import { createHash } from 'node:crypto';
import type { Pool, PoolClient } from 'pg';
import { z } from 'zod';
import { XMR_RPC_UINT64_MAX } from '../../shared/xmr-amount.ts';
import type { MoneroIncomingObservation, MoneroObservationPage } from './monero-observer.ts';
import { CUSTODY_WRITE_LOCK, verifyCustodySchema } from './postgres-schema.ts';
import { verifyPilotCustodyIdentity } from './pilot-custody-identity.ts';
import { validateXmrAddress } from './xmr.ts';

const MAX_SCAN_RANGE = 100_000;
const MAX_OVERLAP = 10_000;
const fingerprintSchema = z.string().regex(/^[a-f0-9]{64}$/);
const scanInputSchema = z.object({
  scanId: z.string().uuid(),
  walletGenerationId: z.string().uuid(),
  walletFingerprint: fingerprintSchema,
  expectedScanHeight: z.number().int().nonnegative().safe(),
  scannedAt: z.number().int().nonnegative().safe(),
  page: z.object({
    network: z.literal('mainnet'), walletHeight: z.number().int().nonnegative().safe(),
    fromHeight: z.number().int().nonnegative().safe(), throughHeight: z.number().int().nonnegative().safe(),
    primaryAddress: z.string().length(95), incoming: z.array(z.unknown()).max(4096),
  }).strict(),
}).strict();
const observationSchema = z.object({
  observationId: z.string().min(1).max(160), transactionId: fingerprintSchema,
  address: z.string().length(95), amountAtomicXmr: z.string().regex(/^[1-9]\d{0,19}$/),
  height: z.number().int().positive().safe(), confirmations: z.number().int().positive().safe(),
  timestamp: z.number().int().nonnegative().safe(), unlockTime: z.string().regex(/^(0|[1-9]\d{0,19})$/),
  accountIndex: z.number().int().nonnegative().max(0xffff_ffff).safe(),
  addressIndex: z.number().int().nonnegative().max(0xffff_ffff).safe(),
  outputPublicKey: fingerprintSchema, globalOutputIndex: z.number().int().nonnegative().safe(),
  locked: z.boolean(), doubleSpendSeen: z.boolean(),
}).strict();

type PersistInput = {
  scanId: string;
  walletGenerationId: string;
  walletFingerprint: string;
  expectedScanHeight: number;
  scannedAt: number;
  page: MoneroObservationPage;
};
export type PilotObservationClassification = 'assigned' | 'unallocated';
export type PilotObservationScanResult = {
  scanId: string; scanHeight: number; observed: number; assigned: number; unallocated: number;
  changed: number; reappeared: number; notSeen: number; replayed: boolean;
};

export async function readPilotMoneroObservationCursor(pool: Pool, walletGenerationId: string,
  walletFingerprint: string): Promise<number> {
  const generation = z.string().uuid().parse(walletGenerationId);
  const fingerprint = fingerprintSchema.parse(walletFingerprint);
  const identity = await verifyPilotCustodyIdentity(pool, (await pool.query(
    'SELECT custodian_id FROM wk_custody.pilot_custody_identity WHERE singleton=1')).rows[0]?.custodian_id ?? '');
  const result = await pool.query(`SELECT wallet_fingerprint,network,scan_height
    FROM wk_custody.pilot_monero_observer_state WHERE wallet_generation_id=$1`, [generation]);
  if (!result.rowCount) return 0;
  if (result.rows[0].wallet_fingerprint !== fingerprint || result.rows[0].network !== identity.moneroNetwork) {
    throw new Error('Pilot observer cursor is bound to a different wallet or Monero network');
  }
  return Number(result.rows[0].scan_height);
}

function digest(value: unknown): string {
  return createHash('sha256').update(JSON.stringify(value)).digest('hex');
}
function observationDigest(observation: MoneroIncomingObservation, classification: PilotObservationClassification): string {
  return digest([classification, observation.observationId, observation.transactionId, observation.amountAtomicXmr,
    observation.height, observation.timestamp, observation.unlockTime, observation.accountIndex,
    observation.addressIndex, observation.outputPublicKey, observation.globalOutputIndex,
    observation.locked, observation.doubleSpendSeen]);
}

/** Persist pilot-wallet observations separately from mainnet custody observations. Never posts money or credits. */
export async function persistPilotMoneroObservationScan(pool: Pool, input: PersistInput): Promise<PilotObservationScanResult> {
  const parsed = scanInputSchema.parse(input), page = parsed.page;
  if (page.fromHeight > page.throughHeight || page.throughHeight > page.walletHeight ||
      page.throughHeight - page.fromHeight > MAX_SCAN_RANGE || parsed.expectedScanHeight - page.fromHeight > MAX_OVERLAP ||
      page.fromHeight > parsed.expectedScanHeight || page.throughHeight < parsed.expectedScanHeight) {
    throw new Error('Pilot Monero observation range is not contiguous with its durable cursor');
  }
  await validateXmrAddress(page.primaryAddress);
  const observations = page.incoming.map(row => observationSchema.parse(row));
  const seen = new Set<string>();
  for (const observation of observations) {
    await validateXmrAddress(observation.address);
    if (observation.observationId !== `${observation.transactionId}:${observation.outputPublicKey}` ||
        BigInt(observation.amountAtomicXmr) > XMR_RPC_UINT64_MAX || observation.height < page.fromHeight ||
        observation.height > page.throughHeight || observation.confirmations !== page.walletHeight - observation.height + 1 ||
        seen.has(observation.observationId)) throw new Error('Pilot Monero observation is inconsistent with its scan page');
    seen.add(observation.observationId);
  }
  const pageDigest = digest([parsed.walletGenerationId, parsed.walletFingerprint, page.fromHeight, page.throughHeight,
    observations.map(row => [row.observationId, row]).sort((a, b) => String(a[0]).localeCompare(String(b[0])))]);

  const client = await pool.connect();
  try {
    await client.query('BEGIN');
    await client.query("SET LOCAL statement_timeout='10s'; SET LOCAL lock_timeout='3s'; SET LOCAL idle_in_transaction_session_timeout='10s'; SET LOCAL synchronous_commit=on");
    await client.query('SELECT pg_advisory_xact_lock($1::bigint)', [CUSTODY_WRITE_LOCK]);
    await verifyCustodySchema(client);
    const identity = await verifyPilotCustodyIdentity(pool, (await client.query(
      'SELECT custodian_id FROM wk_custody.pilot_custody_identity WHERE singleton=1')).rows[0]?.custodian_id ?? '');
    const state = await client.query('SELECT wallet_fingerprint,network,scan_height FROM wk_custody.pilot_monero_observer_state WHERE wallet_generation_id=$1 FOR UPDATE',
      [parsed.walletGenerationId]);
    const prior = await client.query('SELECT payload_digest,through_height FROM wk_custody.pilot_monero_observation_scans WHERE scan_id=$1', [parsed.scanId]);
    if (prior.rowCount) {
      if (prior.rows[0].payload_digest !== pageDigest) throw new Error('Pilot observation scan ID was reused with different data');
      await client.query('COMMIT');
      return { scanId: parsed.scanId, scanHeight: Number(prior.rows[0].through_height), observed: observations.length,
        assigned: 0, unallocated: 0, changed: 0, reappeared: 0, notSeen: 0, replayed: true };
    }
    if (!state.rowCount) {
      if (parsed.expectedScanHeight !== 0) throw new Error('First pilot observation scan must start at wallet history height zero');
      await client.query(`INSERT INTO wk_custody.pilot_monero_observer_state(wallet_generation_id,wallet_fingerprint,network,scan_height,updated_at)
        VALUES($1,$2,'mainnet',0,$3)`, [parsed.walletGenerationId, parsed.walletFingerprint, parsed.scannedAt]);
    } else if (state.rows[0].wallet_fingerprint !== parsed.walletFingerprint || state.rows[0].network !== 'mainnet' ||
        Number(state.rows[0].scan_height) !== parsed.expectedScanHeight) {
      throw new Error('Pilot observer wallet identity or cursor changed; reread durable state before retry');
    }
    const scanState = await client.query('SELECT wallet_fingerprint FROM wk_custody.pilot_monero_observer_state WHERE wallet_generation_id=$1',
      [parsed.walletGenerationId]);
    if (scanState.rows[0].wallet_fingerprint !== parsed.walletFingerprint) throw new Error('Pilot observer wallet fingerprint mismatch');
    await client.query(`INSERT INTO wk_custody.pilot_monero_observation_scans
      (scan_id,wallet_generation_id,from_height,through_height,payload_digest,completed_at) VALUES($1,$2,$3,$4,$5,$6)`,
    [parsed.scanId, parsed.walletGenerationId, page.fromHeight, page.throughHeight, pageDigest, parsed.scannedAt]);

    let assigned = 0, unallocated = 0, changed = 0, reappeared = 0;
    for (const observation of observations) {
      const allocation = await client.query<{ allocation_id: string }>(`SELECT allocation_id
        FROM wk_custody.pilot_monero_deposit_reservations
        WHERE wallet_generation_id=$1 AND wallet_fingerprint=$2 AND account_index=$3 AND address_index=$4
          AND monero_network='mainnet' AND status='registered' AND monero_address=$5`,
      [parsed.walletGenerationId, parsed.walletFingerprint, observation.accountIndex, observation.addressIndex, observation.address]);
      const classification: PilotObservationClassification = allocation.rowCount ? 'assigned' : 'unallocated';
      const allocationId: string | null = allocation.rowCount ? allocation.rows[0].allocation_id : null;
      const payloadDigest = observationDigest(observation, classification);
      const existing = await client.query(`SELECT payload_digest,currently_present,first_observed_at FROM wk_custody.pilot_monero_observations
        WHERE wallet_generation_id=$1 AND observation_id=$2 FOR UPDATE`, [parsed.walletGenerationId, observation.observationId]);
      const eventType = !existing.rowCount ? 'seen' : !existing.rows[0].currently_present ? 'reappeared'
        : existing.rows[0].payload_digest !== payloadDigest ? 'changed' : null;
      await client.query(`INSERT INTO wk_custody.pilot_monero_observations(
        wallet_generation_id,observation_id,allocation_id,classification,txid,monero_address,amount_atomic_xmr,
        block_height,wallet_height,confirmations,timestamp,unlock_time,account_index,address_index,
        output_public_key,global_output_index,locked,double_spend_seen,payload_digest,currently_present,
        last_seen_scan_id,first_observed_at,last_observed_at)
        VALUES($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,true,$20,$21,$21)
        ON CONFLICT(wallet_generation_id,observation_id) DO UPDATE SET
          amount_atomic_xmr=excluded.amount_atomic_xmr,block_height=excluded.block_height,wallet_height=excluded.wallet_height,
          confirmations=excluded.confirmations,timestamp=excluded.timestamp,unlock_time=excluded.unlock_time,
          locked=excluded.locked,double_spend_seen=excluded.double_spend_seen,payload_digest=excluded.payload_digest,
          currently_present=true,last_seen_scan_id=excluded.last_seen_scan_id,last_observed_at=excluded.last_observed_at`,
      [parsed.walletGenerationId, observation.observationId, allocationId, classification, observation.transactionId,
        observation.address, observation.amountAtomicXmr, observation.height, page.walletHeight, observation.confirmations,
        observation.timestamp, observation.unlockTime, observation.accountIndex, observation.addressIndex,
        observation.outputPublicKey, observation.globalOutputIndex, observation.locked, observation.doubleSpendSeen,
        payloadDigest, parsed.scanId, existing.rowCount ? Number(existing.rows[0].first_observed_at) : parsed.scannedAt]);
      if (classification === 'assigned') assigned++; else unallocated++;
      if (eventType) {
        await client.query(`INSERT INTO wk_custody.pilot_monero_observation_events
          (wallet_generation_id,observation_id,scan_id,event_type,payload_digest,wallet_height,observed_at)
          VALUES($1,$2,$3,$4,$5,$6,$7) ON CONFLICT DO NOTHING`,
        [parsed.walletGenerationId, observation.observationId, parsed.scanId, eventType, payloadDigest, page.walletHeight, parsed.scannedAt]);
        if (eventType === 'changed') changed++;
        if (eventType === 'reappeared') reappeared++;
      }
    }

    const missing = await client.query(`SELECT observation_id,payload_digest FROM wk_custody.pilot_monero_observations
      WHERE wallet_generation_id=$1 AND currently_present=true AND block_height BETWEEN $2 AND $3 AND last_seen_scan_id<>$4 FOR UPDATE`,
    [parsed.walletGenerationId, page.fromHeight, page.throughHeight, parsed.scanId]);
    for (const row of missing.rows) {
      await client.query(`UPDATE wk_custody.pilot_monero_observations SET currently_present=false,last_seen_scan_id=$3
        WHERE wallet_generation_id=$1 AND observation_id=$2`, [parsed.walletGenerationId, row.observation_id, parsed.scanId]);
      await client.query(`INSERT INTO wk_custody.pilot_monero_observation_events
        (wallet_generation_id,observation_id,scan_id,event_type,payload_digest,wallet_height,observed_at)
        VALUES($1,$2,$3,'not-seen',$4,$5,$6) ON CONFLICT DO NOTHING`,
      [parsed.walletGenerationId, row.observation_id, parsed.scanId, row.payload_digest, page.walletHeight, parsed.scannedAt]);
    }
    await client.query(`UPDATE wk_custody.pilot_monero_observer_state SET scan_height=GREATEST(scan_height,$2),updated_at=$3
      WHERE wallet_generation_id=$1`, [parsed.walletGenerationId, page.throughHeight, parsed.scannedAt]);
    await client.query('COMMIT');
    return { scanId: parsed.scanId, scanHeight: Math.max(parsed.expectedScanHeight, page.throughHeight), observed: observations.length,
      assigned, unallocated, changed, reappeared, notSeen: missing.rowCount ?? 0, replayed: false };
  } catch (error) { await client.query('ROLLBACK').catch(() => {}); throw error; }
  finally { client.release(); }
}