import { sql, type SQLWrapper } from 'drizzle-orm'
import type { Schema } from '@platform-modules/db'
import {
  isMeteringError,
  MeteringStorageError,
  MeteringValidationError,
  QuotaExhaustedError,
  ReservationConflictError,
  ReservationExpiredError,
  ReservationNotFoundError,
} from './errors.js'
import { isWithinMagnitudeBound, MAX_BIGINT_DIGITS } from './internal/bigint.js'
import { applyDelta, lockCounter, type CounterKey } from './internal/counter.js'
import { findByIdempotencyKey, type Reservation as StoredReservation } from './internal/idempotency.js'
import type { Clock, CommitResult, MeteringDatabase, RatingStore } from './types.js'

export type CommitDeps<S extends Schema = Record<string, never>> = {
  db: MeteringDatabase<S>
  clock?: Clock
  rating?: RatingStore
}

type SqlRow = Record<string, unknown>
type SqlExecutor = { execute(query: SQLWrapper): Promise<unknown> }

function firstRow(result: unknown): SqlRow | undefined {
  const rows = (Array.isArray(result) ? result : (result as { rows?: SqlRow[] }).rows) ?? []
  return rows[0]
}

function assertReservationId(reservationId: string): void {
  if (typeof reservationId !== 'string' || reservationId.trim() === '') {
    throw new MeteringValidationError('reservationId', 'must be non-empty')
  }
}

function keyFromRow(row: SqlRow): CounterKey {
  return {
    tenantId: String(row.tenant_id),
    account: String(row.account),
    meter: String(row.meter),
    periodId: String(row.period_id),
  }
}

async function findReservationLocator(
  tx: SqlExecutor,
  reservationId: string,
): Promise<{ key: CounterKey; idempotencyKey: string } | undefined> {
  const result = await tx.execute(sql`
    SELECT tenant_id, account, meter, period_id, idempotency_key
    FROM metering_reservation
    WHERE id = ${reservationId}
  `)
  const row = firstRow(result)
  if (!row) return undefined
  return {
    key: keyFromRow(row),
    idempotencyKey: String(row.idempotency_key),
  }
}

function publicReservation(reservation: StoredReservation) {
  const { fingerprint: _fingerprint, ...result } = reservation
  return result
}

function effectiveCeiling(reservation: StoredReservation): bigint | null {
  if (reservation.appliedPolicy.overage === 'deny') {
    return reservation.appliedPolicy.includedUnits
  }
  return reservation.appliedPolicy.hardLimit ?? null
}

function maxZero(value: bigint): bigint {
  return value > 0n ? value : 0n
}

function readCommitTime(clock: Clock | undefined): Date {
  const value = clock?.() ?? new Date()
  if (!(value instanceof Date) || Number.isNaN(value.getTime())) {
    throw new MeteringValidationError('clock', 'must return a valid Date')
  }
  return value
}

function throwStorage(error: unknown, operation: string): never {
  if (isMeteringError(error)) throw error
  throw new MeteringStorageError(operation, error)
}

async function commitInTransaction<S extends Schema>(
  deps: CommitDeps<S>,
  tx: SqlExecutor,
  reservationId: string,
  actualUnits: bigint,
): Promise<CommitResult> {
  const locator = await findReservationLocator(tx, reservationId)
  if (!locator) throw new ReservationNotFoundError(reservationId)

  const position = await lockCounter(tx, locator.key)
  const reservation = await findByIdempotencyKey(
    tx,
    locator.key.tenantId,
    locator.idempotencyKey,
  )
  if (!reservation || reservation.id !== reservationId) {
    throw new MeteringStorageError('reservation row changed during commit', new Error(reservationId))
  }

  if (reservation.status === 'committed') {
    const committedUnits = reservation.committedUnits
    if (committedUnits === undefined) {
      throw new MeteringStorageError('committed reservation has no committed units', new Error(reservationId))
    }
    if (committedUnits !== actualUnits) {
      throw new ReservationConflictError({
        reservationId,
        detail: 'different actualUnits for committed reservation',
      })
    }
    return {
      reservation: publicReservation(reservation),
      committedUnits,
      releasedUnits: maxZero(reservation.reservedUnits - committedUnits),
    }
  }

  if (reservation.status === 'expired') {
    throw new ReservationExpiredError(reservationId)
  }
  if (reservation.status !== 'reserved') {
    throw new ReservationConflictError({
      reservationId,
      detail: `cannot commit ${reservation.status} reservation`,
    })
  }

  const releasedUnits = maxZero(reservation.reservedUnits - actualUnits)
  if (actualUnits > reservation.reservedUnits) {
    const requested = actualUnits - reservation.reservedUnits
    const ceiling = effectiveCeiling(reservation)
    const used = position.committed + position.reserved
    const capacityRemaining = ceiling === null ? null : maxZero(ceiling - used)
    if (capacityRemaining !== null && requested > capacityRemaining) {
      throw new QuotaExhaustedError({
        requested,
        includedRemaining: maxZero(reservation.appliedPolicy.includedUnits - used),
        capacityRemaining,
      })
    }
  }
  if (!isWithinMagnitudeBound(position.committed + actualUnits)) {
    throw new MeteringValidationError(
      'committed',
      `result must be below 10^${MAX_BIGINT_DIGITS}`,
    )
  }

  const committedAt = readCommitTime(deps.clock)
  let ruleVersion: string | null = null
  if (deps.rating) {
    const rule = await deps.rating.rule(reservation.meter, committedAt)
    if (rule !== null && (typeof rule.ruleVersion !== 'string' || rule.ruleVersion.trim() === '')) {
      throw new MeteringValidationError('rating.ruleVersion', 'must be non-empty')
    }
    ruleVersion = rule?.ruleVersion ?? null
  }

  await applyDelta(tx, locator.key, {
    committed: actualUnits,
    reserved: -reservation.reservedUnits,
  })
  const updateResult = await tx.execute(sql`
    UPDATE metering_reservation
    SET status = 'committed', committed_units = ${actualUnits}
    WHERE id = ${reservationId} AND status = 'reserved'
    RETURNING id
  `)
  if (!firstRow(updateResult)?.id) {
    throw new MeteringStorageError('reservation disappeared during commit', new Error(reservationId))
  }

  await tx.execute(sql`
    INSERT INTO metering_usage_event (
      id, tenant_id, account, meter, period_id, units, kind,
      reservation_id, rule_version, created_at
    ) VALUES (
      ${`${reservationId}:commit`}, ${locator.key.tenantId}, ${locator.key.account},
      ${locator.key.meter}, ${locator.key.periodId}, ${actualUnits}, 'commit',
      ${reservationId}, ${ruleVersion}, ${committedAt.toISOString()}::timestamptz
    )
  `)

  return {
    reservation: publicReservation({
      ...reservation,
      status: 'committed',
      committedUnits: actualUnits,
    }),
    committedUnits: actualUnits,
    releasedUnits,
  }
}

export async function commit<S extends Schema = Record<string, never>>(
  deps: CommitDeps<S>,
  reservationId: string,
  actualUnits: bigint,
): Promise<CommitResult> {
  assertReservationId(reservationId)
  if (typeof actualUnits !== 'bigint' || actualUnits < 0n) {
    throw new MeteringValidationError('actualUnits', 'must be a non-negative bigint')
  }
  if (!isWithinMagnitudeBound(actualUnits)) {
    throw new MeteringValidationError('actualUnits', `must be below 10^${MAX_BIGINT_DIGITS}`)
  }

  try {
    return await deps.db.transaction((tx) =>
      commitInTransaction(deps, tx, reservationId, actualUnits),
    )
  } catch (error) {
    return throwStorage(error, 'commit')
  }
}