diff --git a/.env.example b/.env.example index 24053240..012a056b 100644 --- a/.env.example +++ b/.env.example @@ -342,6 +342,9 @@ PAID_NOTICE_INTERVAL_MS=30000 # How often the chain worker sends referral bounties; 0 disables it. The operator # key lives only in the worker, so core-api never runs this. REFERRAL_INTERVAL_MS=5000 +# Days after expiry that an unpaid payment is deleted (analytics keeps counting it). +# Never touches a payment with funds, or a FAILED one. 0 turns pruning off. +EXPIRED_PAYMENT_RETENTION_DAYS=2 # ============================================================================= # Quote layer — venue pricing, oracle deviation guard, order signing diff --git a/apps/api/src/chain-worker.ts b/apps/api/src/chain-worker.ts index ed2beef8..90cc5cc1 100644 --- a/apps/api/src/chain-worker.ts +++ b/apps/api/src/chain-worker.ts @@ -233,6 +233,30 @@ if (referralTimer !== undefined) { console.log(`[chain-worker] paying referral bounties every ${config.referralIntervalMs}ms`); } +// Unpaid payments past their retention are deleted; analytics keeps their count. An hour +// is plenty: the batch is small, and a backlog drains one batch per pass. +const PRUNE_INTERVAL_MS = 60 * 60_000; +const PRUNE_BATCH = 100; +const DAY_MS = 24 * 60 * 60_000; +let pruning = false; +const pruneTimer = + config.expiredPaymentRetentionDays <= 0 + ? undefined + : setInterval(() => { + if (pruning) return; + pruning = true; + const cutoff = new Date(Date.now() - config.expiredPaymentRetentionDays * DAY_MS); + void container.expiredPayments + .prune(cutoff, PRUNE_BATCH) + .then((pruned) => { + if (pruned > 0) console.log(`[chain-worker] pruned ${pruned} unpaid payments`); + }) + .catch((error: unknown) => console.error("[chain-worker] prune failed", error)) + .finally(() => { + pruning = false; + }); + }, PRUNE_INTERVAL_MS); + // Funds that reached a deposit address after its payment expired: money with no // payment to settle. Reported once per payment per process. const LATE_DEPOSIT_INTERVAL_MS = 5 * 60_000; @@ -346,6 +370,7 @@ const shutdown = async (signal: NodeJS.Signals): Promise => { clearInterval(operatorFloatTimer); clearInterval(gasRefillTimer); clearInterval(referralTimer); + clearInterval(pruneTimer); clearInterval(resumeTimer); clearInterval(stuckPaymentTimer); if (depositReturnTimer !== undefined) clearInterval(depositReturnTimer); diff --git a/apps/api/src/config.ts b/apps/api/src/config.ts index 32b7775a..8a9b352b 100644 --- a/apps/api/src/config.ts +++ b/apps/api/src/config.ts @@ -351,6 +351,13 @@ const configSchema = z.object({ * to one instance (the cursor is read, then written). */ referralIntervalMs: z.coerce.number().int().min(0).default(5_000), + /** + * Days after its expiry that an unpaid payment (expired, or never confirmed) is + * deleted. Two is past the watcher's 24 hours, so a late deposit is still found + * first. A payment with any sign of funds is never deleted, and `FAILED` is kept. + * `0` turns pruning off. Only the chain worker runs it. + */ + expiredPaymentRetentionDays: z.coerce.number().int().min(0).default(2), // --- Quote layer (RFC #6/#7, wired in the container) ------------------- /** @@ -1520,6 +1527,7 @@ export function loadConfig( webhookIntervalMs: env.WEBHOOK_INTERVAL_MS, paidNoticeIntervalMs: env.PAID_NOTICE_INTERVAL_MS, referralIntervalMs: env.REFERRAL_INTERVAL_MS, + expiredPaymentRetentionDays: env.EXPIRED_PAYMENT_RETENTION_DAYS, quoteEnabled: env.QUOTE_ENABLED, quoteVenues: env.QUOTE_VENUES, quoteOracle: env.QUOTE_ORACLE, diff --git a/apps/api/src/container.ts b/apps/api/src/container.ts index f3bcd01f..8567000e 100644 --- a/apps/api/src/container.ts +++ b/apps/api/src/container.ts @@ -44,6 +44,7 @@ import { DrizzleDepositAddressRepository, DrizzleDepositRepository, DrizzleEscrowDisputeRepository, + DrizzleExpiredPaymentPruner, DrizzleInternalSettlementStore, DrizzleInvoiceRepository, DrizzleLedgerRepository, @@ -89,6 +90,7 @@ import { } from "@mayarin/notifications"; import { DerivedRailCatalog, + type ExpiredPaymentPruner, type MerchantAssetPolicySource, PaymentIntentService, type RailCatalog, @@ -230,6 +232,8 @@ export interface Container { readonly paymentApp: PaymentAppService; /** The dashboard's analytics figures, for the merchant API. */ readonly analytics: AnalyticsService; + /** Deletes abandoned payments after the retention window; the chain worker runs it. */ + readonly expiredPayments: ExpiredPaymentPruner; /** Refunds against settled payments (#12). */ readonly refunds: RefundService; /** Closes the operator key's sent-transaction rows once the chain settles them (SEC-04). */ @@ -930,6 +934,7 @@ export function createContainer({ }); const intentRepository = new DrizzlePaymentIntentRepository(handle.db); + const expiredPayments = new DrizzleExpiredPaymentPruner(handle.db); const intents = new PaymentIntentService({ repository: intentRepository, clock, @@ -1629,8 +1634,10 @@ export function createContainer({ analytics: new AnalyticsService({ intents: intentRepository, clearing: clearingRepository, + pruned: expiredPayments, clock, }), + expiredPayments, escrow: new EscrowService(paymentApp, config.chainRpcUrls, disputes, clock, disputeNotices), payerStanding, riskAssessments, diff --git a/apps/api/src/services/analytics.ts b/apps/api/src/services/analytics.ts index 73c13769..08fb81b2 100644 --- a/apps/api/src/services/analytics.ts +++ b/apps/api/src/services/analytics.ts @@ -10,9 +10,11 @@ import type { ClearingTransaction } from "@mayarin/clearing"; import { + type ExpiredPaymentPruner, isExpired, type PaymentIntent, type PaymentIntentRepository, + type PrunedPaymentCount, } from "@mayarin/payment-intent"; import { assetDecimals, @@ -180,6 +182,7 @@ const emptyBucket = (): Bucket => ({ function gather( payments: readonly PaymentIntent[], byIntent: ReadonlyMap, + pruned: readonly PrunedPaymentCount[], range: Range, interval: AnalyticsInterval, ): ReadonlyMap { @@ -198,6 +201,15 @@ function gather( if (elapsed !== undefined) bucket.completion.push(elapsed); } + // A pruned payment was an abandoned checkout: one of each, as if it were still here. + for (const row of pruned) { + if (!inRange(new Date(timeOf(row.day)), range)) continue; + const bucket = buckets.get(bucketOf(row.day, interval)); + if (bucket === undefined) continue; + bucket.checkouts += row.count; + bucket.failed += row.count; + } + for (const intent of payments) { const tx = byIntent.get(intent.id); if (tx === undefined || !isSettled(tx)) continue; @@ -304,6 +316,7 @@ function shares(keys: readonly string[]): readonly AnalyticsShareDto[] { function performanceBy( payments: readonly PaymentIntent[], keyOf: (intent: PaymentIntent) => string, + prunedBad: ReadonlyMap, ): readonly AnalyticsPerformanceDto[] { const groups = new Map(); for (const intent of payments) { @@ -318,6 +331,11 @@ function performanceBy( } else group.bad += 1; groups.set(key, group); } + for (const [key, count] of prunedBad) { + const group = groups.get(key) ?? { ok: 0, bad: 0, seconds: [] }; + group.bad += count; + groups.set(key, group); + } return [...groups] .map(([key, group]) => ({ key, @@ -337,15 +355,36 @@ function payerCountry(intent: PaymentIntent): string { /* Report */ /* -------------------------------------------------------------------------- */ +/** Pruned payments summed per key; a payer who never chose a rail is "Unknown". */ +function prunedBy( + pruned: readonly PrunedPaymentCount[], + keyOf: (row: PrunedPaymentCount) => string | undefined, +): ReadonlyMap { + const totals = new Map(); + for (const row of pruned) { + const key = keyOf(row) ?? "Unknown"; + totals.set(key, (totals.get(key) ?? 0) + row.count); + } + return totals; +} + export interface AnalyticsInput { /** Every payment of the merchant, expiry already applied. */ readonly payments: readonly PaymentIntent[]; + /** What was deleted as abandoned; counted as failed and expired. */ + readonly pruned: readonly PrunedPaymentCount[]; readonly clearing: readonly ClearingTransaction[]; readonly query: AnalyticsQuery; readonly now: Date; } -export function buildAnalytics({ payments, clearing, query, now }: AnalyticsInput): AnalyticsDto { +export function buildAnalytics({ + payments, + pruned, + clearing, + query, + now, +}: AnalyticsInput): AnalyticsDto { const range = rangeOf(query, now); const { interval } = query; const filtered = payments.filter( @@ -354,9 +393,14 @@ export function buildAnalytics({ payments, clearing, query, now }: AnalyticsInpu (query.asset === undefined || intent.payment?.asset === query.asset), ); const byIntent = new Map(clearing.map((tx) => [tx.paymentIntentId, tx])); + const keptCounts = pruned.filter( + (row) => + (query.chain === undefined || row.chain === query.chain) && + (query.asset === undefined || row.asset === query.asset), + ); - const current = gather(filtered, byIntent, range, interval); - const previous = fold(gather(filtered, byIntent, previousRange(range), interval)); + const current = gather(filtered, byIntent, keptCounts, range, interval); + const previous = fold(gather(filtered, byIntent, keptCounts, previousRange(range), interval)); const whole = fold(current); const keys = query.metrics ?? ANALYTICS_METRICS; @@ -375,6 +419,7 @@ export function buildAnalytics({ payments, clearing, query, now }: AnalyticsInpu const inWindow = filtered.filter((intent) => inRange(intent.createdAt, range)); const completed = inWindow.filter((intent) => intent.status === "COMPLETED"); + const prunedInWindow = keptCounts.filter((row) => inRange(new Date(timeOf(row.day)), range)); return { currency: "USD", range, @@ -386,8 +431,16 @@ export function buildAnalytics({ payments, clearing, query, now }: AnalyticsInpu countries: shares(completed.map(payerCountry)), }, performance: { - networks: performanceBy(inWindow, (intent) => intent.payment?.chain ?? "Unknown"), - assets: performanceBy(inWindow, (intent) => intent.payment?.asset ?? "Unknown"), + networks: performanceBy( + inWindow, + (intent) => intent.payment?.chain ?? "Unknown", + prunedBy(prunedInWindow, (row) => row.chain), + ), + assets: performanceBy( + inWindow, + (intent) => intent.payment?.asset ?? "Unknown", + prunedBy(prunedInWindow, (row) => row.asset), + ), }, }; } @@ -395,15 +448,18 @@ export function buildAnalytics({ payments, clearing, query, now }: AnalyticsInpu export class AnalyticsService { readonly #intents: PaymentIntentRepository; readonly #clearing: AnalyticsClearingReader; + readonly #pruned: Pick; readonly #clock: Clock; constructor(options: { readonly intents: PaymentIntentRepository; readonly clearing: AnalyticsClearingReader; + readonly pruned: Pick; readonly clock: Clock; }) { this.#intents = options.intents; this.#clearing = options.clearing; + this.#pruned = options.pruned; this.#clock = options.clock; } @@ -429,6 +485,7 @@ export class AnalyticsService { isExpired(intent, now) ? { ...intent, status: "EXPIRED" as const } : intent, ); const clearing = await this.#clearing.listByPaymentIntentIds(viewed.map((intent) => intent.id)); - return buildAnalytics({ payments: viewed, clearing, query, now }); + const pruned = await this.#pruned.prunedCounts(merchantId); + return buildAnalytics({ payments: viewed, pruned, clearing, query, now }); } } diff --git a/apps/api/test/analytics.test.ts b/apps/api/test/analytics.test.ts index 7173b575..f3a15ec8 100644 --- a/apps/api/test/analytics.test.ts +++ b/apps/api/test/analytics.test.ts @@ -39,7 +39,7 @@ describe("buildAnalytics", () => { intent("c", at(3), { status: "FAILED", unfinished: true }), intent("d", at(3), { status: "CREATED", unfinished: true }), ]; - const report = buildAnalytics({ payments, clearing: [], query: query(), now: NOW }); + const report = buildAnalytics({ payments, pruned: [], clearing: [], query: query(), now: NOW }); test("revenue sums completed payments per day and zero-fills the rest", () => { expect(metric(report, "revenue")?.points.map((point) => point.value)).toEqual([ @@ -58,8 +58,37 @@ describe("buildAnalytics", () => { expect(metric(report, "time_to_pay_p90")?.total).toBe(30); }); + test("a pruned payment still counts as a failed checkout", () => { + const pruned = [ + { day: "2026-10-03", chain: undefined, asset: undefined, count: 4 }, + { day: "2026-10-03", chain: "base-sepolia", asset: "USDC", count: 1 }, + { day: "2026-09-01", chain: undefined, asset: undefined, count: 9 }, + ]; + const withPruned = buildAnalytics({ payments, pruned, clearing: [], query: query(), now: NOW }); + // 4 kept checkouts + 5 pruned in range; the one outside the range is ignored. + expect(metric(withPruned, "checkouts")?.total).toBe(9); + expect(metric(withPruned, "failed_checkouts")?.total).toBe(6); + expect(metric(withPruned, "conversion_rate")?.total).toBeCloseTo(2 / 9); + expect(withPruned.performance.networks.find((row) => row.key === "Unknown")?.finished).toBe(4); + + const onBase = buildAnalytics({ + payments, + pruned, + clearing: [], + query: query({ chain: "base-sepolia" }), + now: NOW, + }); + expect(metric(onBase, "checkouts")?.total).toBe(5); + }); + test("a rate with nothing behind it is null, not zero", () => { - const empty = buildAnalytics({ payments: [], clearing: [], query: query(), now: NOW }); + const empty = buildAnalytics({ + payments: [], + pruned: [], + clearing: [], + query: query(), + now: NOW, + }); expect(metric(empty, "conversion_rate")?.total).toBeNull(); expect(metric(empty, "time_to_pay")?.points[0]?.value).toBeNull(); }); @@ -67,6 +96,7 @@ describe("buildAnalytics", () => { test("change compares against the range of the same length before", () => { const withPrevious = buildAnalytics({ payments: [...payments, intent("p", new Date(Date.UTC(2026, 8, 28, 10)))], + pruned: [], clearing: [], query: query(), now: NOW, @@ -84,7 +114,13 @@ describe("buildAnalytics", () => { netAmount: { asset: "USDC", amount: 9_900_000n }, fee: { asset: "USDC", amount: 100_000n }, } as ClearingTransaction; - const settled = buildAnalytics({ payments, clearing: [tx], query: query(), now: NOW }); + const settled = buildAnalytics({ + payments, + pruned: [], + clearing: [tx], + query: query(), + now: NOW, + }); expect(metric(settled, "net_revenue")?.total).toBe(990); expect(metric(settled, "fees")?.total).toBe(10); expect(metric(settled, "settlements")?.total).toBe(1); @@ -100,6 +136,7 @@ describe("buildAnalytics", () => { test("a network filter keeps only payments on it", () => { const other = buildAnalytics({ payments, + pruned: [], clearing: [], query: query({ chain: "arc-testnet" }), now: NOW, @@ -110,6 +147,7 @@ describe("buildAnalytics", () => { test("metrics narrows the response to the keys asked for", () => { const some = buildAnalytics({ payments, + pruned: [], clearing: [], query: query({ metrics: "orders, success_rate" }), now: NOW, diff --git a/apps/api/test/harness.ts b/apps/api/test/harness.ts index 991b2b94..bbf25a68 100644 --- a/apps/api/test/harness.ts +++ b/apps/api/test/harness.ts @@ -175,6 +175,7 @@ export function createApiHarness(options: ApiHarnessOptions = {}) { const registry = new InMemoryStablecoinRegistry(config.stablecoins); const intentRepository = new InMemoryPaymentIntentRepository(); + const expiredPayments = { prune: async () => 0, prunedCounts: async () => [] }; // The fraud gate over in-memory history. `market` is built further down, which is // fine: the rules are read when a payment is confirmed, not when this is built. const riskHistory = new InMemoryRiskHistory(); @@ -364,8 +365,10 @@ export function createApiHarness(options: ApiHarnessOptions = {}) { analytics: new AnalyticsService({ intents: intentRepository, clearing: clearingRepository, + pruned: expiredPayments, clock, }), + expiredPayments, escrow: new EscrowService(paymentApp, {}, escrowDisputes, clock, { opened: async () => {}, message: async (dispute, message) => { diff --git a/apps/dashboard/src/components/payment-detail.tsx b/apps/dashboard/src/components/payment-detail.tsx index 6b640626..34ba694e 100644 --- a/apps/dashboard/src/components/payment-detail.tsx +++ b/apps/dashboard/src/components/payment-detail.tsx @@ -43,7 +43,8 @@ function reasonOf(error: unknown): string { return match(error) .when( (e): e is ApiError => e instanceof ApiError && e.status === 404, - () => "No payment with that id belongs to this merchant.", + () => + "This payment is not available. It may belong to another merchant, or it went unpaid and was removed two days after it expired.", ) .when( (e): e is ApiError => e instanceof ApiError, diff --git a/packages/core/payment-intent/src/index.ts b/packages/core/payment-intent/src/index.ts index 4d336fea..4099d865 100644 --- a/packages/core/payment-intent/src/index.ts +++ b/packages/core/payment-intent/src/index.ts @@ -3,6 +3,7 @@ export * from "./from-qr.ts"; export * from "./intent.ts"; export * from "./merchant-policy.ts"; export * from "./payer-standing.ts"; +export * from "./pruning.ts"; export * from "./rail-catalog.ts"; export * from "./repository.ts"; export * from "./risk-gate.ts"; diff --git a/packages/core/payment-intent/src/pruning.ts b/packages/core/payment-intent/src/pruning.ts new file mode 100644 index 00000000..51a46159 --- /dev/null +++ b/packages/core/payment-intent/src/pruning.ts @@ -0,0 +1,29 @@ +/** + * Pruning abandoned payments. + * + * A payment nobody paid — expired, or never confirmed — is rubbish once nothing + * can still arrive for it. `prune` deletes those rows and keeps a count, so the + * merchant's analytics still see a checkout that happened and was not completed. + */ + +/** Payments pruned from one merchant, grouped by the day they were created and their rail. */ +export interface PrunedPaymentCount { + /** UTC day of creation, `YYYY-MM-DD`. */ + readonly day: string; + /** Undefined: the payer never chose a rail. */ + readonly chain: string | undefined; + readonly asset: string | undefined; + readonly count: number; +} + +export interface ExpiredPaymentPruner { + /** + * Deletes up to `limit` unpaid payments that expired before `cutoff` and + * returns how many went. A payment that has any sign of funds is never one of + * them: the deposit address saw a transfer, a clearing step past + * `PAYMENT_PENDING`, or a provider funded it. + */ + prune(cutoff: Date, limit: number): Promise; + /** What `prune` has removed for the merchant. */ + prunedCounts(merchantId: string): Promise; +} diff --git a/packages/db/migrations/0072_pruned_payment_counts.sql b/packages/db/migrations/0072_pruned_payment_counts.sql new file mode 100644 index 00000000..70ee2e01 --- /dev/null +++ b/packages/db/migrations/0072_pruned_payment_counts.sql @@ -0,0 +1,10 @@ +-- Abandoned payments are deleted after a grace period; this keeps the count so +-- analytics still sees them. One row per merchant, creation day and rail. +CREATE TABLE "pruned_payment_counts" ( + "merchant_id" text NOT NULL, + "day" text NOT NULL, + "chain" text NOT NULL, + "asset" text NOT NULL, + "count" integer NOT NULL, + PRIMARY KEY ("merchant_id", "day", "chain", "asset") +); diff --git a/packages/db/migrations/meta/_journal.json b/packages/db/migrations/meta/_journal.json index 69906714..05d6c15d 100644 --- a/packages/db/migrations/meta/_journal.json +++ b/packages/db/migrations/meta/_journal.json @@ -505,6 +505,13 @@ "when": 1793466400000, "tag": "0071_merchant_small_payment_fee", "breakpoints": true + }, + { + "idx": 72, + "version": "7", + "when": 1793552800000, + "tag": "0072_pruned_payment_counts", + "breakpoints": true } ] } diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index 4c872161..e555c12c 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -8,6 +8,7 @@ export * from "./repositories/catalog.ts"; export * from "./repositories/chain.ts"; export * from "./repositories/clearing.ts"; export * from "./repositories/escrow-dispute.ts"; +export * from "./repositories/expired-payment.ts"; export * from "./repositories/fraud.ts"; export * from "./repositories/invoice.ts"; export * from "./repositories/ledger.ts"; diff --git a/packages/db/src/repositories/expired-payment.ts b/packages/db/src/repositories/expired-payment.ts new file mode 100644 index 00000000..b3cceb40 --- /dev/null +++ b/packages/db/src/repositories/expired-payment.ts @@ -0,0 +1,171 @@ +/** + * Postgres adapter for `ExpiredPaymentPruner`. + * + * One transaction per batch: the children go first, then the payment, and the + * count is written in the same commit, so a crash leaves either all of it or none. + */ + +import type { ExpiredPaymentPruner, PrunedPaymentCount } from "@mayarin/payment-intent"; +import { + and, + eq, + inArray, + isNotNull, + lt, + notExists, + notInArray, + or, + type SQL, + sql, +} from "drizzle-orm"; +import type { Database } from "../client.ts"; +import { + chainDeposits, + clearingEvents, + clearingTransactions, + depositAddresses, + paymentFundings, + paymentIntents, + prunedPaymentCounts, + riskAssessments, + webhookDeliveries, +} from "../schema.ts"; + +/** Statuses of a payment nobody paid. `FAILED` is kept: it says why. */ +const UNPAID = ["CREATED", "CONFIRMED", "EXPIRED"]; + +/** Clearing states before any asset arrived. */ +const BEFORE_FUNDS = ["CREATED", "QR_PARSED", "PRICE_LOCKED", "PAYMENT_PENDING"]; + +export class DrizzleExpiredPaymentPruner implements ExpiredPaymentPruner { + readonly #db: Database; + + constructor(db: Database) { + this.#db = db; + } + + async prune(cutoff: Date, limit: number): Promise { + return this.#db.transaction(async (tx) => { + const pastFunds = tx + .select({ one: sql`1` }) + .from(clearingTransactions) + .where( + and( + eq(clearingTransactions.paymentIntentId, paymentIntents.id), + or( + notInArray(clearingTransactions.state, BEFORE_FUNDS), + isNotNull(clearingTransactions.providerReference), + ), + ), + ); + const deposited = tx + .select({ one: sql`1` }) + .from(chainDeposits) + .innerJoin( + depositAddresses, + and( + eq(depositAddresses.chain, chainDeposits.chain), + eq(depositAddresses.address, chainDeposits.address), + ), + ) + .innerJoin( + clearingTransactions, + eq(clearingTransactions.id, depositAddresses.clearingTransactionId), + ) + .where(eq(clearingTransactions.paymentIntentId, paymentIntents.id)); + const funded = tx + .select({ one: sql`1` }) + .from(paymentFundings) + .where(eq(paymentFundings.paymentIntentId, paymentIntents.id)); + + const rows = await tx + .select({ + id: paymentIntents.id, + merchantId: paymentIntents.merchantId, + createdAt: paymentIntents.createdAt, + chain: paymentIntents.paymentChain, + asset: paymentIntents.paymentAsset, + }) + .from(paymentIntents) + .where( + and( + inArray(paymentIntents.status, UNPAID), + lt(paymentIntents.expiresAt, cutoff), + notExists(pastFunds), + notExists(deposited), + notExists(funded), + ), + ) + .orderBy(paymentIntents.expiresAt) + .limit(limit) + .for("update", { skipLocked: true }); + if (rows.length === 0) return 0; + + const ids = rows.map((row) => row.id); + const transactionIds = tx + .select({ id: clearingTransactions.id }) + .from(clearingTransactions) + .where(inArray(clearingTransactions.paymentIntentId, ids)); + const eventIds = tx + .select({ id: clearingEvents.id }) + .from(clearingEvents) + .where(inArray(clearingEvents.clearingTransactionId, transactionIds)); + + await tx.delete(webhookDeliveries).where(inArray(webhookDeliveries.eventId, eventIds)); + await tx + .delete(clearingEvents) + .where(inArray(clearingEvents.clearingTransactionId, transactionIds)); + await tx + .delete(depositAddresses) + .where(inArray(depositAddresses.clearingTransactionId, transactionIds)); + await tx.delete(riskAssessments).where(inArray(riskAssessments.paymentIntentId, ids)); + await tx + .delete(clearingTransactions) + .where(inArray(clearingTransactions.paymentIntentId, ids)); + await tx.delete(paymentIntents).where(inArray(paymentIntents.id, ids)); + + const counts = new Map(); + for (const row of rows) { + const day = row.createdAt.toISOString().slice(0, 10); + const chain = row.chain ?? ""; + const asset = row.asset ?? ""; + const key = [row.merchantId, day, chain, asset].join("|"); + const seen = counts.get(key); + counts.set(key, { + merchantId: row.merchantId, + day, + chain, + asset, + count: (seen?.count ?? 0) + 1, + }); + } + const added: SQL = sql`${prunedPaymentCounts.count} + excluded.count`; + await tx + .insert(prunedPaymentCounts) + .values([...counts.values()]) + .onConflictDoUpdate({ + target: [ + prunedPaymentCounts.merchantId, + prunedPaymentCounts.day, + prunedPaymentCounts.chain, + prunedPaymentCounts.asset, + ], + set: { count: added }, + }); + return rows.length; + }); + } + + async prunedCounts(merchantId: string): Promise { + const rows = await this.#db + .select() + .from(prunedPaymentCounts) + .where(eq(prunedPaymentCounts.merchantId, merchantId)); + return rows.map((row) => ({ + day: row.day, + chain: row.chain === "" ? undefined : row.chain, + asset: row.asset === "" ? undefined : row.asset, + count: row.count, + })); + } +} diff --git a/packages/db/src/schema.ts b/packages/db/src/schema.ts index 3467890f..0f470b12 100644 --- a/packages/db/src/schema.ts +++ b/packages/db/src/schema.ts @@ -1568,3 +1568,21 @@ export const referralCampaign = pgTable("referral_campaign", { chains: text("chains").array().notNull(), updatedAt: timestamp("updated_at", { withTimezone: true, mode: "date" }).notNull(), }); + +/** + * Abandoned payments that were pruned, counted by merchant, creation day and rail. + * Analytics adds these back, so deleting an expired payment never lowers the + * checkout count or the failed-and-expired count. `chain` and `asset` are empty + * when the payer never chose a rail. + */ +export const prunedPaymentCounts = pgTable( + "pruned_payment_counts", + { + merchantId: text("merchant_id").notNull(), + day: text("day").notNull(), + chain: text("chain").notNull(), + asset: text("asset").notNull(), + count: integer("count").notNull(), + }, + (table) => [primaryKey({ columns: [table.merchantId, table.day, table.chain, table.asset] })], +); diff --git a/packages/db/test/postgres.test.ts b/packages/db/test/postgres.test.ts index add0a1c5..b17e9388 100644 --- a/packages/db/test/postgres.test.ts +++ b/packages/db/test/postgres.test.ts @@ -66,6 +66,7 @@ import { } from "../src/repositories/chain.ts"; import { DrizzleClearingRepository } from "../src/repositories/clearing.ts"; import { DrizzleEscrowDisputeRepository } from "../src/repositories/escrow-dispute.ts"; +import { DrizzleExpiredPaymentPruner } from "../src/repositories/expired-payment.ts"; import { DrizzleDenylist, DrizzleRiskAssessments, @@ -88,8 +89,11 @@ import { } from "../src/repositories/wallet.ts"; import { DrizzleX402ResourceRepository } from "../src/repositories/x402.ts"; import { + chainDeposits, + clearingTransactions, loginFailures as loginFailuresTable, operatorTransactions, + paymentIntents, workerFencing, } from "../src/schema.ts"; import { guardedTestDatabaseUrl } from "./database-guard.ts"; @@ -149,7 +153,7 @@ describe.skipIf(TEST_DATABASE_URL === undefined)("Drizzle repositories", () => { */ async function truncateAll(): Promise { await handle.db.execute( - sql`truncate table risk_assessments, address_denylist, worker_fencing, login_failures, operator_transactions, operator_nonces, escrow_dispute_messages, escrow_disputes, x402_resources, webhook_deliveries, webhook_endpoints, webhook_cursors, wallet_withdrawals, wallet_challenges, merchant_wallets, sessions, users, merchants, chain_deposits, deposit_addresses, settlement_events, watcher_cursors, clearing_events, clearing_transactions, ledger_entries, ledger_transactions, ledger_accounts, payment_intents restart identity cascade`, + sql`truncate table risk_assessments, address_denylist, worker_fencing, login_failures, operator_transactions, operator_nonces, escrow_dispute_messages, escrow_disputes, x402_resources, webhook_deliveries, webhook_endpoints, webhook_cursors, wallet_withdrawals, wallet_challenges, merchant_wallets, sessions, users, merchants, chain_deposits, deposit_addresses, settlement_events, watcher_cursors, clearing_events, clearing_transactions, ledger_entries, ledger_transactions, ledger_accounts, payment_intents, pruned_payment_counts restart identity cascade`, ); } @@ -675,6 +679,60 @@ describe.skipIf(TEST_DATABASE_URL === undefined)("Drizzle repositories", () => { expect(again.address).toBe(first.address); }); + test("prunes an unpaid payment and its rows, keeps one a deposit reached, and counts both", async () => { + const pruner = new DrizzleExpiredPaymentPruner(handle.db); + const deriver = new FixedDepositAddressDeriver(); + const allocate = (clearingTransactionId: string) => + new DrizzleDepositAddressRepository(handle.db).allocate({ + clearingTransactionId, + chain: "base-sepolia", + asset: "USDC", + deriver, + now: new Date(), + }); + + // Never paid, with a deposit address and a clearing history. + const abandoned = await seedClearingTransaction(); + await allocate(abandoned); + // A transfer reached this one's address, so nothing about it may go. + const funded = await seedClearingTransaction(); + const address = await allocate(funded); + await handle.db.insert(chainDeposits).values({ + id: "dep-1", + chain: "base-sepolia", + txHash: "0xabc", + logIndex: 0, + address: address.address, + asset: "USDC", + amount: "1", + blockNumber: "1", + blockHash: "0xdef", + status: "confirmed", + firstSeenAt: new Date(), + }); + // Not yet past the cutoff. + const recent = await confirmedIntent(); + + const cutoff = new Date(clock.now().getTime() + 900_000 + 1); + await handle.db + .update(paymentIntents) + .set({ expiresAt: new Date(cutoff.getTime() + 1) }) + .where(eq(paymentIntents.id, recent.id)); + + expect(await pruner.prune(cutoff, 100)).toBe(1); + + const left = ( + await handle.db.select({ id: clearingTransactions.id }).from(clearingTransactions) + ).map((row) => row.id); + expect(left).toEqual([funded]); + expect(await handle.db.select().from(paymentIntents)).toHaveLength(2); + + const counts = await pruner.prunedCounts("ID1020017611473"); + expect(counts).toEqual([{ day: "2026-01-01", chain: undefined, asset: undefined, count: 1 }]); + // Nothing left to prune a second time. + expect(await pruner.prune(cutoff, 100)).toBe(0); + }); + test("records a transfer once however often it is seen", async () => { const repository = new DrizzleDepositRepository(handle.db); const log = {