/**
 * API-key persistence over the app's KV state store (`App.create({ kv })`).
 *
 * Records are keyed by token hash (`ApiKey.keyFor`) — the plaintext token is
 * never stored — and validated against the scope catalog on read; unknown,
 * corrupt, expired, or invalid records resolve to `null`. Mutable billing
 * snapshots persist separately so reconciliation cannot overwrite newer key
 * metadata. An `id → recordKey` index makes revoke-by-id O(1). The persisted
 * {@link Record}/{@link MintInput} shapes live here (derived from
 * `ApiKey.schema` so they cannot drift from the resolved shape), while the
 * shared credential primitives (token format/generation, storage-key
 * derivation, redaction) live in `./ApiKey.js`.
 */

import * as z from 'zod/mini'

import * as ApiKey from './ApiKey.js'
import type * as Db from './db/Db.js'
import * as ApiKeyAdmissions from './db/tables/apiKeyAdmissions.js'
import * as ApiKeyOwnerTombstones from './db/tables/apiKeyOwnerTombstones.js'
import * as ApiKeyRevocations from './db/tables/apiKeyRevocations.js'
import * as Store from './internal/Store.js'
import * as Scope from './Scope.js'

const billingSnapshotSchema = z.object({
  active: z.boolean().check(z.describe('Whether billing is active.')),
  orgId: z.string().check(z.describe('Organization that produced the snapshot.')),
})

/** Maximum live API keys attributed to one organization. */
export const maxLiveKeysPerOrganization = 100

/** Maximum legacy organization-index entries examined during one admission attempt. */
export const maxLegacyAdmissionIndexes = 1_000

/** Zod schemas owned by the API-keys module. */
export namespace schema {
  /**
   * Fields accepted when minting a key, and the single source of truth for the
   * {@link MintInput} type and structural validation. {@link mint} validates
   * scope identifiers against its configured catalog.
   */
  export const MintInput = z.extend(z.omit(ApiKey.schema.ApiKey, { id: true }), {
    createdBy: z
      .optional(z.string())
      .check(z.describe('Identity creating the key (e.g. admin email), recorded for audit.')),
    expiresAt: z
      .optional(z.iso.datetime({ offset: true }))
      .check(z.describe('ISO 8601 expiry timestamp. Omit for a non-expiring key.')),
    // TODO: make required once orgs exist.
    orgId: z.optional(z.string()).check(z.describe('Owning organization id.')),
    scopes: z.readonly(z.array(z.string().check(z.minLength(1)))).check(
      z.refine((scopes) => new Set(scopes).size === scopes.length),
      z.describe("Granted scopes. '*' grants all scopes; duplicate scopes are rejected."),
    ),
  })

  /**
   * A persisted API-key record: the resolved `ApiKey.schema` plus
   * persistence/display fields. Extending the resolved schema keeps the resolved
   * and persisted shapes in lockstep.
   */
  export const Record = z.extend(ApiKey.schema.ApiKey, {
    createdAt: z.string().check(z.describe('ISO 8601 creation timestamp.')),
    createdBy: z.optional(z.string()).check(z.describe('Identity that created the key.')),
    expiresAt: z.optional(z.string()).check(z.describe('ISO 8601 expiry timestamp.')),
    tokenLast4: z.string().check(z.describe('Last 4 chars of the plaintext token.')),
  })
}

/**
 * A durable API-key record as persisted. A superset of the resolved
 * `ApiKey.ApiKey` with provisioning/display metadata. The plaintext token is
 * never stored — records are keyed by its hash (`ApiKey.keyFor`) and carry only
 * `tokenLast4` for display.
 */
export type Record = z.output<typeof schema.Record>

/**
 * Fields accepted when minting a key via {@link mint}. Uses the schema's
 * *input* type so defaulted fields (e.g. `environment`) are optional for
 * callers — the default is applied during `mint`'s validation.
 */
export type MintInput = z.input<typeof schema.MintInput>

/** Result of {@link mint}: the persisted record and its one-time token. */
export type MintResult = {
  /** The persisted key record (metadata only — never the token). */
  record: Record
  /** The plaintext token, shown once and unrecoverable afterward. */
  token: string
}

/**
 * Mints a key: validates the input (e.g. rejects an unknown scope at the write
 * boundary), persists its hashed record, and returns the record plus the
 * one-time plaintext token (shown once, unrecoverable afterward).
 *
 * @param state - The KV state store holding key records.
 * @param input - The key to mint.
 * @param options - Scope catalog used for validation.
 * @returns The persisted record and its one-time token.
 */
export async function mint(
  state: Store.State,
  input: MintInput,
  options: mint.Options = {},
): Promise<MintResult> {
  return mintValidated(state, validateMintInput(input, options), ApiKey.generateId())
}

async function mintValidated(
  state: Store.State,
  data: z.output<typeof schema.MintInput>,
  id: string,
): Promise<MintResult> {
  const orgId = data.orgId ?? id
  if (await isOwnerDeleted(state, { orgId, projectId: data.projectId }))
    throw new OwnerDeletedError()

  const time = new Date()
  const token = ApiKey.generateToken(data.environment)
  const recordKey = ApiKey.keyFor(token)
  const record: Record = {
    allowedIps: data.allowedIps,
    createdAt: time.toISOString(),
    environment: data.environment,
    id,
    orgId,
    scopes: data.scopes,
    tokenLast4: token.slice(-4),
    ...(data.billingActive === undefined ? {} : { billingActive: data.billingActive }),
    ...(data.createdBy === undefined ? {} : { createdBy: data.createdBy }),
    ...(data.expiresAt === undefined ? {} : { expiresAt: data.expiresAt }),
    ...(data.name === undefined ? {} : { name: data.name }),
    ...(data.projectId === undefined ? {} : { projectId: data.projectId }),
    ...(data.rateLimits === undefined ? {} : { rateLimits: data.rateLimits }),
  }

  // Bound the stored record by its expiry so the backend evicts it natively.
  const ttl =
    data.expiresAt === undefined
      ? undefined
      : Math.max(1, Date.parse(data.expiresAt) - time.getTime())
  await state.put(recordKey, JSON.stringify(record), { ttl })
  // Index `id → recordKey` so revoke-by-id is O(1) (records are keyed by
  // token hash, unrecoverable from the id alone). Shares the record's TTL so
  // both entries expire together.
  await state.put(idKeyFor(record.id), recordKey, { ttl })
  // Index `org → recordKey` (one entry per key, so concurrent mints never
  // race a shared value) backing org-scoped listing without a full scan.
  await state.put(orgKeyFor(record.orgId, record.id), recordKey, { ttl })
  return { record, token }
}

/** Mints an attributed key after atomically reserving an organization-wide live-key slot. */
export async function mintBounded(
  db: Db.Db,
  state: Store.State,
  input: MintInput,
  options: mint.Options = {},
): Promise<MintResult> {
  if (input.orgId === undefined) return mint(state, input, options)
  const orgId = input.orgId
  const data = validateMintInput(input, options)
  const id = ApiKey.generateId()
  const result = await db.transaction(async (tx) => {
    const owner = { orgId, projectId: data.projectId }
    if (!(await ApiKeyOwnerTombstones.lockActive(tx, owner))) throw new OwnerDeletedError()
    await ApiKeyAdmissions.lockOrganization(tx, orgId)
    const legacy = (await ApiKeyAdmissions.isBootstrapped(tx, orgId))
      ? { complete: true, records: [] }
      : await admissionRecordsForOrg(state, orgId, options)
    const admitted = await ApiKeyAdmissions.admit(tx, {
      expiresAt: data.expiresAt ?? null,
      id,
      limit: maxLiveKeysPerOrganization,
      legacy: legacy.records,
      legacyComplete: legacy.complete,
      orgId,
      projectId: data.projectId ?? null,
    })
    return admitted ? mintValidated(state, data, id) : null
  })
  if (!result) throw new LiveKeyLimitError(maxLiveKeysPerOrganization)
  return result
}

/** Durably fences key use before deleting its organization or project. The idempotent fence remains on failure so concurrent attempts cannot reopen the owner. */
export async function markOwnerDeleting(
  state: Store.State,
  owner: markOwnerDeleting.Owner,
): Promise<void> {
  await state.put(ownerDeletedKey(owner), 'deleting')
}

export declare namespace markOwnerDeleting {
  /** API-key owner being deleted. */
  type Owner = {
    /** Owning organization id. */
    orgId: string
    /** Project id when deleting only one project. */
    projectId?: string | undefined
  }
}

export declare namespace mint {
  /** Options for {@link mint}. */
  type Options = {
    /** Scope catalog accepted while minting the key. */
    scopeCatalog?: Scope.Catalog | undefined
  }
}

/**
 * Resolves a plaintext API key token to its metadata, or `null`. Expiry is
 * enforced on read — not just via the store's native TTL — so it holds on
 * backends that evict lazily and regardless of TTL granularity.
 *
 * @param state - The KV state store holding key records.
 * @param token - The presented plaintext token.
 * @param options - Scope catalog used for validation.
 * @returns The resolved key metadata, or `null`.
 */
export async function resolve(
  state: Store.State,
  token: string,
  options: resolve.Options = {},
): Promise<ApiKey.ApiKey | null> {
  const record = await read(
    state,
    ApiKey.keyFor(token),
    Date.now(),
    options.scopeCatalog ?? Scope.catalog,
  )
  if (!record || (await isOwnerDeleted(state, record))) return null
  return record
}

export declare namespace resolve {
  /** Options for {@link resolve}. */
  type Options = {
    /** Scope catalog accepted while resolving the key. */
    scopeCatalog?: Scope.Catalog | undefined
  }
}

/**
 * Reads a key record by id (metadata only — never the token), or `null`.
 *
 * @param state - The KV state store holding key records.
 * @param id - The key id (`key_…`).
 * @param options - Scope catalog used for validation.
 * @returns The record, or `null`.
 */
export async function get(
  state: Store.State,
  id: string,
  options: get.Options = {},
): Promise<Record | null> {
  const recordKey = await state.get(idKeyFor(id))
  if (!recordKey) return null
  return read(state, recordKey, Date.now(), options.scopeCatalog ?? Scope.catalog)
}

export declare namespace get {
  /** Options for {@link get}. */
  type Options = {
    /** Scope catalog accepted while reading the key. */
    scopeCatalog?: Scope.Catalog | undefined
  }
}

/**
 * Lists key records (metadata only — never tokens), newest first. Expired or
 * catalog-invalid records are skipped.
 *
 * @param state - The KV state store holding key records.
 * @param options - Optional owner filter.
 * @returns The records.
 */
export async function list(
  state: Store.State,
  options: list.Options = {},
): Promise<readonly Record[]> {
  const { keys } = await state.list({ prefix: ApiKey.recordPrefix })
  const values = await Store.getMany(
    state,
    keys.map(({ name }) => name),
  )
  const now = Date.now()
  const scopeCatalog = options.scopeCatalog ?? Scope.catalog
  const entries: applyBillingOverrides.Entry[] = []
  for (const { name } of keys) {
    const record = parse(values.get(name) ?? null, now, scopeCatalog)
    if (record && (!options.orgId || record.orgId === options.orgId))
      entries.push({ record, recordKey: name })
  }
  const overrides = await applyBillingOverrides(state, entries)
  const records = [...overrides.values()].filter((record) => record !== null)
  // Newest first; ids are random, so sort by creation time.
  return records.sort((a, b) => b.createdAt.localeCompare(a.createdAt))
}

export declare namespace list {
  /** Options for {@link list}. */
  type Options = {
    /** Restrict results to keys owned by this organization. */
    orgId?: string | undefined
    /** Scope catalog accepted while listing keys. */
    scopeCatalog?: Scope.Catalog | undefined
  }
}

/**
 * Lists an organization's key records through the org index (no full scan),
 * newest first. Dangling index entries (revoked or expired records) are
 * skipped and lazily deleted.
 *
 * @param state - The KV state store holding key records.
 * @param orgId - The owning organization id (`org_…`).
 * @param options - Optional environment/project filters.
 * @returns The records.
 */
export async function listByOrg(
  state: Store.State,
  orgId: string,
  options: listByOrg.Options = {},
): Promise<readonly Record[]> {
  return (
    await listByOrgPage(state, orgId, {
      ...options,
      limit: maxLiveKeysPerOrganization,
    })
  ).records
}

/** Reads one bounded page through an organization's key index. */
export async function listByOrgPage(
  state: Store.State,
  orgId: string,
  options: listByOrgPage.Options,
): Promise<listByOrgPage.Result> {
  const page = await state.list({
    ...(options.cursor === undefined ? {} : { cursor: options.cursor }),
    limit: options.limit,
    prefix: `${orgPrefix}${orgId}:`,
  })
  const { keys } = page
  const index = await Store.getMany(
    state,
    keys.map(({ name }) => name),
  )
  const recordKeys = keys.flatMap(({ name }) => {
    const recordKey = index.get(name)
    return recordKey ? [recordKey] : []
  })
  const values = await Store.getMany(state, recordKeys)
  const now = Date.now()
  const scopeCatalog = options.scopeCatalog ?? Scope.catalog
  const entries = recordKeys.flatMap((recordKey) => {
    const record = parse(values.get(recordKey) ?? null, now, scopeCatalog)
    return record ? [{ record, recordKey }] : []
  })
  const overrides = await applyBillingOverrides(state, entries)
  const records: Record[] = []
  for (const { name } of keys) {
    const recordKey = index.get(name)
    const record = recordKey ? (overrides.get(recordKey) ?? null) : null
    if (!record || record.orgId !== orgId) {
      if (options.cleanupDangling !== false) await state.delete(name)
      continue
    }
    if (options.environment && record.environment !== options.environment) continue
    if (!options.projectId || record.projectId === options.projectId) records.push(record)
  }
  return {
    ...(page.cursor === undefined ? {} : { cursor: page.cursor }),
    indexCount: keys.length,
    listComplete: page.listComplete ?? true,
    records: records.sort((a, b) => b.createdAt.localeCompare(a.createdAt)),
  }
}

export declare namespace listByOrgPage {
  /** Bounded organization-key page inputs. */
  type Options = listByOrg.Options & {
    /** Whether dangling index entries should be deleted while reading. */
    cleanupDangling?: boolean | undefined
    /** Continue after a previous page. */
    cursor?: string | undefined
    /** Maximum index entries to read. */
    limit: number
  }

  /** Bounded organization-key page. */
  type Result = {
    /** Cursor for the following page, when present. */
    cursor?: string | undefined
    /** Number of index entries examined in this page. */
    indexCount: number
    /** Whether every matching index was returned. */
    listComplete: boolean
    /** Valid key records in this page. */
    records: readonly Record[]
  }
}

export declare namespace listByOrg {
  /** Options for {@link listByOrg}. */
  type Options = {
    /** Restrict results to keys in this environment. */
    environment?: 'production' | 'sandbox' | undefined
    /** Restrict results to keys attributed to this project. */
    projectId?: string | undefined
    /** Scope catalog accepted while listing keys. */
    scopeCatalog?: Scope.Catalog | undefined
  }
}

/**
 * Revokes a key by id, deleting its record and index entries.
 *
 * @param state - The KV state store holding key records.
 * @param id - The key id (`key_…`).
 * @returns Whether a matching record was deleted.
 */
export async function revoke(state: Store.State, id: string): Promise<boolean> {
  const recordKey = await state.get(idKeyFor(id))
  if (!recordKey) return false
  // Parse loosely for the org index entry: even a record the catalog no
  // longer validates must clean up its index on revoke.
  const raw = await state.get(recordKey)
  const orgId = (() => {
    try {
      return raw ? (JSON.parse(raw) as { orgId?: string }).orgId : undefined
    } catch {
      return undefined
    }
  })()
  await state.delete(recordKey)
  await state.delete(billingKeyFor(recordKey))
  await state.delete(idKeyFor(id))
  if (orgId) await state.delete(orgKeyFor(orgId, id))
  return true
}

/** Fences and revokes a key, then idempotently removes its authoritative live-key slot. */
export async function revokeBounded(db: Db.Db, state: Store.State, id: string): Promise<boolean> {
  const fenced = await db.transaction(async (tx) => {
    await ApiKeyAdmissions.lock(tx, id)
    const [admission, record] = await Promise.all([ApiKeyAdmissions.get(tx, id), get(state, id)])
    const orgId = admission?.orgId ?? record?.orgId ?? null
    const projectId = admission?.projectId ?? record?.projectId ?? undefined
    const active =
      orgId === null ||
      (await ApiKeyOwnerTombstones.lockActive(tx, {
        orgId,
        ...(projectId === undefined || projectId === null ? {} : { projectId }),
      }))
    if ((record || admission) && active) await ApiKeyRevocations.markPending(tx, id, orgId)
    return { found: record !== null || admission !== undefined, orgId }
  })
  const revoked = await revoke(state, id)
  if (revoked)
    await db.transaction(async (tx) => {
      await ApiKeyAdmissions.lock(tx, id)
      await ApiKeyRevocations.mark(tx, id, fenced.orgId)
      await ApiKeyAdmissions.release(tx, id)
    })
  return revoked || fenced.found
}

/**
 * Updates a key's IP allowlist, display name, quotas, scopes, and attribution,
 * rewriting its record in place and moving the org index entry when the org changes.
 *
 * @param state - The KV state store holding key records.
 * @param id - The key id (`key_…`).
 * @param input - Key metadata to apply.
 * @param options - Scope catalog used for validation.
 * @returns The updated record, or `null` when the key is absent.
 */
export async function update(
  state: Store.State,
  id: string,
  input: update.Input,
  options: update.Options = {},
): Promise<Record | null> {
  if (input.allowedIps !== undefined) {
    const valid = ApiKey.schema.AllowedIps.safeParse(input.allowedIps)
    if (!valid.success) throw new Error(`invalid API key input: ${z.prettifyError(valid.error)}`)
  }
  if (input.rateLimits !== undefined) {
    const valid = ApiKey.schema.ApiKey.shape.rateLimits.safeParse(input.rateLimits)
    if (!valid.success) throw new Error(`invalid API key input: ${z.prettifyError(valid.error)}`)
  }
  if (
    input.scopes !== undefined &&
    (!validScopes(input.scopes, options.scopeCatalog ?? Scope.catalog) ||
      new Set(input.scopes).size !== input.scopes.length)
  )
    throw new Error('invalid API key input: invalid scopes')
  const recordKey = await state.get(idKeyFor(id))
  if (!recordKey) return null
  const record = await read(state, recordKey, Date.now(), options.scopeCatalog ?? Scope.catalog)
  if (!record) return null
  const owner = {
    orgId: input.orgId ?? record.orgId,
    projectId: input.projectId ?? record.projectId,
  }
  if (
    (input.orgId !== undefined || input.projectId !== undefined) &&
    (await isOwnerDeleted(state, owner))
  )
    throw new OwnerDeletedError()
  const reassigning = input.orgId !== undefined && input.orgId !== record.orgId
  const recordWithoutBilling = { ...record }
  delete recordWithoutBilling.billingActive

  const updated: Record = {
    ...(reassigning ? recordWithoutBilling : record),
    ...(input.allowedIps === undefined ? {} : { allowedIps: input.allowedIps }),
    ...(input.name === undefined ? {} : { name: input.name }),
    ...(input.orgId === undefined ? {} : { orgId: input.orgId }),
    ...(input.projectId === undefined ? {} : { projectId: input.projectId }),
    ...(input.rateLimits === undefined ? {} : { rateLimits: input.rateLimits }),
    ...(input.scopes === undefined ? {} : { scopes: input.scopes }),
  }
  const ttl =
    updated.expiresAt === undefined
      ? undefined
      : Math.max(1, Date.parse(updated.expiresAt) - Date.now())
  await state.put(recordKey, JSON.stringify(updated), { ttl })
  if (reassigning) {
    await state.delete(billingKeyFor(recordKey))
    await state.delete(orgKeyFor(record.orgId, id))
  }
  await state.put(orgKeyFor(updated.orgId, id), recordKey, { ttl })
  return updated
}

/** Updates a key while atomically enforcing owner fences and admission accounting. */
export async function updateBounded(
  db: Db.Db,
  state: Store.State,
  id: string,
  input: update.Input,
  options: updateBounded.Options = {},
): Promise<Record | null> {
  const result = await db.transaction(async (tx) => {
    await ApiKeyAdmissions.lock(tx, id)
    const record = await get(state, id, options)
    if (!record) return { type: 'missing' } as const
    if (
      options.expectedOwner &&
      (record.orgId !== options.expectedOwner.orgId ||
        record.projectId !== options.expectedOwner.projectId)
    )
      return { type: 'missing' } as const
    if (input.orgId === undefined && input.projectId === undefined)
      return { type: 'updated', updated: await update(state, id, input, options) } as const
    const owner = {
      orgId: input.orgId ?? record.orgId,
      projectId: input.projectId ?? record.projectId,
    }
    const owners = [owner]
    if (record.orgId !== record.id)
      owners.push({ orgId: record.orgId, projectId: record.projectId })
    if (!(await ApiKeyOwnerTombstones.lockActiveMany(tx, owners))) throw new OwnerDeletedError()
    await ApiKeyAdmissions.lockOrganization(tx, owner.orgId)
    const legacy = (await ApiKeyAdmissions.isBootstrapped(tx, owner.orgId))
      ? { complete: true, records: [] }
      : await admissionRecordsForOrg(state, owner.orgId, options)
    const admitted = await ApiKeyAdmissions.admit(tx, {
      currentOrgId: record.orgId === record.id ? undefined : record.orgId,
      expiresAt: normalizeExpiresAt(record.expiresAt),
      id,
      limit: maxLiveKeysPerOrganization,
      legacy: legacy.records,
      legacyComplete: legacy.complete,
      orgId: owner.orgId,
      projectId: owner.projectId ?? null,
    })
    if (!admitted) return { type: 'limit' } as const
    const updated = await update(state, id, input, options)
    if (!updated) await ApiKeyAdmissions.release(tx, id)
    return { type: 'updated', updated } as const
  })
  if (result.type === 'limit') throw new LiveKeyLimitError(maxLiveKeysPerOrganization)
  return result.type === 'missing' ? null : result.updated
}

export declare namespace updateBounded {
  /** Options for an admission-accounted key update. */
  type Options = update.Options & {
    /** Owner the locked record must still belong to. */
    expectedOwner?: Owner | undefined
  }

  /** Expected owner checked under the per-key lock. */
  type Owner = {
    /** Owning organization id (`org_…`). */
    orgId: string
    /** Attributed project id, or absent for an organization key. */
    projectId?: string | undefined
  }
}

export declare namespace update {
  /** Metadata applied by {@link update}; omitted fields keep their current value. */
  type Input = {
    /** Replacement client IP/CIDR allowlist. An empty list clears the restriction. */
    allowedIps?: readonly string[] | undefined
    /** New human-readable key name. */
    name?: string | undefined
    /** New owning organization id (`org_…`). */
    orgId?: string | undefined
    /** New attributed project id (`prj_…`). */
    projectId?: string | undefined
    /** Replacement per-key quota overrides. */
    rateLimits?: ApiKey.ApiKey['rateLimits'] | undefined
    /** Replacement granted scopes. */
    scopes?: readonly Scope.Id[] | undefined
  }

  /** Options for {@link update}. */
  type Options = {
    /** Scope catalog accepted while reading the key. */
    scopeCatalog?: Scope.Catalog | undefined
  }
}

/**
 * Re-attributes a key while preserving the existing public API.
 *
 * @param state - The KV state store holding key records.
 * @param id - The key id (`key_…`).
 * @param input - Attribution to apply.
 * @param options - Scope catalog used for validation.
 * @returns The updated record, or `null` when the key is absent.
 */
export function attribute(
  state: Store.State,
  id: string,
  input: attribute.Input,
  options: attribute.Options = {},
): Promise<Record | null> {
  return update(state, id, input, options)
}

export declare namespace attribute {
  /** Attribution applied by {@link attribute}; omitted fields keep their current value. */
  type Input = Pick<update.Input, 'orgId' | 'projectId'>

  /** Options for reading and updating the key record. */
  type Options = update.Options
}

/**
 * Stamps `billingActive` onto every key an organization owns in an environment,
 * re-syncing the snapshot the auth path reads when billing flips (activation or
 * lapse). The snapshot is stored separately so this background writer never
 * overwrites newer key metadata. Records already carrying the target value are
 * left untouched.
 *
 * @param state - The KV state store holding key records.
 * @param options - Owner, environment, and the new billing-active value.
 * @returns The number of records updated.
 */
export async function setBillingActive(
  state: Store.State,
  options: setBillingActive.Options,
): Promise<number> {
  const { active, environment, orgId, scopeCatalog } = options
  let updated = 0
  let cursor: string | undefined
  for (;;) {
    const page = await listByOrgPage(state, orgId, {
      cursor,
      environment,
      limit: maxLiveKeysPerOrganization,
      scopeCatalog,
    })
    for (const record of page.records) {
      if (record.billingActive === active) continue
      const recordKey = await state.get(idKeyFor(record.id))
      if (!recordKey) continue
      const ttl =
        record.expiresAt === undefined
          ? undefined
          : Math.max(1, Date.parse(record.expiresAt) - Date.now())
      await state.put(billingKeyFor(recordKey), JSON.stringify({ active, orgId }), { ttl })
      updated += 1
    }
    if (page.listComplete) break
    if (page.cursor === undefined) throw new Error('API-key list page is missing a cursor.')
    cursor = page.cursor
  }
  return updated
}

export declare namespace setBillingActive {
  /** Options for {@link setBillingActive}. */
  type Options = {
    /** The new billing-active value to stamp. */
    active: boolean
    /** Environment whose keys are re-stamped. */
    environment: 'production' | 'sandbox'
    /** Owning organization id (`org_…`). */
    orgId: string
    /** Scope catalog accepted while updating keys. */
    scopeCatalog?: Scope.Catalog | undefined
  }
}

/**
 * Rebuilds the `id` and `org` index entries for every stored record — the
 * run-once backfill making legacy keys visible to org-scoped listing. Records
 * themselves are never modified, so existing tokens keep resolving.
 *
 * @param state - The KV state store holding key records.
 * @param options - Scope catalog used for validation.
 * @returns Scanned and indexed record counts.
 */
export async function backfill(
  state: Store.State,
  options: backfill.Options = {},
): Promise<backfill.Result> {
  const { keys } = await state.list({ prefix: ApiKey.recordPrefix })
  let indexed = 0
  for (const { name } of keys) {
    const record = await read(state, name, Date.now(), options.scopeCatalog ?? Scope.catalog)
    if (!record) continue
    const ttl =
      record.expiresAt === undefined
        ? undefined
        : Math.max(1, Date.parse(record.expiresAt) - Date.now())
    await state.put(idKeyFor(record.id), name, { ttl })
    await state.put(orgKeyFor(record.orgId, record.id), name, { ttl })
    indexed += 1
  }
  return { indexed, scanned: keys.length }
}

export declare namespace backfill {
  /** Options for {@link backfill}. */
  type Options = {
    /** Scope catalog accepted while reading keys. */
    scopeCatalog?: Scope.Catalog | undefined
  }

  /** Result of {@link backfill}. */
  type Result = {
    /** Records whose index entries were written. */
    indexed: number
    /** Stored records scanned (including invalid or expired ones, skipped). */
    scanned: number
  }
}

/**
 * Reads and validates a stored record by storage key. Invalid, corrupt, or
 * expired records (`expiresAt <= now`) resolve to `null`.
 */
async function read(
  state: Store.State,
  key: string,
  now: number,
  scopeCatalog: Scope.Catalog,
): Promise<Record | null> {
  const record = parse(await state.get(key), now, scopeCatalog)
  if (!record) return null
  const overrides = await applyBillingOverrides(state, [{ record, recordKey: key }])
  return overrides.get(key) ?? null
}

function parse(raw: null | string, now: number, scopeCatalog: Scope.Catalog): Record | null {
  if (!raw) return null
  const value = (() => {
    try {
      return JSON.parse(raw) as unknown
    } catch {
      return undefined
    }
  })()
  if (value === undefined) return null
  const parsed = schema.Record.safeParse(StoredRecord.normalizeRateLimits(value))
  if (!parsed.success) return null
  const record = parsed.data
  if (!validScopes(record.scopes, scopeCatalog)) return null
  if (record.expiresAt !== undefined && Date.parse(record.expiresAt) <= now) return null
  return record
}

/** Applies separately persisted billing snapshots in one store read. */
async function applyBillingOverrides(
  state: Store.State,
  entries: readonly applyBillingOverrides.Entry[],
): Promise<ReadonlyMap<string, Record | null>> {
  const values = await Store.getMany(
    state,
    entries.map(({ recordKey }) => billingKeyFor(recordKey)),
  )
  return new Map(
    entries.map(({ record, recordKey }) => {
      const raw = values.get(billingKeyFor(recordKey)) ?? null
      if (raw === null) return [recordKey, record]
      const value = (() => {
        try {
          return JSON.parse(raw) as unknown
        } catch {
          return undefined
        }
      })()
      const snapshot = billingSnapshotSchema.safeParse(value)
      if (snapshot.success)
        return [
          recordKey,
          snapshot.data.orgId === record.orgId
            ? { ...record, billingActive: snapshot.data.active }
            : record,
        ]
      // Boolean snapshots predate organization reassignment and remain valid until rewritten.
      return [recordKey, typeof value === 'boolean' ? { ...record, billingActive: value } : null]
    }),
  )
}

declare namespace applyBillingOverrides {
  type Entry = {
    /** Parsed primary API-key record. */
    record: Record
    /** Storage key for the primary record. */
    recordKey: string
  }
}

namespace StoredRecord {
  type UnknownRecord = { [key: string]: unknown }

  /** Normalizes persisted pre-period quotas without accepting legacy fields at the write boundary. */
  export function normalizeRateLimits(value: unknown): unknown {
    if (!value || typeof value !== 'object' || Array.isArray(value)) return value
    const record = value as UnknownRecord
    const rateLimits = record['rateLimits']
    if (!rateLimits || typeof rateLimits !== 'object' || Array.isArray(rateLimits)) return value

    return {
      ...record,
      rateLimits: Object.fromEntries(
        Object.entries(rateLimits).map(([scope, rateLimit]) => {
          if (!rateLimit || typeof rateLimit !== 'object' || Array.isArray(rateLimit))
            return [scope, rateLimit]
          const perMinute = (rateLimit as UnknownRecord)['perMinute']
          return [
            scope,
            typeof perMinute === 'number' ? { limit: perMinute, period: 'minute' } : rateLimit,
          ]
        }),
      ),
    }
  }
}

/** Validates and normalizes API-key mint input. */
function validateMintInput(
  input: MintInput,
  options: mint.Options,
): z.output<typeof schema.MintInput> {
  const valid = schema.MintInput.safeParse(input)
  if (!valid.success) throw new Error(`invalid API key input: ${z.prettifyError(valid.error)}`)
  if (!validScopes(valid.data.scopes, options.scopeCatalog ?? Scope.catalog))
    throw new Error('invalid API key input: unknown scope')
  return {
    ...valid.data,
    ...(valid.data.expiresAt === undefined
      ? {}
      : { expiresAt: new Date(valid.data.expiresAt).toISOString() }),
  }
}

/** Maps bounded live KV records into the authoritative bootstrap shape. */
function admissionRecords(records: readonly Record[]) {
  return records.map((record) => ({
    expiresAt: normalizeExpiresAt(record.expiresAt),
    id: record.id,
    projectId: record.projectId ?? null,
  }))
}

/** Reconciles bounded index pages until 100 live legacy keys or index exhaustion. */
async function admissionRecordsForOrg(state: Store.State, orgId: string, options: get.Options) {
  const records: Record[] = []
  let indexes = 0
  let cursor: string | undefined
  for (;;) {
    const limit = Math.min(maxLiveKeysPerOrganization, maxLegacyAdmissionIndexes - indexes)
    const page = await listByOrgPage(state, orgId, {
      cleanupDangling: false,
      cursor,
      limit,
      scopeCatalog: options.scopeCatalog,
    })
    indexes += page.indexCount
    records.push(...page.records.slice(0, maxLiveKeysPerOrganization - records.length))
    if (page.listComplete) return { complete: true, records: admissionRecords(records) }
    if (records.length === maxLiveKeysPerOrganization)
      return { complete: false, records: admissionRecords(records) }
    if (indexes >= maxLegacyAdmissionIndexes)
      throw new LiveKeyLimitError(maxLiveKeysPerOrganization)
    if (page.cursor === undefined) throw new Error('API-key list page is missing a cursor.')
    cursor = page.cursor
  }
}

/** Normalizes a stored expiry for lexicographic database comparison. */
function normalizeExpiresAt(expiresAt: string | undefined): string | null {
  if (expiresAt === undefined) return null
  const time = new Date(expiresAt)
  return Number.isNaN(time.getTime()) ? null : time.toISOString()
}

/** Returns whether every granted scope belongs to the configured catalog. */
function validScopes(scopes: readonly string[], scopeCatalog: Scope.Catalog): boolean {
  return scopes.every((scope) => scope === Scope.wildcard || Scope.includes(scopeCatalog, scope))
}

/** Storage-key prefix for the `id → recordKey` index. */
const idPrefix = 'apikey_id:'

/** Storage key under which a key's `id → recordKey` index entry is persisted. */
function idKeyFor(id: string): string {
  return `${idPrefix}${id}`
}

/** Storage-key prefix for the per-key `org → recordKey` listing index. */
const orgPrefix = 'apikey_org:'

/** Storage key under which a key's org index entry is persisted. */
function orgKeyFor(orgId: string, id: string): string {
  return `${orgPrefix}${orgId}:${id}`
}

/** Storage-key prefix for durable owner deletion markers. */
const ownerDeletedPrefix = 'apikey_owner_deleted:'

/** Returns whether the organization or attributed project has been deleted. */
export async function isOwnerDeleted(
  state: Store.State,
  owner: isOwnerDeleted.Owner,
): Promise<boolean> {
  if (await state.get(ownerDeletedKey({ orgId: owner.orgId }))) return true
  return owner.projectId === undefined ? false : (await state.get(ownerDeletedKey(owner))) !== null
}

export declare namespace isOwnerDeleted {
  /** API-key owner whose deletion state is checked. */
  type Owner = markOwnerDeleting.Owner
}

/** Storage key for one organization or project deletion marker. */
function ownerDeletedKey(owner: markOwnerDeleting.Owner): string {
  return `${ownerDeletedPrefix}${owner.orgId}${owner.projectId ? `:${owner.projectId}` : ''}`
}

/** Storage-key prefix for mutable billing snapshots. */
const billingPrefix = 'apikey_billing:'

/** Storage key under which a key's current billing snapshot is persisted. */
function billingKeyFor(recordKey: string): string {
  return `${billingPrefix}${recordKey}`
}

/** Thrown when an API key owner is being deleted or is already deleted. */
export class OwnerDeletedError extends Error {
  override name = 'ApiKeys.OwnerDeletedError'
  constructor() {
    super('API key owner is deleted')
  }
}

/** Thrown when an organization has reached its live API-key cap. */
export class LiveKeyLimitError extends Error {
  override name = 'ApiKeys.LiveKeyLimitError'
  constructor(limit: number) {
    super(`Organization has reached its ${limit} live API-key limit`)
  }
}