open-brainy/src/db/generationSegments.ts

/**
 * @module db/generationSegments
 * @description The generation-segment store — Stage-2 D1+D3+repacking's file
 * format (co-frozen 2026-07-19; design: the d1-d3-repacking spec).
 *
 * Packs CONSECUTIVE cold generations' record-sets (before-images + delta)
 * into append-once segment files with derived sidecar indexes, so history
 * scales in SEGMENTS (tens) instead of FILES-PER-GENERATION (hundreds of
 * thousands), and cold-open reads ONE manifest instead of listing the
 * backlog. Layout under `_generations/segments/`:
 *
 * - `seg-<firstGen, zero-padded 20>.bgs` — magic "BGS1", then one frame per
 *   generation: `u32 payloadLen | u32 crc32c | msgpack payload`. Payload is
 *   POSITIONAL: `[generation, timestamp, delta, records[], flags]` with
 *   records `[kindByte, id, record]`. `flags` reserves encoding evolution
 *   (bit 0 = compressed payload — v1 always 0; a future writer upgrade,
 *   never a format break). Sealed segments are IMMUTABLE — the fact log's
 *   own law, generalized.
 * - `seg-<firstGen>.idx` — DERIVED sidecar (msgpack): per-generation frame
 *   offsets (point reads = one ranged read, never a listing) + per-id
 *   generation postings (per-id chain rebuilds read only what they need).
 *   Corrupt/missing → rebuilt from its segment in one sequential read,
 *   loudly.
 * - `manifest.json` — the segment catalogue + `compactedBelow` (D3's
 *   horizon marker). Cold-open reads THIS; the packed backlog is never
 *   listed.
 *
 * D3 semantics carried here: bounded-retention reclaim drops WHOLE segments
 * at boundaries (O(1) per segment, no rewrite); under the archival profile
 * (`retention: 'all'`) nothing here is ever dropped — folding is the only
 * transform (re-representation, never deletion).
 */

import { encode as msgpackEncode, decode as msgpackDecode } from '@msgpack/msgpack'
import { crc32c } from '../utils/crc32c.js'
import type { FactLogStorage } from './factLog.js'
import { prodLog } from '../utils/logger.js'

/** Directory for segment files + manifest, under the generations prefix. */
export const SEGMENTS_PREFIX = '_generations/segments'

/** Target sealed-segment size (co-freeze proposal; tunable on evidence). */
export const SEGMENT_TARGET_BYTES = 64 * 1024 * 1024

const MAGIC = new TextEncoder().encode('BGS1')
const FRAME_PREFIX_BYTES = 8 // u32 payloadLen + u32 crc32c
const MANIFEST_PATH = `${SEGMENTS_PREFIX}/manifest.json`

/** One generation's fold input — exactly what the live tier holds for it. */
export interface FoldGeneration {
  generation: number
  timestamp: number
  /** The tx.json delta object, carried verbatim. */
  delta: unknown
  /** The before-image record-set (empty for record-less generations). */
  records: Array<{ kind: 'noun' | 'verb'; id: string; record: unknown }>
}

/** Manifest entry for one sealed segment. */
export interface SegmentMeta {
  file: string
  firstGeneration: number
  lastGeneration: number
  frames: number
  bytes: number
  /** crc32c of the full segment byte stream — the digest chain's link. */
  checksum: number
}

interface SegmentManifest {
  version: 1
  compactedBelow: number
  segments: SegmentMeta[]
}

interface SidecarIndex {
  version: 1
  /** [generation, frameOffset, frameLen] ascending by generation. */
  generations: Array<[number, number, number]>
  /** `${kindByte}:${id}` → ascending generations holding a record for it. */
  ids: Record<string, number[]>
}

const segmentFileName = (firstGeneration: number): string =>
  `seg-${String(firstGeneration).padStart(20, '0')}.bgs`
const sidecarFileName = (firstGeneration: number): string =>
  `seg-${String(firstGeneration).padStart(20, '0')}.idx`

/**
 * The generation-segment store. Owns the packed tier ONLY — the live
 * per-generation tier and the routing between tiers belong to
 * `GenerationStore`. All mutating entry points here are called under the
 * generation store's commit mutex.
 */
export class GenerationSegmentStore {
  private readonly storage: FactLogStorage
  private manifest: SegmentManifest = { version: 1, compactedBelow: 0, segments: [] }
  /** Sidecar cache — segments are immutable, so entries never invalidate. */
  private readonly sidecars = new Map<string, SidecarIndex>()

  constructor(storage: FactLogStorage) {
    this.storage = storage
  }

  /** Load the manifest (ONE read — never a directory listing). */
  async open(): Promise<void> {
    const raw = (await this.storage.readRawObject(MANIFEST_PATH)) as SegmentManifest | null
    if (raw) {
      if (raw.version !== 1) {
        throw new Error(
          `[GenerationSegments] manifest version ${String(raw.version)} is newer than this ` +
            `engine understands — refusing to serve partial history. Upgrade the engine.`
        )
      }
      this.manifest = raw
    }
  }

  /** The packed tier's catalogue (ascending, immutable snapshot). */
  segments(): readonly SegmentMeta[] {
    return this.manifest.segments
  }

  /** D3's horizon marker: generations below this were reclaimed (bounded profiles only). */
  compactedBelow(): number {
    return this.manifest.compactedBelow
  }

  /** The covering sealed segment for `gen`, or null if it lives outside the packed tier. */
  private coveringSegment(gen: number): SegmentMeta | null {
    // Manifest is ascending and ranges never overlap — binary search.
    const segs = this.manifest.segments
    let lo = 0
    let hi = segs.length - 1
    while (lo <= hi) {
      const mid = (lo + hi) >> 1
      const s = segs[mid]
      if (gen < s.firstGeneration) hi = mid - 1
      else if (gen > s.lastGeneration) lo = mid + 1
      else return s
    }
    return null
  }

  /** True when `gen` is packed (readable from this tier). */
  hasGeneration(gen: number): boolean {
    return this.coveringSegment(gen) !== null
  }

  /**
   * @description True when `meta` declares more generations than it holds
   * frames — a segment sealed by a writer that folded across a hole. The
   * manifest records `frames` at fold time, so this is an O(1) comparison
   * against the declared span and needs no I/O.
   */
  private isSparse(meta: SegmentMeta): boolean {
    return meta.lastGeneration - meta.firstGeneration + 1 !== meta.frames
  }

  /**
   * @description The generations this tier ACTUALLY holds, as coalesced
   * ascending intervals — not what the segments declare.
   *
   * Dense segments (every one a current writer produces) contribute their
   * declared range with no I/O. A SPARSE segment — one sealed before the
   * density law was enforced, whose declared range spans generations it has
   * no frame for — has its real generation list read from its sidecar and
   * contributed instead, with the discrepancy narrated once.
   *
   * This is what keeps a store that already carries the damage from wedging.
   * `open()` seeds `committedRanges` from these intervals, so a hole is never
   * re-admitted as a committed generation, and the auto-compaction pass that
   * used to fail on every run with "packed history is damaged" simply never
   * asks for the missing frame.
   *
   * @returns Ascending, non-overlapping `[first, last]` intervals.
   */
  async actualRanges(): Promise<Array<[number, number]>> {
    const out: Array<[number, number]> = []
    for (const meta of this.manifest.segments) {
      if (!this.isSparse(meta)) {
        out.push([meta.firstGeneration, meta.lastGeneration])
        continue
      }
      const missing = meta.lastGeneration - meta.firstGeneration + 1 - meta.frames
      prodLog.warn(
        `[GenerationSegments] sealed segment ${meta.file} declares generations ` +
          `${meta.firstGeneration}..${meta.lastGeneration} but holds only ${meta.frames} ` +
          `frame(s) — ${missing} generation(s) in that span were never folded into it. ` +
          `Serving the frames it actually holds; the declared span is not treated as ` +
          `committed history. (Written by a pre-density-law writer that folded across a ` +
          `gap; the segment itself is intact and no record is lost.)`
      )
      const idx = await this.sidecarFor(meta)
      for (const [gen] of idx.generations) {
        const last = out[out.length - 1]
        if (last !== undefined && gen === last[1] + 1) last[1] = gen
        else out.push([gen, gen])
      }
    }
    return out
  }

  /**
   * Fold consecutive generations into ONE new sealed segment + sidecar and
   * append it to the manifest atomically. Caller guarantees: `gens` is
   * ascending, contiguous with the packed tier (first = last packed + 1 when
   * segments exist), and already durable in the live tier. Crash between the
   * segment write and the caller's live-tier delete leaves a DUPLICATE
   * representation — resolved live-tier-wins by the reader; never a gap.
   */
  async fold(gens: FoldGeneration[]): Promise<SegmentMeta> {
    if (gens.length === 0) {
      throw new Error('[GenerationSegments] fold() requires at least one generation')
    }
    for (let i = 1; i < gens.length; i++) {
      if (gens[i].generation <= gens[i - 1].generation) {
        throw new Error('[GenerationSegments] fold() input must be strictly ascending')
      }
    }
    // THE DENSITY LAW, MADE MECHANICAL.
    //
    // A sealed segment declares a CONTIGUOUS range [firstGeneration,
    // lastGeneration] and every reader treats that range as containment:
    // `coveringSegment` is an interval test, `hasGeneration` returns true for
    // anything inside it, and `open()` seeds committedRanges from it. So a
    // segment folded from a SPARSE input silently claims generations it does
    // not hold, and the first read of one of those holes throws
    // "inside sealed segment ... but has no frame — packed history is damaged".
    //
    // That is exactly how the damage was produced. `repackHistory` skipped
    // generations mid-batch — ones absent from committedRanges, ones still in
    // the pending buffer, ones whose tx.json would not read — and handed the
    // survivors here, where the range was computed from the first and last of
    // them. Worse, the mis-declared range was then merged back into
    // committedRanges at the next open, which is what turned a quiet hole into
    // a repeating auto-compaction failure on every subsequent run.
    //
    // Callers now split at discontinuities; this refusal is what keeps any
    // future caller from reintroducing the class. A refusal here loses
    // nothing — the generations stay in the live tier, readable, and the next
    // pass folds them correctly.
    for (let i = 1; i < gens.length; i++) {
      if (gens[i].generation !== gens[i - 1].generation + 1) {
        throw new Error(
          `[GenerationSegments] fold() input is not contiguous: ${gens[i - 1].generation} → ` +
            `${gens[i].generation} skips ${gens[i].generation - gens[i - 1].generation - 1} ` +
            `generation(s). A sealed segment declares a dense range, so folding a sparse ` +
            `batch would claim generations it does not hold. Split the batch at the gap.`
        )
      }
    }
    const last = this.manifest.segments[this.manifest.segments.length - 1]
    if (last && gens[0].generation <= last.lastGeneration) {
      throw new Error(
        `[GenerationSegments] fold() overlaps the packed tier: ${gens[0].generation} ≤ ` +
          `sealed ${last.lastGeneration} — segments are immutable, never rewritten`
      )
    }

    const first = gens[0].generation
    const file = segmentFileName(first)
    const sidecar: SidecarIndex = { version: 1, generations: [], ids: {} }

    // Encode all frames, tracking offsets for the sidecar.
    const parts: Uint8Array[] = [MAGIC]
    let offset = MAGIC.length
    for (const g of gens) {
      const payload = msgpackEncode([
        g.generation,
        g.timestamp,
        g.delta,
        g.records.map((r) => [r.kind === 'noun' ? 0 : 1, r.id, r.record]),
        0 // flags: v1 = uncompressed
      ])
      const frame = new Uint8Array(FRAME_PREFIX_BYTES + payload.length)
      const view = new DataView(frame.buffer)
      view.setUint32(0, payload.length, true)
      view.setUint32(4, crc32c(payload), true)
      frame.set(payload, FRAME_PREFIX_BYTES)
      sidecar.generations.push([g.generation, offset, frame.length])
      for (const r of g.records) {
        const key = `${r.kind === 'noun' ? 0 : 1}:${r.id}`
        ;(sidecar.ids[key] ??= []).push(g.generation)
      }
      parts.push(frame)
      offset += frame.length
    }
    const total = parts.reduce((n, p) => n + p.length, 0)
    const bytes = new Uint8Array(total)
    let at = 0
    for (const p of parts) {
      bytes.set(p, at)
      at += p.length
    }

    const meta: SegmentMeta = {
      file,
      firstGeneration: first,
      lastGeneration: gens[gens.length - 1].generation,
      frames: gens.length,
      bytes: total,
      checksum: crc32c(bytes)
    }

    // Durability order: segment + sidecar fsync'd BEFORE the manifest names
    // them (a crash before the manifest = invisible orphan files, harmless);
    // manifest last, atomically.
    const segPath = `${SEGMENTS_PREFIX}/${file}`
    const idxPath = `${SEGMENTS_PREFIX}/${sidecarFileName(first)}`
    await this.storage.writeRawBytes(segPath, bytes)
    await this.storage.writeRawBytes(idxPath, msgpackEncode(sidecar))
    await this.storage.syncRawObjects([segPath, idxPath])
    const next: SegmentManifest = {
      ...this.manifest,
      segments: [...this.manifest.segments, meta]
    }
    await this.storage.writeRawObject(MANIFEST_PATH, next)
    await this.storage.syncRawObjects([MANIFEST_PATH])
    this.manifest = next
    this.sidecars.set(file, sidecar)
    return meta
  }

  /** Load (or rebuild, loudly) a segment's sidecar. */
  private async sidecarFor(meta: SegmentMeta): Promise<SidecarIndex> {
    const cached = this.sidecars.get(meta.file)
    if (cached) return cached
    const idxPath = `${SEGMENTS_PREFIX}/${sidecarFileName(meta.firstGeneration)}`
    const raw = await this.storage.readRawBytes(idxPath)
    if (raw) {
      try {
        const idx = msgpackDecode(raw) as SidecarIndex
        if (idx.version === 1) {
          this.sidecars.set(meta.file, idx)
          return idx
        }
      } catch {
        // fall through to rebuild
      }
    }
    // Sidecars are DERIVED: rebuild from the segment, loudly — never serve
    // wrong offsets silently.
    prodLog.warn(
      `[GenerationSegments] sidecar for ${meta.file} missing or unreadable — rebuilding from the segment`
    )
    const rebuilt = await this.rebuildSidecar(meta)
    await this.storage.writeRawBytes(idxPath, msgpackEncode(rebuilt))
    this.sidecars.set(meta.file, rebuilt)
    return rebuilt
  }

  /** One sequential read of the segment → a fresh sidecar. Verifies every frame CRC. */
  private async rebuildSidecar(meta: SegmentMeta): Promise<SidecarIndex> {
    const frames = await this.readAllFrames(meta)
    const idx: SidecarIndex = { version: 1, generations: [], ids: {} }
    for (const f of frames) {
      idx.generations.push([f.generation, f.offset, f.frameLen])
      for (const r of f.records) {
        const key = `${r.kind === 'noun' ? 0 : 1}:${r.id}`
        ;(idx.ids[key] ??= []).push(f.generation)
      }
    }
    return idx
  }

  private decodeFrame(
    payload: Uint8Array
  ): { generation: number; timestamp: number; delta: unknown; records: FoldGeneration['records'] } {
    const [generation, timestamp, delta, rawRecords] = msgpackDecode(payload) as [
      number,
      number,
      unknown,
      Array<[number, string, unknown]>,
      number
    ]
    return {
      generation,
      timestamp,
      delta,
      records: rawRecords.map(([kindByte, id, record]) => ({
        kind: kindByte === 0 ? ('noun' as const) : ('verb' as const),
        id,
        record
      }))
    }
  }

  private async readAllFrames(meta: SegmentMeta): Promise<
    Array<ReturnType<GenerationSegmentStore['decodeFrame']> & { offset: number; frameLen: number }>
  > {
    const bytes = await this.storage.readRawBytes(`${SEGMENTS_PREFIX}/${meta.file}`)
    if (!bytes) {
      throw new Error(
        `[GenerationSegments] sealed segment ${meta.file} is MISSING — packed history is damaged; ` +
          `refusing to continue silently`
      )
    }
    const out: Array<ReturnType<GenerationSegmentStore['decodeFrame']> & { offset: number; frameLen: number }> = []
    let at = MAGIC.length
    const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
    while (at + FRAME_PREFIX_BYTES <= bytes.length) {
      const payloadLen = view.getUint32(at, true)
      const crc = view.getUint32(at + 4, true)
      const payload = bytes.subarray(at + FRAME_PREFIX_BYTES, at + FRAME_PREFIX_BYTES + payloadLen)
      if (payload.length !== payloadLen || crc32c(payload) !== crc) {
        throw new Error(
          `[GenerationSegments] frame CRC mismatch in ${meta.file} at offset ${at} — ` +
            `packed history is damaged; refusing to serve it`
        )
      }
      out.push({ ...this.decodeFrame(payload), offset: at, frameLen: FRAME_PREFIX_BYTES + payloadLen })
      at += FRAME_PREFIX_BYTES + payloadLen
    }
    return out
  }

  /** Read one packed generation's frame via its sidecar offset (one ranged read). */
  private async readFrame(
    gen: number
  ): Promise<ReturnType<GenerationSegmentStore['decodeFrame']> | null> {
    const meta = this.coveringSegment(gen)
    if (!meta) return null
    const idx = await this.sidecarFor(meta)
    // generations ascending → binary search.
    const gens = idx.generations
    let lo = 0
    let hi = gens.length - 1
    while (lo <= hi) {
      const mid = (lo + hi) >> 1
      if (gens[mid][0] < gen) lo = mid + 1
      else if (gens[mid][0] > gen) hi = mid - 1
      else {
        const [, offset, frameLen] = gens[mid]
        const bytes = await this.storage.readRawBytes(`${SEGMENTS_PREFIX}/${meta.file}`)
        if (!bytes) {
          throw new Error(`[GenerationSegments] sealed segment ${meta.file} is MISSING`)
        }
        const frame = bytes.subarray(offset, offset + frameLen)
        const view = new DataView(frame.buffer, frame.byteOffset, frame.byteLength)
        const payloadLen = view.getUint32(0, true)
        const crc = view.getUint32(4, true)
        const payload = frame.subarray(FRAME_PREFIX_BYTES, FRAME_PREFIX_BYTES + payloadLen)
        if (payload.length !== payloadLen || crc32c(payload) !== crc) {
          throw new Error(
            `[GenerationSegments] frame CRC mismatch for generation ${gen} in ${meta.file} — ` +
              `packed history is damaged; refusing to serve it`
          )
        }
        return this.decodeFrame(payload)
      }
    }
    // Inside the covering range but with no frame. Two very different causes,
    // and conflating them is what made this class wedge every maintenance pass
    // on the affected stores.
    //
    // (1) A SPARSE SEGMENT — the manifest's own `frames` count is smaller than
    //     the span it declares. That segment was sealed by a writer that
    //     folded across a hole (the class this file's density law now bars).
    //     The segment is INTACT and nothing is lost; it simply never held this
    //     generation. Answering "not packed" is the honest answer, and it lets
    //     the caller's two-tier read decide what a genuinely absent generation
    //     means, instead of every compaction pass dying on a repeating throw.
    //     `actualRanges()` keeps such holes out of committedRanges at open, so
    //     in a healed store nobody asks this question in the first place.
    //
    // (2) A DENSE SEGMENT missing a frame it says it has — the manifest and
    //     the sidecar disagree about a segment that claims to be complete.
    //     That IS damage, and it stays loud.
    if (this.isSparse(meta)) {
      prodLog.warn(
        `[GenerationSegments] generation ${gen} falls inside sealed segment ${meta.file}'s ` +
          `declared range ${meta.firstGeneration}..${meta.lastGeneration}, but that segment ` +
          `holds ${meta.frames} frame(s) for a ${meta.lastGeneration - meta.firstGeneration + 1}` +
          `-generation span — it was sealed across a gap and never held this generation. ` +
          `Reporting it as unpacked rather than as damage; no record is lost.`
      )
      return null
    }
    throw new Error(
      `[GenerationSegments] generation ${gen} is inside sealed segment ${meta.file}'s declared ` +
        `range but has no frame, and that segment declares a complete ${meta.frames}-frame ` +
        `span — the manifest and the sidecar disagree; packed history is damaged`
    )
  }

  /** The packed tier's delta for `gen` (null = not packed). */
  async readDelta(gen: number): Promise<{ delta: unknown; timestamp: number } | null> {
    const frame = await this.readFrame(gen)
    return frame ? { delta: frame.delta, timestamp: frame.timestamp } : null
  }

  /** The packed tier's full record-set for `gen` (null = not packed). */
  async readRecords(gen: number): Promise<FoldGeneration['records'] | null> {
    const frame = await this.readFrame(gen)
    return frame ? frame.records : null
  }

  /** One packed before-image (null = not packed OR no record for the id in that generation). */
  async readRecord(gen: number, kind: 'noun' | 'verb', id: string): Promise<unknown | null> {
    const frame = await this.readFrame(gen)
    if (!frame) return null
    const hit = frame.records.find((r) => r.kind === kind && r.id === id)
    return hit ? hit.record : null
  }

  /**
   * D3 reclaim: drop WHOLE segments whose lastGeneration < `belowGeneration`
   * and bump `compactedBelow`. Partial segments are never dropped — the
   * boundary waits. NEVER called under the archival profile (the caller
   * enforces retention semantics; this method only executes boundary drops).
   */
  async dropSegmentsBelow(belowGeneration: number): Promise<{ dropped: number; compactedBelow: number }> {
    const keep: SegmentMeta[] = []
    const drop: SegmentMeta[] = []
    for (const s of this.manifest.segments) {
      ;(s.lastGeneration < belowGeneration ? drop : keep).push(s)
    }
    if (drop.length === 0) {
      return { dropped: 0, compactedBelow: this.manifest.compactedBelow }
    }
    const compactedBelow = Math.max(
      this.manifest.compactedBelow,
      drop[drop.length - 1].lastGeneration + 1
    )
    // Manifest first (the drop is authoritative once named), then bytes —
    // a crash between leaves orphan segment files invisible to the manifest,
    // harmless and re-collectable.
    const next: SegmentManifest = { ...this.manifest, compactedBelow, segments: keep }
    await this.storage.writeRawObject(MANIFEST_PATH, next)
    await this.storage.syncRawObjects([MANIFEST_PATH])
    this.manifest = next
    for (const s of drop) {
      await this.storage.deleteRawObject(`${SEGMENTS_PREFIX}/${s.file}`)
      await this.storage.deleteRawObject(`${SEGMENTS_PREFIX}/${sidecarFileName(s.firstGeneration)}`)
      this.sidecars.delete(s.file)
    }
    return { dropped: drop.length, compactedBelow }
  }

  /**
   * D8 rider — the packed portion of `generationDigest(g)`: a deterministic
   * crc32c chain over sealed-segment checksums fully below `g`, plus the
   * frame CRC of `g`'s own frame when `g` is mid-segment. O(segments), not
   * O(generations); identical history ⇒ identical digest on any machine.
   * The live-tier portion is composed by the caller.
   */
  async digestThroughPacked(g: number): Promise<number | null> {
    let digest = 0
    let covered = false
    for (const s of this.manifest.segments) {
      if (s.lastGeneration <= g) {
        digest = crc32c(new TextEncoder().encode(`${digest}:${s.checksum}`))
        if (s.lastGeneration === g) covered = true
      } else if (s.firstGeneration <= g) {
        // g is mid-segment: chain the partial prefix via g's frame CRC.
        const frame = await this.readFrame(g)
        if (frame === null) return null
        const idx = await this.sidecarFor(s)
        const upTo = idx.generations.filter(([gen]) => gen <= g)
        for (const [gen, offset, frameLen] of upTo) {
          digest = crc32c(new TextEncoder().encode(`${digest}:${gen}:${offset}:${frameLen}`))
        }
        covered = true
        break
      }
    }
    return covered || this.manifest.segments.length > 0 ? digest : null
  }
}