/**
 * # bus/read — the crash-tolerant, schema-validating bus reader (RUNTIME §1, §8)
 *
 * `readBus` turns bus text (or a bus file) into a stream of typed {@link BusEvent}s. It is
 * **crash-tolerant** exactly where a single-writer append-only log can tear: the very last
 * line, if the process died mid-write, is an incomplete JSON record with no trailing
 * newline — that torn final line is **dropped silently**. Any OTHER unparseable line is
 * **mid-file corruption**: it throws loudly with the 1-based line number (RUNTIME §1). A
 * `META` header with an unrecognized `bus_schema` is refused loudly (fail-closed, §8).
 *
 * The distinction is precise: a well-formed writer always ends a record with `\n`
 * ({@link ../bus/write.ts serializeEvent}), so a file that ends WITHOUT a newline has a
 * partially-written tail. Only that tail may be dropped; an interior blank/garbage line
 * means the log itself is damaged and the reader must not paper over it.
 *
 * Two **forward-compatible** tolerances keep an older build reading a newer bus without crashing
 * (fail-closed to a logged skip, never a false success, §8): (1) the META header may carry any
 * schema in {@link SUPPORTED_BUS_SCHEMAS} — a v1 bus (no JOURNAL) reads clean under a v2 reader;
 * (2) a well-formed {@link JournalEvent JOURNAL} with an unrecognized `kind` is **skipped with a
 * logged reason** ({@link BusReadOptions.onSkip}) rather than thrown — its `seq` is still consumed
 * so the log stays gap-free. Neither tolerance touches the engine: JOURNAL is author metadata.
 *
 * Validation is **fail-closed and complete** (RUNTIME §1), in two layers. Per-record
 * ({@link validate}): finite `seq`/`ts`, a legal `stream`, and a `type` that actually pairs
 * with that stream on the discriminated union (`isValidStreamType` — a `TICK` is `SPOT | BOOK
 * | HEARTBEAT`, an `ORDER` a `place | cancel | fill | reject`, …); META always carries the
 * supported `bus_schema`. Cross-record ({@link readBusText}): a well-formed bus opens with
 * **exactly one META** header (first event, no duplicate) and `seq` is **strictly monotonic,
 * gap-free from 0**. Every violation is a loud, line-located {@link BusReadError} — the reader
 * never widens a malformed record onto {@link BusEvent} by an unchecked cast.
 */

import { readFileSync } from "node:fs";

import type { BusEvent, DeliberationEvent, JournalEvent, MetaEvent, WakeEvent } from "./types.ts";
import {
  DELIBERATION_TYPE,
  isStream,
  isValidStreamType,
  JOURNAL_KINDS,
  SUPPORTED_BUS_SCHEMAS,
} from "./types.ts";

/**
 * A sink for a crash-tolerant SKIP: a record that is well-formed but **forward-compatible-
 * unknown** — today only a {@link JournalEvent JOURNAL} whose `kind` this build does not
 * recognize. The record is dropped from the yielded stream and its reason logged; it is never
 * thrown (fail-closed to a logged skip, RUNTIME §8) and its `seq` is still consumed so the log
 * stays gap-free. Defaults to a `console.warn` when the caller supplies none.
 */
export type BusSkipSink = (reason: string, line: number) => void;

/** Options for the bus reader. */
export interface BusReadOptions {
  /** Where a crash-tolerant SKIP is reported (see {@link BusSkipSink}). */
  readonly onSkip?: BusSkipSink;
}

const defaultSkipSink: BusSkipSink = (reason, line) => {
  console.warn(`bus skip at line ${line}: ${reason}`);
};

/** A loud, located bus corruption error (mid-file damage or a schema violation). */
export class BusReadError extends Error {
  readonly line: number;
  constructor(message: string, line: number) {
    super(`bus corruption at line ${line}: ${message}`);
    this.name = "BusReadError";
    this.line = line;
  }
}

/** True when `source` is bus *content* rather than a *path*. A JSONL bus always begins with
 * `{`; a filesystem path never does. This lets one entry point accept either (RUNTIME's
 * "path or string") without an out-of-band flag. */
function looksLikeContent(source: string): boolean {
  return source.trimStart().startsWith("{");
}

/**
 * Validate one event's **intra-record** well-formedness and (for META) its schema, then narrow
 * the record onto the {@link BusEvent} union. Throws {@link BusReadError} on any structural
 * violation — a record that does not inhabit the union is corruption (RUNTIME §1/§8), not a
 * thing to skip silently.
 *
 * The narrowing is **earned**, not asserted: by the time we cast, `seq`/`ts` are finite
 * numbers, `stream` is a legal {@link Stream}, `type` is a string that pairs legally with that
 * stream ({@link isValidStreamType}), and a META carries the supported `bus_schema`. Cross-
 * record invariants (seq monotonicity, exactly-one/META-first) are the reader loop's job — a
 * single record cannot see them — and live in {@link readBusText}.
 */
function validate(obj: unknown, lineNo: number): BusEvent {
  if (typeof obj !== "object" || obj === null || Array.isArray(obj)) {
    throw new BusReadError("event is not a JSON object", lineNo);
  }
  const e = obj as Record<string, unknown>;
  if (typeof e["seq"] !== "number" || !Number.isFinite(e["seq"])) {
    throw new BusReadError("missing/invalid numeric `seq`", lineNo);
  }
  if (typeof e["ts"] !== "number" || !Number.isFinite(e["ts"])) {
    throw new BusReadError("missing/invalid numeric `ts`", lineNo);
  }
  const stream = e["stream"];
  if (!isStream(stream)) {
    throw new BusReadError(`unknown stream ${JSON.stringify(stream)}`, lineNo);
  }
  // JOURNAL is the one stream with no engine `type` — its discriminant is `kind`. Validate it on
  // `kind`/`body` directly (both strings; a non-string is genuine corruption and throws) and
  // return. Whether the `kind` is *recognized* is a separate, softer question the reader loop
  // handles by a logged skip, not a throw — a well-formed-but-unknown journal is forward-compat,
  // not damage (RUNTIME §8).
  if (stream === "JOURNAL") {
    if (typeof e["kind"] !== "string") {
      throw new BusReadError("JOURNAL missing/invalid string `kind`", lineNo);
    }
    if (typeof e["body"] !== "string") {
      throw new BusReadError("JOURNAL missing/invalid string `body`", lineNo);
    }
    return obj as BusEvent;
  }

  const type = e["type"];
  if (typeof type !== "string") {
    throw new BusReadError("missing/invalid string `type`", lineNo);
  }
  if (!isValidStreamType(stream, type)) {
    throw new BusReadError(
      `invalid (stream,type) pairing ${JSON.stringify(stream)}/${JSON.stringify(type)}`,
      lineNo,
    );
  }
  if (stream === "META" && !SUPPORTED_BUS_SCHEMAS.has(e["bus_schema"] as number)) {
    throw new BusReadError(
      `unsupported bus_schema ${JSON.stringify(e["bus_schema"])} (reader supports ${[...SUPPORTED_BUS_SCHEMAS].join(", ")})`,
      lineNo,
    );
  }
  // The v6 deliberation record's INTRA-record shape (clock-honest wakes §1): integer `wake_seq`,
  // non-negative integer costs. The CROSS-record facts — the bus is stamped ≥ v6, `wake_seq` names an
  // earlier WAKE checkpoint, and `ts === checkpoint.ts + measured_ms + buffer_ms` — need state a single
  // record cannot see, so they live in the {@link readBusText} loop (which already walks with state).
  if (stream === "WAKE" && type === DELIBERATION_TYPE) {
    const intField = (name: "wake_seq" | "measured_ms" | "buffer_ms"): number => {
      const v = e[name];
      if (typeof v !== "number" || !Number.isInteger(v)) {
        throw new BusReadError(`deliberation record: missing/non-integer \`${name}\``, lineNo);
      }
      return v;
    };
    if (intField("wake_seq") < 0) throw new BusReadError("deliberation record: negative `wake_seq`", lineNo);
    if (intField("measured_ms") < 0) throw new BusReadError("deliberation record: negative `measured_ms`", lineNo);
    if (intField("buffer_ms") < 0) throw new BusReadError("deliberation record: negative `buffer_ms`", lineNo);
  }
  return obj as BusEvent;
}

/**
 * Read typed events from bus **text** (already-loaded JSONL). Torn final line dropped silently;
 * interior corruption throws loudly with the line number.
 *
 * Beyond per-record validation ({@link validate}), this loop enforces the **cross-record**
 * well-formedness a single record cannot see (RUNTIME §1): a well-formed bus opens with
 * **exactly one META** — the header must be the **first** event and no second META may appear —
 * and `seq` is **strictly monotonic, gap-free from 0** (`0, 1, 2, …`). A backward, skipped, or
 * teleported `seq`, a missing/late header, or a duplicate header is mid-file corruption and
 * raises loudly. (A torn final line is still dropped first — it never reaches these checks.)
 */
export function* readBusText(text: string, opts?: BusReadOptions): Generator<BusEvent> {
  if (text.length === 0) return;
  const onSkip = opts?.onSkip ?? defaultSkipSink;
  const endsWithNewline = text.endsWith("\n");
  const segments = text.split("\n");
  // A trailing "\n" yields a final empty segment that is not a line — drop it.
  if (endsWithNewline) segments.pop();

  let expectedSeq = 0;
  let sawMeta = false;
  // Cross-record state for the v6 deliberation identity (clock-honest wakes §1): the header's
  // declared schema, and every WAKE checkpoint's `seq → ts` so a deliberation record's `wake_seq`
  // join and its `ts = checkpoint.ts + measured_ms + buffer_ms` identity verify against the record
  // it names. Scoped record-vs-checkpoint, never record-vs-landing (the driver owns landing).
  let metaSchema = 0;
  const checkpointTsBySeq = new Map<number, number>();

  for (let i = 0; i < segments.length; i++) {
    const raw = segments[i] ?? "";
    const lineNo = i + 1;
    const isFinalSegment = i === segments.length - 1;
    const torniable = isFinalSegment && !endsWithNewline;

    if (raw.trim() === "") {
      // A blank tail with no newline is a torn write — drop it. A blank interior line is
      // damage in an append-only log — that is loud.
      if (torniable) return;
      throw new BusReadError("unexpected blank line", lineNo);
    }

    let parsed: unknown;
    try {
      parsed = JSON.parse(raw);
    } catch {
      // Only the unterminated final line may be a torn write; anything else is corruption.
      if (torniable) return;
      throw new BusReadError("unparseable JSON", lineNo);
    }
    const event = validate(parsed, lineNo);

    // Cross-record well-formedness (RUNTIME §1): exactly one META, header-first.
    if (!sawMeta) {
      if (event.stream !== "META") {
        throw new BusReadError(
          "first event must be the META session header (a well-formed bus opens with exactly one META)",
          lineNo,
        );
      }
      sawMeta = true;
      metaSchema = (event as MetaEvent).bus_schema;
    } else if (event.stream === "META") {
      throw new BusReadError(
        "a second META event (a well-formed bus opens with exactly one META)",
        lineNo,
      );
    }

    // Cross-record well-formedness (RUNTIME §1): seq strictly monotonic, gap-free from 0.
    if (event.seq !== expectedSeq) {
      throw new BusReadError(
        `non-monotonic seq: expected ${expectedSeq}, got ${event.seq} (seq must be gap-free from 0)`,
        lineNo,
      );
    }
    expectedSeq += 1;

    // The v6 deliberation cross-record identity (clock-honest wakes §1; RUNTIME §8). This check lives
    // HERE — beside the wake_seq join it needs — not in per-record `validate()`, which has no META
    // context. Fail-closed on every leg: a deliberation record on a pre-v6 bus is a corrupt bus; a
    // dangling `wake_seq` (naming no EARLIER WAKE checkpoint) is corrupt; a `ts` off the
    // `checkpoint.ts + measured_ms + buffer_ms` identity is corrupt. Integer-ms addition only.
    if (event.stream === "WAKE") {
      if (event.type === "wake" && (event as WakeEvent).wake === "checkpoint") {
        checkpointTsBySeq.set(event.seq, event.ts);
      } else if (event.type === DELIBERATION_TYPE) {
        if (metaSchema < 6) {
          throw new BusReadError(
            `a deliberation record on a bus stamped bus_schema ${metaSchema} — a clocked record on a pre-clocked (<6) bus is corrupt (clock-honest wakes §1)`,
            lineNo,
          );
        }
        const d = event as DeliberationEvent;
        const cpTs = checkpointTsBySeq.get(d.wake_seq);
        if (cpTs === undefined) {
          throw new BusReadError(
            `deliberation record: dangling wake_seq ${d.wake_seq} — it must name an EARLIER WAKE checkpoint on this bus`,
            lineNo,
          );
        }
        if (d.ts !== cpTs + d.measured_ms + d.buffer_ms) {
          throw new BusReadError(
            `deliberation record: ts ${d.ts} breaks the return-time identity checkpoint.ts + measured_ms + buffer_ms = ${cpTs + d.measured_ms + d.buffer_ms}`,
            lineNo,
          );
        }
      }
    }

    // Crash-tolerant forward-compat (RUNTIME §8): a well-formed JOURNAL with an unrecognized
    // `kind` is author metadata a newer writer emitted — drop it with a logged reason rather than
    // throwing. Its `seq` is already consumed above, so the gap-free invariant survives the skip,
    // and because JOURNAL is never an engine input the emitted stream is unaffected either way.
    if (event.stream === "JOURNAL" && !JOURNAL_KINDS.has((event as JournalEvent).kind)) {
      onSkip(`unknown JOURNAL kind ${JSON.stringify((event as JournalEvent).kind)}`, lineNo);
      continue;
    }

    yield event;
  }
}

/** Read typed events from a bus **file** (synchronous, deterministic — no wall clock). */
export function* readBusFile(path: string, opts?: BusReadOptions): Generator<BusEvent> {
  yield* readBusText(readFileSync(path, "utf8"), opts);
}

/**
 * Read typed events from a bus given **either a path or raw JSONL content** (RUNTIME §1).
 * Content is detected structurally (a JSONL bus begins with `{`); anything else is read as a
 * filesystem path. Crash-tolerant and schema-validating per {@link readBusText}.
 */
export function* readBus(source: string, opts?: BusReadOptions): Generator<BusEvent> {
  if (looksLikeContent(source)) {
    yield* readBusText(source, opts);
  } else {
    yield* readBusFile(source, opts);
  }
}