//! Durable physical-compaction watermark in the shared metadata database.

const std = @import("std");
const meta_store = @import("../storage/meta_store.zig");

const log = std.log.scoped(.stream);
pub const version: u8 = 1;
const key = "compaction/seq";

pub const Store = struct {
    meta: *meta_store.Store,

    pub fn init(meta: *meta_store.Store) Store {
        return .{ .meta = meta };
    }
    pub fn deinit(_: *Store, _: std.Io) void {}

    pub fn load(self: *Store, _: std.Io) !?u64 {
        const maybe_bytes = try self.meta.getAlloc(self.meta.allocator, key);
        const bytes = maybe_bytes orelse return null;
        defer self.meta.allocator.free(bytes);
        if (bytes.len != 9) return error.InvalidCompactionWatermarkLength;
        if (bytes[0] != version) return error.UnsupportedCompactionWatermarkVersion;
        return std.mem.readInt(u64, bytes[1..9], .little);
    }

    pub fn save(self: *Store, _: std.Io, watermark: u64) !void {
        var payload: [9]u8 = undefined;
        payload[0] = version;
        std.mem.writeInt(u64, payload[1..9], watermark, .little);
        try self.meta.putDurable(key, &payload);
    }

    pub fn initFloor(self: *Store, io: std.Io, next_seq: u64) !u64 {
        if (try self.load(io)) |w| return w;
        const w = next_seq -| 1;
        try self.save(io, w);
        log.info("compaction watermark floored at {d}", .{w});
        return w;
    }
};

test "watermark floor and updates" {
    const testing = std.testing;
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
    const path_len = try tmp.dir.realPath(testing.io, &path_buf);
    var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]);
    defer meta.deinit();
    var store = Store.init(&meta);
    try testing.expectEqual(@as(?u64, null), try store.load(testing.io));
    try testing.expectEqual(@as(u64, 0), try store.initFloor(testing.io, 0));
    try store.save(testing.io, 99);
    try testing.expectEqual(@as(?u64, 99), try store.load(testing.io));
}

test "watermark corruption and storage errors fail closed without initialization" {
    const testing = std.testing;
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var path_buf: [std.fs.max_path_bytes]u8 = undefined;
    const path_len = try tmp.dir.realPath(testing.io, &path_buf);
    var meta = try meta_store.Store.open(testing.allocator, path_buf[0..path_len]);
    defer meta.deinit();
    var store = Store.init(&meta);

    try meta.putDurable(key, &.{ 1, 2, 3 });
    try testing.expectError(error.InvalidCompactionWatermarkLength, store.initFloor(testing.io, 100));
    const malformed = (try meta.getAlloc(testing.allocator, key)).?;
    defer testing.allocator.free(malformed);
    try testing.expectEqualSlices(u8, &.{ 1, 2, 3 }, malformed);

    try meta.putDurable(key, &.{ 2, 0, 0, 0, 0, 0, 0, 0, 0 });
    try testing.expectError(error.UnsupportedCompactionWatermarkVersion, store.load(testing.io));
    try testing.expect(meta_store.armReadFault(key, 1));
    defer meta_store.disarmReadFault();
    try testing.expectError(error.InjectedStoreFault, store.initFloor(testing.io, 100));
}