import type { Hex } from 'ox'
import { TxEnvelopeTempo } from 'ox/tempo'
import {
  type Address,
  type Client,
  type EIP1193RequestFn,
  type TransactionReceipt,
  TransactionReceiptNotFoundError,
} from 'viem'
import { getTransactionReceipt } from 'viem/actions'

import * as Db from '../../db/Db.js'
import * as BillingSettings from '../../db/tables/billingSettings.js'
import * as Projects from '../../db/tables/projects.js'
import * as SponsorshipAttributions from '../../db/tables/sponsorshipAttributions.js'
import * as SponsorshipReconciliationCursors from '../../db/tables/sponsorshipReconciliationCursors.js'
import * as SponsoredTransactions from '../../db/tables/sponsoredTransactions.js'
import * as Fees from '../../internal/Fees.js'
import * as Tidx from '../../internal/Tidx.js'
import * as Value from '../../internal/Value.js'
import * as Viem from '../../internal/Viem.js'
import * as Billing from '../management/Billing.js'
import * as Zones from '../Zones.js'

/**
 * Runs one finalization pass over pending sponsored transactions: rows with a
 * receipt finalize with the actual fee (`gasUsed × effectiveGasPrice`, in
 * fee-token base units); rows still unmined past the pending TTL fail, as do
 * hash-less fill intents the TTL outlives. Rows whose receipt lookup errors
 * persist past the error TTL fail so they cannot monopolize every batch.
 *
 * @param db - The database holding `sponsored_transactions`.
 * @param options - Finalization options.
 * @returns Counters for the pass.
 */
export async function finalize(db: Db.Db, options: finalize.Options): Promise<finalize.Result> {
  const {
    batchSize = 100,
    getClient,
    now = () => new Date(),
    pendingTtlMs = 3_600_000,
    receiptErrorTtlMs = 86_400_000,
  } = options

  const rows = await SponsoredTransactions.claimPending(db, {
    attemptedAt: now().toISOString(),
    limit: batchSize,
  })
  const result = { errors: 0, failed: 0, finalized: 0, pending: 0 }

  for (const row of rows) {
    // Fill intents carry no hash until reconciliation matches them on-chain;
    // here only the TTL can settle them.
    if (row.transactionHash === null) {
      if (Date.parse(row.createdAt) + pendingTtlMs <= now().getTime()) {
        await SponsoredTransactions.fail(db, row.id, now().toISOString())
        result.failed++
      } else result.pending++
      continue
    }

    let receipt: TransactionReceipt | null
    try {
      receipt = await getTransactionReceipt(getClient(row.chainId), {
        hash: row.transactionHash as Hex.Hex,
      })
    } catch (error) {
      // Unmined transactions surface as a typed not-found; let the TTL decide.
      if (error instanceof TransactionReceiptNotFoundError) receipt = null
      else {
        result.errors++
        if (Date.parse(row.createdAt) + receiptErrorTtlMs <= now().getTime()) {
          await SponsoredTransactions.fail(db, row.id, now().toISOString())
          result.failed++
        } else result.pending++
        continue
      }
    }

    if (receipt) {
      // A reverted transaction still finalizes: the fee payer paid its fees.
      const feeAmount = Fees.fromGas(receipt.gasUsed, receipt.effectiveGasPrice)
      await SponsoredTransactions.finalize(db, row.id, {
        feeAmount: feeAmount.toString(),
        finalizedAt: now().toISOString(),
      })
      result.finalized++
    } else if (Date.parse(row.createdAt) + pendingTtlMs <= now().getTime()) {
      await SponsoredTransactions.fail(db, row.id, now().toISOString())
      result.failed++
    } else result.pending++
  }

  return result
}

export declare namespace finalize {
  /** Options for {@link finalize}. */
  type Options = {
    /** Maximum pending rows processed per pass. @default 100 */
    batchSize?: number | undefined
    /** Client resolver keyed by the row's chain id. */
    getClient: (chainId: number) => Client
    /** Clock; injectable for tests. */
    now?: (() => Date) | undefined
    /** Age after which an unmined sponsorship fails. @default 1 hour */
    pendingTtlMs?: number | undefined
    /** Age after which repeated receipt lookup errors fail a row. @default 24 hours */
    receiptErrorTtlMs?: number | undefined
  }

  /** Counters returned by one finalization pass. */
  type Result = {
    /** Receipt lookup errors that prevented finalization work. */
    errors: number
    /** Rows marked failed (unmined past the pending TTL). */
    failed: number
    /** Rows finalized with an actual fee. */
    finalized: number
    /** Rows left pending (unmined, or receipt lookup errored within the error TTL). */
    pending: number
  }
}

/**
 * Runs one reconciliation pass over pending fill intents (rows with no
 * transaction hash): fetches the fee payer's recent on-chain transactions
 * from TIDX, recomputes each candidate's fee-payer sign payload, and fills
 * matching intents' hashes so the receipt pass can finalize them. Query or
 * candidate errors leave intents for the next pass.
 *
 * @param db - The database holding `sponsored_transactions`.
 * @param options - Reconciliation options.
 * @returns Counters for the pass.
 */
export async function reconcile(db: Db.Db, options: reconcile.Options): Promise<reconcile.Result> {
  const {
    batchSize = 100,
    candidateCap = 500,
    failedIntentTtlMs = SponsoredTransactions.failedIntentRecoveryTtlMs,
    feePayer,
    getClient,
    getTidx,
    skewMs = 120_000,
  } = options

  const failedSince = new Date(Date.now() - failedIntentTtlMs).toISOString()
  // Old failed intents must not move the active scan behind its candidate cap.
  const [intents, recoveryWindows] = await Promise.all([
    SponsoredTransactions.listIntents(db, { limit: batchSize }),
    SponsoredTransactions.listRecoveryWindows(db, { failedSince }),
  ])
  const result = { errors: 0, matched: 0, scanned: 0 }
  if (intents.length === 0 && recoveryWindows.length === 0) return result

  const groups = new Map<number, SponsoredTransactions.Record[]>()
  for (const intent of intents)
    groups.set(intent.chainId, [...(groups.get(intent.chainId) ?? []), intent])

  for (const [chainId, rows] of groups) {
    if (candidateCap <= 0) continue
    const bySignPayload = new Map(rows.map((row) => [row.signPayload, row]))
    const since = formatTidxTimestamp(
      Math.min(...rows.map((row) => Date.parse(row.createdAt))) - skewMs,
    )

    let candidates: readonly globalThis.Record<string, unknown>[]
    try {
      // Dynamic filter values; the tidx client's query type wants literals.
      const fetched = await getTidx(chainId).fetch({
        chainId,
        query:
          `SELECT hash, "from" FROM txs WHERE fee_payer = '${feePayer.toLowerCase()}' AND block_timestamp >= '${since}' ORDER BY block_timestamp ASC LIMIT ${candidateCap}` as string,
      })
      candidates = fetched.rows
    } catch {
      // Indexer outage must not fail intents; retry on the next pass.
      result.errors++
      continue
    }

    for (const candidate of candidates) {
      if (bySignPayload.size === 0) break
      const hash = candidate['hash'] as Hex.Hex | undefined
      const sender = candidate['from'] as Address | undefined
      if (!hash || !sender) continue
      result.scanned++
      try {
        const serialized = await getRawTransaction(getClient(chainId), hash)
        if (!serialized) {
          result.errors++
          continue
        }
        if (!serialized.startsWith(TxEnvelopeTempo.serializedType)) continue
        const signPayload = TxEnvelopeTempo.getFeePayerSignPayload(
          TxEnvelopeTempo.deserialize(serialized as TxEnvelopeTempo.Serialized),
          { sender },
        )
        const intent = bySignPayload.get(signPayload)
        if (!intent) continue
        await SponsoredTransactions.assignTransactionHash(
          db,
          intent.id,
          hash,
          await options.limitFor?.(intent),
        )
        bySignPayload.delete(signPayload)
        result.matched++
      } catch (error) {
        // Spend-limit enforcement is an expected policy refusal; dependency,
        // lookup, and malformed-envelope failures remain operational errors.
        if (!(error instanceof SponsoredTransactions.PeriodSpendLimitError)) result.errors++
      }
    }
    // A truncated active scan cannot prove that the remaining intents were never mined.
    if (candidates.length === candidateCap && bySignPayload.size > 0) result.errors++
  }

  const feePayer_lower = feePayer.toLowerCase() as Address
  for (const window of recoveryWindows) {
    const chainId = window.chainId
    if (candidateCap <= 0) continue
    const cursor = await SponsorshipReconciliationCursors.load(db, {
      chainId,
      feePayer: feePayer_lower,
    })
    const keyset = !cursor.complete
      ? `AND block_num <= ${cursor.blockNumber} AND (block_num < ${cursor.blockNumber} OR idx < ${cursor.transactionIndex})`
      : ''
    const since = formatTidxTimestamp(Date.parse(window.createdAt) - skewMs)

    let candidates: readonly globalThis.Record<string, unknown>[]
    try {
      const fetched = await getTidx(chainId).fetch({
        chainId,
        query:
          `SELECT hash, "from", block_num, idx FROM txs WHERE fee_payer = '${feePayer_lower}' AND block_timestamp >= '${since}' ${keyset} ORDER BY block_num DESC, idx DESC LIMIT ${candidateCap}` as string,
      })
      candidates = fetched.rows
    } catch {
      result.errors++
      continue
    }

    type Match = {
      hash?: Hex.Hex | undefined
      position: NonNullable<ReturnType<typeof candidatePosition>>
      signPayload?: string | undefined
    }
    const matches: Match[] = []
    let stopped = false
    let retryCount = 0
    let retryKey: string | null = null
    for (const candidate of candidates) {
      const position = candidatePosition(candidate)
      if (!position) {
        result.errors++
        stopped = true
        break
      }
      const hash = candidate['hash'] as Hex.Hex | undefined
      const sender = candidate['from'] as Address | undefined
      result.scanned++
      try {
        if (!hash || !sender) throw new Error('Missing recovery candidate identity.')
        const serialized = await getRawTransaction(getClient(chainId), hash)
        if (!serialized) throw new Error('Recovery candidate unavailable from RPC.')
        if (!serialized.startsWith(TxEnvelopeTempo.serializedType)) {
          matches.push({ position })
          continue
        }
        matches.push({
          hash,
          position,
          signPayload: TxEnvelopeTempo.getFeePayerSignPayload(
            TxEnvelopeTempo.deserialize(serialized as TxEnvelopeTempo.Serialized),
            { sender },
          ),
        })
      } catch {
        result.errors++
        const key = `${position.blockNumber}:${position.transactionIndex}`
        const attempts = cursor.retryKey === key ? cursor.retryCount + 1 : 1
        // Retry transient failures across three passes, then revisit on the next complete scan.
        if (attempts >= 3) {
          matches.push({ position })
          continue
        }
        retryCount = attempts
        retryKey = key
        stopped = true
        break
      }
    }

    let recoverable: SponsoredTransactions.Record[]
    try {
      recoverable = await SponsoredTransactions.listRecoverableIntents(db, {
        failedSince,
        signPayloads: matches.flatMap((match) =>
          match.signPayload === undefined ? [] : [match.signPayload],
        ),
      })
    } catch {
      result.errors++
      continue
    }
    const bySignPayload = new Map(recoverable.map((intent) => [intent.signPayload, intent]))
    let lastProcessed: Match['position'] | undefined
    for (const match of matches) {
      const signPayload = match.signPayload
      const intent = signPayload ? bySignPayload.get(signPayload) : undefined
      if (!signPayload || !intent || !match.hash) {
        lastProcessed = match.position
        continue
      }
      try {
        await SponsoredTransactions.assignTransactionHash(
          db,
          intent.id,
          match.hash,
          await options.limitFor?.(intent),
        )
        bySignPayload.delete(signPayload)
        result.matched++
      } catch (error) {
        // Policy refusals must not pin the cursor; operational errors retry this candidate.
        if (!(error instanceof SponsoredTransactions.PeriodSpendLimitError)) {
          result.errors++
          stopped = true
          break
        }
      }
      lastProcessed = match.position
    }

    const remaining = await SponsoredTransactions.listRecoveryWindows(db, { failedSince })
    const complete =
      (!stopped && candidates.length < candidateCap) ||
      !remaining.some((window) => window.chainId === chainId)
    await SponsorshipReconciliationCursors.save(db, {
      ...cursor,
      ...lastProcessed,
      complete: complete || (!lastProcessed && cursor.complete),
      retryCount: complete ? 0 : retryCount,
      retryKey: complete ? null : retryKey,
    })
  }

  return result
}

export declare namespace reconcile {
  /** Options for {@link reconcile}. */
  type Options = {
    /** Maximum active intents considered per pass. @default 100 */
    batchSize?: number | undefined
    /** Maximum candidates per chain in each active or recovery scan. @default 500 */
    candidateCap?: number | undefined
    /** How long failed intents remain reconciliation candidates. @default 24 hours */
    failedIntentTtlMs?: number | undefined
    /** Fee payer address whose on-chain transactions are candidates. */
    feePayer: Address
    /** Client resolver keyed by the intent's chain id. */
    getClient: (chainId: number) => Client
    /** TIDX client resolver keyed by the intent's chain id. */
    getTidx: (chainId: number) => Tidx.Client
    /** Resolves the current period spend limit before reopening a failed intent. */
    limitFor?:
      | ((
          intent: SponsoredTransactions.Record,
        ) => Promise<SponsoredTransactions.reserve.Limit | undefined>)
      | undefined
    /** Lookback slack before the oldest intent's creation. @default 2 minutes */
    skewMs?: number | undefined
  }

  /** Counters returned by one reconciliation pass. */
  type Result = {
    /** Dependency errors, invalid candidates, or incomplete active scans. */
    errors: number
    /** Intents matched to an on-chain transaction. */
    matched: number
    /** On-chain candidates inspected. */
    scanned: number
  }
}

/** Reads a candidate's stable TIDX keyset position. */
function candidatePosition(candidate: globalThis.Record<string, unknown>) {
  const blockNumber = Value.toIntegerString(candidate['block_num'])
  const transactionIndex = Value.toNumber(candidate['idx'])
  if (
    blockNumber === undefined ||
    transactionIndex === undefined ||
    !Number.isInteger(transactionIndex) ||
    transactionIndex < 0
  )
    return undefined
  return { blockNumber, transactionIndex }
}

/** Formats a timestamp for TIDX's ClickHouse `DateTime64` comparisons. */
function formatTidxTimestamp(value: number) {
  return new Date(value)
    .toISOString()
    .replace('T', ' ')
    .replace(/\.\d+Z$/, '')
}

/**
 * Fetches a mined transaction's canonical envelope bytes.
 * `eth_getRawTransactionByHash` is a node extension absent from viem's typed
 * RPC schema, hence the request-fn narrowing.
 */
function getRawTransaction(client: Client, hash: Hex.Hex) {
  type Request = EIP1193RequestFn<
    [
      {
        Method: 'eth_getRawTransactionByHash'
        Parameters: readonly [Hex.Hex]
        ReturnType: Hex.Hex | null
      },
    ]
  >
  return (client.request as Request)({ method: 'eth_getRawTransactionByHash', params: [hash] })
}

/**
 * Builds a scheduled sponsorship finalizer bound to a database and RPC config.
 * A host runtime drives {@link createFinalizer.Finalizer.tick} on a timer (a
 * Cloudflare `scheduled` cron handler, a self-host `setInterval`, …); each
 * tick reconciles fill intents (when `feePayer` + `tidx` are configured) and
 * runs one {@link finalize} pass, with memoized per-chain clients.
 */
export function createFinalizer(options: createFinalizer.Options): createFinalizer.Finalizer {
  const zones = new Map((options.zones ?? []).map((zone) => [zone.id, zone]))
  const chainIds = Zones.chainIds({ sourceId: Viem.chainId.mainnet, zones: zones.values() })
  const getClient = Viem.createGetClient({
    defaultChainId: options.defaultChainId,
    rpc: options.rpc,
    zones: options.zones,
  })
  const getTidx = options.tidx
    ? Tidx.createGetClient({ defaultChainId: options.defaultChainId, tidx: options.tidx })
    : undefined
  const feePayer = options.feePayer

  return {
    async tick() {
      // Resolve per tick so Workers factories build a fresh Hyperdrive-pooled
      // connection each run.
      const db = Db.get(options.db)
      // Reconcile first so freshly matched intents finalize in the same tick.
      const reconciled =
        feePayer && getTidx
          ? await reconcile(db, {
              batchSize: options.batchSize,
              feePayer,
              getClient,
              getTidx: (chainId) => getTidx(chainId as Viem.ChainId, { zone: zones.get(chainId) }),
              limitFor: async (intent) => {
                if (!Viem.isMainnet(zones.get(intent.chainId)?.sourceId ?? intent.chainId))
                  return undefined
                if (!intent.billable) {
                  if (intent.sponsorshipAttributionId) {
                    const attribution = await SponsorshipAttributions.get(db, {
                      id: intent.sponsorshipAttributionId,
                      orgId: intent.orgId,
                    })
                    // Missing promotion state cannot prove the intent remains within Tempo's budget.
                    if (!attribution)
                      return {
                        sponsorshipAttributionId: intent.sponsorshipAttributionId,
                        billable: false,
                        chainIds,
                        max: -1n,
                        since: intent.createdAt,
                      }
                    if (attribution.startsAt === null)
                      return {
                        sponsorshipAttributionId: intent.sponsorshipAttributionId,
                        billable: false,
                        chainIds,
                        max: -1n,
                        since: intent.createdAt,
                      }
                    if (attribution.spendLimit === null) return undefined
                    return {
                      billable: false,
                      chainIds,
                      max: BigInt(attribution.spendLimit),
                      since: attribution.startsAt,
                      sponsorshipAttributionId: intent.sponsorshipAttributionId,
                    }
                  }
                  if (!intent.projectId) return undefined
                  const project = await Projects.get(db, intent.projectId)
                  // Rows recorded before canonical attribution retain their project-scoped cap.
                  if (!project || project.sponsorshipSubsidyStartsAt === null)
                    return {
                      billable: false,
                      chainIds,
                      max: -1n,
                      projectId: intent.projectId,
                      since: intent.createdAt,
                    }
                  if (project.sponsorshipSpendLimit === null) return undefined
                  return {
                    billable: false,
                    chainIds,
                    max: BigInt(project.sponsorshipSpendLimit),
                    projectId: intent.projectId,
                    since: project.sponsorshipSubsidyStartsAt,
                  }
                }
                if (!intent.billable) return undefined
                const settings = await BillingSettings.get(db, intent.orgId)
                if (settings?.spendLimit == null) return undefined
                return {
                  chainIds,
                  max: Billing.toBaseUnits(settings.spendLimit),
                  since: Billing.periodStart(settings.period),
                }
              },
            })
          : { errors: 0, matched: 0, scanned: 0 }
      const finalized = await finalize(db, {
        batchSize: options.batchSize,
        getClient,
        now: options.now,
        pendingTtlMs: options.pendingTtlMs,
      })
      return {
        ...finalized,
        ...reconciled,
        errors: finalized.errors + reconciled.errors,
      }
    },
  }
}

export declare namespace createFinalizer {
  /** Options for {@link createFinalizer}. */
  type Options = Pick<finalize.Options, 'batchSize' | 'now' | 'pendingTtlMs'> & {
    /** Database holding sponsorship rows, or a factory resolved per tick (Workers/Hyperdrive). */
    db: Db.Source
    /** Default chain id for the TIDX client resolver. */
    defaultChainId?: Viem.ChainId | undefined
    /** Fee payer address whose on-chain transactions reconcile fill intents; omit to skip reconciliation. */
    feePayer?: Address | undefined
    /** RPC options for the receipt-lookup clients. */
    rpc?: Viem.getClient.Rpc | undefined
    /** TIDX query client options for intent reconciliation; omit to skip reconciliation. */
    tidx?: Tidx.getClient.Tidx | undefined
    /** Zone chains used to resolve internal receipt-lookup endpoints. */
    zones?: Viem.createGetClient.Options['zones'] | undefined
  }

  /** A scheduled sponsorship finalizer. */
  type Finalizer = {
    /** Runs one reconcile + finalize pass. */
    tick(): Promise<finalize.Result & reconcile.Result>
  }
}