From b127fcc2750cd3b2a6c61cec89b7e251ee7f5cd8 Mon Sep 17 00:00:00 2001 From: Abdulmumin Yaqeen Date: Wed, 8 Jul 2026 09:49:00 +0100 Subject: [PATCH 1/3] fix(api): make customer auto-create alias-safe --- apps/api/src/lib/customers.ts | 67 ++++++++++++++++- .../runtime/customers-auto-create.test.ts | 74 +++++++++++++++++++ apps/api/test/runtime/helpers/sqlite-d1.ts | 4 + .../0011_customer_alias_uniqueness.sql | 2 + packages/db/src/schema/billing.ts | 7 ++ 5 files changed, 150 insertions(+), 4 deletions(-) create mode 100644 apps/api/test/runtime/customers-auto-create.test.ts create mode 100644 packages/db/migrations/0011_customer_alias_uniqueness.sql diff --git a/apps/api/src/lib/customers.ts b/apps/api/src/lib/customers.ts index d83f02f..3cebd97 100644 --- a/apps/api/src/lib/customers.ts +++ b/apps/api/src/lib/customers.ts @@ -1,10 +1,48 @@ import { eq } from "drizzle-orm"; import { schema, createDb } from "@owostack/db"; import { EntitlementCache } from "./cache"; -import { resolveCustomerByIdentifier } from "./customer-resolution"; +import { + resolveCustomerByEmail, + resolveCustomerByIdentifier, +} from "./customer-resolution"; import { autoAssignPlansToNewCustomer } from "./customer-auto-plans"; type DB = ReturnType; +type Customer = typeof schema.customers.$inferSelect; + +function isUniqueConstraintError(error: unknown): boolean { + if (!error || typeof error !== "object") return false; + const message = "message" in error ? String(error.message) : ""; + return /unique constraint failed|constraint failed/i.test(message); +} + +async function resolveExistingCustomerAfterCreateConflict(opts: { + db: DB; + organizationId: string; + customerId: string; + email: string; + cache?: EntitlementCache | null; + waitUntil?: (promise: Promise) => void; +}): Promise { + const byIdentifier = await resolveCustomerByIdentifier({ + db: opts.db, + organizationId: opts.organizationId, + customerId: opts.customerId, + cache: opts.cache, + waitUntil: opts.waitUntil, + }); + if (byIdentifier?.customer) return byIdentifier.customer; + + const byEmail = await resolveCustomerByEmail({ + db: opts.db, + organizationId: opts.organizationId, + email: opts.email, + cache: opts.cache, + waitUntil: opts.waitUntil, + }); + + return byEmail?.customer ?? null; +} export interface CustomerData { email: string; @@ -133,10 +171,31 @@ export async function resolveOrCreateCustomer( updatedAt: now, }; - await db.insert(schema.customers).values(newCustomer); - customer = newCustomer as unknown as typeof schema.customers.$inferSelect; + let createdCustomer = false; + try { + await db.insert(schema.customers).values(newCustomer); + customer = newCustomer as unknown as typeof schema.customers.$inferSelect; + createdCustomer = true; + } catch (error) { + if (!isUniqueConstraintError(error)) { + throw error; + } + + customer = await resolveExistingCustomerAfterCreateConflict({ + db, + organizationId, + customerId, + email, + cache, + waitUntil: opts.waitUntil, + }); + + if (!customer) { + throw error; + } + } - if (opts.autoApplyPlansOnCreate) { + if (createdCustomer && opts.autoApplyPlansOnCreate) { await autoAssignPlansToNewCustomer({ db, organizationId, diff --git a/apps/api/test/runtime/customers-auto-create.test.ts b/apps/api/test/runtime/customers-auto-create.test.ts new file mode 100644 index 0000000..ae39553 --- /dev/null +++ b/apps/api/test/runtime/customers-auto-create.test.ts @@ -0,0 +1,74 @@ +import { and, eq } from "drizzle-orm"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { schema } from "@owostack/db"; +import { resolveOrCreateCustomer } from "../../src/lib/customers"; +import { createRuntimeBusinessDb } from "./helpers/business-db"; +import { insertOrganization } from "./helpers/workflow-runtime"; + +describe("resolveOrCreateCustomer runtime integration", () => { + let businessDb: ReturnType; + + beforeEach(async () => { + businessDb = createRuntimeBusinessDb(); + await insertOrganization(businessDb.d1, { id: "org_customer_create" }); + }); + + afterEach(() => { + businessDb.close(); + }); + + it("coalesces concurrent auto-create requests for the same organization-scoped aliases", async () => { + const created = await Promise.all( + Array.from({ length: 12 }, () => + resolveOrCreateCustomer({ + db: businessDb.db, + organizationId: "org_customer_create", + customerId: "external_user_123", + customerData: { + email: "USER@example.com", + name: "Runtime User", + }, + }), + ), + ); + + const customerIds = new Set(created.map((customer) => customer?.id)); + expect(customerIds.size).toBe(1); + + const rows = await businessDb.db.query.customers.findMany({ + where: and( + eq(schema.customers.organizationId, "org_customer_create"), + eq(schema.customers.email, "user@example.com"), + ), + }); + + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + organizationId: "org_customer_create", + externalId: "external_user_123", + email: "user@example.com", + name: "Runtime User", + }); + }); + + it("keeps the same email reusable across different organizations", async () => { + await insertOrganization(businessDb.d1, { id: "org_customer_create_2" }); + + const first = await resolveOrCreateCustomer({ + db: businessDb.db, + organizationId: "org_customer_create", + customerId: "shared_user", + customerData: { email: "shared@example.com" }, + }); + const second = await resolveOrCreateCustomer({ + db: businessDb.db, + organizationId: "org_customer_create_2", + customerId: "shared_user", + customerData: { email: "shared@example.com" }, + }); + + expect(first?.id).toBeTruthy(); + expect(second?.id).toBeTruthy(); + expect(first?.id).not.toBe(second?.id); + }); +}); diff --git a/apps/api/test/runtime/helpers/sqlite-d1.ts b/apps/api/test/runtime/helpers/sqlite-d1.ts index c1cb712..9a3eeb6 100644 --- a/apps/api/test/runtime/helpers/sqlite-d1.ts +++ b/apps/api/test/runtime/helpers/sqlite-d1.ts @@ -42,6 +42,10 @@ const MIGRATION_FILES = [ "../../../../../packages/db/migrations/0010_single_default_payment_method.sql", import.meta.url, ), + new URL( + "../../../../../packages/db/migrations/0011_customer_alias_uniqueness.sql", + import.meta.url, + ), ]; type SqliteRunResult = { diff --git a/packages/db/migrations/0011_customer_alias_uniqueness.sql b/packages/db/migrations/0011_customer_alias_uniqueness.sql new file mode 100644 index 0000000..341ea2d --- /dev/null +++ b/packages/db/migrations/0011_customer_alias_uniqueness.sql @@ -0,0 +1,2 @@ +CREATE UNIQUE INDEX `customers_org_email_uniq` ON `customers` (`organization_id`, `email`);--> statement-breakpoint +CREATE UNIQUE INDEX `customers_org_external_uniq` ON `customers` (`organization_id`, `external_id`) WHERE `external_id` IS NOT NULL; diff --git a/packages/db/src/schema/billing.ts b/packages/db/src/schema/billing.ts index 243b35d..b096feb 100644 --- a/packages/db/src/schema/billing.ts +++ b/packages/db/src/schema/billing.ts @@ -46,6 +46,13 @@ export const customers = sqliteTable( index("customers_org_idx").on(table.organizationId), index("customers_email_idx").on(table.email), index("customers_external_idx").on(table.externalId), + uniqueIndex("customers_org_email_uniq").on( + table.organizationId, + table.email, + ), + uniqueIndex("customers_org_external_uniq") + .on(table.organizationId, table.externalId) + .where(sql`${table.externalId} is not null`), ], ); From 315d7d4f3ed812a259d3825ed95339ff58d5f151 Mon Sep 17 00:00:00 2001 From: Abdulmumin Yaqeen Date: Wed, 8 Jul 2026 09:51:27 +0100 Subject: [PATCH 2/3] fix(api): mark usage after invoice commit --- apps/api/src/lib/billing.ts | 49 ++++++ .../billing-invoice-durability.test.ts | 165 ++++++++++++++++++ 2 files changed, 214 insertions(+) create mode 100644 apps/api/test/runtime/billing-invoice-durability.test.ts diff --git a/apps/api/src/lib/billing.ts b/apps/api/src/lib/billing.ts index 099bfdb..75f16cf 100644 --- a/apps/api/src/lib/billing.ts +++ b/apps/api/src/lib/billing.ts @@ -622,6 +622,11 @@ export class BillingService { existingItemKeys.add(itemKey); } + } + }; + + const markInvoiceUsage = async () => { + for (const f of unbilled.features) { const ledgerMarked = await this.deps.markUsageInvoiced( { usageLedger: this.opts?.usageLedger, @@ -653,6 +658,38 @@ export class BillingService { } }; + const voidInvoiceAfterLedgerFailure = async ( + cause: unknown, + invoiceMetadata: Record, + ) => { + const releasedUsageRecords = await this.deps.releaseUsageInvoice( + { + usageLedger: this.opts?.usageLedger, + organizationId, + }, + invoiceId, + ); + + if (releasedUsageRecords === null) { + throw cause; + } + + await this.db + .update(schema.invoices) + .set({ + status: "void", + amountDue: 0, + updatedAt: Date.now(), + metadata: { + ...invoiceMetadata, + sourceTrigger: options.sourceTrigger, + voidedReason: "usage_ledger_mark_failed", + releasedUsageRecords, + }, + }) + .where(eq(schema.invoices.id, invoiceId)); + }; + try { await this.db.transaction(async (tx: any) => { await applyInvoiceWrites(tx); @@ -696,6 +733,18 @@ export class BillingService { ); } + try { + await markInvoiceUsage(); + } catch (error) { + await voidInvoiceAfterLedgerFailure( + error, + typeof finalInvoice.metadata === "object" && finalInvoice.metadata + ? (finalInvoice.metadata as Record) + : {}, + ); + throw error; + } + const featureSlugById = new Map( unbilled.features.map((feature) => [ feature.featureId, diff --git a/apps/api/test/runtime/billing-invoice-durability.test.ts b/apps/api/test/runtime/billing-invoice-durability.test.ts new file mode 100644 index 0000000..2d3d6c9 --- /dev/null +++ b/apps/api/test/runtime/billing-invoice-durability.test.ts @@ -0,0 +1,165 @@ +import { eq } from "drizzle-orm"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { schema } from "@owostack/db"; +import { BillingService } from "../../src/lib/billing"; +import { createRuntimeBusinessDb } from "./helpers/business-db"; +import { insertFeature } from "./helpers/overage-runtime"; +import { insertCustomer, insertOrganization } from "./helpers/workflow-runtime"; + +describe("BillingService invoice usage durability", () => { + let businessDb: ReturnType; + + beforeEach(async () => { + businessDb = createRuntimeBusinessDb(); + await insertOrganization(businessDb.d1, { id: "org_invoice_durability" }); + await insertCustomer(businessDb.d1, { + id: "cust_invoice_durability", + organizationId: "org_invoice_durability", + email: "invoice-durability@example.com", + }); + await insertFeature(businessDb.d1, { + id: "feature_invoice_durability", + organizationId: "org_invoice_durability", + slug: "api-calls", + name: "API Calls", + type: "metered", + }); + }); + + afterEach(() => { + businessDb.close(); + }); + + function createUnbilledUsage() { + return { + customerId: "cust_invoice_durability", + usageWindowEnd: 2_000, + currency: "USD", + totalEstimated: 500, + features: [ + { + featureId: "feature_invoice_durability", + featureSlug: "api-calls", + featureName: "API Calls", + usageModel: "usage_based", + usage: 5, + included: 0, + billableQuantity: 5, + pricePerUnit: 100, + billingUnits: 1, + estimatedAmount: 500, + periodStart: 1_000, + periodEnd: 2_000, + billingGroupKey: "feature_invoice_durability:1000:2000", + }, + ], + }; + } + + it("marks usage only after the invoice row is durably readable", async () => { + const observedInvoiceIds: string[] = []; + const service = new BillingService(businessDb.db, { + deps: { + markUsageInvoiced: async (_ctx, params) => { + const invoice = await businessDb.db.query.invoices.findFirst({ + where: eq(schema.invoices.id, params.invoiceId), + }); + if (!invoice) return null; + observedInvoiceIds.push(invoice.id); + return 1; + }, + releaseUsageInvoice: async () => 0, + releaseCustomerOverageBlockForInvoice: async () => undefined, + sumUsageAmount: async () => 0, + sumUnbilledByFeaturePeriod: async () => [], + }, + }); + + const result = await service.createInvoiceFromUsage( + "cust_invoice_durability", + "org_invoice_durability", + createUnbilledUsage(), + { + idempotencyKey: "manual:org_invoice_durability:cust_invoice_durability:2000", + sourceTrigger: "manual", + }, + ); + + expect(result.invoiceId).toBeTruthy(); + expect(observedInvoiceIds).toEqual([result.invoiceId]); + }); + + it("voids a durable invoice and releases partial ledger marks if a later mark fails", async () => { + await insertFeature(businessDb.d1, { + id: "feature_invoice_durability_2", + organizationId: "org_invoice_durability", + slug: "storage", + name: "Storage", + type: "metered", + }); + + const releasedInvoiceIds: string[] = []; + const service = new BillingService(businessDb.db, { + deps: { + markUsageInvoiced: async (_ctx, params) => { + if (params.featureId === "feature_invoice_durability_2") { + return null; + } + return 1; + }, + releaseUsageInvoice: async (_ctx, invoiceId) => { + releasedInvoiceIds.push(invoiceId); + return 1; + }, + releaseCustomerOverageBlockForInvoice: async () => undefined, + sumUsageAmount: async () => 0, + sumUnbilledByFeaturePeriod: async () => [], + }, + }); + + const unbilled = createUnbilledUsage(); + unbilled.features.push({ + featureId: "feature_invoice_durability_2", + featureSlug: "storage", + featureName: "Storage", + usageModel: "usage_based", + usage: 2, + included: 0, + billableQuantity: 2, + pricePerUnit: 50, + billingUnits: 1, + estimatedAmount: 100, + periodStart: 1_000, + periodEnd: 2_000, + billingGroupKey: "feature_invoice_durability_2:1000:2000", + }); + unbilled.totalEstimated = 600; + + await expect( + service.createInvoiceFromUsage( + "cust_invoice_durability", + "org_invoice_durability", + unbilled, + { + idempotencyKey: "manual:org_invoice_durability:cust_invoice_durability:2000", + sourceTrigger: "manual", + }, + ), + ).rejects.toThrow("Failed to mark usage as invoiced"); + + expect(releasedInvoiceIds).toHaveLength(1); + const invoice = await businessDb.db.query.invoices.findFirst({ + where: eq(schema.invoices.id, releasedInvoiceIds[0]!), + }); + + expect(invoice).toMatchObject({ + status: "void", + amountDue: 0, + }); + expect(invoice?.metadata).toMatchObject({ + sourceTrigger: "manual", + voidedReason: "usage_ledger_mark_failed", + releasedUsageRecords: 1, + }); + }); +}); From bf594741a9574907cedfcef993c92b6596f0f6d3 Mon Sep 17 00:00:00 2001 From: Abdulmumin Yaqeen Date: Wed, 8 Jul 2026 10:00:43 +0100 Subject: [PATCH 3/3] fix(api): recover credit purchase application --- .../lib/webhooks/handlers/charge-success.ts | 79 +++++++++-- apps/api/test/runtime/helpers/sqlite-d1.ts | 8 ++ .../credit-purchase-idempotency.test.ts | 125 ++++++++++++++++++ .../migrations/0011_credit_balance_ledger.sql | 13 ++ ...012_credit_purchase_application_status.sql | 2 + packages/db/src/schema/billing.ts | 29 ++++ 6 files changed, 247 insertions(+), 9 deletions(-) create mode 100644 packages/db/migrations/0011_credit_balance_ledger.sql create mode 100644 packages/db/migrations/0012_credit_purchase_application_status.sql diff --git a/apps/api/src/lib/webhooks/handlers/charge-success.ts b/apps/api/src/lib/webhooks/handlers/charge-success.ts index fa62cde..e5dfbdd 100644 --- a/apps/api/src/lib/webhooks/handlers/charge-success.ts +++ b/apps/api/src/lib/webhooks/handlers/charge-success.ts @@ -882,8 +882,9 @@ async function handleCreditPurchase( } // Claim the provider payment reference before granting credits. The unique - // index on credit_purchases.payment_reference makes this safe across - // concurrent duplicate webhook deliveries; exactly one delivery gets to top up. + // index on credit_purchases.payment_reference makes the reference claim + // durable, while status/appliedAt lets retries finish a claimed-but-unapplied + // purchase without double-applying credits. const purchaseId = crypto.randomUUID(); const purchaseInsert = await (db as any) .insert((schema as any).creditPurchases) @@ -898,26 +899,56 @@ async function handleCreditPurchase( currency: event.payment?.currency || "USD", paymentReference: reference, providerId: event.provider, + status: "pending", + appliedAt: null, metadata: event.raw, }) .onConflictDoNothing({ target: (schema as any).creditPurchases.paymentReference, }) - .returning({ id: (schema as any).creditPurchases.id }); + .returning({ + id: (schema as any).creditPurchases.id, + status: (schema as any).creditPurchases.status, + customerId: (schema as any).creditPurchases.customerId, + creditSystemId: (schema as any).creditPurchases.creditSystemId, + credits: (schema as any).creditPurchases.credits, + }); + + let purchase = purchaseInsert[0] as + | { id: string; status: string; customerId?: string; creditSystemId?: string | null; credits?: number } + | undefined; + if (!purchase) { + purchase = await (db as any).query.creditPurchases.findFirst({ + where: eq((schema as any).creditPurchases.paymentReference, reference), + columns: { + id: true, + status: true, + customerId: true, + creditSystemId: true, + credits: true, + }, + }); + } - if (!purchaseInsert[0]) { + if (!purchase) { + throw new Error( + `[WEBHOOK] Credit purchase claim disappeared after reference claim: ref=${reference}`, + ); + } + + if (purchase?.status === "completed") { console.log( `[WEBHOOK] Credit purchase already processed: ref=${reference}, skipping`, ); return; } - // Atomic upsert into credit_system_balances (uses UNIQUE index) - await chargeSuccessDependencies.topUpScopedBalance( + await applyCreditPurchaseBalance( db, - dbCustomer.id, - creditSystemId, - creditsAmount, + purchase.id, + purchase.customerId ?? dbCustomer.id, + purchase.creditSystemId ?? creditSystemId, + purchase.credits ?? creditsAmount, ); console.log( @@ -925,6 +956,36 @@ async function handleCreditPurchase( ); } +async function applyCreditPurchaseBalance( + db: any, + purchaseId: string, + customerId: string, + creditSystemId: string, + amount: number, +): Promise { + const now = Date.now(); + + await (db as any).run( + sql`INSERT INTO credit_balance_ledger (id, purchase_id, customer_id, credit_system_id, amount, created_at) + VALUES (${crypto.randomUUID()}, ${purchaseId}, ${customerId}, ${creditSystemId}, ${amount}, ${now}) + ON CONFLICT (purchase_id) DO NOTHING`, + ); + + await (db as any).run( + sql`INSERT INTO credit_system_balances (id, customer_id, credit_system_id, balance, updated_at) + SELECT ${crypto.randomUUID()}, ${customerId}, ${creditSystemId}, COALESCE(SUM(amount), 0), ${now} + FROM credit_balance_ledger + WHERE customer_id = ${customerId} AND credit_system_id = ${creditSystemId} + ON CONFLICT (customer_id, credit_system_id) + DO UPDATE SET balance = excluded.balance, updated_at = ${now}`, + ); + + await (db as any) + .update((schema as any).creditPurchases) + .set({ status: "completed", appliedAt: now }) + .where(eq((schema as any).creditPurchases.id, purchaseId)); +} + async function handleOneTimePurchase( ctx: WebhookContext, dbCustomer: any, diff --git a/apps/api/test/runtime/helpers/sqlite-d1.ts b/apps/api/test/runtime/helpers/sqlite-d1.ts index c1cb712..2a4dcdf 100644 --- a/apps/api/test/runtime/helpers/sqlite-d1.ts +++ b/apps/api/test/runtime/helpers/sqlite-d1.ts @@ -42,6 +42,14 @@ const MIGRATION_FILES = [ "../../../../../packages/db/migrations/0010_single_default_payment_method.sql", import.meta.url, ), + new URL( + "../../../../../packages/db/migrations/0011_credit_balance_ledger.sql", + import.meta.url, + ), + new URL( + "../../../../../packages/db/migrations/0012_credit_purchase_application_status.sql", + import.meta.url, + ), ]; type SqliteRunResult = { diff --git a/apps/api/test/runtime/webhooks/credit-purchase-idempotency.test.ts b/apps/api/test/runtime/webhooks/credit-purchase-idempotency.test.ts index 020d53f..e8f9472 100644 --- a/apps/api/test/runtime/webhooks/credit-purchase-idempotency.test.ts +++ b/apps/api/test/runtime/webhooks/credit-purchase-idempotency.test.ts @@ -6,6 +6,7 @@ import { insertCustomer, insertOrganization } from "../helpers/workflow-runtime" type BalanceRow = { balance: number }; type CountRow = { count: number }; +type PurchaseRow = { id: string; status: string; applied_at: number | null }; async function insertCreditSystem(db: D1Database) { const now = Date.now(); @@ -51,6 +52,31 @@ async function countPurchases(db: D1Database): Promise { return row?.count ?? 0; } +async function countLedgerRows(db: D1Database): Promise { + const row = await db + .prepare( + `SELECT COUNT(*) AS count + FROM credit_balance_ledger + WHERE purchase_id IN ( + SELECT id FROM credit_purchases WHERE payment_reference = ? + )`, + ) + .bind("ref_credit_duplicate") + .first(); + return row?.count ?? 0; +} + +async function loadPurchase(db: D1Database): Promise { + return await db + .prepare( + `SELECT id, status, applied_at + FROM credit_purchases + WHERE payment_reference = ?`, + ) + .bind("ref_credit_duplicate") + .first(); +} + function creditPurchaseEvent(): NormalizedWebhookEvent { return { type: "charge.success", @@ -115,6 +141,105 @@ describe("Credit purchase webhook idempotency runtime integration", () => { expect(first.isOk()).toBe(true); expect(second.isOk()).toBe(true); expect(await countPurchases(businessDb.d1)).toBe(1); + expect(await countLedgerRows(businessDb.d1)).toBe(1); expect(await loadBalance(businessDb.d1)).toBe(40); + expect(await loadPurchase(businessDb.d1)).toMatchObject({ + status: "completed", + }); + }); + + it("recovers a claimed credit purchase that failed before applying credits", async () => { + await businessDb.d1 + .prepare( + `INSERT INTO credit_purchases + (id, customer_id, credit_pack_id, credit_system_id, credits, quantity, price, currency, payment_reference, provider_id, status, applied_at, metadata, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + "cp_claimed_before_topup", + "cust_1", + null, + "cs_ai", + 40, + 2, + 7500, + "NGN", + "ref_credit_duplicate", + "paystack", + "pending", + null, + JSON.stringify({}), + Date.now(), + ) + .run(); + + const result = await handler.handle(creditPurchaseEvent()); + + expect(result.isOk()).toBe(true); + expect(await countPurchases(businessDb.d1)).toBe(1); + expect(await countLedgerRows(businessDb.d1)).toBe(1); + expect(await loadBalance(businessDb.d1)).toBe(40); + expect(await loadPurchase(businessDb.d1)).toMatchObject({ + status: "completed", + }); + }); + + it("does not double top up when retrying after balance update but before completion marker", async () => { + await businessDb.d1 + .prepare( + `INSERT INTO credit_purchases + (id, customer_id, credit_pack_id, credit_system_id, credits, quantity, price, currency, payment_reference, provider_id, status, applied_at, metadata, created_at) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, + ) + .bind( + "cp_balance_without_marker", + "cust_1", + null, + "cs_ai", + 40, + 2, + 7500, + "NGN", + "ref_credit_duplicate", + "paystack", + "pending", + null, + JSON.stringify({}), + Date.now(), + ) + .run(); + await businessDb.d1 + .prepare( + `INSERT INTO credit_balance_ledger + (id, purchase_id, customer_id, credit_system_id, amount, created_at) + VALUES (?, ?, ?, ?, ?, ?)`, + ) + .bind( + "cbl_balance_without_marker", + "cp_balance_without_marker", + "cust_1", + "cs_ai", + 40, + Date.now(), + ) + .run(); + await businessDb.d1 + .prepare( + `INSERT INTO credit_system_balances + (id, customer_id, credit_system_id, balance, updated_at) + VALUES (?, ?, ?, ?, ?)`, + ) + .bind("csb_balance_without_marker", "cust_1", "cs_ai", 40, Date.now()) + .run(); + + const result = await handler.handle(creditPurchaseEvent()); + + expect(result.isOk()).toBe(true); + expect(await countPurchases(businessDb.d1)).toBe(1); + expect(await countLedgerRows(businessDb.d1)).toBe(1); + expect(await loadBalance(businessDb.d1)).toBe(40); + expect(await loadPurchase(businessDb.d1)).toMatchObject({ + status: "completed", + }); }); }); diff --git a/packages/db/migrations/0011_credit_balance_ledger.sql b/packages/db/migrations/0011_credit_balance_ledger.sql new file mode 100644 index 0000000..e224588 --- /dev/null +++ b/packages/db/migrations/0011_credit_balance_ledger.sql @@ -0,0 +1,13 @@ +CREATE TABLE `credit_balance_ledger` ( + `id` text PRIMARY KEY NOT NULL, + `purchase_id` text NOT NULL, + `customer_id` text NOT NULL, + `credit_system_id` text NOT NULL, + `amount` integer NOT NULL, + `created_at` integer NOT NULL, + FOREIGN KEY (`purchase_id`) REFERENCES `credit_purchases`(`id`) ON UPDATE no action ON DELETE cascade, + FOREIGN KEY (`customer_id`) REFERENCES `customers`(`id`) ON UPDATE no action ON DELETE cascade, + FOREIGN KEY (`credit_system_id`) REFERENCES `credit_systems`(`id`) ON UPDATE no action ON DELETE cascade +);--> statement-breakpoint +CREATE UNIQUE INDEX `credit_balance_ledger_purchase_uniq` ON `credit_balance_ledger` (`purchase_id`);--> statement-breakpoint +CREATE INDEX `credit_balance_ledger_balance_idx` ON `credit_balance_ledger` (`customer_id`,`credit_system_id`); diff --git a/packages/db/migrations/0012_credit_purchase_application_status.sql b/packages/db/migrations/0012_credit_purchase_application_status.sql new file mode 100644 index 0000000..32a082e --- /dev/null +++ b/packages/db/migrations/0012_credit_purchase_application_status.sql @@ -0,0 +1,2 @@ +ALTER TABLE `credit_purchases` ADD `status` text DEFAULT 'pending' NOT NULL;--> statement-breakpoint +ALTER TABLE `credit_purchases` ADD `applied_at` integer; diff --git a/packages/db/src/schema/billing.ts b/packages/db/src/schema/billing.ts index 243b35d..099a137 100644 --- a/packages/db/src/schema/billing.ts +++ b/packages/db/src/schema/billing.ts @@ -467,6 +467,8 @@ export const creditPurchases = sqliteTable( currency: text("currency").notNull().default("NGN"), paymentReference: text("payment_reference"), providerId: text("provider_id"), + status: text("status").notNull().default("pending"), + appliedAt: integer("applied_at"), metadata: text("metadata", { mode: "json" }).$type< Record >(), @@ -483,6 +485,33 @@ export const creditPurchases = sqliteTable( ], ); +export const creditBalanceLedger = sqliteTable( + "credit_balance_ledger", + { + id: text("id").primaryKey(), + purchaseId: text("purchase_id") + .notNull() + .references(() => creditPurchases.id, { onDelete: "cascade" }), + customerId: text("customer_id") + .notNull() + .references(() => customers.id, { onDelete: "cascade" }), + creditSystemId: text("credit_system_id") + .notNull() + .references(() => creditSystems.id, { onDelete: "cascade" }), + amount: integer("amount").notNull(), + createdAt: integer("created_at") + .notNull() + .$defaultFn(() => Date.now()), + }, + (table) => [ + uniqueIndex("credit_balance_ledger_purchase_uniq").on(table.purchaseId), + index("credit_balance_ledger_balance_idx").on( + table.customerId, + table.creditSystemId, + ), + ], +); + // Per-credit-system addon balances (scoped pools) export const creditSystemBalances = sqliteTable( "credit_system_balances",