/**
 * Cell-level provenance capture (ADR 0019).
 *
 * ADR 0018 answered *which producer* filled a cell. This module carries the
 * other half — *what that producer was given*, *which branch was taken*, and
 * *where the row came from* — as additive fields on writes the runtime already
 * makes. Nothing here performs a round trip, and every value is a fact the
 * runtime held at the write site rather than a declaration it trusted or a
 * mapping it reconstructed afterwards.
 *
 * Shared by the runtime (`ctx-types`/`context`), the run ledger, and the
 * durable sheet parse path, so all three agree on one shape and one bound.
 */

/**
 * One cell version a cell was computed from.
 *
 * Both addressing components are omitted when they match the cell that carries
 * the ref: no `table` means the same table, no `rowKey` means the same row.
 * That is not shorthand for the reader's convenience — it is what keeps the
 * trace inside the retained-row memory budget. Runtime sheet row keys are
 * namespace-salted hashes, and repeating one per ref per cell inflated a
 * mapped row by roughly forty percent, which the streaming-dataset budget test
 * rejects. Readers fill both in from the cell's own address; the durable sheet
 * read model does this so every consumer sees a complete ref.
 *
 * `version` is the attempt/fence identity of the value read and is set only
 * where the runtime already held one: absent means "not witnessed", never
 * "version 0".
 */
export interface PlayCellReadRef {
  table?: string;
  rowKey?: string;
  column: string;
  version?: number;
}

/**
 * The branch a per-row control-flow evaluation selected, with the read-set the
 * evaluator was handed. `branch` is the runtime's own vocabulary, never
 * authored free text.
 */
export interface PlayCellDecision {
  branch: string;
  reads: PlayCellReadRef[];
}

/** Row-grain birth: the source row the runtime held while building this row. */
export type PlayRowBornFromRow = { table: string; rowKey: string };

/**
 * Dataset-grain birth: rows arrived from free-form play-body code and no
 * per-row mapping was witnessed. It names the destination table and how many
 * rows entered, and deliberately says nothing about any single row.
 *
 * There is no `step` field. The only step name available at the write site is
 * `map:<table>`, which restates `table` — a derived join key, not an observed
 * fact — and the record already hangs off the dataset's own lifecycle entry.
 */
export type PlayDatasetBornFrom = {
  table: string;
  rowCountIn: number;
};

export type PlayRowBornFrom = PlayRowBornFromRow | PlayDatasetBornFrom;

/**
 * Per-cell cap on the durable read trace, matching the producer trace bound.
 * The read-set is the row's column list, so an unbounded trace would put a wide
 * row's whole schema into `_cell_meta` once per cell. Truncation keeps the
 * first columns in row order, which is stable across runs.
 */
export const MAX_CELL_READ_REFS = 12;

/**
 * Cap on how many cells one row's read order describes. A dataset has a handful
 * of columns; this only bounds pathological graphs, and it reports what it
 * dropped rather than trimming silently.
 */
export const MAX_ROW_READ_CELLS = 24;

/** Reserved `_cell_meta` key holding row-grain facts rather than a cell. */
export const ROW_META_CELL_KEY = '_row';

/**
 * Row-grain read order (ADR 0019).
 *
 * Every cell in a row reads a prefix of the same list: the row's input columns,
 * then each column this run wrote, in the order they became available. Storing
 * the list once per row and a cut point per cell is the same information as a
 * per-cell array at O(columns) instead of O(columns x cells) — which matters,
 * because this rides the retained-row memory budget on every mapped row.
 *
 * A written column is appended to `columns` before its own cell runs, so its
 * position *is* its cut point: everything before it was available, itself and
 * everything after it was not. `upto` therefore only carries cells the list
 * cannot place — non-persisted fields, and cells past the column cap.
 * `droppedCells` counts cells no cut describes at all, so an incomplete order
 * is visible instead of looking like a cell that read nothing.
 */
export interface PlayRowReadOrder {
  columns: string[];
  upto?: Record<string, number>;
  droppedCells?: number;
}

/**
 * How many leading `columns` entries a cell read, or null when the order
 * describes no read for it. Null is not zero: a cell that read nothing and a
 * cell nobody recorded must not render the same.
 */
export function cellReadCut(
  order: PlayRowReadOrder | null | undefined,
  column: string,
): number | null {
  if (!order) return null;
  const explicit = order.upto?.[column];
  if (typeof explicit === 'number' && Number.isFinite(explicit)) {
    return Math.max(0, Math.min(Math.trunc(explicit), order.columns.length));
  }
  const index = order.columns.indexOf(column);
  return index >= 0 ? index : null;
}

/** Row-grain `_cell_meta._row` record. */
export interface PlayRowMeta {
  reads?: PlayRowReadOrder;
  bornFrom?: PlayRowBornFromRow;
}

/**
 * Expand one cell's read-set out of the row order. Returns undefined when the
 * order describes no read for that cell, which is not the same as an empty
 * read-set and must not be rendered as one.
 */
export function hydrateCellReadRefs(
  order: PlayRowReadOrder | null | undefined,
  column: string,
  rowKey: string,
): Array<PlayCellReadRef & { rowKey: string }> | undefined {
  const cut = cellReadCut(order, column);
  if (cut === null || cut <= 0 || !order) return undefined;
  const refs = order.columns
    .slice(0, Math.min(cut, MAX_CELL_READ_REFS))
    .map((readColumn) => ({ rowKey, column: readColumn }));
  return refs.length > 0 ? refs : undefined;
}

/** Parse a stored row read order off untrusted transport. */
export function normalizeRowReadOrder(value: unknown): PlayRowReadOrder | null {
  if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
  const record = value as Record<string, unknown>;
  const columns = Array.isArray(record.columns)
    ? record.columns
        .filter((column): column is string => typeof column === 'string')
        .slice(0, MAX_CELL_READ_REFS)
    : [];
  if (columns.length === 0) return null;
  const rawUpto =
    record.upto &&
    typeof record.upto === 'object' &&
    !Array.isArray(record.upto)
      ? (record.upto as Record<string, unknown>)
      : {};
  const upto: Record<string, number> = {};
  let entries = 0;
  for (const [column, cut] of Object.entries(rawUpto)) {
    if (entries >= MAX_ROW_READ_CELLS) break;
    if (typeof cut !== 'number' || !Number.isFinite(cut) || cut < 0) continue;
    upto[column] = Math.min(Math.trunc(cut), columns.length);
    entries += 1;
  }
  const droppedCells = finiteNonNegativeInteger(record.droppedCells);
  return {
    columns,
    ...(entries > 0 ? { upto } : {}),
    ...(droppedCells ? { droppedCells } : {}),
  };
}

/** Enforce {@link MAX_CELL_READ_REFS} without reordering. */
export function boundCellReadRefs<T>(refs: readonly T[]): T[] {
  return refs.length > MAX_CELL_READ_REFS
    ? refs.slice(0, MAX_CELL_READ_REFS)
    : [...refs];
}

/**
 * Fold a newly stated dataset birth into the one already on the snapshot.
 *
 * A paged dataset registers more than once, each registration stating the rows
 * admitted so far, so the larger `rowCountIn` is the later truth. A phase event
 * that states no birth never erases one that was witnessed.
 */
export function mergeDatasetBornFrom(
  current: PlayDatasetBornFrom | null | undefined,
  next: PlayDatasetBornFrom | null | undefined,
): PlayDatasetBornFrom | null {
  if (!next) return current ?? null;
  if (!current) return next;
  return next.rowCountIn > current.rowCountIn ? next : current;
}

function finiteNonNegativeInteger(value: unknown): number | null {
  if (typeof value !== 'number' || !Number.isFinite(value) || value < 0) {
    return null;
  }
  return Math.trunc(value);
}

function nonEmptyString(value: unknown): string | null {
  if (typeof value !== 'string') return null;
  const trimmed = value.trim();
  return trimmed ? trimmed : null;
}

/**
 * Parse a dataset-grain birth record off untrusted transport. Returns null for
 * anything that is not a complete record: a partial birth record would claim a
 * lineage the writer never stated.
 */
export function normalizeDatasetBornFrom(
  value: unknown,
): PlayDatasetBornFrom | null {
  if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
  const record = value as Record<string, unknown>;
  const table = nonEmptyString(record.table);
  const rowCountIn = finiteNonNegativeInteger(record.rowCountIn);
  if (!table || rowCountIn === null) return null;
  return { table, rowCountIn };
}