open-brainy/src/db/factLog.ts

/**
 * @module db/factLog
 * @description The generation FACT LOG — an append-only, CRC-framed record of
 * every committed generation as an AFTER-IMAGE "fact": what each touched
 * entity/relationship BECAME (or a body-less tombstone when it was removed).
 * This is the dual-write half of the log-canonical transition: today the
 * before-image history + canonical tree remain authoritative; the fact log is
 * appended at the same commit points and reconciled to committed truth at
 * open, so consumers (index heals, replays, scans) can read one sequential,
 * self-verifying stream instead of walking the entity tree.
 *
 * ## Wire format (frozen; additive-only within a major)
 *
 * Fact (msgpack, POSITIONAL array — the segment header's formatVersion
 * governs the schema):
 *
 *   fact := [ generation:u64, timestamp:u64, ops, meta|nil, blobHashes|nil ]
 *   op   := [ kind:u8 (0=noun, 1=verb), id:bin16 (raw uuid bytes),
 *             record:[metaLeg, vecLeg] | nil ]        // nil = TOMBSTONE
 *
 * Segment file (`_generations/facts/seg-<firstGeneration, zero-padded 20>.bfl`):
 *
 *   header := magic "BFACTS\0\0" (8B) | formatVersion:u32 LE |
 *             firstGeneration:u64 LE | reserved 12B (ZEROED, verified)
 *   frame  := length:u32 LE | crc32c:u32 LE (of payload) | payload
 *
 * A fact is never split across segments; a torn tail (length overrun or CRC
 * mismatch) terminates that segment's scan — everything before it is intact.
 * Zero-padded names make lexicographic order == generation order.
 *
 * ## Invariant
 *
 * After {@link FactLog.open}, the log contains EXACTLY the committed prefix:
 * facts are appended BEFORE the commit point (inside the same durability
 * window), so a crash can only leave the log AHEAD of committed truth — open
 * truncates any fact beyond the committed generation. Absent generation =
 * never committed; present = committed. A scan can never see an uncommitted
 * fact.
 *
 * The manifest (`_generations/facts/manifest.json`, JSON — forensics stay
 * terminal-readable) is the single source of truth for the segment SET;
 * rotation flips it atomically (write-new → fsync → rename) BEFORE the new
 * tail's first byte exists, so no segment file is ever unaccounted for.
 */
import { encode as defaultEncode, decode as defaultDecode } from '@msgpack/msgpack'
import { crc32c } from '../utils/crc32c.js'
import { prodLog } from '../utils/logger.js'

// Swappable msgpack implementation — defaults to the JS codec; a native
// provider (registered via the plugin registry's 'msgpack' key) may replace
// it. Byte-compatibility is the contract (positional arrays, bin16 ids).
let msgpackEncode: (value: unknown) => Uint8Array = defaultEncode
let msgpackDecode: (bytes: Uint8Array) => unknown = defaultDecode

/** Replace the msgpack encode/decode implementation at runtime. */
export function setFactCodec(impl: {
  encode: (value: unknown) => Uint8Array
  decode: (bytes: Uint8Array) => unknown
}): void {
  msgpackEncode = impl.encode
  msgpackDecode = impl.decode
}

/** Storage-root-relative home of the fact log. */
export const FACTS_PREFIX = '_generations/facts'
/** The facts manifest path (JSON). */
export const FACTS_MANIFEST_PATH = `${FACTS_PREFIX}/manifest.json`
/** Current segment format version (header field; additive-only within a major). */
export const FACTS_FORMAT_VERSION = 1
/** Rotation threshold: seal the tail segment once it exceeds this many bytes. */
const SEGMENT_ROTATE_BYTES = 8 * 1024 * 1024
/** Segment header: magic(8) + formatVersion(4) + firstGeneration(8) + reserved(12). */
const HEADER_BYTES = 32
const MAGIC = new Uint8Array([0x42, 0x46, 0x41, 0x43, 0x54, 0x53, 0x00, 0x00]) // "BFACTS\0\0"
/** Frame prefix: length(4) + crc32c(4). */
const FRAME_PREFIX_BYTES = 8

/** One write inside a fact: what the id became (or a tombstone). */
export interface FactOp {
  kind: 'noun' | 'verb'
  id: string
  /** The AFTER-IMAGE legs, or `null` for a tombstone (the id was removed). */
  record: { metadata: unknown | null; vector: unknown | null } | null
}

/** One committed generation, as scanned back out of the log. */
export interface CommitFact {
  generation: number
  timestamp: number
  ops: FactOp[]
  meta?: Record<string, unknown>
  blobHashes?: string[]
}

/** The telemetry a scan batch carries (frozen shape). */
export interface FactScanBatch {
  facts: CommitFact[]
  firstGeneration: number
  lastGeneration: number
  factCount: number
  byteSize: number
  segmentId: string
}

/**
 * Liveness bound on a scan's FIRST batch (Stage-2 co-freeze, D1 contract):
 * `batches()` must yield its first batch — or fail loudly — within this many
 * ms of the first pull. A backlogged or damaged store may be SLOW, but it may
 * never be SILENT: a consumer awaiting the first batch is otherwise
 * indistinguishable from a wedge (the exact failure shape a production heal
 * hit against a generations-backlogged brain).
 */
export const SCANFACTS_FIRST_BATCH_MS = 10_000

/** The telemetry a scan OPEN returns (frozen shape). */
export interface FactScanHandle {
  headGeneration: number
  segmentCount: number
  approxFactCount: number
  /**
   * Ordered batches; a detected gap aborts LOUDLY, never a silent skip.
   * Liveness contract: the FIRST batch resolves or rejects within
   * {@link SCANFACTS_FIRST_BATCH_MS} of the first pull — never a silent hang.
   */
  batches: () => AsyncGenerator<FactScanBatch>
  /** Close telemetry — the invariant cross-check, valid after iteration ends. */
  summary: () => { factsYielded: number; segmentsRead: number }
}

/** Manifest entry for a sealed segment. */
interface SegmentEntry {
  file: string
  firstGeneration: number
  lastGeneration: number
  facts: number
  bytes: number
}

/** The facts manifest (JSON on disk). */
interface FactsManifest {
  formatVersion: number
  segments: SegmentEntry[]
  /** The append target. Its true content is established by scanning (crash tolerance). */
  tailSegment: string | null
  updatedAt: string
}

/** The narrow byte-level storage surface the fact log rides. */
export interface FactLogStorage {
  appendRawBytes(path: string, bytes: Uint8Array): Promise<void>
  readRawBytes(path: string): Promise<Uint8Array | null>
  writeRawBytes(path: string, bytes: Uint8Array): Promise<void>
  rawByteSize(path: string): Promise<number | null>
  readRawObject(path: string): Promise<any | null>
  writeRawObject(path: string, data: any): Promise<void>
  syncRawObjects(paths: string[]): Promise<void>
  deleteRawObject(path: string): Promise<void>
}

/** True when the storage adapter exposes every primitive the fact log needs. */
export function storageSupportsFactLog(storage: unknown): storage is FactLogStorage {
  const s = storage as Record<string, unknown>
  return (
    typeof s.appendRawBytes === 'function' &&
    typeof s.readRawBytes === 'function' &&
    typeof s.writeRawBytes === 'function' &&
    typeof s.rawByteSize === 'function'
  )
}

/** uuid string → 16 raw bytes (bin16 on the wire). */
function uuidToBytes(id: string): Uint8Array {
  const hex = id.replace(/-/g, '')
  if (hex.length !== 32) {
    // Non-uuid ids (legacy/natural keys) ride as UTF-8 with a length prefix
    // marker impossible for uuids: we refuse instead — the write API has
    // guaranteed uuid ids since 8.0, so anything else is a corruption signal.
    throw new Error(`fact log: id is not a uuid: ${id}`)
  }
  const bytes = new Uint8Array(16)
  for (let i = 0; i < 16; i++) {
    bytes[i] = parseInt(hex.slice(i * 2, i * 2 + 2), 16)
  }
  return bytes
}

/** 16 raw bytes → canonical lowercase uuid string. */
function bytesToUuid(bytes: Uint8Array): string {
  let hex = ''
  for (let i = 0; i < 16; i++) hex += bytes[i].toString(16).padStart(2, '0')
  return `${hex.slice(0, 8)}-${hex.slice(8, 12)}-${hex.slice(12, 16)}-${hex.slice(16, 20)}-${hex.slice(20)}`
}

/** Zero-padded segment filename: lexicographic order == generation order. */
function segmentFileName(firstGeneration: number): string {
  return `seg-${String(firstGeneration).padStart(20, '0')}.bfl`
}

/** Build a segment header. Reserved bytes are ZEROED (and verified on open). */
function buildHeader(firstGeneration: number): Uint8Array {
  const header = new Uint8Array(HEADER_BYTES)
  header.set(MAGIC, 0)
  const view = new DataView(header.buffer)
  view.setUint32(8, FACTS_FORMAT_VERSION, true)
  view.setBigUint64(12, BigInt(firstGeneration), true)
  // bytes 20..31 stay zero (reserved)
  return header
}

/** Encode one fact into a framed record (length + crc32c + msgpack payload). */
function encodeFrame(fact: CommitFact): Uint8Array {
  const payload = msgpackEncode([
    fact.generation,
    fact.timestamp,
    fact.ops.map((op) => [
      op.kind === 'noun' ? 0 : 1,
      uuidToBytes(op.id),
      op.record === null ? null : [op.record.metadata, op.record.vector]
    ]),
    fact.meta ?? null,
    fact.blobHashes && fact.blobHashes.length > 0 ? fact.blobHashes : null
  ])
  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)
  return frame
}

/** Decode one msgpack payload back into a CommitFact. */
function decodeFact(payload: Uint8Array): CommitFact {
  const raw = msgpackDecode(payload) as unknown[]
  const [generation, timestamp, ops, meta, blobHashes] = raw as [
    number,
    number,
    Array<[number, Uint8Array, [unknown, unknown] | null]>,
    Record<string, unknown> | null,
    string[] | null
  ]
  return {
    generation: Number(generation),
    timestamp: Number(timestamp),
    ops: ops.map(([kind, idBytes, record]) => ({
      kind: kind === 0 ? ('noun' as const) : ('verb' as const),
      id: bytesToUuid(idBytes),
      record: record === null ? null : { metadata: record[0] ?? null, vector: record[1] ?? null }
    })),
    ...(meta ? { meta } : {}),
    ...(blobHashes && blobHashes.length > 0 ? { blobHashes } : {})
  }
}

/**
 * Parse a segment's bytes: verify the header, then walk frames until the end
 * or a torn tail (length overrun / CRC mismatch), which terminates the walk —
 * everything before it is intact. Returns the decoded facts plus the byte
 * length of the VALID prefix (header + intact frames), which reconciliation
 * uses to cut a torn tail without re-encoding.
 */
function parseSegment(
  file: string,
  bytes: Uint8Array
): { facts: CommitFact[]; validBytes: number } {
  if (bytes.length < HEADER_BYTES) {
    prodLog.warn(`[FactLog] segment ${file} shorter than its header — treating as empty`)
    return { facts: [], validBytes: 0 }
  }
  for (let i = 0; i < MAGIC.length; i++) {
    if (bytes[i] !== MAGIC[i]) {
      throw new Error(`fact log: segment ${file} has a bad magic — not a fact segment`)
    }
  }
  const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength)
  const version = view.getUint32(8, true)
  if (version !== FACTS_FORMAT_VERSION) {
    throw new Error(
      `fact log: segment ${file} has formatVersion ${version}; this build reads ${FACTS_FORMAT_VERSION}`
    )
  }
  for (let i = 20; i < HEADER_BYTES; i++) {
    if (bytes[i] !== 0) {
      // Non-zero reserved bytes = a future format this build cannot verify.
      throw new Error(`fact log: segment ${file} has non-zero reserved header bytes — unverifiable`)
    }
  }

  const facts: CommitFact[] = []
  let offset = HEADER_BYTES
  while (offset + FRAME_PREFIX_BYTES <= bytes.length) {
    const length = view.getUint32(offset, true)
    const expectedCrc = view.getUint32(offset + 4, true)
    const start = offset + FRAME_PREFIX_BYTES
    const end = start + length
    if (end > bytes.length) break // torn tail: frame length overruns the file
    const payload = bytes.subarray(start, end)
    if (crc32c(payload) !== expectedCrc) break // torn tail: payload CRC mismatch
    facts.push(decodeFact(payload))
    offset = end
  }
  return { facts, validBytes: offset }
}

/**
 * The generation fact log. One instance per open store; every method assumes
 * the single-writer discipline the generation store already enforces (calls
 * arrive under its commit mutex).
 */
export class FactLog {
  private readonly storage: FactLogStorage
  /** Rotation threshold (bytes); tests may lower it to exercise rotation. */
  private readonly rotateBytes: number
  private manifest: FactsManifest = {
    formatVersion: FACTS_FORMAT_VERSION,
    segments: [],
    tailSegment: null,
    updatedAt: new Date(0).toISOString()
  }
  /** Decoded facts of the TAIL segment (bounded by the rotation threshold). */
  private tailFacts: CommitFact[] = []
  /** Byte size of the tail segment file (valid prefix). */
  private tailBytes = 0
  /** Highest generation in the log (0 = empty). */
  private head = 0
  /** Segment paths appended since the last sync (the fsync batch). */
  private readonly dirtySegments = new Set<string>()

  constructor(storage: FactLogStorage, options?: { rotateBytes?: number }) {
    this.storage = storage
    this.rotateBytes = options?.rotateBytes ?? SEGMENT_ROTATE_BYTES
  }

  /** The highest committed generation the log holds (0 = empty). */
  headGeneration(): number {
    return this.head
  }

  /**
   * Open the log and reconcile it to committed truth: read the manifest,
   * establish the tail's intact content (torn-tail scan), then TRUNCATE any
   * fact with `generation > committedGeneration` — those never committed (a
   * crash between fact-append and the commit point). After open, the log is
   * exactly the committed prefix.
   */
  async open(committedGeneration: number): Promise<void> {
    const stored = (await this.storage.readRawObject(FACTS_MANIFEST_PATH)) as FactsManifest | null
    if (stored && typeof stored === 'object' && Array.isArray(stored.segments)) {
      if (stored.formatVersion !== FACTS_FORMAT_VERSION) {
        throw new Error(
          `fact log: manifest formatVersion ${stored.formatVersion}; this build reads ${FACTS_FORMAT_VERSION}`
        )
      }
      this.manifest = stored
    }

    // Drop sealed segments that sit ENTIRELY beyond committed truth (a crash
    // right after a rotation whose facts never committed), newest first.
    while (this.manifest.segments.length > 0) {
      const last = this.manifest.segments[this.manifest.segments.length - 1]
      if (last.firstGeneration > committedGeneration) {
        prodLog.warn(
          `[FactLog] dropping sealed segment ${last.file} (generations ${last.firstGeneration}..` +
            `${last.lastGeneration} never committed)`
        )
        await this.storage.deleteRawObject(`${FACTS_PREFIX}/${last.file}`)
        this.manifest.segments.pop()
        await this.persistManifest()
      } else if (last.lastGeneration > committedGeneration) {
        // A sealed segment STRADDLING committed truth: cut it back.
        await this.truncateSegmentTo(last.file, committedGeneration)
        const cut = await this.reloadSegmentEntry(last.file)
        this.manifest.segments[this.manifest.segments.length - 1] = cut
        await this.persistManifest()
        break
      } else {
        break
      }
    }

    // Establish the tail: scan its intact prefix, then truncate beyond
    // committed truth (the common crash shape: buffered single-op facts whose
    // counter never went durable).
    if (this.manifest.tailSegment) {
      const tailPath = `${FACTS_PREFIX}/${this.manifest.tailSegment}`
      const bytes = await this.storage.readRawBytes(tailPath)
      if (bytes === null) {
        // Manifest named a tail whose first byte never landed — an empty tail.
        this.tailFacts = []
        this.tailBytes = 0
      } else {
        const { facts, validBytes } = parseSegment(this.manifest.tailSegment, bytes)
        const kept = facts.filter((f) => f.generation <= committedGeneration)
        if (kept.length !== facts.length || validBytes !== bytes.length) {
          const dropped = facts.length - kept.length
          if (dropped > 0) {
            prodLog.warn(
              `[FactLog] truncating ${dropped} uncommitted fact(s) beyond generation ` +
                `${committedGeneration} from the tail (never committed)`
            )
          }
          await this.rewriteTail(kept)
        } else {
          this.tailFacts = facts
          this.tailBytes = validBytes
        }
      }
    }

    this.head = this.computeHead()
  }

  /**
   * Append one committed generation's fact. NOT durable until {@link sync} —
   * the caller batches durability at its commit barrier (transact syncs in
   * the same call; Model-B group-commit syncs at flush).
   */
  async append(fact: CommitFact): Promise<void> {
    if (fact.generation <= this.head) {
      throw new Error(
        `fact log: non-monotonic append (generation ${fact.generation} ≤ head ${this.head})`
      )
    }
    if (this.manifest.tailSegment === null) {
      await this.startTail(fact.generation)
    } else if (this.tailBytes >= this.rotateBytes) {
      await this.rotate(fact.generation)
    }
    const frame = encodeFrame(fact)
    const tailPath = `${FACTS_PREFIX}/${this.manifest.tailSegment}`
    await this.storage.appendRawBytes(tailPath, frame)
    this.tailFacts.push(fact)
    this.tailBytes += frame.length
    this.head = fact.generation
    this.dirtySegments.add(tailPath)
  }

  /** Fsync every segment appended since the last sync. */
  async sync(): Promise<void> {
    if (this.dirtySegments.size === 0) return
    const paths = [...this.dirtySegments]
    this.dirtySegments.clear()
    await this.storage.syncRawObjects(paths)
  }

  // --- GROUP COMMIT ON THE LOG (durable-at-ack mode) ------------------------
  // Classic group commit: concurrent writers append, then join ONE fsync
  // whose completion releases every covered ack. Two slots — the running
  // sync and at most one queued behind it — give the covering guarantee:
  // an append followed by ensureSynced() is always covered, because the
  // sync it awaits STARTS after the append landed (a running sync that
  // may have snapshotted earlier is never joined; the queued one is).
  private syncRunning: Promise<void> | null = null
  private syncQueued: Promise<void> | null = null

  /**
   * Await a sync that covers every byte appended before this call. Many
   * concurrent callers share one fsync (solo caller = immediate sync). The
   * durability contract of an acked write in log-durable mode: this promise
   * resolving means the caller's frames survive power loss.
   */
  async ensureSynced(): Promise<void> {
    if (this.syncQueued) {
      // A sync that has NOT started yet exists — it will snapshot after our
      // append, so it covers us.
      return this.syncQueued
    }
    if (this.syncRunning) {
      // The running sync may have snapshotted before our append — queue the
      // next one behind it and join that.
      const queued = this.syncRunning
        .catch(() => {})
        .then(() => {
          // Promote: the queued sync becomes the running one.
          this.syncQueued = null
          this.syncRunning = this.sync().finally(() => {
            this.syncRunning = null
          })
          return this.syncRunning
        })
      this.syncQueued = queued
      return queued
    }
    this.syncRunning = this.sync().finally(() => {
      this.syncRunning = null
    })
    return this.syncRunning
  }

  /**
   * Open a scan over committed facts. The scan runs against a MANIFEST
   * SNAPSHOT (sealed segments + the tail's decoded facts at open) — exactly-
   * once per fact, inclusive bounds, stable under concurrent appends. Gaps
   * abort LOUDLY: a missing generation inside a segment's declared range is
   * corruption, never silently skipped.
   */
  scanFacts(options?: {
    fromGeneration?: number
    toGeneration?: number
    kinds?: Array<'noun' | 'verb'>
    batchSize?: number
    /** Test override for the first-batch liveness bound (default {@link SCANFACTS_FIRST_BATCH_MS}). */
    firstBatchTimeoutMs?: number
  }): FactScanHandle {
    const from = options?.fromGeneration ?? 1
    const to = options?.toGeneration ?? this.head
    const kinds = options?.kinds
    const batchSize = Math.max(1, options?.batchSize ?? 256)

    // Snapshot: the segment list + tail content as of NOW.
    const segments = this.manifest.segments.filter(
      (s) => s.lastGeneration >= from && s.firstGeneration <= to
    )
    const tailSnapshot = this.tailFacts.filter((f) => f.generation >= from && f.generation <= to)
    const tailId = this.manifest.tailSegment ?? 'tail'
    const approxFactCount =
      segments.reduce((sum, s) => sum + s.facts, 0) + tailSnapshot.length

    let factsYielded = 0
    let segmentsRead = 0
    const storage = this.storage

    async function* batches(this: void): AsyncGenerator<FactScanBatch> {
      let expectedNext = 0 // gap detection: generations are monotonic, not necessarily dense
      const emit = (facts: CommitFact[], segmentId: string, byteSize: number): FactScanBatch => ({
        facts,
        firstGeneration: facts[0].generation,
        lastGeneration: facts[facts.length - 1].generation,
        factCount: facts.length,
        byteSize,
        segmentId
      })
      const filterOps = (fact: CommitFact): CommitFact =>
        kinds
          ? { ...fact, ops: fact.ops.filter((op) => kinds.includes(op.kind)) }
          : fact

      for (const entry of segments) {
        const bytes = await storage.readRawBytes(`${FACTS_PREFIX}/${entry.file}`)
        if (bytes === null) {
          throw new Error(
            `fact log: sealed segment ${entry.file} is MISSING — the log is damaged; aborting scan`
          )
        }
        const { facts } = parseSegment(entry.file, bytes)
        segmentsRead++
        const inRange = facts.filter((f) => f.generation >= from && f.generation <= to)
        for (const f of inRange) {
          if (f.generation <= expectedNext - 1) {
            throw new Error(`fact log: out-of-order fact ${f.generation} in ${entry.file} — aborting scan`)
          }
          expectedNext = f.generation + 1
        }
        for (let i = 0; i < inRange.length; i += batchSize) {
          const slice = inRange.slice(i, i + batchSize).map(filterOps)
          if (slice.length === 0) continue
          factsYielded += slice.length
          yield emit(slice, entry.file, slice.reduce((n, f) => n + encodeFrame(f).length, 0))
        }
      }

      if (tailSnapshot.length > 0) {
        segmentsRead++
        for (const f of tailSnapshot) {
          if (f.generation <= expectedNext - 1) {
            throw new Error(`fact log: out-of-order fact ${f.generation} in the tail — aborting scan`)
          }
          expectedNext = f.generation + 1
        }
        for (let i = 0; i < tailSnapshot.length; i += batchSize) {
          const slice = tailSnapshot.slice(i, i + batchSize).map(filterOps)
          factsYielded += slice.length
          yield emit(slice, tailId, slice.reduce((n, f) => n + encodeFrame(f).length, 0))
        }
      }
    }

    // Liveness wrapper: the FIRST pull races the contract deadline. Only the
    // first — the bound is time-to-first-batch (proof the producer is alive),
    // not per-batch pacing; and it runs only while a pull is actually pending,
    // so consumer think-time between pulls never counts against the producer.
    const firstBatchTimeoutMs = options?.firstBatchTimeoutMs ?? SCANFACTS_FIRST_BATCH_MS
    async function* batchesWithLiveness(this: void): AsyncGenerator<FactScanBatch> {
      const inner = batches()
      let timer: NodeJS.Timeout | undefined
      try {
        const deadline = new Promise<never>((_, reject) => {
          timer = setTimeout(
            () =>
              reject(
                new Error(
                  `fact log: scanFacts produced no first batch within ${firstBatchTimeoutMs}ms ` +
                    `(liveness contract) — the store is wedged or unreadably slow; aborting scan LOUDLY ` +
                    `instead of hanging the consumer.`
                )
              ),
            firstBatchTimeoutMs
          )
          timer.unref?.()
        })
        const first = await Promise.race([inner.next(), deadline])
        if (first.done) return
        yield first.value
      } finally {
        clearTimeout(timer)
      }
      yield* inner
    }

    return {
      headGeneration: this.head,
      segmentCount: segments.length + (tailSnapshot.length > 0 ? 1 : 0),
      approxFactCount,
      batches: batchesWithLiveness,
      summary: () => ({ factsYielded, segmentsRead })
    }
  }

  /**
   * The mmap fast path (capability handoff): the immutable sealed segment
   * files covering `fromGeneration`, in order. The TAIL is deliberately NOT
   * included — it is append-mutable; consumers read it via {@link scanFacts}.
   */
  segmentPaths(options?: { fromGeneration?: number }): string[] {
    const from = options?.fromGeneration ?? 1
    return this.manifest.segments
      .filter((s) => s.lastGeneration >= from)
      .map((s) => `${FACTS_PREFIX}/${s.file}`)
  }

  /**
   * Drop every fact with `generation > keepThrough` — the in-session abort
   * compensation: a transact appends its fact BEFORE the commit point, so a
   * real (non-crash) abort after the append must take the fact back out. The
   * dropped facts can only live in the TAIL (they were just appended); the
   * rewrite is atomic and bounded by the rotation threshold.
   */
  async dropAbove(keepThrough: number): Promise<void> {
    if (this.head <= keepThrough) return
    const kept = this.tailFacts.filter((f) => f.generation <= keepThrough)
    if (kept.length === this.tailFacts.length) {
      throw new Error(
        `fact log: dropAbove(${keepThrough}) found no droppable facts in the tail ` +
          `(head ${this.head}) — the fact to drop was already sealed; the log needs reopen`
      )
    }
    await this.rewriteTail(kept)
    this.head = this.computeHead()
  }

  // -- internals -------------------------------------------------------------

  private computeHead(): number {
    if (this.tailFacts.length > 0) return this.tailFacts[this.tailFacts.length - 1].generation
    const sealed = this.manifest.segments
    if (sealed.length > 0) return sealed[sealed.length - 1].lastGeneration
    return 0
  }

  /** Create the very first tail segment (manifest-first, then header bytes). */
  private async startTail(firstGeneration: number): Promise<void> {
    const file = segmentFileName(firstGeneration)
    this.manifest.tailSegment = file
    await this.persistManifest()
    await this.storage.appendRawBytes(`${FACTS_PREFIX}/${file}`, buildHeader(firstGeneration))
    this.tailFacts = []
    this.tailBytes = HEADER_BYTES
  }

  /**
   * Seal the tail into the manifest and start a new one. Manifest-first: the
   * flip both seals the old tail AND names the new one atomically, so no
   * segment file ever exists unaccounted for.
   */
  private async rotate(nextGeneration: number): Promise<void> {
    const sealedFile = this.manifest.tailSegment
    if (!sealedFile) return
    // Seal what the tail actually holds.
    await this.sync() // sealed segments are always fully durable
    const entry: SegmentEntry = {
      file: sealedFile,
      firstGeneration: this.tailFacts[0]?.generation ?? nextGeneration,
      lastGeneration: this.tailFacts[this.tailFacts.length - 1]?.generation ?? nextGeneration - 1,
      facts: this.tailFacts.length,
      bytes: this.tailBytes
    }
    const newFile = segmentFileName(nextGeneration)
    this.manifest.segments.push(entry)
    this.manifest.tailSegment = newFile
    await this.persistManifest()
    await this.storage.appendRawBytes(`${FACTS_PREFIX}/${newFile}`, buildHeader(nextGeneration))
    this.tailFacts = []
    this.tailBytes = HEADER_BYTES
  }

  /** Atomically persist the manifest (write-new → fsync → rename downstream). */
  private async persistManifest(): Promise<void> {
    this.manifest.updatedAt = new Date().toISOString()
    await this.storage.writeRawObject(FACTS_MANIFEST_PATH, this.manifest)
    await this.storage.syncRawObjects([FACTS_MANIFEST_PATH])
  }

  /** Rewrite the tail segment to hold exactly `facts` (atomic replace). */
  private async rewriteTail(facts: CommitFact[]): Promise<void> {
    const file = this.manifest.tailSegment
    if (!file) return
    const first = facts[0]?.generation ?? this.segmentFirstGenerationFromName(file)
    const parts: Uint8Array[] = [buildHeader(first)]
    for (const f of facts) parts.push(encodeFrame(f))
    const total = parts.reduce((n, p) => n + p.length, 0)
    const merged = new Uint8Array(total)
    let offset = 0
    for (const p of parts) {
      merged.set(p, offset)
      offset += p.length
    }
    await this.storage.writeRawBytes(`${FACTS_PREFIX}/${file}`, merged)
    this.tailFacts = facts
    this.tailBytes = total
  }

  /** Cut a SEALED segment back to `committedGeneration` (atomic replace). */
  private async truncateSegmentTo(file: string, committedGeneration: number): Promise<void> {
    const path = `${FACTS_PREFIX}/${file}`
    const bytes = await this.storage.readRawBytes(path)
    if (bytes === null) return
    const { facts } = parseSegment(file, bytes)
    const kept = facts.filter((f) => f.generation <= committedGeneration)
    prodLog.warn(
      `[FactLog] truncating sealed segment ${file} to generation ${committedGeneration} ` +
        `(${facts.length - kept.length} uncommitted fact(s) dropped)`
    )
    const first = kept[0]?.generation ?? this.segmentFirstGenerationFromName(file)
    const parts: Uint8Array[] = [buildHeader(first)]
    for (const f of kept) parts.push(encodeFrame(f))
    const total = parts.reduce((n, p) => n + p.length, 0)
    const merged = new Uint8Array(total)
    let offset = 0
    for (const p of parts) {
      merged.set(p, offset)
      offset += p.length
    }
    await this.storage.writeRawBytes(path, merged)
  }

  /** Re-derive a sealed segment's manifest entry from its actual bytes. */
  private async reloadSegmentEntry(file: string): Promise<SegmentEntry> {
    const bytes = await this.storage.readRawBytes(`${FACTS_PREFIX}/${file}`)
    const { facts, validBytes } = bytes
      ? parseSegment(file, bytes)
      : { facts: [] as CommitFact[], validBytes: 0 }
    return {
      file,
      firstGeneration: facts[0]?.generation ?? this.segmentFirstGenerationFromName(file),
      lastGeneration: facts[facts.length - 1]?.generation ?? 0,
      facts: facts.length,
      bytes: validBytes
    }
  }

  /** Parse the zero-padded firstGeneration back out of a segment filename. */
  private segmentFirstGenerationFromName(file: string): number {
    const match = /^seg-(\d{20})\.bfl$/.exec(file)
    return match ? Number(match[1]) : 0
  }
}