Main

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<string, unknown>,
+    ) => {
+      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<string, unknown>)
+          : {},
+      );
+      throw error;
+    }
+
     const featureSlugById = new Map(
       unbilled.features.map((feature) => [
         feature.featureId,

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<typeof createDb>;
+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<unknown>) => void;
+}): Promise<Customer | null> {
+  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,

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,33 +899,93 @@ 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(
     `[WEBHOOK] Credit purchase: customer=${dbCustomer.id}, credits=${creditsAmount}, qty=${resolvedQuantity}, pack=${creditPackId || "manual"}, system=${creditSystemId}`,
   );
 }
 
+async function applyCreditPurchaseBalance(
+  db: any,
+  purchaseId: string,
+  customerId: string,
+  creditSystemId: string,
+  amount: number,
+): Promise<void> {
+  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,