wal.ts

/**
 * The write-ahead log format: one durable record per acknowledged mutation.
 *
 * Phase 0 made each full snapshot crash-safe, but a snapshot is written on a
 * timer and rewrites the whole index, so an insert acknowledged to the caller
 * could still be lost if the tab died before the next commit. The WAL closes
 * that window: every insert/remove is appended (and, in strict mode, flushed)
 * as one small framed record *before* its promise resolves, and the full
 * snapshot becomes a periodic checkpoint that the log is replayed on top of.
 *
 * ## `wal.bin` layout (FVW2, little-endian, except the header magic)
 * ```
 * header, 26 + modelLen bytes
 *   0   4  magic "FVW2"            (raw bytes, so a hex dump reads "FVW2")
 *   4   2  u16 version = 2
 *   6   2  u16 flags = 0           (reserved)
 *   8   8  u64 baseSequence        (FVS2 snapshot sequence this log applies on top of; 0 = none)
 *   16  4  u32 dims
 *   20  2  u16 modelLen            (≤ 256: WAL_MAX_MODEL_LEN)
 *   22  n  model id UTF-8, modelLen bytes (the embedding model; empty = unknown)
 *   22+n 4 u32 headerCrc = crc32(bytes[0..22+n))
 * records, repeated until EOF
 *   u32 payloadLen
 *   u32 payloadCrc = crc32(payload)
 *   payload:
 *     u8  kind                     1 = insert (upsert), 2 = remove
 *     u16 idLen, id UTF-8
 *     insert only: f32[dims] vector, u32 textLen, text UTF-8
 * ```
 *
 * The model id lets a store whose data lives only in the log (killed before
 * its first checkpoint) refuse to be reopened with a different embedding model
 * of the same width, exactly as an FVS2 snapshot's meta does. The model id is
 * bounded to 256 UTF-8 bytes, so a header is never longer than 282 bytes
 * (`WAL_MAX_HEADER_LEN`): a `modelLen` above the bound is corrupt, never "a
 * header whose write was cut short", and a file longer than any header cannot
 * be a torn header write (see `resolveWal` in `persistence.ts`). FVW1 (the 24-byte
 * header without a model id) never shipped and is rejected as `bad-header`.
 *
 * ## Torn tails are normal
 * A tab killed mid-append leaves a partial last record. {@link scanWal} treats
 * the first record that overruns the file, fails its CRC, or does not parse as
 * the end of the log: everything before it is returned, nothing after it is
 * trusted, and the caller truncates the file back to the last good record. A
 * record is therefore applied whole or not at all — replay never sees a torn
 * record. Only a bad *header* is an error ({@link WalCorruptError}), because
 * then the log's base, geometry and model are unknowable.
 *
 * Pure: no DOM, no Worker, no wasm.
 */

import { crc32 } from './crc32.ts';
import { WalCorruptError } from './errors.ts';

/** One logged mutation. `insert` is an upsert, exactly like `Engine.insert`. */
export type WalRecord =
  | { kind: 'insert'; id: string; vector: Float32Array; text: string }
  | { kind: 'remove'; id: string };

/** The decoded log header. */
export interface WalHeader {
  /** The FVS2 snapshot sequence the records apply on top of (`0n` = no snapshot). */
  baseSequence: bigint;
  /** Vector dimensionality of every insert record. */
  dims: number;
  /** Embedding model id the records were embedded with (`''` = unknown). */
  model: string;
  /** Size of the header on disk, in bytes (where the first record starts). */
  length: number;
}

/** Result of {@link scanWal}. */
export interface WalScan {
  /** Every complete, checksummed record, in log order. */
  records: WalRecord[];
  /** Byte offset just past the last good record (≥ the header length). */
  validLength: number;
  /** `true` when bytes follow {@link validLength} (a torn or corrupt tail). */
  truncatedTail: boolean;
}

/** Size of the smallest possible log header (empty model id), in bytes. */
export const WAL_MIN_HEADER_LEN = 26;
/** Longest model id a log header may carry, in UTF-8 bytes. */
export const WAL_MAX_MODEL_LEN = 256;
/** Size of the largest possible log header (a {@link WAL_MAX_MODEL_LEN}-byte model id), in bytes. */
export const WAL_MAX_HEADER_LEN = WAL_MIN_HEADER_LEN + WAL_MAX_MODEL_LEN;
/** Offset of the u16 `modelLen` field (the fixed part of the header ends here). */
const MODEL_LEN_OFFSET = 20;
/** Offset of the model id bytes. */
const MODEL_OFFSET = 22;
/** Size of a record's `payloadLen` + `payloadCrc` prefix, in bytes. */
const RECORD_PREFIX_LEN = 8;

/** `"FVW2"` as raw bytes. */
const MAGIC = [0x46, 0x56, 0x57, 0x32] as const;
const VERSION = 2;
const KIND_INSERT = 1;
const KIND_REMOVE = 2;
const MAX_ID_BYTES = 0xffff;

const encoder = new TextEncoder();
/** Fatal, so a malformed UTF-8 payload is a stop condition rather than U+FFFD. */
const decoder = new TextDecoder('utf-8', { fatal: true });

/** Encode the log header. Throws if the model id exceeds {@link WAL_MAX_MODEL_LEN} UTF-8 bytes. */
export function encodeWalHeader(header: { baseSequence: bigint; dims: number; model: string }): Uint8Array {
  const model = encoder.encode(header.model);
  if (model.length > WAL_MAX_MODEL_LEN) {
    throw new RangeError(
      `[ferrovec] model id is ${model.length} UTF-8 bytes; the log header allows at most ${WAL_MAX_MODEL_LEN}`,
    );
  }
  const crcAt = MODEL_OFFSET + model.length;
  const bytes = new Uint8Array(crcAt + 4);
  const view = new DataView(bytes.buffer);
  bytes.set(MAGIC, 0);
  view.setUint16(4, VERSION, true);
  view.setUint16(6, 0, true);
  view.setBigUint64(8, header.baseSequence, true);
  view.setUint32(16, header.dims, true);
  view.setUint16(MODEL_LEN_OFFSET, model.length, true);
  bytes.set(model, MODEL_OFFSET);
  view.setUint32(crcAt, crc32(bytes.subarray(0, crcAt)), true);
  return bytes;
}

/**
 * The header length a log *claims*, from its magic, version and `modelLen`
 * alone (no checksum), or `null` if `bytes` does not start like an FVW2 header
 * or claims a `modelLen` above {@link WAL_MAX_MODEL_LEN} (corrupt, never a cut
 * write). Lets open tell a header whose write was cut short (the file ends
 * inside the declared header: nothing was ever logged after it) from a corrupt
 * one.
 */
export function declaredWalHeaderLength(bytes: Uint8Array): number | null {
  if (bytes.length < MODEL_OFFSET) return null;
  for (let i = 0; i < MAGIC.length; i++) if (bytes[i] !== MAGIC[i]) return null;
  const view = new DataView(bytes.buffer, bytes.byteOffset, MODEL_OFFSET);
  if (view.getUint16(4, true) !== VERSION) return null;
  const modelLen = view.getUint16(MODEL_LEN_OFFSET, true);
  if (modelLen > WAL_MAX_MODEL_LEN) return null;
  return MODEL_OFFSET + modelLen + 4;
}

/** Decode and validate the log header. Throws `WalCorruptError('bad-header')`. */
export function decodeWalHeader(bytes: Uint8Array): WalHeader {
  if (bytes.length < WAL_MIN_HEADER_LEN) {
    throw new WalCorruptError('bad-header', `short header (${bytes.length} bytes)`);
  }
  for (let i = 0; i < MAGIC.length; i++) {
    if (bytes[i] !== MAGIC[i]) throw new WalCorruptError('bad-header', 'bad magic');
  }
  const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
  const version = view.getUint16(4, true);
  if (version !== VERSION) throw new WalCorruptError('bad-header', `unsupported version ${version}`);
  const modelLen = view.getUint16(MODEL_LEN_OFFSET, true);
  if (modelLen > WAL_MAX_MODEL_LEN) {
    throw new WalCorruptError('bad-header', `model id length ${modelLen} exceeds ${WAL_MAX_MODEL_LEN}`);
  }
  const crcAt = MODEL_OFFSET + modelLen;
  if (bytes.length < crcAt + 4) {
    throw new WalCorruptError('bad-header', `short header (${bytes.length} of ${crcAt + 4} bytes)`);
  }
  if (view.getUint32(crcAt, true) !== crc32(bytes.subarray(0, crcAt))) {
    throw new WalCorruptError('bad-header', 'header checksum mismatch');
  }
  let model: string;
  try {
    model = decoder.decode(bytes.subarray(MODEL_OFFSET, crcAt));
  } catch {
    throw new WalCorruptError('bad-header', 'model id is not valid UTF-8');
  }
  return {
    baseSequence: view.getBigUint64(8, true),
    dims: view.getUint32(16, true),
    model,
    length: crcAt + 4,
  };
}

/**
 * Encode one record as `payloadLen ‖ payloadCrc ‖ payload`. Throws if an insert
 * vector's length is not `dims`, or an id exceeds 65535 UTF-8 bytes.
 */
export function encodeWalRecord(rec: WalRecord, dims: number): Uint8Array {
  const id = encoder.encode(rec.id);
  if (id.length > MAX_ID_BYTES) {
    throw new RangeError(`[ferrovec] id is ${id.length} UTF-8 bytes; the log allows at most ${MAX_ID_BYTES}`);
  }
  let payloadLen = 1 + 2 + id.length;
  let text: Uint8Array | null = null;
  if (rec.kind === 'insert') {
    if (rec.vector.length !== dims) {
      throw new RangeError(
        `[ferrovec] cannot log a ${rec.vector.length}-dim vector into a ${dims}-dim write-ahead log`,
      );
    }
    text = encoder.encode(rec.text);
    payloadLen += dims * 4 + 4 + text.length;
  }

  const out = new Uint8Array(RECORD_PREFIX_LEN + payloadLen);
  const view = new DataView(out.buffer);
  let off = RECORD_PREFIX_LEN;
  view.setUint8(off, rec.kind === 'insert' ? KIND_INSERT : KIND_REMOVE);
  off += 1;
  view.setUint16(off, id.length, true);
  off += 2;
  out.set(id, off);
  off += id.length;
  if (rec.kind === 'insert' && text) {
    for (let i = 0; i < dims; i++, off += 4) view.setFloat32(off, rec.vector[i]!, true);
    view.setUint32(off, text.length, true);
    off += 4;
    out.set(text, off);
  }
  view.setUint32(0, payloadLen, true);
  view.setUint32(4, crc32(out.subarray(RECORD_PREFIX_LEN)), true);
  return out;
}

/** Decode one checksum-verified payload, or `null` if it is malformed. */
function decodePayload(payload: Uint8Array, dims: number): WalRecord | null {
  const view = new DataView(payload.buffer, payload.byteOffset, payload.byteLength);
  if (payload.length < 3) return null;
  const kind = view.getUint8(0);
  const idLen = view.getUint16(1, true);
  let off = 3;
  if (off + idLen > payload.length) return null;
  let id: string;
  try {
    id = decoder.decode(payload.subarray(off, off + idLen));
  } catch {
    return null;
  }
  off += idLen;
  if (kind === KIND_REMOVE) {
    return off === payload.length ? { kind: 'remove', id } : null;
  }
  if (kind !== KIND_INSERT) return null;
  if (off + dims * 4 + 4 > payload.length) return null;
  const vector = new Float32Array(dims);
  for (let i = 0; i < dims; i++, off += 4) vector[i] = view.getFloat32(off, true);
  const textLen = view.getUint32(off, true);
  off += 4;
  if (off + textLen !== payload.length) return null;
  let text: string;
  try {
    text = decoder.decode(payload.subarray(off, off + textLen));
  } catch {
    return null;
  }
  return { kind: 'insert', id, vector, text };
}

/**
 * Decode a whole log. Validates the header (throwing `WalCorruptError` only for
 * a bad one), then reads records until the first one that overruns the file,
 * fails its CRC, or is malformed. Never throws for a torn tail.
 */
export function scanWal(bytes: Uint8Array, dims: number): WalScan {
  const header = decodeWalHeader(bytes);
  const view = new DataView(bytes.buffer, bytes.byteOffset, bytes.byteLength);
  const records: WalRecord[] = [];
  let off = header.length;
  while (off < bytes.length) {
    if (off + RECORD_PREFIX_LEN > bytes.length) break;
    const payloadLen = view.getUint32(off, true);
    const start = off + RECORD_PREFIX_LEN;
    if (payloadLen === 0 || start + payloadLen > bytes.length) break;
    const payload = bytes.subarray(start, start + payloadLen);
    if (crc32(payload) !== view.getUint32(off + 4, true)) break;
    const rec = decodePayload(payload, dims);
    if (!rec) break;
    records.push(rec);
    off = start + payloadLen;
  }
  return { records, validLength: off, truncatedTail: off < bytes.length };
}