diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 601bf7d6..991bf3cd 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -282,6 +282,56 @@ jobs: - name: Run issuer tests (incl. cross-boundary circuit test) run: pnpm --filter @stellarcred/issuer test + indexer: + # The indexer's DB layer supports BOTH SQLite and Postgres, so the same + # test matrix must run against both backends or one silently rots. This job + # spins up a Postgres 16 service container so the parameterized DB suite + # (services/indexer/src/db.test.ts) exercises migrations, upserts, revokes, + # cursor updates, etc. on Postgres while also still running the SQLite leg. + name: Indexer tests (SQLite + Postgres) + runs-on: ubuntu-latest + defaults: + run: + working-directory: services/indexer + services: + postgres: + image: postgres:16 + env: + POSTGRES_USER: indexer + POSTGRES_PASSWORD: indexer + POSTGRES_DB: indexer + ports: + - 5432:5432 + # Gate job steps until Postgres is accepting connections. + options: >- + --health-cmd "pg_isready -U indexer -d indexer" + --health-interval 10s + --health-timeout 5s + --health-retries 5 + steps: + - uses: actions/checkout@v7 + + - uses: actions/setup-node@v7 + with: + node-version: 20 + cache: npm + cache-dependency-path: services/indexer/package-lock.json + + - name: Install dependencies + run: npm ci + + - name: Build + run: npm run build + + - name: Run DB test matrix (SQLite + Postgres) and unit tests + run: npm test + env: + # Dummy contract ID / RPC so the ingester tests run headlessly. + PROOF_REGISTRY_CONTRACT_ID: C000000000000000000000000000000000000000000000000000000000000001 + RPC_URL: https://soroban-testnet.stellar.org + # Point the Postgres leg of the DB matrix at the service container. + TEST_POSTGRES_URL: postgres://indexer:indexer@localhost:5432/indexer + sdk-integration: name: SDK integration tests (testnet) runs-on: ubuntu-latest diff --git a/services/indexer/.env.example b/services/indexer/.env.example index 67595b49..cb8d2f59 100644 --- a/services/indexer/.env.example +++ b/services/indexer/.env.example @@ -1,6 +1,13 @@ +# Backend selection: "sqlite" (default; embedded, zero-infra, single-writer) or +# "postgres" (standalone DB, required for multi-instance production deployments). +# See README "Database Backend Selection & Tradeoffs". DB_DRIVER=sqlite SQLITE_PATH=./data/indexer.db +# Required when DB_DRIVER=postgres, otherwise ignored. DATABASE_URL=postgres://indexer:indexer@localhost:5432/indexer +# Used by the DB test matrix (src/db.test.ts) to exercise the Postgres backend +# locally; falls back to DATABASE_URL. Ignored at runtime. +TEST_POSTGRES_URL=postgres://indexer:indexer@localhost:5432/indexer RPC_URL=https://soroban-testnet.stellar.org NETWORK_PASSPHRASE=Test SDF Network ; September 2015 diff --git a/services/indexer/README.md b/services/indexer/README.md index 081f3f30..0db77e51 100644 --- a/services/indexer/README.md +++ b/services/indexer/README.md @@ -133,6 +133,69 @@ the first page; a `null` `nextCursor` means there are no more claims. `limit` --- +## Database Backend Selection & Tradeoffs + +The indexer is a thin storage layer over one of two backends, selected at +startup with the `DB_DRIVER` environment variable: + +| `DB_DRIVER` | Engine | Connection config | Best for | +|---|---|---|---| +| `sqlite` (default) | better-sqlite3 (`journal_mode=WAL`) | `SQLITE_PATH` | local dev, demos, single-instance / hobby deployments | +| `postgres` | node-postgres pool | `DATABASE_URL` | production multi-instance deployments | + +**How selection works.** `loadConfig()` reads `DB_DRIVER` (defaulting to +`sqlite`) and validates it. `createDb()` then returns the matching adapter and +runs the schema migrations for that engine. `DATABASE_URL` **must** be set +when `DB_DRIVER=postgres` (and is ignored by the SQLite adapter). Everything +above the adapter — the ingester and the HTTP API — is backend-agnostic and +talks only to the `Db` interface, so adding a new backend means implementing +that interface, not touching the business logic. + +**Tradeoffs.** + +- **Operational scale** — SQLite is embedded in the process (zero + infrastructure, single file, WAL for concurrent readers) and is perfect for + local development and single-instance nodes. Postgres is a standalone + service that supports concurrent writers and many readers, which is what a + multi-instance / horizontally-scaled deployment needs. +- **Concurrency** — SQLite allows a single writer process; if you run more than + one indexer instance against the same SQLite file you can corrupt/resolve the + cursor incorrectly. Postgres serializes writes with row-level locking and a + shared cursor row. +- **Operational tooling** — Postgres gives you replication, backups, managed + hosting, and point-in-time recovery out of the box; SQLite needs your own + file-backup strategy. +- **Dependency footprint** — SQLite (via `better-sqlite3`) adds a native + module to `node_modules`; the Postgres driver (`pg`) is pure JS. Choose the + default (`sqlite`) unless you actually need Postgres's scaling and tooling. + +> **Recommendation:** run `sqlite` in development and single-instance +> production; enable `postgres` only when you need multiple reader/writer +> instances or managed database tooling. + +**Testing both backends.** The worker test suite runs the **same DB test +matrix against SQLite and Postgres** (`src/db.test.ts`). Coverage includes +schema migrations (idempotency), ledger-cursor updates, claim upserts, +revokes, `claimsByWallet`, `stats`, paginated `recent`, `deleteClaimsAfter` +and `getMaxClaimLedger`. The SQLite leg always runs locally; the Postgres leg +runs in CI (via the `postgres` service container in `.github/workflows/ci.yml`)and locally whenever `TEST_POSTGRES_URL` (or `DATABASE_URL`) points at a live +Postgres, and is skipped otherwise: + +```bash +# SQLite leg only (no Postgres reachable): +npm test + +# Both legs, against a local Postgres, e.g. `docker run ... -p 5432:5432 postgres`: +TEST_POSTGRES_URL=postgres://user:pass@localhost:5432/db npm test +``` + +Because the two engines use different SQL dialects (`INSERT OR IGNORE` vs +`ON CONFLICT`, `INTEGER` vs `BIGINT`), the matrix is exactly where silent +cross-backend divergences surface (e.g. Postgres returning `BIGINT` columns as +strings) — running it on both is how we keep either backend from rotting. + +--- + ## Consistency, Finality & Reorg Guarantees - **Cursor Progression**: The indexer stores the last successfully processed ledger sequence in database metadata. In the event of a restart, ingestion resumes seamlessly from the saved checkpoint without skipping events. @@ -147,7 +210,8 @@ the first page; a `null` `nextCursor` means there are no more claims. `limit` # Install dependencies npm install -# Run unit and integration tests +# Run unit and integration tests (SQLite by default; add TEST_POSTGRES_URL +# to also exercise the Postgres backend — see "Database Backend Selection") npm test # Build TypeScript to dist/ diff --git a/services/indexer/src/db.test.ts b/services/indexer/src/db.test.ts new file mode 100644 index 00000000..7b3a279f --- /dev/null +++ b/services/indexer/src/db.test.ts @@ -0,0 +1,259 @@ +/** + * db.test.ts — Parameterized database integration tests. + * + * Runs the *same* suite against both supported backends so neither can rot: + * + * - SQLite (better-sqlite3; the default dev / single-instance driver) + * - Postgres (pg pool; the production multi-instance driver) + * + * The two drivers use different SQL dialects (INSERT OR IGNORE vs + * ON CONFLICT, INTEGER vs BIGINT), which is exactly where they silently + * diverge — so these tests exercise migrations, upserts, revokes, and cursor + * updates against both. + * + * The Postgres leg runs in CI via the `postgres` service container in + * `.github/workflows/ci.yml`. Locally it is gated on `TEST_POSTGRES_URL` + * (falling back to `DATABASE_URL`) and is skipped when unset, mirroring how + * the SDK integration suites skip gracefully without configured credentials. + */ + +import path from "path"; +import os from "os"; +import fs from "fs"; +import { describe, it, expect, beforeEach, afterEach } from "@jest/globals"; +import { createDb, type ClaimInput, type Db } from "./db"; +import type { Config } from "./config"; + +// Connection used by the Postgres leg. In CI this is provided by the +// `postgres` service container; locally point it at any reachable Postgres. +const POSTGRES_URL = + process.env.TEST_POSTGRES_URL ?? process.env.DATABASE_URL; + +function baseConfig(overrides: Partial = {}): Config { + return { + stellarNetwork: "testnet", + horizonUrl: "https://horizon-testnet.stellar.org", + rpcUrl: "https://soroban-testnet.stellar.org", + proofRegistryContractId: "CTEST", + dbDriver: "sqlite", + sqlitePath: path.join( + os.tmpdir(), + `db-test-${process.pid}-${Date.now()}-${Math.random().toString(16).slice(2)}.db`, + ), + databaseUrl: undefined, + pollIntervalMs: 6000, + startLedger: 0, + port: 3001, + finalityLag: 6, + corsOrigins: ["http://localhost:3000"], + rateLimitWindowMs: 60_000, + rateLimitMax: 120, + rateLimitEnabled: true, + ...overrides, + } as Config; +} + +function makeClaim(overrides: Partial = {}): ClaimInput { + return { + wallet: "GALICE", + credential_type: "kyc", + issuer: "GISSUER", + verified_at: 1_724_000_000, + expiry: 1_755_000_000, + ledger_sequence: 100, + threshold: null, + revoked: 0, + ...overrides, + }; +} + +/** Register one full suite of DB tests for a given backend. */ +function registerSuite( + d: typeof describe, + driver: string, + make: () => Config, +): void { + d(`DB adapter — ${driver}`, () => { + let db: Db; + let sqliteFile: string | undefined; + + beforeEach(async () => { + const cfg = make(); + if (cfg.dbDriver === "sqlite") sqliteFile = cfg.sqlitePath; + db = createDb(cfg); + await db.migrate(); + // Isolate each test. SQLite gets a fresh temp file per test, but the + // Postgres leg reuses the same database — clear any residue so tests are + // order-independent (delete every claim and reset the cursor). + await db.deleteClaimsAfter(0); + await db.setLastLedger(0); + }); + + afterEach(async () => { + await db.close(); + if (sqliteFile) { + for (const f of [ + sqliteFile, + `${sqliteFile}-wal`, + `${sqliteFile}-shm`, + ]) { + try { + fs.unlinkSync(f); + } catch { + // already gone + } + } + } + }); + + it("migration is idempotent", async () => { + await db.migrate(); // second pass must not throw or corrupt + expect(await db.getLastLedger()).toBe(0); + }); + + it("tracks the ledger cursor (default 0, round-trips)", async () => { + expect(await db.getLastLedger()).toBe(0); + await db.setLastLedger(123_456); + expect(await db.getLastLedger()).toBe(123_456); + }); + + it("upserts a new claim and updates an existing one", async () => { + await db.upsertClaim(makeClaim()); + + let rows = await db.claimsByWallet("GALICE"); + expect(rows).toHaveLength(1); + expect(rows[0].credential_type).toBe("kyc"); + expect(rows[0].revoked).toBe(0); + + // Re-verification updates expiry/sequence and resets revoked to 0. + await db.upsertClaim( + makeClaim({ expiry: 1_999_999_999, ledger_sequence: 500 }), + ); + rows = await db.claimsByWallet("GALICE"); + expect(rows).toHaveLength(1); // still one row — upserted, not duplicated + expect(rows[0].expiry).toBe(1_999_999_999); + expect(rows[0].ledger_sequence).toBe(500); + expect(rows[0].revoked).toBe(0); + }); + + it("persists a threshold when present and nulls it when absent", async () => { + await db.upsertClaim(makeClaim({ credential_type: "income", threshold: 200_000 })); + let rows = await db.claimsByWallet("GALICE"); + expect(rows[0].threshold).toBe(200_000); + + await db.upsertClaim(makeClaim({ credential_type: "income", threshold: null })); + rows = await db.claimsByWallet("GALICE"); + expect(rows[0].threshold).toBeNull(); + }); + + it("revoke sets revoked = 1", async () => { + await db.upsertClaim(makeClaim()); + await db.revokeClaim("GALICE", "kyc"); + + const rows = await db.claimsByWallet("GALICE"); + expect(rows).toHaveLength(1); + expect(rows[0].revoked).toBe(1); + }); + + it("claimsByWallet returns only that wallet's claims", async () => { + await db.upsertClaim(makeClaim({ wallet: "GALICE" })); + await db.upsertClaim(makeClaim({ wallet: "GBOB", credential_type: "age" })); + + const galice = await db.claimsByWallet("GALICE"); + expect(galice).toHaveLength(1); + expect(galice[0].credential_type).toBe("kyc"); + }); + + it("stats aggregates total/active/revoked per type", async () => { + await db.upsertClaim(makeClaim({ credential_type: "kyc" })); + await db.upsertClaim(makeClaim({ credential_type: "kyc", wallet: "GBOB" })); + await db.upsertClaim(makeClaim({ credential_type: "age", wallet: "GCAR" })); + await db.revokeClaim("GALICE", "kyc"); + + const stats = await db.stats(); + const kyc = stats.find((s) => s.credential_type === "kyc"); + expect(kyc).toEqual({ credential_type: "kyc", total: 2, active: 1, revoked: 1 }); + expect(stats.find((s) => s.credential_type === "age")?.total).toBe(1); + }); + + it("recent returns non-revoked claims newest-first with pagination", async () => { + await db.upsertClaim( + makeClaim({ credential_type: "kyc", ledger_sequence: 100, wallet: "GA" }), + ); + await db.upsertClaim( + makeClaim({ credential_type: "age", ledger_sequence: 300, wallet: "GB" }), + ); + await db.upsertClaim( + makeClaim({ credential_type: "income", ledger_sequence: 200, wallet: "GC" }), + ); + await db.upsertClaim( + makeClaim({ credential_type: "funds", ledger_sequence: 150, wallet: "GD" }), + ); + // Revoked claims never appear in recent. + await db.revokeClaim("GA", "kyc"); + + // First page: newest (highest ledger) first, revoked claim excluded. + const firstPage = await db.recent(2, null); + expect(firstPage.claims.map((r) => r.credential_type)).toEqual([ + "age", + "income", + ]); + expect(firstPage.nextCursor).not.toBeNull(); + + // Second page via the keyset cursor: the revoked kyc is still excluded, + // so only funds remains and the cursor is exhausted. + const secondPage = await db.recent(2, firstPage.nextCursor); + expect(secondPage.claims.map((r) => r.credential_type)).toEqual(["funds"]); + expect(secondPage.nextCursor).toBeNull(); + }); + + it("deleteClaimsAfter rolls back un-final claims", async () => { + await db.upsertClaim(makeClaim({ ledger_sequence: 100, wallet: "GA" })); + await db.upsertClaim(makeClaim({ ledger_sequence: 200, wallet: "GB" })); + await db.upsertClaim(makeClaim({ ledger_sequence: 160, wallet: "GC" })); + + await db.deleteClaimsAfter(150); + const ga = await db.claimsByWallet("GA"); + const gc = await db.claimsByWallet("GC"); + expect(ga[0].ledger_sequence).toBe(100); // kept (≤ fromLedger) + expect(gc).toHaveLength(0); // dropped (> fromLedger) + expect(await db.getMaxClaimLedger()).toBe(100); + }); + + it("getMaxClaimLedger returns 0 when empty and the max otherwise", async () => { + expect(await db.getMaxClaimLedger()).toBe(0); + await db.upsertClaim(makeClaim({ ledger_sequence: 1000 })); + await db.upsertClaim( + makeClaim({ ledger_sequence: 2000, wallet: "GBOB" }), + ); + expect(await db.getMaxClaimLedger()).toBe(2000); + }); + + it("re-verification after revocation clears the revoked flag", async () => { + await db.upsertClaim(makeClaim()); + await db.revokeClaim("GALICE", "kyc"); + expect((await db.claimsByWallet("GALICE"))[0].revoked).toBe(1); + + // A fresh verified event for the same (wallet, type) upserts revoked back to 0. + await db.upsertClaim(makeClaim({ ledger_sequence: 999 })); + expect((await db.claimsByWallet("GALICE"))[0].revoked).toBe(0); + }); + }); +} + +registerSuite(describe, "sqlite", () => baseConfig({ dbDriver: "sqlite" })); + +if (POSTGRES_URL) { + registerSuite(describe, "postgres", () => + baseConfig({ dbDriver: "postgres", databaseUrl: POSTGRES_URL }), + ); +} else { + describe.skip( + "DB adapter — postgres", + () => { + it("is skipped when TEST_POSTGRES_URL / DATABASE_URL is not set", () => { + expect(POSTGRES_URL).toBeUndefined(); + }); + }, + ); +} \ No newline at end of file diff --git a/services/indexer/src/db.ts b/services/indexer/src/db.ts index 5529469c..d8269b08 100644 --- a/services/indexer/src/db.ts +++ b/services/indexer/src/db.ts @@ -406,7 +406,17 @@ export function createPostgresDb(config: Config): Db { } // eslint-disable-next-line @typescript-eslint/no-require-imports - const { Pool } = require("pg") as typeof import("pg"); + const pg = require("pg") as typeof import("pg"); + const { Pool } = pg; + + // pg returns INT8/BIGINT and INT4/INTEGER columns as strings by default, + // which would make ClaimRow's numeric fields (verified_at, expiry, + // ledger_sequence, threshold, revoked) come back as strings on Postgres but + // numbers on SQLite. Force them to JS numbers so both backends expose + // identical row shapes. + pg.types.setTypeParser(20, Number); // INT8 / BIGINT + pg.types.setTypeParser(23, Number); // INT4 / INTEGER + const pool = new Pool({ connectionString: config.databaseUrl }); return { diff --git a/services/indexer/src/ingester.test.ts b/services/indexer/src/ingester.test.ts index f0404dec..e27a9e8b 100644 --- a/services/indexer/src/ingester.test.ts +++ b/services/indexer/src/ingester.test.ts @@ -40,7 +40,7 @@ function makeConfig(overrides: Partial = {}): Config { rateLimitMax: 120, rateLimitEnabled: true, ...overrides, - }; + } as Config; } /** diff --git a/services/indexer/src/ingester.ts b/services/indexer/src/ingester.ts index 62a62046..993d8f71 100644 --- a/services/indexer/src/ingester.ts +++ b/services/indexer/src/ingester.ts @@ -194,7 +194,10 @@ function decodeScVal(b64: string): unknown { } return native; } catch { - return null; + // Not valid base64 XDR — treat the raw value as a literal string so that + // already-decoded / plain-string topics (e.g. "proof", "verified") still + // match parseEvent's topic comparisons instead of silently becoming null. + return b64; } } @@ -599,6 +602,11 @@ export function createIngester(config: Config, db: Db): Ingester { try { events = await fetchEvents(cursor, finalityCeiling); } catch (err) { + // Record the error but do NOT advance the cursor — we'll retry next tick. + health.lastError = (err as Error).message; + health.lastErrorTime = Date.now(); + health.consecutiveErrors++; + health.fetchFailures++; console.warn("[indexer] Horizon fetch error:", (err as Error).message); fetchErrorsTotal++; return 0;