import type { Selectable } from 'kysely'
import { sql } from 'kysely'
import { Value } from 'ox'

import * as Fees from '../../internal/Fees.js'
import * as Id from '../../internal/Id.js'
import type * as Db from '../Db.js'
import type * as db_Schema from '../Schema.js'
import * as SponsorshipAttributions from './sponsorshipAttributions.js'

/** Columns of the `sponsored_transactions` table, derived from `Schema.SponsoredTransaction`. */
export type Table = db_Schema.SponsoredTransaction

/** A stored sponsored-transaction row. */
export type Record = Selectable<Table>

/** How long failed fill intents remain eligible for on-chain reconciliation. */
export const failedIntentRecoveryTtlMs = 86_400_000

/**
 * Records one sponsorship commitment as `pending`, keyed by its fee-payer
 * sign payload. Written after fee-payer signing and before any broadcast, so
 * no sponsored transaction goes unrecorded; a later recording site for the
 * same envelope (raw submission after a fill intent) upgrades the row with
 * its transaction hash, reopening failed rows for finalization.
 *
 * @param db - The database.
 * @param input - Attribution and envelope facts captured at signing time.
 * @returns The stored record.
 */
export async function upsert(db: Db.Db, input: upsert.Input): Promise<Record> {
  const record: Record = {
    apiKeyId: input.apiKeyId,
    sponsorshipAttributionId: input.sponsorshipAttributionId ?? null,
    billable: input.billable,
    chainId: input.chainId,
    createdAt: new Date().toISOString(),
    currency: input.currency ?? null,
    environment: input.environment,
    feeAmount: null,
    feeMax: input.feeMax ?? null,
    feeToken: input.feeToken ?? null,
    finalizationAttemptedAt: null,
    finalizedAt: null,
    id: Id.generate('stx'),
    meterReportedAt: null,
    orgId: input.orgId,
    projectId: input.projectId ?? null,
    signPayload: input.signPayload,
    status: 'pending',
    transaction: input.transaction,
    transactionHash: input.transactionHash ?? null,
  }
  const stored = await db.kysely
    .insertInto('sponsored_transactions')
    .values(record)
    .onConflict((oc) =>
      oc
        .column('signPayload')
        .doUpdateSet((eb) => ({
          // Same sign payload = same envelope = same cap; only fills older nulls.
          currency: eb.fn.coalesce(eb.ref('sponsored_transactions.currency'), eb.ref('excluded.currency')), // prettier-ignore
          feeMax: eb.fn.coalesce(
            eb.ref('sponsored_transactions.feeMax'),
            eb.ref('excluded.feeMax'),
          ),
          // A raw submission fills the intent's hash; a duplicate intent never
          // clears an already-known hash.
          transactionHash: eb.fn.coalesce(
            eb.ref('excluded.transactionHash'),
            eb.ref('sponsored_transactions.transactionHash'),
          ),
          createdAt: sql`CASE WHEN ${eb.ref('sponsored_transactions.status')} = 'failed' THEN ${eb.ref('excluded.createdAt')} ELSE ${eb.ref('sponsored_transactions.createdAt')} END`,
          finalizedAt: sql`CASE WHEN ${eb.ref('sponsored_transactions.status')} = 'failed' THEN NULL ELSE ${eb.ref('sponsored_transactions.finalizedAt')} END`,
          status: sql`CASE WHEN ${eb.ref('sponsored_transactions.status')} = 'failed' THEN 'pending' ELSE ${eb.ref('sponsored_transactions.status')} END`,
        }))
        // A shared envelope cannot transfer billing attribution between owners.
        .where('sponsored_transactions.orgId', '=', input.orgId)
        .where(
          sql<boolean>`sponsored_transactions.sponsorship_attribution_id IS NOT DISTINCT FROM ${record.sponsorshipAttributionId}`,
        )
        .where(
          sql<boolean>`sponsored_transactions.project_id IS NOT DISTINCT FROM ${record.projectId}`,
        ),
    )
    .returningAll()
    .executeTakeFirst()
  if (!stored) throw new AttributionConflictError()
  return stored
}

export declare namespace upsert {
  /** Attribution and envelope facts recorded at sponsorship time. */
  type Input = {
    /** API key id (`key_…`) that requested sponsorship. */
    apiKeyId: string
    /** Canonical sponsorship attribution id snapshotted at sponsorship time. */
    sponsorshipAttributionId?: string | undefined
    /** Whether the sponsorship accrues billable spend; false for sandbox. */
    billable: boolean
    /** Chain the sponsored transaction targets. */
    chainId: number
    /** Lowercase fee currency snapshot (`usd`), when resolved. */
    currency?: string | undefined
    /** Key environment the sponsorship was requested under. */
    environment: Table['environment']
    /** Signed fee cap (`gas × maxFeePerGas`) in base units, when known. */
    feeMax?: string | undefined
    /** Fee token the sponsorship resolved, when known. */
    feeToken?: string | undefined
    /** Organization id (`org_…`) the spend attributes to. */
    orgId: string
    /** Project id (`prj_…`) the spend attributes to; omit for organization-level spend. */
    projectId?: string | undefined
    /** Fee-payer sign payload — a stable identity for the sponsored envelope. */
    signPayload: string
    /** Serialized sponsored transaction, kept for observability beyond log retention. */
    transaction: string
    /** Transaction hash; omit for fill intents, whose senders have not signed yet. */
    transactionHash?: string | undefined
  }
}

/**
 * Reads one sponsorship row by id.
 *
 * @param db - The database.
 * @param id - The sponsored-transaction id (`stx_…`).
 * @returns The record, or `undefined` when absent.
 */
export function get(db: Db.Db, id: string): Promise<Record | undefined> {
  return db.kysely
    .selectFrom('sponsored_transactions')
    .selectAll()
    .where('id', '=', id)
    .executeTakeFirst()
}

/** Lists sponsored transactions newest-first for administrative inspection. */
export function list(db: Db.Db, options: list.Options): Promise<Record[]> {
  let query = db.kysely.selectFrom('sponsored_transactions').selectAll()
  const cursor = options.cursor
  if (cursor !== undefined)
    query = query.where((eb) =>
      eb.or([
        eb('createdAt', '<', cursor.createdAt),
        eb.and([eb('createdAt', '=', cursor.createdAt), eb('id', '<', cursor.id)]),
      ]),
    )
  return query.orderBy('createdAt', 'desc').orderBy('id', 'desc').limit(options.limit).execute()
}

export declare namespace list {
  /** Pagination options for sponsored transactions. */
  type Options = {
    /** Exclusive lower bound from the previous page. */
    cursor?: Cursor | undefined
    /** Maximum rows returned. */
    limit: number
  }

  /** Cursor fields for the last transaction on the previous page. */
  type Cursor = {
    /** Sponsorship creation time. */
    createdAt: string
    /** Sponsored transaction id, used as a deterministic tie-breaker. */
    id: string
  }
}

/** Lists pending sponsorships, oldest first. */
export function listPending(db: Db.Db, options: listPending.Options = {}): Promise<Record[]> {
  let query = db.kysely
    .selectFrom('sponsored_transactions')
    .selectAll()
    .where('status', '=', 'pending')
    .orderBy('createdAt', 'asc')
    .orderBy('id', 'asc')
  if (options.limit !== undefined) query = query.limit(options.limit)
  return query.execute()
}

export declare namespace listPending {
  /** Options for {@link listPending}. */
  type Options = {
    /** Maximum rows to return. */
    limit?: number | undefined
  }
}

/** Claims the least-recently attempted pending sponsorships for one finalization pass. */
export function claimPending(db: Db.Db, options: claimPending.Options): Promise<Record[]> {
  return db.transaction(async (db) => {
    const rows = await db.kysely
      .selectFrom('sponsored_transactions')
      .selectAll()
      .where('status', '=', 'pending')
      .orderBy(sql`finalization_attempted_at nulls first`)
      .orderBy('createdAt', 'asc')
      .orderBy('id', 'asc')
      .limit(options.limit)
      .forUpdate()
      .skipLocked()
      .execute()
    if (rows.length === 0) return []
    await db.kysely
      .updateTable('sponsored_transactions')
      .set({ finalizationAttemptedAt: options.attemptedAt })
      .where(
        'id',
        'in',
        rows.map((row) => row.id),
      )
      .execute()
    return rows.map((row) => ({ ...row, finalizationAttemptedAt: options.attemptedAt }))
  })
}

export declare namespace claimPending {
  /** Claim timestamp and batch bound. */
  type Options = {
    /** Timestamp recorded before receipt lookups begin. */
    attemptedAt: string
    /** Maximum rows claimed by this pass. */
    limit: number
  }
}

/**
 * Lists active hash-less fill intents, oldest first, for reconciliation.
 *
 * @param db - The database.
 * @param options - Paging options.
 * @returns The fill intents.
 */
export function listIntents(db: Db.Db, options: listIntents.Options = {}): Promise<Record[]> {
  let query = db.kysely
    .selectFrom('sponsored_transactions')
    .selectAll()
    .where('status', '=', 'pending')
    .where('transactionHash', 'is', null)
    .orderBy('createdAt', 'asc')
    .orderBy('id', 'asc')
  if (options.limit !== undefined) query = query.limit(options.limit)
  return query.execute()
}

export declare namespace listIntents {
  /** Options for {@link listIntents}. */
  type Options = listPending.Options
}

/** Lists the oldest recoverable failed-intent timestamp for each chain. */
export function listRecoveryWindows(
  db: Db.Db,
  options: listRecoveryWindows.Options,
): Promise<listRecoveryWindows.Window[]> {
  return db.kysely
    .selectFrom('sponsored_transactions')
    .select('chainId')
    .select((eb) => eb.fn.min<string>('createdAt').as('createdAt'))
    .where('finalizedAt', '>=', options.failedSince)
    .where('status', '=', 'failed')
    .where('transactionHash', 'is', null)
    .groupBy('chainId')
    .execute()
}

export declare namespace listRecoveryWindows {
  /** Failed-intent recovery window options. */
  type Options = {
    /** Earliest terminal timestamp eligible for recovery. */
    failedSince: string
  }

  /** One chain's failed-intent recovery window. */
  type Window = {
    /** Chain containing recoverable failed intents. */
    chainId: number
    /** Oldest eligible intent creation timestamp. */
    createdAt: string
  }
}

/** Lists pending and recoverable failed intents matching candidate sign payloads. */
export function listRecoverableIntents(
  db: Db.Db,
  options: listRecoverableIntents.Options,
): Promise<Record[]> {
  if (options.signPayloads.length === 0) return Promise.resolve([])
  return db.kysely
    .selectFrom('sponsored_transactions')
    .selectAll()
    .where('signPayload', 'in', [...options.signPayloads])
    .where((eb) =>
      eb.or([
        eb('status', '=', 'pending'),
        eb.and([eb('finalizedAt', '>=', options.failedSince), eb('status', '=', 'failed')]),
      ]),
    )
    .where('transactionHash', 'is', null)
    .execute()
}

export declare namespace listRecoverableIntents {
  /** Reconciliation candidate lookup options. */
  type Options = {
    /** Earliest terminal timestamp eligible for recovery. */
    failedSince: string
    /** Candidate fee-payer sign payloads. */
    signPayloads: readonly string[]
  }
}

/**
 * Fills a reconciled intent's transaction hash and reopens failed intents.
 * A no-op once another pass already assigned the hash.
 *
 * @param db - The database.
 * @param id - The sponsored-transaction id (`stx_…`).
 * @param transactionHash - The matched on-chain transaction hash.
 * @param limit - The active period spend limit, when enforced.
 */
export async function assignTransactionHash(
  db: Db.Db,
  id: string,
  transactionHash: string,
  limit?: reserve.Limit,
): Promise<void> {
  try {
    await db.transaction(async (tx) => {
      const record = await get(tx, id)
      if (
        !record ||
        record.transactionHash !== null ||
        !['failed', 'pending'].includes(record.status)
      )
        return
      const failedSince = new Date(Date.now() - failedIntentRecoveryTtlMs).toISOString()
      if (
        record.status === 'failed' &&
        (record.finalizedAt === null || record.finalizedAt < failedSince)
      )
        return
      const limited =
        limit !== undefined &&
        (limit.billable === undefined ? record.billable : record.billable === limit.billable)
      if (record.status === 'failed' && limited) {
        const scope = limit.sponsorshipAttributionId
          ? `${record.orgId}:${limit.sponsorshipAttributionId}`
          : limit.projectId
            ? `${record.orgId}:${limit.projectId}`
            : record.orgId
        await sql`select pg_advisory_xact_lock(${billingLockClass}, hashtext(${scope}))`.execute(tx.kysely) // prettier-ignore
      }
      let query = tx.kysely
        .updateTable('sponsored_transactions')
        .set({
          ...(record.status === 'failed' ? { createdAt: new Date().toISOString() } : {}),
          finalizedAt: null,
          status: 'pending',
          transactionHash,
        })
        .where('id', '=', id)
        .where('status', '=', record.status)
        .where('transactionHash', 'is', null)
      if (record.status === 'failed')
        query = query.where((eb) =>
          eb.exists(
            eb
              .selectFrom('organizations')
              .select('organizations.id')
              .whereRef('organizations.id', '=', 'sponsored_transactions.orgId'),
          ),
        )
      const updated = await query.executeTakeFirst()
      if (updated.numUpdatedRows === 0n) return
      if (record.status === 'failed' && limited) {
        const committed = await spend(tx, {
          ...(limit.billable === undefined ? {} : { billable: limit.billable }),
          ...(limit.sponsorshipAttributionId === undefined
            ? {}
            : { sponsorshipAttributionId: limit.sponsorshipAttributionId }),
          chainIds: limit.chainIds,
          orgId: record.orgId,
          ...(limit.projectId === undefined ? {} : { projectId: limit.projectId }),
          since: limit.since,
        })
        if (committed > limit.max) throw new PeriodSpendLimitError()
      }
    })
  } catch (error) {
    // Keep the on-chain match eligible until budget becomes available.
    if (error instanceof PeriodSpendLimitError)
      await db.kysely
        .updateTable('sponsored_transactions')
        .set({ finalizedAt: new Date().toISOString() })
        .where('id', '=', id)
        .where('status', '=', 'failed')
        .where('transactionHash', 'is', null)
        .execute()
    throw error
  }
}

/**
 * Transitions a pending sponsorship to `finalized` with the receipt's actual
 * fee. A no-op when the row is no longer pending (idempotent re-runs).
 *
 * @param db - The database.
 * @param id - The sponsored-transaction id (`stx_…`).
 * @param input - The finalized fee facts.
 */
export async function finalize(db: Db.Db, id: string, input: finalize.Input): Promise<void> {
  await db.kysely
    .updateTable('sponsored_transactions')
    .set({ feeAmount: input.feeAmount, finalizedAt: input.finalizedAt, status: 'finalized' })
    .where('id', '=', id)
    .where('status', '=', 'pending')
    .execute()
}

export declare namespace finalize {
  /** Fee facts resolved from the transaction receipt. */
  type Input = {
    /** Actual fee paid in fee-token base units. */
    feeAmount: string
    /** When the sponsorship finalized (ISO 8601). */
    finalizedAt: string
  }
}

/**
 * Transitions a pending sponsorship to `failed` (never landed within the
 * pending TTL). A no-op when the row is no longer pending.
 *
 * @param db - The database.
 * @param id - The sponsored-transaction id (`stx_…`).
 * @param finalizedAt - When the sponsorship was marked failed (ISO 8601).
 */
export async function fail(db: Db.Db, id: string, finalizedAt: string): Promise<void> {
  await db.kysely
    .updateTable('sponsored_transactions')
    .set({ finalizedAt, status: 'failed' })
    .where('id', '=', id)
    .where('status', '=', 'pending')
    .execute()
}

/**
 * Records a sponsorship and enforces the period spend limit atomically. A
 * per-org advisory lock serializes concurrent sponsorships so each one's
 * signed cap is counted before the limit check; without it a burst reads
 * pre-burst spend and overshoots. Throws {@link PeriodSpendLimitError} (rolling
 * back the row) when the limit would be exceeded.
 *
 * @param db - The database (primary; never a cached replica).
 * @param input - The sponsorship to record.
 * @param limit - The period spend limit to enforce.
 * @returns The stored record.
 */
export async function reserve(
  db: Db.Db,
  input: upsert.Input,
  limit: reserve.Limit,
): Promise<Record> {
  return db.transaction(async (tx) => {
    // `hashtext` maps the org id into the advisory-lock keyspace; the class
    // namespaces it against other advisory-lock users. Released at tx end.
    const scope = limit.sponsorshipAttributionId
      ? `${input.orgId}:${limit.sponsorshipAttributionId}`
      : limit.projectId
        ? `${input.orgId}:${limit.projectId}`
        : input.orgId
    await sql`select pg_advisory_xact_lock(${billingLockClass}, hashtext(${scope}))`.execute(tx.kysely) // prettier-ignore
    const record = await upsert(tx, input)
    const limited =
      limit.billable === undefined ? record.billable : record.billable === limit.billable
    if (limited) {
      const committed = await spend(tx, {
        ...(limit.billable === undefined ? {} : { billable: limit.billable }),
        ...(limit.sponsorshipAttributionId === undefined
          ? {}
          : { sponsorshipAttributionId: limit.sponsorshipAttributionId }),
        chainIds: limit.chainIds,
        orgId: input.orgId,
        ...(limit.projectId === undefined ? {} : { projectId: limit.projectId }),
        since: limit.since,
      })
      if (committed > limit.max) throw new PeriodSpendLimitError()
    }
    return record
  })
}

/** Advisory-lock class namespacing per-org sponsorship serialization. */
const billingLockClass = 1112294220

export declare namespace reserve {
  /** The period spend limit enforced by {@link reserve}. */
  type Limit = {
    /** Canonical attribution scope for independent limits. */
    sponsorshipAttributionId?: string | undefined
    /** Billing snapshot whose rows consume this limit; omit for customer-billable spend. */
    billable?: boolean | undefined
    /** Chains whose spend counts (mainnet chain ids). */
    chainIds: readonly number[]
    /** Limit in fee-token base units. */
    max: bigint
    /** Project scope for independent limits; omit for the organization-wide budget. */
    projectId?: string | undefined
    /** Window start (ISO 8601). */
    since: string
  }
}

/**
 * Applies the current promotion policy and records the sponsorship atomically.
 * Exhausted promotions retain their window and fall back to customer billing.
 */
export function reservePromotion(
  db: Db.Db,
  input: reservePromotion.Input,
  options: reservePromotion.Options,
): Promise<reservePromotion.Result> {
  return db.transaction(async (tx) => {
    const scope = `${input.orgId}:${input.sponsorshipAttributionId}`
    await sql`select pg_advisory_xact_lock(${billingLockClass}, hashtext(${scope}))`.execute(tx.kysely) // prettier-ignore
    const existing = await tx.kysely
      .selectFrom('sponsored_transactions')
      .selectAll()
      .where('signPayload', '=', input.signPayload)
      .forUpdate()
      .executeTakeFirst()
    if (existing) {
      if (
        existing.orgId !== input.orgId ||
        (existing.sponsorshipAttributionId !== null &&
          existing.sponsorshipAttributionId !== input.sponsorshipAttributionId) ||
        existing.projectId !== (input.projectId ?? null)
      )
        throw new AttributionConflictError()
      // A pre-promotion row keeps its null attribution and billable snapshot
      // when policy changes before an idempotent retry.
      const stored = await upsert(
        tx,
        existing.sponsorshipAttributionId === null
          ? { ...input, billable: existing.billable, sponsorshipAttributionId: undefined }
          : input,
      )
      if (stored.billable && options.billingLimit) {
        await sql`select pg_advisory_xact_lock(${billingLockClass}, hashtext(${input.orgId}))`.execute(tx.kysely) // prettier-ignore
        const billed = await spend(tx, {
          chainIds: options.billingLimit.chainIds,
          orgId: input.orgId,
          since: options.billingLimit.since,
        })
        if (billed > options.billingLimit.max) throw new PeriodSpendLimitError()
      }
      if (!stored.billable && stored.sponsorshipAttributionId !== null) {
        const attribution = await SponsorshipAttributions.get(tx, {
          id: input.sponsorshipAttributionId,
          orgId: input.orgId,
        })
        if (!attribution?.startsAt) throw new PromotionActivationError()
        if (attribution.spendLimit !== null) {
          const committed = await spend(tx, {
            sponsorshipAttributionId: input.sponsorshipAttributionId,
            billable: false,
            chainIds: options.chainIds,
            orgId: input.orgId,
            since: attribution.startsAt,
          })
          if (committed > BigInt(attribution.spendLimit)) throw new PeriodSpendLimitError()
        }
      }
      return { record: stored, status: stored.billable ? 'billed' : 'subsidized' }
    }

    const organization = await tx.kysely
      .selectFrom('organizations')
      .select(['sponsorshipSubsidyDurationDays', 'sponsorshipSubsidyProjectSpendLimit'])
      .where('id', '=', input.orgId)
      .forShare()
      .executeTakeFirst()
    if (!organization || organization.sponsorshipSubsidyDurationDays === null)
      return { status: 'ineligible' }

    const current = await SponsorshipAttributions.get(tx, {
      id: input.sponsorshipAttributionId,
      orgId: input.orgId,
    })
    if (current?.endsAt && Date.parse(options.at) >= Date.parse(current.endsAt))
      return { status: 'ineligible' }

    const endsAt = new Date(
      Date.parse(options.at) + organization.sponsorshipSubsidyDurationDays * 86_400_000,
    ).toISOString()
    const attribution = await SponsorshipAttributions.activate(tx, {
      id: input.sponsorshipAttributionId,
      endsAt,
      orgId: input.orgId,
      spendLimit:
        organization.sponsorshipSubsidyProjectSpendLimit === null
          ? null
          : Value.from(
              organization.sponsorshipSubsidyProjectSpendLimit,
              Fees.tokenDecimals,
            ).toString(),
      startsAt: options.at,
    })
    let record = await upsert(tx, input)
    if (!attribution.startsAt) throw new PromotionActivationError()
    if (attribution.spendLimit !== null) {
      const committed = await spend(tx, {
        sponsorshipAttributionId: input.sponsorshipAttributionId,
        billable: false,
        chainIds: options.chainIds,
        orgId: input.orgId,
        since: attribution.startsAt,
      })
      if (committed > BigInt(attribution.spendLimit)) {
        record = await tx.kysely
          .updateTable('sponsored_transactions')
          .set({ billable: true })
          .where('id', '=', record.id)
          .returningAll()
          .executeTakeFirstOrThrow()
        if (options.billingLimit) {
          await sql`select pg_advisory_xact_lock(${billingLockClass}, hashtext(${input.orgId}))`.execute(tx.kysely) // prettier-ignore
          const billed = await spend(tx, {
            chainIds: options.billingLimit.chainIds,
            orgId: input.orgId,
            since: options.billingLimit.since,
          })
          if (billed > options.billingLimit.max) throw new PeriodSpendLimitError()
        }
      }
    }
    return { record, status: record.billable ? 'billed' : 'subsidized' }
  })
}

export declare namespace reservePromotion {
  /** Sponsored transaction with its canonical attribution snapshot. */
  type Input = upsert.Input & {
    /** Canonical sponsorship attribution id. */
    sponsorshipAttributionId: string
  }

  /** Attribution promotion window and cap. */
  type Options = {
    /** Timestamp of the recorded sponsorship that may activate the window. */
    at: string
    /** Customer billing limit used when the attribution promotion is exhausted. */
    billingLimit?: reserve.Limit | undefined
    /** Mainnet chains whose promotional spend counts. */
    chainIds: readonly number[]
  }

  /** Promotion reservation outcome. */
  type Result =
    | {
        /** Stored sponsorship. */
        record: Record
        /** Whether Tempo or the customer pays for the stored sponsorship. */
        status: 'billed' | 'subsidized'
      }
    | {
        /** The current organization or attribution policy does not grant a promotion. */
        status: 'ineligible'
      }
}

/**
 * Billable spend committed since a cutoff, in fee-token base units: finalized
 * rows at their actual fee plus pending rows at their signed fee cap, so
 * in-flight sponsorships consume limit budget until receipts land.
 *
 * @param db - The database.
 * @param options - Aggregation scope.
 * @returns The committed spend in base units.
 */
export async function spend(db: Db.Db, options: spend.Options): Promise<bigint> {
  const {
    sponsorshipAttributionId,
    billable,
    chainIds,
    environment = 'production',
    orgId,
    projectId,
    since,
  } = options
  let query = db.kysely
    .selectFrom('sponsored_transactions')
    .select([
      sql<string>`coalesce(sum(case when status = 'finalized' then fee_amount::numeric else 0 end), 0)`.as('finalized'), // prettier-ignore
      sql<string>`coalesce(sum(case when status = 'pending' then fee_max::numeric else 0 end), 0)`.as('pending'), // prettier-ignore
    ])
    .where('orgId', '=', orgId)
    .where('environment', '=', environment)
    // Redundant with the CASE arms, but lets the billing index's
    // (org_id, status, created_at) shape serve the range scan.
    .where('status', 'in', ['finalized', 'pending'])
    .where('chainId', 'in', [...chainIds])
    .where('createdAt', '>=', since)
  if (billable !== undefined) query = query.where('billable', '=', billable)
  if (sponsorshipAttributionId !== undefined)
    query = query.where('sponsorshipAttributionId', '=', sponsorshipAttributionId)
  if (projectId !== undefined) query = query.where('projectId', '=', projectId)
  // Production mirrors the sponsorship gate — only billable rows count. Sandbox
  // rows are non-billable by construction, so the figure is display-only.
  const row = await (
    environment === 'production' && billable === undefined
      ? query.where('billable', '=', true)
      : query
  ).executeTakeFirstOrThrow()
  return BigInt(row.finalized) + BigInt(row.pending)
}

export declare namespace spend {
  /** Options for {@link spend}. */
  type Options = {
    /** Canonical sponsorship attribution id whose spend counts. */
    sponsorshipAttributionId?: string | undefined
    /** Billing snapshot to count; omit for the existing production-billable behavior. */
    billable?: boolean | undefined
    /** Chains whose spend counts (mainnet chain ids). */
    chainIds: readonly number[]
    /** Environment whose spend counts; production mirrors the billing gate, sandbox is display-only. Defaults to `production`. */
    environment?: Table['environment'] | undefined
    /** Organization id (`org_…`) the spend attributes to. */
    orgId: string
    /** Project whose spend counts; omit for organization-wide spend. */
    projectId?: string | undefined
    /** Window start (ISO 8601); rows created earlier are out of scope. */
    since: string
  }
}

/**
 * Sponsorship usage bucketed by time for one organization: row counts and
 * committed fees per bucket. Fee semantics mirror {@link spend}: finalized
 * rows at their actual fee, pending rows at their signed cap, failed rows at
 * zero. Buckets with no rows are omitted.
 *
 * @param db - The database.
 * @param options - Aggregation scope.
 * @returns Time-ordered usage buckets.
 */
export async function usage(db: Db.Db, options: usage.Options): Promise<usage.Bucket[]> {
  // TODO: convert to an OLAP-friendly approach (rollup table or columnar store)
  // when per-org range scans grow slow.
  const { environment, from, interval, orgId, projectId, to } = options
  // `sql.lit` inlines the enum-validated interval so the GROUP BY expression
  // matches the SELECT expression textually; truncation happens in UTC wall
  // time regardless of the session time zone.
  const bucket = sql`date_trunc(${sql.lit(interval)}, (created_at::timestamptz) at time zone 'UTC')`
  let query = db.kysely
    .selectFrom('sponsored_transactions')
    .select([
      sql<string>`to_char(${bucket}, 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"')`.as('timestamp'),
      sql<string>`count(*)`.as('count'),
      sql<string>`count(*) filter (where status = 'failed')`.as('failed'),
      sql<string>`coalesce(sum(case when status = 'finalized' then fee_amount::numeric when status = 'pending' then fee_max::numeric else 0 end), 0)`.as('feeTotal'), // prettier-ignore
    ])
    .where('orgId', '=', orgId)
    // Text-range bounds keep the (org_id, created_at) index serving the scan.
    .where('createdAt', '>=', from)
    .where('createdAt', '<', to)
    .groupBy(bucket)
    .orderBy(bucket)
  if (environment !== undefined) query = query.where('environment', '=', environment)
  if (projectId !== undefined) query = query.where('projectId', '=', projectId)
  const rows = await query.execute()
  return rows.map((row) => ({
    count: Number(row.count),
    failed: Number(row.failed),
    feeTotal: BigInt(row.feeTotal),
    timestamp: row.timestamp,
  }))
}

export declare namespace usage {
  /** One time bucket of sponsorship usage. */
  type Bucket = {
    /** Sponsored transactions recorded in the bucket, any status. */
    count: number
    /** Sponsored transactions in the bucket whose status is `failed`. */
    failed: number
    /** Committed fees in fee-token base units: finalized fees plus pending caps. */
    feeTotal: bigint
    /** Bucket start (ISO 8601), aligned to UTC calendar boundaries. */
    timestamp: string
  }

  /** Options for {@link usage}. */
  type Options = {
    /** Key environment to restrict to; omit for all. */
    environment?: Table['environment'] | undefined
    /** Window start (ISO 8601), inclusive. */
    from: string
    /** Bucket width, aligned to UTC calendar boundaries. */
    interval: 'day' | 'hour' | 'month' | 'week'
    /** Organization id (`org_…`) the usage attributes to. */
    orgId: string
    /** Project id (`prj_…`) to restrict to; omit for all. */
    projectId?: string | undefined
    /** Window end (ISO 8601), exclusive. */
    to: string
  }
}

/**
 * Lists finalized billable rows not yet reported to the billing meter, oldest
 * first, for the metering job.
 *
 * @param db - The database.
 * @param options - Reporting scope.
 * @returns The unreported records.
 */
export function listUnreported(db: Db.Db, options: listUnreported.Options): Promise<Record[]> {
  let query = db.kysely
    .selectFrom('sponsored_transactions')
    .selectAll()
    .where('status', '=', 'finalized')
    .where('billable', '=', true)
    .where('chainId', 'in', [...options.chainIds])
    .where('meterReportedAt', 'is', null)
    // `id` breaks same-millisecond `finalizedAt` ties for a deterministic scan.
    .orderBy('finalizedAt', 'asc')
    .orderBy('id', 'asc')
  if (options.limit !== undefined) query = query.limit(options.limit)
  return query.execute()
}

export declare namespace listUnreported {
  /** Options for {@link listUnreported}. */
  type Options = {
    /** Chains whose spend is metered (mainnet chain ids). */
    chainIds: readonly number[]
    /** Maximum rows to return. */
    limit?: number | undefined
  }
}

/**
 * Marks a row as reported to the billing meter. A no-op once marked — the
 * durable exactly-once guard for the metering job.
 *
 * @param db - The database.
 * @param id - The sponsored-transaction id (`stx_…`).
 * @param at - When the report was acknowledged (ISO 8601).
 */
export async function markReported(db: Db.Db, id: string, at: string): Promise<void> {
  await db.kysely
    .updateTable('sponsored_transactions')
    .set({ meterReportedAt: at })
    .where('id', '=', id)
    .where('meterReportedAt', 'is', null)
    .execute()
}

/** Thrown by {@link reserve} when recording would exceed the org's period spend limit. */
export class PeriodSpendLimitError extends Error {
  override name = 'SponsoredTransactions.PeriodSpendLimitError'
}

/** Thrown when a promotional window cannot be activated consistently. */
export class PromotionActivationError extends Error {
  override name = 'SponsoredTransactions.PromotionActivationError'
}

/** Thrown when one signed envelope is presented with conflicting attribution. */
export class AttributionConflictError extends Error {
  override name = 'SponsoredTransactions.AttributionConflictError'
}