Add OData filtering and multi-source ingest failover

README.md

@@ -258,6 +258,9 @@ const res = await fetch(
 const data = await res.json();
 console.log(data);
 
+// OData-style filters and projections are also supported:
+// `http://localhost:3000/transfers/address/${ADDRESS}?$filter=ledger gt 1000 and contains(contractId,'CB64')&$select=contractId,amount&cursor=...`
+
 // Expected response  (same shape as /transfers/incoming — adds "direction" per row)
 // {
 //   "total": 85,
@@ -401,6 +404,8 @@ curl "http://localhost:3000/transfers/tx/abcdef1234567890..."
 | `STELLAR_NETWORK`     | —             | `testnet` or `mainnet`. Testnet auto-configures the default RPC URL.                          |
 | `SOROBAN_RPC_URL`     | *(see below)* | Soroban RPC endpoint. Overrides any network default. Required when `STELLAR_NETWORK=mainnet`. |
 | `STELLAR_RPC_URL`     | —             | Backward-compat alias for `SOROBAN_RPC_URL`. Used when `SOROBAN_RPC_URL` is unset.            |
+| `HORIZON_URL`         | —             | Optional Horizon endpoint used as a fallback source when RPC is unhealthy.                    |
+| `HORIZON_EVENTS_PATH`  | `/events`     | Horizon contract-events path used by the fallback source.                                      |
 | `START_LEDGER`        | *(tip)*       | Ledger to start indexing from. Leave blank to resume from DB state or start near the tip.     |
 | `POLL_INTERVAL_MS`    | `6000`        | Polling interval in ms (\~1 ledger ≈ 6 s)                                                     |
 | `CONTRACT_IDS`        | *(all)*       | Comma-separated token contract IDs to watch. Empty = watch all (very heavy on mainnet)        |
@@ -418,6 +423,10 @@ Wraith resolves the RPC endpoint in this order and fails fast at startup if noth
 4. `STELLAR_NETWORK=mainnet` → **error**: requires explicit `SOROBAN_RPC_URL`
 5. Nothing set → **error**: clear message explaining what to configure
 
+### Indexer Source Fallback
+
+If `HORIZON_URL` is set, the indexer checks the RPC source first and switches to Horizon when the RPC health check fails. It switches back automatically once RPC becomes healthy again.
+
 ### Mainnet RPC Providers
 
 | Provider           | URL pattern                                               |

src/__tests__/indexerSources.test.ts

@@ -0,0 +1,55 @@
+jest.mock("../rpc", () => ({
+  getLatestLedger: jest.fn(),
+  fetchEventsSafe: jest.fn(),
+}));
+
+import { createSourceSwitcherWithConfig } from "../indexer/sources";
+import { getLatestLedger, fetchEventsSafe } from "../rpc";
+
+const mockGetLatestLedger = getLatestLedger as jest.MockedFunction<typeof getLatestLedger>;
+const mockFetchEventsSafe = fetchEventsSafe as jest.MockedFunction<typeof fetchEventsSafe>;
+
+describe("Indexer source switcher", () => {
+  beforeEach(() => {
+    mockGetLatestLedger.mockReset();
+    mockFetchEventsSafe.mockReset();
+  });
+
+  it("falls back to Horizon when RPC is unhealthy", async () => {
+    mockGetLatestLedger.mockRejectedValue(new Error("rpc down"));
+
+    const fetchImpl = jest.fn()
+      .mockResolvedValueOnce({
+        ok: true,
+        json: async () => ({ _embedded: { records: [{ sequence: 123 }] } }),
+      })
+      .mockResolvedValueOnce({
+        ok: true,
+        json: async () => ({
+          records: [
+            {
+              id: "evt-1",
+              ledger: 122,
+              ledgerCloseTime: "2025-01-01T00:00:00Z",
+              contractId: "C123",
+              txHash: "tx-1",
+              topic: [],
+              value: {},
+            },
+          ],
+        }),
+      });
+
+    const switcher = createSourceSwitcherWithConfig({
+      horizonUrl: "https://horizon.example",
+      fetchImpl: fetchImpl as never,
+    });
+    const result = await switcher.fetchEvents(120, 123, ["C123"], 50);
+
+    expect(result.highestLedger).toBe(122);
+    expect(result.events).toHaveLength(1);
+    expect(result.events[0].contractId).toBe("C123");
+    expect(fetchImpl).toHaveBeenCalledTimes(2);
+    expect(mockFetchEventsSafe).not.toHaveBeenCalled();
+  });
+});

src/__tests__/odata.test.ts

@@ -0,0 +1,29 @@
+import { parseODataFilter, parseODataSelect } from "../lib/odata";
+
+describe("OData helper", () => {
+  const fields = {
+    contractId: { type: "string" as const },
+    ledger: { type: "number" as const },
+    ledgerClosedAt: { type: "date" as const },
+    eventType: { type: "string" as const },
+  };
+
+  it("parses a safe AND-only filter", () => {
+    expect(parseODataFilter("ledger gt 100 and contains(contractId,'C')", fields)).toEqual({
+      AND: [
+        { ledger: { gt: 100 } },
+        { contractId: { contains: "C", mode: "insensitive" } },
+      ],
+    });
+  });
+
+  it("rejects unsafe or unsupported filter expressions", () => {
+    expect(() => parseODataFilter("ledger gt 100 or 1 eq 1", fields)).toThrow(/AND combinations/i);
+    expect(() => parseODataFilter("contains(ledger,'1')", fields)).toThrow(/string fields/i);
+  });
+
+  it("parses a projection list and rejects unknown fields", () => {
+    expect(parseODataSelect("contractId, ledger", ["contractId", "ledger"])) .toEqual(["contractId", "ledger"]);
+    expect(() => parseODataSelect("contractId, hacked", ["contractId"])) .toThrow(/Unsupported \$select field/i);
+  });
+});

src/__tests__/routes/transfers.test.ts

@@ -112,7 +112,7 @@
   describe("GET /transfers/incoming/:address", () => {
     it("returns all incoming transfers for a known address", async () => {
       const incoming = SEED_TRANSFERS.filter((t) => t.toAddress === ALICE);
       mockQueryTransfers.mockResolvedValue({ total: incoming.length, transfers: incoming });
 
       const res = await request(app).get(`/transfers/incoming/${ALICE}`);
 
@@ -125,7 +125,7 @@
 
     it("attaches displayAmount to every transfer", async () => {
       const transfer = makeTransfer({ amount: "10000000" });
       mockQueryTransfers.mockResolvedValue({ total: 1, transfers: [transfer] });
 
       const res = await request(app).get(`/transfers/incoming/${ALICE}`);
 
@@ -134,7 +134,7 @@
     });
 
     it("returns empty array for an unknown address", async () => {
       mockQueryTransfers.mockResolvedValue({ total: 0, transfers: [] });
 
       const res = await request(app).get("/transfers/incoming/GUNKNOWNADDRESS");
 
@@ -147,7 +147,7 @@
       const filtered = SEED_TRANSFERS.filter(
         (t) => t.toAddress === ALICE && t.contractId === CONTRACT_A
       );
       mockQueryTransfers.mockResolvedValue({ total: filtered.length, transfers: filtered });
 
       const res = await request(app)
         .get(`/transfers/incoming/${ALICE}`)
@@ -159,8 +159,35 @@
       );
     });
 
+    it("forwards OData filter, select, and cursor params", async () => {
+      mockQueryTransfers.mockResolvedValue({
+        total: 1,
+        transfers: [makeTransfer({ amount: "10000000" })],
+        nextCursor: "cursor-1",
+      });
+
+      const res = await request(app)
+        .get(`/transfers/incoming/${ALICE}`)
+        .query({
+          $filter: "ledger gt 1000 and contains(contractId,'C')",
+          $select: "contractId,amount",
+          cursor: "cursor-0",
+        });
+
+      expect(res.status).toBe(200);
+      expect(res.body.nextCursor).toBe("cursor-1");
+      expect(mockQueryTransfers).toHaveBeenCalledWith(
+        expect.objectContaining({
+          filter: "ledger gt 1000 and contains(contractId,'C')",
+          select: ["contractId", "amount"],
+          cursor: "cursor-0",
+        })
+      );
+      expect(res.body.transfers[0].displayAmount).toBe("1.0000000");
+    });
+
     it("passes fromDate and toDate to queryTransfers", async () => {
       mockQueryTransfers.mockResolvedValue({ total: 2, transfers: SEED_TRANSFERS.slice(14, 16) });
 
       const res = await request(app)
         .get(`/transfers/incoming/${ALICE}`)
@@ -200,7 +227,7 @@
     });
 
     it("accepts valid eventType values", async () => {
       mockQueryTransfers.mockResolvedValue({ total: 1, transfers: [makeTransfer({ eventType: "mint" })] });
 
       const res = await request(app)
         .get(`/transfers/incoming/${ALICE}`)
@@ -213,7 +240,7 @@
     });
 
     it("accepts comma-separated eventType values", async () => {
       mockQueryTransfers.mockResolvedValue({ total: 2, transfers: [] });
 
       const res = await request(app)
         .get(`/transfers/incoming/${ALICE}`)
@@ -227,7 +254,7 @@
 
     it("honours limit and offset for pagination", async () => {
       const page = SEED_TRANSFERS.slice(0, 5);
       mockQueryTransfers.mockResolvedValue({ total: 20, transfers: page });
 
       const res = await request(app)
         .get(`/transfers/incoming/${ALICE}`)
@@ -242,7 +269,7 @@
     });
 
     it("falls back to limit=50, offset=0 when not provided", async () => {
       mockQueryTransfers.mockResolvedValue({ total: 0, transfers: [] });
 
       await request(app).get(`/transfers/incoming/${ALICE}`);
 
@@ -252,7 +279,7 @@
     });
 
     it("forwards fromLedger and toLedger filters", async () => {
       mockQueryTransfers.mockResolvedValue({ total: 3, transfers: SEED_TRANSFERS.slice(0, 3) });
 
       await request(app)
         .get(`/transfers/incoming/${ALICE}`)
@@ -343,14 +370,14 @@
     });
 
     it("honours pagination params", async () => {
-      mockQueryAllTransfers.mockResolvedValue({ total: 20, transfers: [] });
+      mockQueryAllTransfers.mockResolvedValue({ total: 20, transfers: [], nextCursor: "cursor-2" });
 
       await request(app)
         .get(`/transfers/address/${ALICE}`)
-        .query({ limit: "10", offset: "5" });
+        .query({ limit: "10", offset: "5", cursor: "cursor-1", $select: "contractId,direction" });
 
       expect(mockQueryAllTransfers).toHaveBeenCalledWith(
-        expect.objectContaining({ limit: 10, offset: 5 })
+        expect.objectContaining({ limit: 10, offset: 5, cursor: "cursor-1", select: ["contractId", "direction"] })
       );
     });
 

src/api.ts

@@ -37,6 +37,11 @@ const withDisplay = <T extends { amount: string }>(t: T) => ({
   displayAmount: toDisplayAmount(t.amount),
 });
 
+function parseSelectQuery(value: unknown): string[] | undefined {
+  if (typeof value !== "string" || !value.trim()) return undefined;
+  return value.split(",").map((item) => item.trim()).filter(Boolean);
+}
+
 const VALID_EVENT_TYPES = new Set(["transfer", "mint", "burn", "clawback"]);
 
 // ── CSV utilities ─────────────────────────────────────────────────────────────
@@ -216,7 +221,7 @@ export function createApp(): express.Application {
     async (req: Request, res: Response, next: NextFunction) => {
       try {
         const { address } = req.params;
-        const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset } = req.query;
+        const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset, cursor, $filter, $select } = req.query;
 
         const fromDateVal = parseDateParam(fromDate, res);
         if (fromDateVal === null) return;
@@ -232,6 +237,9 @@ export function createApp(): express.Application {
           address,
           direction: "incoming",
           contractId: contractId as string | undefined,
+          filter: $filter as string | undefined,
+          select: parseSelectQuery($select),
+          cursor: cursor as string | undefined,
           fromLedger: fromLedger ? parseIntParam(fromLedger, 0) : undefined,
           toLedger: toLedger ? parseIntParam(toLedger, 0) : undefined,
           fromDate: fromDateVal,
@@ -241,7 +249,17 @@ export function createApp(): express.Application {
           offset: off,
         });
 
-        res.json({ ...result, transfers: result.transfers.map(withDisplay), limit: lim, offset: off });
+        res.json({
+          ...result,
+          transfers: result.transfers.map((transfer) => {
+            if (transfer && typeof (transfer as { amount?: unknown }).amount === "string") {
+              return withDisplay(transfer as { amount: string });
+            }
+            return transfer;
+          }),
+          limit: lim,
+          offset: off,
+        });
       } catch (err) {
         next(err);
       }
@@ -258,7 +276,7 @@ export function createApp(): express.Application {
     async (req: Request, res: Response, next: NextFunction) => {
       try {
         const { address } = req.params;
-        const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset } = req.query;
+        const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset, cursor, $filter, $select } = req.query;
 
         const fromDateVal = parseDateParam(fromDate, res);
         if (fromDateVal === null) return;
@@ -274,6 +292,9 @@ export function createApp(): express.Application {
           address,
           direction: "outgoing",
           contractId: contractId as string | undefined,
+          filter: $filter as string | undefined,
+          select: parseSelectQuery($select),
+          cursor: cursor as string | undefined,
           fromLedger: fromLedger ? parseIntParam(fromLedger, 0) : undefined,
           toLedger: toLedger ? parseIntParam(toLedger, 0) : undefined,
           fromDate: fromDateVal,
@@ -283,7 +304,17 @@ export function createApp(): express.Application {
           offset: off,
         });
 
-        res.json({ ...result, transfers: result.transfers.map(withDisplay), limit: lim, offset: off });
+        res.json({
+          ...result,
+          transfers: result.transfers.map((transfer) => {
+            if (transfer && typeof (transfer as { amount?: unknown }).amount === "string") {
+              return withDisplay(transfer as { amount: string });
+            }
+            return transfer;
+          }),
+          limit: lim,
+          offset: off,
+        });
       } catch (err) {
         next(err);
       }
@@ -311,7 +342,7 @@ export function createApp(): express.Application {
     async (req: Request, res: Response, next: NextFunction) => {
       try {
         const { address } = req.params;
-        const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset } = req.query;
+        const { contractId, fromLedger, toLedger, fromDate, toDate, eventType, limit, offset, cursor, $filter, $select } = req.query;
 
         const fromDateVal = parseDateParam(fromDate, res);
         if (fromDateVal === null) return;
@@ -326,6 +357,9 @@ export function createApp(): express.Application {
         const result = await queryAllTransfers({
           address,
           contractId: contractId as string | undefined,
+          filter: $filter as string | undefined,
+          select: parseSelectQuery($select),
+          cursor: cursor as string | undefined,
           fromLedger: fromLedger ? parseIntParam(fromLedger, 0) : undefined,
           toLedger: toLedger ? parseIntParam(toLedger, 0) : undefined,
           fromDate: fromDateVal,
@@ -335,7 +369,17 @@ export function createApp(): express.Application {
           offset: off,
         });
 
-        res.json({ ...result, transfers: result.transfers.map(withDisplay), limit: lim, offset: off });
+        res.json({
+          ...result,
+          transfers: result.transfers.map((transfer) => {
+            if (transfer && typeof (transfer as { amount?: unknown }).amount === "string") {
+              return withDisplay(transfer as { amount: string });
+            }
+            return transfer;
+          }),
+          limit: lim,
+          offset: off,
+        });
       } catch (err) {
         next(err);
       }
@@ -521,14 +565,17 @@ export function createApp(): express.Application {
     "/nfts/transfers",
     async (req: Request, res: Response, next: NextFunction) => {
       try {
-        const { contract, token_id, address, fromLedger, toLedger, limit, offset } = req.query;
+        const { contract, token_id, address, fromLedger, toLedger, limit, offset, cursor, $filter, $select } = req.query;
         const lim = parseIntParam(limit, 50);
         const off = parseIntParam(offset, 0);
 
         const result = await queryNftTransfers({
           contractId: contract as string | undefined,
           tokenId: token_id as string | undefined,
           address: address as string | undefined,
+          filter: $filter as string | undefined,
+          select: parseSelectQuery($select),
+          cursor: cursor as string | undefined,
           fromLedger: fromLedger ? parseIntParam(fromLedger, 0) : undefined,
           toLedger: toLedger ? parseIntParam(toLedger, 0) : undefined,
           limit: lim,