//! The single durable metadata database for Stream.
//!
//! Segment files own event bytes. Everything structured that cannot be
//! recovered cheaply from a segment scan belongs here. A synced metadata
//! batch may acknowledge segment bytes only after those bytes have been
//! fsynced; callers use `Batch.commitDurable` for that boundary.

const std = @import("std");
const rocksdb = @import("rocksdb");
const metrics = @import("../runtime/metrics.zig");

const Allocator = std.mem.Allocator;
const Io = std.Io;
const log = std.log.scoped(.stream_meta);

/// Deterministic, process-wide metadata fault seam used only by the offline
/// restart oracle. Production leaves this unarmed. The operation still runs
/// through the real Store and RocksDB schema; the selected durable mutation
/// returns an I/O failure immediately before it reaches RocksDB so callers'
/// fail-loud and restart behavior can be exercised without corrupting bytes.
const Fault = struct {
    prefix: ?[]const u8 = null,
    ordinal: u64 = 0,
    seen: std.atomic.Value(u64) = .init(0),

    fn check(self: *Fault, key: []const u8) !void {
        const prefix = self.prefix orelse return;
        if (!std.mem.startsWith(u8, key, prefix)) return;
        const seen = self.seen.fetchAdd(1, .monotonic) + 1;
        if (seen != self.ordinal) return;
        log.warn("store fault: injected I/O failure for prefix {s} at ordinal {d}", .{ prefix, seen });
        return error.InjectedStoreFault;
    }
};

var fault: Fault = .{};
var read_fault: Fault = .{};

/// Arm one matching durable metadata operation. Must be called during process
/// startup, before worker threads exist. Returns false for an empty prefix or
/// zero ordinal so an invalid oracle configuration fails closed.
pub fn armFault(prefix: []const u8, ordinal: u64) bool {
    if (prefix.len == 0 or ordinal == 0) return false;
    fault = .{ .prefix = prefix, .ordinal = ordinal };
    return true;
}

pub fn disarmFault() void {
    fault = .{};
}

/// Independent read-side fault seam. Keeping it separate from durable mutation
/// faults preserves the existing crash schedules while allowing startup code
/// to prove that a storage error is not misclassified as a missing key.
pub fn armReadFault(prefix: []const u8, ordinal: u64) bool {
    if (prefix.len == 0 or ordinal == 0) return false;
    read_fault = .{ .prefix = prefix, .ordinal = ordinal };
    return true;
}

pub fn disarmReadFault() void {
    read_fault = .{};
}

pub const subdir = "meta.rocksdb";

pub const Store = struct {
    allocator: Allocator,
    db: rocksdb.DB,
    default_cf: rocksdb.ColumnFamilyHandle,
    io: ?Io = null,
    stats: ?*metrics.Stats = null,

    pub fn open(allocator: Allocator, data_dir: []const u8) !Store {
        return openInner(allocator, data_dir, null, null);
    }

    pub fn openWithStats(allocator: Allocator, io: Io, data_dir: []const u8, stats: *metrics.Stats) !Store {
        return openInner(allocator, data_dir, io, stats);
    }

    fn openInner(allocator: Allocator, data_dir: []const u8, io: ?Io, stats: ?*metrics.Stats) !Store {
        const path = try std.fs.path.join(allocator, &.{ data_dir, subdir });
        defer allocator.free(path);

        var err_data: ?rocksdb.Data = null;
        defer if (err_data) |e| e.deinit();
        const db, const families = rocksdb.DB.open(
            allocator,
            path,
            .{
                .create_if_missing = true,
                .create_missing_column_families = true,
            },
            &.{.{ .name = "default", .options = .{ .bloom_bits_per_key = 10 } }},
            false,
            &err_data,
        ) catch |err| {
            if (err_data) |e| log.err("open {s}: {s}", .{ path, e.data });
            return err;
        };
        defer allocator.free(families);
        if (families.len != 1) return error.InvalidMetadataStore;

        log.info("metadata store opened at {s}", .{path});
        return .{
            .allocator = allocator,
            .db = db,
            .default_cf = families[0].handle,
            .io = io,
            .stats = stats,
        };
    }

    pub fn deinit(self: *Store) void {
        self.db.deinit();
    }

    /// One-way import for canaries created before the unified metadata DB.
    /// The legacy files remain untouched as rollback material; existing DB
    /// keys always win, making this safe on every restart.
    pub fn migrateLegacy(self: *Store, io: std.Io, data_dir: []const u8) !void {
        if (try self.getAlloc(self.allocator, "relay/cursor")) |existing| {
            defer self.allocator.free(existing);
            // Stream previously stored a bare signed little-endian value.
            // Convert that known schema exactly once; all other malformed or
            // unknown-version values remain untouched so the strict loader
            // fails closed instead of laundering corruption.
            if (existing.len == 8) {
                const seq = std.mem.readInt(i64, existing[0..8], .little);
                if (seq < 0) return error.NegativeRelayCursor;
                var versioned: [9]u8 = undefined;
                versioned[0] = 1;
                std.mem.writeInt(u64, versioned[1..9], @intCast(seq), .little);
                try self.putDurable("relay/cursor", &versioned);
                log.info("migrated relay cursor to versioned encoding", .{});
            }
        } else {
            var path_buf: [std.fs.max_path_bytes]u8 = undefined;
            const path = try std.fmt.bufPrint(&path_buf, "{s}/cursor", .{data_dir});
            var text_buf: [64]u8 = undefined;
            if (std.Io.Dir.cwd().readFile(io, path, &text_buf)) |text| {
                if (std.fmt.parseInt(i64, std.mem.trim(u8, text, " \n\t"), 10)) |seq| {
                    if (seq < 0) return error.NegativeRelayCursor;
                    var bytes: [9]u8 = undefined;
                    bytes[0] = 1;
                    std.mem.writeInt(u64, bytes[1..9], @intCast(seq), .little);
                    try self.putDurable("relay/cursor", &bytes);
                    log.info("imported legacy relay cursor {d}", .{seq});
                } else |_| {}
            } else |_| {}
        }

        if (try self.getAlloc(self.allocator, "phase")) |existing| {
            self.allocator.free(existing);
        } else {
            var path_buf: [std.fs.max_path_bytes]u8 = undefined;
            const path = try std.fmt.bufPrint(&path_buf, "{s}/phase", .{data_dir});
            var text_buf: [32]u8 = undefined;
            if (std.Io.Dir.cwd().readFile(io, path, &text_buf)) |text| {
                const phase = std.mem.trim(u8, text, " \n\t");
                if (phase.len > 0) try self.putDurable("phase", phase);
            } else |_| {}
        }

        if (try self.getAlloc(self.allocator, "compaction/seq")) |existing| {
            self.allocator.free(existing);
        } else {
            var path_buf: [std.fs.max_path_bytes]u8 = undefined;
            const path = try std.fmt.bufPrint(&path_buf, "{s}/compaction_watermark", .{data_dir});
            var bytes: [9]u8 = undefined;
            if (std.Io.Dir.cwd().readFile(io, path, &bytes)) |value| {
                if (value.len == bytes.len) try self.putDurable("compaction/seq", value);
            } else |_| {}
        }

        if (try self.getAlloc(self.allocator, "merge/next_source_idx")) |existing| {
            defer self.allocator.free(existing);
            if (existing.len == 8) {
                var versioned: [9]u8 = undefined;
                versioned[0] = 1;
                @memcpy(versioned[1..9], existing);
                try self.putDurable("merge/next_source_idx", &versioned);
                log.info("migrated merge cursor to versioned encoding", .{});
            }
        }
    }

    pub fn getAlloc(self: *Store, allocator: Allocator, key: []const u8) !?[]u8 {
        const started_at = self.started();
        read_fault.check(key) catch |err| {
            self.observe(.get, .error_result, started_at);
            return err;
        };
        var err_data: ?rocksdb.Data = null;
        defer if (err_data) |e| e.deinit();
        const value = self.db.get(self.default_cf, key, &err_data) catch |err| {
            self.observe(.get, .error_result, started_at);
            if (err_data) |e| log.err("get {s}: {s}", .{ key, e.data });
            return err;
        } orelse {
            self.observe(.get, .notfound, started_at);
            return null;
        };
        self.observe(.get, .ok, started_at);
        defer value.deinit();
        return try allocator.dupe(u8, value.data);
    }

    pub fn putDurable(self: *Store, key: []const u8, value: []const u8) !void {
        const started_at = self.started();
        fault.check(key) catch |err| {
            self.observe(.set, .error_result, started_at);
            return err;
        };
        var err_data: ?rocksdb.Data = null;
        defer if (err_data) |e| e.deinit();
        self.db.putWithOptions(self.default_cf, key, value, .{ .sync = true }, &err_data) catch |err| {
            self.observe(.set, .error_result, started_at);
            if (err_data) |e| log.err("durable put {s}: {s}", .{ key, e.data });
            return err;
        };
        self.observe(.set, .ok, started_at);
    }

    pub fn deleteDurable(self: *Store, key: []const u8) !void {
        const started_at = self.started();
        fault.check(key) catch |err| {
            self.observe(.delete, .error_result, started_at);
            return err;
        };
        var err_data: ?rocksdb.Data = null;
        defer if (err_data) |e| e.deinit();
        self.db.deleteWithOptions(self.default_cf, key, .{ .sync = true }, &err_data) catch |err| {
            self.observe(.delete, .error_result, started_at);
            if (err_data) |e| log.err("durable delete {s}: {s}", .{ key, e.data });
            return err;
        };
        self.observe(.delete, .ok, started_at);
    }

    pub fn iterator(self: *Store, start: ?[]const u8) rocksdb.Iterator {
        return self.db.iterator(self.default_cf, .forward, start);
    }

    pub fn batch(self: *Store) Batch {
        return .{ .store = self, .inner = rocksdb.WriteBatch.init() };
    }

    fn started(self: *const Store) ?Io.Timestamp {
        const io = self.io orelse return null;
        return Io.Timestamp.now(io, .awake);
    }

    fn observe(self: *Store, op: metrics.StoreOp, status: metrics.StoreStatus, started_at: ?Io.Timestamp) void {
        const stats = self.stats orelse return;
        const io = self.io orelse return;
        const started_value = started_at orelse return;
        const elapsed = started_value.durationTo(Io.Timestamp.now(io, .awake)).toMicroseconds();
        stats.observeStore(op, status, @intCast(@max(elapsed, 0)));
    }
};

pub const Batch = struct {
    store: *Store,
    inner: rocksdb.WriteBatch,
    fault_key: ?[]const u8 = null,

    pub fn deinit(self: *Batch) void {
        self.inner.deinit();
    }

    pub fn put(self: *Batch, key: []const u8, value: []const u8) void {
        if (fault.prefix) |prefix| {
            if (self.fault_key == null and std.mem.startsWith(u8, key, prefix)) self.fault_key = key;
        }
        self.inner.put(self.store.default_cf, key, value);
    }

    pub fn delete(self: *Batch, key: []const u8) void {
        if (fault.prefix) |prefix| {
            if (self.fault_key == null and std.mem.startsWith(u8, key, prefix)) self.fault_key = key;
        }
        self.inner.delete(self.store.default_cf, key);
    }

    pub fn commitDurable(self: *Batch) !void {
        const started_at = self.store.started();
        if (self.fault_key) |key| fault.check(key) catch |err| {
            self.store.observe(.batch_commit, .error_result, started_at);
            return err;
        };
        var err_data: ?rocksdb.Data = null;
        defer if (err_data) |e| e.deinit();
        self.store.db.writeWithOptions(self.inner, .{ .sync = true }, &err_data) catch |err| {
            self.store.observe(.batch_commit, .error_result, started_at);
            if (err_data) |e| log.err("durable batch commit: {s}", .{e.data});
            return err;
        };
        self.store.observe(.batch_commit, .ok, started_at);
    }
};

test "durable point reads and atomic batches survive reopen" {
    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);
    const path = path_buf[0..path_len];

    {
        var store = try Store.open(testing.allocator, path);
        defer store.deinit();
        var batch = store.batch();
        defer batch.deinit();
        batch.put("relay/cursor", "41");
        batch.put("repo/did:plc:alice", "complete\x003lrev");
        try batch.commitDurable();
    }
    {
        var store = try Store.open(testing.allocator, path);
        defer store.deinit();
        const cursor = (try store.getAlloc(testing.allocator, "relay/cursor")).?;
        defer testing.allocator.free(cursor);
        try testing.expectEqualStrings("41", cursor);
        const repo = (try store.getAlloc(testing.allocator, "repo/did:plc:alice")).?;
        defer testing.allocator.free(repo);
        try testing.expectEqualStrings("complete\x003lrev", repo);
    }
}

test "real RocksDB operations populate canonical store metric outcomes" {
    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 stats: metrics.Stats = .{};
    var store = try Store.openWithStats(testing.allocator, testing.io, path_buf[0..path_len], &stats);
    defer store.deinit();

    try store.putDurable("point", "value");
    const found = (try store.getAlloc(testing.allocator, "point")).?;
    testing.allocator.free(found);
    try testing.expectEqual(null, try store.getAlloc(testing.allocator, "missing"));
    try store.deleteDurable("point");
    var writes = store.batch();
    defer writes.deinit();
    writes.put("batch/a", "1");
    writes.put("batch/b", "2");
    try writes.commitDurable();

    try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.get, .ok)].load(.monotonic));
    try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.get, .notfound)].load(.monotonic));
    try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.set, .ok)].load(.monotonic));
    try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.delete, .ok)].load(.monotonic));
    try testing.expectEqual(1, stats.store_op_duration_count[storeMetricIndex(.batch_commit, .ok)].load(.monotonic));
}

test "armed durable store fault fires once at matching batch ordinal" {
    const testing = std.testing;
    disarmFault();
    defer disarmFault();
    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 store = try Store.open(testing.allocator, path_buf[0..path_len]);
    defer store.deinit();

    try testing.expect(armFault("merge/next_source_idx", 2));
    for (0..2) |i| {
        var writes = store.batch();
        defer writes.deinit();
        writes.put("repo/did:plc:alice", "complete");
        writes.put("merge/next_source_idx", "1");
        if (i == 0) {
            try writes.commitDurable();
        } else {
            try testing.expectError(error.InjectedStoreFault, writes.commitDurable());
        }
    }

    // The rejected batch never reached RocksDB. Disarming and retrying the
    // exact real batch succeeds, which is the recovery contract used by the
    // process oracle.
    disarmFault();
    var retry = store.batch();
    defer retry.deinit();
    retry.put("merge/next_source_idx", "2");
    try retry.commitDurable();
    const cursor = (try store.getAlloc(testing.allocator, "merge/next_source_idx")).?;
    defer testing.allocator.free(cursor);
    try testing.expectEqualStrings("2", cursor);
}

fn storeMetricIndex(op: metrics.StoreOp, status: metrics.StoreStatus) usize {
    return @backingInt(op) * @typeInfo(metrics.StoreStatus).@"enum".field_names.len + @backingInt(status);
}