1+ #!/usr/bin/env node
2+
3+ const { program } = require ( 'commander' ) ;
4+ const cliProgress = require ( 'cli-progress' ) ;
5+ const dotenv = require ( 'dotenv' ) ;
6+ const path = require ( 'path' ) ;
7+
8+ // Load environment variables
9+ dotenv . config ( { path : path . resolve ( __dirname , '../.env' ) } ) ;
10+
11+ // Import core services from your existing structure
12+ const { loadConfig } = require ( '../src/config' ) ;
13+ const { SorobanIndexerWorker } = require ( '../src/services/sorobanIndexerWorker' ) ; // Adjust if path differs
14+ const { getVaultService } = require ( '../src/services/vaultService' ) ;
15+
16+ // CLI Setup
17+ program
18+ . name ( 'sync:historical' )
19+ . description ( 'Historical backfill tool for SubStream Soroban subscription events' )
20+ . option ( '--start-ledger <number>' , 'Starting ledger sequence (inclusive)' , parseInt )
21+ . option ( '--end-ledger <number>' , 'Ending ledger sequence (inclusive)' , parseInt )
22+ . option ( '--start <number>' , 'Alias for --start-ledger' , parseInt )
23+ . option ( '--end <number>' , 'Alias for --end-ledger' , parseInt )
24+ . option ( '--batch-size <number>' , 'Number of events per DB batch' , parseInt )
25+ . option ( '--contract-ids <string>' , 'Comma-separated Soroban contract IDs to filter' )
26+ . parse ( process . argv ) ;
27+
28+ const opts = program . opts ( ) ;
29+ const startLedger = opts . startLedger || opts . start ;
30+ const endLedger = opts . endLedger || opts . end ;
31+ const batchSize = opts . batchSize || 800 ; // Safe default to avoid overwhelming DB/RPC
32+
33+ if ( ! startLedger || ! endLedger || startLedger > endLedger ) {
34+ console . error ( '❌ Error: --start-ledger and --end-ledger are required and start must be <= end' ) ;
35+ console . error ( 'Example: node scripts/sync-historical.js --start 45000000 --end 45100000' ) ;
36+ process . exit ( 1 ) ;
37+ }
38+
39+ console . log ( `🚀 SubStream Historical Backfill Started` ) ;
40+ console . log ( ` Range: ${ startLedger } ${ endLedger } ) ;
41+ console . log ( ` Batch size: ${ batchSize } ) ;
42+
43+ const progressBar = new cliProgress . SingleBar ( {
44+ format : 'Backfill Progress | {bar} | {percentage}% | Ledger {value}/{total} | ETA: {eta}s' ,
45+ barCompleteChar : '█' ,
46+ barIncompleteChar : '░' ,
47+ hideCursor : true
48+ } ) ;
49+
50+ async function main ( ) {
51+ let vaultService = null ;
52+ let config = null ;
53+
54+ try {
55+ // Initialize Vault if enabled (same as your worker.js)
56+ if ( process . env . VAULT_ENABLED === 'true' ) {
57+ vaultService = getVaultService ( {
58+ vaultAddr : process . env . VAULT_ADDR ,
59+ vaultRole : process . env . VAULT_ROLE ,
60+ authPath : process . env . VAULT_AUTH_PATH ,
61+ secretPath : process . env . VAULT_SECRET_PATH ,
62+ dbPath : process . env . VAULT_DB_PATH
63+ } ) ;
64+ await vaultService . initialize ( ) ;
65+ console . log ( '[Vault] Secrets loaded successfully' ) ;
66+ }
67+
68+ config = await loadConfig ( process . env , vaultService ) ;
69+
70+ // Initialize the same SorobanIndexerWorker used in live indexing
71+ const indexer = new SorobanIndexerWorker ( config ) ; // Pass config if constructor accepts it
72+
73+ progressBar . start ( endLedger - startLedger + 1 , 0 ) ;
74+
75+ let currentLedger = startLedger ;
76+ let totalProcessed = 0 ;
77+
78+ while ( currentLedger <= endLedger ) {
79+ const batchEnd = Math . min ( currentLedger + 99 , endLedger ) ; // Fetch in safe ledger chunks (RPC limit friendly)
80+
81+ console . log ( `\nFetching events from ledger ${ currentLedger } ${ batchEnd } ) ;
82+
83+ // Use the same method your SorobanIndexerWorker uses internally for fetching events
84+ const events = await fetchEventsInRange ( indexer , currentLedger , batchEnd , opts . contractIds ) ;
85+
86+ if ( events && events . length > 0 ) {
87+ // Push through the EXACT same ingestion pipeline as live events
88+ const processed = await processEventsThroughIndexer ( indexer , events , batchSize ) ;
89+ totalProcessed += processed ;
90+ console . log ( ` ✓ Processed ${ processed } ) ;
91+ } else {
92+ console . log ( ` No relevant events in this range` ) ;
93+ }
94+
95+ progressBar . update ( currentLedger - startLedger + 1 ) ;
96+ currentLedger = batchEnd + 1 ;
97+ }
98+
99+ progressBar . stop ( ) ;
100+ console . log ( `\n✅ Historical backfill completed successfully!` ) ;
101+ console . log ( ` Total events processed: ${ totalProcessed } ) ;
102+ console . log ( ` Range: ${ startLedger } ${ endLedger } ) ;
103+
104+ } catch ( error ) {
105+ progressBar . stop ( ) ;
106+ console . error ( '❌ Backfill failed:' , error . message ) ;
107+ if ( error . stack ) console . error ( error . stack ) ;
108+ process . exit ( 1 ) ;
109+ } finally {
110+ if ( vaultService ) {
111+ try { await vaultService . cleanup ( ) ; } catch ( e ) { }
112+ }
113+ }
114+ }
115+
116+ /**
117+ * Fetch events using the same Soroban RPC logic as your live indexer
118+ */
119+ async function fetchEventsInRange ( indexer , startLedger , endLedger , contractIdsStr ) {
120+ try {
121+ // Prefer using internal method from SorobanIndexerWorker if exposed
122+ // Otherwise fall back to direct RPC call
123+ if ( typeof indexer . fetchEvents === 'function' ) {
124+ return await indexer . fetchEvents ( {
125+ startLedger,
126+ endLedger,
127+ contractIds : contractIdsStr ? contractIdsStr . split ( ',' ) : ( process . env . SUBSTREAM_CONTRACT_IDS || '' ) . split ( ',' )
128+ } ) ;
129+ }
130+
131+ // Direct RPC fallback using soroban-client (you already have this dependency)
132+ const { Server } = require ( 'soroban-client' ) ;
133+ const rpcUrl = process . env . SOROBAN_RPC_URL || config ?. soroban ?. rpcUrl || 'https://rpc.mainnet.stellar.org' ;
134+ const server = new Server ( rpcUrl ) ;
135+
136+ const filters = [ {
137+ type : 'contract' ,
138+ contractIds : contractIdsStr ? contractIdsStr . split ( ',' ) : [ ] ,
139+ // Add topic filters for subscription events if needed, e.g.:
140+ // topics: [[Buffer.from('subscribe').toString('base64')]]
141+ } ] ;
142+
143+ const response = await server . getEvents ( {
144+ startLedger : startLedger . toString ( ) ,
145+ endLedger : endLedger . toString ( ) ,
146+ filters,
147+ pagination : { limit : 5000 } // High but safe limit
148+ } ) ;
149+
150+ return response . events || [ ] ;
151+ } catch ( err ) {
152+ console . warn ( `Warning: Failed to fetch events ${ startLedger } ${ endLedger } , err . message ) ;
153+ return [ ] ;
154+ }
155+ }
156+
157+ /**
158+ * Process events using the exact same ingestion pipeline as live SorobanIndexerWorker
159+ * This ensures full consistency and reuses all business logic + idempotency
160+ */
161+ async function processEventsThroughIndexer ( indexer , events , batchSize ) {
162+ if ( ! events . length ) return 0 ;
163+
164+ let processedCount = 0 ;
165+ const chunks = chunkArray ( events , batchSize ) ;
166+
167+ for ( const chunk of chunks ) {
168+ // If your SorobanIndexerWorker has a method like ingestEvents or processEventBatch, use it
169+ if ( typeof indexer . processEvents === 'function' ) {
170+ await indexer . processEvents ( chunk ) ;
171+ processedCount += chunk . length ;
172+ }
173+ else if ( typeof indexer . ingestEvent === 'function' ) {
174+ // Fallback: process one by one (slower but safe)
175+ for ( const event of chunk ) {
176+ try {
177+ await indexer . ingestEvent ( event ) ; // This should contain your idempotency logic
178+ processedCount ++ ;
179+ } catch ( e ) {
180+ console . warn ( 'Failed to ingest single event:' , e . message ) ;
181+ }
182+ }
183+ }
184+ else {
185+ console . error ( 'Error: SorobanIndexerWorker does not expose processEvents or ingestEvent method.' ) ;
186+ console . error ( 'Please expose a public method in SorobanIndexerWorker for batch ingestion.' ) ;
187+ process . exit ( 1 ) ;
188+ }
189+ }
190+
191+ return processedCount ;
192+ }
193+
194+ function chunkArray ( array , size ) {
195+ const result = [ ] ;
196+ for ( let i = 0 ; i < array . length ; i += size ) {
197+ result . push ( array . slice ( i , i + size ) ) ;
198+ }
199+ return result ;
200+ }
201+
202+ // Run the tool
203+ main ( ) . catch ( err => {
204+ console . error ( 'Fatal error:' , err ) ;
205+ process . exit ( 1 ) ;
206+ } ) ;
0 commit comments