import { and, eq, isNull, lt, or, sql } from 'drizzle-orm';
import { certificates } from '@rareindex/database';
import { createCrawlContext, listConnectorMeta, loadConnector, missingRequirements, type RareIndexConnector } from '@rareindex/connectors';
import { logger, type NormalizedRecord } from '@rareindex/shared';
import { db } from './lib/db.ts';
import { getRouter } from './lib/router.ts';
import { flushCosts } from './lib/costs.ts';

/**
 * Certificate verification job (SPEC §22): for cert numbers seen on marketplaces that have not been
 * verified yet, call the grader's public verification page through its cert_lookup connector
 * (convention: meta.capabilities includes "cert_lookup", meta.config.grader = grader slug, and the
 * connector module exports `certUrl(cert)`). Results are stored on the certificate row; the cert
 * connector's own normalised output (catalog_item / population_report) is NOT written here — that
 * happens through the regular crawl pipeline when the connector is scheduled.
 */
export async function verifyCertificates(opts: { limit?: number; grader?: string; recheckDays?: number } = {}): Promise<{ checked: number; verified: number; notFound: number; failed: number; skippedGraders: string[] }> {
  const log = logger.child({ component: 'certs-verify' });
  const limit = opts.limit ?? 100;
  const recheckDays = opts.recheckDays ?? 180;
  const lookups = new Map<string, { connector: RareIndexConnector; certUrl: (cert: string, grade?: string | null) => string }>();
  const skippedGraders: string[] = [];
  for (const meta of listConnectorMeta({ enabled: true })) {
    if (!meta.capabilities.includes('cert_lookup') && !(meta.config as { grader?: string }).grader) continue;
    const grader = (meta.config as { grader?: string }).grader;
    if (!grader || (opts.grader && grader !== opts.grader)) continue;
    if (missingRequirements(meta).length) {
      skippedGraders.push(grader);
      continue;
    }
    try {
      const connector = await loadConnector(meta.id);
      const mod = (await import(`${new URL('../connectors/', import.meta.url).pathname}${meta.module}/index.ts`)) as { certUrl?: (cert: string, grade?: string | null) => string };
      if (!connector.lookup || !mod.certUrl) {
        skippedGraders.push(grader);
        continue;
      }
      lookups.set(grader, { connector, certUrl: mod.certUrl });
    } catch (err) {
      log.warn({ err, connector: meta.id }, 'cert connector unavailable');
      skippedGraders.push(grader);
    }
  }
  const res = { checked: 0, verified: 0, notFound: 0, failed: 0, skippedGraders };
  if (!lookups.size) return res;
  const graders = [...lookups.keys()];
  const rows = await db()
    .select()
    .from(certificates)
    .where(and(sql`${certificates.grader} = any(${sql.raw(`'{${graders.join(',')}}'::text[]`)})`, or(isNull(certificates.verifiedAt), lt(certificates.verifiedAt, new Date(Date.now() - recheckDays * 86_400_000)))))
    .orderBy(sql`${certificates.verifiedAt} nulls first`, sql`${certificates.sightings} desc`)
    .limit(limit);
  for (const row of rows) {
    const l = lookups.get(row.grader)!;
    const ctx = createCrawlContext({ router: getRouter(l.connector.meta.id), meta: l.connector.meta, options: { mode: 'probe', limit: 1 } });
    res.checked++;
    try {
      // NGC/PMG encode the holder grade in the URL ("cert/grade"); other graders ignore the suffix.
      const certArg = (row.grader === 'ngc' || row.grader === 'pmg') && row.grade ? `${row.certNumber}/${row.grade.replace(/\s+/g, '')}` : row.certNumber;
      const raws = await l.connector.lookup!(l.certUrl(certArg, row.grade), ctx);
      const records: NormalizedRecord[] = [];
      for (const raw of raws) records.push(...(await l.connector.normalize({ ...raw, externalId: raw.externalId ?? null, fetchedAt: raw.fetchedAt ?? new Date() })));
      const item = records.find((r) => r.kind === 'catalog_item');
      const pop = records.find((r) => r.kind === 'population_report');
      const found = Boolean(item);
      const verification = found
        ? { status: 'verified', title: item && 'rawTitle' in item ? item.rawTitle : null, grade: item && 'grade' in item ? item.grade : null, attributes: item?.attributes ?? null, population: pop && pop.kind === 'population_report' ? { total: pop.total, byGrade: pop.byGrade, reportDate: pop.reportDate } : null, checkedAt: new Date().toISOString(), connector: l.connector.meta.id }
        : (row.grader === 'ngc' || row.grader === 'pmg') && !row.grade
          ? { status: 'needs_grade', note: 'NGC/PMG verification URLs require the holder grade; none was captured with this cert', checkedAt: new Date().toISOString(), connector: l.connector.meta.id }
          : { status: 'not_found', checkedAt: new Date().toISOString(), connector: l.connector.meta.id, anomalies: ctx.anomalies };
      await db()
        .update(certificates)
        .set({
          verifiedAt: new Date(),
          verification,
          grade: found && item && 'grade' in item && item.grade.grade ? sql`coalesce(${certificates.grade}, ${item.grade.grade})` : sql`${certificates.grade}`,
          qualifier: found && item && 'grade' in item && item.grade.qualifier ? sql`coalesce(${certificates.qualifier}, ${item.grade.qualifier})` : sql`${certificates.qualifier}`,
        })
        .where(eq(certificates.id, row.id));
      if (found) res.verified++;
      else res.notFound++;
    } catch (err) {
      res.failed++;
      log.warn({ err, cert: row.certNumber, grader: row.grader }, 'cert verification failed');
      await db().update(certificates).set({ verification: { status: 'error', error: err instanceof Error ? err.message : String(err), checkedAt: new Date().toISOString() } }).where(eq(certificates.id, row.id));
    }
  }
  await flushCosts();
  log.info(res, 'certificate verification done');
  return res;
}

/**
 * One-off/idempotent backfill: register certificates + sightings from sales/listings/auction lots that already
 * carry a certification number (rows ingested before the certificates table existed). Safe to re-run.
 */
export async function backfillCertificates(): Promise<{ certificates: number; sightings: number }> {
  const norm = (col: string) => sql.raw(`upper(regexp_replace(${col}, '[^A-Za-z0-9]', '', 'g'))`);
  const where = (tbl: string) => sql.raw(`${tbl}.certification_number is not null and ${tbl}.grader is not null and length(regexp_replace(${tbl}.certification_number, '[^A-Za-z0-9]', '', 'g')) between 5 and 20`);
  const inserted = (await db().execute(sql`
    with src as (
      select lower(s.grader) as grader, ${norm('s.certification_number')} as cert, s.asset_id, s.variant_id, s.grade, s.source_id, s.source_url, s.price_usd, s.sale_date as observed_at
      from sales s where ${where('s')}
      union all
      select lower(l.grader), ${norm('l.certification_number')}, l.asset_id, l.variant_id, l.grade, l.source_id, l.source_url, l.price_usd, l.last_seen_at
      from listings l where ${where('l')}
    ), agg as (
      select grader, cert, min(asset_id) as asset_id, min(variant_id) as variant_id, min(grade) as grade, min(observed_at) as first_seen, max(observed_at) as last_seen, count(*)::int as n,
        array_agg(distinct source_id) as sources, (array_agg(source_url order by observed_at desc))[1] as last_url, (array_agg(price_usd order by observed_at desc))[1] as last_price
      from src group by grader, cert
    )
    insert into certificates (id, grader, cert_number, asset_id, variant_id, grade, first_seen_at, last_seen_at, sightings, source_ids, last_source_url, last_price_usd)
    select 'cert_' || substr(md5(grader || '|' || cert), 1, 20), grader, cert, asset_id, variant_id, grade, first_seen, last_seen, n, sources, last_url, last_price from agg
    on conflict (grader, cert_number) do update set sightings = greatest(certificates.sightings, excluded.sightings), first_seen_at = least(certificates.first_seen_at, excluded.first_seen_at), last_seen_at = greatest(certificates.last_seen_at, excluded.last_seen_at), asset_id = coalesce(certificates.asset_id, excluded.asset_id)
    returning id`)) as unknown as unknown[];
  const sightings = (await db().execute(sql`
    insert into certificate_sightings (id, certificate_id, kind, target_id, source_id, connector_id, source_url, price_usd, currency, price, observed_at)
    select 'evt_' || substr(md5('sale' || s.id), 1, 20), c.id, 'sale', s.id, s.source_id, s.connector_id, s.source_url, s.price_usd, s.currency, s.price, s.sale_date
    from sales s join certificates c on c.grader = lower(s.grader) and c.cert_number = ${norm('s.certification_number')} where ${where('s')}
    union all
    select 'evt_' || substr(md5('listing' || l.id), 1, 20), c.id, 'listing', l.id, l.source_id, l.connector_id, l.source_url, l.price_usd, l.currency, l.price, l.last_seen_at
    from listings l join certificates c on c.grader = lower(l.grader) and c.cert_number = ${norm('l.certification_number')} where ${where('l')}
    on conflict do nothing returning id`)) as unknown as unknown[];
  return { certificates: inserted.length, sightings: sightings.length };
}