import type pg from 'pg';
import { foodWords } from '../../core/food-search-name.ts';
import { singular } from './storage.ts';

// The merge snapshot of a food release (20 FD-62, FD-65, FD-42; migration 0073): which visible foods the search
// shows instead of which. It is computed over the latest version of every food in the published releases and the
// named draft releases, and written as one complete snapshot release. Nothing in the versions changes.

export const COMMON_NAME_SIMILARITY = 0.8; // FD-62 [20]: share of common trigrams of the two names
export const COMMON_ENERGY_AGREEMENT = 0.25; // a concept in two tables keeps its energy within a quarter (and 10 kcal)
export const NEAR_DUPLICATE_ENERGY = 0.02; // FD-65 [20]

export type MergeRule = 'FD-62' | 'FD-65' | 'FD-42';
export type MergeRow = { foodId: string; intoFoodId: string; rule: MergeRule; similarity: number | null; family: string };
export type HeldPair = { hidden: string; shown: string; hiddenName: string; shownName: string; similarity: number; why: string };
export type MergePlan = { merges: MergeRow[]; held: HeldPair[]; heldCount: number };

// FD-03 order of tables inside one market: the market's own table first. USDA Foundation is the newest analytical
// table, SR Legacy is frozen since 2018, FNDDS is a survey of prepared foods; AUSNUT before AFCD (FD-03).
const tableRank = `case s.src when 'SRC-01' then case convert_from(s.raw_record, 'UTF8')::jsonb ->> 'dataType'
    when 'Foundation' then 10 when 'SR Legacy' then 20 else 30 end
  when 'SRC-09' then 10 when 'SRC-05' then 20 else 10 end`;
// Branded: USDA Branded before Open Food Facts (FD-03).
const brandedRank = `case s.src when 'SRC-02' then 10 when 'SRC-06' then 20 else 30 end`;

// A word bag: the same words whatever the order, punctuation or plural ("Fennel bulb, raw" = "Bulb, fennel, raw").
// Trigram similarity alone lets "round steak" and "round roast" or "choice" and "select" merge, which are
// different foods; the second-model check of FD-62 is the final judge, until then they stay apart.
const wordList = (name: string) => foodWords(name).split(' ').filter(Boolean).map(singular);
const wordBag = (name: string) => wordList(name).sort().join(' ');
// pg_trgm's measure (shared trigrams over all trigrams, words padded with two spaces before and one after) on the
// words in the singular, so "Banana, raw" and "Bananas, raw" are the same name for FD-62 as they are for FD-24a.
export function trigramSimilarity(left: string, right: string) {
  const grams = (name: string) => new Set(wordList(name).flatMap(word => {
    const padded = `  ${word} `;
    return Array.from({ length: padded.length - 2 }, (_, at) => padded.slice(at, at + 3));
  }));
  const a = grams(left), b = grams(right);
  let shared = 0;
  for (const gram of a) if (b.has(gram)) shared++;
  return a.size + b.size - shared === 0 ? 0 : shared / (a.size + b.size - shared);
}
const kcal = (row: { enerc_kcal: number | null; enerc_kj: number | null }) =>
  row.enerc_kcal ?? (row.enerc_kj === null ? 0 : row.enerc_kj / 4.184);

// Builds the plan and, with `write`, the snapshot release that holds it. One transaction.
// `rejected`: foods the second model of FD-62 found to be another concept than their pair; they are not merged by FD-62.
export async function buildMergeSnapshot(client: pg.Client, draftReleaseIds: string[], write: boolean, rejected: ReadonlySet<string> = new Set()) {
  await client.query('begin');
  try {
    // The visible state of the catalogue: newest version of each food, with the search's own visibility rule.
    await client.query(`create temporary table merge_visible on commit drop as
      select * from (
        select distinct on (v.food_id) v.food_id, v.licence_family, f.market, f.layer, v.food_group, v.state, v.preparation, v.search_name,
          v.enerc_kcal, v.enerc_kj, v.food_status, v.source_released_at, v.source_record_id, s.src, s.record_id,
          (v.procnt_g is not null and v.fat_g is not null and v.chocdf_g is not null) as macros,
          case when f.layer = 'common' then ${tableRank} when f.layer = 'branded' then ${brandedRank} else 0 end as rank,
          case when f.layer = 'branded' then lower(btrim(coalesce(
            (select a.fields ->> 'brand' from food.food_added_off a where a.source_record_id = v.source_record_id and a.food_release_id = v.food_release_id),
            (select a.fields ->> 'brand' from food.food_added_open a where a.source_record_id = v.source_record_id and a.food_release_id = v.food_release_id)))) end as brand
        from food.food_version v
        join food.food f on f.id = v.food_id
        join food.food_release r on r.id = v.food_release_id
        join food.food_source_record s on s.id = v.source_record_id
        where r.state = 'published' or r.id = any($1::uuid[])
        order by v.food_id, r.version desc, v.id) latest
      where food_status = 'active' and food_group is not null and (enerc_kcal is not null or enerc_kj is not null)
        and (layer = 'chain' or macros)`, [draftReleaseIds]);
    await client.query('create index on merge_visible (market, food_group, state, preparation) where layer = \'common\'');
    await client.query('analyze merge_visible');
    const hidden = new Map<string, MergeRow>();

    // FD-42: one product per code and market; the layer order of FD-03 chooses which record is shown.
    const codes = await client.query(`select b.gtin, b.market, v.food_id, v.licence_family, v.rank, v.source_released_at from food.food_barcode b
      join merge_visible v on v.food_id = b.food_id and v.market = b.market and v.layer = 'branded'
      where (b.gtin, b.market) in (select b2.gtin, b2.market from food.food_barcode b2 join merge_visible v2
        on v2.food_id = b2.food_id and v2.market = b2.market and v2.layer = 'branded' group by b2.gtin, b2.market having count(distinct b2.food_id) > 1)
      order by b.gtin, b.market, v.rank, v.source_released_at desc,
        -- Inside one USDA Branded release the newer product card (higher FDC id) holds a reused code.
        case when v.src = 'SRC-02' and v.record_id ~ '^[0-9]+$' then v.record_id::bigint end desc nulls last, v.food_id`);
    let shownKey = '', shownFood = '';
    for (const row of codes.rows) {
      const key = `${row.gtin}|${row.market}`;
      if (shownKey !== key) { shownKey = key; shownFood = row.food_id; }
      else if (row.food_id !== shownFood)
        hidden.set(row.food_id, { foodId: row.food_id, intoFoodId: shownFood, rule: 'FD-42', similarity: null, family: row.licence_family });
    }

    // FD-65: same brand, same normalised name, energy within 2%: one row, the freshest record of the newest release.
    const near = await client.query(`select food_id, licence_family, market, brand, search_name, enerc_kcal, enerc_kj, source_released_at
      from merge_visible where layer = 'branded' and brand is not null and brand <> ''
        and (market, brand, search_name) in (select market, brand, search_name from merge_visible
          where layer = 'branded' and brand is not null and brand <> '' group by 1, 2, 3 having count(*) > 1)
      order by market, brand, search_name, source_released_at desc, food_id`);
    const clusters = new Map<string, typeof near.rows>();
    for (const row of near.rows) {
      if (hidden.has(row.food_id)) continue;
      const key = `${row.market}|${foodWords(row.brand)}|${row.search_name}`;
      clusters.set(key, [...clusters.get(key) ?? [], row]);
    }
    for (const rows of clusters.values()) {
      let rest = rows;
      while (rest.length > 1) {
        const [anchor, ...others] = rest;
        const energy = kcal(anchor);
        rest = [];
        for (const other of others) {
          if (Math.abs(kcal(other) - energy) <= NEAR_DUPLICATE_ENERGY * Math.max(energy, kcal(other)))
            hidden.set(other.food_id, { foodId: other.food_id, intoFoodId: anchor.food_id, rule: 'FD-65', similarity: null, family: other.licence_family });
          else rest.push(other);
        }
      }
    }

    // FD-62: one general record per concept and market; the market's own table is shown. Two records of one table
    // with the same words in the name and agreeing energy are one concept too (a table that names every cultivar
    // "Apple" lists it seven times): the earlier record is shown. The rule has no "other table" condition.
    const pairs = await client.query(`select a.food_id as hidden, a.licence_family as hidden_family, b.food_id as shown, a.search_name as hidden_name,
        b.search_name as shown_name, a.rank as hidden_rank, b.rank as shown_rank, a.enerc_kcal as hidden_kcal, a.enerc_kj as hidden_kj,
        b.enerc_kcal as shown_kcal, b.enerc_kj as shown_kj, food.similarity(a.search_name, b.search_name) as similarity
      from merge_visible a join merge_visible b on a.layer = 'common' and b.layer = 'common' and a.market = b.market
        and a.food_group = b.food_group and a.state = b.state and a.preparation = b.preparation
        and (a.rank > b.rank or (a.rank = b.rank and a.food_id > b.food_id))
      where food.similarity(a.search_name, b.search_name) >= $1
      order by a.rank, a.food_id, similarity desc, b.rank, b.food_id`, [COMMON_NAME_SIMILARITY - 0.3]);
    const held: HeldPair[] = [];
    let heldCount = 0;
    // The database prefilters at a lower bound; the singular-word measure decides.
    const candidates = pairs.rows.map(pair => ({ ...pair, similarity: trigramSimilarity(pair.hidden_name, pair.shown_name) }))
      .filter(pair => pair.similarity >= COMMON_NAME_SIMILARITY)
      .sort((a, b) => a.hidden_rank - b.hidden_rank || (a.hidden < b.hidden ? -1 : a.hidden > b.hidden ? 1 : 0) || b.similarity - a.similarity
        || a.shown_rank - b.shown_rank || (a.shown < b.shown ? -1 : a.shown > b.shown ? 1 : 0));
    for (const pair of candidates) {
      if (hidden.has(pair.hidden) || rejected.has(pair.hidden)) continue;
      const a = kcal({ enerc_kcal: pair.hidden_kcal, enerc_kj: pair.hidden_kj });
      const b = kcal({ enerc_kcal: pair.shown_kcal, enerc_kj: pair.shown_kj });
      const why = wordBag(pair.hidden_name) !== wordBag(pair.shown_name) ? 'different words'
        : Math.abs(a - b) > Math.max(10, COMMON_ENERGY_AGREEMENT * Math.max(a, b)) ? 'energy differs' : '';
      // Two similar names inside one table are simply two foods of that table; only a pair that is one concept by words
      // and differs in energy, or a pair from two tables, waits for the second-model check.
      if (why === 'different words' && pair.hidden_rank === pair.shown_rank) continue;
      if (why) { heldCount++; if (held.length < 40) held.push({ hidden: pair.hidden, shown: pair.shown, hiddenName: pair.hidden_name,
        shownName: pair.shown_name, similarity: pair.similarity, why }); continue; }
      // Processed from the first table to the last, so a shown record is never hidden itself.
      if (hidden.has(pair.shown)) continue;
      hidden.set(pair.hidden, { foodId: pair.hidden, intoFoodId: pair.shown, rule: 'FD-62', similarity: pair.similarity, family: pair.hidden_family });
    }

    // A merge target is never hidden: follow a hidden target to the record that stands for it.
    const root = (id: string): string => { const next = hidden.get(id); return next ? root(next.intoFoodId) : id; };
    const merges = [...hidden.values()].map(row => ({ ...row, intoFoodId: root(row.intoFoodId) }));
    let releaseId: string | null = null;
    if (write) {
      releaseId = (await client.query('insert into food.food_release (merge_snapshot) values (true) returning id')).rows[0].id;
      for (let at = 0; at < merges.length; at += 20000) {
        const part = merges.slice(at, at + 20000);
        await client.query(`insert into food.food_merge (food_release_id, food_id, into_food_id, rule, similarity, licence_family)
          select $1, * from unnest($2::uuid[], $3::uuid[], $4::text[], $5::numeric[], $6::text[])`,
        [releaseId, part.map(row => row.foodId), part.map(row => row.intoFoodId), part.map(row => row.rule), part.map(row => row.similarity), part.map(row => row.family)]);
      }
    }
    await client.query('commit');
    return { releaseId, plan: { merges, held, heldCount } as MergePlan };
  } catch (error) { await client.query('rollback'); throw error; }
}