import type { Pool } from 'pg';
import { randomUUID } from 'node:crypto';
import { z } from 'zod';
import { MONERO_OBSERVER_WORKER_LOCK } from './postgres-schema.ts';
import { persistMoneroObservationScan, type PersistObservationScanResult } from './monero-observation-store.ts';
import type { MoneroObservationScanner } from './monero-observer.ts';

const fingerprintSchema = z.string().regex(/^[a-f0-9]{64}$/);
export const MONERO_OBSERVER_MAX_BLOCKS_PER_PAGE = 100_000;
export const MONERO_OBSERVER_OVERLAP_BLOCKS = 1_000;
export const MONERO_OBSERVER_MAX_PAGES_PER_BATCH = 4;

export type MoneroObserverBatchResult = {
  walletFingerprint: string;
  pages: PersistObservationScanResult[];
  scanHeight: number;
  walletHeight: number;
  caughtUp: boolean;
};

/**
 * Perform a small bounded catch-up batch. The database observation role supplies the durable cursor;
 * the scanner never receives a spend or address-allocation capability.
 */
export async function runMoneroObserverBatch(pool: Pool, walletFingerprint: string, scanner: MoneroObservationScanner,
  options: { now?: () => number; maxPages?: number } = {}): Promise<MoneroObserverBatchResult> {
  fingerprintSchema.parse(walletFingerprint);
  const maxPages = options.maxPages ?? MONERO_OBSERVER_MAX_PAGES_PER_BATCH;
  if (!Number.isSafeInteger(maxPages) || maxPages < 1 || maxPages > MONERO_OBSERVER_MAX_PAGES_PER_BATCH) {
    throw new Error('Invalid Monero observer page budget');
  }
  const pages: PersistObservationScanResult[] = [];
  let walletHeight = 0, scanHeight = 0;
  for (let pageNumber = 0; pageNumber < maxPages; pageNumber++) {
    const state = await pool.query<{ scan_height: string }>(
      'SELECT scan_height::text FROM wk_custody.monero_observer_state WHERE wallet_fingerprint=$1', [walletFingerprint]);
    const expectedScanHeight = state.rowCount ? Number(state.rows[0].scan_height) : 0;
    if (!Number.isSafeInteger(expectedScanHeight) || expectedScanHeight < 0) throw new Error('Invalid durable Monero observer cursor');
    const fromHeight = Math.max(0, expectedScanHeight - MONERO_OBSERVER_OVERLAP_BLOCKS);
    const page = await scanner.scanIncoming(fromHeight, MONERO_OBSERVER_MAX_BLOCKS_PER_PAGE);
    if (page.network !== 'mainnet' || page.fromHeight !== fromHeight || page.throughHeight < expectedScanHeight ||
        page.throughHeight - page.fromHeight >= MONERO_OBSERVER_MAX_BLOCKS_PER_PAGE) {
      throw new Error('Monero observer returned a page outside the worker’s bounded cursor window');
    }
    const result = await persistMoneroObservationScan(pool, { scanId: randomUUID(), walletFingerprint,
      expectedScanHeight, scannedAt: (options.now ?? Date.now)(), page });
    pages.push(result); scanHeight = result.scanHeight; walletHeight = page.walletHeight;
    if (page.throughHeight >= page.walletHeight) break;
  }
  return { walletFingerprint, pages, scanHeight, walletHeight, caughtUp: scanHeight >= walletHeight };
}

export type ObserverWorkerOptions = {
  pollIntervalMs?: number;
  onBatch?: (result: MoneroObserverBatchResult) => void;
  onFailure?: () => void;
  signal?: AbortSignal;
};

function sleep(ms: number, signal?: AbortSignal): Promise<void> {
  if (signal?.aborted) return Promise.resolve();
  return new Promise(resolve => {
    const timer = setTimeout(done, ms); timer.unref();
    function done() { clearTimeout(timer); signal?.removeEventListener('abort', done); resolve(); }
    signal?.addEventListener('abort', done, { once: true });
  });
}

/**
 * Explicit opt-in worker loop. A PostgreSQL session advisory lock prevents duplicate scanner processes;
 * scan failures are redacted and retried with bounded backoff. No journal/credit/release APIs are called.
 */
export async function runMoneroObserverWorker(pool: Pool, walletFingerprint: string, scanner: MoneroObservationScanner,
  options: ObserverWorkerOptions = {}): Promise<void> {
  fingerprintSchema.parse(walletFingerprint);
  const pollIntervalMs = options.pollIntervalMs ?? 30_000;
  if (!Number.isSafeInteger(pollIntervalMs) || pollIntervalMs < 1_000 || pollIntervalMs > 15 * 60_000) {
    throw new Error('Invalid Monero observer polling interval');
  }
  const leader = await pool.connect();
  let acquired = false;
  let retryDelay = pollIntervalMs;
  try {
    const lock = await leader.query<{ locked: boolean }>('SELECT pg_try_advisory_lock($1::bigint) AS locked', [MONERO_OBSERVER_WORKER_LOCK]);
    acquired = lock.rows[0]?.locked === true;
    if (!acquired) throw new Error('A Monero observer worker already holds the database lock');
    while (!options.signal?.aborted) {
      try {
        const result = await runMoneroObserverBatch(pool, walletFingerprint, scanner);
        options.onBatch?.(result);
        retryDelay = pollIntervalMs;
        if (!result.caughtUp) continue;
      } catch {
        options.onFailure?.();
        retryDelay = Math.min(Math.max(pollIntervalMs, retryDelay * 2), 5 * 60_000);
      }
      await sleep(retryDelay, options.signal);
    }
  } finally {
    if (acquired) await leader.query('SELECT pg_advisory_unlock($1::bigint)', [MONERO_OBSERVER_WORKER_LOCK]).catch(() => {});
    leader.release();
  }
}