acuity_index_substrate/
shared.rs

1use byteorder::BigEndian;
2use serde::{Deserialize, Serialize};
3use sled::{Db, Tree};
4use std::fmt;
5use std::hash::Hash;
6use subxt::utils::H256;
7use tokio::sync::mpsc::UnboundedSender;
8use tokio_tungstenite::tungstenite;
9use zerocopy::{
10    AsBytes,
11    byteorder::{U16, U32},
12};
13use zerocopy_derive::*;
14
15/// Errors this crate can return
16#[derive(thiserror::Error, Debug)]
17pub enum IndexError {
18    #[error("database error")]
19    Sled(#[from] sled::Error),
20    #[error("connection error")]
21    Subxt(#[from] subxt::Error),
22    #[error("connection error")]
23    Tungstenite(#[from] tungstenite::Error),
24    #[error("parse error")]
25    Hex(#[from] hex::FromHexError),
26    #[error("parse error")]
27    ParseError,
28    #[error("connection error")]
29    BlockNotFound(u32),
30    #[error("RPC error")]
31    RpcError(#[from] subxt::ext::subxt_rpcs::Error),
32    #[error("codec error")]
33    CodecError(#[from] subxt::ext::codec::Error),
34    #[error("metadata error")]
35    MetadataError(#[from] subxt::error::MetadataTryFromError),
36}
37
38/// Indexer for a specific chain
39pub trait RuntimeIndexer {
40    type RuntimeConfig: subxt::Config;
41    type ChainKey: IndexKey
42        + Serialize
43        + for<'a> Deserialize<'a>
44        + Clone
45        + Eq
46        + PartialEq
47        + Hash
48        + Send;
49
50    fn get_name() -> &'static str;
51
52    fn get_genesis_hash() -> H256;
53
54    fn get_versions() -> &'static [u32];
55
56    fn get_default_url() -> &'static str;
57
58    fn process_event(
59        indexer: &crate::Indexer<Self>,
60        block_number: u32,
61        event_index: u16,
62        event: subxt::events::EventDetails<Self::RuntimeConfig>,
63    ) -> Result<u32, IndexError>;
64}
65
66pub trait IndexTrees {
67    fn open(db: &Db) -> Result<Self, sled::Error>
68    where
69        Self: Sized;
70    fn flush(&self) -> Result<(), sled::Error>;
71}
72
73/// Database trees for built-in Substrate keys
74#[derive(Clone)]
75pub struct SubstrateTrees {
76    pub account_id: Tree,
77    pub account_index: Tree,
78    pub bounty_index: Tree,
79    pub era_index: Tree,
80    pub message_id: Tree,
81    pub pool_id: Tree,
82    pub preimage_hash: Tree,
83    pub proposal_hash: Tree,
84    pub proposal_index: Tree,
85    pub ref_index: Tree,
86    pub registrar_index: Tree,
87    pub session_index: Tree,
88    pub tip_hash: Tree,
89    pub spend_index: Tree,
90}
91
92impl SubstrateTrees {
93    pub fn open(db: &Db) -> Result<Self, sled::Error> {
94        Ok(SubstrateTrees {
95            account_id: db.open_tree(b"account_id")?,
96            account_index: db.open_tree(b"account_index")?,
97            bounty_index: db.open_tree(b"bounty_index")?,
98            era_index: db.open_tree(b"era_index")?,
99            message_id: db.open_tree(b"message_id")?,
100            pool_id: db.open_tree(b"pool_id")?,
101            preimage_hash: db.open_tree(b"preimage_hash")?,
102            proposal_hash: db.open_tree(b"proposal_hash")?,
103            proposal_index: db.open_tree(b"proposal_index")?,
104            ref_index: db.open_tree(b"ref_index")?,
105            registrar_index: db.open_tree(b"registrar_index")?,
106            session_index: db.open_tree(b"session_index")?,
107            tip_hash: db.open_tree(b"tip_hash")?,
108            spend_index: db.open_tree(b"spend_index")?,
109        })
110    }
111
112    pub fn flush(&self) -> Result<(), sled::Error> {
113        self.account_id.flush()?;
114        self.account_index.flush()?;
115        self.bounty_index.flush()?;
116        self.era_index.flush()?;
117        self.message_id.flush()?;
118        self.pool_id.flush()?;
119        self.preimage_hash.flush()?;
120        self.proposal_hash.flush()?;
121        self.proposal_index.flush()?;
122        self.ref_index.flush()?;
123        self.registrar_index.flush()?;
124        self.session_index.flush()?;
125        self.tip_hash.flush()?;
126        self.spend_index.flush()?;
127        Ok(())
128    }
129}
130
131/// Database trees for the indexer
132#[derive(Clone)]
133pub struct Trees<CT> {
134    pub root: sled::Db,
135    pub span: Tree,
136    pub variant: Tree,
137    pub substrate: SubstrateTrees,
138    pub chain: CT,
139    pub block_events: Tree,
140}
141
142/// On-disk format for variant keys
143#[derive(FromZeroes, FromBytes, AsBytes, Unaligned, PartialEq, Debug)]
144#[repr(C)]
145pub struct VariantKey {
146    pub pallet_index: u8,
147    pub variant_index: u8,
148    pub block_number: U32<BigEndian>,
149    pub event_index: U16<BigEndian>,
150}
151
152/// On-disk format for 32-byte keys
153#[derive(FromZeroes, FromBytes, AsBytes, Unaligned, PartialEq, Debug)]
154#[repr(C)]
155pub struct Bytes32Key {
156    pub key: [u8; 32],
157    pub block_number: U32<BigEndian>,
158    pub event_index: U16<BigEndian>,
159}
160
161/// On-disk format for u32 keys
162#[derive(FromZeroes, FromBytes, AsBytes, Unaligned, PartialEq, Debug)]
163#[repr(C)]
164pub struct U32Key {
165    pub key: U32<BigEndian>,
166    pub block_number: U32<BigEndian>,
167    pub event_index: U16<BigEndian>,
168}
169
170/// Datatype to hold 32-byte keys
171#[derive(Copy, Clone, Debug, PartialEq, Hash, Eq)]
172pub struct Bytes32(pub [u8; 32]);
173
174impl AsRef<[u8; 32]> for Bytes32 {
175    fn as_ref(&self) -> &[u8; 32] {
176        &self.0
177    }
178}
179
180impl From<[u8; 32]> for Bytes32 {
181    fn from(x: [u8; 32]) -> Self {
182        Bytes32(x)
183    }
184}
185
186impl AsRef<[u8]> for Bytes32 {
187    fn as_ref(&self) -> &[u8] {
188        &self.0[..]
189    }
190}
191
192impl Serialize for Bytes32 {
193    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
194    where
195        S: serde::Serializer,
196    {
197        let mut hex_string = "0x".to_owned();
198        hex_string.push_str(&hex::encode(self.0));
199        serializer.serialize_str(&hex_string)
200    }
201}
202
203impl<'de> Deserialize<'de> for Bytes32 {
204    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
205    where
206        D: serde::Deserializer<'de>,
207    {
208        match String::deserialize(deserializer)?.get(2..66) {
209            Some(message_id) => match hex::decode(message_id) {
210                Ok(message_id) => Ok(Bytes32(message_id.try_into().unwrap())),
211                Err(_error) => Err(serde::de::Error::custom("error")),
212            },
213            None => Err(serde::de::Error::custom("error")),
214        }
215    }
216}
217
218impl std::str::FromStr for Bytes32 {
219    type Err = IndexError;
220
221    fn from_str(s: &str) -> Result<Self, Self::Err> {
222        Ok(Bytes32(
223            hex::decode(s)?
224                .try_into()
225                .map_err(|_| IndexError::ParseError)?,
226        ))
227    }
228}
229
230/// All the key types that are built-in to Substrate
231#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, Hash)]
232#[serde(tag = "type", content = "value")]
233pub enum SubstrateKey {
234    AccountId(Bytes32),
235    AccountIndex(u32),
236    BountyIndex(u32),
237    EraIndex(u32),
238    MessageId(Bytes32),
239    PoolId(u32),
240    PreimageHash(Bytes32),
241    ProposalHash(Bytes32),
242    ProposalIndex(u32),
243    RefIndex(u32),
244    RegistrarIndex(u32),
245    SessionIndex(u32),
246    TipHash(Bytes32),
247    SpendIndex(u32),
248}
249
250impl SubstrateKey {
251    pub fn write_db_key(
252        &self,
253        trees: &SubstrateTrees,
254        block_number: u32,
255        event_index: u16,
256    ) -> Result<(), sled::Error> {
257        let block_number = block_number.into();
258        let event_index = event_index.into();
259        match self {
260            SubstrateKey::AccountId(account_id) => {
261                let key = Bytes32Key {
262                    key: account_id.0,
263                    block_number,
264                    event_index,
265                };
266                trees.account_id.insert(key.as_bytes(), &[])?
267            }
268            SubstrateKey::AccountIndex(account_index) => {
269                let key = U32Key {
270                    key: (*account_index).into(),
271                    block_number,
272                    event_index,
273                };
274                trees.account_index.insert(key.as_bytes(), &[])?
275            }
276            SubstrateKey::BountyIndex(bounty_index) => {
277                let key = U32Key {
278                    key: (*bounty_index).into(),
279                    block_number,
280                    event_index,
281                };
282                trees.bounty_index.insert(key.as_bytes(), &[])?
283            }
284            SubstrateKey::EraIndex(era_index) => {
285                let key = U32Key {
286                    key: (*era_index).into(),
287                    block_number,
288                    event_index,
289                };
290                trees.era_index.insert(key.as_bytes(), &[])?
291            }
292            SubstrateKey::MessageId(message_id) => {
293                let key = Bytes32Key {
294                    key: message_id.0,
295                    block_number,
296                    event_index,
297                };
298                trees.message_id.insert(key.as_bytes(), &[])?
299            }
300            SubstrateKey::PoolId(pool_id) => {
301                let key = U32Key {
302                    key: (*pool_id).into(),
303                    block_number,
304                    event_index,
305                };
306                trees.pool_id.insert(key.as_bytes(), &[])?
307            }
308            SubstrateKey::PreimageHash(preimage_hash) => {
309                let key = Bytes32Key {
310                    key: preimage_hash.0,
311                    block_number,
312                    event_index,
313                };
314                trees.preimage_hash.insert(key.as_bytes(), &[])?
315            }
316            SubstrateKey::ProposalHash(proposal_hash) => {
317                let key = Bytes32Key {
318                    key: proposal_hash.0,
319                    block_number,
320                    event_index,
321                };
322                trees.proposal_hash.insert(key.as_bytes(), &[])?
323            }
324            SubstrateKey::ProposalIndex(proposal_index) => {
325                let key = U32Key {
326                    key: (*proposal_index).into(),
327                    block_number,
328                    event_index,
329                };
330                trees.proposal_index.insert(key.as_bytes(), &[])?
331            }
332            SubstrateKey::RefIndex(ref_index) => {
333                let key = U32Key {
334                    key: (*ref_index).into(),
335                    block_number,
336                    event_index,
337                };
338                trees.ref_index.insert(key.as_bytes(), &[])?
339            }
340            SubstrateKey::RegistrarIndex(registrar_index) => {
341                let key = U32Key {
342                    key: (*registrar_index).into(),
343                    block_number,
344                    event_index,
345                };
346                trees.registrar_index.insert(key.as_bytes(), &[])?
347            }
348            SubstrateKey::SessionIndex(session_index) => {
349                let key = U32Key {
350                    key: (*session_index).into(),
351                    block_number,
352                    event_index,
353                };
354                trees.session_index.insert(key.as_bytes(), &[])?
355            }
356            SubstrateKey::TipHash(tip_hash) => {
357                let key = Bytes32Key {
358                    key: tip_hash.0,
359                    block_number,
360                    event_index,
361                };
362                trees.tip_hash.insert(key.as_bytes(), &[])?
363            }
364            SubstrateKey::SpendIndex(spend_index) => {
365                let key = U32Key {
366                    key: (*spend_index).into(),
367                    block_number,
368                    event_index,
369                };
370                trees.spend_index.insert(key.as_bytes(), &[])?
371            }
372        };
373        Ok(())
374    }
375}
376
377pub trait IndexKey {
378    type ChainTrees: IndexTrees + Send + Sync + Clone;
379
380    fn write_db_key(
381        &self,
382        trees: &Self::ChainTrees,
383        block_number: u32,
384        event_index: u16,
385    ) -> Result<(), sled::Error>;
386
387    fn get_key_events(&self, trees: &Self::ChainTrees) -> Vec<Event>;
388}
389
390/// All the key types for the chain
391#[derive(Serialize, Deserialize, Clone, Debug, Eq, PartialEq, Hash)]
392#[serde(tag = "type", content = "value")]
393pub enum Key<CK: IndexKey> {
394    Variant(u8, u8),
395    Substrate(SubstrateKey),
396    Chain(CK),
397}
398
399impl<CK: IndexKey> Key<CK> {
400    pub fn write_db_key(
401        &self,
402        trees: &Trees<CK::ChainTrees>,
403        block_number: u32,
404        event_index: u16,
405    ) -> Result<(), sled::Error> {
406        match self {
407            Key::Variant(pallet_index, variant_index) => {
408                let key = VariantKey {
409                    pallet_index: *pallet_index,
410                    variant_index: *variant_index,
411                    block_number: block_number.into(),
412                    event_index: event_index.into(),
413                };
414                trees.variant.insert(key.as_bytes(), &[])?;
415            }
416            Key::Substrate(substrate_key) => {
417                substrate_key.write_db_key(&trees.substrate, block_number, event_index)?;
418            }
419            Key::Chain(chain_key) => {
420                chain_key.write_db_key(&trees.chain, block_number, event_index)?;
421            }
422        };
423        Ok(())
424    }
425}
426
427/// JSON request messages
428#[derive(Deserialize, Debug, Clone)]
429#[serde(tag = "type")]
430pub enum RequestMessage<CK: IndexKey> {
431    Status,
432    SubscribeStatus,
433    UnsubscribeStatus,
434    Variants,
435    GetEvents { key: Key<CK> },
436    SubscribeEvents { key: Key<CK> },
437    UnsubscribeEvents { key: Key<CK> },
438    SizeOnDisk,
439}
440
441/// Identifies an event by block number and event index
442#[derive(Serialize, Debug, Clone, Deserialize, PartialEq)]
443#[serde(rename_all = "camelCase")]
444pub struct Event {
445    pub block_number: u32,
446    pub event_index: u16,
447}
448
449#[derive(Serialize, Debug, Clone, Deserialize, PartialEq)]
450#[serde(rename_all = "camelCase")]
451pub struct Block {
452    pub block_number: u32,
453    pub bytes: Vec<u8>,
454}
455
456impl fmt::Display for Event {
457    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
458        write!(
459            f,
460            "block number: {}, event index: {}",
461            self.block_number, self.event_index
462        )
463    }
464}
465
466/// Index and name of an event type
467#[derive(Deserialize, Serialize, Debug, Clone, PartialEq)]
468pub struct EventMeta {
469    pub index: u8,
470    pub name: String,
471}
472
473/// Index, name and list of event types for a pallet
474#[derive(Serialize, Debug, Clone, Deserialize, PartialEq)]
475pub struct PalletMeta {
476    pub index: u8,
477    pub name: String,
478    pub events: Vec<EventMeta>,
479}
480
481/// On-disk format for span value
482#[derive(FromZeroes, FromBytes, AsBytes, Unaligned, PartialEq, Debug)]
483#[repr(C)]
484pub struct SpanDbValue {
485    pub start: U32<BigEndian>,
486    pub version: U16<BigEndian>,
487    pub index_variant: u8,
488    pub store_events: u8,
489}
490
491/// Start and end block number for a span of blocks
492#[derive(Serialize, Debug, Clone, PartialEq, Deserialize)]
493pub struct Span {
494    pub start: u32,
495    pub end: u32,
496}
497
498impl fmt::Display for Span {
499    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
500        write!(f, "start: {}, end: {}", self.start, self.end)
501    }
502}
503
504/// JSON response messages
505#[derive(Serialize, Debug, Clone)]
506#[serde(tag = "type", content = "data")]
507#[serde(rename_all = "camelCase")]
508pub enum ResponseMessage<CK: IndexKey> {
509    Status(Vec<Span>),
510    Variants(Vec<PalletMeta>),
511    Events {
512        key: Key<CK>,
513        events: Vec<Event>,
514        block_events: Vec<Block>,
515    },
516    Subscribed,
517    Unsubscribed,
518    SizeOnDisk(u64),
519    //    Error,
520}
521
522/// Subscription message sent from a WebSocket connection thread to the indexer thread
523#[derive(Debug)]
524pub enum SubscriptionMessage<CK: IndexKey> {
525    SubscribeStatus {
526        sub_response_tx: UnboundedSender<ResponseMessage<CK>>,
527    },
528    UnsubscribeStatus {
529        sub_response_tx: UnboundedSender<ResponseMessage<CK>>,
530    },
531    SubscribeEvents {
532        key: Key<CK>,
533        sub_response_tx: UnboundedSender<ResponseMessage<CK>>,
534    },
535    UnsubscribeEvents {
536        key: Key<CK>,
537        sub_response_tx: UnboundedSender<ResponseMessage<CK>>,
538    },
539}
acuity_index_substrate/shared.rs

acuity_index_substrate/
shared.rs
