import { sql, type JSONColumnType, type Selectable } from 'kysely'
import * as pg from 'pg'

import type * as Db from '../Db.js'
import type * as db_Schema from '../Schema.js'
import * as RoutesIdempotency from './routesIdempotency.js'
import * as RoutesSubsidies from './routesSubsidies.js'

/** Columns of the `routes_deposit_addresses` table. */
export type Table = Omit<db_Schema.RoutesDepositAddress, 'providerRequestIds'> & {
  /** Private provider request identifiers stored as JSON. */
  providerRequestIds: JSONColumnType<readonly string[], string, string>
}

/** A stored route deposit address. */
export type Record = Selectable<Table>

/** Ownership filter derived from the authenticated API key. */
export type Owner = {
  /** Key environment. */
  environment: Record['environment']
  /** Owning organization id (`org_…`). */
  orgId: string
  /** Attributed project id for project-scoped API keys. */
  projectId?: string | undefined
}

/** Identity fields that make one reusable address match a caller request. */
export type Match = Owner & {
  /** Customer destination token key. */
  destinationTokenKey: string
  /** Tempo account that receives completed deposits. */
  recipient: string
  /** Source-chain account that receives refunds. */
  refundAddress: string
  /** Source chain CAIP-2 id. */
  sourceChainId: string
  /** Source token key. */
  sourceTokenKey: string
  /** Whether Tempo guarantees normalized 1:1 delivery. */
  subsidize: boolean
}

/** Counts durable route deposit addresses owned by an organization creator. */
async function countByOwner(db: Db.Db, owner: countByOwner.Owner): Promise<number> {
  let query = db.kysely
    .selectFrom('routes_deposit_addresses')
    .select(({ fn }) => fn.countAll<string>().as('count'))
  const userId = owner.userId
  query = userId
    ? query.where('creatorUserId', '=', userId)
    : query.where('orgId', '=', owner.orgId)
  const result = await query.executeTakeFirstOrThrow()
  return Number(result.count)
}

declare namespace countByOwner {
  /** Canonical creator scope, with organization fallback for unowned rows. */
  type Owner = {
    /** Organization id (`org_…`). */
    orgId: string
    /** Canonical creator id (`usr_…`). */
    userId: string | null
  }
}

/** Counts durable route deposit addresses for one organization. */
export async function countByOrganization(db: Db.Db, orgId: string): Promise<number> {
  const result = await db.kysely
    .selectFrom('routes_deposit_addresses')
    .select(({ fn }) => fn.countAll<string>().as('count'))
    .where('orgId', '=', orgId)
    .executeTakeFirstOrThrow()
  return Number(result.count)
}

/** Lists active addresses that organization deletion will deactivate. */
export function listActiveByOrganization(db: Db.Db, orgId: string): Promise<readonly Record[]> {
  return db.kysely
    .selectFrom('routes_deposit_addresses')
    .selectAll()
    .where('orgId', '=', orgId)
    .where('status', '!=', 'deactivated')
    .execute()
}

/** Claims address creation under the organization deletion fence. */
export function claimCreation(
  db: Db.Db,
  options: claimCreation.Options,
): Promise<RoutesIdempotency.claim.Result> {
  return db.transaction(async (tx) => {
    await RoutesSubsidies.lockOrganization(tx, options.orgId)
    const organization = await getOrCreateCapacityOwner(tx, {
      apiKeyId: options.claim.apiKeyId,
      orgId: options.orgId,
    })
    if (!organization) throw new OwnerNotFoundError()
    const claim = await RoutesIdempotency.claim(tx, {
      ...options.claim,
      operation: RoutesIdempotency.operations.depositAddress,
      orgId: options.orgId,
    })
    return claim
  })
}

export declare namespace claimCreation {
  /** Claim identity and organization deletion fence. */
  type Options = {
    /** Idempotency identity and lease. */
    claim: Omit<RoutesIdempotency.claim.Input, 'operation' | 'orgId'>
    /** Organization reserving address capacity. */
    orgId: string
  }
}

/** Validates reserved address capacity, excluding an already reusable match. */
export function validateCreationCapacity(
  db: Db.Db,
  options: validateCreationCapacity.Options,
): Promise<validateCreationCapacity.Result> {
  return db
    .transaction(async (tx) => {
      await RoutesSubsidies.lockOrganization(tx, options.orgId)
      const organization = await getCapacityOwner(tx, options.orgId)
      if (!organization) throw new OwnerNotFoundError()
      await lockCapacityOwner(tx, organization)
      const matching = await getMatching(tx, options.match)
      if (matching) {
        const response = options.serializeMatchingResponse(matching)
        const completed = await RoutesIdempotency.complete(tx, {
          ...options.claim,
          replayTtlMs: options.replayTtlMs,
          response,
        })
        if (!completed) throw new CapacityClaimNotFoundError()
        if (options.recoveredClaim)
          await RoutesIdempotency.complete(tx, {
            ...options.recoveredClaim,
            replayTtlMs: options.replayTtlMs,
            response,
          })
        return { allowed: true, result: { record: matching, type: 'existing' } } as const
      }
      // Old workers cannot prove a checkpoint's owner or match; exact same-key retries attribute it before this fence.
      if (await RoutesIdempotency.hasUnattributedInFlightClaim(tx)) throw new MatchInProgressError()
      const reservation = await RoutesIdempotency.reserveDepositAddressMatch(tx, {
        ...options.claim,
        matchHash: options.matchHash,
        orgId: options.orgId,
        ttlMs: options.pendingTtlMs,
      })
      if (reservation.type === 'lost') throw new CapacityClaimNotFoundError()
      if (reservation.type === 'pending') throw new MatchInProgressError()
      if (reservation.type === 'resume') return { allowed: true, result: reservation } as const
      if (reservation.type === 'provisioned' || options.recoveredClaim !== undefined)
        return { allowed: true, result: { type: 'reserved' } } as const
      const addresses = await countByOwner(tx, organization)
      const inFlight = await RoutesIdempotency.countInFlightDepositAddresses(tx, organization)
      if (addresses + inFlight <= options.maxPerCreator)
        return { allowed: true, result: { type: 'reserved' } } as const
      await RoutesIdempotency.release(tx, options.claim)
      return { allowed: false, result: undefined } as const
    })
    .then((result) => {
      if (!result.allowed) throw new LimitExceededError(options.maxPerCreator)
      return result.result
    })
}

export declare namespace validateCreationCapacity {
  /** Reusable identity and organization capacity bound. */
  type Options = {
    /** Pending claim released atomically when the creator-wide cap is full. */
    claim: RoutesIdempotency.release.Input
    /** Maximum durable and in-flight addresses for one canonical creator. */
    maxPerCreator: number
    /** Stable reusable-address identity that does not consume new capacity when present. */
    match: Match
    /** SHA-256 fingerprint of the canonical reusable match. */
    matchHash: string
    /** Organization reserving address capacity. */
    orgId: string
    /** Renewed pending lease duration, in milliseconds. */
    pendingTtlMs: number
    /** Successful-response replay retention from completion, in milliseconds. */
    replayTtlMs: number
    /** Older provider checkpoint completed when its durable match already exists. */
    recoveredClaim?: RoutesIdempotency.release.Input | undefined
    /** Serializes a reusable address into the idempotent response. */
    serializeMatchingResponse: (record: Record) => string
  }

  /** Capacity reservation or reusable recovery result. */
  type Result =
    | { record: Record; type: 'existing' }
    | { type: 'reserved' }
    | {
        checkpointFresh: boolean
        claim: RoutesIdempotency.release.Input
        response: string
        type: 'resume'
      }
}

/** Deactivates an organization's addresses so deletion leaves no recurring work. */
export function deactivateByOrganization(
  db: Db.Db,
  options: deactivateByOrganization.Options,
): Promise<readonly Record[]> {
  return db.kysely
    .updateTable('routes_deposit_addresses')
    .set({
      pollLeaseUntil: null,
      pollLeaseVersion: sql<number>`poll_lease_version + 1`,
      status: 'deactivated',
      statusUpdatedAt: options.now,
      updatedAt: options.now,
      version: sql<number>`version + 1`,
    })
    .where('orgId', '=', options.orgId)
    .where('status', '!=', 'deactivated')
    .returningAll()
    .execute()
}

export declare namespace deactivateByOrganization {
  /** Organization and timestamp for one bulk deactivation. */
  type Options = {
    /** Timestamp committed to each changed address. */
    now: string
    /** Owning organization id (`org_…`). */
    orgId: string
  }
}

/** Returns whether an organization owns an irrevocable provider subsidy request. */
export async function hasSubsidizedRelayAddress(db: Db.Db, orgId: string): Promise<boolean> {
  return Boolean(
    await db.kysely
      .selectFrom('routes_deposit_addresses')
      .select('id')
      .where('orgId', '=', orgId)
      .where('providerId', '=', 'relay')
      .where('subsidize', '=', true)
      .limit(1)
      .executeTakeFirst(),
  )
}

/** Aggregates current deposit-address states for bounded operational metrics. */
export function summarizeStates(db: Db.Db) {
  return db.kysely
    .selectFrom('routes_deposit_addresses')
    .select(({ fn }) => [
      'environment',
      'providerId',
      'status',
      fn.countAll<string>().as('count'),
      sql<string>`min(coalesce(status_updated_at, updated_at))`.as('oldestStatusUpdatedAt'),
    ])
    .where('status', '!=', 'deactivated')
    .groupBy(['environment', 'providerId', 'status'])
    .execute()
}

/** Aggregates active deposit-address checks that are due without a live lease. */
export function summarizeOverdue(db: Db.Db, options: summarizeOverdue.Options) {
  return db.kysely
    .selectFrom('routes_deposit_addresses')
    .select(({ fn }) => [
      'environment',
      'providerId',
      fn.countAll<string>().as('count'),
      fn.min<string>('nextPollAt').as('oldestNextPollAt'),
    ])
    .where('nextPollAt', '<', options.now)
    .where('status', '=', 'active')
    .where((eb) =>
      eb.or([eb('pollLeaseUntil', 'is', null), eb('pollLeaseUntil', '<=', options.now)]),
    )
    .groupBy(['environment', 'providerId'])
    .execute()
}

export declare namespace summarizeOverdue {
  /** Overdue-address snapshot options. */
  type Options = {
    /** Current timestamp used to classify due and leased addresses. */
    now: string
  }
}

/** Inserts a provisioned route deposit address. */
export function insert(db: Db.Db, record: Record): Promise<Record> {
  return db.kysely
    .insertInto('routes_deposit_addresses')
    .values({
      ...record,
      providerRequestIds: JSON.stringify(record.providerRequestIds),
    })
    .returningAll()
    .executeTakeFirstOrThrow()
}

/** Inserts an address or returns the concurrently-created matching resource. */
export async function insertOrGet(db: Db.Db, options: insertOrGet.Options): Promise<Record> {
  const record = await db.kysely
    .insertInto('routes_deposit_addresses')
    .values({
      ...options.record,
      providerRequestIds: JSON.stringify(options.record.providerRequestIds),
    })
    .onConflict((oc) =>
      oc
        .columns([
          'orgId',
          'environment',
          'projectId',
          'sourceChainId',
          'sourceTokenKey',
          'destinationTokenKey',
          'recipient',
          'refundAddress',
          'subsidize',
        ])
        .doNothing(),
    )
    .returningAll()
    .executeTakeFirst()
    .catch((cause) => {
      if (
        cause instanceof pg.DatabaseError &&
        cause.code === '23505' &&
        cause.constraint === 'routes_deposit_addresses_provider_address_key'
      )
        throw new ConflictError()
      throw cause
    })
  if (record) return record
  const existing = await getMatching(db, options.match)
  if (existing) return existing
  throw new ConflictError()
}

/**
 * Inserts or returns a matching address while enforcing its creator's cap.
 * The caller must provide a transaction so the advisory lock covers the insert.
 */
export async function insertOrGetWithinLimit(
  db: Db.Db,
  options: insertOrGetWithinLimit.Options,
): Promise<Record> {
  await RoutesSubsidies.lockOrganization(db, options.record.orgId)
  const organization = await getCapacityOwner(db, options.record.orgId)
  if (!organization) throw new OwnerNotFoundError()
  await lockCapacityOwner(db, organization)
  const existing = await getMatching(db, options.match)
  if (existing) return existing
  if ((await countByOwner(db, organization)) >= options.maxPerCreator)
    throw new LimitExceededError(options.maxPerCreator)
  return insertOrGet(db, options)
}

export declare namespace insertOrGetWithinLimit {
  /** Record, stable identity, and creator-wide cap for an atomic insert. */
  type Options = insertOrGet.Options & {
    /** Maximum durable deposit addresses one canonical creator may own. */
    maxPerCreator: number
  }
}

/** Inserts an already-provisioned address under the creator locks without applying a new-work cap. */
export async function insertProvisionedOrGet(
  db: Db.Db,
  options: insertProvisionedOrGet.Options,
): Promise<Record> {
  await RoutesSubsidies.lockOrganization(db, options.record.orgId)
  const organization = await getCapacityOwner(db, options.record.orgId)
  if (!organization) throw new OwnerNotFoundError()
  await lockCapacityOwner(db, organization)
  return insertOrGet(db, options)
}

export declare namespace insertProvisionedOrGet {
  /** Already-provisioned record and stable reusable identity. */
  type Options = insertOrGet.Options
}

export declare namespace insertOrGet {
  /** Record and stable identity used by create-or-get. */
  type Options = {
    /** Stable customer-visible address identity. */
    match: Match
    /** Provisioned address record to insert. */
    record: Record
  }
}

function getCapacityOwner(db: Db.Db, orgId: string): Promise<countByOwner.Owner | undefined> {
  return db.kysely
    .selectFrom('organizations')
    .select(['id as orgId', 'userId'])
    .where('id', '=', orgId)
    .executeTakeFirst()
}

async function getOrCreateCapacityOwner(
  db: Db.Db,
  options: getOrCreateCapacityOwner.Options,
): Promise<countByOwner.Owner | undefined> {
  const existing = await getCapacityOwner(db, options.orgId)
  if (existing || options.apiKeyId !== options.orgId) return existing
  // Legacy organization-less keys use their key id as orgId; materialize it so lifecycle fences remain enforceable.
  const now = new Date().toISOString()
  await db.kysely
    .insertInto('organizations')
    .values({
      createdAt: now,
      createdBy: null,
      id: options.orgId,
      name: options.orgId,
      sponsorshipSubsidyDurationDays: null,
      sponsorshipSubsidyProjectSpendLimit: null,
      updatedAt: now,
      userId: null,
    })
    .onConflict((oc) => oc.column('id').doNothing())
    .execute()
  return getCapacityOwner(db, options.orgId)
}

declare namespace getOrCreateCapacityOwner {
  /** Legacy API-key fallback identity. */
  type Options = {
    /** API key id (`key_…`). */
    apiKeyId: string
    /** API key organization fallback. */
    orgId: string
  }
}

async function lockCapacityOwner(db: Db.Db, owner: countByOwner.Owner): Promise<void> {
  const key = owner.userId
    ? `routes-deposit-address-capacity:user:${owner.userId}`
    : `routes-deposit-address-capacity:organization:${owner.orgId}`
  await sql`SELECT pg_advisory_xact_lock(hashtextextended(${key}, 0))`.execute(db.kysely)
}

/** Gets a route deposit address without ownership scoping. */
export function get(db: Db.Db, id: string): Promise<Record | undefined> {
  return db.kysely
    .selectFrom('routes_deposit_addresses')
    .selectAll()
    .where('id', '=', id)
    .executeTakeFirst()
}

/** Gets a route deposit address by its provider-owned address. */
export function getByProviderAddress(
  db: Db.Db,
  options: getByProviderAddress.Options,
): Promise<Record | undefined> {
  const address = /^0x[0-9a-fA-F]{40}$/.test(options.address)
    ? options.address.toLowerCase()
    : options.address
  return db.kysely
    .selectFrom('routes_deposit_addresses')
    .selectAll()
    .where('address', '=', address)
    .where('providerId', '=', options.providerId)
    .where('status', '!=', 'deactivated')
    .executeTakeFirst()
}

export declare namespace getByProviderAddress {
  /** Provider address identity received from a reconciliation trigger. */
  type Options = {
    /** Provider-owned reusable deposit address. */
    address: string
    /** Route provider id. */
    providerId: string
  }
}

/** Gets an owner-visible route deposit address. */
export function getOwned(db: Db.Db, owner: Owner, id: string): Promise<Record | undefined> {
  let query = db.kysely
    .selectFrom('routes_deposit_addresses')
    .selectAll()
    .where('environment', '=', owner.environment)
    .where('id', '=', id)
    .where('orgId', '=', owner.orgId)
    .where('status', '!=', 'deactivated')
  if (owner.projectId !== undefined) query = query.where('projectId', '=', owner.projectId)
  return query.executeTakeFirst()
}

/** Lists owner-visible route deposit addresses newest-first. */
export function listByOwner(db: Db.Db, options: listByOwner.Options): Promise<Record[]> {
  const { cursor, limit, owner } = options
  let query = db.kysely
    .selectFrom('routes_deposit_addresses')
    .selectAll()
    .where('environment', '=', owner.environment)
    .where('orgId', '=', owner.orgId)
    .where('status', '!=', 'deactivated')
  if (owner.projectId !== undefined) query = query.where('projectId', '=', owner.projectId)
  if (cursor)
    query = query.where((eb) =>
      eb.or([
        eb('createdAt', '<', cursor.createdAt),
        eb.and([eb('createdAt', '=', cursor.createdAt), eb('id', '<', cursor.id)]),
      ]),
    )
  return query.orderBy('createdAt', 'desc').orderBy('id', 'desc').limit(limit).execute()
}

export declare namespace listByOwner {
  /** Ownership and paging options for route deposit addresses. */
  type Options = {
    /** Exclusive lower bound from the previous page. */
    cursor?: Cursor | undefined
    /** Maximum rows to return. */
    limit: number
    /** Ownership filter derived from the authenticated API key. */
    owner: Owner
  }

  /** Cursor fields for the last address returned by the previous page. */
  type Cursor = {
    /** Deposit address creation time. */
    createdAt: string
    /** Deposit address id, used as a deterministic tie-breaker. */
    id: string
  }
}

/** Lists filtered route deposit addresses, newest update first. */
export function list(db: Db.Db, options: list.Options): Promise<Record[]> {
  let query = db.kysely.selectFrom('routes_deposit_addresses').selectAll()
  if (options.environments !== undefined)
    query = query.where('environment', 'in', options.environments)
  else if (options.environment !== undefined)
    query = query.where('environment', '=', options.environment)
  if (options.orgId !== undefined) query = query.where('orgId', '=', options.orgId)
  if (options.projectId !== undefined) query = query.where('projectId', '=', options.projectId)
  if (options.providerIds !== undefined)
    query = query.where('providerId', 'in', options.providerIds)
  else if (options.providerId !== undefined)
    query = query.where('providerId', '=', options.providerId)
  if (options.sourceChainIds !== undefined)
    query = query.where('sourceChainId', 'in', options.sourceChainIds)
  else if (options.sourceChainId !== undefined)
    query = query.where('sourceChainId', '=', options.sourceChainId)
  if (options.statuses !== undefined) query = query.where('status', 'in', options.statuses)
  else if (options.status !== undefined) query = query.where('status', '=', options.status)
  if (options.query !== undefined) {
    const exact = options.query
    const address = /^0x[0-9a-fA-F]{40}$/.test(options.query)
      ? options.query.toLowerCase()
      : options.query
    const transactionRef = /^(?:0x)?[0-9a-fA-F]{64}$/.test(options.query)
      ? options.query.toLowerCase()
      : options.query
    query = query.where((eb) =>
      eb.or([
        eb('address', '=', address),
        eb('id', '=', exact),
        eb.exists(
          eb
            .selectFrom('routes_deposits')
            .select('routes_deposits.id')
            .whereRef('routes_deposits.depositAddressId', '=', 'routes_deposit_addresses.id')
            .where((fb) =>
              fb.or([
                fb('routes_deposits.settlementTransactionHash', '=', transactionRef),
                fb('routes_deposits.sourceTransactionHash', '=', transactionRef),
                sql<boolean>`routes_deposits.provider_transaction_hashes @> ${JSON.stringify([transactionRef])}::jsonb`,
                sql<boolean>`routes_deposits.snapshot->'destinationTransactionHashes' @> ${JSON.stringify([transactionRef])}::jsonb`,
                sql<boolean>`routes_deposits.snapshot->'refundTransactionHashes' @> ${JSON.stringify([transactionRef])}::jsonb`,
              ]),
            ),
        ),
      ]),
    )
  }
  const cursor = options.cursor
  if (cursor !== undefined)
    query = query.where((eb) =>
      eb.or([
        eb('updatedAt', '<', cursor.updatedAt),
        eb.and([eb('updatedAt', '=', cursor.updatedAt), eb('id', '<', cursor.id)]),
      ]),
    )
  return query.orderBy('updatedAt', 'desc').orderBy('id', 'desc').limit(options.limit).execute()
}

export declare namespace list {
  /** Deposit-address filters and page bound. */
  type Options = {
    /** Exclusive lower bound from the previous page. */
    cursor?: Cursor | undefined
    /** Restricts addresses to one route environment. */
    environment?: Record['environment'] | undefined
    /** Restricts addresses to any listed route environment. */
    environments?: readonly Record['environment'][] | undefined
    /** Maximum rows returned. */
    limit: number
    /** Restricts addresses to one organization. */
    orgId?: string | undefined
    /** Restricts addresses to one project. */
    projectId?: string | undefined
    /** Restricts addresses to one provider. */
    providerId?: string | undefined
    /** Restricts addresses to any listed provider. */
    providerIds?: readonly string[] | undefined
    /** Exact address resource id, deposit address, or related transaction reference. */
    query?: string | undefined
    /** Restricts addresses to one source chain. */
    sourceChainId?: string | undefined
    /** Restricts addresses to any listed source chain. */
    sourceChainIds?: readonly string[] | undefined
    /** Restricts addresses to one lifecycle status. */
    status?: Record['status'] | undefined
    /** Restricts addresses to any listed lifecycle status. */
    statuses?: readonly Record['status'][] | undefined
  }

  /** Cursor fields for the last address on the previous page. */
  type Cursor = {
    /** Deposit address id, used as a deterministic tie-breaker. */
    id: string
    /** Latest material update time. */
    updatedAt: string
  }
}

/** Summarizes deposit addresses that require operator action. */
export async function summarize(
  db: Db.Db,
  options: summarize.Options = {},
): Promise<summarize.Result> {
  let query = db.kysely.selectFrom('routes_deposit_addresses')
  if (options.environments !== undefined)
    query = query.where('environment', 'in', options.environments)
  else if (options.environment !== undefined)
    query = query.where('environment', '=', options.environment)
  if (options.orgId !== undefined) query = query.where('orgId', '=', options.orgId)
  if (options.projectId !== undefined) query = query.where('projectId', '=', options.projectId)
  if (options.providerIds !== undefined)
    query = query.where('providerId', 'in', options.providerIds)
  else if (options.providerId !== undefined)
    query = query.where('providerId', '=', options.providerId)
  if (options.sourceChainIds !== undefined)
    query = query.where('sourceChainId', 'in', options.sourceChainIds)
  else if (options.sourceChainId !== undefined)
    query = query.where('sourceChainId', '=', options.sourceChainId)
  const row = await query
    .select((eb) => [
      eb.fn.countAll<number>().filterWhere('status', '=', 'action-required').as('actionRequired'),
      eb.fn
        .min(eb.fn.coalesce('statusUpdatedAt', 'updatedAt'))
        .filterWhere('status', '=', 'action-required')
        .as('oldestActiveAt'),
    ])
    .executeTakeFirstOrThrow()
  return {
    actionRequired: Number(row.actionRequired),
    inProgress: 0,
    oldestActiveAt: row.oldestActiveAt,
  }
}

export declare namespace summarize {
  /** Attribution and route filters. */
  type Options = {
    /** Restricts addresses to one route environment. */
    environment?: Record['environment'] | undefined
    /** Restricts addresses to any listed route environment. */
    environments?: readonly Record['environment'][] | undefined
    /** Restricts addresses to one organization. */
    orgId?: string | undefined
    /** Restricts addresses to one project. */
    projectId?: string | undefined
    /** Restricts addresses to one provider. */
    providerId?: string | undefined
    /** Restricts addresses to any listed provider. */
    providerIds?: readonly string[] | undefined
    /** Restricts addresses to one source chain. */
    sourceChainId?: string | undefined
    /** Restricts addresses to any listed source chain. */
    sourceChainIds?: readonly string[] | undefined
  }

  /** Current deposit-address work totals. */
  type Result = {
    /** Addresses that require operator action. */
    actionRequired: number
    /** Deposit addresses never count as processing work. */
    inProgress: 0
    /** When the oldest address entered action-required status. */
    oldestActiveAt: string | null
  }
}

/** Gets the reusable address that matches one owner and route identity. */
export function getMatching(db: Db.Db, match: Match): Promise<Record | undefined> {
  let query = db.kysely
    .selectFrom('routes_deposit_addresses')
    .selectAll()
    .where('destinationTokenKey', '=', match.destinationTokenKey)
    .where('environment', '=', match.environment)
    .where('orgId', '=', match.orgId)
    .where('recipient', '=', match.recipient)
    .where('refundAddress', '=', match.refundAddress)
    .where('sourceChainId', '=', match.sourceChainId)
    .where('sourceTokenKey', '=', match.sourceTokenKey)
    .where('subsidize', '=', match.subsidize)
  query =
    match.projectId === undefined
      ? query.where('projectId', 'is', null)
      : query.where('projectId', '=', match.projectId)
  return query.executeTakeFirst()
}

/** Lists active provider addresses for one source-chain token. */
export function listActiveSources(
  db: Db.Db,
  options: listActiveSources.Options,
): Promise<readonly Record[]> {
  return db.kysely
    .selectFrom('routes_deposit_addresses')
    .selectAll()
    .where('providerId', '=', options.providerId)
    .where('sourceChainId', '=', options.sourceChainId)
    .where('sourceTokenKey', '=', options.sourceTokenKey)
    .where('status', '=', 'active')
    .orderBy('id', 'asc')
    .execute()
}

export declare namespace listActiveSources {
  /** Provider and source asset identifying one observer shard. */
  type Options = {
    /** Route provider that owns the deposit addresses. */
    providerId: string
    /** Source chain CAIP-2 id. */
    sourceChainId: string
    /** Source token key. */
    sourceTokenKey: string
  }
}

/** Claims a bounded batch of active addresses due for provider reconciliation. */
export function claimDue(db: Db.Db, options: claimDue.Options): Promise<readonly Record[]> {
  return db.transaction(async (tx) => {
    const due = await tx.kysely
      .selectFrom('routes_deposit_addresses')
      .select('id')
      .where('nextPollAt', '<=', options.now)
      .where('providerId', '=', options.providerId)
      .where('status', '=', 'active')
      .where((eb) =>
        eb.or([eb('pollLeaseUntil', 'is', null), eb('pollLeaseUntil', '<=', options.now)]),
      )
      .orderBy('nextPollAt', 'asc')
      .orderBy('id', 'asc')
      .limit(options.limit)
      .forUpdate()
      .skipLocked()
      .execute()
    if (due.length === 0) return []
    return tx.kysely
      .updateTable('routes_deposit_addresses')
      .set({
        pollLeaseUntil: options.leaseUntil,
        pollLeaseVersion: sql<number>`poll_lease_version + 1`,
      })
      .where(
        'id',
        'in',
        due.map((record) => record.id),
      )
      .returningAll()
      .execute()
  })
}

export declare namespace claimDue {
  /** Provider and lease bounds for one reconciliation claim. */
  type Options = {
    /** Exclusive upper bound on rows claimed in one tick. */
    limit: number
    /** Expiry time assigned to each claimed lease. */
    leaseUntil: string
    /** Current time used for due and expired-lease comparisons. */
    now: string
    /** Route provider whose addresses should be claimed. */
    providerId: string
  }
}

/** Claims one active address unless another reconciliation owns its lease. */
export function claim(db: Db.Db, options: claim.Options): Promise<Record | undefined> {
  return db.kysely
    .updateTable('routes_deposit_addresses')
    .set({
      pollLeaseUntil: options.leaseUntil,
      pollLeaseVersion: sql<number>`poll_lease_version + 1`,
    })
    .where('id', '=', options.id)
    .where('status', '=', 'active')
    .where((eb) =>
      eb.or([eb('pollLeaseUntil', 'is', null), eb('pollLeaseUntil', '<=', options.now)]),
    )
    .returningAll()
    .executeTakeFirst()
}

export declare namespace claim {
  /** Identity and lease bounds for one reconciliation claim. */
  type Options = {
    /** Route deposit address id (`rda_…`). */
    id: string
    /** Expiry time assigned to the claimed lease. */
    leaseUntil: string
    /** Current time used for expired-lease comparison. */
    now: string
  }
}

/** Extends an active reconciliation lease without changing its fence. */
export async function renewPoll(db: Db.Db, options: renewPoll.Options): Promise<boolean> {
  return Boolean(
    await db.kysely
      .updateTable('routes_deposit_addresses')
      .set({ pollLeaseUntil: options.leaseUntil })
      .where('id', '=', options.id)
      .where('pollLeaseUntil', 'is not', null)
      .where('pollLeaseVersion', '=', options.pollLeaseVersion)
      .where('status', '=', 'active')
      .returning('id')
      .executeTakeFirst(),
  )
}

export declare namespace renewPoll {
  /** Lease identity and renewed expiry guarded by the current fence. */
  type Options = {
    /** Route deposit address id (`rda_…`). */
    id: string
    /** Replacement reconciliation lease expiry. */
    leaseUntil: string
    /** Fencing token returned by the claim operation. */
    pollLeaseVersion: number
  }
}

/** Saves reconciliation progress before processing a page under an owned lease. */
export function checkpointPoll(db: Db.Db, options: checkpointPoll.Options) {
  return db.kysely
    .updateTable('routes_deposit_addresses')
    .set({ providerState: options.providerState })
    .where('id', '=', options.id)
    .where('pollLeaseVersion', '=', options.pollLeaseVersion)
    .returningAll()
    .executeTakeFirst()
}

export declare namespace checkpointPoll {
  /** Progress update fenced by the address reconciliation lease. */
  type Options = Pick<completePoll.Options, 'id' | 'pollLeaseVersion' | 'providerState'>
}

/** Completes an owned reconciliation lease and schedules the address again. */
export function completePoll(
  db: Db.Db,
  options: completePoll.Options,
): Promise<Record | undefined> {
  return db.kysely
    .updateTable('routes_deposit_addresses')
    .set({
      lastPolledAt: options.lastPolledAt,
      nextPollAt: options.nextPollAt,
      pollFailureCount: 0,
      pollLeaseUntil: null,
      providerRequestIds: JSON.stringify(options.providerRequestIds),
      providerState: options.providerState,
    })
    .where('id', '=', options.id)
    .where('pollLeaseVersion', '=', options.pollLeaseVersion)
    .returningAll()
    .executeTakeFirst()
}

export declare namespace completePoll {
  /** Successful reconciliation result guarded by the claimed lease version. */
  type Options = {
    /** Route deposit address id (`rda_…`). */
    id: string
    /** Time when provider reconciliation completed. */
    lastPolledAt: string
    /** Time when this address becomes due again. */
    nextPollAt: string
    /** Fencing token returned by {@link claimDue}. */
    pollLeaseVersion: number
    /** Provider request identifiers retained from address provisioning. */
    providerRequestIds: readonly string[]
    /** Bounded provider reconciliation state. */
    providerState: Record['providerState']
  }
}

/** Releases an owned reconciliation lease after failure and schedules a retry. */
export function failPoll(db: Db.Db, options: failPoll.Options): Promise<Record | undefined> {
  return db.kysely
    .updateTable('routes_deposit_addresses')
    .set({
      nextPollAt: options.nextPollAt,
      pollFailureCount: sql<number>`poll_failure_count + 1`,
      pollLeaseUntil: null,
    })
    .where('id', '=', options.id)
    .where('pollLeaseVersion', '=', options.pollLeaseVersion)
    .returningAll()
    .executeTakeFirst()
}

export declare namespace failPoll {
  /** Failed reconciliation result guarded by the claimed lease version. */
  type Options = {
    /** Route deposit address id (`rda_…`). */
    id: string
    /** Time when this address becomes eligible for retry. */
    nextPollAt: string
    /** Fencing token returned by {@link claimDue}. */
    pollLeaseVersion: number
  }
}

/** Applies a version-guarded deposit address status update. */
export function update(db: Db.Db, options: update.Options): Promise<Record | undefined> {
  return db.kysely
    .updateTable('routes_deposit_addresses')
    .set({
      status: options.status,
      statusUpdatedAt: options.statusUpdatedAt,
      updatedAt: options.updatedAt,
      version: options.version,
    })
    .where('id', '=', options.id)
    .where('version', '=', options.expectedVersion)
    .returningAll()
    .executeTakeFirst()
}

export declare namespace update {
  /** Guarded address update fields. */
  type Options = {
    /** Stored version required for the update. */
    expectedVersion: number
    /** Route deposit address id (`rda_…`). */
    id: string
    /** Replacement lifecycle status. */
    status: Record['status']
    /** When the address entered its current status. */
    statusUpdatedAt: Record['statusUpdatedAt']
    /** New material update time. */
    updatedAt: string
    /** New material version. */
    version: number
  }
}

/** The capacity claim was replaced before it could be reserved. */
export class CapacityClaimNotFoundError extends Error {
  override readonly name = 'RoutesDepositAddresses.CapacityClaimNotFoundError'
}

/** Provider address conflicted with a different customer-visible identity. */
export class ConflictError extends Error {
  override name = 'RoutesDepositAddresses.ConflictError'
}

/** Organization deposit-address cap reached. */
export class LimitExceededError extends Error {
  override name = 'RoutesDepositAddresses.LimitExceededError'

  /** Maximum durable deposit addresses the organization may own. */
  limit: number

  constructor(limit: number) {
    super(`Route deposit address limit reached (${limit}).`)
    this.limit = limit
  }
}

/** Another request owns provider provisioning for the same reusable match. */
export class MatchInProgressError extends Error {
  override readonly name = 'RoutesDepositAddresses.MatchInProgressError'
}

/** Owning organization disappeared before address persistence. */
export class OwnerNotFoundError extends Error {
  override name = 'RoutesDepositAddresses.OwnerNotFoundError'
}