Verify and reconcile payments

import { DatabaseSync } from "node:sqlite"

const db = new DatabaseSync("payments.sqlite")
db.exec("PRAGMA foreign_keys = ON")
db.exec("PRAGMA journal_mode = WAL")
db.exec("PRAGMA synchronous = FULL")
db.exec("PRAGMA busy_timeout = 5000")

const schema = [
  `CREATE TABLE IF NOT EXISTS orders (
    id TEXT PRIMARY KEY,
    status TEXT NOT NULL CHECK (status IN ('unpaid', 'paid'))
  )`,
  `CREATE TABLE IF NOT EXISTS instructions (
    order_id TEXT PRIMARY KEY REFERENCES orders(id),
    chain_id INTEGER NOT NULL,
    token TEXT NOT NULL,
    sender TEXT NOT NULL,
    recipient TEXT NOT NULL,
    amount TEXT NOT NULL,
    start_block TEXT NOT NULL
  )`,
  `CREATE TABLE IF NOT EXISTS verified_events (
    event_id TEXT PRIMARY KEY,
    order_id TEXT NOT NULL UNIQUE REFERENCES instructions(order_id),
    block_number TEXT NOT NULL,
    block_hash TEXT NOT NULL,
    tx_hash TEXT NOT NULL,
    log_index INTEGER NOT NULL,
    evidence_json TEXT NOT NULL
  )`,
  `CREATE TABLE IF NOT EXISTS application_effects (
    order_id TEXT PRIMARY KEY REFERENCES orders(id),
    event_id TEXT NOT NULL UNIQUE REFERENCES verified_events(event_id)
  )`,
  `CREATE TABLE IF NOT EXISTS quarantine_records (
    chain_id INTEGER NOT NULL,
    policy TEXT NOT NULL,
    receipt_id TEXT NOT NULL,
    order_id TEXT NOT NULL REFERENCES instructions(order_id),
    evidence_json TEXT NOT NULL,
    state TEXT NOT NULL CHECK (state IN ('quarantined', 'recovered')),
    PRIMARY KEY (chain_id, policy, receipt_id)
  )`,
  `CREATE TABLE IF NOT EXISTS cursors (
    stream TEXT PRIMARY KEY,
    block_number TEXT NOT NULL,
    block_hash TEXT NOT NULL
  )`,
  `CREATE TABLE IF NOT EXISTS outgoing_attempts (
    operation_id TEXT PRIMARY KEY,
    chain_id INTEGER NOT NULL,
    signer TEXT NOT NULL,
    intent_json TEXT NOT NULL,
    nonce TEXT,
    tx_hash TEXT UNIQUE,
    raw TEXT,
    state TEXT NOT NULL CHECK (state IN ('preparing', 'signed', 'submitted', 'final', 'reverted')),
    UNIQUE (chain_id, signer, nonce)
  )`,
  `CREATE UNIQUE INDEX IF NOT EXISTS signer_busy
    ON outgoing_attempts(chain_id, signer)
    WHERE state IN ('preparing', 'signed', 'submitted')`
]
for (const statement of schema) db.exec(statement)

import { decodeEventLog, parseAbiItem, type Address, type Hash, type PublicClient } from "viem"

const transferEvent = parseAbiItem(
  "event Transfer(address indexed from, address indexed to, uint256 value)"
)

type Instruction = {
  chainId: number
  token: Address
  sender: Address
  recipient: Address
  amount: bigint
  startBlock: bigint
}

export async function readFinalizedHeight(rpcUrl: string) {
  const response = await fetch(rpcUrl, {
    method: "POST",
    headers: { "content-type": "application/json" },
    body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "consensus_status", params: [] }),
    signal: AbortSignal.timeout(10_000)
  })
  const body = await response.json()
  const number = body.result?.finalized?.number
  if (!response.ok || body.error || number === undefined) {
    throw new Error("Finalized progress is unavailable")
  }
  if (typeof number === "number" && !Number.isSafeInteger(number)) {
    throw new Error("Unsafe finalized height")
  }
  const height = BigInt(number)
  if (height < 0n) throw new Error("Invalid finalized height")
  return height
}

export async function verifyFinalizedTransfer(
  client: PublicClient,
  rpcUrl: string,
  expected: Instruction,
  hash: Hash,
  logIndex: number
) {
  if (expected.amount <= 0n) throw new Error("Expected amount must be positive")
  if (await client.getChainId() !== expected.chainId) throw new Error("Wrong network")
  const receipt = await client.getTransactionReceipt({ hash })
  if (receipt.transactionHash !== hash || receipt.status !== "success") {
    throw new Error("Transaction hash mismatch or failed execution")
  }
  if (receipt.blockNumber < expected.startBlock) throw new Error("Transfer predates instruction")
  if (await readFinalizedHeight(rpcUrl) < receipt.blockNumber) {
    throw new Error("Transfer is awaiting finality")
  }
  const block = await client.getBlock({ blockNumber: receipt.blockNumber })
  if (block.hash !== receipt.blockHash) throw new Error("Noncanonical receipt")
  const log = receipt.logs.find((entry) => entry.logIndex === logIndex)
  if (!log || log.removed || log.address.toLowerCase() !== expected.token.toLowerCase() ||
      log.transactionHash !== hash || log.blockHash !== receipt.blockHash) {
    throw new Error("Unexpected event identity")
  }
  const { args } = decodeEventLog({
    abi: [transferEvent], data: log.data, topics: log.topics, strict: true
  })
  if (args.from.toLowerCase() !== expected.sender.toLowerCase() ||
      args.to.toLowerCase() !== expected.recipient.toLowerCase() ||
      args.value !== expected.amount) {
    throw new Error("Transfer does not match instruction")
  }
  return {
    eventId: `${expected.chainId}:${hash}:${logIndex}`,
    blockNumber: receipt.blockNumber.toString(),
    blockHash: receipt.blockHash,
    txHash: hash,
    logIndex,
    chainId: expected.chainId,
    token: expected.token.toLowerCase(),
    sender: expected.sender.toLowerCase(),
    recipient: expected.recipient.toLowerCase(),
    amount: expected.amount.toString()
  }
}

type Evidence = Awaited<ReturnType<typeof verifyFinalizedTransfer>>

function commitBlock(
  stream: string,
  previous: { number: string, hash: string },
  block: { number: bigint, hash: string, parentHash: string },
  matches: { orderId: string, evidence: Evidence }[]
) {
  db.exec("BEGIN IMMEDIATE")
  try {
    const cursor = db.prepare("SELECT block_number, block_hash FROM cursors WHERE stream = ?")
      .get(stream) as { block_number: string, block_hash: string } | undefined
    if (!cursor || cursor.block_number !== previous.number || cursor.block_hash !== previous.hash ||
        block.number !== BigInt(previous.number) + 1n || block.parentHash !== previous.hash) {
      throw new Error("Cursor changed or block continuity failed")
    }
    for (const { orderId, evidence } of matches) {
      if (evidence.blockNumber !== block.number.toString() || evidence.blockHash !== block.hash) {
        throw new Error("Evidence belongs to another block")
      }
      const old = db.prepare("SELECT order_id FROM verified_events WHERE event_id = ?")
        .get(evidence.eventId) as { order_id: string } | undefined
      if (old) {
        if (old.order_id !== orderId) throw new Error("Event already assigned")
        continue
      }
      const instruction = db.prepare("SELECT * FROM instructions WHERE order_id = ?")
        .get(orderId) as Record<string, string | number> | undefined
      if (!instruction || instruction.chain_id !== evidence.chainId ||
          instruction.token !== evidence.token || instruction.sender !== evidence.sender ||
          instruction.recipient !== evidence.recipient || instruction.amount !== evidence.amount ||
          BigInt(instruction.start_block) > block.number) {
        throw new Error("Instruction changed or evidence mismatched")
      }
      db.prepare("INSERT INTO verified_events VALUES (?, ?, ?, ?, ?, ?, ?)").run(
        evidence.eventId, orderId, evidence.blockNumber, evidence.blockHash,
        evidence.txHash, evidence.logIndex, JSON.stringify(evidence)
      )
      db.prepare("INSERT INTO application_effects VALUES (?, ?)").run(orderId, evidence.eventId)
      const changed = db.prepare("UPDATE orders SET status = 'paid' WHERE id = ? AND status = 'unpaid'")
        .run(orderId).changes
      if (changed !== 1) throw new Error("Order was already completed")
    }
    db.prepare("UPDATE cursors SET block_number = ?, block_hash = ? WHERE stream = ?")
      .run(block.number.toString(), block.hash, stream)
    db.exec("COMMIT")
  } catch (error) {
    db.exec("ROLLBACK")
    throw error
  }
}

db.prepare(`INSERT INTO outgoing_attempts
  (operation_id, chain_id, signer, intent_json, state)
  VALUES (?, ?, ?, ?, 'preparing')`)
  .run(operationId, chainId, signer.toLowerCase(), JSON.stringify(intent))

const saved = db.prepare(`UPDATE outgoing_attempts
  SET nonce = ?, tx_hash = ?, raw = ?, state = 'signed'
  WHERE operation_id = ? AND state = 'preparing'`)
  .run(nonce.toString(), signed.txHash, signed.raw, operationId)
if (saved.changes !== 1) throw new Error("Attempt ownership lost")

const returnedHash = await signed.send()
if (returnedHash !== signed.txHash) throw new Error("Unexpected transaction hash")
db.prepare("UPDATE outgoing_attempts SET state = 'submitted' WHERE operation_id = ? AND state = 'signed'")
  .run(operationId)

const hash = await client.sendRawTransaction({ serializedTransaction: savedRaw })
if (hash !== savedHash) throw new Error("Unexpected transaction hash")