import { sql, type Selectable } from 'kysely'

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

/** Columns of the `routes_transfer_subsidy_nonces` table. */
export type Table = db_Schema.RoutesTransferSubsidyNonce
/** One persisted destination signer nonce cursor. */
export type Record = Selectable<Table>

/** Serializes inventory and nonce operations for one signer and chain. */
async function lock(db: Db.Db, options: Lane): Promise<void> {
  const key = `${options.chainId}:${options.account.toLowerCase()}`
  await sql`SELECT pg_advisory_xact_lock(hashtextextended(${key}, 0))`.execute(db.kysely)
}

type Lane = {
  /** Destination subsidy signer address. */
  account: string
  /** Destination EVM chain CAIP-2 id. */
  chainId: string
}

/** Locks a signer lane and rejects new work when its nonce cannot be recovered. */
export async function lockAvailableIn(db: Db.Db, options: Lane): Promise<void> {
  await lock(db, options)
  const current = await db.kysely
    .selectFrom('routes_transfer_subsidy_nonces')
    .selectAll()
    .where('account', '=', options.account.toLowerCase())
    .where('chainId', '=', options.chainId)
    .forUpdate()
    .executeTakeFirst()
  if (current?.blockedAt) throw new BlockedError()
}

/** Blocks a signer lane after a reserved nonce is permanently rejected. */
export async function block(db: Db.Db, options: block.Options): Promise<void> {
  return db.transaction((tx) => blockIn(tx, options))
}

/** Blocks a signer lane inside a caller-owned transaction. */
export async function blockIn(db: Db.Db, options: block.Options): Promise<void> {
  await lock(db, options)
  const now = new Date().toISOString()
  const updated = await db.kysely
    .updateTable('routes_transfer_subsidy_nonces')
    .set({
      blockedAt: now,
      blockedReason: options.reason,
      blockedTransferId: options.transferId,
      updatedAt: now,
    })
    .where('account', '=', options.account.toLowerCase())
    .where('chainId', '=', options.chainId)
    .executeTakeFirst()
  if (updated.numUpdatedRows !== 1n) throw new MissingCursorError()
}

export declare namespace block {
  /** Signer lane and permanent rejection evidence. */
  type Options = Lane & {
    /** Bounded reason for blocking the signer lane. */
    reason: 'broadcast_rejected'
    /** Transfer holding the rejected nonce. */
    transferId: string
  }
}

/** Reserves one nonce under a transaction-scoped signer and chain lock. */
export async function reserve(db: Db.Db, options: reserve.Options): Promise<number> {
  const account = options.account.toLowerCase()
  await lock(db, options)
  const current = await db.kysely
    .selectFrom('routes_transfer_subsidy_nonces')
    .selectAll()
    .where('account', '=', account)
    .where('chainId', '=', options.chainId)
    .forUpdate()
    .executeTakeFirst()
  if (current?.blockedAt) throw new BlockedError()
  const nonce = Math.max(current ? Number(current.nextNonce) : 0, options.pendingNonce)
  await db.kysely
    .insertInto('routes_transfer_subsidy_nonces')
    .values({
      account,
      blockedAt: null,
      blockedReason: null,
      blockedTransferId: null,
      chainId: options.chainId,
      nextNonce: String(nonce + 1),
      updatedAt: new Date().toISOString(),
    })
    .onConflict((conflict) =>
      conflict.columns(['chainId', 'account']).doUpdateSet({
        nextNonce: String(nonce + 1),
        updatedAt: new Date().toISOString(),
      }),
    )
    .execute()
  return nonce
}

/** The signer lane is blocked by an unrecoverable reserved nonce. */
export class BlockedError extends Error {
  override name = 'Routes.TransferSubsidyNonce.BlockedError'
}

/** A rejected subsidy transaction had no persisted nonce cursor. */
export class MissingCursorError extends Error {
  override name = 'Routes.TransferSubsidyNonce.MissingCursorError'
}

export declare namespace reserve {
  /** Signer and chain cursor used to reserve one nonce. */
  type Options = {
    /** Destination subsidy signer address. */
    account: string
    /** Destination EVM chain CAIP-2 id. */
    chainId: string
    /** Next nonce observed from the pending RPC state. */
    pendingNonce: number
  }
}