feat(listeners): implement idempotent restart from persisted cursor

docs/horizon-listener.md

@@ -655,6 +655,7 @@ This module listens for bond creation events from Stellar/Horizon and syncs iden
 - Subscribes to Horizon for bond creation events
 - Parses event payload (identity, amount, duration, etc.)
 - Upserts identity and bond records in PostgreSQL
+- **Idempotent Restart & Gap-Free Resumption**: Loads the saved `bond_creation` cursor checkpoint from the `horizon_cursors` table on startup (falling back to `'now'` only on the first boot or DB error) to ensure no blockchain operations are missed across process restarts.
 - Handles reconnection and backfill
 - Comprehensive tests with mocked Horizon
 

src/listeners/__tests__/horizonBondEvents.test.ts

@@ -1,4 +1,5 @@
-import { describe, it, expect, vi, afterEach } from "vitest";
+import { describe, it, expect, vi, afterEach, beforeEach } from "vitest";
+import { CursorRepository } from "../../db/repositories/cursorRepository.js";
 
 vi.mock("prom-client", () => ({
   register: {},
@@ -19,6 +20,7 @@ vi.mock("../../observability/horizonMetrics.js", () => ({
 
 let capturedHandlers: { onmessage?: any; onerror?: any } = {};
 let streamCallCount = 0;
+let lastCursorVal: string | undefined;
 
 vi.mock("@stellar/stellar-sdk", () => {
   return {
@@ -27,7 +29,10 @@ vi.mock("@stellar/stellar-sdk", () => {
         return {
           operations: vi.fn().mockReturnValue({
             forAsset: vi.fn().mockReturnThis(),
-            cursor: vi.fn().mockReturnThis(),
+            cursor: vi.fn().mockImplementation(function(c: string) {
+              lastCursorVal = c;
+              return this;
+            }),
             stream: vi.fn().mockImplementation(function(handlers: any) {
               streamCallCount++;
               capturedHandlers.onmessage = handlers.onmessage;
@@ -52,6 +57,7 @@ describe("subscribeBondCreationEvents", () => {
     vi.clearAllMocks();
     capturedHandlers = {};
     streamCallCount = 0;
+    lastCursorVal = undefined;
   });
 
   it("opens exactly ONE stream on subscribe", () => {
@@ -110,4 +116,61 @@ describe("subscribeBondCreationEvents", () => {
     await capturedHandlers.onerror?.(new Error("test"));
     expect(streamCallCount).toBe(countBefore);
   });
+
+  describe("Idempotent restart with database cursor", () => {
+    let mockPool: any;
+
+    beforeEach(() => {
+      mockPool = {} as any;
+    });
+
+    it("resumes from saved cursor when present in CursorRepository", async () => {
+      const spy = vi.spyOn(CursorRepository.prototype, "findByStreamName")
+        .mockResolvedValue({
+          streamName: "bond_creation",
+          pagingToken: "9876543210",
+          lastCheckpoint: new Date(),
+          createdAt: new Date(),
+          updatedAt: new Date(),
+        });
+
+      const h = subscribeBondCreationEvents({ captureFailure: vi.fn() }, undefined, mockPool);
+
+      // Wait for the async initAndStart microtasks to run
+      await new Promise((resolve) => process.nextTick(resolve));
+
+      expect(spy).toHaveBeenCalledWith("bond_creation");
+      expect(lastCursorVal).toBe("9876543210");
+      expect(streamCallCount).toBe(1);
+      h.stop();
+    });
+
+    it("falls back to 'now' if CursorRepository.findByStreamName returns null", async () => {
+      const spy = vi.spyOn(CursorRepository.prototype, "findByStreamName")
+        .mockResolvedValue(null);
+
+      const h = subscribeBondCreationEvents({ captureFailure: vi.fn() }, undefined, mockPool);
+
+      await new Promise((resolve) => process.nextTick(resolve));
+
+      expect(spy).toHaveBeenCalledWith("bond_creation");
+      expect(lastCursorVal).toBe("now");
+      expect(streamCallCount).toBe(1);
+      h.stop();
+    });
+
+    it("falls back to 'now' if CursorRepository.findByStreamName throws an error", async () => {
+      const spy = vi.spyOn(CursorRepository.prototype, "findByStreamName")
+        .mockRejectedValue(new Error("Database connection error"));
+
+      const h = subscribeBondCreationEvents({ captureFailure: vi.fn() }, undefined, mockPool);
+
+      await new Promise((resolve) => process.nextTick(resolve));
+
+      expect(spy).toHaveBeenCalledWith("bond_creation");
+      expect(lastCursorVal).toBe("now");
+      expect(streamCallCount).toBe(1);
+      h.stop();
+    });
+  });
 });

src/listeners/horizonBondEvents.ts

@@ -116,8 +116,26 @@ export function subscribeBondCreationEvents(
       });
   };
 
+  const initAndStart = async () => {
+    if (cursorRepo) {
+      try {
+        const savedCursor = await cursorRepo.findByStreamName(STREAM_NAME);
+        if (savedCursor) {
+          cursor = savedCursor.pagingToken;
+          console.log(`[${STREAM_NAME}] Resuming from saved cursor: ${cursor}`);
+        } else {
+          console.log(`[${STREAM_NAME}] No saved cursor found, starting from: ${cursor}`);
+        }
+      } catch (err) {
+        console.error(`[${STREAM_NAME}] Failed to load saved cursor, falling back to: ${cursor}`, err);
+      }
+    }
+    startStream();
+  };
+
   // Start exactly ONE stream
-  startStream();
+  initAndStart();
+
 
   return {
     stop: () => {