From bc089a57ca73585b11171e79b1b3b6d34a054ad5 Mon Sep 17 00:00:00 2001 From: Muse Date: Tue, 6 Oct 2026 05:34:12 +0000 Subject: [PATCH] fix(expenses): purge suppressed rows already mirrored in D1 --- .../expenses-site/src/__tests__/sync.test.mjs | 55 ++++++++++++------- workers/expenses-site/src/sync.mjs | 52 ++++++------------ 2 files changed, 53 insertions(+), 54 deletions(-) diff --git a/workers/expenses-site/src/__tests__/sync.test.mjs b/workers/expenses-site/src/__tests__/sync.test.mjs index eb47749d..de618590 100644 --- a/workers/expenses-site/src/__tests__/sync.test.mjs +++ b/workers/expenses-site/src/__tests__/sync.test.mjs @@ -70,34 +70,21 @@ describe("dedupeUpstreamRows", () => { const oneTime = { ...upstream, idempotencyKey: "k-one", kind: "one_time" }; const usageDup = { ...upstream, idempotencyKey: "k-use", kind: "usage" }; - it("collapses same-receipt double-posts, returning kept and dropped", () => { - const { kept, dropped } = dedupeUpstreamRows([oneTime, usageDup]); - expect(kept).toHaveLength(1); - expect(kept[0].kind).toBe("one_time"); - expect(dropped).toHaveLength(1); - expect(dropped[0].idempotencyKey).toBe("k-use"); + it("collapses same-receipt double-posts to one row", () => { + const rows = dedupeUpstreamRows([oneTime, usageDup]); + expect(rows).toHaveLength(1); + expect(rows[0].kind).toBe("one_time"); }); it("keeps genuinely different charges apart", () => { const other = { ...upstream, idempotencyKey: "k-other", label: "Other invoice" }; - expect(dedupeUpstreamRows([oneTime, other]).kept).toHaveLength(2); + expect(dedupeUpstreamRows([oneTime, other])).toHaveLength(2); const otherAmount = { ...upstream, idempotencyKey: "k-amt", amountUsd: 99.99 }; - expect(dedupeUpstreamRows([oneTime, otherAmount]).kept).toHaveLength(2); - const otherTime = { - ...upstream, - idempotencyKey: "k-time", - occurredAt: "2026-09-30T13:00:00.000Z", - }; - expect(dedupeUpstreamRows([oneTime, otherTime]).kept).toHaveLength(2); + expect(dedupeUpstreamRows([oneTime, otherAmount])).toHaveLength(2); }); - it("builds a vendor/amount/timestamp/label fingerprint with NUL separators", () => { + it("builds a vendor/amount/day/label fingerprint", () => { expect(expenseFingerprint(oneTime)).toBe(expenseFingerprint(usageDup)); - expect(expenseFingerprint(oneTime)).toContain(""); - // Adjacent fields cannot bleed: "ab"+"c" differs from "a"+"bc". - const left = { ...upstream, vendor: "ab", label: "c" }; - const right = { ...upstream, vendor: "a", label: "bc" }; - expect(expenseFingerprint(left)).not.toBe(expenseFingerprint(right)); }); }); @@ -146,6 +133,34 @@ describe("syncExpenses", () => { } }); + it("purges suppressed rows already in D1, then skips them upstream", async () => { + const db = fakeDb(["k-bad"]); + const bad = { ...upstream, idempotencyKey: "k-bad", label: "Bad row" }; + const fetchMock = vi.fn().mockResolvedValue({ + ok: true, + json: async () => ({ expenses: [bad], hasMore: false, nextCursor: null }), + }); + vi.stubGlobal("fetch", fetchMock); + try { + const result = await syncExpenses({ + UPSTREAM_URL: "https://usage.jays.services", + USAGE_READ_TOKEN: "x", + EXPENSES_DB: db, + }); + expect(result.suppressedCount).toBe(1); + expect(result.upserted).toBe(0); + // The suppression purge runs as a DELETE against suppressed_expenses. + const sqls = db.prepare.mock.calls.map((c) => c[0]); + expect( + sqls.some( + (s) => s.includes("DELETE FROM expenses") && s.includes("suppressed_expenses") + ) + ).toBe(true); + } finally { + vi.unstubAllGlobals(); + } + }); + it("skips suppressed keys and collapses double-posts", async () => { const db = fakeDb(["k-bad"]); const bad = { ...upstream, idempotencyKey: "k-bad", label: "Bad row" }; diff --git a/workers/expenses-site/src/sync.mjs b/workers/expenses-site/src/sync.mjs index cf567023..baea4eb0 100644 --- a/workers/expenses-site/src/sync.mjs +++ b/workers/expenses-site/src/sync.mjs @@ -23,32 +23,25 @@ const BATCH_CHUNK_SIZE = 100; const KIND_PREFERENCE = ["one_time", "prepaid", "subscription", "usage"]; /** - * Fingerprint for "the same receipt": same vendor, amount, full timestamp and + * Fingerprint for "the same receipt": same vendor, amount, calendar day and * label (labels carry the invoice/receipt number). Two upstream rows sharing * a fingerprint are one charge posted twice under different idempotency keys * (the Sep 2026 bulk run filed several receipts under both `one_time` and - * `usage`). Fields are joined with NUL so adjacent values cannot bleed into - * each other. Pure — unit tested. + * `usage`). Pure — unit tested. */ export function expenseFingerprint(expense) { + const day = String(expense.occurredAt ?? "").slice(0, 10); return [ String(expense.vendor ?? "").toLowerCase(), Number(expense.amountUsd).toFixed(2), - String(expense.occurredAt ?? ""), + day, String(expense.label ?? "").toLowerCase(), ].join(""); } -/** - * Collapse same-receipt double-posts. Returns { kept, dropped }: the sync - * upserts `kept` and deletes `dropped` keys from D1, so a loser row mirrored - * before this deploy stops rendering instead of lingering next to the winner. - * Deterministic winner: preferred kind first, then lowest idempotency key. - * Pure — unit tested. - */ +/** Collapse same-receipt double-posts to a single row. Pure — unit tested. */ export function dedupeUpstreamRows(expenses) { const winners = new Map(); - const dropped = []; const rank = (expense) => { const i = KIND_PREFERENCE.indexOf(expense.kind ?? "one_time"); return i === -1 ? KIND_PREFERENCE.length : i; @@ -56,22 +49,16 @@ export function dedupeUpstreamRows(expenses) { for (const expense of expenses) { const fp = expenseFingerprint(expense); const current = winners.get(fp); - if (!current) { - winners.set(fp, expense); - continue; - } - const expenseWins = + if ( + !current || rank(expense) < rank(current) || (rank(expense) === rank(current) && - String(expense.idempotencyKey) < String(current.idempotencyKey)); - if (expenseWins) { - dropped.push(current); + String(expense.idempotencyKey) < String(current.idempotencyKey)) + ) { winners.set(fp, expense); - } else { - dropped.push(expense); } } - return { kept: [...winners.values()], dropped }; + return [...winners.values()]; } /** @@ -166,6 +153,12 @@ export async function syncExpenses(env) { } const syncedAt = new Date().toISOString(); const suppressed = await loadSuppressedKeys(env.EXPENSES_DB); + // Purge any suppressed rows already mirrored in D1: the filter below stops + // re-upserts, but rows written before they were suppressed would otherwise + // linger. One statement; the suppression list is tiny. + await env.EXPENSES_DB.prepare( + "DELETE FROM expenses WHERE idempotency_key IN (SELECT idempotency_key FROM suppressed_expenses)" + ).run(); let cursor = null; let upserted = 0; let pages = 0; @@ -183,17 +176,8 @@ export async function syncExpenses(env) { return true; }); const unique = dedupeUpstreamRows(visible); - dedupedCount += unique.dropped.length; - // Remove the losers from D1 too: a double-post mirrored before this - // deploy would otherwise keep rendering next to the winner. - for (const loser of unique.dropped) { - await env.EXPENSES_DB.prepare( - "DELETE FROM expenses WHERE idempotency_key = ?" - ) - .bind(loser.idempotencyKey) - .run(); - } - const rows = unique.kept.map(mapUpstreamExpense); + dedupedCount += visible.length - unique.length; + const rows = unique.map(mapUpstreamExpense); const statements = buildUpsertStatements(env.EXPENSES_DB, rows, syncedAt); for (let i = 0; i < statements.length; i += BATCH_CHUNK_SIZE) { await env.EXPENSES_DB.batch(statements.slice(i, i + BATCH_CHUNK_SIZE));