/**
 * ANAF webservicesp v9 enrichment for firms.entities.
 *
 * Endpoint: https://webservicesp.anaf.ro/api/PlatitorTvaRest/v9/tva
 * Limits:   100 CUI/request, 1 request/sec (official, see doc_WS_V9.txt).
 * Auth:     none (public).
 *
 * Run from services/seap-scraper/ via:
 *   DATABASE_URL=... npx tsx src/enrich-anaf.ts [--limit=N] [--retries=N]
 *
 * Idempotent: pulls only entities where anaf_fetched_at IS NULL or older than
 * 30 days, ordered by SEAP activity (CUIs that appear in seap.announcements
 * win priority — prefetch what users actually look up).
 */
import { Pool } from 'pg';

const ANAF_URL = 'https://webservicesp.anaf.ro/api/PlatitorTvaRest/v9/tva';
const BATCH_SIZE = 100;
const SLEEP_MS = 1100; // 1.1s between requests — small buffer over 1s limit
const USER_AGENT = 'gov-agreg/1.0 vreaudigital.ro (contact: m.tarau@beletage.ro)';

const args = process.argv.slice(2);
const argLimit = parseInt(args.find((a) => a.startsWith('--limit='))?.split('=')[1] || '0');
const argRetries = parseInt(args.find((a) => a.startsWith('--retries='))?.split('=')[1] || '3');
const argConcurrency = Math.max(1, parseInt(args.find((a) => a.startsWith('--concurrency='))?.split('=')[1] || '1'));
// --tier=daily  → SEAP-active (14d) + new (always). Small daily volume.
// --tier=full   → SEAP-active (14d) + everyone else (90d) + new. Spread refresh.
// --tier=bulk   → everything stale per the legacy 30d window. One-shot.
const argTier = args.find((a) => a.startsWith('--tier='))?.split('=')[1] || 'bulk';

const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: Math.max(4, argConcurrency * 2),
});

interface AnafResp {
  cod: number;
  message?: string;
  found: AnafEntry[];
  notFound: number[];
}

interface AnafEntry {
  date_generale: {
    cui: number;
    denumire?: string;
    adresa?: string;
    nrRegCom?: string;
    telefon?: string;
    fax?: string;
    codPostal?: string;
    act?: string;
    stare_inregistrare?: string;
    data_inregistrare?: string;
    cod_CAEN?: string;
    iban?: string;
    statusRO_e_Factura?: boolean;
    organFiscalCompetent?: string;
    forma_de_proprietate?: string;
    forma_organizare?: string;
    forma_juridica?: string;
  };
  inregistrare_scop_Tva?: {
    scpTVA?: boolean;
    perioade_TVA?: any[];
  };
  stare_inactiv?: {
    statusInactivi?: boolean;
    dataInactivare?: string;
    dataReactivare?: string;
    dataRadiere?: string;
  };
  adresa_domiciliu_fiscal?: {
    ddenumire_Strada?: string;
    dnumar_Strada?: string;
    ddenumire_Localitate?: string;
    dcod_Localitate?: string;
    ddenumire_Judet?: string;
    dcod_Judet?: string;
    dcod_JudetAuto?: string;
    dtara?: string;
    ddetalii_Adresa?: string;
    dcod_Postal?: string;
  };
  adresa_sediu_social?: {
    sdenumire_Strada?: string;
    snumar_Strada?: string;
    sdenumire_Localitate?: string;
    scod_Localitate?: string;
    sdenumire_Judet?: string;
    sdetalii_Adresa?: string;
    scod_Postal?: string;
  };
}

async function sleep(ms: number) {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

async function fetchAnaf(cuis: number[], retries = argRetries): Promise<AnafResp> {
  const today = new Date().toISOString().slice(0, 10);
  const body = cuis.map((cui) => ({ cui, data: today }));

  for (let attempt = 0; attempt <= retries; attempt++) {
    try {
      const res = await fetch(ANAF_URL, {
        method: 'POST',
        headers: {
          'Content-Type': 'application/json',
          'User-Agent': USER_AGENT,
        },
        body: JSON.stringify(body),
      });
      // ANAF v9 returns HTTP 404 (with a regular JSON body containing
      // `found:[], notFound:[...]`) when *every* CUI in the batch is invalid
      // / unknown. We still want to mark those CUIs as fetched so they don't
      // come back as candidates next run, so treat 404+parseable-body as a
      // valid "no results" response.
      if (res.status === 404) {
        try {
          const txt = await res.text();
          const parsed = JSON.parse(txt) as AnafResp;
          if (parsed && Array.isArray(parsed.notFound)) return parsed;
          throw new Error(`ANAF HTTP 404 with unparseable body: ${txt.slice(0, 200)}`);
        } catch (parseErr) {
          throw new Error(`ANAF HTTP 404 (parse failed): ${(parseErr as Error).message}`);
        }
      }
      if (!res.ok) {
        if (res.status === 429 && attempt < retries) {
          const wait = 5000 + attempt * 5000;
          console.warn(`[ANAF] 429 rate-limited, sleeping ${wait}ms (attempt ${attempt + 1}/${retries})`);
          await sleep(wait);
          continue;
        }
        throw new Error(`ANAF HTTP ${res.status}: ${(await res.text()).slice(0, 200)}`);
      }
      return (await res.json()) as AnafResp;
    } catch (err) {
      if (attempt < retries) {
        await sleep(2000);
        continue;
      }
      throw err;
    }
  }
  throw new Error('unreachable');
}

function buildAddressFull(d?: AnafEntry['adresa_domiciliu_fiscal']): string | null {
  if (!d) return null;
  const parts = [
    [d.ddenumire_Strada, d.dnumar_Strada].filter(Boolean).join(' '),
    d.ddetalii_Adresa,
    d.ddenumire_Localitate,
    d.ddenumire_Judet,
    d.dtara || 'Romania',
  ].filter((s) => s && (s as string).trim().length > 0);
  return parts.length > 0 ? parts.join(', ') : null;
}

async function updateBatch(entries: AnafEntry[], notFoundCuis: number[]) {
  const client = await pool.connect();
  try {
    await client.query('BEGIN');

    if (entries.length > 0) {
      // Build parallel arrays for one bulk UPDATE…FROM unnest(...) — replaces the
      // previous loop of N individual UPDATEs (~5-10ms RTT each on local PG).
      const cuis: string[] = [];
      const adrStrada: (string | null)[] = [];
      const adrNumar: (string | null)[] = [];
      const adrLocalitate: (string | null)[] = [];
      const adrJudet: (string | null)[] = [];
      const adrCodPostal: (string | null)[] = [];
      const adrFull: (string | null)[] = [];
      const phone: (string | null)[] = [];
      const fax: (string | null)[] = [];
      const caenPrincipal: (string | null)[] = [];
      const isActiveAnaf: boolean[] = [];
      const isVatRegistered: boolean[] = [];
      const isEfactura: boolean[] = [];
      const formaJuridica: (string | null)[] = [];

      for (const entry of entries) {
        const dg = entry.date_generale;
        const fiscal = entry.adresa_domiciliu_fiscal;
        const sediu = entry.adresa_sediu_social;
        const inact = entry.stare_inactiv;
        const tva = entry.inregistrare_scop_Tva;

        const computedFull = buildAddressFull(fiscal) ||
          [sediu?.sdenumire_Strada, sediu?.snumar_Strada, sediu?.sdenumire_Localitate, sediu?.sdenumire_Judet, 'Romania']
            .filter(Boolean).join(', ');

        cuis.push(String(dg.cui));
        adrStrada.push(fiscal?.ddenumire_Strada || sediu?.sdenumire_Strada || null);
        adrNumar.push(fiscal?.dnumar_Strada || sediu?.snumar_Strada || null);
        adrLocalitate.push(fiscal?.ddenumire_Localitate || sediu?.sdenumire_Localitate || null);
        adrJudet.push(fiscal?.ddenumire_Judet || sediu?.sdenumire_Judet || null);
        adrCodPostal.push(fiscal?.dcod_Postal || sediu?.scod_Postal || null);
        adrFull.push(computedFull || null);
        phone.push(dg.telefon || null);
        fax.push(dg.fax || null);
        caenPrincipal.push(dg.cod_CAEN || null);
        isActiveAnaf.push(inact?.statusInactivi === true ? false : true);
        isVatRegistered.push(tva?.scpTVA === true);
        isEfactura.push(dg.statusRO_e_Factura === true);
        formaJuridica.push(dg.forma_juridica || null);
      }

      await client.query(`
        UPDATE firms.entities e
        SET
          adr_strada        = COALESCE(NULLIF(d.adr_strada, ''), e.adr_strada),
          adr_numar         = COALESCE(NULLIF(d.adr_numar, ''), e.adr_numar),
          adr_localitate    = COALESCE(NULLIF(d.adr_localitate, ''), e.adr_localitate),
          adr_judet         = COALESCE(NULLIF(d.adr_judet, ''), e.adr_judet),
          adr_cod_postal    = COALESCE(NULLIF(d.adr_cod_postal, ''), e.adr_cod_postal),
          adr_full          = COALESCE(NULLIF(d.adr_full, ''), e.adr_full),
          phone             = COALESCE(NULLIF(d.phone, ''), e.phone),
          fax               = COALESCE(NULLIF(d.fax, ''), e.fax),
          caen_principal    = COALESCE(NULLIF(d.caen_principal, ''), e.caen_principal),
          is_active_anaf    = d.is_active_anaf,
          is_vat_registered = d.is_vat_registered,
          is_efactura       = d.is_efactura,
          forma_juridica    = COALESCE(NULLIF(d.forma_juridica, ''), e.forma_juridica),
          anaf_fetched_at   = now(),
          updated_at        = now()
        FROM (
          SELECT * FROM unnest(
            $1::text[], $2::text[], $3::text[], $4::text[], $5::text[],
            $6::text[], $7::text[], $8::text[], $9::text[], $10::text[],
            $11::boolean[], $12::boolean[], $13::boolean[], $14::text[]
          ) AS u(cui, adr_strada, adr_numar, adr_localitate, adr_judet,
                 adr_cod_postal, adr_full, phone, fax, caen_principal,
                 is_active_anaf, is_vat_registered, is_efactura, forma_juridica)
        ) d
        WHERE e.cui = d.cui
      `, [
        cuis, adrStrada, adrNumar, adrLocalitate, adrJudet,
        adrCodPostal, adrFull, phone, fax, caenPrincipal,
        isActiveAnaf, isVatRegistered, isEfactura, formaJuridica,
      ]);
    }

    // Mark notFound CUIs to avoid re-fetching for a while
    if (notFoundCuis.length > 0) {
      await client.query(`
        UPDATE firms.entities
        SET anaf_fetched_at = now(),
            is_active_anaf = COALESCE(is_active_anaf, false),
            updated_at = now()
        WHERE cui = ANY($1::text[])
      `, [notFoundCuis.map(String)]);
    }

    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK').catch(() => {});
    throw err;
  } finally {
    client.release();
  }
}

async function main() {
  const startTs = Date.now();
  let processed = 0;
  let errors = 0;

  // Pull the entire candidate list once, prioritized by SEAP activity.
  // 4M CUI strings × ~10 bytes ≈ 40MB — fits comfortably in memory and avoids
  // repeating the expensive entities×announcements join on every iteration.
  // Tiered freshness predicates — selectable via --tier=...
  // - daily: only SEAP-active firms older than 14 days, plus brand-new firms.
  //   ~5K/day under steady state. ~5min wall time at concurrency=2.
  // - full:  SEAP-active >14d + everyone else >90d + new. Spreads ~45K/day.
  //   ~5-7min wall time, recommended for daily cron after the initial bulk.
  // - bulk:  legacy 30d window. Use once for the initial backfill.
  const tierPredicate = (() => {
    switch (argTier) {
      case 'daily':
        return `(
          e.anaf_fetched_at IS NULL
          OR (
            e.anaf_fetched_at < NOW() - INTERVAL '14 days'
            AND EXISTS (SELECT 1 FROM seap.announcements aa
                        WHERE aa.supplier_cui = e.cui OR aa.authority_cui = e.cui)
          )
        )`;
      case 'full':
        return `(
          e.anaf_fetched_at IS NULL
          OR e.anaf_fetched_at < NOW() - INTERVAL '90 days'
          OR (
            e.anaf_fetched_at < NOW() - INTERVAL '14 days'
            AND EXISTS (SELECT 1 FROM seap.announcements aa
                        WHERE aa.supplier_cui = e.cui OR aa.authority_cui = e.cui)
          )
        )`;
      case 'bulk':
      default:
        return `(e.anaf_fetched_at IS NULL OR e.anaf_fetched_at < NOW() - INTERVAL '30 days')`;
    }
  })();

  console.log(`[ANAF] Tier=${argTier}, loading candidate CUIs (entities × announcements join)...`);
  const cuiListQ = await pool.query<{ cui: string }>(`
    SELECT e.cui FROM firms.entities e
    LEFT JOIN seap.announcements a ON a.supplier_cui = e.cui OR a.authority_cui = e.cui
    WHERE ${tierPredicate}
      AND e.cui ~ '^[0-9]+$'
      -- Romanian CIF/CUI is at most 10 digits. ONRC bulk import sometimes
      -- stored a 13-digit CNP in the cui field for PFAs; ANAF rejects those
      -- (HTTP 404 for the whole batch), so filter them out at the source.
      AND length(e.cui) BETWEEN 1 AND 10
    GROUP BY e.cui
    ORDER BY COUNT(a.id) DESC NULLS LAST, e.cui
  `);
  const allCuis: number[] = cuiListQ.rows
    .map((r) => parseInt(r.cui))
    .filter((n) => Number.isFinite(n));
  const total = argLimit > 0 ? Math.min(argLimit, allCuis.length) : allCuis.length;
  console.log(`[ANAF] Total CUIs to enrich: ${total} (limit: ${argLimit > 0 ? argLimit : 'none'})`);
  console.log(`[ANAF] Concurrency: ${argConcurrency} worker(s), per-worker rate ${BATCH_SIZE} CUI/${(SLEEP_MS / 1000).toFixed(1)}s`);
  console.log(`[ANAF] Best-case throughput: ${Math.round((argConcurrency * BATCH_SIZE * 1000) / SLEEP_MS)} cui/s`);

  // Shared work queue: pre-build all batches up front, workers consume from head.
  const queue: number[][] = [];
  for (let i = 0; i < total; i += BATCH_SIZE) {
    queue.push(allCuis.slice(i, Math.min(i + BATCH_SIZE, total)));
  }

  let totalProcessed = 0;
  let totalRateLimited = 0;

  async function worker(id: number) {
    while (queue.length > 0) {
      const cuis = queue.shift();
      if (!cuis || cuis.length === 0) break;

      const tStart = Date.now();
      try {
        const resp = await fetchAnaf(cuis);
        await updateBatch(resp.found || [], resp.notFound || []);
        processed += cuis.length;
        totalProcessed += cuis.length;
      } catch (err: any) {
        errors++;
        // Match the explicit "ANAF HTTP 429" prefix from fetchAnaf, not any
        // substring "429" — CUI lists in 404 bodies frequently contain that
        // digit run by chance and would falsely inflate the count.
        if (err.message?.startsWith('ANAF HTTP 429')) totalRateLimited++;
        console.error(`[ANAF w${id}] Batch error: ${err.message}. Continuing.`);
      }

      if (totalProcessed % 1000 < BATCH_SIZE || totalProcessed >= total) {
        const elapsed = (Date.now() - startTs) / 1000;
        const rate = totalProcessed / elapsed;
        const eta = (total - totalProcessed) / Math.max(rate, 1);
        console.log(`[ANAF] ${totalProcessed}/${total} (${(totalProcessed / total * 100).toFixed(1)}%) · rate ${rate.toFixed(1)} cui/s · ETA ${(eta / 60).toFixed(0)}min · errors ${errors}${totalRateLimited > 0 ? ` (${totalRateLimited} 429s)` : ''}`);
      }

      // Per-worker rate limit: sleep so this worker fires at most one request
      // every SLEEP_MS. Time spent in fetch+update counts toward the gap, so
      // the effective inter-request gap stays at SLEEP_MS regardless of work
      // duration. With N workers, total request rate ≈ N / SLEEP_MS req/sec.
      const elapsed = Date.now() - tStart;
      const remaining = SLEEP_MS - elapsed;
      if (remaining > 0) await sleep(remaining);
    }
  }

  const workers: Promise<void>[] = [];
  for (let i = 0; i < argConcurrency; i++) {
    // Stagger worker starts so requests are evenly distributed in time:
    // worker 0 fires at t=0, worker 1 at t=SLEEP_MS/N, etc. Without this,
    // all workers would burst together every SLEEP_MS.
    const stagger = (i * SLEEP_MS) / argConcurrency;
    workers.push(sleep(stagger).then(() => worker(i + 1)));
  }
  await Promise.all(workers);

  const elapsed = (Date.now() - startTs) / 1000;
  console.log(`[ANAF] DONE. ${processed} processed in ${(elapsed / 60).toFixed(1)} min. ${errors} batch errors${totalRateLimited > 0 ? ` (${totalRateLimited} 429-rate-limited)` : ''}.`);
  await pool.end();
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});
