From e46334bd573bceb32aa4ac03d7b50b68631fc30c Mon Sep 17 00:00:00 2001 From: Salmatcre8 Date: Sun, 30 Aug 2026 08:46:05 +0100 Subject: [PATCH 1/2] feat(db): add network column across all Prisma models (#159) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Wraith's schema had no network dimension, so a mainnet instance and a testnet instance could not share a database. Three collisions made it impossible rather than merely untidy: - eventId was globally @unique on TokenTransfer/HostFnLog/NftTransfer. It is an RPC paging token, unique only within a network, so the same token on both chains silently overwrote one network's row. - IndexerState and BackfillCursor were singleton rows (@id @default(1)). One cursor cannot describe how far two independent chains have been indexed. - AccountSummary and NftMetadata keyed on (address, contractId) and (contractId, tokenId), merging two chains into one aggregate. Every model now carries `network` ("testnet" | "mainnet"), folded into every uniqueness constraint and every query. IndexerState and BackfillCursor are keyed by network instead of the singleton id. network also leads every composite index. Each query filters on it now, and a trailing position would leave Postgres unable to use the index for that filter. src/network.ts is the single place that decides what "the current network" means. Every db.ts function takes an optional trailing network defaulting to STELLAR_NETWORK, so existing single-network callers are unchanged and #161/#163 have somewhere explicit to pass. Two things the type checker could not have caught: - commitBatch wrote transfers, NFT transfers and host-fn logs without a network. The column has a DEFAULT of 'testnet', so that compiles and files mainnet events under testnet. Now stamped explicitly. - updateAccountSummaries in checkpoint.ts was a byte-identical copy of upsertAccountSummaries in db.ts, including the raw ON CONFLICT target. That target must name the same columns as the unique index, which widened here, so the copy would have thrown on every write. It now delegates to the single implementation. The migration is hand-written. `prisma migrate diff` emits ADD COLUMN "network" TEXT NOT NULL with no default for IndexerState and BackfillCursor, because the new column is their primary key. Postgres rejects that on a non-empty table, verified: ERROR: column "network" of relation "pkproof" contains null values Both are therefore added WITH a default, populated, then the default is dropped before the key is applied, which preserves the indexer cursor instead of forcing a re-index from genesis. Verified against a Postgres 15 instance: pushed the pre-#159 schema, seeded IndexerState (lastIndexedLedger 987654) and BackfillCursor, applied the migration, and confirmed both survived as network='testnet'. `prisma migrate diff --from-url` against the result reports an empty migration, so the SQL lands exactly on the target schema with no drift. Tests: 270 pass (was 244). The 26 new ones assert the predicate is present in each where clause rather than that queries merely succeed — deleting one network filter from queryTransfers fails three of them. Claude-Session: https://claude.ai/code/session_01USgemLt4Rnz4SGB1Srf3GB --- docs/DUAL_NETWORK.md | 13 +- .../20260829120000_add_network/migration.sql | 108 ++++++ prisma/schema.prisma | 178 ++++++---- src/__tests__/network.test.ts | 315 ++++++++++++++++++ src/__tests__/routes/transfers.test.ts | 1 + src/db.ts | 190 ++++++++--- src/indexer/checkpoint.ts | 134 +++----- src/network.ts | 55 +++ 8 files changed, 777 insertions(+), 217 deletions(-) create mode 100644 prisma/migrations/20260829120000_add_network/migration.sql create mode 100644 src/__tests__/network.test.ts create mode 100644 src/network.ts diff --git a/docs/DUAL_NETWORK.md b/docs/DUAL_NETWORK.md index 5df9813c..70f9e305 100644 --- a/docs/DUAL_NETWORK.md +++ b/docs/DUAL_NETWORK.md @@ -11,9 +11,14 @@ the safe demo/QA surface while mainnet handles real value. issues below (network-aware storage, per-network RPC, per-network loops, a network selector on the API). -Either way the storage must be network-segregated: today the schema has **no -`network` column**, so two networks in one DB collide (`eventId` is globally -unique; `IndexerState`/`BackfillCursor` are singleton rows). +Either way the storage must be network-segregated. As of #159 it is: every +model carries a `network` column, `eventId` is unique per `(network, eventId)` +rather than globally, and `IndexerState`/`BackfillCursor` hold one row per +network instead of the old singleton `id = 1`. + +Every `db.ts` function takes an optional trailing `network`, defaulting to +`STELLAR_NETWORK` via `src/network.ts`. Single-network deployments therefore +behave exactly as before; #161 and #163 pass it explicitly. ## Env matrix @@ -34,7 +39,7 @@ Dependencies: **#159 → #161** and #160 before #161. | # | Issue | Dep | |---|-------|-----| -| [#159](../../issues/159) | `network` column across all Prisma models | — | +| ~~[#159](../../issues/159)~~ | ~~`network` column across all Prisma models~~ (done) | — | | [#160](../../issues/160) | Per-network `getRpc(network)` factory | — | | [#161](../../issues/161) | One indexer loop per network | #159, #160 | | [#162](../../issues/162) | Per-network SAC/NFT watch-lists | — | diff --git a/prisma/migrations/20260829120000_add_network/migration.sql b/prisma/migrations/20260829120000_add_network/migration.sql new file mode 100644 index 00000000..8c83adc3 --- /dev/null +++ b/prisma/migrations/20260829120000_add_network/migration.sql @@ -0,0 +1,108 @@ +-- Add a `network` dimension to every table so one database can hold both +-- testnet and mainnet without collisions (#159). +-- +-- Existing rows are all testnet, so DEFAULT 'testnet' back-fills them +-- correctly rather than guessing. +-- +-- NOTE ON IndexerState / BackfillCursor: `prisma migrate diff` generates +-- `ADD COLUMN "network" TEXT NOT NULL` for these two (no default, because the +-- new column is the primary key). That statement fails on a non-empty table — +-- Postgres cannot add a NOT NULL column with no default to existing rows. Both +-- are therefore hand-written below: add WITH a default, let the existing row +-- inherit it, then drop the default so future inserts must be explicit. This +-- preserves the indexer cursor; regenerating this migration with +-- `prisma migrate dev` would silently reintroduce the broken form and force a +-- full re-index from genesis. + +-- ─── Drop old indexes (superseded by network-leading equivalents) ──────────── +DROP INDEX "wraith"."TokenTransfer_eventId_key"; +DROP INDEX "wraith"."TokenTransfer_toAddress_idx"; +DROP INDEX "wraith"."TokenTransfer_fromAddress_idx"; +DROP INDEX "wraith"."TokenTransfer_contractId_idx"; +DROP INDEX "wraith"."TokenTransfer_ledger_idx"; +DROP INDEX "wraith"."TokenTransfer_txHash_idx"; +DROP INDEX "wraith"."TokenTransfer_toAddress_contractId_idx"; +DROP INDEX "wraith"."TokenTransfer_fromAddress_contractId_idx"; +DROP INDEX "wraith"."HostFnLog_eventId_key"; +DROP INDEX "wraith"."HostFnLog_contractId_idx"; +DROP INDEX "wraith"."HostFnLog_contractId_functionName_idx"; +DROP INDEX "wraith"."HostFnLog_ledger_idx"; +DROP INDEX "wraith"."HostFnLog_txHash_idx"; +DROP INDEX "wraith"."NftTransfer_eventId_key"; +DROP INDEX "wraith"."NftTransfer_contractId_idx"; +DROP INDEX "wraith"."NftTransfer_tokenId_idx"; +DROP INDEX "wraith"."NftTransfer_toAddress_idx"; +DROP INDEX "wraith"."NftTransfer_fromAddress_idx"; +DROP INDEX "wraith"."NftTransfer_contractId_tokenId_idx"; +DROP INDEX "wraith"."NftMetadata_contractId_tokenId_key"; +DROP INDEX "wraith"."AccountSummary_address_idx"; +DROP INDEX "wraith"."AccountSummary_lastActivityAt_idx"; +DROP INDEX "wraith"."AccountSummary_address_contractId_key"; +DROP INDEX "wraith"."WebhookSubscription_active_idx"; +DROP INDEX "wraith"."WebhookDelivery_eventId_idx"; +DROP INDEX "wraith"."IndexerCheckpoint_batchId_key"; + +-- ─── Add the network column ────────────────────────────────────────────────── +ALTER TABLE "wraith"."TokenTransfer" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; +ALTER TABLE "wraith"."HostFnLog" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; +ALTER TABLE "wraith"."NftTransfer" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; +ALTER TABLE "wraith"."NftMetadata" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; +ALTER TABLE "wraith"."AccountSummary" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; +ALTER TABLE "wraith"."WebhookSubscription" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; +ALTER TABLE "wraith"."WebhookDelivery" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; +ALTER TABLE "wraith"."IndexerCheckpoint" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; +ALTER TABLE "wraith"."RetentionJobRun" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; + +-- ─── IndexerState: singleton (id = 1) → one row per network ────────────────── +-- Ordering matters: the column must exist and be populated before it can carry +-- a primary key, and `id` must go before the new key is added. +ALTER TABLE "wraith"."IndexerState" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; +ALTER TABLE "wraith"."IndexerState" DROP CONSTRAINT "IndexerState_pkey"; +ALTER TABLE "wraith"."IndexerState" DROP COLUMN "id"; +ALTER TABLE "wraith"."IndexerState" ALTER COLUMN "network" DROP DEFAULT; +-- The table held at most one row (id defaulted to 1 and every caller wrote +-- `where: { id: 1 }`), so this cannot collide. If it somehow does, the +-- migration fails loudly here rather than silently discarding a cursor. +ALTER TABLE "wraith"."IndexerState" ADD CONSTRAINT "IndexerState_pkey" PRIMARY KEY ("network"); + +-- ─── BackfillCursor: same singleton → per-network conversion ───────────────── +ALTER TABLE "wraith"."BackfillCursor" ADD COLUMN "network" TEXT NOT NULL DEFAULT 'testnet'; +ALTER TABLE "wraith"."BackfillCursor" DROP CONSTRAINT "BackfillCursor_pkey"; +ALTER TABLE "wraith"."BackfillCursor" DROP COLUMN "id"; +ALTER TABLE "wraith"."BackfillCursor" ALTER COLUMN "network" DROP DEFAULT; +ALTER TABLE "wraith"."BackfillCursor" ADD CONSTRAINT "BackfillCursor_pkey" PRIMARY KEY ("network"); + +-- ─── Recreate indexes with network leading ─────────────────────────────────── +CREATE INDEX "TokenTransfer_network_toAddress_idx" ON "wraith"."TokenTransfer"("network", "toAddress"); +CREATE INDEX "TokenTransfer_network_fromAddress_idx" ON "wraith"."TokenTransfer"("network", "fromAddress"); +CREATE INDEX "TokenTransfer_network_contractId_idx" ON "wraith"."TokenTransfer"("network", "contractId"); +CREATE INDEX "TokenTransfer_network_ledger_idx" ON "wraith"."TokenTransfer"("network", "ledger"); +CREATE INDEX "TokenTransfer_network_txHash_idx" ON "wraith"."TokenTransfer"("network", "txHash"); +CREATE INDEX "TokenTransfer_network_toAddress_contractId_idx" ON "wraith"."TokenTransfer"("network", "toAddress", "contractId"); +CREATE INDEX "TokenTransfer_network_fromAddress_contractId_idx" ON "wraith"."TokenTransfer"("network", "fromAddress", "contractId"); +CREATE UNIQUE INDEX "TokenTransfer_network_eventId_key" ON "wraith"."TokenTransfer"("network", "eventId"); + +CREATE INDEX "HostFnLog_network_contractId_idx" ON "wraith"."HostFnLog"("network", "contractId"); +CREATE INDEX "HostFnLog_network_contractId_functionName_idx" ON "wraith"."HostFnLog"("network", "contractId", "functionName"); +CREATE INDEX "HostFnLog_network_ledger_idx" ON "wraith"."HostFnLog"("network", "ledger"); +CREATE INDEX "HostFnLog_network_txHash_idx" ON "wraith"."HostFnLog"("network", "txHash"); +CREATE UNIQUE INDEX "HostFnLog_network_eventId_key" ON "wraith"."HostFnLog"("network", "eventId"); + +CREATE INDEX "NftTransfer_network_contractId_idx" ON "wraith"."NftTransfer"("network", "contractId"); +CREATE INDEX "NftTransfer_network_tokenId_idx" ON "wraith"."NftTransfer"("network", "tokenId"); +CREATE INDEX "NftTransfer_network_toAddress_idx" ON "wraith"."NftTransfer"("network", "toAddress"); +CREATE INDEX "NftTransfer_network_fromAddress_idx" ON "wraith"."NftTransfer"("network", "fromAddress"); +CREATE INDEX "NftTransfer_network_contractId_tokenId_idx" ON "wraith"."NftTransfer"("network", "contractId", "tokenId"); +CREATE UNIQUE INDEX "NftTransfer_network_eventId_key" ON "wraith"."NftTransfer"("network", "eventId"); + +CREATE UNIQUE INDEX "NftMetadata_network_contractId_tokenId_key" ON "wraith"."NftMetadata"("network", "contractId", "tokenId"); + +CREATE INDEX "AccountSummary_network_address_idx" ON "wraith"."AccountSummary"("network", "address"); +CREATE INDEX "AccountSummary_network_lastActivityAt_idx" ON "wraith"."AccountSummary"("network", "lastActivityAt"); +CREATE UNIQUE INDEX "AccountSummary_network_address_contractId_key" ON "wraith"."AccountSummary"("network", "address", "contractId"); + +CREATE INDEX "WebhookSubscription_network_active_idx" ON "wraith"."WebhookSubscription"("network", "active"); +CREATE INDEX "WebhookDelivery_network_eventId_idx" ON "wraith"."WebhookDelivery"("network", "eventId"); + +CREATE UNIQUE INDEX "IndexerCheckpoint_network_batchId_key" ON "wraith"."IndexerCheckpoint"("network", "batchId"); +CREATE INDEX "RetentionJobRun_network_startedAt_idx" ON "wraith"."RetentionJobRun"("network", "startedAt"); diff --git a/prisma/schema.prisma b/prisma/schema.prisma index de8a5d67..b728dcd4 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -14,9 +14,28 @@ datasource db { schemas = ["wraith"] } +// ─── Network dimension ─────────────────────────────────────────────────────── +// Every table carries `network` ("testnet" | "mainnet") so one database can +// hold both without collisions. Two properties make this mandatory rather than +// cosmetic: +// +// 1. `eventId` is an RPC paging token. Those are only unique *within* a +// network, so the same token can legitimately appear on both chains — a +// global @unique silently overwrites one network row with the other. +// 2. IndexerState and BackfillCursor used to be singleton rows (id = 1). +// One cursor cannot track two chains, so both are now keyed by network. +// +// The default is "testnet": every row that exists today was indexed from +// testnet, so back-filling the column with that value is correct, not a guess. +// +// `network` also leads every composite index. Every query filters on it now, +// and a trailing position would leave Postgres unable to use the index for +// that filter — the column would be recorded but not exploited. + // ─── Token Transfers ────────────────────────────────────────────────────────── model TokenTransfer { - id Int @id @default(autoincrement()) + id Int @id @default(autoincrement()) + network String @default("testnet") contractId String eventType String fromAddress String? @@ -25,19 +44,20 @@ model TokenTransfer { ledger Int ledgerClosedAt DateTime txHash String - eventId String @unique + eventId String // True when the emitting contract is a Stellar Asset Contract (SAC) wrapping a // classic asset, as opposed to a native Soroban token. Set by sac-detect (#136). - isSac Boolean @default(false) - createdAt DateTime @default(now()) - - @@index([toAddress]) - @@index([fromAddress]) - @@index([contractId]) - @@index([ledger]) - @@index([txHash]) - @@index([toAddress, contractId]) - @@index([fromAddress, contractId]) + isSac Boolean @default(false) + createdAt DateTime @default(now()) + + @@unique([network, eventId]) + @@index([network, toAddress]) + @@index([network, fromAddress]) + @@index([network, contractId]) + @@index([network, ledger]) + @@index([network, txHash]) + @@index([network, toAddress, contractId]) + @@index([network, fromAddress, contractId]) @@schema("wraith") } @@ -45,49 +65,54 @@ model TokenTransfer { // One row per contract event — includes token events and all other contracts. // Allows downstream consumers to interpret events from arbitrary contracts. model HostFnLog { - id Int @id @default(autoincrement()) + id Int @id @default(autoincrement()) + + // Which chain this event came from. + network String @default("testnet") // The contract that emitted the event (C...) - contractId String + contractId String // topics[0] decoded as a symbol string (e.g. "transfer", "swap", "deposit") functionName String // topics[1..n] serialised via scValToNative — BigInt values become strings - args Json + args Json // Event value serialised via scValToNative; null if the value is scvVoid - result Json? + result Json? // Gas consumed by the invocation — populated externally when tx metadata is // available; null otherwise - gasUsed BigInt? + gasUsed BigInt? // Ledger sequence number where the event was emitted - ledger Int + ledger Int // UTC close time of the ledger ledgerClosedAt DateTime // Transaction hash (SHA-256 hex, no 0x prefix) - txHash String - - // Stellar RPC paging token — unique per event, used for deduplication - eventId String @unique + txHash String - createdAt DateTime @default(now()) + // Stellar RPC paging token — unique per event *within a network*, used for + // deduplication. See the network note at the top of this file. + eventId String - @@index([contractId]) - @@index([contractId, functionName]) - @@index([ledger]) - @@index([txHash]) + createdAt DateTime @default(now()) + @@unique([network, eventId]) + @@index([network, contractId]) + @@index([network, contractId, functionName]) + @@index([network, ledger]) + @@index([network, txHash]) @@schema("wraith") } // ─── NFT Transfers ──────────────────────────────────────────────────────────── model NftTransfer { id Int @id @default(autoincrement()) + network String @default("testnet") contractId String tokenId String fromAddress String? @@ -95,33 +120,36 @@ model NftTransfer { ledger Int ledgerClosedAt DateTime txHash String - eventId String @unique + eventId String createdAt DateTime @default(now()) - @@index([contractId]) - @@index([tokenId]) - @@index([toAddress]) - @@index([fromAddress]) - @@index([contractId, tokenId]) + @@unique([network, eventId]) + @@index([network, contractId]) + @@index([network, tokenId]) + @@index([network, toAddress]) + @@index([network, fromAddress]) + @@index([network, contractId, tokenId]) @@schema("wraith") } // ─── NFT Metadata Cache ─────────────────────────────────────────────────────── model NftMetadata { id Int @id @default(autoincrement()) + network String @default("testnet") contractId String tokenId String name String? tokenUri String? fetchedAt DateTime @default(now()) - @@unique([contractId, tokenId]) + @@unique([network, contractId, tokenId]) @@schema("wraith") } // ─── Account Summaries ──────────────────────────────────────────────────────── model AccountSummary { id Int @id @default(autoincrement()) + network String @default("testnet") address String contractId String totalSent String @default("0") @@ -131,78 +159,93 @@ model AccountSummary { lastActivityAt DateTime updatedAt DateTime @updatedAt - @@unique([address, contractId]) - @@index([address]) - @@index([lastActivityAt]) + @@unique([network, address, contractId]) + @@index([network, address]) + @@index([network, lastActivityAt]) @@schema("wraith") } // ─── Webhook Subscriptions ──────────────────────────────────────────────────── // Each row is one subscriber endpoint. The secret is used to HMAC-sign payloads. +// A subscription is scoped to one network: a testnet subscriber must not be +// woken by mainnet value moving, and vice versa. model WebhookSubscription { - id Int @id @default(autoincrement()) + id Int @id @default(autoincrement()) + network String @default("testnet") url String secret String filter Json? - active Boolean @default(true) - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + active Boolean @default(true) + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt deliveries WebhookDelivery[] - @@index([active]) + @@index([network, active]) @@schema("wraith") } // ─── Webhook Deliveries ─────────────────────────────────────────────────────── model WebhookDelivery { - id Int @id @default(autoincrement()) + id Int @id @default(autoincrement()) + // Denormalised from the subscription so a delivery can be looked up by + // (network, eventId) without a join. + network String @default("testnet") subscriptionId Int - subscription WebhookSubscription @relation(fields: [subscriptionId], references: [id], onDelete: Cascade) + subscription WebhookSubscription @relation(fields: [subscriptionId], references: [id], onDelete: Cascade) eventId String payload Json - status String @default("pending") - attempts Int @default(0) + status String @default("pending") + attempts Int @default(0) nextRetryAt DateTime? lastStatusCode Int? lastError String? deliveredAt DateTime? - createdAt DateTime @default(now()) - updatedAt DateTime @updatedAt + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt @@index([subscriptionId]) + // The retry sweeper drains every network queue in one pass, so this index + // deliberately stays network-agnostic. @@index([status, nextRetryAt]) - @@index([eventId]) + @@index([network, eventId]) @@schema("wraith") } // ─── Indexer State ──────────────────────────────────────────────────────────── +// One row per network. This was a singleton (id = 1) — a single cursor cannot +// describe how far two independent chains have been indexed. model IndexerState { - id Int @id @default(1) - lastIndexedLedger Int - updatedAt DateTime @updatedAt + network String @id + lastIndexedLedger Int + updatedAt DateTime @updatedAt @@schema("wraith") } // ─── Indexer Checkpoint ────────────────────────────────────────────────────── // Tracks the cursor position for exactly-once idempotent batch processing. -// Each batch is keyed by batchId (e.g., "sac:6000-7000") to enable parallel -// workers to checkpoint independently. Atomic upsert of events + cursor ensures -// a crash mid-batch either completes or rolls back, never leaving a gap. +// Each batch is keyed by (network, batchId) (e.g., "sac:6000-7000") to enable +// parallel workers to checkpoint independently. Atomic upsert of events + +// cursor ensures a crash mid-batch either completes or rolls back, never +// leaving a gap. Batch ids are only unique within a network — the same ledger +// range exists on both chains. model IndexerCheckpoint { - id Int @id @default(autoincrement()) - batchId String @unique - lastLedger Int - processedAt DateTime @default(now()) - updatedAt DateTime @updatedAt - + id Int @id @default(autoincrement()) + network String @default("testnet") + batchId String + lastLedger Int + processedAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@unique([network, batchId]) @@schema("wraith") } // ─── Partition Retention Runs ─────────────────────────────────────────────── model RetentionJobRun { id Int @id @default(autoincrement()) + network String @default("testnet") startedAt DateTime @default(now()) finishedAt DateTime @updatedAt compressed Int @default(0) @@ -211,18 +254,19 @@ model RetentionJobRun { skipped Int @default(0) status String @default("completed") + @@index([network, startedAt]) @@schema("wraith") } // ─── Backfill Cursor ─────────────────────────────────────────────────────────── // Durable cursor so the backfill job can resume mid-range after a crash. -// Always row ID 1 — singleton, one backfill at a time. +// One row per network — one backfill at a time *per chain*, not globally. model BackfillCursor { - id Int @id @default(1) - startLedger Int - endLedger Int - nextLedger Int - updatedAt DateTime @updatedAt + network String @id + startLedger Int + endLedger Int + nextLedger Int + updatedAt DateTime @updatedAt @@schema("wraith") } diff --git a/src/__tests__/network.test.ts b/src/__tests__/network.test.ts new file mode 100644 index 00000000..d08d8f8e --- /dev/null +++ b/src/__tests__/network.test.ts @@ -0,0 +1,315 @@ +/** + * Network-scoping tests for #159. + * + * The failure this guards against is quiet: `network` carries a DEFAULT of + * 'testnet', so a query that forgets to filter on it — or a write that forgets + * to set it — still compiles, still passes a type check, and still returns + * rows. It just returns (or writes) the wrong chain's rows. Nothing surfaces + * until mainnet and testnet share a database, at which point balances are + * wrong rather than missing. + * + * So these assert the predicate is actually present in each where clause, + * rather than asserting that the queries merely succeed. + * + * db.ts resolves its client from `globalThis.prisma` before constructing one, + * which lets us inject a recording stub and keep this a unit test with no + * database. + */ + +type AnyRecord = Record; + +/** Records every call so a test can inspect the `where` that was built. */ +function recorder() { + const calls: Array<{ model: string; op: string; args: AnyRecord }> = []; + + const model = (name: string, results: AnyRecord = {}) => + new Proxy( + {}, + { + get: (_t, op: string) => (args: AnyRecord) => { + calls.push({ model: name, op, args: args ?? {} }); + if (op === "count") return Promise.resolve(0); + if (op === "findMany") return Promise.resolve([]); + if (op === "deleteMany") return Promise.resolve({ count: 0 }); + if (op === "createMany") return Promise.resolve({ count: 0 }); + if (op === "findUnique" || op === "findFirst") { + return Promise.resolve(results[op] ?? null); + } + return Promise.resolve(results[op] ?? null); + }, + } + ) as AnyRecord; + + const raw: Array<{ strings: string[]; values: unknown[] }> = []; + const rawFn = (strings: TemplateStringsArray | string[], ...values: unknown[]) => { + raw.push({ strings: Array.from(strings as string[]), values }); + return Promise.resolve([]); + }; + + const stub: AnyRecord = { + tokenTransfer: model("tokenTransfer"), + nftTransfer: model("nftTransfer"), + nftMetadata: model("nftMetadata"), + hostFnLog: model("hostFnLog"), + accountSummary: model("accountSummary"), + indexerState: model("indexerState", { findUnique: { lastIndexedLedger: 42 } }), + backfillCursor: model("backfillCursor"), + indexerCheckpoint: model("indexerCheckpoint"), + $transaction: (ops: unknown) => + Array.isArray(ops) ? Promise.all(ops) : (ops as (tx: AnyRecord) => unknown)(stub), + $queryRaw: rawFn, + $executeRaw: rawFn, + }; + + return { stub, calls, raw }; +} + +const { stub, calls, raw } = recorder(); +(globalThis as AnyRecord).prisma = stub; + +// Imported after the stub is installed — db.ts binds its client at module load. +import * as db from "../db"; +import { currentNetwork, isNetwork, parseNetwork, resolveNetwork } from "../network"; + +/** The `where` of the last recorded call against `model`. */ +function lastWhere(model: string, op?: string): AnyRecord { + const match = [...calls].reverse().find((c) => c.model === model && (!op || c.op === op)); + if (!match) throw new Error(`no recorded ${op ?? "any"} call on ${model}`); + return match.args.where ?? {}; +} + +/** Every value passed into a raw SQL template, flattened across Prisma.Sql. */ +function rawValues(): unknown[] { + return raw.flatMap((r) => + r.values.flatMap((v) => + v && typeof v === "object" && Array.isArray((v as AnyRecord).values) + ? (v as AnyRecord).values + : [v] + ) + ); +} + +const ADDR = "GDWCO35QUYQLGO6P7OLW4BZWNMMGGUWNPLRVPLCBVG7YNVDZKUDIW4KN"; +const CONTRACT = "CBC42KFZO33TYVFDOUXFRWXYYXHFGH7W5GM4IJQSXKGFINKL2XPP4XTE"; + +beforeEach(() => { + calls.length = 0; + raw.length = 0; + delete process.env.STELLAR_NETWORK; +}); + +describe("network resolution", () => { + it("defaults to testnet when STELLAR_NETWORK is unset", () => { + expect(currentNetwork()).toBe("testnet"); + }); + + it("reads STELLAR_NETWORK, case-insensitively and trimmed", () => { + process.env.STELLAR_NETWORK = " MAINNET "; + expect(currentNetwork()).toBe("mainnet"); + }); + + it("falls back to testnet for an unrecognised value rather than passing it through", () => { + // A typo like STELLAR_NETWORK=main must not become a third network that + // matches no rows and no index. + process.env.STELLAR_NETWORK = "main"; + expect(currentNetwork()).toBe("testnet"); + expect(parseNetwork("main")).toBeNull(); + }); + + it("re-reads the environment on every call", () => { + process.env.STELLAR_NETWORK = "mainnet"; + expect(currentNetwork()).toBe("mainnet"); + process.env.STELLAR_NETWORK = "testnet"; + expect(currentNetwork()).toBe("testnet"); + }); + + it("lets an explicit argument win over the environment", () => { + process.env.STELLAR_NETWORK = "testnet"; + expect(resolveNetwork("mainnet")).toBe("mainnet"); + expect(resolveNetwork(undefined)).toBe("testnet"); + }); + + it("narrows only the two real networks", () => { + expect(isNetwork("mainnet")).toBe(true); + expect(isNetwork("futurenet")).toBe(false); + expect(isNetwork(undefined)).toBe(false); + }); +}); + +describe("reads are network-scoped", () => { + it("queryTransfers filters on the configured network", async () => { + process.env.STELLAR_NETWORK = "mainnet"; + await db.queryTransfers({ address: ADDR, direction: "incoming" }); + expect(lastWhere("tokenTransfer", "findMany").network).toBe("mainnet"); + }); + + it("queryTransfers honours an explicit network over the environment", async () => { + process.env.STELLAR_NETWORK = "mainnet"; + await db.queryTransfers({ address: ADDR, direction: "incoming", network: "testnet" }); + expect(lastWhere("tokenTransfer", "findMany").network).toBe("testnet"); + }); + + it("counts and rows agree on the network, so total matches the page", async () => { + process.env.STELLAR_NETWORK = "mainnet"; + await db.queryTransfers({ address: ADDR, direction: "incoming" }); + expect(lastWhere("tokenTransfer", "count").network).toBe("mainnet"); + expect(lastWhere("tokenTransfer", "findMany").network).toBe("mainnet"); + }); + + it("queryAllTransfers scopes alongside its OR on address", async () => { + await db.queryAllTransfers({ address: ADDR, network: "mainnet" }); + const where = lastWhere("tokenTransfer", "findMany"); + expect(where.network).toBe("mainnet"); + // The address OR must stay a sibling of the network filter, not replace it. + expect(where.OR).toHaveLength(2); + }); + + it("queryNftTransfers scopes", async () => { + await db.queryNftTransfers({ contractId: CONTRACT, network: "mainnet" }); + expect(lastWhere("nftTransfer", "findMany").network).toBe("mainnet"); + }); + + it("queryAccountSummaries scopes", async () => { + await db.queryAccountSummaries({ address: ADDR, network: "mainnet" }); + expect(lastWhere("accountSummary", "findMany").network).toBe("mainnet"); + }); + + it("getAccountSummary scopes", async () => { + await db.getAccountSummary(ADDR, undefined, "mainnet"); + expect(lastWhere("accountSummary", "findMany").network).toBe("mainnet"); + }); + + it("queryByTxHash scopes — a tx hash alone is not unique across chains", async () => { + await db.queryByTxHash("abc123", "mainnet"); + expect(lastWhere("tokenTransfer", "findMany").network).toBe("mainnet"); + }); + + it("getNftOwner scopes", async () => { + await db.getNftOwner(CONTRACT, "1", "mainnet"); + expect(lastWhere("nftTransfer", "findFirst").network).toBe("mainnet"); + }); +}); + +describe("cursors are per-network, not singletons", () => { + it("getLastIndexedLedger keys on network instead of id = 1", async () => { + await db.getLastIndexedLedger("mainnet"); + const where = lastWhere("indexerState", "findUnique"); + expect(where).toEqual({ network: "mainnet" }); + expect(where.id).toBeUndefined(); + }); + + it("setLastIndexedLedger upserts the network row", async () => { + await db.setLastIndexedLedger(999, "mainnet"); + const call = calls.find((c) => c.model === "indexerState" && c.op === "upsert")!; + expect(call.args.where).toEqual({ network: "mainnet" }); + expect(call.args.create).toMatchObject({ network: "mainnet", lastIndexedLedger: 999 }); + }); + + it("backfill cursor reads, writes and clears per network", async () => { + await db.getBackfillCursor("mainnet"); + expect(lastWhere("backfillCursor", "findUnique")).toEqual({ network: "mainnet" }); + + await db.setBackfillCursor({ startLedger: 1, endLedger: 9, nextLedger: 5 }, "mainnet"); + const upsert = calls.find((c) => c.model === "backfillCursor" && c.op === "upsert")!; + expect(upsert.args.create).toMatchObject({ network: "mainnet", nextLedger: 5 }); + + await db.clearBackfillCursor("mainnet"); + expect(lastWhere("backfillCursor", "deleteMany")).toEqual({ network: "mainnet" }); + }); +}); + +describe("destructive operations cannot cross networks", () => { + it("rollbackToLedger scopes all three deletes", async () => { + // The dangerous case: ledger sequences are per-chain and testnet runs far + // ahead, so an unscoped `ledger > target` would delete real mainnet rows + // during a testnet reorg. + await db.rollbackToLedger(500, "testnet"); + + for (const model of ["tokenTransfer", "nftTransfer", "hostFnLog"]) { + const where = lastWhere(model, "deleteMany"); + expect(where.network).toBe("testnet"); + expect(where.ledger).toEqual({ gt: 500 }); + } + }); + + it("pruneOldTransfers scopes, so pruning testnet leaves mainnet history alone", async () => { + await db.pruneOldTransfers("testnet"); + expect(lastWhere("tokenTransfer", "deleteMany").network).toBe("testnet"); + }); +}); + +describe("writes stamp the network explicitly", () => { + const record = { + contractId: CONTRACT, + eventType: "transfer", + fromAddress: ADDR, + toAddress: null, + amount: "100", + ledger: 10, + ledgerClosedAt: new Date("2026-01-01T00:00:00Z"), + txHash: "abc", + eventId: "evt-1", + }; + + it("upsertTransfers sets network on every row rather than relying on the column default", async () => { + await db.upsertTransfers([record, { ...record, eventId: "evt-2" }], "mainnet"); + const call = calls.find((c) => c.model === "tokenTransfer" && c.op === "createMany")!; + expect(call.args.data).toHaveLength(2); + for (const row of call.args.data) expect(row.network).toBe("mainnet"); + }); + + it("upsertNftTransfers stamps the network", async () => { + await db.upsertNftTransfers( + [{ contractId: CONTRACT, tokenId: "1", fromAddress: null, toAddress: ADDR, ledger: 1, ledgerClosedAt: new Date(), txHash: "t", eventId: "n-1" } as any], + "mainnet" + ); + const call = calls.find((c) => c.model === "nftTransfer" && c.op === "createMany")!; + expect(call.args.data[0].network).toBe("mainnet"); + }); + + it("NFT metadata is keyed by the compound (network, contractId, tokenId)", async () => { + await db.getNftMetadata(CONTRACT, "7", "mainnet"); + expect(lastWhere("nftMetadata", "findUnique")).toEqual({ + network_contractId_tokenId: { network: "mainnet", contractId: CONTRACT, tokenId: "7" }, + }); + }); +}); + +describe("raw SQL carries the network predicate", () => { + it("querySummary passes the network as a bound parameter", async () => { + // Raw SQL bypasses Prisma's where-builder entirely, so this is the path + // most likely to be missed — and it aggregates balances. + await db.querySummary({ address: ADDR, network: "mainnet" }); + expect(rawValues()).toContain("mainnet"); + }); + + it("queryPopularAssets scopes both the count and the page", async () => { + await db.queryPopularAssets({ fromDate: new Date(0), by: "volume", limit: 10, offset: 0, network: "mainnet" }); + // Two statements run: the DISTINCT count and the grouped page. If only one + // were scoped, `total` would disagree with the rows returned. + expect(raw).toHaveLength(2); + for (const statement of raw) expect(statement.values).toContain("mainnet"); + }); + + it("upsertAccountSummaries binds the network into the INSERT", async () => { + await db.upsertAccountSummaries( + [{ ...record_(), fromAddress: ADDR, toAddress: null }], + "mainnet" + ); + expect(rawValues()).toContain("mainnet"); + }); + + function record_() { + return { + contractId: CONTRACT, + eventType: "transfer", + fromAddress: ADDR, + toAddress: null, + amount: "100", + ledger: 10, + ledgerClosedAt: new Date("2026-01-01T00:00:00Z"), + txHash: "abc", + eventId: "evt-1", + }; + } +}); diff --git a/src/__tests__/routes/transfers.test.ts b/src/__tests__/routes/transfers.test.ts index d744ed49..425135c9 100644 --- a/src/__tests__/routes/transfers.test.ts +++ b/src/__tests__/routes/transfers.test.ts @@ -52,6 +52,7 @@ function makeTransfer(overrides: TransferOverrides = {}) { function baseTransfer() { return { id: 1, + network: "testnet", contractId: CONTRACT_A, eventType: "transfer", fromAddress: BOB, diff --git a/src/db.ts b/src/db.ts index bdd57540..4ca80066 100644 --- a/src/db.ts +++ b/src/db.ts @@ -1,6 +1,12 @@ import { PrismaClient, Prisma } from "@prisma/client"; import type { NftTransferRecord, NftMetadataPayload } from "./ingester/nft"; import { decodeCursor, encodeCursor, parseODataFilter, parseODataSelect, projectRecord } from "./lib/odata"; +import { resolveNetwork, type Network } from "./network"; + +// Every function below takes an optional trailing `network`. Omitting it means +// "the network this process is configured for" (STELLAR_NETWORK), which is +// exactly the pre-#159 behaviour for single-network deployments. The per-network +// indexer loop (#161) and the API selector (#163) pass it explicitly. const STROOPS = 10_000_000n; @@ -163,15 +169,19 @@ const ACCOUNT_SUMMARY_FIELD_TYPES = { // ─── Upsert helper ───────────────────────────────────────────────────────── /** * Idempotently insert a batch of transfer events. - * Conflicts on `eventId` are silently ignored — safe to call multiple times - * with overlapping ledger ranges. + * Conflicts on `(network, eventId)` are silently ignored — safe to call + * multiple times with overlapping ledger ranges. */ -export async function upsertTransfers(records: TransferRecord[]): Promise { +export async function upsertTransfers( + records: TransferRecord[], + network?: Network +): Promise { if (records.length === 0) return 0; + const net = resolveNetwork(network); // Prisma's createMany with skipDuplicates is the most efficient bulk path. const result = await prisma.tokenTransfer.createMany({ - data: records, + data: records.map((r) => ({ ...r, network: net })), skipDuplicates: true, }); @@ -181,28 +191,35 @@ export async function upsertTransfers(records: TransferRecord[]): Promise { - const state = await prisma.indexerState.findUnique({ where: { id: 1 } }); +export async function getLastIndexedLedger(network?: Network): Promise { + const state = await prisma.indexerState.findUnique({ + where: { network: resolveNetwork(network) }, + }); return state?.lastIndexedLedger ?? null; } /** * Read the last indexed ledger and state details from DB. */ -export async function getLastIndexedState(): Promise<{ lastIndexedLedger: number | null }> { - const state = await prisma.indexerState.findUnique({ where: { id: 1 } }); +export async function getLastIndexedState( + network?: Network +): Promise<{ lastIndexedLedger: number | null }> { + const state = await prisma.indexerState.findUnique({ + where: { network: resolveNetwork(network) }, + }); return { lastIndexedLedger: state?.lastIndexedLedger ?? null }; } /** * Persist the last successfully indexed ledger sequence number. */ -export async function setLastIndexedLedger(ledger: number): Promise { +export async function setLastIndexedLedger(ledger: number, network?: Network): Promise { + const net = resolveNetwork(network); await prisma.indexerState.upsert({ - where: { id: 1 }, - create: { id: 1, lastIndexedLedger: ledger }, + where: { network: net }, + create: { network: net, lastIndexedLedger: ledger }, update: { lastIndexedLedger: ledger }, }); } @@ -214,23 +231,31 @@ export interface BackfillCursorState { nextLedger: number; } -export async function getBackfillCursor(): Promise { - const state = await prisma.backfillCursor.findUnique({ where: { id: 1 } }); +export async function getBackfillCursor( + network?: Network +): Promise { + const state = await prisma.backfillCursor.findUnique({ + where: { network: resolveNetwork(network) }, + }); return state ? { startLedger: state.startLedger, endLedger: state.endLedger, nextLedger: state.nextLedger } : null; } -export async function setBackfillCursor(cursor: BackfillCursorState): Promise { +export async function setBackfillCursor( + cursor: BackfillCursorState, + network?: Network +): Promise { + const net = resolveNetwork(network); await prisma.backfillCursor.upsert({ - where: { id: 1 }, - create: { id: 1, ...cursor }, + where: { network: net }, + create: { network: net, ...cursor }, update: cursor, }); } -export async function clearBackfillCursor(): Promise { - await prisma.backfillCursor.deleteMany({ where: { id: 1 } }); +export async function clearBackfillCursor(network?: Network): Promise { + await prisma.backfillCursor.deleteMany({ where: { network: resolveNetwork(network) } }); } // ─── Data retention ───────────────────────────────────────────────────────── @@ -240,17 +265,20 @@ const RETENTION_DAYS = parseInt(process.env.RETENTION_DAYS ?? "30", 10); * Delete transfers older than RETENTION_DAYS to keep the DB within free-tier limits. * Returns the number of rows deleted. */ -export async function pruneOldTransfers(): Promise { +export async function pruneOldTransfers(network?: Network): Promise { + const net = resolveNetwork(network); const cutoff = new Date(); cutoff.setDate(cutoff.getDate() - RETENTION_DAYS); + // Scoped to one network: pruning testnet must never delete mainnet history, + // which is the more expensive of the two to re-index. const result = await prisma.tokenTransfer.deleteMany({ - where: { ledgerClosedAt: { lt: cutoff } }, + where: { network: net, ledgerClosedAt: { lt: cutoff } }, }); if (result.count > 0) { console.log( - `[prune] Deleted ${result.count} transfers older than ${RETENTION_DAYS} days (before ${cutoff.toISOString()})` + `[prune] Deleted ${result.count} ${net} transfers older than ${RETENTION_DAYS} days (before ${cutoff.toISOString()})` ); } @@ -259,6 +287,7 @@ export async function pruneOldTransfers(): Promise { // ─── Query helpers ───────────────────────────────────────────────────────── export type TransferQueryParams = { + network?: Network; address: string; direction: "incoming" | "outgoing"; contractId?: string; @@ -277,6 +306,7 @@ export type TransferQueryParams = { export async function queryTransfers(params: TransferQueryParams) { const { + network, address, direction, contractId, @@ -294,6 +324,7 @@ export async function queryTransfers(params: TransferQueryParams) { } = params; const baseWhere: Prisma.TokenTransferWhereInput = { + network: resolveNetwork(network), ...(direction === "incoming" ? { toAddress: address } : { fromAddress: address }), ...(contractId ? { contractId } : {}), ...(token ? { contractId: token } : {}), @@ -364,15 +395,16 @@ export async function queryTransfers(params: TransferQueryParams) { }; } -export async function queryByTxHash(txHash: string) { +export async function queryByTxHash(txHash: string, network?: Network) { return prisma.tokenTransfer.findMany({ - where: { txHash }, + where: { network: resolveNetwork(network), txHash }, orderBy: { id: "asc" }, }); } // ─── Summary aggregate query ─────────────────────────────────────────────── export type SummaryQueryParams = { + network?: Network; address: string; contractId?: string; fromDate?: Date; @@ -391,9 +423,12 @@ type SummaryRow = { * Uses a raw SQL query because Prisma cannot SUM string-typed columns. */ export async function querySummary(params: SummaryQueryParams): Promise { - const { address, contractId, fromDate, toDate } = params; + const { network, address, contractId, fromDate, toDate } = params; + // Raw SQL bypasses Prisma's where-builder, so the network predicate has to be + // added by hand here — a missing one would silently sum both chains together. const conditions: Prisma.Sql[] = [ + Prisma.sql`"network" = ${resolveNetwork(network)}`, Prisma.sql`("toAddress" = ${address} OR "fromAddress" = ${address})`, ]; if (contractId) conditions.push(Prisma.sql`"contractId" = ${contractId}`); @@ -417,10 +452,14 @@ export async function querySummary(params: SummaryQueryParams): Promise { +export async function upsertNftTransfers( + records: NftTransferRecord[], + network?: Network +): Promise { if (records.length === 0) return 0; + const net = resolveNetwork(network); const result = await prisma.nftTransfer.createMany({ - data: records, + data: records.map((r) => ({ ...r, network: net })), skipDuplicates: true, }); return result.count; @@ -428,10 +467,13 @@ export async function upsertNftTransfers(records: NftTransferRecord[]): Promise< export async function getNftMetadata( contractId: string, - tokenId: string + tokenId: string, + network?: Network ): Promise<{ name: string | null; tokenUri: string | null } | null> { return prisma.nftMetadata.findUnique({ - where: { contractId_tokenId: { contractId, tokenId } }, + where: { + network_contractId_tokenId: { network: resolveNetwork(network), contractId, tokenId }, + }, select: { name: true, tokenUri: true }, }); } @@ -442,15 +484,22 @@ export async function getNftMetadata( * and atomically updates the indexer state to reflect the new tip. * Returns the number of deleted rows (sum across tables). */ -export async function rollbackToLedger(targetLedger: number): Promise { - // Perform deletes and state update atomically. +export async function rollbackToLedger( + targetLedger: number, + network?: Network +): Promise { + const net = resolveNetwork(network); + + // Every delete is network-scoped. Ledger sequences are per-chain and testnet + // runs far ahead of mainnet, so an unscoped `ledger > target` would let a + // testnet reorg delete real mainnet history. const [deletedTransfers, deletedNftTransfers, deletedHostFnLogs, _state] = await prisma.$transaction([ - prisma.tokenTransfer.deleteMany({ where: { ledger: { gt: targetLedger } } }), - prisma.nftTransfer.deleteMany({ where: { ledger: { gt: targetLedger } } }), - prisma.hostFnLog.deleteMany({ where: { ledger: { gt: targetLedger } } }), + prisma.tokenTransfer.deleteMany({ where: { network: net, ledger: { gt: targetLedger } } }), + prisma.nftTransfer.deleteMany({ where: { network: net, ledger: { gt: targetLedger } } }), + prisma.hostFnLog.deleteMany({ where: { network: net, ledger: { gt: targetLedger } } }), prisma.indexerState.upsert({ - where: { id: 1 }, - create: { id: 1, lastIndexedLedger: targetLedger }, + where: { network: net }, + create: { network: net, lastIndexedLedger: targetLedger }, update: { lastIndexedLedger: targetLedger }, }), ]); @@ -459,9 +508,9 @@ export async function rollbackToLedger(targetLedger: number): Promise { (deletedTransfers?.count ?? 0) + (deletedNftTransfers?.count ?? 0) + (deletedHostFnLogs?.count ?? 0); if (totalDeleted > 0) { - console.log(`[reorg] Rolled back to ledger ${targetLedger}, deleted ${totalDeleted} rows`); + console.log(`[reorg] Rolled back ${net} to ledger ${targetLedger}, deleted ${totalDeleted} rows`); } else { - console.log(`[reorg] Rolled back to ledger ${targetLedger}, no rows deleted`); + console.log(`[reorg] Rolled back ${net} to ledger ${targetLedger}, no rows deleted`); } return totalDeleted; @@ -470,16 +519,25 @@ export async function rollbackToLedger(targetLedger: number): Promise { export async function upsertNftMetadata( contractId: string, tokenId: string, - data: NftMetadataPayload + data: NftMetadataPayload, + network?: Network ): Promise { + const net = resolveNetwork(network); await prisma.nftMetadata.upsert({ - where: { contractId_tokenId: { contractId, tokenId } }, - create: { contractId, tokenId, name: data.name ?? null, tokenUri: data.tokenUri ?? null }, + where: { network_contractId_tokenId: { network: net, contractId, tokenId } }, + create: { + network: net, + contractId, + tokenId, + name: data.name ?? null, + tokenUri: data.tokenUri ?? null, + }, update: { name: data.name ?? null, tokenUri: data.tokenUri ?? null, fetchedAt: new Date() }, }); } export type NftTransferQueryParams = { + network?: Network; contractId?: string; tokenId?: string; address?: string; @@ -494,6 +552,7 @@ export type NftTransferQueryParams = { export async function queryNftTransfers(params: NftTransferQueryParams) { const { + network, contractId, tokenId, address, @@ -507,6 +566,7 @@ export async function queryNftTransfers(params: NftTransferQueryParams) { } = params; const baseWhere: Prisma.NftTransferWhereInput = { + network: resolveNetwork(network), ...(contractId ? { contractId } : {}), ...(tokenId ? { tokenId } : {}), ...(address ? { OR: [{ fromAddress: address }, { toAddress: address }] } : {}), @@ -568,10 +628,11 @@ export async function queryNftTransfers(params: NftTransferQueryParams) { */ export async function getNftOwner( contractId: string, - tokenId: string + tokenId: string, + network?: Network ): Promise { const latest = await prisma.nftTransfer.findFirst({ - where: { contractId, tokenId, toAddress: { not: null } }, + where: { network: resolveNetwork(network), contractId, tokenId, toAddress: { not: null } }, orderBy: [{ ledger: "desc" }, { id: "desc" }], select: { toAddress: true }, }); @@ -591,8 +652,14 @@ export async function getNftOwner( * * Using raw SQL because Prisma cannot do arithmetic on string-typed NUMERIC columns. */ -export async function upsertAccountSummaries(records: TransferRecord[]): Promise { +export async function upsertAccountSummaries( + records: TransferRecord[], + network?: Network +): Promise { if (records.length === 0) return; + // Deliberately not called `net` — this table already has a `net` column + // holding a balance, and the SQL below references both. + const networkName = resolveNetwork(network); // Accumulate deltas keyed by "address|contractId" const deltas = new Map< @@ -624,12 +691,17 @@ export async function upsertAccountSummaries(records: TransferRecord[]): Promise const receivedStr = received.toString(); const netStr = (received - sent).toString(); + // The ON CONFLICT target must name the same columns as the unique index, + // which #159 widened to (network, address, contractId). Leaving the old + // two-column target here would raise + // "no unique or exclusion constraint matching the ON CONFLICT specification" + // on every write, not merely mis-scope the aggregate. await prisma.$executeRaw` INSERT INTO wraith."AccountSummary" - (address, "contractId", "totalSent", "totalReceived", net, "txCount", "lastActivityAt", "updatedAt") + ("network", address, "contractId", "totalSent", "totalReceived", net, "txCount", "lastActivityAt", "updatedAt") VALUES - (${address}, ${contractId}, ${sentStr}, ${receivedStr}, ${netStr}, ${count}, ${lastAt}, NOW()) - ON CONFLICT (address, "contractId") DO UPDATE SET + (${networkName}, ${address}, ${contractId}, ${sentStr}, ${receivedStr}, ${netStr}, ${count}, ${lastAt}, NOW()) + ON CONFLICT ("network", address, "contractId") DO UPDATE SET "totalSent" = (wraith."AccountSummary"."totalSent"::NUMERIC + ${sentStr}::NUMERIC)::TEXT, "totalReceived" = (wraith."AccountSummary"."totalReceived"::NUMERIC + ${receivedStr}::NUMERIC)::TEXT, net = (wraith."AccountSummary"."totalReceived"::NUMERIC + ${receivedStr}::NUMERIC @@ -645,9 +717,14 @@ export async function upsertAccountSummaries(records: TransferRecord[]): Promise * Return all asset rows for a given address, optionally filtered to one contract. * O(1) — reads directly from the materialized AccountSummary table. */ -export async function getAccountSummary(address: string, contractId?: string) { +export async function getAccountSummary( + address: string, + contractId?: string, + network?: Network +) { return prisma.accountSummary.findMany({ where: { + network: resolveNetwork(network), address, ...(contractId ? { contractId } : {}), }, @@ -664,6 +741,7 @@ export async function getAccountSummary(address: string, contractId?: string) { } export type AccountSummaryQueryParams = { + network?: Network; address: string; contractId?: string; filter?: string; @@ -674,9 +752,10 @@ export type AccountSummaryQueryParams = { }; export async function queryAccountSummaries(params: AccountSummaryQueryParams) { - const { address, contractId, filter, select, cursor, limit = 50, offset = 0 } = params; + const { network, address, contractId, filter, select, cursor, limit = 50, offset = 0 } = params; const baseWhere: Prisma.AccountSummaryWhereInput = { + network: resolveNetwork(network), address, ...(contractId ? { contractId } : {}), }; @@ -729,6 +808,7 @@ export async function queryAccountSummaries(params: AccountSummaryQueryParams) { // ─── Combined address query ──────────────────────────────────────────────── export type AllTransfersQueryParams = { + network?: Network; address: string; contractId?: string; token?: string; @@ -746,6 +826,7 @@ export type AllTransfersQueryParams = { export async function queryAllTransfers(params: AllTransfersQueryParams) { const { + network, address, contractId, token, @@ -762,6 +843,7 @@ export async function queryAllTransfers(params: AllTransfersQueryParams) { } = params; const baseWhere: Prisma.TokenTransferWhereInput = { + network: resolveNetwork(network), OR: [{ toAddress: address }, { fromAddress: address }], ...(contractId ? { contractId } : {}), ...(token ? { contractId: token } : {}), @@ -832,6 +914,7 @@ export async function queryAllTransfers(params: AllTransfersQueryParams) { // ─── Popular assets query ────────────────────────────────────────────────── export type PopularAssetsQueryParams = { + network?: Network; fromDate: Date; by: string; limit: number; @@ -845,17 +928,20 @@ type PopularAssetRow = { }; export async function queryPopularAssets(params: PopularAssetsQueryParams) { - const { fromDate, by, limit, offset } = params; + const { network, fromDate, by, limit, offset } = params; + const net = resolveNetwork(network); const cap = Math.min(limit, 100); const orderClause = by === "volume" ? Prisma.sql`SUM(CAST("amount" AS NUMERIC)) DESC` : Prisma.sql`COUNT(*) DESC`; + // Both halves filter on network. Leaving it off the count but not the page + // (or vice versa) would return a total that disagrees with the rows. const countResult = await prisma.$queryRaw>` SELECT COUNT(DISTINCT "contractId")::INT8 AS "total" FROM "wraith"."TokenTransfer" - WHERE "ledgerClosedAt" >= ${fromDate} + WHERE "network" = ${net} AND "ledgerClosedAt" >= ${fromDate} `; const total = Number(countResult[0]?.total ?? 0); @@ -865,7 +951,7 @@ export async function queryPopularAssets(params: PopularAssetsQueryParams) { COUNT(*)::INT8 AS "transferCount", COALESCE(SUM(CAST("amount" AS NUMERIC)), 0)::TEXT AS "volume" FROM "wraith"."TokenTransfer" - WHERE "ledgerClosedAt" >= ${fromDate} + WHERE "network" = ${net} AND "ledgerClosedAt" >= ${fromDate} GROUP BY "contractId" ORDER BY ${orderClause} LIMIT ${cap} diff --git a/src/indexer/checkpoint.ts b/src/indexer/checkpoint.ts index 0abb650f..f1e8288a 100644 --- a/src/indexer/checkpoint.ts +++ b/src/indexer/checkpoint.ts @@ -1,8 +1,9 @@ import { Prisma } from "@prisma/client"; -import { prisma } from "../db"; +import { prisma, upsertAccountSummaries } from "../db"; import type { TransferRecord } from "../db"; import type { NftTransferRecord } from "../ingester/nft"; import type { HostFnRecord } from "./host-fn-log"; +import { resolveNetwork, type Network } from "../network"; /** * Batch metadata for atomic processing. @@ -29,20 +30,26 @@ export interface BatchPayload { * Useful for idempotent restart: if we crash mid-batch, resuming with the same * batchId allows us to skip re-processing. */ -export async function hasCheckpoint(batchId: string): Promise { +export async function hasCheckpoint(batchId: string, network?: Network): Promise { const checkpoint = await prisma.indexerCheckpoint.findUnique({ - where: { batchId }, + where: { network_batchId: { network: resolveNetwork(network), batchId } }, select: { id: true }, }); return checkpoint !== null; } /** - * Get the most recent checkpoint across all batches (for single-worker resume). - * Returns the last ledger we successfully processed, or null if no checkpoints exist. + * Get the most recent checkpoint across all batches on one network (for + * single-worker resume). Returns the last ledger we successfully processed, or + * null if no checkpoints exist. + * + * Scoping is not optional here: ledger sequences are per-chain, so an unscoped + * "highest lastLedger" would hand a mainnet worker testnet's far-ahead tip and + * skip every mainnet ledger in between. */ -export async function getLastCheckpoint(): Promise { +export async function getLastCheckpoint(network?: Network): Promise { const checkpoint = await prisma.indexerCheckpoint.findFirst({ + where: { network: resolveNetwork(network) }, orderBy: { lastLedger: "desc" }, select: { lastLedger: true }, }); @@ -66,37 +73,44 @@ export async function getLastCheckpoint(): Promise { export async function commitBatch( metadata: BatchMetadata, payload: BatchPayload, + network?: Network, ): Promise<{ transferred: number; nftTransferred: number; hostFnLogs: number; }> { + // Stamped explicitly on every row. The column has a DEFAULT of 'testnet', so + // omitting it compiles and silently files mainnet events under testnet — + // the one failure mode in #159 that no type error would catch. + const net = resolveNetwork(network); + const result = await prisma.$transaction(async (tx) => { - // Upsert token transfers (idempotent by eventId) + // Upsert token transfers (idempotent by (network, eventId)) const transferred = payload.transfers.length ? ( await tx.tokenTransfer.createMany({ - data: payload.transfers, + data: payload.transfers.map((r) => ({ ...r, network: net })), skipDuplicates: true, }) ).count : 0; - // Upsert NFT transfers (idempotent by eventId) + // Upsert NFT transfers (idempotent by (network, eventId)) const nftTransferred = payload.nftTransfers.length ? ( await tx.nftTransfer.createMany({ - data: payload.nftTransfers, + data: payload.nftTransfers.map((r) => ({ ...r, network: net })), skipDuplicates: true, }) ).count : 0; - // Upsert host function logs (idempotent by eventId) + // Upsert host function logs (idempotent by (network, eventId)) const hostFnLogs = payload.hostFnLogs.length ? ( await tx.hostFnLog.createMany({ data: payload.hostFnLogs.map((r) => ({ + network: net, contractId: r.contractId, functionName: r.functionName, args: r.args as Prisma.InputJsonValue, @@ -118,8 +132,9 @@ export async function commitBatch( // Atomically advance the checkpoint. On reprocessing the same batchId, // this upsert will update the timestamp but keep the same lastLedger. await tx.indexerCheckpoint.upsert({ - where: { batchId: metadata.batchId }, + where: { network_batchId: { network: net, batchId: metadata.batchId } }, create: { + network: net, batchId: metadata.batchId, lastLedger: metadata.toLedger, }, @@ -137,90 +152,21 @@ export async function commitBatch( /** * Update account summaries for the given transfer records. - * This is called separately after the main batch commit because it's a derived - * table that aggregates from transfers. If this fails, we don't lose data. + * This is called separately after the main batch commit because it is a + * derived table that aggregates from transfers. If this fails, we do not lose + * data. + * + * Delegates to `upsertAccountSummaries` in db.ts rather than keeping a second + * copy of the raw upsert. The two had drifted into byte-identical duplicates, + * and #159 made that actively dangerous: the statement carries an + * `ON CONFLICT (network, address, contractId)` target that has to match the + * unique index exactly, so a copy left behind would throw + * "no unique or exclusion constraint matching the ON CONFLICT specification" + * on every write. */ export async function updateAccountSummaries( records: TransferRecord[], + network?: Network, ): Promise { - if (records.length === 0) return; - - // Accumulate deltas keyed by "address|contractId" - const deltas = new Map< - string, - { - address: string; - contractId: string; - sent: bigint; - received: bigint; - count: number; - lastAt: Date; - } - >(); - - const touch = ( - address: string, - contractId: string, - sent: bigint, - received: bigint, - at: Date, - ) => { - const key = `${address}|${contractId}`; - const prev = deltas.get(key) ?? { - address, - contractId, - sent: 0n, - received: 0n, - count: 0, - lastAt: at, - }; - deltas.set(key, { - address, - contractId, - sent: prev.sent + sent, - received: prev.received + received, - count: prev.count + 1, - lastAt: at > prev.lastAt ? at : prev.lastAt, - }); - }; - - for (const { - contractId, - fromAddress, - toAddress, - amount, - ledgerClosedAt, - } of records) { - const amt = BigInt(amount); - if (fromAddress) touch(fromAddress, contractId, amt, 0n, ledgerClosedAt); - if (toAddress) touch(toAddress, contractId, 0n, amt, ledgerClosedAt); - } - - for (const { - address, - contractId, - sent, - received, - count, - lastAt, - } of deltas.values()) { - const sentStr = sent.toString(); - const receivedStr = received.toString(); - const netStr = (received - sent).toString(); - - await prisma.$executeRaw` - INSERT INTO wraith."AccountSummary" - (address, "contractId", "totalSent", "totalReceived", net, "txCount", "lastActivityAt", "updatedAt") - VALUES - (${address}, ${contractId}, ${sentStr}, ${receivedStr}, ${netStr}, ${count}, ${lastAt}, NOW()) - ON CONFLICT (address, "contractId") DO UPDATE SET - "totalSent" = (wraith."AccountSummary"."totalSent"::NUMERIC + ${sentStr}::NUMERIC)::TEXT, - "totalReceived" = (wraith."AccountSummary"."totalReceived"::NUMERIC + ${receivedStr}::NUMERIC)::TEXT, - net = (wraith."AccountSummary"."totalReceived"::NUMERIC + ${receivedStr}::NUMERIC - - wraith."AccountSummary"."totalSent"::NUMERIC - ${sentStr}::NUMERIC)::TEXT, - "txCount" = wraith."AccountSummary"."txCount" + ${count}, - "lastActivityAt" = GREATEST(wraith."AccountSummary"."lastActivityAt", ${lastAt}), - "updatedAt" = NOW() - `; - } + await upsertAccountSummaries(records, resolveNetwork(network)); } diff --git a/src/network.ts b/src/network.ts new file mode 100644 index 00000000..1a15d637 --- /dev/null +++ b/src/network.ts @@ -0,0 +1,55 @@ +/** + * The network dimension. + * + * Wraith stores testnet and mainnet rows in the same tables, discriminated by + * a `network` column (#159). This module is the single place that decides what + * "the current network" means, so the answer cannot drift between the indexer, + * the API and the jobs. + * + * Every db.ts function takes an optional `network` argument that defaults to + * {@link currentNetwork}. That keeps today's single-network callers working + * untouched while giving the per-network indexer loop (#161) and the API + * network selector (#163) somewhere explicit to pass. + */ + +export type Network = "testnet" | "mainnet"; + +export const NETWORKS: readonly Network[] = ["testnet", "mainnet"] as const; + +/** The value back-filled onto every pre-existing row by the #159 migration. */ +export const DEFAULT_NETWORK: Network = "testnet"; + +export function isNetwork(value: unknown): value is Network { + return value === "testnet" || value === "mainnet"; +} + +/** + * Coerce arbitrary input (env var, query string, header) to a Network. + * Returns null rather than throwing so callers can decide between a 400 and a + * fallback — an indexer wants to fail loudly, an HTTP route wants to answer. + */ +export function parseNetwork(value: unknown): Network | null { + if (typeof value !== "string") return null; + const normalised = value.trim().toLowerCase(); + return isNetwork(normalised) ? normalised : null; +} + +/** + * The network this process is configured for, from `STELLAR_NETWORK`. + * + * Read from the environment on every call rather than cached at import time: + * caching would freeze whatever the environment happened to be when the first + * module imported this one, which makes the value untestable and surprising in + * workers that set their environment after startup. + * + * Falls back to testnet — the same default the column carries — so an unset + * variable behaves exactly like the pre-#159 code did. + */ +export function currentNetwork(): Network { + return parseNetwork(process.env.STELLAR_NETWORK) ?? DEFAULT_NETWORK; +} + +/** Resolve an optional explicit network against the configured default. */ +export function resolveNetwork(network?: Network): Network { + return network ?? currentNetwork(); +} From 32b59f367a2bad893c44dac102b8dd92a28083a5 Mon Sep 17 00:00:00 2001 From: Salmatcre8 Date: Sun, 30 Aug 2026 09:42:35 +0100 Subject: [PATCH 2/2] fix(tests): key integration IndexerState by network, not id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The integration suite still created and read IndexerState with the old singleton key: prisma.indexerState.create({ data: { id: 1, lastIndexedLedger: 101 } }) prisma.indexerState.findUnique({ where: { id: 1 } }) `id` no longer exists on that model, so every integration file failed at setup with PrismaClientValidationError. Why the unit run and typecheck missed it: tsconfig.json has `include: ["src/**/*"]`, so tests/ is never typechecked, and the integration suite is a separate vitest config that does not run in the unit job. `tsc --noEmit` was clean and 270 unit tests passed while six integration suites were broken. Widening `include` to tests/ conflicts with `rootDir: "src"`, so a proper fix needs a separate tsconfig for typechecking tests — worth its own issue rather than changing the build layout here. Claude-Session: https://claude.ai/code/session_01USgemLt4Rnz4SGB1Srf3GB --- tests/integration/reorg-rollover.test.ts | 4 ++-- tests/integration/setup.ts | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/tests/integration/reorg-rollover.test.ts b/tests/integration/reorg-rollover.test.ts index 53d70452..8865abd1 100644 --- a/tests/integration/reorg-rollover.test.ts +++ b/tests/integration/reorg-rollover.test.ts @@ -42,7 +42,7 @@ describe("automatic reorg rollback", () => { ], }); - await prisma.indexerState.create({ data: { id: 1, lastIndexedLedger: 101 } }); + await prisma.indexerState.create({ data: { network: 'testnet', lastIndexedLedger: 101 } }); // Build in-memory buffer reflecting observed ledgers const reorg = new ReorgHandler(16); @@ -85,7 +85,7 @@ describe("automatic reorg rollback", () => { expect(rows100[0].eventId).toBe("evt-100"); // Verify indexer state updated to 101 - const state = await prisma.indexerState.findUnique({ where: { id: 1 } }); + const state = await prisma.indexerState.findUnique({ where: { network: 'testnet' } }); expect(state?.lastIndexedLedger).toBe(101); expect((result as any).action).toBe("reorg"); diff --git a/tests/integration/setup.ts b/tests/integration/setup.ts index 3a4b7ec9..a608279b 100644 --- a/tests/integration/setup.ts +++ b/tests/integration/setup.ts @@ -36,7 +36,7 @@ export async function seedIntegrationFixtures(): Promise { await prisma.tokenTransfer.deleteMany(); await prisma.indexerState.deleteMany(); await prisma.tokenTransfer.createMany({ data: seedTransfers }); - await prisma.indexerState.create({ data: { id: 1, lastIndexedLedger: 2006 } }); + await prisma.indexerState.create({ data: { network: 'testnet', lastIndexedLedger: 2006 } }); } finally { await prisma.$disconnect(); }