import { createReadStream } from 'node:fs';
import { createHash, randomBytes } from 'node:crypto';
import { createGunzip } from 'node:zlib';
import { foldFoodText } from '../../../core/food-search-name.ts';
import type pg from 'pg';
import { normalizeBarcode } from '../barcode.ts';
import { isExcluded, offGroup, storedGroup, UNMAPPED_REASON } from '../groups.ts';
import type { FoodGroup } from '../groups.ts';
import { assignedGroup } from './assigned-groups.ts';

// SRC-06 Open Food Facts, branded products with a barcode for every country the export lists (spec 20).
// Owner decision 2026-10-13: Calsa ships everywhere, so each product becomes a food for each market in
// its countries_tags (market = ISO 3166-1 alpha-2). The country scope is a parameter, not a list.
// Input is the official daily JSONL export (one JSON object per line, optionally gzipped).
// The file is untrusted data: it is streamed line by line, never loaded whole, and nothing in it
// is executed. Imports only create DRAFT releases. Raw lines are stored exactly as delivered in
// food_source_record (FD-57); everything we add or decide lives in food_version and food_added_off.

export type Market = string;
export type Scope = ReadonlySet<Market> | null;
export const OFF_ATTRIBUTION = '© Open Food Facts contributors; database under ODbL 1.0, contents under DbCL 1.0';

// OFF country tags are English taxonomy names ("en:united-kingdom"). The ISO code comes from the
// ICU region names; the aliases are the names where OFF and ICU differ. Other tags ("en:world",
// regions, free text typed into the field) map to nothing and are ignored.
const slug = (name: string) => name.normalize('NFKD').replace(/\p{M}/gu, '').toLowerCase()
  .replace(/[^a-z0-9]+/g, '-').replace(/^-|-$/g, '');
const NOT_COUNTRIES = new Set(['EU', 'UN', 'ZZ', 'EZ', 'QO', 'XA', 'XB', 'AC', 'CP', 'DG', 'EA', 'IC', 'TA',
  // Retired or exceptional codes that ICU still names; the live code of the same country is kept.
  'UK', 'FX', 'SU', 'YU', 'ZR', 'TP', 'BU', 'DD', 'NT', 'CS', 'AN', 'CQ', 'DY', 'HV', 'NH', 'RH', 'VD', 'YD']);
const ALIASES: Record<string, Market> = {
  'czech-republic': 'CZ', turkey: 'TR', 'hong-kong': 'HK', 'bosnia-and-herzegovina': 'BA',
  'virgin-islands-of-the-united-states': 'VI', 'trinidad-and-tobago': 'TT', myanmar: 'MM',
  'democratic-republic-of-the-congo': 'CD', 'republic-of-the-congo': 'CG', 'the-bahamas': 'BS',
  macau: 'MO', 'saint-pierre-and-miquelon': 'PM', 'republic-of-macedonia': 'MK', 'saint-lucia': 'LC',
  'saint-martin': 'MF', 'antigua-and-barbuda': 'AG', 'state-of-palestine': 'PS', 'wallis-and-futuna': 'WF',
  'saint-vincent-and-the-grenadines': 'VC', 'saint-kitts-and-nevis': 'KN', 'st-kitts-and-nevis': 'KN',
  'st-vincent-and-grenadines': 'VC', 'turks-and-caicos-islands': 'TC', swaziland: 'SZ',
  'united-states-minor-outlying-islands': 'UM', 'svalbard-and-jan-mayen': 'SJ', 'saint-barthelemy': 'BL',
  'sao-tome-and-principe': 'ST', 'french-southern-and-antarctic-lands': 'TF',
};
const TAG_TO_MARKET = new Map<string, Market>();
const names = new Intl.DisplayNames('en', { type: 'region' });
for (let a = 65; a <= 90; a++) for (let b = 65; b <= 90; b++) {
  const code = String.fromCharCode(a, b);
  const name = NOT_COUNTRIES.has(code) ? code : names.of(code);
  if (name && name !== code && !TAG_TO_MARKET.has(slug(name))) TAG_TO_MARKET.set(slug(name), code);
}
for (const [name, code] of Object.entries(ALIASES)) TAG_TO_MARKET.set(name, code);
export const OFF_COUNTRY_CODES: readonly Market[] = [...new Set(TAG_TO_MARKET.values())].sort();
const marketOfTag = (tag: string): Market | undefined =>
  tag.startsWith('en:') ? TAG_TO_MARKET.get(slug(tag.slice(3))) : undefined;

// "US,GB" -> a scope; "all" (or nothing) -> null, every country. An unknown code is an error, not a
// silently empty import.
export function parseScope(list: string | undefined): Scope {
  if (list === undefined || list.trim() === '' || list.trim().toLowerCase() === 'all') return null;
  const codes = list.split(',').map(c => c.trim().toUpperCase());
  const unknown = codes.filter(c => !OFF_COUNTRY_CODES.includes(c));
  if (unknown.length) throw new Error(`Unknown country code: ${unknown.join(', ')}`);
  return new Set(codes);
}
// Only for a restricted scope: the English tags whose bytes decide whether a line is worth parsing.
const scopeTags = (scope: ReadonlySet<Market>) => [...TAG_TO_MARKET].filter(([, code]) => scope.has(code))
  .map(([name]) => `en:${name}`);
// Markets whose labels state total carbohydrate (fibre included); everywhere else carbohydrate is
// "available" (fibre apart), as in the EU, GB and AU rules.
const TOTAL_CARB: ReadonlySet<Market> = new Set(['US', 'CA']);
const MAX_LINE_BYTES = 8_000_000;
const BATCH = 500;

const GRAMS: Record<string, number> = { g: 1, mg: 1e-3, ug: 1e-6 };
// column -> [OFF nutrient keys in order of preference, column unit]
const COLUMNS = {
  enerc_kcal: [['energy-kcal'], 'kcal'], enerc_kj: [['energy-kj', 'energy'], 'kj'],
  procnt_g: [['proteins'], 'g'], fat_g: [['fat'], 'g'], chocdf_g: [['carbohydrates'], 'g'],
  fibtg_g: [['fiber'], 'g'], sugar_g: [['sugars'], 'g'], sugad_g: [['added-sugars'], 'g'],
  fasat_g: [['saturated-fat'], 'g'], na_mg: [['sodium'], 'mg'], chole_mg: [['cholesterol'], 'mg'],
  k_mg: [['potassium'], 'mg'], ca_mg: [['calcium'], 'mg'], fe_mg: [['iron'], 'mg'],
  vitd_ug: [['vitamin-d'], 'ug'], caffn_mg: [['caffeine'], 'mg'],
} as const;
export type Column = keyof typeof COLUMNS;
export const OFF_COLUMNS = Object.keys(COLUMNS) as Column[];
type Values = Record<Column, number | null>;
type Evidence = { type: string; off_source: string };

type Entry = { value?: unknown; unit?: unknown; source?: unknown; source_per?: unknown; modifier?: unknown };
type OffRecord = {
  code?: unknown; product_type?: unknown; countries_tags?: unknown; states_tags?: unknown;
  data_quality_errors_tags?: unknown; product_name?: unknown; product_name_en?: unknown; lang?: unknown;
  brands?: unknown; categories_tags?: unknown; allergens_tags?: unknown; quantity?: unknown;
  serving_size?: unknown; images?: { selected?: { nutrition?: unknown } };
  nutrition?: { aggregated_set?: { per?: unknown; preparation?: unknown; nutrients?: Record<string, Entry> } };
  nutriments?: Record<string, unknown>; nutrition_data_per?: unknown;
};
type NutritionSet = { per?: unknown; preparation?: unknown; nutrients?: Record<string, Entry> };

export type MarketPlan = {
  market: Market; status: 'active' | 'quarantine'; position: 'confirmed' | 'estimate';
  reasons: string[]; carbohydrate_definition: 'total' | 'available';
};
export type OffCandidate = {
  raw: Buffer; code: string; gtin: string; name: string; searchName: string; basis: '100g' | '100ml';
  values: Values; evidence: Partial<Record<Column, Evidence>>; markets: MarketPlan[];
  added: Record<string, unknown>; group: FoodGroup | null;
};
type Evaluation =
  | { ok: false; reason: string; markets: Market[]; valueRejects: string[] }
  | { ok: true; candidate: OffCandidate; mismatched: Market[]; valueRejects: string[] };

const round4 = (x: number) => Math.round(x * 1e4) / 1e4;
const stringList = (v: unknown, max: number) => Array.isArray(v)
  ? v.filter((x): x is string => typeof x === 'string' && x.length <= 120).slice(0, max) : [];
const text = (v: unknown, max: number) => {
  if (typeof v !== 'string') return null;
  const t = v.replace(/\s+/g, ' ').trim();
  return t && t.length <= max ? t : null;
};
const unitFactor = (column: Column, unit: unknown) => {
  const target = COLUMNS[column][1];
  const u = typeof unit === 'string' ? unit.toLowerCase().replace(/[µμ]/g, 'u') : '';
  if (target === 'kcal') return u === 'kcal' ? 1 : null;
  if (target === 'kj') return u === 'kj' ? 1 : null;
  return u === 'mcg' ? GRAMS.ug! / GRAMS[target]! : GRAMS[u] !== undefined ? GRAMS[u]! / GRAMS[target]! : null;
};

// FD-07/FD-40: OFF nutrient source "packaging"/"manufacturer" is a label value, "USDA" is borrowed,
// "estimate" and anything unknown never enters. Values shown per serving are label_per_serving (FD-06).
function valueType(source: unknown, per: unknown): { type: string } | { reject: string } {
  if (typeof source !== 'string') return { reject: 'source_missing' };
  const s = source.toLowerCase();
  if (s === 'packaging' || s === 'manufacturer') return { type: per === 'serving' ? 'label_per_serving' : 'label' };
  if (s === 'usda') return { type: 'borrowed' };
  // A record that Open Food Facts never converted to its source-tagged table: its contributors typed the values
  // from the pack (the table is marked completed, FD-40); without a nutrition photo the position stays "estimate".
  if (s === 'legacy') return { type: per === 'serving' ? 'label_per_serving' : 'label' };
  return { reject: /^[a-z_-]{1,20}$/.test(s) ? `source_${s}` : 'source_other' };
}

// The old Open Food Facts layout: `nutriments` holds per-100 g values in grams (energy in kcal and kJ), with no
// source per nutrient. The basis is 100 ml for a product sold by volume.
const LEGACY_UNIT: Record<string, string> = { 'energy-kcal': 'kcal', 'energy-kj': 'kj', energy: 'kj' };
const SOLD_BY_VOLUME = /\b\d+(?:[.,]\d+)?\s*(?:ml|cl|dl|l|fl\.?\s*oz)\b/i;
function legacySet(rec: OffRecord): NutritionSet | null {
  const nutriments = rec.nutriments;
  if (!nutriments || typeof nutriments !== 'object') return null;
  const per = rec.nutrition_data_per === 'serving' ? 'serving' : '100g';
  const nutrients: Record<string, Entry> = {};
  for (const column of OFF_COLUMNS) for (const key of COLUMNS[column][0]) {
    const value = nutriments[`${key}_100g`];
    if (value === undefined) continue;
    const modifier = nutriments[`${key}_modifier`];
    nutrients[key] = { value: typeof value === 'string' ? Number(value) : value, unit: LEGACY_UNIT[key] ?? 'g',
      source: 'legacy', source_per: per, ...modifier ? { modifier } : {} };
  }
  const quantity = typeof rec.quantity === 'string' ? rec.quantity : '';
  return { per: SOLD_BY_VOLUME.test(quantity) ? '100ml' : '100g', preparation: 'as_sold', nutrients };
}

function extract(nutrients: Record<string, Entry>) {
  const values = Object.fromEntries(OFF_COLUMNS.map(c => [c, null])) as Values;
  const evidence: Partial<Record<Column, Evidence>> = {};
  const rejects: string[] = [];
  const perOf: Partial<Record<Column, unknown>> = {};
  for (const column of OFF_COLUMNS) {
    for (const key of COLUMNS[column][0]) {
      const entry = nutrients[key];
      if (!entry || typeof entry !== 'object') continue;
      const verdict = valueType(entry.source, entry.source_per);
      if ('reject' in verdict) { rejects.push(verdict.reject); continue; }
      if (entry.modifier !== undefined && entry.modifier !== null && entry.modifier !== '') {
        rejects.push('modifier'); continue;
      }
      const factor = unitFactor(column, entry.unit);
      if (factor === null) { rejects.push('unit'); continue; }
      if (typeof entry.value !== 'number' || !Number.isFinite(entry.value) || entry.value < 0) {
        rejects.push('value'); continue;
      }
      values[column] = round4(entry.value * factor);
      evidence[column] = { type: verdict.type, off_source: String(entry.source).toLowerCase() };
      perOf[column] = entry.source_per;
      break;
    }
  }
  return { values, evidence, rejects, perOf };
}

export function implausible(v: Values, market: Market) {
  const p = v.procnt_g!, f = v.fat_g!, c = v.chocdf_g!;
  const available = !TOTAL_CARB.has(market);
  const fibre = v.fibtg_g ?? 0;
  if (p + f + c + (available ? fibre : 0) > 102) return true;
  if (v.sugar_g !== null && v.sugar_g > c + 0.5) return true;
  if (!available && v.fibtg_g !== null && v.fibtg_g > c) return true;
  const au = market === 'AU';
  const energy = au ? v.enerc_kj! : v.enerc_kcal!;
  const expected = au ? 17 * p + 37 * f + 17 * c + 8 * fibre
    : 4 * p + 9 * f + 4 * c + (available ? 2 * fibre : 0);
  const gap = Math.abs(energy - expected);
  return gap > expected * 0.2 && gap > (au ? 15 * 4.184 : 15);
}

export function evaluateOffLine(raw: Buffer, scope: Scope = null): Evaluation {
  const fail = (reason: string, markets: Market[] = [], valueRejects: string[] = []): Evaluation =>
    ({ ok: false, reason, markets, valueRejects });
  let rec: OffRecord;
  try {
    const parsed: unknown = JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(raw));
    if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) return fail('malformed');
    rec = parsed as OffRecord;
  } catch { return fail('malformed'); }
  const listed = [...new Set(stringList(rec.countries_tags, 400).map(marketOfTag)
    .filter((m): m is Market => m !== undefined && (!scope || scope.has(m))))];
  if (!listed.length) return fail('out_of_scope');
  if (rec.product_type !== undefined && rec.product_type !== 'food') return fail('not_food', listed);

  const code = typeof rec.code === 'string' ? rec.code : '';
  const barcode = normalizeBarcode(code);
  if (barcode.kind !== 'valid') return fail(`barcode_${barcode.kind}`, listed);

  // FD-40: nutrition table marked completed, no quality errors, no estimate values.
  const states = stringList(rec.states_tags, 200);
  if (!states.includes('en:nutrition-facts-completed')) return fail('nutrition_not_completed', listed);
  if (!Array.isArray(rec.data_quality_errors_tags)) return fail('quality_unknown', listed);
  if (rec.data_quality_errors_tags.length) return fail('quality_error', listed);

  // C-23, C-26: baby food and supplements are not part of the base.
  const categories = stringList(rec.categories_tags, 40);
  const entry = offGroup(categories);
  if (isExcluded(entry)) return fail(`excluded_${entry}`, listed);
  const group = storedGroup(entry) ?? assignedGroup('SRC-06', code);

  // The English name when the record has one, else the name as delivered with its language; no translation.
  const nameEn = text(rec.product_name_en, 300);
  const name = nameEn ?? text(rec.product_name, 300);
  const nameLang = nameEn ? 'en' : typeof rec.lang === 'string' && /^[a-z]{2,3}(-[A-Za-z0-9]{2,8})?$/.test(rec.lang)
    ? rec.lang : null;
  const searchName = foldFoodText(name ?? '').replace(/\s+/g, ' ').trim();
  if (!name || !searchName) return fail('no_name', listed);

  const set: NutritionSet | null | undefined = rec.nutrition?.aggregated_set ?? legacySet(rec);
  if (!set || typeof set !== 'object' || !set.nutrients || typeof set.nutrients !== 'object')
    return fail('no_source_info', listed);
  if (set.preparation !== undefined && set.preparation !== 'as_sold') return fail('not_as_sold', listed);
  if (set.per !== '100g' && set.per !== '100ml') return fail('basis_not_100', listed);

  const { values, evidence, rejects, perOf } = extract(set.nutrients);
  const hasEnergy = values.enerc_kcal !== null || values.enerc_kj !== null;
  if (!hasEnergy || values.procnt_g === null || values.fat_g === null || values.chocdf_g === null)
    return fail(rejects.length ? 'required_missing_after_fd40' : 'required_missing', listed, rejects);
  // FD-12: a missing unit is derived by 4.184 and typed calculated.
  if (values.enerc_kj === null) {
    values.enerc_kj = round4(values.enerc_kcal! * 4.184);
    evidence.enerc_kj = { type: 'calculated', off_source: evidence.enerc_kcal!.off_source };
  } else if (values.enerc_kcal === null) {
    values.enerc_kcal = round4(values.enerc_kj / 4.184);
    evidence.enerc_kcal = { type: 'calculated', off_source: evidence.enerc_kj!.off_source };
  }
  // FD-13: optional nutrients outside their bounds become "no data" and the record stays.
  const drop = (c: Column) => { values[c] = null; delete evidence[c]; };
  if (values.sugad_g !== null && values.sugar_g !== null && values.sugad_g > values.sugar_g) drop('sugad_g');
  if (values.fasat_g !== null && values.fasat_g > values.fat_g) drop('fasat_g');
  if (values.na_mg !== null && values.na_mg > 39300) drop('na_mg');

  // FD-60: the label decides the market of the values.
  const energyPer = perOf.enerc_kcal ?? perOf.enerc_kj;
  const kjDeclared = evidence.enerc_kj?.type !== 'calculated';
  const labelMarkets: Market[] = energyPer === 'serving' ? ['US', 'CA']
    : energyPer !== undefined && kjDeclared ? ['GB', 'AU'] : [];
  const matched = listed.filter(m => labelMarkets.includes(m));
  const unknownLabel = matched.length === 0;
  const targets = unknownLabel ? listed : matched;
  const photo = states.includes('en:nutrition-photo-selected') || Boolean(rec.images?.selected?.nutrition);
  // FD-09, FD-43: a product without a group is quarantined and stays out of search.
  const markets: MarketPlan[] = targets.map(market => ({
    market,
    status: implausible(values, market) || group === null ? 'quarantine' : 'active',
    position: photo && !unknownLabel ? 'confirmed' : 'estimate',
    reasons: [...(photo ? [] : ['no_nutrition_photo']), ...(unknownLabel ? ['label_market_unknown'] : []),
      ...(group === null ? [UNMAPPED_REASON] : [])],
    carbohydrate_definition: TOTAL_CARB.has(market) ? 'total' : 'available',
  }));
  const sources = Object.fromEntries(OFF_COLUMNS.filter(c => evidence[c]).map(c => [c, evidence[c]!.off_source]));
  const added = {
    off_code: code, gtin: barcode.gtin, brand: text(rec.brands, 300), name_lang: nameLang,
    categories_tags: categories, allergens_tags: stringList(rec.allergens_tags, 20),
    quantity: text(rec.quantity, 60), serving_size: text(rec.serving_size, 60),
    nutrition_photo: photo, label_kind: unknownLabel ? null : labelMarkets[0] === 'US' ? 'per_serving' : 'per_100g_kj',
    listed_markets: listed, nutrient_sources: sources,
    markets: Object.fromEntries(markets.map(m => [m.market, { status: m.status, position: m.position, reasons: m.reasons }])),
  };
  return { ok: true, candidate: { raw, code, gtin: barcode.gtin, name, searchName, basis: set.per, values, evidence, markets, added, group },
    mismatched: unknownLabel ? [] : listed.filter(m => !matched.includes(m)), valueRejects: rejects };
}

// Splits a byte stream into lines without decoding; a line over the cap is skipped, not buffered.
export async function* offLines(input: AsyncIterable<Buffer | string>): AsyncGenerator<Buffer | null> {
  let pending: Buffer[] = [];
  let size = 0;
  let skipping = false;
  for await (const piece of input) {
    let chunk = typeof piece === 'string' ? Buffer.from(piece) : piece;
    for (;;) {
      const at = chunk.indexOf(10);
      if (at < 0) {
        if (!skipping) {
          pending.push(chunk); size += chunk.length;
          if (size > MAX_LINE_BYTES) { pending = []; size = 0; skipping = true; yield null; }
        }
        break;
      }
      if (!skipping) {
        const line = Buffer.concat([...pending, chunk.subarray(0, at)]);
        if (line.length > MAX_LINE_BYTES) yield null; else if (line.length) yield line;
      }
      pending = []; size = 0; skipping = false;
      chunk = chunk.subarray(at + 1);
    }
  }
  if (!skipping && size) yield Buffer.concat(pending);
}

export function openOffExport(path: string): AsyncIterable<Buffer> {
  const file = createReadStream(path);
  if (!path.endsWith('.gz')) return file;
  const gunzip = createGunzip();
  file.on('error', error => gunzip.destroy(error));
  return file.pipe(gunzip);
}

export type MarketOutcome = 'new' | 'unchanged' | 'duplicate_gtin' | 'db_error';
// A sink with `flush` takes candidates in batches: scanOff does not wait for each one, it settles them after a flush.
export type Sink = ((candidate: OffCandidate) => Promise<Record<Market, MarketOutcome>>) & { flush?: () => Promise<void> };
type Tally = { seen: number; accepted: number; active: number; quarantine: number; estimate: number;
  newVersions: number; rejected: Record<string, number> };
export type OffReport = {
  lines: number; malformed: number; oversize: number; outOfScope: number;
  markets: Record<Market, Tally>; valueRejects: Record<string, number>;
  newRaw: number; releaseId: string | null;
};
const bump = (m: Record<string, number>, k: string, n = 1) => { m[k] = (m[k] ?? 0) + n; };

// `legacyOnly` reads only the records that carry no source-tagged table: a catch-up pass after the old layout was added.
export async function scanOff(input: AsyncIterable<Buffer | string>, sink: Sink, limit = Infinity, scope: Scope = null,
  legacyOnly = false) {
  const report: OffReport = { lines: 0, malformed: 0, oversize: 0, outOfScope: 0,
    markets: {}, valueRejects: {}, newRaw: 0, releaseId: null };
  const tally = (market: Market): Tally => report.markets[market] ??=
    { seen: 0, accepted: 0, active: 0, quarantine: 0, estimate: 0, newVersions: 0, rejected: {} };
  const tags = scope ? scopeTags(scope) : null;
  const settleOne = (candidate: OffCandidate, outcomes: Record<Market, MarketOutcome>) => {
    for (const plan of candidate.markets) {
      const t = tally(plan.market);
      t.seen++;
      const outcome = outcomes[plan.market] ?? 'db_error';
      if (outcome === 'duplicate_gtin' || outcome === 'db_error') { bump(t.rejected, outcome); continue; }
      t.accepted++; t[plan.status]++;
      if (plan.position === 'estimate') t.estimate++;
      if (outcome === 'new') t.newVersions++;
    }
  };
  const inflight: { candidate: OffCandidate; outcomes: Promise<Record<Market, MarketOutcome>> }[] = [];
  const settle = async () => { for (const item of inflight.splice(0)) settleOne(item.candidate, await item.outcomes); };
  for await (const line of offLines(input)) {
    if (report.lines >= limit) break;
    report.lines++;
    if (line === null) { report.oversize++; continue; }
    // A restricted scope tests bytes first: lines sold elsewhere are never parsed.
    if (tags && !tags.some(tag => line.includes(tag))) { report.outOfScope++; continue; }
    if (legacyOnly && line.includes('"aggregated_set"')) { report.outOfScope++; continue; }
    const result = evaluateOffLine(line, scope);
    for (const reason of result.valueRejects) bump(report.valueRejects, reason);
    if (!result.ok) {
      if (result.reason === 'malformed') report.malformed++;
      else if (result.reason === 'out_of_scope') report.outOfScope++;
      else for (const m of result.markets) { tally(m).seen++; bump(tally(m).rejected, result.reason); }
      continue;
    }
    const { candidate } = result;
    for (const m of result.mismatched) {
      tally(m).seen++; bump(tally(m).rejected, 'label_of_other_market');
    }
    if (sink.flush) {
      inflight.push({ candidate, outcomes: sink(candidate) });
      if (inflight.length >= BATCH) { await sink.flush(); await settle(); }
    } else settleOne(candidate, await sink(candidate));
  }
  if (sink.flush) { await sink.flush(); await settle(); }
  return report;
}

const NUTRIENT_SQL = OFF_COLUMNS.join(',');
const sameNumber = (a: unknown, b: number | null) => (a === null || a === undefined ? null : Number(a)) === b;

// Draft-only writer. Batches commit every BATCH records; one draft release is created lazily
// by the first record that changes anything. Each record runs in a savepoint so one rejected
// row (a constraint, a NUL byte) costs that row, not the batch.
export function offWriter(client: pg.Client, releasedAt: string) {
  const release = releasedAt;
  let open = 0;
  let releaseId: string | null = null;
  let newRaw = 0;
  const sink: Sink = async (c) => {
    if (!open) await client.query('begin');
    open++;
    const outcomes: Record<Market, MarketOutcome> = {};
    let added = 0;
    let newRelease = false;
    await client.query('savepoint record');
    try {
      const prev = (await client.query(`select distinct on (v.food_id) v.food_id, f.market, v.name,
          v.search_name, v.state, v.preparation, v.basis, v.food_status, v.food_group, v.carbohydrate_definition,
          ${OFF_COLUMNS.map(col => `v.${col}`).join(',')}, a.fields
        from food.food_source_record s
        join food.food_version v on v.source_record_id=s.id and v.licence_family='off'
        join food.food f on f.id=v.food_id
        join food.food_release r on r.id=v.food_release_id
        left join food.food_added_off a on a.source_record_id=v.source_record_id and a.food_release_id=v.food_release_id
        where s.src='SRC-06' and s.record_id=$1
        order by v.food_id, r.version desc`, [c.code])).rows;
      const byMarket = new Map<string, Record<string, unknown>>(prev.map(row => [row.market as string, row]));
      const changed: MarketPlan[] = [];
      for (const plan of c.markets) {
        const old = byMarket.get(plan.market);
        const marketAdded = (c.added.markets as Record<string, unknown>)[plan.market];
        const same = old && old.name === c.name && old.search_name === c.searchName && old.state === 'as_sold'
          && old.preparation === 'none' && old.basis === c.basis && old.food_status === plan.status
          && old.carbohydrate_definition === plan.carbohydrate_definition && (old.food_group ?? null) === c.group
          && OFF_COLUMNS.every(col => sameNumber(old[col], c.values[col]))
          && sameJson((old.fields as { markets?: Record<string, unknown> } | null)?.markets?.[plan.market], marketAdded);
        if (same) { outcomes[plan.market] = 'unchanged'; continue; }
        if (!old) {
          // FD-42: one product per GTIN and market inside SRC-06.
          const clash = await client.query(`select 1 from food.food_barcode b
            join food.food f on f.id=b.food_id and f.market=b.market
            join food.food_version v on v.food_id=f.id and v.licence_family='off'
            join food.food_source_record s on s.id=v.source_record_id
            where b.gtin=$1 and b.market=$2 and s.src='SRC-06' and s.record_id<>$3 limit 1`,
          [c.gtin, plan.market, c.code]);
          if (clash.rowCount) { outcomes[plan.market] = 'duplicate_gtin'; continue; }
        }
        changed.push(plan);
      }
      if (changed.length) {
        const source = await client.query(`insert into food.food_source_record
          (src,record_id,raw_record,source_release,source_released_at)
          values('SRC-06',$1,$2,$3,$4) on conflict (src,record_id,content_sha256) do nothing returning id`,
        [c.code, c.raw, release, releasedAt]);
        if (source.rowCount) added = 1;
        // Raw bytes already stored keep their release: a new version of them names that release (the version guard).
        const stored = source.rows[0] ? { id: source.rows[0].id as string, source_release: release, source_released_at: releasedAt }
          : (await client.query(`select id, source_release, source_released_at::text from food.food_source_record
          where src='SRC-06' and record_id=$1 and content_sha256=encode(sha256($2::bytea),'hex')`, [c.code, c.raw])).rows[0];
        const sourceId: string = stored.id;
        if (!releaseId) {
          releaseId = (await client.query('insert into food.food_release default values returning id')).rows[0].id;
          newRelease = true;
        }
        const provenance = Object.fromEntries(OFF_COLUMNS.filter(col => c.evidence[col]).map(col =>
          [col, { src: 'SRC-06', record_id: c.code, release: stored.source_release, ...c.evidence[col] }]));
        for (const plan of changed) {
          const foodId: string = (byMarket.get(plan.market)?.food_id as string | undefined)
            ?? (await client.query("insert into food.food(market,layer) values($1,'branded') returning id",
              [plan.market])).rows[0].id;
          await client.query(`insert into food.food_barcode(gtin,market,food_id) values($1,$2,$3)
            on conflict do nothing`, [c.gtin, plan.market, foodId]);
          const params = [foodId, releaseId, sourceId, c.name, c.searchName, stored.source_release, stored.source_released_at,
            plan.status, c.basis, plan.carbohydrate_definition, ...OFF_COLUMNS.map(col => c.values[col]),
            provenance, c.group];
          const slots = params.map((_, i) => `$${i + 1}`);
          await client.query(`insert into food.food_version
            (licence_family,food_id,food_release_id,source_record_id,name,search_name,source_name,
              source_release,source_released_at,food_status,state,preparation,basis,carbohydrate_definition,
              allergens,${NUTRIENT_SQL},value_provenance,food_group)
            values('off',${slots.slice(0, 5).join(',')},'Open Food Facts',${slots.slice(5, 8).join(',')},
              'as_sold','none',${slots.slice(8, 10).join(',')},'{}',${slots.slice(10).join(',')})`, params);
          outcomes[plan.market] = 'new';
        }
        // FD-57: what we add to the raw record is its own row, one per source record and release.
        await client.query(`insert into food.food_added_off(source_record_id,food_release_id,fields)
          values($1,$2,$3) on conflict do nothing`, [sourceId, releaseId, c.added]);
      }
      await client.query('release savepoint record');
    } catch {
      // The reason is not kept: database messages can quote untrusted record data.
      await client.query('rollback to savepoint record');
      for (const plan of c.markets) outcomes[plan.market] = 'db_error';
      added = 0;
      if (newRelease) releaseId = null;
    }
    newRaw += added;
    if (open >= BATCH) { await client.query('commit'); open = 0; }
    return outcomes;
  };
  const finish = async () => {
    if (open) { await client.query('commit'); open = 0; }
    return { newRaw, releaseId };
  };
  return { sink, finish };
}

// jsonb keeps object keys in its own order (shorter keys first), not in the order they were written: stored added fields
// equal fresh ones when every key and value does, whatever the order.
function sameJson(a: unknown, b: unknown): boolean {
  const canonical = (value: unknown): unknown => Array.isArray(value) ? value.map(canonical)
    : value && typeof value === 'object' ? Object.fromEntries(Object.keys(value).sort().map(key => [key, canonical((value as Record<string, unknown>)[key])]))
    : value;
  return JSON.stringify(canonical(a)) === JSON.stringify(canonical(b));
}

function uuidv7() {
  const bytes = randomBytes(16);
  bytes.writeUIntBE(Date.now(), 0, 6);
  bytes[6] = (bytes[6]! & 0x0f) | 0x70; bytes[8] = (bytes[8]! & 0x3f) | 0x80;
  const hex = bytes.toString('hex');
  return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`;
}

// The same writer as offWriter, for a whole batch at once: one read of what the batch already has, one decision
// per record in memory, then set-based inserts. A weekly export of two million records is a handful of
// statements per 500 records instead of eight per record. A batch that fails is retried record by record
// with offWriter, so one bad row still costs only that row.
export function offBatchWriter(client: pg.Client, releasedAt: string) {
  const release = releasedAt;
  let releaseId: string | null = null;
  let newRaw = 0;
  let pending: { c: OffCandidate; resolve: (o: Record<Market, MarketOutcome>) => void }[] = [];
  let slow: ReturnType<typeof offWriter> | null = null;
  let slowDone: { newRaw: number; releaseId: string | null } = { newRaw: 0, releaseId: null };
  const byMarketKey = (gtin: string, market: string) => `${gtin}|${market}`;

  async function processBatch(batch: OffCandidate[]) {
    const outcomes = batch.map(() => ({}) as Record<Market, MarketOutcome>);
    let createdRelease = false;
    await client.query('begin');
    try {
      const prevRows = (await client.query(`select distinct on (s.record_id, v.food_id) s.record_id, v.food_id, f.market,
          v.name, v.search_name, v.state, v.preparation, v.basis, v.food_status, v.food_group, v.carbohydrate_definition,
          ${OFF_COLUMNS.map(col => `v.${col}`).join(',')}, a.fields
        from food.food_source_record s
        join food.food_version v on v.source_record_id=s.id and v.licence_family='off'
        join food.food f on f.id=v.food_id
        join food.food_release r on r.id=v.food_release_id
        left join food.food_added_off a on a.source_record_id=v.source_record_id and a.food_release_id=v.food_release_id
        where s.src='SRC-06' and s.record_id = any($1::text[])
        order by s.record_id, v.food_id, r.version desc`, [batch.map(c => c.code)])).rows;
      const previous = new Map<string, Map<string, Record<string, unknown>>>();
      for (const row of prevRows) {
        const code = row.record_id as string;
        (previous.get(code) ?? previous.set(code, new Map()).get(code)!).set(row.market as string, row);
      }
      // Records with no earlier version of the market: their GTIN must not belong to another record (FD-42).
      const wanted = batch.flatMap(c => c.markets.filter(plan => !previous.get(c.code)?.has(plan.market))
        .map(plan => ({ gtin: c.gtin, market: plan.market })));
      const taken = new Map<string, Set<string>>();
      if (wanted.length) {
        const clashes = (await client.query(`select b.gtin, b.market, s.record_id from food.food_barcode b
          join food.food f on f.id=b.food_id and f.market=b.market
          join food.food_version v on v.food_id=f.id and v.licence_family='off'
          join food.food_source_record s on s.id=v.source_record_id
          where s.src='SRC-06' and (b.gtin, b.market) in (select t.g, t.m from unnest($1::text[], $2::text[]) as t(g, m))`,
        [wanted.map(w => w.gtin), wanted.map(w => w.market)])).rows;
        for (const row of clashes) {
          const key = byMarketKey(row.gtin as string, row.market as string);
          (taken.get(key) ?? taken.set(key, new Set()).get(key)!).add(row.record_id as string);
        }
      }
      const claimed = new Map<string, string>();
      const changed: { c: OffCandidate; index: number; plan: MarketPlan; foodId: string; isNew: boolean }[] = [];
      batch.forEach((c, index) => {
        for (const plan of c.markets) {
          const old = previous.get(c.code)?.get(plan.market);
          const marketAdded = (c.added.markets as Record<string, unknown>)[plan.market];
          const same = old && old.name === c.name && old.search_name === c.searchName && old.state === 'as_sold'
            && old.preparation === 'none' && old.basis === c.basis && old.food_status === plan.status
            && old.carbohydrate_definition === plan.carbohydrate_definition && (old.food_group ?? null) === c.group
            && OFF_COLUMNS.every(col => sameNumber(old[col], c.values[col]))
            && sameJson((old.fields as { markets?: Record<string, unknown> } | null)?.markets?.[plan.market], marketAdded);
          if (same) { outcomes[index]![plan.market] = 'unchanged'; continue; }
          if (!old) {
            const key = byMarketKey(c.gtin, plan.market);
            const other = [...(taken.get(key) ?? [])].some(code => code !== c.code) || (claimed.has(key) && claimed.get(key) !== c.code);
            if (other) { outcomes[index]![plan.market] = 'duplicate_gtin'; continue; }
            claimed.set(key, c.code);
          }
          changed.push({ c, index, plan, foodId: (old?.food_id as string | undefined) ?? uuidv7(), isNew: !old });
        }
      });
      if (changed.length) {
        if (!releaseId) {
          releaseId = (await client.query('insert into food.food_release default values returning id')).rows[0].id;
          createdRelease = true;
        }
        const sources = [...new Map(changed.map(x => [x.c.code, x.c])).values()];
        const inserted = await client.query(`insert into food.food_source_record
          (src,record_id,raw_record,source_release,source_released_at)
          select 'SRC-06', t.code, t.raw, $3, $4::date from unnest($1::text[], $2::bytea[]) as t(code, raw)
          on conflict (src,record_id,content_sha256) do nothing`,
        [sources.map(c => c.code), sources.map(c => c.raw), release, releasedAt]);
        // The hash is computed here, not by the database: the bytes would cross the wire a second time.
        // Raw bytes already stored keep their release: a new version of them names that release (the version guard).
        const stored = new Map((await client.query(`select s.id, s.record_id, s.source_release, s.source_released_at::text
          from food.food_source_record s join unnest($1::text[], $2::text[]) as t(code, sha)
            on s.src='SRC-06' and s.record_id=t.code and s.content_sha256=t.sha`,
        [sources.map(c => c.code), sources.map(c => createHash('sha256').update(c.raw).digest('hex'))])).rows
          .map(row => [row.record_id as string, row as { id: string; source_release: string; source_released_at: string }]));
        const ids = new Map([...stored].map(([code, row]) => [code, row.id]));
        const fresh = changed.filter(x => x.isNew);
        if (fresh.length) await client.query(`insert into food.food(id,market,layer)
          select t.id, t.market, 'branded' from unnest($1::uuid[], $2::text[]) as t(id, market)`,
        [fresh.map(x => x.foodId), fresh.map(x => x.plan.market)]);
        await client.query(`insert into food.food_barcode(gtin,market,food_id)
          select t.g, t.m, t.f from unnest($1::text[], $2::text[], $3::uuid[]) as t(g, m, f) on conflict do nothing`,
        [changed.map(x => x.c.gtin), changed.map(x => x.plan.market), changed.map(x => x.foodId)]);
        const rows = changed.map(({ c, plan, foodId }) => ({
          food_id: foodId, source_record_id: ids.get(c.code), name: c.name, search_name: c.searchName,
          source_release: stored.get(c.code)!.source_release, source_released_at: stored.get(c.code)!.source_released_at,
          food_status: plan.status, basis: c.basis, cd: plan.carbohydrate_definition, food_group: c.group,
          provenance: Object.fromEntries(OFF_COLUMNS.filter(col => c.evidence[col]).map(col =>
            [col, { src: 'SRC-06', record_id: c.code, release: stored.get(c.code)!.source_release, ...c.evidence[col] }])),
          ...Object.fromEntries(OFF_COLUMNS.map(col => [col, c.values[col]])) }));
        await client.query(`insert into food.food_version
          (licence_family,food_id,food_release_id,source_record_id,name,search_name,source_name,source_release,
            source_released_at,food_status,state,preparation,basis,carbohydrate_definition,allergens,${NUTRIENT_SQL},
            value_provenance,food_group)
          select 'off', x.food_id, $2::uuid, x.source_record_id, x.name, x.search_name, 'Open Food Facts', x.source_release, x.source_released_at::date,
            x.food_status, 'as_sold', 'none', x.basis, x.cd, '{}'::jsonb, ${OFF_COLUMNS.map(col => `x.${col}`).join(',')},
            x.provenance, x.food_group
          from jsonb_to_recordset($1::jsonb) as x(food_id uuid, source_record_id uuid, name text, search_name text,
            source_release text, source_released_at text,
            food_status text, basis text, cd text, food_group text, provenance jsonb,
            ${OFF_COLUMNS.map(col => `${col} numeric`).join(',')})`,
        [JSON.stringify(rows), releaseId]);
        await client.query(`insert into food.food_added_off(source_record_id,food_release_id,fields)
          select x.source_record_id, $2::uuid, x.fields
          from jsonb_to_recordset($1::jsonb) as x(source_record_id uuid, fields jsonb) on conflict do nothing`,
        [JSON.stringify(sources.map(c => ({ source_record_id: ids.get(c.code), fields: c.added }))), releaseId]);
        for (const { index, plan } of changed) outcomes[index]![plan.market] = 'new';
        newRaw += inserted.rowCount ?? 0;
      }
      await client.query('commit');
      return outcomes;
    } catch (error) {
      await client.query('rollback').catch(() => undefined);
      if (createdRelease) releaseId = null;
      throw error;
    }
  }

  const sink: Sink = (c) => new Promise(resolve => { pending.push({ c, resolve }); });
  sink.flush = async () => {
    const batch = pending; pending = [];
    if (!batch.length) return;
    try {
      const outcomes = await processBatch(batch.map(item => item.c));
      batch.forEach((item, i) => item.resolve(outcomes[i]!));
    } catch (error) {
      // A database message can quote untrusted record data, so it is not kept; a failure of our own code is shown,
      // because the slow path below hides it as a lost speed.
      if (!(error as { severity?: unknown }).severity) console.error(`batch writer failed: ${(error as Error).message}`);
      slow ??= offWriter(client, releasedAt);
      for (const item of batch) item.resolve(await slow.sink(item.c));
      // Commit its open transaction now: a later failed batch must not roll these records back.
      slowDone = await slow.finish();
    }
  };
  const finish = async () => {
    await sink.flush!();
    return { newRaw: newRaw + slowDone.newRaw, releaseId: releaseId ?? slowDone.releaseId };
  };
  return { sink, finish };
}

export async function importOffFile(client: pg.Client | null, path: string, releasedAt: string, limit = Infinity,
  scope: Scope = null, legacyOnly = false) {
  if (!/^\d{4}-\d{2}-\d{2}$/.test(releasedAt) || Number.isNaN(Date.parse(releasedAt)))
    throw new Error('Open Food Facts export date must be YYYY-MM-DD');
  const seen = new Set<string>();
  const dry: Sink = async (c) => Object.fromEntries(c.markets.map(p => {
    const key = `${c.gtin}|${p.market}`;
    if (seen.has(key)) return [p.market, 'duplicate_gtin'];
    seen.add(key); return [p.market, 'new'];
  }));
  const writer = client ? offBatchWriter(client, releasedAt) : null;
  try {
    const report = await scanOff(openOffExport(path), writer?.sink ?? dry, limit, scope, legacyOnly);
    const done = await writer?.finish();
    return { ...report, newRaw: done?.newRaw ?? 0, releaseId: done?.releaseId ?? null };
  } catch (error) {
    if (client) await client.query('rollback').catch(() => undefined);
    throw error;
  }
}