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(); +} 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(); }