import { and, eq, inArray, sql } from 'drizzle-orm';
import { normalizedRecords } from '@rareindex/database';
import { NormalizedRecordSchema, logger } from '@rareindex/shared';
import { db } from '../lib/db.ts';
import { emitMany } from '../lib/events.ts';
import { applyRecord, FxMissingError } from './writers.ts';
import { syncFx } from '../fx.ts';

export interface ResolveResult {
  processed: number;
  applied: number;
  duplicates: number;
  unmatched: number;
  rejected: number;
  fxMissing: number;
  touchedAssets: string[];
}

/**
 * Entity resolution + canonical writes (§108 steps 5–9). Pending normalized records are applied in
 * kind order (catalog first so identifiers exist before sales/listings resolve against them).
 */
export async function resolveBatch(opts: { limit?: number; connectorId?: string } = {}): Promise<ResolveResult> {
  const log = logger.child({ component: 'resolver' });
  const limit = opts.limit ?? 500;
  // Atomically claim a batch (status pending → processing) so several resolver workers can run
  // concurrently without processing the same rows; rows are released back to pending on crash.
  const connectorFilter = opts.connectorId ? sql` and connector_id = ${opts.connectorId}` : sql``;
  // Kind priority (catalog first so identifiers exist before sales/listings resolve against them).
  // Claim per kind so the query uses the (status, kind, created_at) index instead of sorting every pending row.
  const KIND_ORDER = ['catalog_item', 'population_report', 'sale', 'price_observation', 'listing', 'auction_lot', 'news_item'];
  let rows: Array<typeof normalizedRecords.$inferSelect> = [];
  for (const kind of KIND_ORDER) {
    rows = (await db().execute(sql`
      with claimed as (
        select id from normalized_records
        where status = 'pending' and kind = ${kind}${connectorFilter}
        order by created_at
        limit ${limit}
        for update skip locked
      )
      update normalized_records n set status = 'processing'
      from claimed where n.id = claimed.id
      returning n.id, n.raw_record_id as "rawRecordId", n.connector_id as "connectorId", n.source_id as "sourceId", n.kind, n.payload, n.seq, n.asset_id as "assetId", n.variant_id as "variantId", n.match_method as "matchMethod", n.match_confidence as "matchConfidence", n.status, n.reject_reason as "rejectReason", n.target_id as "targetId", n.created_at as "createdAt", n.processed_at as "processedAt"
    `)) as unknown as Array<typeof normalizedRecords.$inferSelect>;
    if (rows.length) break;
  }
  const res: ResolveResult = { processed: 0, applied: 0, duplicates: 0, unmatched: 0, rejected: 0, fxMissing: 0, touchedAssets: [] };
  if (rows.length === 0) return res;
  const touched = new Set<string>();
  const events: Parameters<typeof emitMany>[0] = [];
  const missingFx = new Set<string>();
  const updates: Array<{ id: string; set: Partial<typeof normalizedRecords.$inferInsert> }> = [];

  for (const row of rows) {
    res.processed++;
    const parsed = NormalizedRecordSchema.safeParse(row.payload);
    if (!parsed.success) {
      res.rejected++;
      updates.push({ id: row.id, set: { status: 'rejected', rejectReason: `schema: ${parsed.error.issues[0]?.message ?? 'invalid'}`, processedAt: new Date() } });
      continue;
    }
    try {
      const out = await applyRecord(parsed.data);
      if (out.assetId) touched.add(out.assetId);
      if (out.event) events.push({ type: out.event, entityType: out.event === 'entity_created' ? 'asset' : out.event === 'sale_detected' ? 'sale' : 'listing', entityId: out.targetId ?? out.assetId ?? undefined, payload: { assetId: out.assetId, connectorId: row.connectorId } });
      const status = out.status === 'applied' ? 'applied' : out.status === 'duplicate' ? 'applied' : out.status === 'unmatched' ? 'unmatched' : 'rejected';
      if (out.status === 'applied') res.applied++;
      else if (out.status === 'duplicate') res.duplicates++;
      else if (out.status === 'unmatched') res.unmatched++;
      else res.rejected++;
      updates.push({ id: row.id, set: { status, assetId: out.assetId, variantId: out.variantId, targetId: out.targetId, matchMethod: out.method, matchConfidence: out.confidence, rejectReason: out.status === 'duplicate' ? 'duplicate' : (out.reason ?? null), processedAt: new Date() } });
    } catch (err) {
      if (err instanceof FxMissingError) {
        res.fxMissing++;
        missingFx.add(err.currency);
        if (row.rejectReason?.startsWith('fx_missing')) {
          // second attempt after a sync: no rate exists for this currency/date → reject honestly
          res.rejected++;
          updates.push({ id: row.id, set: { status: 'rejected', rejectReason: `${err.message} (no rate after sync)`, processedAt: new Date() } });
        } else {
          updates.push({ id: row.id, set: { status: 'pending', rejectReason: err.message } }); // back to pending until fx sync
        }
        continue;
      }
      const msg = err instanceof Error ? err.message : String(err);
      log.warn({ err, record: row.id }, 'apply failed');
      res.rejected++;
      updates.push({ id: row.id, set: { status: 'rejected', rejectReason: msg.slice(0, 500), processedAt: new Date() } });
    }
  }
  for (const u of updates) await db().update(normalizedRecords).set(u.set).where(eq(normalizedRecords.id, u.id));
  await emitMany(events);
  if (missingFx.size) {
    log.warn({ currencies: [...missingFx] }, 'fx rates missing; syncing');
    try {
      await syncFx({ backfill: true });
    } catch (err) {
      log.error({ err }, 'fx sync failed');
    }
  }
  res.touchedAssets = [...touched];
  log.info({ ...res, touchedAssets: res.touchedAssets.length }, 'resolve batch done');
  return res;
}

export async function pendingCount(): Promise<number> {
  const [row] = await db().select({ n: sql<number>`count(*)::int` }).from(normalizedRecords).where(eq(normalizedRecords.status, 'pending'));
  return row?.n ?? 0;
}

export async function markRejected(ids: string[], reason: string): Promise<void> {
  if (!ids.length) return;
  await db().update(normalizedRecords).set({ status: 'rejected', rejectReason: reason, processedAt: new Date() }).where(inArray(normalizedRecords.id, ids));
}