import { Hash, Hex, Value } from 'ox'
import type Stripe from 'stripe'

import * as ApiKey from '../../ApiKey.js'
import * as ApiKeys from '../../ApiKeys.js'
import type * as BillingSettings from '../../db/tables/billingSettings.js'
import * as Db from '../../db/Db.js'
import * as Fees from '../../internal/Fees.js'
import * as RequestUsage from '../../db/tables/requestUsage.js'
import type * as Scope from '../../Scope.js'
import * as SponsoredTransactions from '../../db/tables/sponsoredTransactions.js'
import type * as Store from '../../internal/Store.js'
import * as StripeCustomers from '../../db/tables/stripeCustomers.js'
import * as Viem from '../../internal/Viem.js'

/**
 * Whether the organization has any active billing source — the production
 * sponsorship gate's read. Storage stays one table per source; this resolver
 * is the union across them (Stripe today).
 *
 * @param db - The database.
 * @param orgId - The organization id (`org_…`).
 * @returns Whether any billing source is active.
 */
export async function active(
  db: Db.Db,
  orgId: string,
  environment: StripeCustomers.Record['environment'] = 'production',
): Promise<boolean> {
  return (await status(db, orgId, environment)) === 'active'
}

/**
 * The organization's billing status derived from its stored billing source,
 * or undefined when it has none. The sponsorship gate maps non-`active`
 * statuses to distinct refusal reasons.
 *
 * @param db - The database.
 * @param orgId - The organization id (`org_…`).
 * @returns The stored status, or undefined.
 */
export async function status(
  db: Db.Db,
  orgId: string,
  environment: StripeCustomers.Record['environment'] = 'production',
): Promise<StripeCustomers.Record['status'] | undefined> {
  const record = await StripeCustomers.get(db, orgId, environment)
  return record?.status
}

/**
 * Converts a decimal currency string (e.g. `'0.50'`) to fee-token base units.
 *
 * @param amount - The decimal string; at most {@link Fees.tokenDecimals} fraction digits.
 * @returns The amount in base units.
 */
export function toBaseUnits(amount: string): bigint {
  return Value.from(amount, Fees.tokenDecimals)
}

/**
 * Converts fee-token base units to a decimal currency string with trailing
 * zeros trimmed (e.g. `350000n` → `'0.35'`).
 *
 * @param units - The amount in base units.
 * @returns The decimal string.
 */
export function fromBaseUnits(units: bigint): string {
  return Value.format(units, Fees.tokenDecimals)
}

/**
 * Start of the limit window containing `now` (ISO 8601); `month` is the UTC
 * calendar month.
 *
 * @param period - The configured window.
 * @param now - Reference time; defaults to now.
 * @returns The window start.
 */
export function periodStart(period: BillingSettings.Record['period'], now = new Date()): string {
  // Single-member today; a switch lands with the second period.
  void period
  return new Date(Date.UTC(now.getUTCFullYear(), now.getUTCMonth(), 1)).toISOString()
}

/**
 * A metered billing dimension: a Stripe meter plus its metered monthly price.
 * Each kind bills as a distinct invoice line item on one shared subscription.
 */
export type MeterKind = 'apiRequests' | 'feePayerSpend'

/** Descriptor for one metered dimension. */
export type MeterDescriptor = {
  /** Human-readable meter/product display name. */
  displayName: string
  /** Meter event name; the Stripe meter aggregates `sum(value)` per customer per period. */
  eventName: string
  /** Versioned price lookup key; changing economics ships a new key, never a mutated price. */
  lookupKey: string
  /** Price per meter unit in cents (`unit_amount_decimal`). */
  unitAmountDecimal: string
}

/**
 * The metered dimensions and their unit economics. A price pins cents per meter
 * unit; changing them ships a new versioned `lookupKey`, never a mutated price.
 */
export const meters = {
  apiRequests: {
    displayName: 'API Requests',
    eventName: 'api_request_count',
    lookupKey: 'api-request-count-v1',
    // $0.0001/request → 0.01¢ per unit; the meter value is the integer request count.
    unitAmountDecimal: '0.01',
  },
  feePayerSpend: {
    displayName: 'Sponsor Usage',
    eventName: 'fee_payer_spend',
    lookupKey: 'fee-payer-spend-v1',
    // Meter values are 6-decimal base units: 100¢ per 1e6 units.
    unitAmountDecimal: '0.0001',
  },
} as const satisfies Record<MeterKind, MeterDescriptor>

/**
 * The metered dimensions billed in an environment. Production bills sponsorship
 * spend and request counts; sandbox bills request counts only.
 *
 * @param environment - The environment.
 * @returns The metered dimensions for that environment.
 */
export function kindsFor(environment: StripeCustomers.Record['environment']): readonly MeterKind[] {
  return environment === 'production' ? ['feePayerSpend', 'apiRequests'] : ['apiRequests']
}

/** Back-compat: the fee-payer meter event name. */
export const meterEventName = meters.feePayerSpend.eventName

/** Back-compat: the fee-payer metered price lookup key. */
export const priceLookupKey = meters.feePayerSpend.lookupKey

/** Fixture promises memoized per client, per kind; a failed ensure retries on the next call. */
const fixturesByClient = new WeakMap<Stripe, Map<MeterKind, Promise<ensureFixtures.Fixtures>>>()

/**
 * Ensures the account-level metering fixtures exist for a kind: its meter and
 * metered monthly price. Idempotent by lookup (meter `event_name`, price
 * `lookup_key`) and memoized per `(client, kind)`.
 *
 * @param stripe - The Stripe client.
 * @param kind - The metered dimension; defaults to fee-payer spend.
 * @returns The fixture ids.
 */
export function ensureFixtures(
  stripe: Stripe,
  kind: MeterKind = 'feePayerSpend',
): Promise<ensureFixtures.Fixtures> {
  let byKind = fixturesByClient.get(stripe)
  if (!byKind) {
    byKind = new Map()
    fixturesByClient.set(stripe, byKind)
  }
  const pending = byKind.get(kind)
  if (pending) return pending
  const descriptor = meters[kind]
  const created = (async () => {
    const meterList = await stripe.billing.meters.list({ limit: 100, status: 'active' })
    const meter =
      meterList.data.find((meter) => meter.event_name === descriptor.eventName) ??
      (await stripe.billing.meters.create({
        customer_mapping: { event_payload_key: 'stripe_customer_id', type: 'by_id' },
        default_aggregation: { formula: 'sum' },
        display_name: descriptor.displayName,
        event_name: descriptor.eventName,
        value_settings: { event_payload_key: 'value' },
      }))
    const prices = await stripe.prices.list({ limit: 1, lookup_keys: [descriptor.lookupKey] })
    const price =
      prices.data[0] ??
      (await stripe.prices.create({
        billing_scheme: 'per_unit',
        currency: 'usd',
        lookup_key: descriptor.lookupKey,
        product_data: { name: descriptor.displayName },
        recurring: { interval: 'month', meter: meter.id, usage_type: 'metered' },
        unit_amount_decimal: descriptor.unitAmountDecimal,
      }))
    return { meterId: meter.id, priceId: price.id }
  })()
  byKind.set(kind, created)
  created.catch(() => byKind.delete(kind))
  return created
}

export declare namespace ensureFixtures {
  /** Account-level metering fixture ids. */
  type Fixtures = {
    /** Spend meter id (`mtr_…`). */
    meterId: string
    /** Metered price id (`price_…`). */
    priceId: string
  }
}

/** Subscription statuses that no longer bill; a new subscription must be created. */
const terminalStatuses = ['canceled', 'incomplete_expired']

/**
 * Ensures the customer holds exactly one live subscription carrying one metered
 * item per requested {@link MeterKind}, creating the subscription or adding
 * missing items as needed. Metered usage bills the customer's default payment
 * method monthly.
 *
 * Fail-closed on more than one live managed subscription: duplicates would
 * double-bill the same meter usage, so the caller must reconcile manually.
 *
 * @param stripe - The Stripe client.
 * @param customerId - The Stripe customer id.
 * @param kinds - The metered dimensions to ensure; defaults to fee-payer spend.
 * @returns The per-kind creation state and whether a subscription was created.
 */
export async function ensureSubscription(
  stripe: Stripe,
  customerId: string,
  kinds: readonly MeterKind[] = ['feePayerSpend'],
): Promise<ensureSubscription.Result> {
  const requested = kinds.length ? kinds : (['feePayerSpend'] as const)
  // Resolve every requested kind's price; managed prices are what we reconcile.
  const fixtures = new Map<MeterKind, ensureFixtures.Fixtures>()
  for (const kind of requested) fixtures.set(kind, await ensureFixtures(stripe, kind))
  const priceToKind = new Map<string, MeterKind>()
  for (const [kind, fixture] of fixtures) priceToKind.set(fixture.priceId, kind)

  const subscriptions = await stripe.subscriptions.list({
    customer: customerId,
    limit: 100,
    status: 'all',
  })
  const liveManaged = subscriptions.data.filter(
    (subscription) =>
      !terminalStatuses.includes(subscription.status) &&
      subscription.items.data.some((item) => priceToKind.has(item.price.id)),
  )
  if (liveManaged.length > 1) throw new DuplicateSubscriptionError(customerId)

  const items: Record<string, ensureSubscription.ItemResult> = {}
  const subscription = liveManaged[0]
  if (subscription) {
    const present = new Set(
      subscription.items.data
        .map((item) => priceToKind.get(item.price.id))
        .filter((kind): kind is MeterKind => kind !== undefined),
    )
    for (const kind of requested) {
      if (present.has(kind)) {
        items[kind] = { created: false }
        continue
      }
      // Add the missing item; the idempotency key absorbs concurrent adds.
      await stripe.subscriptionItems.create(
        { price: fixtures.get(kind)!.priceId, subscription: subscription.id },
        { idempotencyKey: `${customerId}-item-${meters[kind].lookupKey}` },
      )
      items[kind] = { created: true }
    }
    return { created: false, items }
  }

  // The idempotency key absorbs concurrent activations: a duplicate metered
  // subscription would double-bill the same meter usage.
  await stripe.subscriptions.create(
    {
      collection_method: 'charge_automatically',
      customer: customerId,
      items: requested.map((kind) => ({ price: fixtures.get(kind)!.priceId })),
    },
    { idempotencyKey: `${customerId}-metered-sub-v2` },
  )
  for (const kind of requested) items[kind] = { created: true }
  return { created: true, items }
}

export declare namespace ensureSubscription {
  /** Per-kind item reconciliation state. */
  type ItemResult = {
    /** Whether this call added the item (a fresh item bills nothing before its start). */
    created: boolean
  }

  /** Outcome of {@link ensureSubscription}. */
  type Result = {
    /** Whether this call created the subscription (none was live before). */
    created: boolean
    /** Per-kind item state, keyed by {@link MeterKind}. */
    items: Partial<Record<MeterKind, ItemResult>>
  }
}

/** More than one live managed subscription for a customer; requires manual reconcile. */
export class DuplicateSubscriptionError extends Error {
  constructor(customerId: string) {
    super(`multiple live metered subscriptions for customer ${customerId}`)
    this.name = 'Billing.DuplicateSubscriptionError'
  }
}

/**
 * Creates the metering job: each tick reports finalized, unreported billable
 * mainnet spend to the Stripe meter and marks rows reported. A host runtime
 * drives {@link createReporter.Reporter.tick} on a timer, like the
 * sponsorship finalizer.
 */
export function createReporter(options: createReporter.Options): createReporter.Reporter {
  const { batchSize = 100, chainIds = [Viem.chainId.mainnet], db, stripe } = options
  return {
    async tick() {
      // Resolve per tick so Workers factories build a fresh Hyperdrive-pooled
      // connection each run.
      const database = Db.get(db)
      const rows = await SponsoredTransactions.listUnreported(database, {
        chainIds,
        limit: batchSize,
      })
      // One customer lookup and one subscription ensure per org per tick.
      const customers = new Map<string, StripeCustomers.Record | undefined>()
      const subscriptions = new Map<string, ensureSubscription.Result>()
      let failed = 0
      let reported = 0
      let skipped = 0
      for (const row of rows) {
        // One poisoned row must not stall the queue: it stays unmarked for
        // retry while the rest keep reporting.
        try {
          let customer = customers.get(row.orgId)
          if (!customers.has(row.orgId)) {
            customer = await StripeCustomers.get(database, row.orgId)
            customers.set(row.orgId, customer)
          }
          if (customer && customer.status !== 'canceled') {
            // Self-heal: guarantee the metered subscription before reporting;
            // meter events without one never reach an invoice. Pass the full
            // production kind set so concurrent creates share idempotency.
            let subscription = subscriptions.get(row.orgId)
            if (!subscription) {
              subscription = await ensureSubscription(
                stripe,
                customer.stripeCustomerId,
                kindsFor('production'),
              )
              subscriptions.set(row.orgId, subscription)
            }
            // Meter timestamps must fall inside Stripe's 35-day ingestion
            // window; older backlog bills into the current period instead. A
            // just-created fee-payer item bills nothing before its start, so
            // its backlog stamps at now too.
            const finalizedAt = new Date(row.finalizedAt!).getTime()
            const recent = Date.now() - finalizedAt < 34 * 24 * 60 * 60 * 1000
            const stamp = recent && !subscription.items.feePayerSpend?.created
            // The row id makes short-window retries idempotent on Stripe's
            // side; `meter_reported_at` is the durable exactly-once guard.
            try {
              await stripe.billing.meterEvents.create({
                event_name: meterEventName,
                identifier: row.id,
                payload: { stripe_customer_id: customer.stripeCustomerId, value: row.feeAmount! },
                ...(stamp ? { timestamp: Math.floor(finalizedAt / 1000) } : {}),
              })
            } catch (error) {
              // Stripe may persist the event before the Worker records its ack.
              // Its duplicate response proves this exact row was already metered.
              if (!isDuplicateMeterEvent(error, row.id)) throw error
            }
            reported++
          } else {
            // Deleted orgs and Stripe-deleted (canceled) customers keep their
            // ledger rows; nobody remains to invoice.
            skipped++
          }
          // Mark only after the Stripe ack (or a definitive no-customer skip).
          await SponsoredTransactions.markReported(database, row.id, new Date().toISOString())
        } catch (error) {
          failed++
          console.error(`Billing.createReporter: reporting row ${row.id} failed`, error)
        }
      }
      return { failed, reported, skipped }
    },
  }
}

/** Whether Stripe confirms that the exact meter event identifier already exists. */
function isDuplicateMeterEvent(error: unknown, identifier: string): boolean {
  return (
    error instanceof Error &&
    error.message === `An event already exists with identifier ${identifier}.`
  )
}

export declare namespace createReporter {
  /** Options for {@link createReporter}. */
  type Options = {
    /** Maximum rows reported per tick. @default 100 */
    batchSize?: number | undefined
    /** Chains whose spend is metered. @default mainnet */
    chainIds?: readonly number[] | undefined
    /** Database holding sponsorship rows, or a factory resolved per tick (Workers/Hyperdrive). */
    db: Db.Source
    /** Stripe client used to send meter events. */
    stripe: Stripe
  }

  /** A scheduled metering job. */
  type Reporter = {
    /** Runs one reporting pass. Failed rows stay unmarked and retry next tick. */
    tick(): Promise<{ failed: number; reported: number; skipped: number }>
  }
}

/** Default settlement delay: only seal buckets older than this, absorbing normal ingest lag. */
const defaultSettlementDelayMs = 15 * 60 * 1000

/** Default correction window: how far back a tick reconciles late-arriving deltas. */
const defaultWindowMs = 48 * 60 * 60 * 1000

/** One hour in milliseconds. */
const hourMs = 60 * 60 * 1000

/**
 * Creates the request-metered billing job: each tick reports deduped billable
 * request counts to the `api_request_count` meter, per environment, using that
 * environment's Stripe client. Exactly-once from the request analytics stream
 * via a Postgres outbox with frozen counts:
 *
 * - Deduped counts come from `request_events` (`uniqExact(request_id)`), never
 *   a raw `count()`, so queue retries never over-bill.
 * - Sandbox rows bill only when billing was active at request time
 *   (`billing_active = 1`); production rows always bill.
 * - Each tick first re-sends any `pending` events (crash recovery, idempotent
 *   on `identifier`), then seals new additive deltas for buckets older than the
 *   settlement delay whose count has grown since the last report.
 *
 * A host runtime drives {@link createRequestUsageReporter.Reporter.tick} on a
 * timer, like the fee-payer meter reporter.
 */
export function createRequestUsageReporter(
  options: createRequestUsageReporter.Options,
): createRequestUsageReporter.Reporter {
  const {
    db,
    now = () => Date.now(),
    readCounts,
    sandboxStripe,
    settlementDelayMs = defaultSettlementDelayMs,
    stripe,
    windowMs = defaultWindowMs,
  } = options
  const clients: { client: Stripe; environment: StripeCustomers.Record['environment'] }[] = [
    { client: stripe, environment: 'production' },
    ...(sandboxStripe ? [{ client: sandboxStripe, environment: 'sandbox' as const }] : []),
  ]
  return {
    async pending() {
      return Promise.all(
        clients.map(async ({ environment }) => {
          const status = await RequestUsage.getPendingStatus(Db.get(db), environment)
          return {
            count: status.count,
            environment,
            oldestPendingAgeSeconds: status.oldestAttemptedAt
              ? Math.max(
                  0,
                  Math.floor((now() - new Date(status.oldestAttemptedAt).getTime()) / 1_000),
                )
              : 0,
          }
        }),
      )
    },
    async tick() {
      const database = Db.get(db)
      let failed = 0
      let reported = 0
      let skipped = 0
      for (const { client, environment } of clients) {
        const nowMs = now()
        const sealCutoff = startOfHour(nowMs - settlementDelayMs)
        const windowStart = new Date(new Date(sealCutoff).getTime() - windowMs).toISOString()

        // Pass A — crash recovery: re-send frozen pending events. Idempotent on
        // `identifier` within Stripe's dedup window; the minutely cadence
        // re-marks well inside it. Buckets that still fail stay blocked so the
        // seal pass never stacks a second delta on them.
        const blocked = new Set<string>()
        for (const event of await RequestUsage.listPending(database, environment)) {
          try {
            await client.billing.meterEvents.create({
              event_name: meters.apiRequests.eventName,
              identifier: event.identifier,
              payload: {
                stripe_customer_id: event.stripeCustomerId,
                value: String(event.deltaCount),
              },
            })
            if (await RequestUsage.settle(database, event.identifier, new Date().toISOString()))
              reported++
          } catch (error) {
            failed++
            blocked.add(bucketKey(event.orgId, event.bucketStart))
            await RequestUsage.markError(database, event.identifier, String(error))
          }
        }

        // Pass B — seal new deltas for closed buckets whose count has grown.
        const counts = await readCounts({ environment, from: windowStart, to: sealCutoff })
        const customers = new Map<string, StripeCustomers.Record | undefined>()
        const subscriptions = new Map<string, ensureSubscription.Result>()
        for (const { bucketStart, count, orgId } of counts) {
          const key = { bucketStart, environment, orgId }
          if (blocked.has(bucketKey(orgId, bucketStart))) continue
          try {
            const watermark = await RequestUsage.getBucket(database, key)
            const delta = count - (watermark?.reportedCount ?? 0)
            if (delta <= 0) continue

            let customer = customers.get(orgId)
            if (!customers.has(orgId)) {
              customer = await StripeCustomers.get(database, orgId, environment)
              customers.set(orgId, customer)
            }
            if (!customer || customer.status === 'canceled') {
              skipped++
              continue
            }

            let subscription = subscriptions.get(orgId)
            if (!subscription) {
              subscription = await ensureSubscription(
                client,
                customer.stripeCustomerId,
                kindsFor(environment),
              )
              subscriptions.set(orgId, subscription)
            }

            // The observed watermark is stable for every tick claiming this delta.
            const sequence = watermark?.reportedCount ?? 0
            const identifier = usageIdentifier(environment, orgId, bucketStart, sequence)
            const inserted = await RequestUsage.insertPending(database, {
              ...key,
              deltaCount: delta,
              identifier,
              sequence,
              stripeCustomerId: customer.stripeCustomerId,
            })
            // A concurrent tick already claimed this delta; leave it to that run.
            if (!inserted) continue

            // Backdate to the bucket hour unless the item was just created (it
            // bills nothing before its start, so its backlog stamps at now).
            const stamp = !subscription.items.apiRequests?.created
            await client.billing.meterEvents.create({
              event_name: meters.apiRequests.eventName,
              identifier,
              payload: { stripe_customer_id: customer.stripeCustomerId, value: String(delta) },
              ...(stamp ? { timestamp: Math.floor(new Date(bucketStart).getTime() / 1000) } : {}),
            })
            if (await RequestUsage.settle(database, identifier, new Date().toISOString()))
              reported++
          } catch (error) {
            failed++
            // Cloudflare flattens multiple console arguments, so keep the driver details on one structured entry.
            const cause = error instanceof Error ? error : undefined
            console.error({
              bucketStart,
              component: 'request_usage_meter',
              environment,
              error: ApiKey.redact(cause?.message ?? String(error)).slice(0, 500),
              ...(cause &&
              'code' in cause &&
              (typeof cause.code === 'string' || typeof cause.code === 'number')
                ? { errorCode: ApiKey.redact(String(cause.code)).slice(0, 100) }
                : {}),
              errorType: cause?.name ?? typeof error,
              orgId,
              ...(cause?.stack ? { stack: ApiKey.redact(cause.stack).slice(0, 4_000) } : {}),
            })
          }
        }
      }
      return { failed, reported, skipped }
    },
  }
}

export declare namespace createRequestUsageReporter {
  /** Deduped billable request count for one `(org, hour)` bucket. */
  type Count = {
    /** UTC hour start (ISO 8601). */
    bucketStart: string
    /** Deduped billable request count. */
    count: number
    /** Organization id (`org_…`). */
    orgId: string
  }

  /** Options for {@link createRequestUsageReporter}. */
  type Options = {
    /** Database holding the outbox/watermarks, or a factory resolved per tick (Workers/Hyperdrive). */
    db: Db.Source
    /** Clock seam for tests; defaults to `Date.now`. */
    now?: (() => number) | undefined
    /** Reads deduped billable counts per `(org, hour)` for an environment window. */
    readCounts: (options: {
      environment: StripeCustomers.Record['environment']
      from: string
      to: string
    }) => Promise<readonly Count[]>
    /** Sandbox (test-mode) Stripe client; omit to skip sandbox billing. */
    sandboxStripe?: Stripe | undefined
    /** Only seal buckets older than this. @default 15min */
    settlementDelayMs?: number | undefined
    /** Production Stripe client. */
    stripe: Stripe
    /** How far back a tick reconciles late deltas. @default 48h */
    windowMs?: number | undefined
  }

  /** A scheduled request-metering job. */
  type Reporter = {
    /** Reads the current retry backlog without sending meter events. */
    pending(): Promise<readonly Pending[]>
    /** Runs one reporting pass across production (and sandbox when configured). */
    tick(): Promise<{ failed: number; reported: number; skipped: number }>
  }

  /** Current durable retry backlog for one billing environment. */
  type Pending = {
    count: number
    environment: StripeCustomers.Record['environment']
    oldestPendingAgeSeconds: number
  }
}

/** Composite key for the in-tick blocked-bucket set. */
function bucketKey(orgId: string, bucketStart: string) {
  return `${orgId}|${bucketStart}`
}

/** Start of the UTC hour containing `ms` (ISO 8601). */
function startOfHour(ms: number) {
  return new Date(Math.floor(ms / hourMs) * hourMs).toISOString()
}

/**
 * Deterministic meter-event identifier, bounded under Stripe's 100-char limit:
 * `rq_${env}_${sha(orgId)[:12]}_${bucketEpochHour}_${observedWatermark}`.
 */
function usageIdentifier(
  environment: StripeCustomers.Record['environment'],
  orgId: string,
  bucketStart: string,
  observedWatermark: number,
) {
  const hash = Hash.keccak256(Hex.fromString(orgId)).slice(2, 14)
  const epochHour = Math.floor(new Date(bucketStart).getTime() / hourMs)
  return `rq_${environment}_${hash}_${epochHour}_${observedWatermark}`
}

/**
 * Every organization's billing-active flag in an environment — the driver set
 * for reconciling cached snapshots against this table. Provider-agnostic seam:
 * callers reconcile against `{ orgId, active }`, never the Stripe source shape.
 *
 * @param db - The database.
 * @param environment - The environment to enumerate.
 * @returns One `{ orgId, active }` per billing source in the environment.
 */
export async function listActive(
  db: Db.Db,
  environment: StripeCustomers.Record['environment'],
): Promise<{ active: boolean; orgId: string }[]> {
  const records = await StripeCustomers.listByEnvironment(db, environment)
  return records.map((record) => ({ active: record.status === 'active', orgId: record.orgId }))
}

/**
 * Creates the snapshot reconciler: each tick re-stamps the `billingActive`
 * flag cached on API-key records from the authoritative billing table, so a
 * missed or failed webhook re-sync self-heals within a tick. A host runtime
 * drives {@link createKeyBillingSyncer.Syncer.tick} on a timer, like the meter
 * reporter. Off the request path, so the auth hot path never reads billing.
 *
 * Only environments where keys carry a billing snapshot are reconciled
 * (sandbox today); production keys carry none.
 */
export function createKeyBillingSyncer(
  options: createKeyBillingSyncer.Options,
): createKeyBillingSyncer.Syncer {
  const { db, environments = ['sandbox'], kv, scopeCatalog } = options
  return {
    async tick() {
      // Resolve per tick so Workers factories build a fresh Hyperdrive-pooled
      // connection each run.
      const database = Db.get(db)
      let checked = 0
      let updated = 0
      for (const environment of environments) {
        const rows = await listActive(database, environment)
        for (const { active, orgId } of rows) {
          checked += 1
          // `setBillingActive` writes only drifted records, so a converged
          // fleet costs list reads and no writes.
          updated += await ApiKeys.setBillingActive(kv, {
            active,
            environment,
            orgId,
            scopeCatalog,
          })
        }
      }
      return { checked, updated }
    },
  }
}

export declare namespace createKeyBillingSyncer {
  /** Options for {@link createKeyBillingSyncer}. */
  type Options = {
    /** Database holding billing sources, or a factory resolved per tick (Workers/Hyperdrive). */
    db: Db.Source
    /** Environments to reconcile. @default ['sandbox'] */
    environments?: readonly StripeCustomers.Record['environment'][] | undefined
    /** KV state store holding API-key records. */
    kv: Store.State
    /** Scope catalog accepted while reconciling API-key records. */
    scopeCatalog?: Scope.Catalog | undefined
  }

  /** A scheduled snapshot-reconcile job. */
  type Syncer = {
    /** Runs one reconcile pass, re-stamping drifted key records. */
    tick(): Promise<{ checked: number; updated: number }>
  }
}