import {
  q,
  type ColumnMeta,
  type NormalizeCluster,
  type NormalizeColumnPlan,
  type NormalizeMerge,
  type NormalizePlan,
  type NormalizeVariant,
} from '@wellformed/shared';
import { badRequest, notFound } from '../errors.js';
import { invalidateMeta, transact, type Db } from '../db/driver.js';
import { findColumn, getTable, loadMeta } from '../db/meta.js';
import { recordAudit, type Actor } from './audit.js';

/**
 * Normalising: the tidy-up the tightening cannot do.
 *
 * A tightening asks whether every row fits a constraint, and `c` fits a text
 * column perfectly well — so the one lower-case `c` in 2,294 rows of `C`
 * survives every check while it splits filters, pollutes option lists and
 * makes two answers out of one. That mess has no error to trip on, which is
 * why nobody opens a dialog to look for it. This module is the looking.
 *
 * The scan (`planNormalize`) reads every text and select column and clusters
 * the spellings that read as the same answer — same letters once case is
 * ignored and whitespace is straightened. Nothing cleverer: no edit distance,
 * no diacritics, because those are guesses of a different order and a merge is
 * only safe to offer when a person can verify it at a glance. The merge
 * (`applyNormalize`) then rewrites the also-ran spellings to the survivor, in
 * one transaction, one audit line per merge, and a select column's option
 * list follows its values.
 *
 * Like the tightening next door, the writes here go below the mutate path on
 * purpose: a merge touching 4,000 rows must not fire a workflow rule 4,000
 * times, and the audit lines are the record of what happened instead.
 */

/** Clusters reported per column. Enough to work through; the rest waits for a rescan. */
const MAX_CLUSTERS = 50;

/**
 * A column holding any value longer than this is a prose column, and prose
 * columns are left alone — the whole column, not just its long values. A
 * person can compare `c` with `C`, or `Betaald` with `betaald`, at a glance;
 * that glance is what makes the merge safe to offer, and a column of
 * sentences never gets it.
 */
export const MAX_VALUE_LENGTH = 40;

/**
 * Straightened whitespace: edges trimmed, runs collapsed. Stray spaces are
 * not the kind of mess worth a question, so everything here — the clusters
 * the plan reports and the rewrites the apply runs — speaks in these tidied
 * spellings, and the raw whitespace variants fold in automatically.
 */
function tidy(value: string): string {
  return value.replace(/\s+/g, ' ').trim();
}

/**
 * Which columns the scan reads. Text and select hold spellings; every other
 * type stores values the grid already writes one way. A unique column is
 * skipped even when it is text: merging two spellings there collides with the
 * index, and on a unique column two near-identical values are a question
 * about the rows, not about the spelling.
 */
export function scannable(col: ColumnMeta): boolean {
  return (col.type === 'text' || col.type === 'select') && !col.unique;
}

// ---- the scan -------------------------------------------------------------

export function planNormalize(db: Db, tableName: string): NormalizePlan {
  const table = getTable(loadMeta(db), tableName);
  const scanned = table.columns.filter(scannable);

  const columns: NormalizeColumnPlan[] = [];
  for (const col of scanned) {
    const clusters = scanColumn(db, table.name, col);
    if (clusters.length) {
      columns.push({
        column: col.name,
        clusters: clusters.slice(0, MAX_CLUSTERS),
        more: Math.max(0, clusters.length - MAX_CLUSTERS),
      });
    }
  }

  return { columns, scanned: scanned.length };
}

/** The spelling groups a column's values fold into, keyed by folded form. */
export interface ColumnSpellings {
  groups: Map<string, Map<string, number>>;
  /** Cells holding anything — the denominator for "do the values repeat". */
  present: number;
}

/**
 * Fold counted raw values (and a select's options, at count zero) into
 * spelling groups. Stray whitespace is not worth a question — `Betaald` and
 * ` Betaald` fold into one spelling here, and the apply straightens them
 * without asking — so everything downstream speaks in tidied spellings.
 */
export function groupSpellings(
  counted: Iterable<{ value: string; count: number }>,
  options: string[],
): ColumnSpellings {
  const groups = new Map<string, Map<string, number>>();
  let present = 0;
  const note = (value: string, count: number) => {
    const spelling = tidy(value);
    // A value that is nothing but whitespace has no spelling at all; emptying
    // it is a clearing, which is the tightening's move, not this one's.
    if (spelling === '') return;
    present += count;
    const key = spelling.toLowerCase();
    const group = groups.get(key) ?? new Map<string, number>();
    group.set(spelling, (group.get(spelling) ?? 0) + count);
    groups.set(key, group);
  };
  for (const { value, count } of counted) note(value, count);
  // A select's options are spellings too — `betaald` beside `Betaald` in the
  // list is the same mess with zero rows attached, and merging it must also
  // tidy the list a picker sees.
  for (const option of options) note(option, 0);
  return { groups, present };
}

/**
 * Normalising exists to prepare a tightening: merged spellings are what a
 * select's options or a foreign key's names are made from. So a text column
 * is only worth the trouble when its values repeat the way an enum's or a
 * key's do — a column of one-off names is neither.
 */
export function repeats(spellings: ColumnSpellings): boolean {
  return spellings.groups.size > 0 && spellings.groups.size * 2 <= spellings.present;
}

/** One group's spellings, the proposed survivor first: majority, then order. */
export function rankSpellings(group: Map<string, number>): NormalizeVariant[] {
  return [...group.entries()]
    .map(([value, count]) => ({ value, count }))
    .sort((a, b) => b.count - a.count || a.value.localeCompare(b.value));
}

/** The groups where spellings genuinely disagree, biggest messes first. */
export function clustersFrom(spellings: ColumnSpellings): NormalizeCluster[] {
  const clusters: NormalizeCluster[] = [];
  for (const group of spellings.groups.values()) {
    if (group.size < 2) continue;
    const variants = rankSpellings(group);
    clusters.push({ to: variants[0]!.value, variants });
  }
  return clusters.sort(
    (a, b) =>
      b.variants.reduce((n, v) => n + v.count, 0) - a.variants.reduce((n, v) => n + v.count, 0),
  );
}

export function scanColumn(db: Db, tableName: string, col: ColumnMeta): NormalizeCluster[] {
  const name = q(col.name);
  // GROUP BY instead of reading rows: the question is about spellings and how
  // often each occurs, and that is exactly what the database can answer.
  const rows = db.sqlite
    .prepare<[], { value: unknown; n: number }>(
      `SELECT ${name} AS value, COUNT(*) AS n FROM ${q(tableName)}
        WHERE ${name} IS NOT NULL AND ${name} != '' GROUP BY ${name}`,
    )
    .all();

  // One long value marks the whole column as prose. Checked over everything
  // the column holds — values and, for a select, its options.
  if (
    rows.some((r) => String(r.value).length > MAX_VALUE_LENGTH) ||
    (col.options ?? []).some((o) => o.length > MAX_VALUE_LENGTH)
  ) {
    return [];
  }

  const spellings = groupSpellings(
    rows.map((r) => ({ value: String(r.value), count: r.n })),
    // A select already declared itself an enum, so the repeat rule below
    // does not apply to it and its options join the groups.
    col.type === 'select' ? (col.options ?? []) : [],
  );
  if (col.type === 'text' && !repeats(spellings)) return [];

  return clustersFrom(spellings);
}

// ---- the merge ------------------------------------------------------------

export interface NormalizeResult {
  column: ColumnMeta;
  /** Rows whose value was rewritten. */
  changed: number;
}

export function applyNormalize(
  db: Db,
  tableName: string,
  columnName: string,
  merges: NormalizeMerge[],
  actor: Actor,
): NormalizeResult {
  const meta = loadMeta(db);
  const table = getTable(meta, tableName);
  const col = findColumn(table, columnName);
  if (!col) throw notFound(`"${tableName}" has no column "${columnName}".`);
  if (!scannable(col)) {
    throw badRequest(
      col.unique
        ? `"${col.label}" is unique; there, two near-identical values are a question about the rows, not the spelling.`
        : `"${col.label}" is a ${col.type} column; only text and select columns hold spellings to merge.`,
    );
  }

  // Every merge speaks in tidied spellings, and none may be circular: a
  // spelling may not be rewritten twice, and may not both survive one merge
  // and vanish in another.
  const cleaned: NormalizeMerge[] = [];
  const rewritten = new Set<string>();
  for (const merge of merges) {
    const to = tidy(merge.to);
    if (to === '') throw badRequest('A spelling cannot be merged into nothing.');
    const from = [...new Set(merge.from.map(tidy))].filter((f) => f !== '' && f !== to);
    if (!from.length) continue;
    for (const f of from) {
      if (rewritten.has(f)) throw badRequest(`"${f}" is named in two merges.`);
      rewritten.add(f);
    }
    cleaned.push({ to, from });
  }
  for (const merge of cleaned) {
    if (rewritten.has(merge.to)) {
      throw badRequest(`"${merge.to}" cannot both survive a merge and be rewritten by one.`);
    }
  }

  // What is stored, exactly as stored. Every raw value maps to one target:
  // its merge's survivor when its spelling was merged, its own tidied self
  // otherwise — stray whitespace is not a question, so it straightens here
  // automatically, alongside whatever was decided.
  const raws = db.sqlite
    .prepare<[], { value: unknown; n: number }>(
      `SELECT ${q(columnName)} AS value, COUNT(*) AS n FROM ${q(tableName)}
        WHERE ${q(columnName)} IS NOT NULL AND ${q(columnName)} != '' GROUP BY ${q(columnName)}`,
    )
    .all();
  if (raws.some((r) => String(r.value).length > MAX_VALUE_LENGTH)) {
    throw badRequest(`"${col.label}" holds long texts; those are left exactly as written.`);
  }

  const map = new Map<string, string>();
  for (const merge of cleaned) for (const f of merge.from) map.set(f, merge.to);

  const before = col.options ?? [];
  const after = col.type === 'select' ? mergedOptions(before, map) : null;

  let changed = 0;
  transact(db, () => {
    const write = db.sqlite.prepare(
      `UPDATE ${q(tableName)} SET ${q(columnName)} = ? WHERE ${q(columnName)} = ?`,
    );

    const merged = new Map<string, number>();
    let straightened = 0;
    for (const raw of raws) {
      const value = String(raw.value);
      const spelling = tidy(value);
      const target = map.get(spelling) ?? spelling;
      if (target === value) continue;
      write.run(target, value);
      changed += raw.n;
      if (map.has(spelling)) merged.set(spelling, (merged.get(spelling) ?? 0) + raw.n);
      else straightened += raw.n;
    }

    // One line per merge, not per row: the trail should read "c became C,
    // once", the way the person decided it.
    for (const merge of cleaned) {
      const n = merge.from.reduce((sum, f) => sum + (merged.get(f) ?? 0), 0);
      const touchedOptions =
        after !== null && (before.some((o) => merge.from.includes(tidy(o))) || !before.includes(merge.to));
      if (n > 0 || touchedOptions) {
        recordAudit(db, actor, {
          kind: 'edit',
          table: tableName,
          column: columnName,
          old: merge.from.join(' | '),
          nw: merge.to + (n > 0 ? ` — ${n} row${n === 1 ? '' : 's'}` : ''),
        });
      }
    }
    if (straightened > 0) {
      recordAudit(db, actor, {
        kind: 'edit',
        table: tableName,
        column: columnName,
        nw: `straightened whitespace — ${straightened} row${straightened === 1 ? '' : 's'}`,
      });
    }

    if (after !== null && !sameList(before, after)) {
      db.sqlite
        .prepare(`UPDATE _wellformed_column SET options = ? WHERE table_name = ? AND name = ?`)
        .run(JSON.stringify(after), tableName, columnName);
    }
  });

  invalidateMeta(db);
  const updated = findColumn(getTable(loadMeta(db), tableName), columnName);
  return { column: updated ?? col, changed };
}

/**
 * The option list after the merge: each option is straightened, a merged
 * spelling becomes its survivor in place, duplicates collapse onto the first,
 * and a survivor whose spelling lived only in the cells is appended — the
 * merged cells now hold it, and a list that refuses its own column's values
 * is broken.
 */
function mergedOptions(options: string[], map: Map<string, string>): string[] {
  const out: string[] = [];
  for (const option of options) {
    const spelling = tidy(option);
    const next = map.get(spelling) ?? spelling;
    if (next && !out.includes(next)) out.push(next);
  }
  for (const to of new Set(map.values())) {
    if (!out.includes(to)) out.push(to);
  }
  return out;
}

function sameList(a: string[], b: string[]): boolean {
  return a.length === b.length && a.every((v, i) => v === b[i]);
}