Key duplicate-payment protection by operation, not by transaction

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<void> {
   }
 }
 
+/**
+ * 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<void> {
+  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<void> {
+  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<void> {
   await migrateLegacyWebhooksTable(client);
+  await migrateLegacyProcessedTxTable(client);
+  await migrateLegacyLinkPaymentsTable(client);
   for (const sql of BOOTSTRAP_SQL) {
     try {
       await client.execute(sql);

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

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,14 +268,15 @@ 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,
         assetIssuer: payment.asset.issuer,
         ledger: payment.ledger,
         createdAt: payment.createdAt,
       })
-      .onConflictDoNothing({ target: linkPayments.txHash });
+      .onConflictDoNothing({ target: [linkPayments.txHash, linkPayments.operationId] });
   }
 
   async paymentLedger(txHash: string): Promise<number | null> {
@@ -551,20 +552,27 @@ export class DrizzleWatcherStateRepository implements WatcherStateRepository {
       });
   }
 
-  async isProcessed(txHash: string): Promise<boolean> {
+  async isProcessed(txHash: string, operationId: string): Promise<boolean> {
     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<void> {
+  async markProcessed(txHash: string, operationId: string, linkId: string | null): Promise<void> {
     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] });
   }
 }
 

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,

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;

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);

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<boolean> {
+    // 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<boolean> {
       return processed.has(txHash);
     },
-    async markProcessed(txHash: string): Promise<void> {
+    async markProcessed(txHash: string, _operationId: string): Promise<void> {
       processed.add(txHash);
     },
   };