Feat/indexer polling and event filter enhancements

src/indexer/db.ts

@@ -75,7 +75,7 @@
 } as const;
 
 /** Index names created by the schema-manager migration (#259). */
 export const SCHEMA_MANAGER_INDEXES = {
   monitoredContractsActive: "idx_monitored_contracts_active",
   eventsCreatedAt: "idx_events_created_at",
   eventsContractTypeLedger: "idx_events_contract_type_ledger",
@@ -201,7 +201,7 @@
 ];
 
 /** Index names created by the SQLite schema manager lookup-index migration (#259). */
 export const SCHEMA_MANAGER_INDEXES = {
   monitoredContractsActive: "idx_monitored_contracts_active",
   eventsCreatedAt: "idx_events_created_at",
   eventsContractTypeLedger: "idx_events_contract_type_ledger",
@@ -216,7 +216,7 @@
 }
 
 /** Index names created by the version-5 migration (#259), for test assertions. */
 export const SCHEMA_MANAGER_INDEXES = {
   monitoredContractsActive: "idx_monitored_contracts_active",
   eventsCreatedAt: "idx_events_created_at",
   eventsContractTypeLedger: "idx_events_contract_type_ledger",
@@ -499,7 +499,7 @@
  * and tests can assert the exact lookup indexes the schema manager relies on
  * without hardcoding names (#259).
  */
 export const SCHEMA_MANAGER_INDEXES = [
   "idx_events_contract_id",
   "idx_events_ledger_sequence",
   "idx_events_contract_ledger",
@@ -998,7 +998,7 @@
  * Returns the number of rows actually inserted (excludes rows ignored as
  * duplicates).
  */
 export function insertHistoricalEventBatch(events: EventRow[]): number {
   const db = getDb();
 
   const insertStmt = db.prepare(`
@@ -1149,6 +1149,47 @@
   });
 }
 
+/**
+ * Insert a batch of events WITHOUT touching the live indexer_state ledger
+ * pointer. Used for custom historical event imports (event_type_filter's
+ * dynamic start/end ledger support) so a backfill over an arbitrary past
+ * range can never advance or rewind last_ledger_sequence - only the live
+ * poller (insertEventBatch, driven strictly by lastLedger+1..currentLedger)
+ * is allowed to move that pointer. Rows still go through INSERT OR IGNORE
+ * against the same UNIQUE(contract_id, ledger_sequence, event_type)
+ * constraint, so re-running a historical import is idempotent exactly like
+ * the live poller.
+ *
+ * Returns the number of rows actually inserted (excludes rows ignored as
+ * duplicates).
+ */
+export function insertHistoricalEventBatch(events: EventRow[]): number {
+  const db = getDb();
+
+  const insertStmt = db.prepare(`
+    INSERT OR IGNORE INTO events
+    (contract_id, event_type, ledger_sequence, timestamp, data_json)
+    VALUES (?, ?, ?, ?, ?)
+  `);
+
+  const batchTransaction = db.transaction(() => {
+    let inserted = 0;
+    for (const ev of events) {
+      const result = insertStmt.run(
+        ev.contractId,
+        ev.eventType,
+        ev.ledgerSequence,
+        ev.timestamp,
+        ev.dataJson
+      );
+      if (result.changes > 0) inserted++;
+    }
+    return inserted;
+  });
+
+  return batchTransaction();
+}
+
 // ---------------------------------------------------------------------------
 // Event queries
 // ---------------------------------------------------------------------------
@@ -1548,7 +1589,7 @@
  * it rejects any event whose ledger_sequence falls outside the declared
  * range so a mistyped range can't silently import the wrong window.
  */
 export function insertHistoricalEventBatch(
   events: EventRow[],
   range: { startLedger: unknown; endLedger: unknown },
   options: InsertHistoricalEventBatchOptions = {},