import { type ColumnType, type Selectable, sql } from 'kysely'

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

/** A `bigint` count column: read as a string from pg, written as a number. */
type Count = ColumnType<string, number, number>

/** Environment a request-usage row attributes to. */
export type Environment = db_Schema.RequestUsageBucket['environment']

/** Exactly-once state of a meter event. */
export type State = db_Schema.RequestUsageMeterEvent['state']

/** Columns of the `request_usage_buckets` table (pg representation). */
export type BucketTable = {
  bucketStart: string
  environment: Environment
  orgId: string
  reportedCount: Count
  updatedAt: string
}

/** Columns of the `request_usage_meter_events` table (pg representation). */
export type MeterEventTable = {
  bucketStart: string
  deltaCount: Count
  environment: Environment
  error: string | null
  firstAttemptedAt: string
  identifier: string
  orgId: string
  reason: string | null
  reportedAt: string | null
  sequence: Count
  state: State
  stripeCustomerId: string
}

/** A stored per-hour request-count watermark. */
export type Bucket = db_Schema.RequestUsageBucket

/** A stored request-count meter event. */
export type MeterEvent = db_Schema.RequestUsageMeterEvent

/** Aggregate state for pending meter events in one billing environment. */
export type PendingStatus = {
  count: number
  oldestAttemptedAt: string | undefined
}

/** Identifies one hourly bucket for an org in an environment. */
export type BucketKey = {
  /** UTC hour start (ISO 8601). */
  bucketStart: string
  /** Environment the requests were served under. */
  environment: Environment
  /** Organization id (`org_…`). */
  orgId: string
}

/**
 * Reads a bucket watermark.
 *
 * @param db - The database.
 * @param key - The bucket key.
 * @returns The watermark, or `undefined` when none has been reported yet.
 */
export async function getBucket(db: Db.Db, key: BucketKey): Promise<Bucket | undefined> {
  const row = await db.kysely
    .selectFrom('request_usage_buckets')
    .selectAll()
    .where('orgId', '=', key.orgId)
    .where('environment', '=', key.environment)
    .where('bucketStart', '=', key.bucketStart)
    .executeTakeFirst()
  return row ? toBucket(row) : undefined
}

/**
 * Lists the meter events still awaiting a Stripe acknowledgement in an
 * environment, oldest first — the reporter's crash-recovery drive.
 *
 * @param db - The database.
 * @param environment - The environment to scan.
 * @returns The pending meter events.
 */
export async function listPending(db: Db.Db, environment: Environment): Promise<MeterEvent[]> {
  const rows = await db.kysely
    .selectFrom('request_usage_meter_events')
    .selectAll()
    .where('environment', '=', environment)
    .where('state', '=', 'pending')
    .orderBy('firstAttemptedAt', 'asc')
    .orderBy('identifier', 'asc')
    .execute()
  return rows.map(toMeterEvent)
}

/**
 * Reads the number and age anchor of meter events still awaiting Stripe.
 *
 * @param db - The database.
 * @param environment - The billing environment to inspect.
 * @returns The pending-event count and oldest first attempt.
 */
export async function getPendingStatus(db: Db.Db, environment: Environment): Promise<PendingStatus> {
  const row = await db.kysely
    .selectFrom('request_usage_meter_events')
    .select((eb) => [eb.fn.countAll().as('count'), eb.fn.min('firstAttemptedAt').as('oldest')])
    .where('environment', '=', environment)
    .where('state', '=', 'pending')
    .executeTakeFirstOrThrow()
  return { count: Number(row.count), oldestAttemptedAt: row.oldest ?? undefined }
}

/**
 * Inserts a frozen `pending` meter event, claiming a delta. Idempotent on the
 * deterministic `identifier`: a re-seal after a crash resolves to the existing
 * row rather than duplicating the delta.
 *
 * @param db - The database.
 * @param input - The event to insert.
 * @returns Whether this call inserted the row.
 */
export async function insertPending(db: Db.Db, input: insertPending.Input): Promise<boolean> {
  const now = new Date().toISOString()
  const inserted = await db.kysely
    .insertInto('request_usage_meter_events')
    .values({
      bucketStart: input.bucketStart,
      deltaCount: input.deltaCount,
      environment: input.environment,
      error: null,
      firstAttemptedAt: now,
      identifier: input.identifier,
      orgId: input.orgId,
      reason: null,
      reportedAt: null,
      sequence: input.sequence,
      state: 'pending',
      stripeCustomerId: input.stripeCustomerId,
    })
    .onConflict((oc) => oc.doNothing())
    .returning('identifier')
    .executeTakeFirst()
  return Boolean(inserted)
}

export declare namespace insertPending {
  /** Fields accepted when sealing a pending meter event. */
  type Input = {
    /** UTC hour start the delta bills for (ISO 8601). */
    bucketStart: string
    /** Frozen request count this event bills (positive). */
    deltaCount: number
    /** Environment the requests were served under. */
    environment: Environment
    /** Deterministic id; also the Stripe identifier and idempotency key. */
    identifier: string
    /** Organization id (`org_…`). */
    orgId: string
    /** Reported-count watermark observed when this delta was claimed. */
    sequence: number
    /** Stripe customer id the event bills against. */
    stripeCustomerId: string
  }
}

/**
 * Marks a pending event reported and advances its bucket watermark by the
 * event's frozen delta, atomically. A no-op once reported, so a retry after a
 * crash between the Stripe ack and this write never double-advances.
 *
 * @param db - The database.
 * @param identifier - The event identifier.
 * @param at - When Stripe acknowledged the event (ISO 8601).
 * @returns Whether this call settled the event (false when already reported).
 */
export function settle(db: Db.Db, identifier: string, at: string): Promise<boolean> {
  return db.transaction(async (tx) => {
    const updated = await tx.kysely
      .updateTable('request_usage_meter_events')
      .set({ error: null, reportedAt: at, state: 'reported' })
      .where('identifier', '=', identifier)
      .where('state', '=', 'pending')
      .returningAll()
      .executeTakeFirst()
    if (!updated) return false
    await tx.kysely
      .insertInto('request_usage_buckets')
      .values({
        bucketStart: updated.bucketStart,
        environment: updated.environment,
        orgId: updated.orgId,
        reportedCount: Number(updated.deltaCount),
        updatedAt: at,
      })
      .onConflict((oc) =>
        oc.columns(['orgId', 'environment', 'bucketStart']).doUpdateSet({
          reportedCount: sql`request_usage_buckets.reported_count + ${updated.deltaCount}`,
          updatedAt: at,
        }),
      )
      .execute()
    return true
  })
}

/**
 * Records a transient failure on a pending event; it stays pending for retry.
 *
 * @param db - The database.
 * @param identifier - The event identifier.
 * @param error - The failure detail.
 */
export async function markError(db: Db.Db, identifier: string, error: string): Promise<void> {
  await db.kysely
    .updateTable('request_usage_meter_events')
    .set({ error: error.slice(0, 500) })
    .where('identifier', '=', identifier)
    .where('state', '=', 'pending')
    .execute()
}

/** Maps a bucket row to its domain record. */
function toBucket(row: Selectable<BucketTable>): Bucket {
  return { ...row, reportedCount: Number(row.reportedCount) }
}

/** Maps a meter-event row to its domain record. */
function toMeterEvent(row: Selectable<MeterEventTable>): MeterEvent {
  return { ...row, deltaCount: Number(row.deltaCount), sequence: Number(row.sequence) }
}