diff --git a/apps/api/src/db/client.ts b/apps/api/src/db/client.ts index af47ffaa3..dbf6b518a 100644 --- a/apps/api/src/db/client.ts +++ b/apps/api/src/db/client.ts @@ -32,14 +32,20 @@ const BOOTSTRAP_SQL = [ expires_at INTEGER, created_at INTEGER NOT NULL, updated_at INTEGER NOT NULL )`, // Cumulative payment ledger (issue 1.4) — one row per payment ever recorded - // against a link, `tx_hash` unique so a reprocessed payment can't double-count. + // against a link. Unique on (tx_hash, operation_id), not tx_hash alone + // (issue 4.11): a transaction can carry more than one payment operation to + // the same link, and each must land its own row rather than being dropped + // as a false duplicate of the other. `CREATE TABLE IF NOT EXISTS link_payments ( - id TEXT PRIMARY KEY, link_id TEXT NOT NULL, tx_hash TEXT NOT NULL UNIQUE, + id TEXT PRIMARY KEY, link_id TEXT NOT NULL, tx_hash TEXT NOT NULL, + operation_id TEXT, payer TEXT NOT NULL, amount TEXT NOT NULL, asset_code TEXT NOT NULL, asset_issuer TEXT, ledger INTEGER, created_at INTEGER NOT NULL )`, `CREATE INDEX IF NOT EXISTS link_payments_link_id_idx ON link_payments (link_id)`, + `CREATE UNIQUE INDEX IF NOT EXISTS link_payments_tx_hash_operation_id_unique + ON link_payments (tx_hash, operation_id)`, `CREATE TABLE IF NOT EXISTS webhooks ( id TEXT PRIMARY KEY, seller_id TEXT NOT NULL, url TEXT NOT NULL, secret_encrypted TEXT NOT NULL, secret_last4 TEXT NOT NULL, @@ -74,9 +80,16 @@ const BOOTSTRAP_SQL = [ `CREATE TABLE IF NOT EXISTS watcher_cursors ( account TEXT PRIMARY KEY, cursor TEXT NOT NULL, updated_at INTEGER NOT NULL )`, + // Watcher dedup ledger (issue 4.11). Keyed on (tx_hash, operation_id), not + // tx_hash alone: a transaction can carry more than one payment operation, + // and each must dedupe independently. operation_id NULL (only possible via + // migrateLegacyProcessedTxTable, never written by new code) means "the + // whole transaction", preserving old behavior for pre-migration rows. `CREATE TABLE IF NOT EXISTS processed_tx ( - tx_hash TEXT PRIMARY KEY, link_id TEXT, created_at INTEGER NOT NULL + tx_hash TEXT NOT NULL, operation_id TEXT, link_id TEXT, created_at INTEGER NOT NULL, + PRIMARY KEY (tx_hash, operation_id) )`, + `CREATE INDEX IF NOT EXISTS processed_tx_tx_hash_idx ON processed_tx (tx_hash)`, `CREATE TABLE IF NOT EXISTS idempotency_keys ( key TEXT NOT NULL, seller_id TEXT NOT NULL, endpoint TEXT NOT NULL, request_hash TEXT NOT NULL, response_status INTEGER NOT NULL, @@ -206,8 +219,81 @@ async function migrateLegacyWebhooksTable(client: Client): Promise { } } +/** + * Rebuilds `processed_tx` around (tx_hash, operation_id) instead of tx_hash + * alone (issue 4.11). SQLite can't ALTER a column into/out of a PRIMARY KEY, + * and the whole point here is that tx_hash must stop being unique by itself — + * two payment operations sharing one transaction need two rows — so this is a + * rename/recreate/copy/drop, not an ADD COLUMN. Legacy rows get + * operation_id = NULL, which `DrizzleWatcherStateRepository.isProcessed` + * treats as "the whole transaction was processed", preserving exactly the + * dedup behavior anything already settled before this migration runs had. + * + * No-op on a fresh database (BOOTSTRAP_SQL below creates the current shape + * directly) and on a database already migrated. + */ +async function migrateLegacyProcessedTxTable(client: Client): Promise { + const info = await client.execute("PRAGMA table_info(processed_tx)"); + const columns = new Set(info.rows.map((r) => String(r.name))); + if (columns.size === 0 || columns.has("operation_id")) return; // fresh table, or already migrated + + await client.execute("ALTER TABLE processed_tx RENAME TO processed_tx_legacy_4_11"); + await client.execute(`CREATE TABLE processed_tx ( + tx_hash TEXT NOT NULL, operation_id TEXT, link_id TEXT, created_at INTEGER NOT NULL, + PRIMARY KEY (tx_hash, operation_id) + )`); + await client.execute("CREATE INDEX IF NOT EXISTS processed_tx_tx_hash_idx ON processed_tx (tx_hash)"); + await client.execute( + `INSERT INTO processed_tx (tx_hash, operation_id, link_id, created_at) + SELECT tx_hash, NULL, link_id, created_at FROM processed_tx_legacy_4_11`, + ); + await client.execute("DROP TABLE processed_tx_legacy_4_11"); +} + +/** + * Same rebuild as `migrateLegacyProcessedTxTable`, for `link_payments` (issue + * 4.11): the unique constraint moves from tx_hash alone to + * (tx_hash, operation_id), so a split payment's second operation gets its own + * ledger row instead of being silently dropped by `onConflictDoNothing`. + * Legacy rows get operation_id = NULL; unlike processed_tx, nothing reads + * link_payments as a dedup gate, so NULL there carries no special meaning — + * it just fills the new column on rows written before it existed. + */ +async function migrateLegacyLinkPaymentsTable(client: Client): Promise { + const info = await client.execute("PRAGMA table_info(link_payments)"); + const columns = new Set(info.rows.map((r) => String(r.name))); + if (columns.size === 0 || columns.has("operation_id")) return; // fresh table, or already migrated + + // `ledger` (issue 9.2) may or may not be present yet depending on how old + // this particular database is — don't assume either way, select it if it's + // there and NULL otherwise, same as `operation_id` always being NULL here. + const ledgerSelect = columns.has("ledger") ? "ledger" : "NULL"; + + await client.execute("ALTER TABLE link_payments RENAME TO link_payments_legacy_4_11"); + await client.execute(`CREATE TABLE link_payments ( + id TEXT PRIMARY KEY, link_id TEXT NOT NULL, tx_hash TEXT NOT NULL, + operation_id TEXT, + payer TEXT NOT NULL, amount TEXT NOT NULL, + asset_code TEXT NOT NULL, asset_issuer TEXT, ledger INTEGER, + created_at INTEGER NOT NULL + )`); + await client.execute("CREATE INDEX IF NOT EXISTS link_payments_link_id_idx ON link_payments (link_id)"); + await client.execute( + `CREATE UNIQUE INDEX IF NOT EXISTS link_payments_tx_hash_operation_id_unique + ON link_payments (tx_hash, operation_id)`, + ); + await client.execute( + `INSERT INTO link_payments (id, link_id, tx_hash, operation_id, payer, amount, asset_code, asset_issuer, ledger, created_at) + SELECT id, link_id, tx_hash, NULL, payer, amount, asset_code, asset_issuer, ${ledgerSelect}, created_at + FROM link_payments_legacy_4_11`, + ); + await client.execute("DROP TABLE link_payments_legacy_4_11"); +} + export async function bootstrap(client: Client): Promise { await migrateLegacyWebhooksTable(client); + await migrateLegacyProcessedTxTable(client); + await migrateLegacyLinkPaymentsTable(client); for (const sql of BOOTSTRAP_SQL) { try { await client.execute(sql); diff --git a/apps/api/src/db/schema.ts b/apps/api/src/db/schema.ts index 9a2ebb7bf..7ee4f7cdd 100644 --- a/apps/api/src/db/schema.ts +++ b/apps/api/src/db/schema.ts @@ -65,11 +65,16 @@ export const links = sqliteTable("links", { // Authoritative ledger of every payment recorded against a link — cumulative // accounting (issue 1.4) sums these rather than trusting a single payment. -// `txHash` is unique so a reprocessed payment can never double-count. +// Unique on (tx_hash, operation_id) — see `migrateLegacyLinkPaymentsTable` in +// db/client.ts — not tx_hash alone: a transaction can carry more than one +// payment operation to the same link (issue 4.11), and each is its own row. export const linkPayments = sqliteTable("link_payments", { id: text("id").primaryKey(), linkId: text("link_id").notNull(), - txHash: text("tx_hash").notNull().unique(), + txHash: text("tx_hash").notNull(), + /** Horizon's per-operation pagingToken. NULL only on rows written before + * this column existed — see `migrateLegacyLinkPaymentsTable`. */ + operationId: text("operation_id"), payer: text("payer").notNull(), amount: text("amount").notNull(), assetCode: text("asset_code").notNull(), @@ -150,8 +155,16 @@ export const watcherCursors = sqliteTable("watcher_cursors", { updatedAt: integer("updated_at").notNull(), }); +// Dedup ledger for the watcher (issue 4.11). Keyed on (tx_hash, operation_id) +// — see `migrateLegacyProcessedTxTable` in db/client.ts — not tx_hash alone: a +// Stellar transaction can carry up to 100 operations, and a payment is one +// operation, not the whole transaction. `operation_id` is Horizon's +// pagingToken; it's NULL only on rows migrated from before this column +// existed, and `isProcessed` treats a NULL row as "the whole transaction was +// processed" to preserve the dedup behavior anything already settled had. export const processedTx = sqliteTable("processed_tx", { - txHash: text("tx_hash").primaryKey(), + txHash: text("tx_hash").notNull(), + operationId: text("operation_id"), linkId: text("link_id"), createdAt: integer("created_at").notNull(), }); diff --git a/apps/api/src/repos/index.ts b/apps/api/src/repos/index.ts index 128ee408d..d3b91071e 100644 --- a/apps/api/src/repos/index.ts +++ b/apps/api/src/repos/index.ts @@ -1,4 +1,4 @@ -import { and, desc, eq, inArray, isNotNull, isNull, lt } from "drizzle-orm"; +import { and, desc, eq, inArray, isNotNull, isNull, lt, or } from "drizzle-orm"; import type { ApiKeyScope } from "../services/api-keys"; import { decodeScopesFromDb, encodeScopesForDb } from "../services/api-keys"; import type { @@ -268,6 +268,7 @@ export class DrizzleLinkRepository implements LinkRepository { id: newId("pmt"), linkId: payment.linkId, txHash: payment.txHash, + operationId: payment.operationId, payer: payment.payer, amount: payment.amount, assetCode: payment.asset.code, @@ -275,7 +276,7 @@ export class DrizzleLinkRepository implements LinkRepository { ledger: payment.ledger, createdAt: payment.createdAt, }) - .onConflictDoNothing({ target: linkPayments.txHash }); + .onConflictDoNothing({ target: [linkPayments.txHash, linkPayments.operationId] }); } async paymentLedger(txHash: string): Promise { @@ -551,20 +552,27 @@ export class DrizzleWatcherStateRepository implements WatcherStateRepository { }); } - async isProcessed(txHash: string): Promise { + async isProcessed(txHash: string, operationId: string): Promise { const rows = await this.db .select({ txHash: processedTx.txHash }) .from(processedTx) - .where(eq(processedTx.txHash, txHash)) + .where( + and( + eq(processedTx.txHash, txHash), + // Exact operation already recorded, OR a pre-migration row marked + // the whole transaction (operation_id NULL) — see schema.ts. + or(eq(processedTx.operationId, operationId), isNull(processedTx.operationId)), + ), + ) .limit(1); return rows.length > 0; } - async markProcessed(txHash: string, linkId: string | null): Promise { + async markProcessed(txHash: string, operationId: string, linkId: string | null): Promise { await this.db .insert(processedTx) - .values({ txHash, linkId, createdAt: Date.now() }) - .onConflictDoNothing(); + .values({ txHash, operationId, linkId, createdAt: Date.now() }) + .onConflictDoNothing({ target: [processedTx.txHash, processedTx.operationId] }); } } diff --git a/apps/api/src/services/link-service.ts b/apps/api/src/services/link-service.ts index bb97aa231..e776e431a 100644 --- a/apps/api/src/services/link-service.ts +++ b/apps/api/src/services/link-service.ts @@ -633,6 +633,7 @@ export class LinkService { await this.deps.links.recordPayment({ linkId: link.id, txHash: payment.txHash, + operationId: payment.pagingToken, payer: payment.from, amount: normalizeAmount(payment.amount), asset: payment.asset, @@ -675,6 +676,7 @@ export class LinkService { await this.deps.links.recordPayment({ linkId: link.id, txHash: payment.txHash, + operationId: payment.pagingToken, payer: payment.from, amount: normalizeAmount(payment.amount), asset: payment.asset, diff --git a/apps/api/src/worker/watcher-loop.ts b/apps/api/src/worker/watcher-loop.ts index fd9a63ebe..4de828774 100644 --- a/apps/api/src/worker/watcher-loop.ts +++ b/apps/api/src/worker/watcher-loop.ts @@ -415,7 +415,7 @@ export class WatcherLoop { for (const payment of payments) { lastToken = payment.pagingToken; const child = log.child({ txHash: payment.txHash, pagingToken: payment.pagingToken }); - if (await this.deps.state.isProcessed(payment.txHash)) { + if (await this.deps.state.isProcessed(payment.txHash, payment.pagingToken)) { child.info({ event: "payment.duplicate" }, "skipping already-processed payment"); continue; } @@ -461,7 +461,7 @@ export class WatcherLoop { } } - await this.deps.state.markProcessed(payment.txHash, linkId); + await this.deps.state.markProcessed(payment.txHash, payment.pagingToken, linkId); } pageCursor = lastToken; diff --git a/apps/api/test/bootstrap-migrations.test.ts b/apps/api/test/bootstrap-migrations.test.ts index aabf21d4c..1660fe5c1 100644 --- a/apps/api/test/bootstrap-migrations.test.ts +++ b/apps/api/test/bootstrap-migrations.test.ts @@ -1,6 +1,8 @@ import { describe, it, expect } from "vitest"; import { createClient } from "@libsql/client"; +import { drizzle } from "drizzle-orm/libsql"; import { bootstrap } from "../src/db/client"; +import { DrizzleWatcherStateRepository } from "../src/repos/index"; import * as schema from "../src/db/schema"; import { getTableConfig } from "drizzle-orm/sqlite-core"; @@ -36,6 +38,11 @@ const LEGACY_LINK_PAYMENTS = `CREATE TABLE link_payments ( created_at INTEGER NOT NULL )`; +/** `processed_tx` as it shipped before issue 4.11 — tx_hash alone as the key. */ +const LEGACY_PROCESSED_TX = `CREATE TABLE processed_tx ( + tx_hash TEXT PRIMARY KEY, link_id TEXT, created_at INTEGER NOT NULL +)`; + // Copied verbatim from the production database's own sqlite_master, not // guessed: `wallet` has NO UNIQUE here. That single missing constraint is what // made every wallet login 500, and a fixture that quietly adds it back tests @@ -143,6 +150,89 @@ describe("bootstrap() against a pre-existing database", () => { expect(await columnsOf(client, "sellers")).toContain("payout_fields_json"); }); + // Issue 4.11 rebuilds processed_tx and link_payments rather than ALTERing + // them — SQLite cannot move a column into or out of a PRIMARY KEY. A rebuild + // that drops rows is a money bug: processed_tx is the dedup ledger, so + // losing it re-credits payments that already settled. + it("preserves every processed_tx row through the 4.11 rebuild", async () => { + const client = createClient({ url: "file::memory:" }); + await client.execute(LEGACY_PROCESSED_TX); + await client.execute( + `INSERT INTO processed_tx (tx_hash, link_id, created_at) VALUES + ('tx_old_1', 'lnk_a', 1000), + ('tx_old_2', NULL, 2000)`, + ); + + await bootstrap(client); + + const rows = await client.execute("SELECT tx_hash, operation_id, link_id, created_at FROM processed_tx ORDER BY tx_hash"); + expect(rows.rows.map((r) => String(r.tx_hash))).toEqual(["tx_old_1", "tx_old_2"]); + // Legacy rows carry no operation id — that is what makes them mean + // "the whole transaction was processed". + expect(rows.rows.every((r) => r.operation_id === null)).toBe(true); + expect(String(rows.rows[0]!.link_id)).toBe("lnk_a"); + expect(Number(rows.rows[0]!.created_at)).toBe(1000); + }); + + it("preserves every link_payments row through the 4.11 rebuild", async () => { + const client = createClient({ url: "file::memory:" }); + await client.execute(LEGACY_LINK_PAYMENTS); + await client.execute( + `INSERT INTO link_payments (id, link_id, tx_hash, payer, amount, asset_code, asset_issuer, created_at) + VALUES ('pmt_1', 'lnk_a', 'tx_old_1', 'GPAYER', '10', 'USDC', 'GISSUER', 1000)`, + ); + + await bootstrap(client); + + const rows = await client.execute("SELECT id, tx_hash, operation_id, amount, ledger FROM link_payments"); + expect(rows.rows).toHaveLength(1); + expect(String(rows.rows[0]!.id)).toBe("pmt_1"); + expect(String(rows.rows[0]!.amount)).toBe("10"); + expect(rows.rows[0]!.operation_id).toBe(null); + // The legacy fixture predates the `ledger` column; the rebuild must not + // assume it was there. + expect(rows.rows[0]!.ledger).toBe(null); + }); + + // The dedup semantics the migration promises: a pre-4.11 row means the whole + // transaction is done, so a replay of any operation in it is still a + // duplicate. Without this, every payment settled before the migration could + // be credited a second time. + it("treats a migrated NULL-operation row as covering the whole transaction", async () => { + const client = createClient({ url: "file::memory:" }); + await client.execute(LEGACY_PROCESSED_TX); + await client.execute( + "INSERT INTO processed_tx (tx_hash, link_id, created_at) VALUES ('tx_settled', 'lnk_a', 1000)", + ); + await bootstrap(client); + + const state = new DrizzleWatcherStateRepository(drizzle(client, { schema })); + + expect(await state.isProcessed("tx_settled", "any-operation-id")).toBe(true); + expect(await state.isProcessed("tx_settled", "another-one")).toBe(true); + expect(await state.isProcessed("tx_never_seen", "1")).toBe(false); + }); + + // Two operations in one transaction must each get their own row — that is + // the entire point of the re-key. + it("records two operations of one transaction independently after migrating", async () => { + const client = createClient({ url: "file::memory:" }); + await client.execute(LEGACY_PROCESSED_TX); + await bootstrap(client); + + const state = new DrizzleWatcherStateRepository(drizzle(client, { schema })); + await state.markProcessed("tx_split", "op_1", "lnk_a"); + + expect(await state.isProcessed("tx_split", "op_1")).toBe(true); + expect(await state.isProcessed("tx_split", "op_2")).toBe(false); + + await state.markProcessed("tx_split", "op_2", "lnk_a"); + expect(await state.isProcessed("tx_split", "op_2")).toBe(true); + + const rows = await client.execute("SELECT operation_id FROM processed_tx WHERE tx_hash = 'tx_split'"); + expect(rows.rows).toHaveLength(2); + }); + it("is idempotent — a second run over a migrated database is a no-op", async () => { const client = createClient({ url: "file::memory:" }); await client.execute(LEGACY_LINKS); @@ -163,12 +253,13 @@ describe("bootstrap() against a pre-existing database", () => { await legacy.execute(LEGACY_LINKS); await legacy.execute(LEGACY_LINK_PAYMENTS); await legacy.execute(LEGACY_SELLERS); + await legacy.execute(LEGACY_PROCESSED_TX); await bootstrap(legacy); const fresh = createClient({ url: "file::memory:" }); await bootstrap(fresh); - for (const table of ["links", "link_payments", "sellers"]) { + for (const table of ["links", "link_payments", "sellers", "processed_tx"]) { const a = (await columnsOf(legacy, table)).slice().sort(); const b = (await columnsOf(fresh, table)).slice().sort(); expect(a, `${table} columns drifted between the fresh and migrated paths`).toEqual(b); diff --git a/apps/api/test/watcher-loop.test.ts b/apps/api/test/watcher-loop.test.ts index 9ae4edde5..13c6daf9c 100644 --- a/apps/api/test/watcher-loop.test.ts +++ b/apps/api/test/watcher-loop.test.ts @@ -71,10 +71,14 @@ function makeFakeStateRepo() { cursors.set(account, cursor); setCursorCalls.push(cursor); }, - async isProcessed(txHash: string): Promise { + // Keyed by txHash alone here (not the real per-operation key) — this file + // tests backlog draining/pagination, not issue 4.11's multi-operation + // dedup, and every payment in it has a unique txHash, so this stays + // behaviorally equivalent while satisfying the WatcherStateRepository shape. + async isProcessed(txHash: string, _operationId: string): Promise { return processed.has(txHash); }, - async markProcessed(txHash: string): Promise { + async markProcessed(txHash: string, _operationId: string): Promise { processed.add(txHash); }, }; diff --git a/apps/api/test/worker/watcher-loop.test.ts b/apps/api/test/worker/watcher-loop.test.ts index 16cc5bcba..f2c751096 100644 --- a/apps/api/test/worker/watcher-loop.test.ts +++ b/apps/api/test/worker/watcher-loop.test.ts @@ -6,6 +6,7 @@ import type { DrizzleLinkRepository, DrizzleSellerRepository, DrizzleWebhookRepo import { DrizzleOffRampStateRepository } from "../../src/repos/index"; import { NoKycRequired } from "@checkout/offramp"; import { FakeTelemetryRepository } from "../fakes"; +import type { NormalizedPayment, WatcherPort } from "@checkout/core"; // --------------------------------------------------------------------------- // WatcherLoop tests @@ -150,7 +151,7 @@ describe("WatcherLoop", () => { const ref = "idem_ref_1"; await createActiveLink(ref); const txHash = "tx_idem_1"; - await stateRepo.markProcessed(txHash, null); + await stateRepo.markProcessed(txHash, "400", null); await stateRepo.setCursor(DEST, "399"); watcher.setPayments([ @@ -248,7 +249,7 @@ describe("WatcherLoop", () => { await loop.runOnce(); - const processed = await stateRepo.isProcessed("tx_crash_1"); + const processed = await stateRepo.isProcessed("tx_crash_1", "801"); expect(processed).toBe(true); const cursor = await stateRepo.getCursor(DEST); @@ -260,7 +261,7 @@ describe("WatcherLoop", () => { await createActiveLink(ref); const txHash = "tx_crash_2"; - await stateRepo.markProcessed(txHash, `lnk_${ref}`); + await stateRepo.markProcessed(txHash, "901", `lnk_${ref}`); await stateRepo.setCursor(DEST, "900"); watcher.setPayments([ @@ -344,7 +345,7 @@ describe("WatcherLoop", () => { await loop.runOnce(); - const processed = await stateRepo.isProcessed("tx_unknown_1"); + const processed = await stateRepo.isProcessed("tx_unknown_1", "1201"); expect(processed).toBe(true); }); }); @@ -454,3 +455,183 @@ describe("WatcherLoop — crash between markProcessed and setCursor", () => { await repos.client.close(); }); }); + +// --------------------------------------------------------------------------- +// Operation-level dedup (issue 4.11) +// +// A Stellar transaction can carry up to 100 operations, and a payment is one +// operation, not the whole transaction. The old `processed_tx` (and +// `link_payments`) dedup keyed on tx_hash alone, so a second payment +// operation sharing a transaction with the first was discarded as a false +// duplicate — permanently: there's no re-processing path once a hash is +// marked done. +// --------------------------------------------------------------------------- + +/** Routes scripted payments per destination, unlike `FakeWatcherPort` (whose + * single shared queue would hand every payment to whichever account's tick + * asks first) — needed to exercise two destinations settling within the + * same `runOnce()`. */ +class RoutedFakeWatcherPort implements WatcherPort { + private byAccount = new Map(); + + setPaymentsFor(account: string, payments: NormalizedPayment[]): void { + this.byAccount.set(account, payments); + } + + async latestCursor(_account: string): Promise { + return "100"; + } + + async fetchSince(account: string, _cursor: string, _limit?: number): Promise { + const result = this.byAccount.get(account) ?? []; + this.byAccount.set(account, []); + return result; + } +} + +describe("WatcherLoop — operation-level dedup (issue 4.11)", () => { + it("a two-operation transaction paying one link settles it at the full amount", async () => { + const repos = await withTestDb(); + const watcher = new FakeWatcherPort(); + const rail = new FakeRailPort(); + const offramp = new FakeOffRampPort(); + + const service = new LinkService({ + links: repos.links, + sellers: repos.sellers, + webhooks: repos.webhooks, + rail, + offramp, + offrampState: new DrizzleOffRampStateRepository(repos.db), + kyc: new NoKycRequired(), + stellar: testStellarConfig, + telemetry: new FakeTelemetryRepository(), + correlation: "memo", + webhookGuard: async () => ({ ok: true }) as const, + }); + + const ref = "split_ref_1"; + const seller = await repos.sellers.getDefault(); + await repos.links.create({ + id: `lnk_${ref}`, + reference: ref, + sellerId: seller.id, + destination: DEST, + muxedId: null, + title: "Split payment", + amount: "10", + asset: { code: "USDC", issuer: ISSUER }, + expiresAt: null, + }); + + await repos.state.setCursor(DEST, "2000"); + + // A wallet splitting one payment across two operations in one atomic + // transaction: same tx_hash, two distinct pagingTokens (Horizon's own + // per-operation id). + const txHash = "tx_split_1"; + watcher.setPayments([ + FakeWatcherPort.payment({ txHash, pagingToken: "2001", memo: ref, amount: "6" }), + FakeWatcherPort.payment({ txHash, pagingToken: "2002", memo: ref, amount: "4" }), + ]); + + const loop = new WatcherLoop({ + watcher, + links: repos.links, + state: repos.state, + service, + pollMs: 60_000, + }); + await loop.runOnce(); + + const link = await repos.links.findByReference(ref); + expect(link!.status).toBe("paid"); + expect(link!.paidAmount).toBe("10"); // both operations credited, not just the first + + // Each operation dedupes independently by its own pagingToken, not by the + // shared tx_hash — replaying either one alone is still recognized as + // already processed. + expect(await repos.state.isProcessed(txHash, "2001")).toBe(true); + expect(await repos.state.isProcessed(txHash, "2002")).toBe(true); + + await repos.client.close(); + }); + + it("one transaction paying two different sellers' destinations settles both links", async () => { + const repos = await withTestDb(); + const watcher = new RoutedFakeWatcherPort(); + const rail = new FakeRailPort(); + const offramp = new FakeOffRampPort(); + + const service = new LinkService({ + links: repos.links, + sellers: repos.sellers, + webhooks: repos.webhooks, + rail, + offramp, + offrampState: new DrizzleOffRampStateRepository(repos.db), + kyc: new NoKycRequired(), + stellar: testStellarConfig, + telemetry: new FakeTelemetryRepository(), + correlation: "memo", + webhookGuard: async () => ({ ok: true }) as const, + }); + + const seller = await repos.sellers.getDefault(); + const DEST_B = "GDESTBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB"; + + await repos.links.create({ + id: "lnk_multi_a", + reference: "multi_ref_a", + sellerId: seller.id, + destination: DEST, + muxedId: null, + title: "Batched payout — seller A", + amount: "10", + asset: { code: "USDC", issuer: ISSUER }, + expiresAt: null, + }); + await repos.links.create({ + id: "lnk_multi_b", + reference: "multi_ref_b", + sellerId: seller.id, + destination: DEST_B, + muxedId: null, + title: "Batched payout — seller B", + amount: "20", + asset: { code: "USDC", issuer: ISSUER }, + expiresAt: null, + }); + + await repos.state.setCursor(DEST, "3000"); + await repos.state.setCursor(DEST_B, "3000"); + + // Ordinary fee-saving batching: one transaction, two operations, two + // different watched destinations — same tx_hash both times. + const txHash = "tx_batch_1"; + watcher.setPaymentsFor(DEST, [ + FakeWatcherPort.payment({ txHash, pagingToken: "3001", to: DEST, memo: "multi_ref_a", amount: "10" }), + ]); + watcher.setPaymentsFor(DEST_B, [ + FakeWatcherPort.payment({ txHash, pagingToken: "3002", to: DEST_B, memo: "multi_ref_b", amount: "20" }), + ]); + + const loop = new WatcherLoop({ + watcher, + links: repos.links, + state: repos.state, + service, + pollMs: 60_000, + }); + await loop.runOnce(); + + const linkA = await repos.links.findByReference("multi_ref_a"); + const linkB = await repos.links.findByReference("multi_ref_b"); + // Old bug: whichever destination's tick reached the hash first claimed it + // system-wide, so the other seller's operation was invisibly dropped. + expect(linkA!.status).toBe("paid"); + expect(linkB!.status).toBe("paid"); + + await repos.client.close(); + }); +}); diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 45d2b50fb..f7ad40aa5 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -146,14 +146,16 @@ sequenceDiagram LS->>LS: canTransition() guard, then links.save() LS->>Hooks: fireWebhook("link.paid" | "link.underpaid") end - Loop->>Loop: state.markProcessed(txHash), state.setCursor(account, lastToken) + Loop->>Loop: state.markProcessed(txHash, operationId), state.setCursor(account, lastToken) end ``` Idempotency is layered on purpose: the persisted **cursor** avoids refetching old operations, the **processed-tx ledger** guards the crash window before a cursor is saved, and the domain's `canTransition()` guard means a duplicate payment can never double-apply -even if both of the above somehow let it through. +even if both of the above somehow let it through. The processed-tx ledger keys on +`(txHash, operationId)`, not `txHash` alone — a transaction can carry more than one +payment operation, and each dedupes independently (issue 4.11). ### 3. Cash-out — SEP-10 → SEP-38 → SEP-6 (`TestAnchorOffRamp`, today's real adapter) diff --git a/docs/FIXLOG.md b/docs/FIXLOG.md index ebb4c0d3e..b33004955 100644 --- a/docs/FIXLOG.md +++ b/docs/FIXLOG.md @@ -37,6 +37,7 @@ Columns: | 2026-08-06 | BUG-8.10 | `turbo run test` intermittently failed two `packages/core` fast-check property tests with `Error: Test timed out in 5000ms` — a random red build with no counterexample and no code change behind it | `packages/core/vitest.config.ts` set no `testTimeout`, so it used vitest's 5s default. The property suites take ~1.2–1.7s in isolation but exceed 5s when all six package suites run concurrently under turbo. `apps/api/vitest.config.ts` already carried this exact fix and its rationale; core never got it | `01c5d72` | `packages/core/vitest.config.ts` — `testTimeout`/`hookTimeout` raised to 10s, matching apps/api. Verified by reproducing the failure deterministically with `turbo run test --force`, then three consecutive clean loaded runs | | 2026-08-06 | BUG-6.6 | `GET /seller/kyc` and `PUT /seller/kyc` both returned `200` to a caller sending no credentials of any kind — verified against a running instance. On the deployed configuration (`OFFRAMP=testanchor`) the GET served the seller's decrypted SEP-12 identity to anyone on the internet, and the PUT let anyone overwrite it and submit it to the live anchor | `routes/kyc.ts` was mounted in `index.ts` with no auth middleware, and resolved the seller with `sellers.getDefault()` rather than from a token. Every other seller route had been moved behind `buildAuthMiddleware` when wallet-native auth landed; this one was missed, so the AES-256-GCM at-rest encryption in `crypto/pii.ts` was protecting data that an unauthenticated GET then handed over in plaintext | `5a01d02` | `apps/api/test/kyc-route-auth.test.ts` · *"refuses an unauthenticated read of the seller's identity"* (also asserts the PII never appears in the response body), *"refuses an unauthenticated write of the seller's identity"*, *"refuses an authenticated key that lacks offramp:initiate"*, *"serves the authenticated seller, resolved from the token rather than getDefault()"* | | 2026-08-06 | BUG-8.11 | `turbo run test` intermittently failed one of two `anchor-health` circuit-breaker tests with `AssertionError: expected 0 to be greater than 0`, or a false `isAvailable()` — a different test each time, with no code change behind it | `snapshot()` and `isAvailable()` both call `tickState()`, which auto-promotes `open → half_open` once `cooldownMs` has elapsed. The two tests asserted the breaker was *still open* while configured with 5ms and 50ms cooldowns — windows shorter than ordinary scheduler jitter — so under concurrent suite execution the promotion won the race. One test expressed its assertion as a `Date.now()` conditional, which turned the failure into an opaque `0 > 0` | `dbaa194` | `apps/api/test/anchor-health.test.ts` — both cooldowns raised to a shared 400ms constant and the `Date.now()` conditional replaced with a direct state assertion plus an explicit `half_open` step. Verified by reproducing under CPU contention, then three consecutive clean loaded runs | +| 2026-08-28 | BUG-152 | A wallet that split one payment across two operations in the same transaction, to the same watched destination, saw the link stuck `underpaid` forever even though the full amount arrived on-chain. Worse: one transaction paying two different watched sellers' destinations (ordinary fee-saving batching) left whichever seller's poll tick lost the race with `paid` never reached, no webhook fired, and the payment fully verifiable on-chain yet invisible to the app — no re-processing path existed once a hash was marked done | `processed_tx` (dedup) and `link_payments` (the cumulative-accounting ledger, issue 1.4) both keyed uniqueness on `tx_hash` alone. A Stellar transaction can carry up to 100 operations, and a payment is one operation, not the whole transaction — the second operation sharing a transaction with the first was discarded as a false duplicate of it, or silently dropped by `onConflictDoNothing`, in both tables | `9650020` | `apps/api/test/worker/watcher-loop.test.ts` · *"a two-operation transaction paying one link settles it at the full amount"* and *"one transaction paying two different sellers' destinations settles both links"* — both fail against the old `tx_hash`-only keying, crediting only the first operation and dropping the second (or the second seller's link) entirely | --- diff --git a/packages/core/src/ports/index.ts b/packages/core/src/ports/index.ts index 027b09740..ed70403f4 100644 --- a/packages/core/src/ports/index.ts +++ b/packages/core/src/ports/index.ts @@ -501,11 +501,16 @@ export interface CreateLinkInput { } /** One incoming payment recorded against a link — the authoritative ledger - * row cumulative accounting sums over (issue 1.4). `txHash` is unique so a - * reprocessed payment can never double-count. */ + * row cumulative accounting sums over (issue 1.4). Unique on + * `(txHash, operationId)`, not `txHash` alone (issue 4.11): a transaction + * can carry more than one payment operation to the same link (a split + * payment), and each must be recorded, not dropped as a false duplicate of + * the other. A reprocessed *operation* still never double-counts. */ export interface LinkPaymentRecord { linkId: string; txHash: string; + /** The chain's per-operation identifier (Horizon's `pagingToken`). */ + operationId: string; payer: string; amount: string; asset: AssetRef; @@ -535,8 +540,10 @@ export interface LinkRepository { /** Active (or underpaid) links whose value lands in `destination`. */ openLinksForDestination(destination: string): Promise; save(link: PaymentLink): Promise; - /** Append a payment to the link's ledger. A duplicate `txHash` is a no-op — - * cumulative accounting must never double-count a reprocessed payment. */ + /** Append a payment to the link's ledger. A duplicate `(txHash, operationId)` + * is a no-op — cumulative accounting must never double-count a reprocessed + * operation. Two *different* operations that happen to share a `txHash` + * (a split payment) are two rows, not one (issue 4.11). */ recordPayment(payment: LinkPaymentRecord): Promise; /** Sum of every payment ever recorded for this link, as a decimal string * ("0" if none). The authoritative source `paidAmount` is cached from. */ @@ -632,12 +639,22 @@ export interface WebhookRepository { listDeliveriesByLinkId(linkId: string): Promise; } -/** Watcher bookkeeping: per-account cursor + processed-tx ledger for idempotency. */ +/** + * Watcher bookkeeping: per-account cursor + processed-payment ledger for + * idempotency. + * + * Keyed by operation, not transaction (issue 4.11): a Stellar transaction can + * carry up to 100 operations, and a payment is one operation, not the whole + * transaction. `operationId` is the chain's per-operation identifier (Horizon's + * `pagingToken` for Stellar) — two payment operations sharing one `txHash` + * (a split payment, or a batch that happens to pay two different watched + * destinations) must dedupe independently, not collide on the shared hash. + */ export interface WatcherStateRepository { getCursor(account: string): Promise; setCursor(account: string, cursor: string): Promise; - isProcessed(txHash: string): Promise; - markProcessed(txHash: string, linkId: string | null): Promise; + isProcessed(txHash: string, operationId: string): Promise; + markProcessed(txHash: string, operationId: string, linkId: string | null): Promise; } /** Session-JWT revocation, keyed by the token's own `jti` — logout and