diff --git a/src/services/stellar/ledgerScanner.ts b/src/services/stellar/ledgerScanner.ts new file mode 100644 index 00000000..e03023fb --- /dev/null +++ b/src/services/stellar/ledgerScanner.ts @@ -0,0 +1,140 @@ +import { getStellarServer } from "../../config/stellar"; +import { + getEventSyncCursor, + setEventSyncCursor, + INITIAL_SYNC_CURSOR, +} from "../../database/eventSyncStateRepository"; +import logger from "../../utils/logger"; + +export type LedgerTransactionHandler = ( + tx: any, +) => Promise | void; + +export interface LedgerScannerConfig { + /** Stream key or account identifier for cursor persistence. */ + streamKey: string; + /** Stellar account/contract address to scan transactions for. */ + contractId?: string; + /** Number of transactions per page (max 200). */ + chunkSize?: number; + /** Poll interval in milliseconds when at the ledger tip. */ + pollIntervalMs?: number; + /** Injected Horizon server for testing. */ + horizon?: any; +} + +export interface LedgerScannerResult { + pages: number; + transactions: number; +} + +/** + * LedgerScanner + * Optimizes the ledger listener loop to ensure no transactions are missed + * during brief database reconnects by storing the last scanned cursor in the database + * and fetching ledger records sequentially using transaction paging tokens. + */ +export class LedgerScanner { + private readonly streamKey: string; + private readonly contractId?: string; + private readonly chunkSize: number; + private readonly pollIntervalMs: number; + private readonly horizon: any; + + private pollTimer: ReturnType | null = null; + private stopped = false; + + constructor(config: LedgerScannerConfig) { + this.streamKey = config.streamKey; + this.contractId = config.contractId; + this.chunkSize = Math.min(Math.max(1, config.chunkSize ?? 200), 200); + this.pollIntervalMs = config.pollIntervalMs ?? 5000; + this.horizon = config.horizon ?? getStellarServer(); + } + + /** + * Runs a single sequential scanning pass from the persisted database cursor. + */ + async scanOnce(handler: LedgerTransactionHandler): Promise { + let pages = 0; + let transactions = 0; + let cursor = ( + (await getEventSyncCursor(this.streamKey)) ?? INITIAL_SYNC_CURSOR + ); + + while (!this.stopped) { + const response = await this.fetchPage(cursor); + const records = response.records ?? []; + + if (records.length === 0) break; + + for (const tx of records) { + await handler(tx); + transactions++; + } + + const lastRecord = records[records.length - 1]; + cursor = lastRecord.paging_token ?? cursor; + + // Persist last scanned cursor in database to prevent duplicates and handle reconnects + await setEventSyncCursor(this.streamKey, cursor); + pages++; + + // If records fetched are less than chunk size, we have reached the tip of the ledger + if (records.length < this.chunkSize) break; + } + + return { pages, transactions }; + } + + /** + * Starts the continuous ledger scanning loop. + */ + start(handler: LedgerTransactionHandler): void { + if (this.pollTimer) return; + + this.stopped = false; + void this.runLoop(handler); + + this.pollTimer = setInterval(() => { + void this.runLoop(handler); + }, this.pollIntervalMs); + } + + /** + * Stops the ledger scanning loop. + */ + stop(): void { + this.stopped = true; + if (this.pollTimer) { + clearInterval(this.pollTimer); + this.pollTimer = null; + } + } + + private async runLoop(handler: LedgerTransactionHandler): Promise { + if (this.stopped) return; + try { + await this.scanOnce(handler); + } catch (error) { + logger.error( + `Ledger scanner failed for stream "${this.streamKey}":`, + error, + ); + } + } + + private fetchPage(cursor: string): Promise { + let call = this.horizon.transactions(); + + if (this.contractId) { + call = call.forAccount(this.contractId); + } + + return call + .limit(this.chunkSize) + .order("asc") + .cursor(cursor) + .call(); + } +}