ghost-crab 0.3.0

ethereum indexer SDK 👻🦀
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112

use crate::block_handler::{process_blocks, BlockHandlerInstance, ProcessBlocksInput};
use crate::cache::manager::RPCManager;
use crate::event_handler::{process_events, EventHandlerInstance, ProcessEventsInput};
use alloy::primitives::Address;
use tokio::sync::mpsc::error::SendError;
use tokio::sync::mpsc::{self, Receiver, Sender};

pub struct Template {
    pub start_block: u64,
    pub address: Address,
    pub handler: EventHandlerInstance,
}

#[derive(Clone)]
pub struct TemplateManager {
    tx: Sender<Template>,
}

impl TemplateManager {
    pub async fn start(&self, template: Template) -> Result<(), SendError<Template>> {
        self.tx.send(template).await
    }
}

pub struct Indexer {
    handlers: Vec<ProcessEventsInput>,
    rx: Receiver<Template>,
    block_handlers: Vec<ProcessBlocksInput>,
    templates: TemplateManager,
    rpc_manager: RPCManager,
}

impl Indexer {
    pub fn new() -> Indexer {
        let (tx, rx) = mpsc::channel::<Template>(1);

        Indexer {
            handlers: Vec::new(),
            block_handlers: Vec::new(),
            templates: TemplateManager { tx },
            rpc_manager: RPCManager::new(),
            rx,
        }
    }

    pub async fn load_event_handler(&mut self, handler: EventHandlerInstance) {
        if handler.is_template() {
            return;
        }

        let provider = self.rpc_manager.get_or_create(handler.network(), handler.rpc_url()).await;

        self.handlers.push(ProcessEventsInput {
            start_block: handler.start_block(),
            address: handler.address(),
            step: 10_000,
            handler,
            templates: self.templates.clone(),
            provider,
        });
    }

    pub async fn load_block_handler(&mut self, handler: BlockHandlerInstance) {
        let provider = self.rpc_manager.get_or_create(handler.network(), handler.rpc_url()).await;

        self.block_handlers.push(ProcessBlocksInput {
            handler,
            templates: self.templates.clone(),
            provider,
        });
    }

    pub async fn start(mut self) {
        for block_handler in self.block_handlers {
            tokio::spawn(async move {
                if let Err(error) = process_blocks(block_handler).await {
                    println!("Error processing logs for block handler: {error}");
                }
            });
        }

        for handler in self.handlers {
            tokio::spawn(async move {
                if let Err(error) = process_events(handler).await {
                    println!("Error processing logs for handler: {error}");
                }
            });
        }

        // For dynamic sources (Templates)
        while let Some(template) = self.rx.recv().await {
            let network = template.handler.network();
            let rpc_url = template.handler.rpc_url();
            let provider = self.rpc_manager.get_or_create(network, rpc_url).await;

            let handler = ProcessEventsInput {
                start_block: template.start_block,
                address: template.address,
                step: 10_000,
                handler: template.handler,
                templates: self.templates.clone(),
                provider,
            };

            tokio::spawn(async move {
                if let Err(error) = process_events(handler).await {
                    println!("Error processing logs for handler: {error}");
                }
            });
        }
    }
}