Function recordCumulativeUsage

apps/sim/lib/billing/core/usage-log.ts:451–518

(
  params: RecordCumulativeUsageParams
)

 * process dies between legs — each leg's cost is durably recorded as it lands.
 */
export async function recordCumulativeUsage(
  params: RecordCumulativeUsageParams
): Promise<RecordCumulativeUsageResult> {
  const { userId, workspaceId, source, model, cost, eventKey, metadata } = params

  // Resolved before the locked transaction on purpose: resolving inside it
  // ran the subscription lookups on the global pool while this tx already
  // held a pooled connection plus the advisory lock, so under load N
  // first-flush transactions each pinned a connection while waiting for one
  // more — starving the pool and queueing every same-key flush (and the Go
  // side's retries) behind the stall.
  const billingContext = await resolveBillingContext(userId)

  return db.transaction(async (tx) => {
    // Serialize all flushes for this request (lock auto-releases at tx end),
    // with a bounded wait so a pathological holder fails this flush fast and
    // lets the caller retry instead of hanging the connection.
    await tx.execute(
      sql`select set_config('lock_timeout', ${`${CUMULATIVE_FLUSH_LOCK_TIMEOUT_MS}ms`}, true)`
    )
    await tx.execute(sql`select pg_advisory_xact_lock(hashtextextended(${eventKey}, 0))`)

    const [existing] = await tx
      .select({ id: usageLog.id, cost: usageLog.cost })
      .from(usageLog)
      .where(eq(usageLog.eventKey, eventKey))
      .limit(1)

    const recorded = existing ? Number.parseFloat(existing.cost) : 0
    const { shouldBill, delta, newTotal } = resolveCumulativeTopUp(recorded, cost)

    if (!shouldBill) {
      return { billed: false, delta: 0, total: recorded }
    }

    if (existing) {
      // Top up the single row to the new (higher) cumulative; the
      // period total is SUM(usage_log.cost), so this lifts it by the delta.
      await tx
        .update(usageLog)
        .set({ cost: newTotal.toString(), metadata: metadata ?? null })
        .where(eq(usageLog.id, existing.id))
    } else {
      // First flush for this request: insert the canonical row with the
      // pre-resolved billing context. Runs in the same tx + advisory lock.
      await recordUsage({
        userId,
        workspaceId,
        tx,
        billingEntity: billingContext.billingEntity,
        billingPeriod: billingContext.billingPeriod,
        entries: [
          {
            category: 'model',
            source,
            description: model,
            cost: newTotal,
            eventKey,