/**
 * What the record store says about one test, and what that is worth.
 *
 * A test earns its place in a pull request by having had new real test failures
 * before, so the score is built on new real test failures and decays slowly.
 * Deciding whether a failure is a new real test failure is the delicate part: a
 * test that is flaky on `main` produces one failure per commit and would
 * otherwise look like the most valuable test in the repository.
 * `foldObservations` is where that judgement is made, once per observation,
 * with the state it needs carried forward from the day before.
 */

import type {
  FlakeEvidence,
  ScoreInputs,
  TestIdentity,
} from "@commonfabric/test-support/records";
import { testIdentityKey } from "@commonfabric/test-support/records";
import { minOf } from "@commonfabric/utils/math";
import { isObjectNotArray, isObjectOrArray } from "@commonfabric/utils/types";
import {
  BREADTH_SATURATION,
  CHURN_HALF_LIFE_DAYS,
  CHURN_WINDOW_DAYS,
  COMMIT_REACH_DAYS,
  COST_WINDOW_DAYS,
  ENVIRONMENTAL_MIN_SOURCES,
  FLAKE_COMMIT_REACH,
  FLAKE_HALF_LIFE_RUNS,
  FLAKE_WINDOW_DAYS,
  FRESHNESS_FLOOR,
  FRESHNESS_HALF_LIFE_DAYS,
  MASS_FAILURE_MIN_IDENTITIES,
  NEW_REAL_TEST_FAILURE_BREADTH_WINDOW_DAYS,
  NEW_REAL_TEST_FAILURE_WEIGHT_LOCAL,
  NEW_REAL_TEST_FAILURE_WEIGHT_MAIN,
  NEW_REAL_TEST_FAILURE_WEIGHT_PR,
  PROVEN_SATURATION,
  SAME_COMMIT_REACH_DAYS,
  VALUE_FLOOR,
  WEIGHT_BREADTH,
  WEIGHT_CHURN,
  WEIGHT_PROVEN,
} from "./policy.ts";

/** Where a failure happened, which changes what it means. */
export type NewRealTestFailurePlace = "local" | "pr" | "main";

/** One execution of one test, as the publisher reads it from a report. */
export interface Observation {
  test: TestIdentity;
  outcome: "pass" | "fail" | "skip";

  /** UTC calendar day, "yyyy-mm-dd". */
  day: string;

  /**
   * When the run this came from started, ISO 8601 UTC. The day is what
   * the counters are kept by; this is what the order is taken from, which
   * a day is too coarse for.
   */
  startedAt: string;

  /** The commit the tests ran against. */
  commit: string;

  /**
   * The seed the run shuffled its tests by, absent for a run that ran
   * them in the order they were declared. Two observations at one commit
   * ran under the same conditions only where this agrees too: an
   * order-dependent test passes in one order and fails in another, and
   * that is a bug in the test rather than chance.
   */
  seed?: number;

  /**
   * Who saw it: the branch for a continuous-integration run, the
   * reporting person's login for a local one.
   */
  source: string;

  /** Where it ran, which is what a new real test failure there is worth. */
  place: NewRealTestFailurePlace;

  /**
   * What a passing execution on a lane's runner took, in milliseconds,
   * which is what its cost is measured from. Absent for any other
   * execution.
   */
  durationMs?: number;
}

/**
 * What one identity's history has accumulated.
 *
 * The fields counting new real test failures are named `...Catches`, and
 * the most recent one's day `lastCatch`, because those are the names the
 * stored aggregate holds them under. Every publisher that reads a stored
 * aggregate looks for them by those names.
 */
export interface IdentityState {
  /** New real test failures on a workstation. */
  localCatches: number;

  /** New real test failures on a pull request. */
  prCatches: number;

  /** New real test failures on the default branch. */
  mainCatches: number;

  /**
   * The day of the most recent new real test failure, absent when there
   * are none.
   */
  lastCatch?: string;

  /** The distinct sources among the new real test failures. */
  sources: string[];

  /**
   * The new real test failures by the day each is dated to. The four
   * fields above are what these add up to, kept beside them because the
   * score reads those and every publisher that reads a state reads them.
   * The days are what history is divided by: an alias moves the days
   * before its date to the new name.
   */
  newRealTestFailureDays: Record<string, NewRealTestFailureDay>;

  /** Failures per day, the numerator of the churn term. */
  failuresByDay: Record<string, number>;

  /** Runs per day, the denominator of that term and of the flake rate. */
  runsByDay: Record<string, number>;

  /** Flake observations per day, the numerator of the flake rate. */
  flakesByDay: Record<string, number>;

  /**
   * The day's passing durations, counted by bucket. Counted rather than
   * kept, because keeping every duration would make the state object
   * grow with the number of runs rather than with the number of tests.
   * A day is the unit because that is what ages out of the cost window.
   */
  costByDay: Record<string, DaySamples>;

  /**
   * The outcome of the most recent `main` run this identity appeared in,
   * in each of the last `MAIN_ORDERS_KEPT` orders `main` ran it in, the
   * most recent last. See `lastMainIn`.
   */
  lastMainByOrder?: MainOutcomes;

  /**
   * Failures on `main` that no later `main` run in their order has judged yet.
   * A failure that a later `main` run in its order passes was fixed by the
   * change between them, which makes it a new real test failure unless it
   * arrived in a crowd; one that is still failing is the same breakage
   * continuing, and waits.
   */
  pendingMain: Array<PendingMainFailure>;
}

/** What `main` last said about one identity, in each order it ran it in. */
export type MainOutcomes = Array<{ seed?: number; outcome: "pass" | "fail" }>;

/**
 * How many orders `main` is remembered in. It runs in two at once: the
 * day's, and the next day's in the daily run of every suite under that
 * day's seed.
 */
const MAIN_ORDERS_KEPT = 2;

/**
 * What `main` last said about an identity in one order, or, where it has
 * not run the identity in that order, in whichever order it ran it last.
 * A test that depends on its order can be failing in one and passing in
 * the other, so what `main` says in another order says nothing about this
 * one while this one has an answer of its own.
 */
export function lastMainIn(
  outcomes: MainOutcomes | undefined,
  seed: number | undefined,
): "pass" | "fail" | undefined {
  return (outcomes?.find((held) => held.seed === seed) ?? outcomes?.at(-1))
    ?.outcome;
}

/** `outcomes` with what `main` said in one order as its most recent. */
function withMainOutcome(
  outcomes: MainOutcomes | undefined,
  seed: number | undefined,
  outcome: "pass" | "fail",
): MainOutcomes {
  return [
    ...(outcomes ?? []).filter((held) => held.seed !== seed),
    seed === undefined ? { outcome } : { seed, outcome },
  ].slice(-MAIN_ORDERS_KEPT);
}

/** A failure on `main` waiting for a later `main` run to judge it. */
export interface PendingMainFailure {
  day: string;
  commit: string;
  seed?: number;
  source: string;

  /**
   * Whether the breakage arrived in a crowd: a run that newly broke at
   * least `MASS_FAILURE_MIN_IDENTITIES` identities. Whichever of them a
   * change runs sees that breakage, so it is no one test's new real test
   * failure.
   */
  crowded?: true;
}

/**
 * Which set of rules credited the new real test failures an aggregate
 * holds: what decides whether a failure is a new real test failure, and
 * when and where it is counted, which includes the day each is dated to.
 * Change it to any other value in the same change that alters those. A new
 * real test failure is held as a count rather than as the failure it came
 * from, so a count credited under other rules cannot be judged again by
 * itself, and the publisher folds the whole history behind it again
 * instead. The values are not ordered and nothing but equality is asked
 * of them; an aggregate carrying none was written before the stamps
 * began. It is stored under the name `catchRule`.
 */
export const NEW_REAL_TEST_FAILURE_RULE = 5;

/** One day's new real test failures of one identity. */
export interface NewRealTestFailureDay {
  local: number;
  pr: number;
  main: number;

  /** The distinct sources among that day's new real test failures. */
  sources: string[];
}

/** A fresh, empty history. */
export function emptyState(): IdentityState {
  return {
    localCatches: 0,
    prCatches: 0,
    mainCatches: 0,
    sources: [],
    newRealTestFailureDays: {},
    failuresByDay: {},
    runsByDay: {},
    flakesByDay: {},
    costByDay: {},
    pendingMain: [],
  };
}

/** Days between two "yyyy-mm-dd" days, `later` minus `earlier`. */
export function daysBetween(earlier: string, later: string): number {
  const from = Date.parse(`${earlier}T00:00:00Z`);
  const to = Date.parse(`${later}T00:00:00Z`);
  return Math.round((to - from) / 86_400_000);
}

function addSource(state: IdentityState, source: string): void {
  if (!state.sources.includes(source)) state.sources.push(source);
}

function creditNewRealTestFailure(
  state: IdentityState,
  place: NewRealTestFailurePlace,
  day: string,
  source: string,
): void {
  const failed = state.newRealTestFailureDays[day] ??
    (state.newRealTestFailureDays[day] = {
      local: 0,
      pr: 0,
      main: 0,
      sources: [],
    });
  failed[place]++;
  if (!failed.sources.includes(source)) failed.sources.push(source);
  if (place === "local") state.localCatches++;
  else if (place === "pr") state.prCatches++;
  else state.mainCatches++;
  if (state.lastCatch === undefined || state.lastCatch < day) {
    state.lastCatch = day;
  }
  addSource(state, source);
}

/**
 * Sets the totals of a state's new real test failures from its days of
 * new real test failures, which is what they are the totals of.
 */
function totalNewRealTestFailures(state: IdentityState): void {
  state.localCatches = 0;
  state.prCatches = 0;
  state.mainCatches = 0;
  state.sources = [];
  delete state.lastCatch;
  for (const [day, failed] of Object.entries(state.newRealTestFailureDays)) {
    state.localCatches += failed.local;
    state.prCatches += failed.pr;
    state.mainCatches += failed.main;
    for (const source of failed.sources) addSource(state, source);
    if (state.lastCatch === undefined || state.lastCatch < day) {
      state.lastCatch = day;
    }
  }
}

/** Whether a stored figure is a count. */
function isCount(value: unknown): value is number {
  return typeof value === "number" && Number.isInteger(value) && value >= 0;
}

/**
 * Whether a stored value is a day of new real test failures this can
 * count with.
 */
function isNewRealTestFailureDay(
  value: unknown,
): value is NewRealTestFailureDay {
  if (!isObjectNotArray(value)) return false;
  return isCount(value.local) && isCount(value.pr) && isCount(value.main) &&
    Array.isArray(value.sources) &&
    value.sources.every((source) => typeof source === "string");
}

/**
 * Reads a state's new real test failures forward. Its days of new real test
 * failures are kept where they and the day they are held against read, and any
 * other is dropped. The totals are then set from the days, so the two agree. A
 * state stored without days was credited under other rules, so the fold that
 * reads it folds its whole history again rather than counting it.
 */
export function readNewRealTestFailuresForward(state: IdentityState): void {
  const days: Record<string, NewRealTestFailureDay> = {};
  const held: unknown = state.newRealTestFailureDays;
  if (isObjectNotArray(held)) {
    for (const [day, failed] of Object.entries(held)) {
      if (isDay(day) && isNewRealTestFailureDay(failed)) days[day] = failed;
    }
  }
  state.newRealTestFailureDays = days;
  totalNewRealTestFailures(state);
}

/** Two counts of one day's new real test failures, as one. */
function mergeNewRealTestFailureDays(
  a: NewRealTestFailureDay | undefined,
  b: NewRealTestFailureDay,
): NewRealTestFailureDay {
  if (a === undefined) return b;
  return {
    local: a.local + b.local,
    pr: a.pr + b.pr,
    main: a.main + b.main,
    sources: [...new Set([...a.sources, ...b.sources])],
  };
}

/**
 * Every day a state holds anything for, in any of its histories. A day
 * is what an alias is applied by, so these are what a state is divided
 * along when part of its history belongs to another name.
 */
export function daysHeld(state: IdentityState): Set<string> {
  const days = new Set<string>([
    ...Object.keys(state.newRealTestFailureDays),
    ...Object.keys(state.failuresByDay),
    ...Object.keys(state.runsByDay),
    ...Object.keys(state.flakesByDay),
    ...Object.keys(state.costByDay),
  ]);
  for (const pending of state.pendingMain) days.add(pending.day);
  return days;
}

/**
 * Moves the days of `from` that `moves` answers yes for into `into`,
 * adding each to whatever `into` holds for the same day. Afterwards each
 * day is in one of the two states and not both, and each state's totals
 * are what its own days add up to.
 *
 * A day's cost is combined only where both were sealed under the same
 * set of cost rules. Where they were not, the day `into` holds stays and
 * the moved one is dropped, since a state holds one set's days.
 *
 * Which outcome `main` last gave goes with the history once `from` has
 * nothing left, and only to a state that has none of its own.
 */
export function moveHistory(
  from: IdentityState,
  into: IdentityState,
  moves: (day: string) => boolean,
): void {
  for (const [day, failed] of Object.entries(from.newRealTestFailureDays)) {
    if (!moves(day)) continue;
    into.newRealTestFailureDays[day] = mergeNewRealTestFailureDays(
      into.newRealTestFailureDays[day],
      failed,
    );
    delete from.newRealTestFailureDays[day];
  }
  for (
    const field of ["failuresByDay", "runsByDay", "flakesByDay"] as const
  ) {
    for (const [day, count] of Object.entries(from[field])) {
      if (!moves(day)) continue;
      into[field][day] = (into[field][day] ?? 0) + count;
      delete from[field][day];
    }
  }
  for (const [day, samples] of Object.entries(from.costByDay)) {
    if (!moves(day)) continue;
    const held = into.costByDay[day];
    if (held === undefined) into.costByDay[day] = samples;
    else if (held.rule === samples.rule) {
      into.costByDay[day] = {
        ...mergeSamples(held, samples),
        ...(held.rule === undefined ? {} : { rule: held.rule }),
      };
    }
    delete from.costByDay[day];
  }
  for (const pending of from.pendingMain) {
    if (moves(pending.day)) into.pendingMain.push(pending);
  }
  from.pendingMain = from.pendingMain.filter((pending) => !moves(pending.day));
  if (daysHeld(from).size === 0 && into.lastMainByOrder === undefined) {
    if (from.lastMainByOrder !== undefined) {
      into.lastMainByOrder = from.lastMainByOrder;
    }
    delete from.lastMainByOrder;
  }
  totalNewRealTestFailures(from);
  totalNewRealTestFailures(into);
}

/**
 * Renames the identities a fold context remembers, each to what `rename`
 * gives for it and the day the context holds it against, so that what the
 * context says about a test's recent runs follows its history to a new
 * name (see `moveHistory`). Each key of `mainAtCommit` and `credited`
 * starts with an identity key and goes on with parts that hold no space.
 *
 * What the context remembers at one point is held against the day it first
 * saw that point, and is renamed by that day. A record of a later day at
 * that point, a rerun of the commit, is the same test at the same commit,
 * so its outcomes go with the rest rather than leave the new name unable to
 * tell a rerun that disagrees from a first failure.
 */
export function renameInContext(
  context: FoldContext,
  rename: (key: string, day: string) => string,
): void {
  for (const seen of context.outcomesAtCommit.values()) {
    for (const [key, outcomes] of [...seen.identities]) {
      const to = rename(key, seen.day);
      if (to === key) continue;
      seen.identities.delete(key);
      const held = seen.identities.get(to) ?? new Set<string>();
      for (const outcome of outcomes) held.add(outcome);
      seen.identities.set(to, held);
    }
  }
  for (const [at, said] of [...context.mainAtCommit]) {
    const split = at.lastIndexOf(" ");
    const to = rename(at.slice(0, split), said.day) + at.slice(split);
    if (to === at) continue;
    context.mainAtCommit.delete(at);
    // A failure at a point is the commit being broken there, whichever
    // name saw it.
    if (context.mainAtCommit.get(to)?.outcome !== "fail") {
      context.mainAtCommit.set(to, said);
    }
  }
  for (const [attribution, day] of [...context.credited]) {
    const split = attribution.lastIndexOf(
      " ",
      attribution.lastIndexOf(" ") - 1,
    );
    const to = rename(attribution.slice(0, split), day) +
      attribution.slice(split);
    if (to === attribution) continue;
    context.credited.delete(attribution);
    // The earlier credit is the one a single fold would have kept.
    const held = context.credited.get(to);
    if (held === undefined || day < held) context.credited.set(to, day);
  }
  const failures = new Map<string, SeenFailure[]>();
  for (const [key, seen] of context.failures) {
    for (const failure of seen) {
      const to = rename(key, failure.day);
      const list = failures.get(to);
      if (list === undefined) failures.set(to, [failure]);
      else list.push(failure);
    }
  }
  context.failures.clear();
  for (const [key, seen] of failures) context.failures.set(key, seen);
  for (const crowd of context.crowds.values()) {
    const renamed = [...crowd.identities].map((key) => rename(key, crowd.day));
    crowd.identities = new Set(renamed);
  }
}

function bump(counts: Record<string, number>, day: string): void {
  counts[day] = (counts[day] ?? 0) + 1;
}

/**
 * The longest window any of a state's per-day counters is kept for, which
 * is also how long a failure on the default branch waits to be judged.
 */
const LONGEST_WINDOW_DAYS = Math.max(CHURN_WINDOW_DAYS, FLAKE_WINDOW_DAYS);

/**
 * Judges the failures on `main` that were waiting for a later `main` run. A
 * failure the next run still shows is the same breakage continuing, so it keeps
 * waiting and nothing new is learned. A failure the next run does not show is a
 * flake when that run is at the commit of any of them in the same order, or
 * `disagreed` says the test also failed at the pass's point, and a new real
 * test failure when it is at a later commit in the same order. A pass judges
 * only the failures in its own order: an order-dependent test stops failing
 * when the order moves on, so a pass in another order says nothing about
 * whether a change fixed anything. A pass with no failure in its own order
 * leaves every failure waiting, for a run in its own order or until
 * `trimWindows` ages it out. A pass that judges any of them ends the stretch,
 * and every failure in other orders goes with it, since they are the same
 * breakage seen in another order. Nothing separates a failure a change fixed
 * from one that healed itself, so a failure that healed itself in one order is
 * credited as a new real test failure as well. A breakage that arrived in a
 * crowd is no new real test failure however it ends, since whichever of the
 * crowd a change runs would have found it.
 */
function resolvePendingMain(
  state: IdentityState,
  observation: Observation,
  disagreed: boolean,
): void {
  if (observation.outcome === "fail") return;
  const inOrder = state.pendingMain.filter((pending) =>
    pending.seed === observation.seed
  );
  if (inOrder.length === 0) return;
  // A failure nothing has judged inside the longest window a state keeps is one
  // the default branch has carried for that long, and crediting a new real test
  // failure for it now would credit the test with a fix it did not find.
  // `trimWindows` ages these by the same window, and applies it too late to
  // answer this on its own: a fold resolves every observation it reads before
  // it ages anything. A pass that finds only those in its order judges nothing,
  // so it drops them and leaves the other orders waiting.
  const live = inOrder.filter((pending) =>
    daysBetween(pending.day, observation.day) <= LONGEST_WINDOW_DAYS
  );
  if (live.length === 0) {
    state.pendingMain = state.pendingMain.filter((pending) =>
      pending.seed !== observation.seed
    );
    return;
  }
  state.pendingMain = [];
  const here = live.find((pending) => pending.commit === observation.commit);
  if (here !== undefined) {
    // The same commit, passing now and failing before, is the test
    // disagreeing with itself there, wherever that commit sits in the
    // stretch. The two runs can arrive in separate batches, so the
    // same-commit check that catches this within one batch does not see
    // it, and dropping the pending failure silently would lose the flake
    // as well as the new real test failure.
    bump(state.flakesByDay, here.day);
    return;
  }
  // A pass at a point where the test also failed, with no failure there
  // waiting, is one whose failure was read as the test disagreeing with
  // itself and counted then. It ends the stretch as a flake rather than
  // a fix.
  if (disagreed) return;
  // One breakage, one new real test failure. A test that stays red across
  // several runs on the default branch has one thing wrong with it, and the
  // change that makes it green fixed that one thing; crediting every run it
  // failed would make a long outage look like the most valuable test in the
  // repository. The first failure is the one the new real test failure belongs
  // to, since that is where the breakage entered.
  const first = live.reduce((earliest, pending) =>
    pending.day < earliest.day ? pending : earliest
  );
  if (first.crowded === true) return;
  creditNewRealTestFailure(state, "main", first.day, first.source);
}

/**
 * Where an observation was made, as the rules that compare two runs need
 * it: the commit, and the order its tests ran in. A run with no seed ran
 * in declaration order, which is an order of its own, so its point is
 * the bare commit and matches no seeded run's.
 */
function pointOf(observation: Pick<Observation, "commit" | "seed">): string {
  return observation.seed === undefined
    ? observation.commit
    : `${observation.commit}#${observation.seed}`;
}

/**
 * The run an observation belongs to, as the rule about crowds counts
 * one: what one source saw at one point (see `pointOf`), and for a
 * workstation, at one start. A workstation runs its uncommitted changes
 * on top of the commit it names, so two of its runs at one commit ran
 * two different trees. A continuous-integration run is held together by
 * its point and source alone, so its lanes, and a rerun of some of them,
 * count as one run.
 */
function runOf(observation: Observation): string {
  const run = `${pointOf(observation)} ${observation.source}`;
  return observation.place === "local"
    ? `${run} ${observation.startedAt}`
    : run;
}

/**
 * The cross-batch context two of the rules need. A batch cannot be judged
 * on its own: whether an identity disagreed with itself at a commit, and
 * whether a failure spans enough sources to read as the environment, are
 * both questions about observations that may sit in another batch. A
 * caller folding a stream carries one of these along and trims it.
 */
export interface FoldContext {
  /**
   * The outcomes seen at one commit in one order, by identity, with the
   * commit's day. Keyed by that point (see `pointOf`) rather than by the
   * pair of point and identity so that the window below can drop a whole
   * point at once, and so that a point's name is stored once instead of
   * against every identity that ran at it.
   */
  outcomesAtCommit: Map<
    string,
    { day: string; identities: Map<string, Set<string>> }
  >;

  /**
   * The most recently seen points (see `pointOf`), oldest first, at most
   * `FLAKE_COMMIT_REACH` of them. Outcomes are remembered at a point
   * while it is in here, and afterwards only for identities the failure
   * witness still names. A commit run in one order is one point, which is
   * every commit that runs without an override.
   */
  recentCommits: string[];

  /**
   * What the default branch said about one identity at one point (see
   * `pointOf`), with the day. A rerun of a commit can arrive long after the run it
   * repeats, so this outlives the batch: without it, a later pass
   * elsewhere would make that rerun look like the first failure at a
   * commit the branch had already shown broken.
   */
  mainAtCommit: Map<string, { day: string; outcome: "pass" | "fail" }>;

  /**
   * The commit and source pairs a new real test failure has already been
   * credited to, against the day it was credited on. That lets the set be
   * aged like everything else here rather than growing with every new
   * real test failure ever credited.
   */
  credited: Map<string, string>;

  /** Where and when each identity has been seen failing. */
  failures: Map<string, SeenFailure[]>;

  /**
   * The identities each run (see `runOf`) newly broke, with the run's
   * day. A run's objects can arrive in more than one batch, so this
   * outlives the batch and ages with the outcomes at a commit.
   */
  crowds: Map<string, { day: string; identities: Set<string> }>;
}

/** One failure of an identity, as the environmental rule counts it. */
export interface SeenFailure {
  day: string;
  source: string;

  /**
   * Whether this failure was away from the default branch while that branch was
   * already failing this identity, which is also what stops it being a new real
   * test failure. A pull request's run tests the pull request merged into the
   * default branch, and a workstation runs a change on top of one of its
   * commits, so either fails a test the default branch is failing. Such a
   * failure is the default branch's breakage seen again, and the environmental
   * rule leaves it out when it counts sources.
   */
  brokenOnMain?: true;
}

/** A context holding nothing, for a fold with no history behind it. */
export function emptyContext(): FoldContext {
  return {
    outcomesAtCommit: new Map(),
    recentCommits: [],
    mainAtCommit: new Map(),
    credited: new Map(),
    failures: new Map(),
    crowds: new Map(),
  };
}

/** A fold context as it travels between runs. */
export interface StoredFoldContext {
  outcomesAtCommit: Array<
    [string, { day: string; identities: Array<[string, string[]]> }]
  >;
  recentCommits: string[];
  mainAtCommit: Array<[string, { day: string; outcome: "pass" | "fail" }]>;
  credited: Array<[string, string]>;
  failures: Array<[string, SeenFailure[]]>;
  crowds?: Array<[string, { day: string; identities: string[] }]>;
}

/** The outcomes a record may carry, for validating a stored context. */
const OUTCOMES = new Set(["pass", "fail", "skip"]);

/**
 * Whether a value is a "yyyy-mm-dd" day this reader can measure from.
 * The parse alone is not enough: a date the calendar does not have rolls
 * into the next month, so "2026-02-31" parses as March 3rd and would be
 * aged from three days later than it claims. What round-trips is a day.
 */
export function isDay(value: unknown): value is string {
  if (typeof value !== "string" || !/^\d{4}-\d{2}-\d{2}$/.test(value)) {
    return false;
  }
  const at = new Date(`${value}T00:00:00Z`);
  return Number.isFinite(at.getTime()) &&
    at.toISOString().slice(0, 10) === value;
}

/** The context, flattened for the aggregate that carries it. */
export function serializeContext(context: FoldContext): StoredFoldContext {
  return {
    outcomesAtCommit: [...context.outcomesAtCommit].map(([commit, seen]) => [
      commit,
      {
        day: seen.day,
        identities: [...seen.identities].map((
          [key, outcomes],
        ) => [key, [...outcomes]] as [string, string[]]),
      },
    ]),
    recentCommits: [...context.recentCommits],
    mainAtCommit: [...context.mainAtCommit],
    credited: [...context.credited],
    failures: [...context.failures],
    crowds: [...context.crowds].map(([run, crowd]) => [
      run,
      { day: crowd.day, identities: [...crowd.identities] },
    ]),
  };
}

/**
 * The context a previous run left. Anything malformed yields an empty
 * one: a context is an optimization over re-reading, and losing it costs
 * the two cross-run rules their reach rather than any stored fact.
 */
export function parseContext(value: unknown): FoldContext {
  const context = emptyContext();
  if (!isObjectOrArray(value)) return context;
  const stored = value as Partial<StoredFoldContext>;
  const pairs = (raw: unknown): Array<[string, unknown]> =>
    Array.isArray(raw)
      ? raw.filter((entry): entry is [string, unknown] =>
        Array.isArray(entry) && typeof entry[0] === "string"
      )
      : [];

  for (const [commit, seen] of pairs(stored.outcomesAtCommit)) {
    if (!isObjectOrArray(seen)) continue;
    const held = seen as { day?: unknown; identities?: unknown };
    if (!isDay(held.day)) continue;
    const identities = new Map<string, Set<string>>();
    for (const [key, raw] of pairs(held.identities)) {
      if (!Array.isArray(raw)) continue;
      // An outcome this reader does not know would read as one more
      // thing the identity did at that commit, and two of them is the
      // test disagreeing with itself — which suppresses a new real test
      // failure.
      const outcomes = raw.filter((outcome): outcome is string =>
        typeof outcome === "string" && OUTCOMES.has(outcome)
      );
      if (outcomes.length > 0) identities.set(key, new Set(outcomes));
    }
    if (identities.size === 0) continue;
    context.outcomesAtCommit.set(commit, { day: held.day, identities });
  }
  if (Array.isArray(stored.recentCommits)) {
    context.recentCommits = stored.recentCommits
      .filter((commit): commit is string => typeof commit === "string")
      .slice(-FLAKE_COMMIT_REACH);
  }
  for (const [at, seen] of pairs(stored.mainAtCommit)) {
    if (!isObjectOrArray(seen)) continue;
    const held = seen as { day?: unknown; outcome?: unknown };
    if (!isDay(held.day)) continue;
    if (held.outcome !== "pass" && held.outcome !== "fail") continue;
    context.mainAtCommit.set(at, { day: held.day, outcome: held.outcome });
  }
  for (const [attribution, day] of pairs(stored.credited)) {
    // A day that cannot be read cannot be aged, and an entry that is
    // never aged suppresses that new real test failure for good.
    if (isDay(day)) context.credited.set(attribution, day);
  }
  for (const [key, seen] of pairs(stored.failures)) {
    if (!Array.isArray(seen)) continue;
    const kept: SeenFailure[] = [];
    for (const failure of seen) {
      if (!isObjectOrArray(failure)) continue;
      const held = failure as {
        day?: unknown;
        source?: unknown;
        brokenOnMain?: unknown;
      };
      if (!isDay(held.day) || typeof held.source !== "string") continue;
      kept.push({
        day: held.day,
        source: held.source,
        ...(held.brokenOnMain === true ? { brokenOnMain: true as const } : {}),
      });
    }
    if (kept.length > 0) context.failures.set(key, kept);
  }
  for (const [run, crowd] of pairs(stored.crowds)) {
    if (!isObjectOrArray(crowd)) continue;
    const held = crowd as { day?: unknown; identities?: unknown };
    if (!isDay(held.day) || !Array.isArray(held.identities)) continue;
    const identities = held.identities.filter((key): key is string =>
      typeof key === "string"
    );
    context.crowds.set(run, { day: held.day, identities: new Set(identities) });
  }
  return context;
}

/**
 * Drops what the rules can no longer reach, each on the span its own
 * question needs. A stream that never trimmed would grow with the number
 * of runs rather than with the number of tests.
 */
export function trimContext(context: FoldContext, today: string): void {
  const beyondBreadth = (day: string) =>
    daysBetween(day, today) > NEW_REAL_TEST_FAILURE_BREADTH_WINDOW_DAYS;
  for (const [key, seen] of context.failures) {
    const kept = seen.filter((failure) => !beyondBreadth(failure.day));
    if (kept.length === 0) context.failures.delete(key);
    else context.failures.set(key, kept);
  }
  for (const [commit, seen] of context.outcomesAtCommit) {
    if (daysBetween(seen.day, today) > SAME_COMMIT_REACH_DAYS) {
      context.outcomesAtCommit.delete(commit);
    }
  }
  for (const [run, crowd] of context.crowds) {
    if (daysBetween(crowd.day, today) > SAME_COMMIT_REACH_DAYS) {
      context.crowds.delete(run);
    }
  }
  context.recentCommits = context.recentCommits.filter((commit) =>
    context.outcomesAtCommit.has(commit)
  );
  // A commit is re-run within days of the run it repeats, not months, so
  // these age on the same window. Kept forever they would grow with the
  // number of new real test failures the repository has ever had.
  const old = (day: string) => daysBetween(day, today) > COMMIT_REACH_DAYS;
  for (const [at, seen] of context.mainAtCommit) {
    if (old(seen.day)) context.mainAtCommit.delete(at);
  }
  for (const [attribution, day] of context.credited) {
    if (old(day)) context.credited.delete(attribution);
  }
}

/** What a fold may be told beyond the observations themselves. */
export interface FoldOptions {
  /** The state each identity's history had reached before this batch. */
  prior?: Map<string, IdentityState>;

  /** What earlier batches of the same stream saw. */
  context?: FoldContext;

  /**
   * Called with the duration of each passing execution a state counts, by
   * identity key and day, for an observation that carries one.
   */
  sampled?: (key: string, day: string, durationMs: number) => void;
}

/**
 * Folds a batch of observations into per-identity state.
 *
 * The observations must be in ascending time order, because the rules
 * that decide whether a failure is a new real test failure look backwards
 * at what `main` last said and forwards at what it says next.
 * Observations at one commit in one order are considered together: an
 * identity that both passed and failed there disagreed with itself, which
 * is a flake observation and never a new real test failure.
 *
 * Both of those need the whole batch in view before any one observation is
 * judged, so this walks the batch more than once and the iterable has to
 * replay it each time. A one-shot iterator would leave every pass after
 * the first with nothing to read and score the batch as though most of it
 * had never run, so one is refused rather than folded.
 *
 * Returns the state each identity was folded into, which is the map
 * `options.prior` names when a caller carries one across batches.
 */
export function foldObservations(
  observations: Iterable<Observation>,
  options: FoldOptions = {},
): Map<string, IdentityState> {
  // An iterator is its own iterable, which is what tells the two apart.
  if (Object.is(observations[Symbol.iterator](), observations)) {
    throw new Error(
      "foldObservations needs an iterable that replays, not an iterator.",
    );
  }
  const states = options.prior ?? new Map<string, IdentityState>();
  const context = options.context ?? emptyContext();
  const stateOf = (key: string): IdentityState => {
    let state = states.get(key);
    if (state === undefined) {
      state = emptyState();
      states.set(key, state);
    }
    return state;
  };

  // Only the default branch says which tests there are, so a record from
  // anywhere else is evidence about an identity the default branch
  // recorded before it, and about nothing else. This says, observation by
  // observation, which records that leaves: every one from the default
  // branch, and the rest whose identity a state already held or the
  // default branch recorded earlier in the batch. Every pass walks the
  // batch in the same order and asks one of its own.
  const counting = () => {
    const recorded = new Set<string>();
    return (key: string, observation: Observation): boolean => {
      if (observation.place === "main") recorded.add(key);
      return states.has(key) || recorded.has(key);
    };
  };

  // Same-commit disagreement, and the sources a failing identity appeared
  // on inside the environmental window, both need the whole batch in view
  // before any one observation can be judged.
  const outcomesAtCommit = context.outcomesAtCommit;
  const failures = context.failures;
  const recent = context.recentCommits;

  // The failure witness first, so that the pass below can tell a test that
  // has failed from one that never has. What the default branch said at
  // each commit is gathered alongside it: only the classification pass
  // reads that, and it depends on nothing gathered here, so it rides along
  // rather than walking the batch again. Gathering it before anything is
  // judged is what stops a failure elsewhere at one commit being weighed
  // against whatever `main` happened to say last in this batch.
  const mainAtCommit = new Map<string, "pass" | "fail" | "skip">();
  const failedHere = new Set<string>();
  const witnessing = counting();
  for (const observation of observations) {
    const key = testIdentityKey(observation.test);
    const counted = witnessing(key, observation);
    if (observation.outcome === "fail" && counted) failedHere.add(key);
    if (observation.place !== "main" || observation.outcome === "skip") {
      continue;
    }
    const at = `${key} ${pointOf(observation)}`;
    // A failure anywhere at one commit in one order is the commit being
    // broken; a pass beside it does not clear that.
    if (observation.outcome === "fail" || !mainAtCommit.has(at)) {
      mainAtCommit.set(at, observation.outcome);
    }
  }

  // Already broken on the default branch, so a run elsewhere learned
  // nothing about the change in front of it. What that branch says at
  // this very commit outranks what it last said, and is known ahead of
  // time so that the order this batch happened to arrive in cannot decide
  // the verdict.
  const brokenOnMain = (
    key: string,
    observation: Observation,
    lastMain: "pass" | "fail" | undefined,
  ): boolean => {
    const here = mainAtCommit.get(`${key} ${pointOf(observation)}`);
    return here === "fail" || (here === undefined && lastMain === "fail");
  };

  // Outcomes at a commit, for as long as they can still be asked about.
  // Every identity runs at nearly every commit, so keeping all of them
  // costs the corpus times the commits: one measured day is 2.6 million
  // pairs and 442MB, which is more than a string can hold. A commit is
  // remembered whole while it is one of the last `FLAKE_COMMIT_REACH`
  // seen, and once it slides out it keeps only the identities the
  // failure witness names — so a test that has failed is exact until the
  // commit ages out at `SAME_COMMIT_REACH_DAYS`, and one that never has
  // is remembered only as long as its rerun could plausibly still
  // arrive.
  //
  // What each run newly broke rides along: on the default branch, an
  // identity whose previous run there in its order did not fail; anywhere
  // else, one the default branch was not already failing. Whether another rule sets a
  // failure aside needs the whole batch, so `crowded` asks that later.
  // Each failure is also recorded here for `environmental`, which reads
  // the record only after this pass. A failure elsewhere that the default
  // branch was already failing is marked as such.
  //
  // A record `counting` passes over still counts toward its run's crowd,
  // which says how much the run broke rather than anything about one test,
  // and toward nothing else. Its identity reaches neither `failures` nor
  // `outcomesAtCommit`, so no rule sets its failure aside, and a failure
  // the default branch was not already failing is always in its run's
  // crowd.
  const crowds = context.crowds;
  const lastMainOf = new Map<string, MainOutcomes>();
  const hasFailed = (key: string) => failures.has(key) || failedHere.has(key);
  const evidencing = counting();
  for (const observation of observations) {
    const key = testIdentityKey(observation.test);
    const counted = evidencing(key, observation);
    // A skip is a test that deliberately did not run, so it agrees with
    // nothing and contradicts nothing; counting it as disagreement would
    // read a skip beside a failure as the test disagreeing with itself.
    if (observation.outcome === "skip") continue;
    const said = lastMainOf.get(key) ?? states.get(key)?.lastMainByOrder;
    const lastMain = lastMainIn(said, observation.seed);
    if (observation.place === "main") {
      lastMainOf.set(
        key,
        withMainOutcome(said, observation.seed, observation.outcome),
      );
    }
    if (observation.outcome === "fail") {
      const onMain = observation.place === "main";
      const continued = onMain
        ? lastMain === "fail"
        : brokenOnMain(key, observation, lastMain);
      if (counted) {
        const list = failures.get(key) ?? [];
        list.push({
          day: observation.day,
          source: observation.source,
          ...(continued && !onMain ? { brokenOnMain: true as const } : {}),
        });
        failures.set(key, list);
      }
      if (!continued) {
        const run = runOf(observation);
        let crowd = crowds.get(run);
        if (crowd === undefined) {
          crowd = { day: observation.day, identities: new Set() };
          crowds.set(run, crowd);
        }
        crowd.identities.add(key);
      }
    }
    if (!counted) continue;
    const point = pointOf(observation);
    let seen = outcomesAtCommit.get(point);
    if (seen === undefined) {
      seen = { day: observation.day, identities: new Map() };
      outcomesAtCommit.set(point, seen);
      recent.push(point);
      while (recent.length > FLAKE_COMMIT_REACH) {
        const dropped = recent.shift()!;
        const held = outcomesAtCommit.get(dropped);
        if (held === undefined) continue;
        for (const identity of [...held.identities.keys()]) {
          if (!hasFailed(identity)) held.identities.delete(identity);
        }
        if (held.identities.size === 0) outcomesAtCommit.delete(dropped);
      }
    }
    if (!recent.includes(point) && !hasFailed(key)) continue;
    const outcomes = seen.identities.get(key) ?? new Set<string>();
    outcomes.add(observation.outcome);
    seen.identities.set(key, outcomes);
  }

  // It passed and failed at one commit in one order, with nothing between
  // the two runs but chance.
  const disagreed = (key: string, observation: Observation): boolean =>
    (outcomesAtCommit.get(pointOf(observation))?.identities.get(key)?.size ??
      0) > 1;

  // A failure the default branch was already failing is that branch's
  // breakage, counted once through the branch's own source rather than
  // once for every pull request or workstation that carried it.
  const environmental = (key: string, day: string, source: string): boolean => {
    const nearby = new Set<string>([source]);
    for (const failure of failures.get(key) ?? []) {
      if (
        failure.brokenOnMain === true ||
        Math.abs(daysBetween(failure.day, day)) >
          NEW_REAL_TEST_FAILURE_BREADTH_WINDOW_DAYS
      ) {
        continue;
      }
      nearby.add(failure.source);
    }
    return nearby.size >= ENVIRONMENTAL_MIN_SOURCES;
  };

  // A breakage a crowd of tests sees is found by whichever of them a
  // change runs, so it says nothing about any one of them. The crowd is
  // what one run newly broke and no other rule sets aside, counted once
  // per run and batch.
  const crowdSizes = new Map<string, number>();
  const crowded = (observation: Observation): boolean => {
    const run = runOf(observation);
    let size = crowdSizes.get(run);
    if (size === undefined) {
      size = 0;
      for (const key of crowds.get(run)?.identities ?? []) {
        if (
          !disagreed(key, observation) &&
          !environmental(key, observation.day, observation.source)
        ) {
          size++;
        }
      }
      crowdSizes.set(run, size);
    }
    return size >= MASS_FAILURE_MIN_IDENTITIES;
  };

  // A new real test failure is attributed to the pair of the commit and
  // the source that saw it, so re-running one broken commit ten times
  // counts once.
  const credited = context.credited;

  const keeping = counting();
  for (const observation of observations) {
    const key = testIdentityKey(observation.test);
    if (!keeping(key, observation)) continue;
    const state = stateOf(key);
    const day = observation.day;

    // A skip is a test that deliberately did not run. It says nothing
    // about the test and nothing about the change, so it does not reach
    // `lastMainByOrder` either: a test skipped on the default branch has
    // not been shown to be fixed, and the last run that did execute it is
    // the last thing known about it. A failure elsewhere therefore goes
    // on being read as the default branch's, and is not credited to the
    // change in front of it.
    if (observation.outcome === "skip") continue;

    bump(state.runsByDay, day);
    if (observation.durationMs !== undefined) {
      options.sampled?.(key, day, observation.durationMs);
    }
    const lastMain = lastMainIn(state.lastMainByOrder, observation.seed);
    // Whether `main` has run this identity in this order, which decides
    // whether what it last said is about this order or about another.
    const ownOrder = state.lastMainByOrder?.some((held) =>
      held.seed === observation.seed
    ) === true;
    if (observation.place === "main") {
      resolvePendingMain(state, observation, disagreed(key, observation));
      state.lastMainByOrder = withMainOutcome(
        state.lastMainByOrder,
        observation.seed,
        observation.outcome,
      );
    }
    if (observation.outcome === "pass") continue;

    bump(state.failuresByDay, day);

    if (disagreed(key, observation)) {
      bump(state.flakesByDay, day);
      continue;
    }
    if (environmental(key, day, observation.source)) continue;
    if (observation.place === "main") {
      // Whether this was a new real test failure depends on what the next
      // `main` run says, so it waits. A failure continuing a breakage, which
      // `main` has not passed since in this order, is part of that breakage,
      // crowd and all. Where `main` has not run it in this order, what it
      // continues is the breakage in whichever order `main` last ran it. A
      // failure after a pass begins a breakage of its own.
      const waiting = state.pendingMain;
      const continued = lastMain === "fail"
        ? waiting.findLast((pending) =>
          !ownOrder || pending.seed === observation.seed
        )
        : undefined;
      const inCrowd = continued !== undefined
        ? continued.crowded === true
        : crowded(observation);
      waiting.push({
        day,
        commit: observation.commit,
        ...(observation.seed === undefined ? {} : { seed: observation.seed }),
        source: observation.source,
        ...(inCrowd ? { crowded: true as const } : {}),
      });
      continue;
    }
    if (brokenOnMain(key, observation, lastMain) || crowded(observation)) {
      continue;
    }
    const attribution = `${key} ${observation.commit} ${observation.source}`;
    if (credited.has(attribution)) continue;
    credited.set(attribution, day);
    creditNewRealTestFailure(state, observation.place, day, observation.source);
  }

  return states;
}

/**
 * How finely a day's durations are counted: this many buckets to each
 * doubling of a duration. A bucket is read as the largest duration it
 * counts, so a percentile read from buckets is at most one bucket's
 * width, about 2.2%, above the exact one, and never below it.
 */
export const COST_BUCKETS_PER_DOUBLING = 32;

/**
 * The passing executions of one identity on one day, counted by bucket.
 * Counts add, so any number of days, and any parts of one day, combine
 * into exactly the counts one accumulation of all of them would give.
 */
export interface DaySamples {
  /** The bucket `counts[0]` counts. */
  lowest: number;

  /** How many executions each bucket from `lowest` upward counts. */
  counts: number[];

  /**
   * Which set of cost rules measured them, on a day a state holds. A
   * batch on its way into one carries none, since the day it is sealed
   * into is what records the set that sealed it.
   */
  rule?: number;
}

/**
 * Which set of cost rules a day a state holds was sealed under: which
 * executions reach a day's sample. Change it to any other value in the
 * same change that alters that. A change to how a day's sample is stored
 * leaves it alone and reads the days stored before it forward instead,
 * as `readCostsForward()` does. The values are not ordered and nothing
 * but equality is asked of them; a day carrying none was sealed before
 * the stamps began, under the first set.
 */
export const COST_RULE = 4;

/** The largest duration a bucket counts, in milliseconds. */
function bucketBound(bucket: number): number {
  return 2 ** (bucket / COST_BUCKETS_PER_DOUBLING);
}

/**
 * The bucket a duration is counted in: the lowest whose bound is at
 * least the duration. Everything up to a millisecond is counted in
 * bucket zero.
 */
function bucketOf(durationMs: number): number {
  if (!(durationMs > 1)) return 0;
  const bucket = Math.ceil(Math.log2(durationMs) * COST_BUCKETS_PER_DOUBLING);
  // The logarithm can round a duration just past a bound down onto it.
  return bucketBound(bucket) < durationMs ? bucket + 1 : bucket;
}

/** A fresh, empty sample. */
export function emptySamples(): DaySamples {
  return { lowest: 0, counts: [] };
}

/** Adds `count` executions to one bucket of a sample. */
function countInto(samples: DaySamples, bucket: number, count: number): void {
  if (count === 0) return;
  if (samples.counts.length === 0) samples.lowest = bucket;
  if (bucket < samples.lowest) {
    const gap = new Array<number>(samples.lowest - bucket).fill(0);
    samples.counts = [...gap, ...samples.counts];
    samples.lowest = bucket;
  }
  const at = bucket - samples.lowest;
  while (samples.counts.length <= at) samples.counts.push(0);
  samples.counts[at]! += count;
}

/** Counts one duration into a day's sample. */
export function sampleDuration(samples: DaySamples, durationMs: number): void {
  countInto(samples, bucketOf(durationMs), 1);
}

/**
 * The ninetieth percentile of a list of values in ascending order, by
 * nearest rank: the value at that rank rather than one between two.
 */
export function percentile90(sorted: readonly number[]): number {
  if (sorted.length === 0) return 0;
  return sorted[Math.ceil(0.9 * sorted.length) - 1]!;
}

/**
 * The ninetieth percentile of the executions a sample counts, by nearest
 * rank, read as the bound of the bucket that rank falls in.
 */
export function sampledPercentile90(samples: DaySamples): number {
  const count = samples.counts.reduce((sum, each) => sum + each, 0);
  if (count === 0) return 0;
  const rank = Math.ceil(0.9 * count);
  let seen = 0;
  const at = samples.counts.findIndex((each) => (seen += each) >= rank);
  return bucketBound(samples.lowest + at);
}

/** One list of a day's durations, as the day's sample of them. */
export function samplesOf(durationsMs: readonly number[]): DaySamples {
  const samples = emptySamples();
  for (const durationMs of durationsMs) sampleDuration(samples, durationMs);
  return samples;
}

/** Two samples read as one: the counts of both, bucket by bucket. */
export function mergeSamples(a: DaySamples, b: DaySamples): DaySamples {
  const merged = emptySamples();
  for (const part of [a, b]) {
    for (const [at, count] of part.counts.entries()) {
      countInto(merged, part.lowest + at, count);
    }
  }
  return merged;
}

/**
 * Folds a batch of one day's durations into that day's sample.
 *
 * A day is read across as many runs as it takes for its objects to
 * arrive, so this is given part of a day at a time and combines rather
 * than replaces. Combining the samples rather than a figure taken from
 * them is what makes the cost a percentile of every execution it covers:
 * a batch carrying one execution contributes one duration, where a
 * percentile of that batch would be that one duration standing for every
 * execution the day holds.
 */
export function sealDay(
  state: IdentityState,
  day: string,
  batch: DaySamples,
): void {
  if (batch.counts.length === 0) return;
  // A day another set of rules sealed answers only until these rules
  // have sealed one, and this is that sealing, so the rest of what that
  // set left goes here. What is left afterwards is one set's days.
  for (const [sealed, samples] of Object.entries(state.costByDay)) {
    if (samples.rule !== COST_RULE) delete state.costByDay[sealed];
  }
  // The only writer of a day's sample, so what is already there is
  // another sealing of the same day from an earlier run and can be
  // combined with this one. Nothing writes a provisional value alongside
  // it, whose counts would then be added to counts that already include
  // it.
  state.costByDay[day] = {
    ...mergeSamples(state.costByDay[day] ?? emptySamples(), batch),
    rule: COST_RULE,
  };
}

/**
 * A day as it was stored before it was counted by bucket: its slowest
 * durations, or its percentile alone, and how many executions it held.
 * It has no `counts`, which is what tells it from a day counted by
 * bucket.
 */
type StoredSlowest =
  | { slowest: number[]; count: number; rule?: number; counts?: undefined }
  | { p90: number; count: number; rule?: number; counts?: undefined };

/**
 * Reads a state's stored days forward. A day stored as its slowest
 * durations and a count is read as those durations, with each execution
 * that was not kept counted at the smallest that was. A day stored as a
 * percentile and a count is read the same way, as one kept duration.
 * Either reading is at or above the executions it stands for, so a
 * percentile read from it errs high, and the whole day's count stays in
 * it: one execution standing for the day would be outweighed by the
 * first part to arrive after it, which is how a day of slow runs would
 * come to report a fast one. The day keeps the stamp of the rules that
 * sealed it, since those decided which executions it holds, and how they
 * were stored did not.
 */
export function readCostsForward(state: IdentityState): void {
  const held = state.costByDay;
  const days = isObjectNotArray(held) ? held : {};
  state.costByDay = days;
  // A stored day is one shape or another, which the state's own declared
  // type cannot say.
  const read: Record<string, DaySamples | StoredSlowest> = days;
  for (const [day, held] of Object.entries(read)) {
    // A day that is not a record of figures at all is read as a day
    // with nothing in it, which is what a day this cannot make sense of
    // is worth. Ending the read of the whole state is not, and a state
    // is read back through this before anything has looked at what it
    // holds.
    if (!isObjectOrArray(held)) {
      days[day] = emptySamples();
      continue;
    }
    if (held.counts === undefined) days[day] = readSlowest(held);
    else if (!isDaySamples(held)) days[day] = emptySamples();
  }
}

/**
 * Whether a stored day is one counted by bucket that holds figures this can
 * count with: a bucket index and a count per bucket, each a whole number
 * and none negative. A day that does not is read as a day with nothing in
 * it, for the reason a day that is not a record is.
 */
export function isDaySamples(held: unknown): held is DaySamples {
  const whole = (value: unknown) =>
    typeof value === "number" && Number.isInteger(value) && value >= 0;
  return isObjectOrArray(held) && whole(held.lowest) &&
    Array.isArray(held.counts) && held.counts.every(whole);
}

/**
 * One day as earlier rules stored it, as these rules count it. A day
 * whose stored figures are not numbers is read as a day with nothing in
 * it, for the reason a day that is not a record is.
 */
function readSlowest(held: StoredSlowest): DaySamples {
  const kept = "slowest" in held ? held.slowest : [held.p90];
  if (
    !Array.isArray(kept) || kept.length === 0 ||
    !kept.every(Number.isFinite) || !Number.isInteger(held.count) ||
    held.count < kept.length
  ) {
    return emptySamples();
  }
  const samples = samplesOf(kept);
  countInto(
    samples,
    bucketOf(minOf(kept)),
    held.count - kept.length,
  );
  if (held.rule !== undefined) samples.rule = held.rule;
  return samples;
}

/** Ages a state's per-day counters, dropping days past their windows. */
export function trimWindows(state: IdentityState, today: string): void {
  const drop = (counts: Record<string, unknown>, windowDays: number): void => {
    for (const day of Object.keys(counts)) {
      if (daysBetween(day, today) > windowDays) delete counts[day];
    }
  };
  // The run counts answer three questions with two windows: the churn
  // term's denominator, the flake rate's, and how recently `lastRun` saw
  // the test. They are kept for the longer of the two windows, and each
  // reader reads back only as far as its own.
  drop(state.runsByDay, LONGEST_WINDOW_DAYS);
  drop(state.failuresByDay, CHURN_WINDOW_DAYS);
  drop(state.flakesByDay, FLAKE_WINDOW_DAYS);
  drop(state.costByDay, COST_WINDOW_DAYS);
  // A failure on the default branch waits here for a later run to judge
  // it, and a test the branch does not go red for is one no run has to
  // arrive for. So this is aged like everything else, over the longest
  // window a state keeps: a failure nothing has judged in that time is
  // one the branch has carried for that long, and crediting a new real
  // test failure for it afterwards would credit the test with a fix it
  // did not find.
  state.pendingMain = state.pendingMain.filter((pending) =>
    daysBetween(pending.day, today) <= LONGEST_WINDOW_DAYS
  );
}

/**
 * A share of a test's runs over the days inside `windowDays`, each day's
 * counts weighed by `weigh` against how old the day is and how many runs
 * have followed it.
 *
 * Decayed rather than summed flat, because a sum cannot tell two
 * histories apart that a reader would never confuse. A test that
 * disagreed twice and then passed two hundred times has settled; one
 * that passed two hundred times and then disagreed twice has just
 * started. The same counts, and not the same test.
 *
 * Read over a window as well, so what is measured stays bounded. Past
 * four half-lives a day is worth under one part in sixteen, which makes
 * the window a performance choice rather than a policy one.
 */
function decayedShare(
  counted: Record<string, number>,
  state: IdentityState,
  today: string,
  windowDays: number,
  weigh: (ageDays: number, runsSince: number) => number,
): number {
  // Newest day first, so the runs a day has been followed by are known
  // by the time that day is weighed. A day's own runs are weighed
  // together, as though all of them happened at the end of it, which is
  // as fine a grain as counters kept per day can answer at.
  const days = Object.keys(state.runsByDay).sort().reverse();
  let runsSince = 0;
  let top = 0;
  let runs = 0;
  for (const day of days) {
    const age = daysBetween(day, today);
    if (age > windowDays) continue;
    const count = state.runsByDay[day] ?? 0;
    const weight = weigh(age, runsSince);
    runs += count * weight;
    top += (counted[day] ?? 0) * weight;
    runsSince += count;
  }
  return runs === 0 ? 0 : top / runs;
}

/**
 * The churn term: recent failures over recent runs. A ratio over a long
 * undecayed window measures total historical brokenness rather than the
 * current rate, and a week of failures eight months ago would otherwise
 * outrank a test that is failing right now.
 */
export function churn(state: IdentityState, today: string): number {
  return decayedShare(
    state.failuresByDay,
    state,
    today,
    CHURN_WINDOW_DAYS,
    (age) => 0.5 ** (age / CHURN_HALF_LIFE_DAYS),
  );
}

/**
 * The share of a test's runs it was seen disagreeing with itself over.
 *
 * Runs rather than failures in the denominator, because what the rate
 * decides is whether running this test once fails somebody's change for
 * something its author cannot act on, and that is a chance per run. The
 * two differ by everything a test's passes say: a test that failed once
 * in ten thousand runs, and passed on the rerun, has every one of its
 * failures a flake, and a share of failures would read it as wholly
 * unreliable. Counting runs is also what lets an exclusion reverse, since
 * a run that does not disagree lowers the share.
 *
 * A disagreement's weight halves every `FLAKE_HALF_LIFE_RUNS` runs that
 * follow it, so a test that has settled since is not judged as though it
 * had just started. Runs rather than days, because what shows a test has
 * settled is running without disagreeing. A test that has not run has
 * shown nothing, and time alone should not clear it.
 *
 * A lower bound on how often the test fails on its own, because the only
 * spurious failure this can count is one with a pass beside it at the
 * same commit. What raises the bound is repeats, which put several
 * executions at one commit, and which a rising share is what buys.
 *
 * Nothing is charged against the count, and no belief about how tests
 * usually behave survives into it. A disagreement is not a sample from
 * an unknown rate the way a failure is; it is a proof, since a test that
 * is deterministic cannot pass and fail at one commit. Shrinking the
 * share toward zero would be shrinking it toward what the observation
 * has already ruled out. So a test seen twice that disagreed once reads
 * a half, and one that disagreed once in ten thousand runs reads a
 * ten-thousandth: what separates them is how much each has been run, not
 * how much either is believed.
 */
export function flakeRate(state: IdentityState, today: string): number {
  return decayedShare(
    state.flakesByDay,
    state,
    today,
    FLAKE_WINDOW_DAYS,
    (_age, runsSince) => 0.5 ** (runsSince / FLAKE_HALF_LIFE_RUNS),
  );
}

/**
 * What a person is shown beside the share: the disagreements inside the
 * flake window and the runs they were seen among, counted flat. A share
 * cannot be weighed without them, and they are not what the share
 * divides — the share weights recent days more heavily, so a test with
 * these counts reads higher when the disagreements are the recent part
 * of them.
 */
export function flakeCounts(
  state: IdentityState,
  today: string,
): FlakeEvidence {
  let flakes = 0;
  for (const [day, count] of Object.entries(state.flakesByDay)) {
    if (daysBetween(day, today) > FLAKE_WINDOW_DAYS) continue;
    flakes += count;
  }
  let runs = 0;
  for (const [day, count] of Object.entries(state.runsByDay)) {
    if (daysBetween(day, today) > FLAKE_WINDOW_DAYS) continue;
    runs += count;
  }
  return { flakes, runs };
}

/**
 * What one execution of this identity costs, in seconds: the ninetieth
 * percentile of every passing execution inside the cost window, taken
 * together. Each execution counts once, however the days fall, so a slow
 * day raises the cost by as much of the window as it holds rather than
 * setting the cost alone. The ninetieth percentile rather than the mean,
 * because a cost model that under-estimates blows the time budget.
 */
export function costSeconds(state: IdentityState, today: string): number {
  let window = emptySamples();
  for (const [day, samples] of Object.entries(state.costByDay)) {
    if (daysBetween(day, today) > COST_WINDOW_DAYS) continue;
    window = mergeSamples(window, samples);
  }
  return sampledPercentile90(window) / 1000;
}

export type { FlakeEvidence, ScoreInputs };

/** The score's inputs for one identity, as of a given day. */
export function scoreInputs(
  state: IdentityState,
  today: string,
): ScoreInputs {
  const inputs: ScoreInputs = {
    catches: NEW_REAL_TEST_FAILURE_WEIGHT_LOCAL * state.localCatches +
      NEW_REAL_TEST_FAILURE_WEIGHT_PR * state.prCatches +
      NEW_REAL_TEST_FAILURE_WEIGHT_MAIN * state.mainCatches,
    sources: state.sources.length,
    churn: churn(state, today),
  };
  if (state.lastCatch !== undefined) inputs.lastCatch = state.lastCatch;
  return inputs;
}

/**
 * What a test is worth running.
 *
 * The branch for no new real test failures is not decoration. A test with none
 * has no `lastCatch`, so the age of its most recent one does not exist, and
 * multiplying zero by a missing number yields a missing number rather than
 * zero. A missing score sorts unpredictably against real ones, so the branch is
 * written rather than left to the algebra.
 */
export function value(inputs: ScoreInputs, today: string): number {
  let record = 0;
  if (inputs.catches > 0 && inputs.lastCatch !== undefined) {
    const proven = 1 - 0.5 ** (inputs.catches / PROVEN_SATURATION);
    const age = daysBetween(inputs.lastCatch, today);
    const freshness = FRESHNESS_FLOOR +
      (1 - FRESHNESS_FLOOR) * 0.5 ** (age / FRESHNESS_HALF_LIFE_DAYS);
    record = proven * freshness;
  }
  const breadth = 1 - 0.5 ** (inputs.sources / BREADTH_SATURATION);
  return VALUE_FLOOR + WEIGHT_PROVEN * record + WEIGHT_BREADTH * breadth +
    WEIGHT_CHURN * inputs.churn;
}