//! Persistence — save/load the whole Store to a single file.
//!
//! Custom byte-safe binary format. We don't need wire compatibility with
//! upstream burner-redis (different language, different process), so we
//! skip MessagePack and serialise to a compact format defined here. The
//! benefit: no external dep, no need for a MessagePack zig library, and
//! every []const u8 is preserved bit-for-bit (no UTF-8 detour).
//!
//! Format (all integers little-endian, lengths prefixed):
//!
//!   header: "BRDB" (4 bytes magic) | u32 version
//!   then:
//!     u64 n_entries
//!     for each entry:
//!       u64 key_len, key_bytes
//!       u8  has_ttl (0/1); if 1 then i128 expires_at_ns
//!       u8  variant tag (see ValueTag below)
//!       (variant-specific payload)
//!     u64 n_scripts
//!     for each script:
//!       u64 sha_len, sha_bytes
//!       u64 body_len, body_bytes
//!
//! Variant payloads:
//!   string:     u64 len, bytes
//!   hash:       u64 n, (u64 fk, field, u64 vk, value)*
//!   set:        u64 n, (u64 lk, bytes)*
//!   sorted_set: u64 n, (f64 score, u64 lk, bytes)*       (sorted by (score, member))
//!   stream:     u64 n_entries, (u64 ms, u64 seq, u64 n_fields, (u64 fk, field, u64 vk, value)*)*
//!               u64 last_ms, u64 last_seq
//!               u64 n_groups, (u64 lk, group_name, group_data)*
//!     group_data: u64 last_delivered_ms, u64 last_delivered_seq
//!                 u64 n_consumers, (u64 lk, consumer_name, consumer_data)*
//!     consumer_data: u64 n_pending, (u64 ms, u64 seq, i128 delivered_at_ns, u64 delivery_count)*
//!   list:       u64 n, (u64 lk, bytes)*
//!
//! Pub/sub state is in-memory only; subscribers and their channel/pattern
//! routing aren't persisted (they're tied to live process state).
//!
//! Write strategy is crash-safe: write the whole blob to `<path>.tmp`,
//! fsync, then rename onto `<path>`. The intermediate is removed if the
//! write fails.

const std = @import("std");
const Allocator = std.mem.Allocator;
const store_mod = @import("store.zig");
const Store = store_mod.Store;
const StreamId = store_mod.StreamId;

const MAGIC: [4]u8 = .{ 'B', 'R', 'D', 'B' };
const VERSION: u32 = 1;

pub const PersistenceError = error{
    BadMagic,
    UnsupportedVersion,
    Truncated,
    InvalidVariant,
    /// The bytes parse, but not as a store this code could have saved.
    Corrupt,
} || std.Io.File.OpenError || std.Io.File.Writer.Error || std.Io.File.SyncError || std.Io.Dir.RenameError || std.Io.File.StatError || Allocator.Error || std.Io.Cancelable;

const ValueTag = enum(u8) {
    string = 0,
    hash = 1,
    set = 2,
    sorted_set = 3,
    stream = 4,
    list = 5,
};

// ----------------------------------------------------------------------------
// Save
// ----------------------------------------------------------------------------

/// Snapshot `store` to `path`. Writes to a temp file + atomic rename; safe
/// to call concurrently with reads but assumes the store mutex is held
/// (use `Store.save` which locks for you).
pub fn save(store: *Store, path: []const u8) PersistenceError!void {
    var arena = std.heap.ArenaAllocator.init(store.allocator);
    defer arena.deinit();
    const a = arena.allocator();

    // Build the entire snapshot in memory, then write+rename. For docket
    // workloads the snapshot is tiny (kilobytes); this keeps the file
    // layer simple.
    var buf: std.ArrayList(u8) = .empty;
    defer buf.deinit(a);

    try writeAll(&buf, a, store);

    const tmp_path = try std.fmt.allocPrint(a, "{s}.tmp", .{path});
    const cwd = std.Io.Dir.cwd();
    errdefer cwd.deleteFile(store.io, tmp_path) catch {};
    {
        const f = try cwd.createFile(store.io, tmp_path, .{ .truncate = true });
        defer f.close(store.io);
        try f.writeStreamingAll(store.io, buf.items);
        try f.sync(store.io);
    }
    try cwd.rename(tmp_path, cwd, path, store.io);
}

fn writeAll(buf: *std.ArrayList(u8), a: Allocator, store: *Store) PersistenceError!void {
    try buf.appendSlice(a, &MAGIC);
    try writeU32(buf, a, VERSION);

    const now_ns = store.nowNs();
    var n_live: u64 = 0;
    var it = store.data.iterator();
    while (it.next()) |kv| {
        if (kv.value_ptr.isExpired(now_ns)) continue;
        n_live += 1;
    }
    try writeU64(buf, a, n_live);

    it = store.data.iterator();
    while (it.next()) |kv| {
        if (kv.value_ptr.isExpired(now_ns)) continue;
        try writeBytes(buf, a, kv.key_ptr.*);
        if (kv.value_ptr.expires_at_ns) |exp| {
            try buf.append(a, 1);
            try writeI128(buf, a, exp);
        } else {
            try buf.append(a, 0);
        }
        try writeValueData(buf, a, &kv.value_ptr.data);
    }

    // SCRIPT LOAD registry — matches upstream's "scripts persisted alongside
    // data" behaviour so EVALSHA still finds known scripts after restore.
    try writeU64(buf, a, store.scripts.count());
    var sit = store.scripts.iterator();
    while (sit.next()) |kv| {
        try writeBytes(buf, a, kv.key_ptr.*);
        try writeBytes(buf, a, kv.value_ptr.*);
    }
}

fn writeValueData(buf: *std.ArrayList(u8), a: Allocator, vd: *const store_mod.ValueData) PersistenceError!void {
    switch (vd.*) {
        .string => |s| {
            try buf.append(a, @backingInt(ValueTag.string));
            try writeBytes(buf, a, s);
        },
        .hash => |*h| {
            try buf.append(a, @backingInt(ValueTag.hash));
            try writeU64(buf, a, h.count());
            var hit = h.iterator();
            while (hit.next()) |fv| {
                try writeBytes(buf, a, fv.key_ptr.*);
                try writeBytes(buf, a, fv.value_ptr.*);
            }
        },
        .set => |*s| {
            try buf.append(a, @backingInt(ValueTag.set));
            try writeU64(buf, a, s.count());
            var sit = s.keyIterator();
            while (sit.next()) |m| try writeBytes(buf, a, m.*);
        },
        .sorted_set => |*z| {
            try buf.append(a, @backingInt(ValueTag.sorted_set));
            try writeU64(buf, a, z.by_score.count());
            var zit = z.by_score.iterator();
            while (zit.next()) |sm| {
                try writeF64(buf, a, sm.score);
                try writeBytes(buf, a, sm.member);
            }
        },
        .stream => |*s| {
            try buf.append(a, @backingInt(ValueTag.stream));
            try writeU64(buf, a, s.entries.items.len);
            for (s.entries.items) |e| {
                try writeU64(buf, a, e.id.ms);
                try writeU64(buf, a, e.id.seq);
                try writeU64(buf, a, e.fields.count());
                var fit = e.fields.iterator();
                while (fit.next()) |fv| {
                    try writeBytes(buf, a, fv.key_ptr.*);
                    try writeBytes(buf, a, fv.value_ptr.*);
                }
            }
            try writeU64(buf, a, s.last_id.ms);
            try writeU64(buf, a, s.last_id.seq);
            try writeU64(buf, a, s.groups.count());
            var git = s.groups.iterator();
            while (git.next()) |gkv| {
                try writeBytes(buf, a, gkv.key_ptr.*);
                try writeU64(buf, a, gkv.value_ptr.last_delivered_id.ms);
                try writeU64(buf, a, gkv.value_ptr.last_delivered_id.seq);
                try writeU64(buf, a, gkv.value_ptr.consumers.count());
                var cit = gkv.value_ptr.consumers.iterator();
                while (cit.next()) |ckv| {
                    try writeBytes(buf, a, ckv.key_ptr.*);
                    try writeU64(buf, a, ckv.value_ptr.pending.count());
                    var pit = ckv.value_ptr.pending.iterator();
                    while (pit.next()) |pkv| {
                        try writeU64(buf, a, pkv.key_ptr.ms);
                        try writeU64(buf, a, pkv.key_ptr.seq);
                        try writeI128(buf, a, pkv.value_ptr.delivered_at_ns);
                        try writeU64(buf, a, pkv.value_ptr.delivery_count);
                    }
                }
            }
        },
        .list => |*l| {
            try buf.append(a, @backingInt(ValueTag.list));
            const n = listLen(l);
            try writeU64(buf, a, n);
            var node = l.first;
            while (node) |nd| : (node = nd.next) {
                const ln: *store_mod.ListNode = @fieldParentPtr("node", nd);
                try writeBytes(buf, a, ln.data);
            }
        },
    }
}

fn listLen(l: *const std.DoublyLinkedList) usize {
    var n: usize = 0;
    var nd = l.first;
    while (nd) |x| : (nd = x.next) n += 1;
    return n;
}

// ----------------------------------------------------------------------------
// Load
// ----------------------------------------------------------------------------

/// Restore a previously-saved snapshot from `path` into `store`. Returns
/// false if `path` is missing (treated as "no snapshot yet"); true if
/// successfully loaded. A snapshot that is refused leaves `store` as it was.
pub fn load(store: *Store, path: []const u8) PersistenceError!bool {
    const cwd = std.Io.Dir.cwd();
    const f = cwd.openFile(store.io, path, .{}) catch |err| switch (err) {
        error.FileNotFound => return false,
        else => return err,
    };
    defer f.close(store.io);

    const st = try f.stat(store.io);
    const bytes = try store.allocator.alloc(u8, @intCast(st.size));
    defer store.allocator.free(bytes);
    var reader_buf: [4096]u8 = undefined;
    var reader = f.reader(store.io, &reader_buf);
    reader.interface.readSliceAll(bytes) catch |err| switch (err) {
        error.EndOfStream, error.ReadFailed => return PersistenceError.Truncated,
    };

    try restore(store, bytes);
    return true;
}

const Keyspace = store_mod.BytesHashMap(store_mod.ValueEntry);
const Scripts = std.StringHashMap([]const u8);

fn freeKeyspace(store: *Store, data: *Keyspace) void {
    var it = data.iterator();
    while (it.next()) |kv| {
        store.allocator.free(kv.key_ptr.*);
        store.freeValueData(&kv.value_ptr.data);
    }
    data.deinit();
}

fn freeScripts(store: *Store, scripts: *Scripts) void {
    var it = scripts.iterator();
    while (it.next()) |kv| {
        store.allocator.free(kv.key_ptr.*);
        store.allocator.free(kv.value_ptr.*);
    }
    scripts.deinit();
}

/// Replace what `store` holds with the snapshot in `bytes`. The snapshot is
/// read whole before the store is touched, and it is refused if it is cut
/// short or does not describe a store `save` could have written: nothing
/// else stands between a damaged file and the store's own assumptions.
fn restore(store: *Store, bytes: []const u8) PersistenceError!void {
    var p = Parser{ .buf = bytes, .pos = 0 };

    var magic: [4]u8 = undefined;
    try p.readInto(&magic);
    if (!std.mem.eql(u8, &magic, &MAGIC)) return PersistenceError.BadMagic;
    const ver = try p.readU32();
    if (ver != VERSION) return PersistenceError.UnsupportedVersion;

    var data = Keyspace.init(store.allocator);
    errdefer freeKeyspace(store, &data);
    const n_entries = try p.readU64();
    var i: u64 = 0;
    while (i < n_entries) : (i += 1) {
        const key = try p.readBytesDuped(store.allocator);
        errdefer store.allocator.free(key);

        const expires_at_ns: ?i128 = switch (try p.readU8()) {
            0 => null,
            1 => try p.readI128(),
            else => return PersistenceError.Corrupt,
        };

        const tag_raw = try p.readU8();
        if (tag_raw > @backingInt(ValueTag.list)) return PersistenceError.InvalidVariant;
        const tag: ValueTag = @fromBackingInt(@intCast(tag_raw));
        var value = try readValueData(&p, store, tag);
        errdefer store.freeValueData(&value);

        const slot = try data.getOrPut(key);
        if (slot.found_existing) return PersistenceError.Corrupt;
        slot.value_ptr.* = .{ .data = value, .expires_at_ns = expires_at_ns };
    }

    // If the file ends here (older format that didn't write scripts), tolerate it.
    var scripts = Scripts.init(store.allocator);
    errdefer freeScripts(store, &scripts);
    if (p.pos < p.buf.len) {
        const n_scripts = try p.readU64();
        var j: u64 = 0;
        while (j < n_scripts) : (j += 1) {
            const sha = try p.readBytesDuped(store.allocator);
            errdefer store.allocator.free(sha);
            const body = try p.readBytesDuped(store.allocator);
            errdefer store.allocator.free(body);
            const slot = try scripts.getOrPut(sha);
            if (slot.found_existing) return PersistenceError.Corrupt;
            slot.value_ptr.* = body;
        }
    }

    store.clearZsetCache();
    freeKeyspace(store, &store.data);
    freeScripts(store, &store.scripts);
    store.data = data;
    store.scripts = scripts;
}

const Parser = struct {
    buf: []const u8,
    pos: usize,

    fn need(self: *Parser, n: u64) PersistenceError!void {
        // `pos + n` would wrap for a length a damaged file can hold
        if (n > self.buf.len - self.pos) return PersistenceError.Truncated;
    }

    /// A count of items that each take at least `item_bytes` of what is left.
    fn readCount(self: *Parser, item_bytes: usize) PersistenceError!usize {
        const n = try self.readU64();
        if (n > (self.buf.len - self.pos) / item_bytes) return PersistenceError.Truncated;
        return @intCast(n);
    }

    fn readInto(self: *Parser, dst: []u8) PersistenceError!void {
        try self.need(dst.len);
        @memcpy(dst, self.buf[self.pos..][0..dst.len]);
        self.pos += dst.len;
    }

    fn readU8(self: *Parser) PersistenceError!u8 {
        try self.need(1);
        const v = self.buf[self.pos];
        self.pos += 1;
        return v;
    }

    fn readU32(self: *Parser) PersistenceError!u32 {
        try self.need(4);
        const v = std.mem.readInt(u32, self.buf[self.pos..][0..4], .little);
        self.pos += 4;
        return v;
    }

    fn readU64(self: *Parser) PersistenceError!u64 {
        try self.need(8);
        const v = std.mem.readInt(u64, self.buf[self.pos..][0..8], .little);
        self.pos += 8;
        return v;
    }

    fn readI128(self: *Parser) PersistenceError!i128 {
        try self.need(16);
        const v = std.mem.readInt(i128, self.buf[self.pos..][0..16], .little);
        self.pos += 16;
        return v;
    }

    fn readF64(self: *Parser) PersistenceError!f64 {
        const bits = try self.readU64();
        return @bitCast(bits);
    }

    fn readBytesDuped(self: *Parser, alloc: Allocator) PersistenceError![]u8 {
        const len64 = try self.readU64();
        try self.need(len64);
        const len: usize = @intCast(len64);
        const out = try alloc.dupe(u8, self.buf[self.pos..][0..len]);
        self.pos += len;
        return out;
    }
};

fn freeFields(alloc: Allocator, fields: *store_mod.StreamFields) void {
    var it = fields.iterator();
    while (it.next()) |kv| {
        alloc.free(kv.key_ptr.*);
        alloc.free(kv.value_ptr.*);
    }
    fields.deinit();
}

/// Field-value pairs, as a hash and a stream entry both hold them. No field twice.
fn readFields(p: *Parser, alloc: Allocator) PersistenceError!store_mod.StreamFields {
    const n = try p.readCount(16);
    var fields = store_mod.StreamFields.init(alloc);
    errdefer freeFields(alloc, &fields);
    for (0..n) |_| {
        const f = try p.readBytesDuped(alloc);
        errdefer alloc.free(f);
        const v = try p.readBytesDuped(alloc);
        errdefer alloc.free(v);
        const slot = try fields.getOrPut(f);
        if (slot.found_existing) return PersistenceError.Corrupt;
        slot.value_ptr.* = v;
    }
    return fields;
}

fn readStreamId(p: *Parser) PersistenceError!StreamId {
    const ms = try p.readU64();
    return .{ .ms = ms, .seq = try p.readU64() };
}

fn readValueData(p: *Parser, store: *Store, tag: ValueTag) PersistenceError!store_mod.ValueData {
    const alloc = store.allocator;
    return switch (tag) {
        .string => .{ .string = try p.readBytesDuped(alloc) },
        .hash => blk: {
            var h = try readFields(p, alloc);
            errdefer freeFields(alloc, &h);
            // the store deletes a key when its last field goes
            if (h.count() == 0) return PersistenceError.Corrupt;
            break :blk .{ .hash = h };
        },
        .set => blk: {
            const n = try p.readCount(8);
            if (n == 0) return PersistenceError.Corrupt;
            var s = store_mod.BytesHashSet.init(alloc);
            errdefer {
                var it = s.keyIterator();
                while (it.next()) |k| alloc.free(k.*);
                s.deinit();
            }
            for (0..n) |_| {
                const m = try p.readBytesDuped(alloc);
                errdefer alloc.free(m);
                const slot = try s.getOrPut(m);
                if (slot.found_existing) return PersistenceError.Corrupt;
            }
            break :blk .{ .set = s };
        },
        .sorted_set => blk: {
            const n = try p.readCount(16);
            if (n == 0) return PersistenceError.Corrupt;
            var z = store_mod.SortedSet.init(alloc);
            errdefer z.deinit(alloc);
            try z.by_score.ensureTotalCapacity(alloc, n);
            var before: ?store_mod.ScoredMember = null;
            for (0..n) |_| {
                const score = try p.readF64();
                const member = try p.readBytesDuped(z.memberAllocator());
                const member_hash = store_mod.zset_index.hash(member);
                const entry: store_mod.ScoredMember = .{ .score = score, .member = member, .member_hash = member_hash };
                // saved in (score, member) order, each member once. NaN fails the first test.
                if (before) |b| {
                    if (!store_mod.ScoredMember.lessThan({}, b, entry)) return PersistenceError.Corrupt;
                } else if (std.math.isNan(score)) return PersistenceError.Corrupt;
                if (z.by_member.get(member) != null) return PersistenceError.Corrupt;
                try z.by_member.put(.{ .hash = member_hash, .bytes = member }, score);
                try z.by_score.appendAssumeCapacity(alloc, entry);
                before = entry;
            }
            break :blk .{ .sorted_set = z };
        },
        .stream => blk: {
            var s = store_mod.Stream.init(alloc);
            errdefer s.deinit(alloc);

            const n_entries = try p.readCount(24);
            var before: ?StreamId = null;
            for (0..n_entries) |_| {
                const id = try readStreamId(p);
                if (before) |b| if (b.cmp(id) != .lt) return PersistenceError.Corrupt;
                before = id;
                var fields = try readFields(p, alloc);
                errdefer freeFields(alloc, &fields);
                try s.entries.append(alloc, .{ .id = id, .fields = fields });
            }
            s.last_id = try readStreamId(p);
            if (before) |last| if (s.last_id.cmp(last) == .lt) return PersistenceError.Corrupt;

            const n_groups = try p.readCount(32);
            for (0..n_groups) |_| {
                const gname = try p.readBytesDuped(alloc);
                errdefer alloc.free(gname);
                var cg = store_mod.ConsumerGroup.init(alloc);
                errdefer cg.deinit(alloc);
                cg.last_delivered_id = try readStreamId(p);

                const n_cons = try p.readCount(16);
                for (0..n_cons) |_| {
                    const cname = try p.readBytesDuped(alloc);
                    errdefer alloc.free(cname);
                    var cons = store_mod.Consumer.init(alloc);
                    errdefer cons.deinit();
                    const n_pending = try p.readCount(40);
                    for (0..n_pending) |_| {
                        const id = try readStreamId(p);
                        const at = try p.readI128();
                        const dc = try p.readU64();
                        const slot = try cons.pending.getOrPut(id);
                        if (slot.found_existing) return PersistenceError.Corrupt;
                        slot.value_ptr.* = .{ .delivered_at_ns = at, .delivery_count = dc };
                    }
                    const slot = try cg.consumers.getOrPut(cname);
                    if (slot.found_existing) return PersistenceError.Corrupt;
                    slot.value_ptr.* = cons;
                }
                const slot = try s.groups.getOrPut(gname);
                if (slot.found_existing) return PersistenceError.Corrupt;
                slot.value_ptr.* = cg;
            }
            break :blk .{ .stream = s };
        },
        .list => blk: {
            const n = try p.readCount(8);
            if (n == 0) return PersistenceError.Corrupt;
            var list: std.DoublyLinkedList = .{};
            errdefer {
                while (list.popFirst()) |nd| {
                    const ln: *store_mod.ListNode = @fieldParentPtr("node", nd);
                    alloc.free(ln.data);
                    alloc.destroy(ln);
                }
            }
            for (0..n) |_| {
                const item = try p.readBytesDuped(alloc);
                errdefer alloc.free(item);
                const ln = try alloc.create(store_mod.ListNode);
                ln.* = .{ .data = item };
                list.append(&ln.node);
            }
            break :blk .{ .list = list };
        },
    };
}

// ----------------------------------------------------------------------------
// Buffer write helpers
// ----------------------------------------------------------------------------

fn writeU32(buf: *std.ArrayList(u8), a: Allocator, v: u32) PersistenceError!void {
    var b: [4]u8 = undefined;
    std.mem.writeInt(u32, &b, v, .little);
    try buf.appendSlice(a, &b);
}

fn writeU64(buf: *std.ArrayList(u8), a: Allocator, v: u64) PersistenceError!void {
    var b: [8]u8 = undefined;
    std.mem.writeInt(u64, &b, v, .little);
    try buf.appendSlice(a, &b);
}

fn writeI128(buf: *std.ArrayList(u8), a: Allocator, v: i128) PersistenceError!void {
    var b: [16]u8 = undefined;
    std.mem.writeInt(i128, &b, v, .little);
    try buf.appendSlice(a, &b);
}

fn writeF64(buf: *std.ArrayList(u8), a: Allocator, v: f64) PersistenceError!void {
    try writeU64(buf, a, @bitCast(v));
}

fn writeBytes(buf: *std.ArrayList(u8), a: Allocator, bytes: []const u8) PersistenceError!void {
    try writeU64(buf, a, bytes.len);
    try buf.appendSlice(a, bytes);
}

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

test "save + load roundtrips a mix of value variants" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();

    var path_buf: [128]u8 = undefined;
    const ns: i128 = std.Io.Timestamp.now(threaded.io(), .real).nanoseconds;
    const path = try std.fmt.bufPrint(&path_buf, "/tmp/burner-test-{x}.brdb", .{ns});
    defer std.Io.Dir.cwd().deleteFile(threaded.io(), path) catch {};

    // Original store: hand-crafted entries across every variant.
    {
        var s = Store.init(std.testing.allocator, threaded.io());
        defer s.deinit();

        _ = try s.set("a", "1", .{});
        _ = try s.hset("h", &.{ "f1", "v1", "f2", "v2" });
        _ = try s.sadd("st", &.{ "m1", "m2" });
        _ = try s.zadd("z", &.{
            .{ .score = 1, .member = "x" },
            .{ .score = 2, .member = "y" },
        }, .{});
        _ = try s.rpush("l", &.{ "a", "b", "c" });
        _ = try s.xadd("xs", .{ .ms = 1, .seq = 0 }, &.{ "k", "v" });
        try s.xgroupCreate("xs", "g1", .{ .id = StreamId.zero() });
        const got = try s.xreadgroupOne(std.testing.allocator, "xs", "g1", "c1", .new_messages, 10);
        defer store_mod.freeDeliveredEntries(std.testing.allocator, got);

        try save(&s, path);
    }

    // Restore into a fresh store; check that every variant came back.
    var s2 = Store.init(std.testing.allocator, threaded.io());
    defer s2.deinit();
    const loaded = try load(&s2, path);
    try std.testing.expect(loaded);

    const a_v = try s2.get(std.testing.allocator, "a");
    defer if (a_v) |x| std.testing.allocator.free(x);
    try std.testing.expectEqualStrings("1", a_v.?);

    const hpairs = try s2.hgetall(std.testing.allocator, "h");
    defer Store.freeHashPairs(std.testing.allocator, hpairs);
    try std.testing.expectEqual(@as(usize, 2), hpairs.len);

    try std.testing.expectEqual(@as(i64, 2), try s2.scard("st"));
    try std.testing.expectEqual(@as(i64, 2), try s2.zcard("z"));
    try std.testing.expectEqual(@as(i64, 3), try s2.llen("l"));
    try std.testing.expectEqual(@as(i64, 1), try s2.xlen("xs"));

    // The consumer group and its pending entry should have persisted too.
    const groups = try s2.xinfoGroups(std.testing.allocator, "xs");
    defer Store.freeGroupInfos(std.testing.allocator, groups);
    try std.testing.expectEqual(@as(usize, 1), groups.len);
    try std.testing.expectEqual(@as(i64, 1), groups[0].pending);
}

test "load returns false when the file doesn't exist" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();
    var s = Store.init(std.testing.allocator, threaded.io());
    defer s.deinit();

    const present = try load(&s, "/tmp/burner-does-not-exist.brdb");
    try std.testing.expect(!present);
}

/// a snapshot of one key, for the tests below to damage
const TestSnapshot = struct {
    buf: std.ArrayList(u8) = .empty,

    const a = std.testing.allocator;

    fn begin(key: []const u8, tag: ValueTag) !TestSnapshot {
        var snap: TestSnapshot = .{};
        errdefer snap.buf.deinit(a);
        try snap.buf.appendSlice(a, &MAGIC);
        try writeU32(&snap.buf, a, VERSION);
        try writeU64(&snap.buf, a, 1);
        try writeBytes(&snap.buf, a, key);
        try snap.buf.append(a, 0);
        try snap.buf.append(a, @backingInt(tag));
        return snap;
    }

    fn int(self: *TestSnapshot, v: u64) !void {
        try writeU64(&self.buf, a, v);
    }

    fn bytes(self: *TestSnapshot, v: []const u8) !void {
        try writeBytes(&self.buf, a, v);
    }

    /// what restoring it into a store that already holds `kept` does
    fn restoreOver(self: *TestSnapshot) !PersistenceError!void {
        defer self.buf.deinit(a);
        var threaded = std.Io.Threaded.init(a, .{});
        defer threaded.deinit();
        var s = Store.init(a, threaded.io());
        defer s.deinit();
        _ = try s.set("kept", "v", .{});
        const result = restore(&s, self.buf.items);
        // a snapshot that is refused leaves the store as it was
        if (result) |_| {} else |_| try std.testing.expectEqual(@as(i64, 1), s.exists(&.{"kept"}));
        return result;
    }
};

test "a length too large to add to the position is a truncated snapshot, not a read past the end" {
    var snap = try TestSnapshot.begin("k", .string);
    try snap.int(std.math.maxInt(u64));
    try std.testing.expectError(PersistenceError.Truncated, try snap.restoreOver());

    var counted = try TestSnapshot.begin("k", .sorted_set);
    try counted.int(std.math.maxInt(u64));
    try std.testing.expectError(PersistenceError.Truncated, try counted.restoreOver());
}

test "a stream whose ids do not ascend, or whose last id is below an entry, is refused" {
    var unordered = try TestSnapshot.begin("k", .stream);
    try unordered.int(2);
    for ([_]u64{ 2, 1 }) |ms| {
        try unordered.int(ms);
        try unordered.int(0);
        try unordered.int(0);
    }
    try unordered.int(2);
    try unordered.int(0);
    try unordered.int(0);
    try std.testing.expectError(PersistenceError.Corrupt, try unordered.restoreOver());

    var behind = try TestSnapshot.begin("k", .stream);
    try behind.int(1);
    try behind.int(5);
    try behind.int(0);
    try behind.int(0);
    try behind.int(4);
    try behind.int(0);
    try behind.int(0);
    try std.testing.expectError(PersistenceError.Corrupt, try behind.restoreOver());
}

test "a sorted set that is empty, out of order, or holds a member twice is refused" {
    var empty = try TestSnapshot.begin("k", .sorted_set);
    try empty.int(0);
    try empty.int(0);
    try std.testing.expectError(PersistenceError.Corrupt, try empty.restoreOver());

    for ([_][2]struct { f64, []const u8 }{
        .{ .{ 2, "a" }, .{ 1, "b" } },
        .{ .{ 1, "a" }, .{ 2, "a" } },
        .{ .{ std.math.nan(f64), "a" }, .{ 2, "b" } },
    }) |pair| {
        var snap = try TestSnapshot.begin("k", .sorted_set);
        try snap.int(2);
        for (pair) |entry| {
            try snap.int(@bitCast(entry[0]));
            try snap.bytes(entry[1]);
        }
        try snap.int(0);
        try std.testing.expectError(PersistenceError.Corrupt, try snap.restoreOver());
    }
}

test "a hash that is empty or holds a field twice is refused" {
    var empty = try TestSnapshot.begin("k", .hash);
    try empty.int(0);
    try empty.int(0);
    try std.testing.expectError(PersistenceError.Corrupt, try empty.restoreOver());

    var twice = try TestSnapshot.begin("k", .hash);
    try twice.int(2);
    for (0..2) |_| {
        try twice.bytes("f");
        try twice.bytes("v");
    }
    try twice.int(0);
    try std.testing.expectError(PersistenceError.Corrupt, try twice.restoreOver());
}

test "a save that fails leaves no temporary file" {
    var threaded = std.Io.Threaded.init(std.testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var s = Store.init(std.testing.allocator, io);
    defer s.deinit();
    _ = try s.set("a", "1", .{});

    // a directory where the snapshot should go: the rename onto it fails
    var path_buf: [128]u8 = undefined;
    const path = try std.fmt.bufPrint(&path_buf, "/tmp/burner-test-dir-{x}", .{std.Io.Timestamp.now(io, .real).nanoseconds});
    const cwd = std.Io.Dir.cwd();
    try cwd.createDir(io, path, .default_dir);
    defer cwd.deleteDir(io, path) catch {};

    try std.testing.expect(std.meta.isError(save(&s, path)));
    var tmp_buf: [140]u8 = undefined;
    const tmp = try std.fmt.bufPrint(&tmp_buf, "{s}.tmp", .{path});
    try std.testing.expectError(error.FileNotFound, cwd.access(io, tmp, .{}));
}