//! Append-only record log, block-framed and compressed (roadmap item 3).
//!
//! File layout:
//!   [0..4)    u32 magic "MLOG"
//!   [4]       u8 version (1)
//!   [5]       u8 codec (0 raw, 1 lz4) — the preferred codec for new blocks
//!   [6..8)    reserved
//!   [8..16)   u64 block_target — soft fill target per block (256 KiB)
//!   [16..)    blocks...
//!
//! Block layout:
//!   [0..4)    u32 total — block bytes including this header
//!   [4..12)   u64 hash over bytes [12..total) — the stored (compressed)
//!             payload exactly as it sits on disk, so the decompressor only
//!             ever sees input already proven intact
//!   [12]      u8 codec (0 raw, 1 lz4); a block falls back to raw when
//!             compression does not help
//!   [13..16)  reserved
//!   [16..)    payload — the block's records, raw or LZ4-compressed
//!
//! Each block holds the pre-existing record framing unchanged:
//!   [0..4)   u32 len — total record bytes
//!   [4..12)  u64 hash over bytes [12..len)
//!   [12..20) u64 seq
//!   [20]     u8 type
//!   [21..24) reserved
//!   [24..)   db\0 coll\0 bson doc
//! All integers little-endian. Records never straddle blocks: appends
//! accumulate into an in-memory block that is sealed (compressed, written,
//! and, when not group-committed, synced) once the next record would push
//! it past the target. Engine.apply_record is untouched — it still sees the
//! same records.
//!
//! Torn tail versus interior corruption stay distinguishable, exactly as
//! before: a short read, an impossible length, or a hash mismatch in the
//! final block means a crash mid-append — replay stops there and later
//! appends overwrite the garbage. A hash mismatch in any earlier block is
//! error.InvalidLog.
//!
//! The integrity hash is XxHash3, not CRC32. Both are checked for the same
//! thing — did these bytes survive intact — but std.hash.Crc32 is the
//! table-driven byte-at-a-time Crc32IsoHdlc, measured here at 408 MB/s
//! against XxHash3's 31 GB/s. On a 16 KiB document that is 38 us versus
//! 0.5 us, which was roughly two thirds of the entire bulk-insert cost.

const std = @import("std");
const bson = @import("bson.zig");

pub const record_type_upsert: u8 = 1;
pub const record_type_delete: u8 = 2;
pub const record_type_index_create: u8 = 3;
pub const record_type_index_drop: u8 = 4;

pub const header_len: usize = 24; // per-record: len + hash + seq + type + reserved
pub const block_header_len: usize = 16; // block: total + hash + codec + reserved
pub const file_header_len: usize = 16; // magic + version + codec + block_target

const file_magic: u32 = 0x4D4C4F47; // "MLOG"
const file_version: u8 = 1;
pub const codec_raw: u8 = 0;
pub const codec_lz4: u8 = 1;

/// Soft fill target for a block's record region. Sealed when the next
/// record would push the block past this; a single oversized record gets
/// its own block.
pub const block_target: usize = 256 * 1024;

/// Integrity hash over a record's bytes after the length and hash fields,
/// or over a block's payload bytes after its header fields.
fn record_hash(bytes: []const u8) u64 {
    return std.hash.XxHash3.hash(0, bytes);
}

/// Largest record payload we will accept during replay. Matches the
/// announced maxBsonObjectSize with room for names and header.
pub const max_record_payload: usize = 16 * 1024 * 1024 + 64 * 1024;

/// Largest stored block payload: block_target plus one oversized record,
/// plus LZ4's worst-case expansion of it.
const max_block_payload: usize = max_record_payload + 1024 * 1024;

pub const Record = struct {
    seq: u64,
    type: u8,
    db: []const u8, // transient: valid only during replay callback
    coll: []const u8,
};

pub const Error = error{
    InvalidLog, // corrupt interior block or record (bad hash or impossible length)
    CorruptLz4, // the decoder saw structurally invalid input
};

/// Callback receives transient slices and a heap-allocated, freshly parsed
/// document. The callback takes ownership of the document (must deinit).
pub const ReplayFn = *const fn (ctx: *anyopaque, record: Record, doc: *bson.Document) anyerror!void;

pub const Log = struct {
    gpa: std.mem.Allocator,
    io: std.Io,
    file: std.Io.File,
    path: []const u8,
    end_pos: u64,
    log_bytes: u64, // bytes written since the log was last rewritten
    /// Uncompressed record bytes appended since the log was last rewritten —
    /// the data volume, which the compaction threshold is really about (the
    /// on-disk size shrinks with compression and would under-trigger).
    data_bytes: u64 = 0,
    codec: u8,
    // Reused record-framing buffer. Appends are single-writer (the engine's
    // exclusive lock), so one buffer avoids a realloc cycle per record.
    scratch: std.ArrayListUnmanaged(u8),
    /// The current block's records, uncompressed, until the block is sealed.
    block: std.ArrayListUnmanaged(u8),
    /// LZ4 output buffer; also the sealed block's payload when raw.
    compressed: std.ArrayListUnmanaged(u8),
    /// LZ4 hash table (positions of recent 4-byte sequences).
    lz4_table: []u32,

    /// Whether an existing file's contents are kept or discarded.
    ///
    /// `.keep` is the database's own log: its bytes are the database, and
    /// `replay` reads them. `.truncate` is for a file being written from
    /// scratch (compaction's tmp), where leftover bytes from an earlier,
    /// longer file would survive past the new content as intact blocks and be
    /// replayed as live records.
    const OpenMode = enum { keep, truncate };

    /// Open the log at `path`, keeping whatever is already there for `replay`.
    pub fn open(gpa: std.mem.Allocator, io: std.Io, path: []const u8) !Log {
        return open_mode(gpa, io, path, .keep);
    }

    /// Open `path` as a brand-new empty log, discarding anything already there.
    ///
    /// Compaction's tmp file must start empty. `open` keeps an existing file's
    /// bytes and only rewinds end_pos to the header, so a longer previous tmp
    /// (a retried or crashed compaction) would leave valid, hash-correct
    /// blocks past the new content -- which `replay` applies as live records
    /// once the rename publishes the file as the database, resurrecting
    /// documents that were deleted.
    pub fn create(gpa: std.mem.Allocator, io: std.Io, path: []const u8) !Log {
        return open_mode(gpa, io, path, .truncate);
    }

    fn open_mode(gpa: std.mem.Allocator, io: std.Io, path: []const u8, mode: OpenMode) !Log {
        // Resolve to an absolute path so compaction can rename the file
        // without depending on the caller's working directory.
        const abs_path = blk: {
            if (path.len > 0 and path[0] == '/') break :blk try gpa.dupe(u8, path);
            const cwd = try std.process.currentPathAlloc(io, gpa);
            defer gpa.free(cwd);
            break :blk try std.fmt.allocPrint(gpa, "{s}/{s}", .{ cwd, path });
        };
        errdefer gpa.free(abs_path);

        const dir = std.Io.Dir.cwd();
        const file: std.Io.File = switch (mode) {
            // createFile truncates by default, so this both creates a missing
            // file and empties an existing one -- the whole point of .truncate.
            .truncate => try dir.createFile(io, abs_path, .{ .read = true, .truncate = true }),
            .keep => dir.openFile(io, abs_path, .{ .mode = .read_write }) catch |err| switch (err) {
                error.FileNotFound => try dir.createFile(io, abs_path, .{ .read = true }),
                else => return err,
            },
        };

        var self: Log = .{
            .gpa = gpa,
            .io = io,
            .file = file,
            .path = abs_path,
            .end_pos = file_header_len,
            .log_bytes = 0,
            .codec = codec_lz4,
            .scratch = .empty,
            .block = .empty,
            .compressed = .empty,
            .lz4_table = undefined,
        };
        errdefer {
            self.scratch.deinit(gpa);
            self.block.deinit(gpa);
            self.compressed.deinit(gpa);
            file.close(io);
        }
        self.lz4_table = try gpa.alloc(u32, lz4_table_size);
        errdefer gpa.free(self.lz4_table);

        // Read or write the file header. A fresh file gets one; an existing
        // file must carry a valid one (a torn header on a never-opened db is
        // unrecoverable and reported as invalid).
        var hdr: [file_header_len]u8 = undefined;
        const n = file.readPositionalAll(io, &hdr, 0) catch return error.InvalidLog;
        if (n == 0) {
            write_file_header(&hdr, codec_lz4);
            try file.writePositionalAll(io, &hdr, 0);
        } else {
            if (n < file_header_len) return error.InvalidLog;
            if (std.mem.readInt(u32, hdr[0..4], .little) != file_magic) return error.InvalidLog;
            if (hdr[4] != file_version) return error.InvalidLog;
            self.codec = hdr[5];
            if (self.codec != codec_raw and self.codec != codec_lz4) return error.InvalidLog;
        }
        return self;
    }

    pub fn close(self: *Log) void {
        // Best-effort: flush an unsealed block so nothing acknowledged is
        // left only in memory. The caller syncs before close on the paths
        // that care about durability.
        self.seal_block() catch {};
        self.block.deinit(self.gpa);
        self.compressed.deinit(self.gpa);
        self.gpa.free(self.lz4_table);
        self.scratch.deinit(self.gpa);
        self.file.close(self.io);
        self.gpa.free(self.path);
    }

    /// Replay all valid records from the beginning of the file.
    /// Replay every record with `seq > from_seq`. Zero replays everything, which
    /// is what an engine with no checkpoint does.
    ///
    /// The walk still visits every block regardless, because that is what leaves
    /// `end_pos` correct for the next append. Skipping is about not *applying*
    /// records the data file already contains; making opens fast is the job of
    /// truncating the log once a checkpoint covers it.
    pub fn replay(self: *Log, ctx: *anyopaque, callback: ReplayFn, from_seq: u64) !void {
        var decomp: std.ArrayListUnmanaged(u8) = .empty;
        defer decomp.deinit(self.gpa);
        var pos: u64 = file_header_len;

        while (true) {
            var hdr: [block_header_len]u8 = undefined;
            const n = self.file.readPositionalAll(self.io, &hdr, pos) catch |err| {
                std.debug.print("multiforadb: log read error at {d}: {s}\n", .{
                    pos,
                    @errorName(err),
                });
                return error.InvalidLog;
            };
            if (n == 0) return; // clean end
            if (n < block_header_len) return; // torn tail: partial block header
            const total: u32 = std.mem.readInt(u32, hdr[0..4], .little);
            if (total < block_header_len or total - block_header_len > max_block_payload) {
                std.debug.print("multiforadb: corrupt block length {d} at {d}\n", .{ total, pos });
                return; // torn tail: impossible length, nothing to validate
            }
            const payload_len: usize = total - block_header_len;
            const payload = try self.gpa.alloc(u8, payload_len);
            defer self.gpa.free(payload);
            const payload_read = self.file.readPositionalAll(self.io, payload, pos + block_header_len) catch return error.InvalidLog;
            if (payload_read < payload_len) return; // torn tail: crash mid-append

            const hash_stored: u64 = std.mem.readInt(u64, hdr[4..12], .little);
            const hash_actual = record_hash(payload);
            if (hash_stored != hash_actual) {
                // A mismatch in the final block is a torn append (the block
                // was partially flushed); a mismatch before more bytes
                // follow is interior corruption. The block header hash
                // covers the stored bytes, so the decompressor below only
                // runs on input already proven intact.
                const file_len = self.file.length(self.io) catch return error.InvalidLog;
                if (pos + total >= file_len) return;
                return error.InvalidLog;
            }

            const codec = hdr[12];
            decomp.clearRetainingCapacity();
            switch (codec) {
                codec_raw => try decomp.appendSlice(self.gpa, payload),
                codec_lz4 => try lz4_decompress(self.gpa, payload, &decomp),
                else => {
                    std.debug.print("multiforadb: unknown block codec {d} at {d}\n", .{
                        codec,
                        pos,
                    });
                    return error.InvalidLog;
                },
            }

            var idx: usize = 0;
            while (idx < decomp.items.len) {
                idx += try self.parse_record(decomp.items[idx..], pos + idx, ctx, callback, from_seq);
            }
            pos += total;
            self.end_pos = pos;
        }
    }

    /// Parse one record from `bytes` (a block's decompressed record region)
    /// and deliver it. Returns the record's byte length. Any framing failure
    /// here is interior corruption: a block was sealed only with complete
    /// records, and its hash proved the stored bytes intact.
    fn parse_record(
        self: *Log,
        bytes: []const u8,
        pos: u64,
        ctx: *anyopaque,
        callback: ReplayFn,
        from_seq: u64,
    ) !usize {
        if (bytes.len < 4) return error.InvalidLog;
        const total: u32 = std.mem.readInt(u32, bytes[0..4], .little);
        if (total < header_len) {
            std.debug.print("multiforadb: corrupt record length {d} at {d}\n", .{ total, pos });
            return error.InvalidLog;
        }
        const payload_len: usize = total - 4;
        if (payload_len > max_record_payload) {
            std.debug.print("multiforadb: record too large at {d}\n", .{pos});
            return error.InvalidLog;
        }
        if (bytes.len < total) return error.InvalidLog; // record crosses the block end
        const payload = bytes[4..total];
        const hash_stored: u64 = std.mem.readInt(u64, payload[0..8], .little);
        const hash_actual = record_hash(payload[8..]);
        if (hash_stored != hash_actual) return error.InvalidLog;

        var idx: usize = header_len - 4;
        const seq: u64 = std.mem.readInt(u64, payload[8..16], .little);
        // Below the checkpoint: the data file already holds its effect. Skipped
        // before the BSON parse, which is the expensive part.
        if (seq <= from_seq) return total;
        const rtype = payload[16];
        const db = read_cstring(payload, &idx) orelse return error.InvalidLog;
        const coll = read_cstring(payload, &idx) orelse return error.InvalidLog;
        if (idx > payload_len) return error.InvalidLog;
        const doc_bytes = payload[idx..payload_len];

        const doc = try self.gpa.create(bson.Document);
        doc.* = bson.Document.parse(self.gpa, doc_bytes) catch {
            self.gpa.destroy(doc);
            std.debug.print("multiforadb: unparseable doc in log at {d}\n", .{pos});
            return error.InvalidLog;
        };
        try callback(ctx, .{
            .seq = seq,
            .type = rtype,
            .db = db,
            .coll = coll,
        }, doc);
        return total;
    }

    pub fn append_upsert(
        self: *Log,
        db: []const u8,
        coll: []const u8,
        doc: []const u8,
        seq: u64,
    ) !void {
        try self.append(record_type_upsert, db, coll, doc, seq);
    }

    pub fn append_delete(
        self: *Log,
        db: []const u8,
        coll: []const u8,
        doc: []const u8,
        seq: u64,
    ) !void {
        try self.append(record_type_delete, db, coll, doc, seq);
    }

    /// The payload is the canonical index spec document ({v, key, name,
    /// unique?, sparse?}); only apply_record interprets it.
    pub fn append_index_create(
        self: *Log,
        db: []const u8,
        coll: []const u8,
        doc: []const u8,
        seq: u64,
    ) !void {
        try self.append(record_type_index_create, db, coll, doc, seq);
    }

    /// The payload is {name: "..."}; only apply_record interprets it.
    pub fn append_index_drop(
        self: *Log,
        db: []const u8,
        coll: []const u8,
        doc: []const u8,
        seq: u64,
    ) !void {
        try self.append(record_type_index_drop, db, coll, doc, seq);
    }

    fn append(
        self: *Log,
        rtype: u8,
        db: []const u8,
        coll: []const u8,
        doc: []const u8,
        seq: u64,
    ) !void {
        if (std.mem.indexOfScalar(u8, db, 0) != null or std.mem.indexOfScalar(u8, coll, 0) != null) {
            return error.NulInName;
        }
        const buf = &self.scratch;
        buf.clearRetainingCapacity();
        try buf.appendNTimes(self.gpa, 0, header_len);
        std.mem.writeInt(u64, buf.items[12..20], seq, .little);
        buf.items[20] = rtype;
        try buf.appendSlice(self.gpa, db);
        try buf.append(self.gpa, 0);
        try buf.appendSlice(self.gpa, coll);
        try buf.append(self.gpa, 0);
        try buf.appendSlice(self.gpa, doc);
        if (buf.items.len > std.math.maxInt(u32)) return error.LogTooLarge;
        const total: u32 = @intCast(buf.items.len);
        std.mem.writeInt(u32, buf.items[0..4], total, .little);
        std.mem.writeInt(u64, buf.items[4..12], record_hash(buf.items[12..]), .little);

        self.data_bytes += buf.items.len;
        // Seal the current block when the next record would push it past the
        // target; a single oversized record keeps its own block.
        if (self.block.items.len > 0 and self.block.items.len + buf.items.len > block_target) {
            try self.seal_block();
        }
        try self.block.appendSlice(self.gpa, buf.items);
        // No sync here: durability is the commit point (Log.sync), which
        // runs once per write command and coalesces across connections.
    }

    /// Compress and write the current block, then reset it. No-op when it is
    /// empty. Blocks fall back to raw when compression does not help.
    fn seal_block(self: *Log) !void {
        if (self.block.items.len == 0) return;
        defer self.block.clearRetainingCapacity();

        var codec: u8 = codec_raw;
        var payload: []const u8 = self.block.items;
        if (self.codec == codec_lz4) {
            const worst = self.block.items.len + self.block.items.len / 255 + 16;
            try self.compressed.ensureTotalCapacity(self.gpa, worst);
            const out = self.compressed.allocatedSlice()[0..worst];
            const n = lz4_compress(self.block.items, out, self.lz4_table);
            if (n < self.block.items.len) {
                payload = out[0..n];
                codec = codec_lz4;
            }
        }

        var hdr: [block_header_len]u8 = undefined;
        const total: u32 = @intCast(block_header_len + payload.len);
        std.mem.writeInt(u32, hdr[0..4], total, .little);
        std.mem.writeInt(u64, hdr[4..12], record_hash(payload), .little);
        hdr[12] = codec;
        hdr[13..block_header_len].* = [_]u8{ 0, 0, 0 };
        try self.file.writePositionalAll(self.io, &hdr, self.end_pos);
        try self.file.writePositionalAll(self.io, payload, self.end_pos + block_header_len);
        self.end_pos += total;
        self.log_bytes += total;
    }

    /// The log's only commit point: seal the open block, then one fsync.
    ///
    /// Appends never sync (see `append_record`), so nothing is durable until
    /// this returns -- which is why Engine.commit calls it exactly once per
    /// write command, coalescing every writer in flight into a single fsync.
    /// Discard every record, keeping only the file header.
    ///
    /// Only a checkpoint may call this, and only after the watermark that covers
    /// these records is durable. The whole point of a lagging checkpoint is that
    /// the log can be reclaimed once the data file holds its effect -- and the
    /// order is not negotiable: truncate before the watermark is durable and a
    /// crash in between leaves records gone from the log and absent from the
    /// image.
    ///
    /// Caller holds the log lock. The open block is dropped rather than sealed:
    /// its records are below the watermark too, so writing them out would only
    /// be work the next open throws away.
    pub fn truncate_to_header(self: *Log) !void {
        self.block.clearRetainingCapacity();
        try self.file.setLength(self.io, file_header_len);
        try self.file.sync(self.io);
        self.end_pos = file_header_len;
        self.log_bytes = 0;
        self.data_bytes = 0;
    }

    pub fn sync(self: *Log) !void {
        try self.seal_block();
        try self.file.sync(self.io);
    }

    fn read_cstring(bytes: []const u8, idx: *usize) ?[]const u8 {
        const start = idx.*;
        while (idx.* < bytes.len and bytes[idx.*] != 0) idx.* += 1;
        if (idx.* >= bytes.len) return null;
        idx.* += 1;
        return bytes[start .. idx.* - 1];
    }
};

fn write_file_header(hdr: *[file_header_len]u8, codec: u8) void {
    std.mem.writeInt(u32, hdr[0..4], file_magic, .little);
    hdr[4] = file_version;
    hdr[5] = codec;
    hdr[6] = 0;
    hdr[7] = 0;
    std.mem.writeInt(u64, hdr[8..16], block_target, .little);
}

// ---------------------------------------------------------------------------
// LZ4 block codec
// ---------------------------------------------------------------------------
//
// The standard LZ4 block format: a sequence of (token, literals, offset,
// match-length) items. The token's high nibble is the literal length
// (15 = extended by bytes), the low nibble is the match length minus 4
// (15 = extended). Matches reference earlier output by a 16-bit offset and
// may overlap it. Self-contained: only this decoder reads this log.

const lz4_table_size: usize = 4096;

fn lz4_hash(v: u32) u32 {
    return (v *% 2654435761) >> 20;
}

/// Compress `src` into `dst`, which must hold at least
/// src.len + src.len/255 + 16 bytes. Returns the compressed length.
pub fn lz4_compress(src: []const u8, dst: []u8, table: []u32) usize {
    if (src.len == 0) return 0;
    @memset(table, 0);
    var out: usize = 0;
    var anchor: usize = 0; // start of the pending literal run
    var ip: usize = 0;
    // Matches may not start in the final 5 bytes (they end there), keeping
    // the stream standard-compliant: the last 5 bytes are always literals.
    const limit = if (src.len >= 5) src.len - 5 else 0;

    while (ip < limit) {
        // Scan for a 4-byte match via the hash table.
        var match_pos: ?usize = null;
        while (ip + 4 <= src.len) : (ip += 1) {
            const h = lz4_hash(std.mem.readInt(u32, src[ip..][0..4], .little));
            const cand = table[h];
            table[h] = @intCast(ip + 1); // 0 means empty; positions are stored +1
            if (cand != 0) {
                const cp = cand - 1;
                const back = ip - cp;
                if (back >= lz4_min_match and back <= 65535 and
                    std.mem.readInt(u32, src[cp..][0..4], .little) == std.mem.readInt(u32, src[ip..][0..4], .little))
                {
                    match_pos = cp;
                    break;
                }
            }
        }
        const mp = match_pos orelse break;
        // Extend the match, stopping at the end or the 5-byte literal tail.
        var match_len: usize = lz4_min_match;
        const match_limit = src.len - 5;
        while (ip + match_len < match_limit and src[ip + match_len] == src[mp + match_len]) : (match_len += 1) {}
        out += emit_sequence(dst[out..], src[anchor..ip], @intCast(ip - mp), match_len);
        ip += match_len;
        anchor = ip;
    }
    out += emit_literals(dst[out..], src[anchor..]);
    return out;
}

const lz4_min_match: usize = 4;

/// Emit one (token, literals, offset, match length) item.
fn emit_sequence(dst: []u8, literals: []const u8, offset: u16, match_len: usize) usize {
    var out: usize = 0;
    const lit_len = literals.len;
    var token: u8 = if (lit_len >= 15) 0xF0 else @intCast(lit_len << 4);
    const mlen = match_len - lz4_min_match;
    token |= if (mlen >= 15) 0x0F else @intCast(mlen);
    dst[out] = token;
    out += 1;
    if (lit_len >= 15) {
        var rem = lit_len - 15;
        while (rem >= 255) : (rem -= 255) {
            dst[out] = 255;
            out += 1;
        }
        dst[out] = @intCast(rem);
        out += 1;
    }
    @memcpy(dst[out .. out + lit_len], literals);
    out += lit_len;
    std.mem.writeInt(u16, dst[out..][0..2], offset, .little);
    out += 2;
    if (mlen >= 15) {
        var rem = mlen - 15;
        while (rem >= 255) : (rem -= 255) {
            dst[out] = 255;
            out += 1;
        }
        dst[out] = @intCast(rem);
        out += 1;
    }
    return out;
}

/// Emit a trailing literal run (a final item with no match).
fn emit_literals(dst: []u8, literals: []const u8) usize {
    if (literals.len == 0) return 0;
    var out: usize = 0;
    dst[out] = if (literals.len >= 15) 0xF0 else @intCast(literals.len << 4);
    out += 1;
    if (literals.len >= 15) {
        var rem = literals.len - 15;
        while (rem >= 255) : (rem -= 255) {
            dst[out] = 255;
            out += 1;
        }
        dst[out] = @intCast(rem);
        out += 1;
    }
    @memcpy(dst[out .. out + literals.len], literals);
    out += literals.len;
    return out;
}

/// Decompress an LZ4 block into `out` (appended). The block hash has
/// already proven the input intact when this runs during replay, so
/// structural failures here mean a bug or a raw-codec mismatch.
fn lz4_decompress(
    gpa: std.mem.Allocator,
    src: []const u8,
    out: *std.ArrayListUnmanaged(u8),
) error{ CorruptLz4, OutOfMemory }!void {
    var ip: usize = 0;
    while (ip < src.len) {
        const token = src[ip];
        ip += 1;
        var lit_len: usize = token >> 4;
        if (lit_len == 15) {
            while (true) {
                if (ip >= src.len) return error.CorruptLz4;
                const b = src[ip];
                ip += 1;
                lit_len += b;
                if (b != 255) break;
            }
        }
        if (ip + lit_len > src.len) return error.CorruptLz4;
        try out.appendSlice(gpa, src[ip .. ip + lit_len]);
        ip += lit_len;
        if (ip >= src.len) break; // trailing literals only
        if (ip + 2 > src.len) return error.CorruptLz4;
        const offset = std.mem.readInt(u16, src[ip..][0..2], .little);
        ip += 2;
        if (offset == 0 or offset > out.items.len) return error.CorruptLz4;
        var match_len: usize = token & 0x0F;
        if (match_len == 15) {
            while (true) {
                if (ip >= src.len) return error.CorruptLz4;
                const b = src[ip];
                ip += 1;
                match_len += b;
                if (b != 255) break;
            }
        }
        match_len += lz4_min_match;
        const base = out.items.len - offset;
        // Grow once, then copy in place: the source cannot alias the
        // destination's old buffer across a realloc. The non-overlapping
        // prefix copies in one shot; the overlapping tail (offset < match
        // length) copies itself forward byte by byte.
        const region = try out.addManyAsSlice(gpa, match_len);
        const non_overlap = @min(match_len, offset);
        @memcpy(region[0..non_overlap], out.items[base .. base + non_overlap]);
        var k = non_overlap;
        while (k < match_len) : (k += 1) {
            region[k] = out.items[base + k];
        }
    }
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

const testing = std.testing;

/// Throwaway log file in the test temp dir, shared by the storage and
/// engine test suites.
pub const TmpLog = struct {
    tmp: std.testing.TmpDir,
    path: []u8,

    pub fn init(gpa: std.mem.Allocator) !TmpLog {
        const tmp = std.testing.tmpDir(.{});
        const path = try std.fmt.allocPrint(gpa, ".zig-cache/tmp/{s}/test.log", .{tmp.sub_path});
        return .{ .tmp = tmp, .path = path };
    }

    pub fn deinit(self: *TmpLog, gpa: std.mem.Allocator) void {
        self.tmp.cleanup();
        gpa.free(self.path);
    }
};

test "lz4 round-trips, including long runs and incompressible input" {
    const gpa = testing.allocator;
    var prng = std.Random.DefaultPrng.init(0x1A2B3C4D);
    const table = try gpa.alloc(u32, lz4_table_size);
    defer gpa.free(table);

    var src: std.ArrayListUnmanaged(u8) = .empty;
    defer src.deinit(gpa);
    // Highly compressible: a repeating pattern with some noise.
    for (0..100_000) |i| {
        try src.append(gpa, if (i % 37 == 0) @as(u8, @intCast(prng.next() & 0xFF)) else 'a' + @as(u8, @intCast(i % 26)));
    }
    const worst = src.items.len + src.items.len / 255 + 16;
    const dst = try gpa.alloc(u8, worst);
    defer gpa.free(dst);
    const n = lz4_compress(src.items, dst, table);
    try testing.expect(n < src.items.len / 3); // actually compressed
    var out: std.ArrayListUnmanaged(u8) = .empty;
    defer out.deinit(gpa);
    try lz4_decompress(gpa, dst[0..n], &out);
    try testing.expectEqualSlices(u8, src.items, out.items);

    // Incompressible input still round-trips (and expands).
    out.clearRetainingCapacity();
    src.clearRetainingCapacity();
    for (0..10_000) |_| try src.append(gpa, @intCast(prng.next() & 0xFF));
    const n2 = lz4_compress(src.items, dst, table);
    try testing.expect(n2 >= src.items.len);
    try lz4_decompress(gpa, dst[0..n2], &out);
    try testing.expectEqualSlices(u8, src.items, out.items);

    // Truncated input is rejected, not read out of bounds: a token
    // promising four literals with only two bytes left, and a match
    // sequence with no offset.
    try testing.expectError(error.CorruptLz4, lz4_decompress(gpa, &[_]u8{ 0x40, 1, 2 }, &out));
    try testing.expectError(error.CorruptLz4, lz4_decompress(gpa, &[_]u8{ 0x00, 0x00 }, &out));
}

test "append, replay, torn tail" {
    var threaded: std.Io.Threaded = .init_single_threaded;
    defer threaded.deinit();
    const io = threaded.io();
    const gpa = testing.allocator;

    var tmp = try TmpLog.init(gpa);
    defer tmp.deinit(gpa);
    var log = try Log.open(gpa, io, tmp.path);
    defer log.close();

    const doc_bytes = [_]u8{
        0x0E, 0x00, 0x00, 0x00, // len 14
        0x10, '_', 'i', 'd', 0, 0x2A, 0x00, 0x00, 0x00, // _id: 42
        0x00,
    };
    try log.append_upsert("db1", "coll1", &doc_bytes, 1);
    try log.append_delete("db1", "coll1", &doc_bytes, 2);
    try log.sync();

    var seen: std.ArrayListUnmanaged(u8) = .empty;
    defer seen.deinit(gpa);
    const Ctx = struct {
        seen: *std.ArrayListUnmanaged(u8),
        gpa: std.mem.Allocator,
        fn apply(ctx: *anyopaque, record: Record, doc: *bson.Document) anyerror!void {
            const self: *@This() = @ptrCast(@alignCast(ctx));
            try self.seen.append(self.gpa, record.type);
            try self.seen.append(self.gpa, @intCast(record.seq));
            doc.deinit();
            self.gpa.destroy(doc);
        }
    };
    var ctx = Ctx{ .seen = &seen, .gpa = gpa };
    try log.replay(@ptrCast(&ctx), Ctx.apply, 0);

    try testing.expectEqualSlices(u8, &[_]u8{ record_type_upsert, 1, record_type_delete, 2 }, seen.items);
}

test "record larger than the read chunk replays" {
    var threaded: std.Io.Threaded = .init_single_threaded;
    defer threaded.deinit();
    const io = threaded.io();
    const gpa = testing.allocator;

    var tmp = try TmpLog.init(gpa);
    defer tmp.deinit(gpa);
    var log = try Log.open(gpa, io, tmp.path);
    defer log.close();

    // Build a doc whose payload pushes the record past the block target,
    // so it gets its own oversized block.
    const big = try gpa.alloc(u8, 80 * 1024);
    defer gpa.free(big);
    @memset(big, 'x');
    const pairs = [_]bson.Pair{
        .{ .key = "_id", .value = .{ .int32 = 1 } },
        .{ .key = "blob", .value = .{ .binary = .{ .subtype = 0, .data = big } } },
    };
    var out: std.ArrayListUnmanaged(u8) = .empty;
    defer out.deinit(gpa);
    try bson.write_doc(&pairs, gpa, &out);
    try log.append_upsert("db", "big", out.items, 1);
    try log.sync();

    var count: usize = 0;
    const Ctx = struct {
        count: *usize,
        gpa: std.mem.Allocator,
        fn apply(ctx: *anyopaque, _: Record, doc: *bson.Document) anyerror!void {
            const self: *@This() = @ptrCast(@alignCast(ctx));
            try testing.expectEqual(@as(usize, 80 * 1024), doc.get("blob").?.binary.data.len);
            self.count.* += 1;
            doc.deinit();
            self.gpa.destroy(doc);
        }
    };
    var ctx = Ctx{ .count = &count, .gpa = gpa };
    try log.replay(@ptrCast(&ctx), Ctx.apply, 0);
    try testing.expectEqual(@as(usize, 1), count);
}

test "reject corrupt interior block" {
    var threaded: std.Io.Threaded = .init_single_threaded;
    defer threaded.deinit();
    const io = threaded.io();
    const gpa = testing.allocator;

    var tmp = try TmpLog.init(gpa);
    defer tmp.deinit(gpa);
    const path = tmp.path;

    var log = try Log.open(gpa, io, path);
    const doc_bytes = [_]u8{ 0x0E, 0, 0, 0, 0x10, '_', 'i', 'd', 0, 42, 0, 0, 0, 0 };
    try log.append_upsert("db", "c", &doc_bytes, 1);
    try log.append_upsert("db", "c", &doc_bytes, 2); // a second block
    try log.sync();
    log.close();

    // Corrupt the FIRST block's first record: flip a byte in the hashed
    // region. The first block's payload starts at the file header plus the
    // block header; the record framing puts db\0coll\0 before the doc.
    const corrupt_at = file_header_len + block_header_len + header_len + 5;
    const dir = std.Io.Dir.cwd();
    var f = try dir.openFile(io, path, .{ .mode = .read_write });
    var buf: [128]u8 = undefined;
    const n = try f.readPositionalAll(io, &buf, 0);
    _ = n;
    buf[corrupt_at] ^= 0xFF;
    try f.writePositionalAll(io, buf[0..128], 0);
    f.close(io);

    var log2 = try Log.open(gpa, io, path);
    defer log2.close();
    defer dir.deleteFile(io, path) catch {};
    var count: usize = 0;
    const Ctx = struct {
        count: *usize,
        gpa: std.mem.Allocator,
        fn apply(ctx: *anyopaque, _: Record, doc: *bson.Document) anyerror!void {
            const self: *@This() = @ptrCast(@alignCast(ctx));
            self.count.* += 1;
            doc.deinit();
            self.gpa.destroy(doc);
        }
    };
    var ctx = Ctx{ .count = &count, .gpa = gpa };
    try testing.expectError(error.InvalidLog, log2.replay(@ptrCast(&ctx), Ctx.apply, 0));
}

test "torn tail truncates cleanly and appends overwrite it" {
    var threaded: std.Io.Threaded = .init_single_threaded;
    defer threaded.deinit();
    const io = threaded.io();
    const gpa = testing.allocator;

    var tmp = try TmpLog.init(gpa);
    defer tmp.deinit(gpa);
    const path = tmp.path;

    var log = try Log.open(gpa, io, path);
    const doc_bytes = [_]u8{ 0x0E, 0, 0, 0, 0x10, '_', 'i', 'd', 0, 42, 0, 0, 0, 0 };
    // One sync per record gives two blocks: the first is complete, the
    // second is the one torn by the truncation.
    try log.append_upsert("db", "c", &doc_bytes, 1);
    try log.sync();
    try log.append_upsert("db", "c", &doc_bytes, 2);
    try log.sync();
    log.close();

    // Cut the file in the middle of the second block: a crash mid-append.
    const dir = std.Io.Dir.cwd();
    var f = try dir.openFile(io, path, .{ .mode = .read_write });
    const full_len = try f.length(io);
    try f.setLength(io, full_len - 10);
    f.close(io);

    var log2 = try Log.open(gpa, io, path);
    defer log2.close();
    defer dir.deleteFile(io, path) catch {};
    var seen: std.ArrayListUnmanaged(u8) = .empty;
    defer seen.deinit(gpa);
    const Ctx = struct {
        seen: *std.ArrayListUnmanaged(u8),
        gpa: std.mem.Allocator,
        fn apply(ctx: *anyopaque, record: Record, doc: *bson.Document) anyerror!void {
            const self: *@This() = @ptrCast(@alignCast(ctx));
            try self.seen.append(self.gpa, @intCast(record.seq));
            doc.deinit();
            self.gpa.destroy(doc);
        }
    };
    var ctx = Ctx{ .seen = &seen, .gpa = gpa };
    try log2.replay(@ptrCast(&ctx), Ctx.apply, 0);
    try testing.expectEqualSlices(u8, &[_]u8{1}, seen.items);

    // A new append overwrites from the replay end and replays cleanly.
    try log2.append_upsert("db", "c", &doc_bytes, 3);
    try log2.sync();
    seen.clearRetainingCapacity();
    var log3 = try Log.open(gpa, io, path);
    defer log3.close();
    ctx.seen = &seen;
    try log3.replay(@ptrCast(&ctx), Ctx.apply, 0);
    try testing.expectEqualSlices(u8, &[_]u8{ 1, 3 }, seen.items);
}

test "Log.create discards a leftover file; Log.open keeps it" {
    // Compaction reuses one tmp path, so a crashed or retried rewrite can leave
    // a *longer* file there. `open` only rewinds end_pos to the header, so the
    // predecessor's trailing blocks would survive past the new content -- and
    // they are intact and hash-correct, so replay applies them as live records
    // once the rename publishes the file. `create` is what prevents that.
    var threaded: std.Io.Threaded = .init_single_threaded;
    defer threaded.deinit();
    const io = threaded.io();
    const gpa = testing.allocator;

    var tmp = try TmpLog.init(gpa);
    defer tmp.deinit(gpa);
    const path = tmp.path;
    const doc_bytes = [_]u8{ 0x0E, 0, 0, 0, 0x10, '_', 'i', 'd', 0, 42, 0, 0, 0, 0 };

    // Stand in for the abandoned rewrite: three records, synced, then closed.
    {
        var old = try Log.open(gpa, io, path);
        defer old.close();
        try old.append_upsert("db", "c", &doc_bytes, 1);
        try old.append_upsert("db", "c", &doc_bytes, 2);
        try old.append_upsert("db", "c", &doc_bytes, 3);
        try old.sync();
        try testing.expect(try old.file.length(io) > file_header_len);
    }

    const Ctx = struct {
        seen: *std.ArrayListUnmanaged(u8),
        gpa: std.mem.Allocator,
        fn apply(ctx: *anyopaque, record: Record, doc: *bson.Document) anyerror!void {
            const self: *@This() = @ptrCast(@alignCast(ctx));
            try self.seen.append(self.gpa, @intCast(record.seq));
            doc.deinit();
            self.gpa.destroy(doc);
        }
    };
    var seen: std.ArrayListUnmanaged(u8) = .empty;
    defer seen.deinit(gpa);
    var ctx = Ctx{ .seen = &seen, .gpa = gpa };

    // `open` keeps the leftover bytes: this is the hazard being guarded against.
    {
        var kept = try Log.open(gpa, io, path);
        defer kept.close();
        try testing.expect(try kept.file.length(io) > file_header_len);
        try kept.replay(@ptrCast(&ctx), Ctx.apply, 0);
        try testing.expectEqualSlices(u8, &[_]u8{ 1, 2, 3 }, seen.items);
    }

    // `create` leaves an empty log: nothing past the header, on disk or in the
    // append position, so no stale record can be replayed.
    {
        var fresh = try Log.create(gpa, io, path);
        defer fresh.close();
        try testing.expectEqual(@as(u64, file_header_len), try fresh.file.length(io));
        try testing.expectEqual(@as(u64, file_header_len), fresh.end_pos);

        // One short record where three used to be: replay must see only it,
        // proving the old tail is gone rather than merely skipped.
        try fresh.append_upsert("db", "c", &doc_bytes, 9);
        try fresh.sync();
        try testing.expectEqual(try fresh.file.length(io), fresh.end_pos);
    }
    seen.clearRetainingCapacity();
    var reopened = try Log.open(gpa, io, path);
    defer reopened.close();
    try reopened.replay(@ptrCast(&ctx), Ctx.apply, 0);
    try testing.expectEqualSlices(u8, &[_]u8{9}, seen.items);
}