import { sql } from 'kysely'

import type * as Db from '../Db.js'
import type * as db_Schema from '../Schema.js'
import * as ApiKeyRevocations from './apiKeyRevocations.js'

/** Columns of the `api_key_admissions` table. */
export type Table = db_Schema.ApiKeyAdmission

/** Returns whether an organization completed its bounded legacy-key import. */
export async function isBootstrapped(db: Db.Db, orgId: string): Promise<boolean> {
  const record = await db.kysely
    .selectFrom('api_key_admission_bootstraps')
    .select('orgId')
    .where('orgId', '=', orgId)
    .executeTakeFirst()
  return record !== undefined
}

/** Returns whether one key has an authoritative admission row. */
export async function exists(db: Db.Db, id: string): Promise<boolean> {
  return (await get(db, id)) !== undefined
}

/** Reads one authoritative admission row. */
export async function get(db: Db.Db, id: string): Promise<Table | undefined> {
  return db.kysely
    .selectFrom('api_key_admissions')
    .selectAll()
    .where('id', '=', id)
    .executeTakeFirst()
}

/** Atomically admits a new key or moves an existing key into an organization. */
export async function admit(db: Db.Db, options: admit.Options): Promise<boolean> {
  await lockOrganization(db, options.orgId)
  await lock(db, options.id)
  const now = new Date().toISOString()
  await ApiKeyRevocations.prune(db)
  if (options.legacy.length > 0)
    await db.kysely
      .insertInto('api_key_admissions')
      .values(
        options.legacy.map((record) => ({
          expiresAt: record.expiresAt,
          id: record.id,
          orgId: options.orgId,
          projectId: record.projectId,
        })),
      )
      .onConflict((conflict) => conflict.column('id').doNothing())
      .execute()
  if (options.legacyComplete)
    await db.kysely
      .insertInto('api_key_admission_bootstraps')
      .values({ bootstrappedAt: now, orgId: options.orgId })
      .onConflict((conflict) => conflict.column('orgId').doUpdateSet({ bootstrappedAt: now }))
      .execute()
  await db.kysely
    .deleteFrom('api_key_admissions')
    .where('orgId', '=', options.orgId)
    .where('expiresAt', '<=', now)
    .execute()
  const existing = await db.kysely
    .selectFrom('api_key_admissions')
    .select(['id', 'orgId'])
    .where('id', '=', options.id)
    .executeTakeFirst()
  if (existing?.orgId === options.orgId) {
    await db.kysely
      .updateTable('api_key_admissions')
      .set({ expiresAt: options.expiresAt, projectId: options.projectId })
      .where('id', '=', options.id)
      .execute()
    return true
  }

  // The ledger wins over KV attribution left by a partially failed external write.
  const currentOrgId = existing?.orgId ?? options.currentOrgId
  if (currentOrgId !== options.orgId) {
    const count = await db.kysely
      .selectFrom('api_key_admissions')
      .select((eb) => eb.fn.countAll<number>().as('count'))
      .where('orgId', '=', options.orgId)
      .where(
        'id',
        'not in',
        db.kysely
          .selectFrom('api_key_revocations')
          .select('id')
          .where('expiresAt', '>', now)
          .where('expiresAt', '!=', ApiKeyRevocations.pendingExpiresAt),
      )
      .executeTakeFirstOrThrow()
    if (Number(count.count) >= options.limit) return false
  }

  if (existing)
    await db.kysely.deleteFrom('api_key_admissions').where('id', '=', options.id).execute()
  await db.kysely
    .insertInto('api_key_admissions')
    .values({
      expiresAt: options.expiresAt,
      id: options.id,
      orgId: options.orgId,
      projectId: options.projectId,
    })
    .execute()
  return true
}

/** Serializes admission and legacy reconciliation for one organization. */
export async function lockOrganization(db: Db.Db, orgId: string): Promise<void> {
  await sql`SELECT pg_advisory_xact_lock(hashtextextended(${`api-key-admission-org:${orgId}`}, 0))`.execute(
    db.kysely,
  )
}

export declare namespace admit {
  /** Key admission inputs. */
  type Options = {
    /** Existing organization for attribution moves, if any. */
    currentOrgId?: string | undefined
    /** Key expiry, or null for a non-expiring key. */
    expiresAt: string | null
    /** Stable API-key id. */
    id: string
    /** Maximum live keys across the organization. */
    limit: number
    /** Bounded live legacy records used only for the first organization bootstrap. */
    legacy: readonly Legacy[]
    /** Whether the legacy scan reached index exhaustion and can complete bootstrap. */
    legacyComplete: boolean
    /** Destination organization id. */
    orgId: string
    /** Destination project id, or null for an organization-level key. */
    projectId: string | null
  }

  /** Existing live key imported during an organization's first admission. */
  type Legacy = {
    /** Key expiry, or null for a non-expiring key. */
    expiresAt: string | null
    /** Stable API-key id. */
    id: string
    /** Project attribution, or null for an organization-level key. */
    projectId: string | null
  }
}

/** Returns whether a ledger-managed key has the supplied attribution. */
export async function matches(db: Db.Db, owner: matches.Owner): Promise<boolean> {
  if (await ApiKeyRevocations.exists(db, owner.id)) return false
  const record = await db.kysely
    .selectFrom('api_key_admissions')
    .select(['orgId', 'projectId'])
    .where('id', '=', owner.id)
    .executeTakeFirst()
  if (!record) return !(await isBootstrapped(db, owner.orgId))
  return record.orgId === owner.orgId && record.projectId === (owner.projectId ?? null)
}

export declare namespace matches {
  /** API-key attribution compared with the authoritative ledger. */
  type Owner = {
    /** Stable API-key id. */
    id: string
    /** Owning organization id. */
    orgId: string
    /** Project attribution, when present. */
    projectId?: string | undefined
  }
}

/** Idempotently removes one key's authoritative admission slot. */
export async function release(db: Db.Db, id: string): Promise<boolean> {
  await lock(db, id)
  const result = await db.kysely
    .deleteFrom('api_key_admissions')
    .where('id', '=', id)
    .executeTakeFirst()
  return result.numDeletedRows > 0n
}

/** Removes every slot attributed to a deleted organization or project. */
export async function releaseOwner(db: Db.Db, owner: releaseOwner.Owner): Promise<void> {
  let query = db.kysely
    .selectFrom('api_key_admissions')
    .select('id')
    .where('orgId', '=', owner.orgId)
  if (owner.projectId !== undefined) query = query.where('projectId', '=', owner.projectId)
  const ids = (await query.orderBy('id').execute()).map(({ id }) => id)
  await ApiKeyRevocations.deletePending(db, ids)
  if (ids.length > 0)
    await db.kysely.deleteFrom('api_key_admissions').where('id', 'in', ids).execute()
  if (owner.projectId === undefined)
    await db.kysely
      .deleteFrom('api_key_admission_bootstraps')
      .where('orgId', '=', owner.orgId)
      .execute()
}

export declare namespace releaseOwner {
  /** Owner whose admitted keys are no longer usable. */
  type Owner = {
    /** Owning organization id. */
    orgId: string
    /** Deleted project id, or omitted for the whole organization. */
    projectId?: string | undefined
  }
}

/** Serializes admission changes for one stable key id in the caller's transaction. */
export async function lock(db: Db.Db, id: string): Promise<void> {
  await sql`SELECT pg_advisory_xact_lock(hashtextextended(${`api-key-admission-id:${id}`}, 0))`.execute(
    db.kysely,
  )
}