// The view join model, as GSF 8.15.x views behave (genesis-pal
// view/join/*.kt, ViewStructureReaderImpl.kt; genesis-codegen
// ViewRepoConfigGenerator.kt): a base table, joins that each start from the
// base or an earlier join, matched on one or more key links, INNER or OUTER,
// plus derived fields. A ViewDef is planned into the form the store
// materializes from, the schema types from and live propagation reads.

import type {
  AntiJoinDef,
  JoinDef,
  JoinFieldLink,
  JoinLink,
  Row,
  ViewDef,
  ViewRowSources,
} from '../types.ts';

// A field link reads its source field from the table `fromAlias` names, or,
// with none, from the view row built so far (see linksOf).
export type PlannedLink =
  | { kind: 'field'; fromAlias: string | undefined; from: string; to: string }
  | { kind: 'value'; to: string; value: unknown };

export interface PlannedJoin {
  def: JoinDef;
  table: string;
  // The name `tables` holds its row under, and joins and links read it by.
  // Undefined for a join with no alias whose table name an earlier table of
  // the view already goes by (see planView).
  alias: string | undefined;
  // The table the join starts from: its own `fromAlias`, else the base.
  fromAlias: string;
  links: PlannedLink[];
  inner: boolean;
  backwardsJoin: boolean;
  // Whether its columns keep their own nullability: true when the join is
  // reached from the base through INNER joins only. Any other join's columns
  // are nullable, as an OUTER join (or anything under one) may find nothing
  // (ViewStructureReaderImpl.kt transformOutputFields / getInnerJoinAliases,
  // GSF v8.15.29).
  alwaysMatched: boolean;
}

export interface PlannedAntiJoin {
  def: AntiJoinDef;
  table: string;
  links: PlannedLink[];
}

export interface ViewPlan {
  baseAlias: string;
  joins: PlannedJoin[];
  antiJoins: PlannedAntiJoin[];
}

const isFieldPair = (on: JoinDef['on']): on is JoinFieldLink =>
  on.length === 2 && typeof on[0] === 'string' && typeof on[1] === 'string';

// A pair, and an object link with no `fromAlias`, read from the join's own
// `fromAlias`. A join with none reads them from the view row built so far —
// the base row's columns, overlaid by the selects of the joins before it —
// as joins always have, so one written before `fromAlias` existed can still
// key on an earlier join's selected column. For a GSF view that is the base
// row's field (an alias never repeats a base column the view exposes); name
// `fromAlias` to read one table's own row.
function linksOf(on: JoinDef['on'], fromAlias: string | undefined): PlannedLink[] {
  const links: JoinLink[] = isFieldPair(on) ? [on] : on;
  return links.map((link): PlannedLink => {
    if (Array.isArray(link)) return { kind: 'field', fromAlias, from: link[0], to: link[1] };
    if ('value' in link) return { kind: 'value', to: link.to, value: link.value };
    return { kind: 'field', fromAlias: link.fromAlias ?? fromAlias, from: link.from, to: link.to };
  });
}

// The plan of a view. Planned on every use (it costs next to nothing), so a
// definition changed in place takes effect, as it always has. Throws on a
// definition GSF would refuse to compile: a join with no `on` link, an alias
// given twice, or a join (or one of its links) reading from a table that is
// neither the base nor an earlier join.
//
// Before aliases existed, a table could be joined twice (or the base table
// joined to itself) with no alias, and such configs keep working: a join
// with no alias whose table name an earlier table already goes by (one
// without an explicit alias either) is joined as before, but goes by no name
// — the name keeps meaning the earlier table. Give it an `alias` to read it.
export function planView(viewName: string, view: ViewDef): ViewPlan {
  const fail = (message: string): never => {
    throw new Error(`[mock-server] View ${viewName}: ${message}`);
  };
  const baseAlias = view.baseAlias ?? view.base;
  // Every name so far: whether it was given explicitly, and whether a row
  // under it always exists when the view row does.
  const names = new Map<string, { explicit: boolean; alwaysMatched: boolean }>([
    [baseAlias, { explicit: view.baseAlias !== undefined, alwaysMatched: true }],
  ]);
  const checkSource = (label: string, alias: string | undefined) => {
    if (alias !== undefined && !names.has(alias)) {
      fail(`${label} reads from '${alias}', which is neither the base nor an earlier join`);
    }
  };
  const joins = (view.joins ?? []).map((def, index): PlannedJoin => {
    const name = def.alias ?? def.table;
    const fromAlias = def.fromAlias ?? baseAlias;
    const label = `join ${index + 1} (${name})`;
    checkSource(label, fromAlias);
    const links = linksOf(def.on, def.fromAlias);
    if (links.length === 0) fail(`${label} has no 'on' link`);
    for (const link of links) if (link.kind === 'field') checkSource(label, link.fromAlias);
    const taken = names.get(name);
    if (taken && (taken.explicit || def.alias !== undefined)) {
      fail(`${label}: alias '${name}' is already used; give each table its own alias`);
    }
    const inner = def.type === 'INNER';
    const alwaysMatched = inner && names.get(fromAlias)!.alwaysMatched;
    if (!taken) names.set(name, { explicit: def.alias !== undefined, alwaysMatched });
    return {
      def,
      table: def.table,
      alias: taken ? undefined : name,
      fromAlias,
      links,
      inner,
      backwardsJoin: def.backwardsJoin === true,
      alwaysMatched,
    };
  });
  const antiJoins = (view.antiJoins ?? []).map((def, index): PlannedAntiJoin => {
    const label = `anti-join ${index + 1} (${def.table})`;
    const links = linksOf(def.on, undefined);
    if (links.length === 0) fail(`${label} has no 'on' link`);
    for (const link of links) {
      if (link.kind === 'field' && link.fromAlias !== undefined && link.fromAlias !== baseAlias) {
        fail(`${label} can only read from the base table`);
      }
    }
    return { def, table: def.table, links };
  });
  return { baseAlias, joins, antiJoins };
}

// --- materialization ---------------------------------------------------------

type Where = ((joinRow: Row, viewRow: Row, sources: ViewRowSources) => boolean) | undefined;

const KEY_SEPARATOR = '\u0000';

// One index key for a list of values, or undefined when one is null or
// missing: GSF builds no join request from a null key, so it matches nothing.
// Typed, so 1 and '1' differ, as === tells them apart.
function keyOf(values: unknown[]): string | undefined {
  if (values.some((value) => value === undefined || value === null)) return undefined;
  return values.map((value) => `${typeof value}:${String(value)}`).join(KEY_SEPARATOR);
}

// Finds a join's match for one view row: the joined rows whose fields equal
// the source fields (an index over them, built once), narrowed by the
// constants and the `where` predicate. GSF only lets a dataserver read a view
// whose joins are one-to-one (their links cover a unique index), so a valid
// view has at most one match; when there are more, the last one in table
// order wins — with a `where`, the first that passes — as before this model.
function matcher(
  links: PlannedLink[],
  where: Where,
  rows: Row[],
): (viewRow: Row, sources: ViewRowSources) => Row | undefined {
  const fieldLinks = links.flatMap((link) => (link.kind === 'field' ? [link] : []));
  const valueLinks = links.flatMap((link) => (link.kind === 'value' ? [link] : []));
  let index: Map<string, Row[]> | undefined;
  const candidatesFor = (viewRow: Row, sources: ViewRowSources): Row[] => {
    if (fieldLinks.length === 0) return rows;
    if (!index) {
      index = new Map();
      for (const row of rows) {
        const key = keyOf(fieldLinks.map((link) => row[link.to]));
        if (key === undefined) continue;
        const bucket = index.get(key);
        if (bucket) bucket.push(row);
        else index.set(key, [row]);
      }
    }
    const key = keyOf(
      fieldLinks.map((link) =>
        link.fromAlias === undefined
          ? viewRow[link.from]
          : sources.tables[link.fromAlias]?.[link.from],
      ),
    );
    return key === undefined ? [] : (index.get(key) ?? []);
  };
  const constantsHold = (row: Row) => valueLinks.every((link) => row[link.to] === link.value);
  return (viewRow, sources) => {
    const candidates = candidatesFor(viewRow, sources);
    if (where) {
      return candidates.find((row) => constantsHold(row) && where(row, viewRow, sources));
    }
    return candidates.findLast(constantsHold);
  };
}

// Where a materialization reads its rows: the base table's, and each join's
// and anti-join's by its planned entry (so one alias of a table can read a
// different state of it than another).
export interface ViewRowSource {
  base: Row[];
  rowsOf: (entry: PlannedJoin | PlannedAntiJoin) => Row[];
}

// A view row, with the row each planned join matched (undefined for none),
// in plan order.
export interface ViewEntry {
  row: Row;
  matches: Array<Row | undefined>;
}

// The view's rows: each base row without an anti-joined match, joined in
// order (an INNER join with no match drops the row; a join whose source
// found nothing finds nothing either), plus the base `select` aliases, each
// join's `select` and the derived fields.
export function materializeEntries(
  viewName: string,
  view: ViewDef,
  source: ViewRowSource,
): ViewEntry[] {
  const plan = planView(viewName, view);
  const antiJoins = plan.antiJoins.map((antiJoin) =>
    matcher(antiJoin.links, antiJoin.def.where, source.rowsOf(antiJoin)),
  );
  const joins = plan.joins.map((join) => matcher(join.links, join.def.where, source.rowsOf(join)));
  const entries: ViewEntry[] = [];
  for (const base of source.base) {
    const sources: ViewRowSources = { tables: { [plan.baseAlias]: base } };
    if (antiJoins.some((match) => match(base, sources))) continue;
    const row: Row = { ...base };
    for (const [name, field] of Object.entries(view.select ?? {})) row[name] = base[field];
    const matches: Array<Row | undefined> = [];
    let dropped = false;
    for (const [index, join] of plan.joins.entries()) {
      // `where` sees the view row so far, as it always has: the base row's
      // columns and the selects before this join.
      const match = sources.tables[join.fromAlias] ? joins[index](row, sources) : undefined;
      if (!match && join.inner) {
        dropped = true;
        break;
      }
      matches.push(match);
      if (join.alias !== undefined) sources.tables[join.alias] = match;
      for (const [name, field] of Object.entries(join.def.select ?? {})) row[name] = match?.[field];
    }
    if (dropped) continue;
    if (!view.derived) {
      entries.push({ row, matches });
      continue;
    }
    const derived: Row = { ...row };
    for (const [name, compute] of Object.entries(view.derived)) {
      derived[name] = compute(row, sources);
    }
    entries.push({ row: derived, matches });
  }
  return entries;
}

// The planned joins and anti-joins that read a table.
export function entriesReading(
  plan: ViewPlan,
  tableName: string,
): Array<PlannedJoin | PlannedAntiJoin> {
  return [...plan.joins, ...plan.antiJoins].filter((entry) => entry.table === tableName);
}