//! Phase B timestamp-import bucketing.
//!
//! A concrete sealed-segment catalog loads the resident DID blooms without
//! reading segment bodies. Valid CSV rows are routed to per-segment packed
//! offset files through bounded least-recently-used DID and file caches. The
//! import lifecycle holds the archive/compaction topology lock while this
//! point-in-time catalog is in use; rows born after the preceding force-seal
//! are stamped by the durable append rule store.

const std = @import("std");
const gloom = @import("../storage/gloom.zig");
const segment = @import("../storage/segment.zig");
const segment_io = @import("../storage/segment_io.zig");
const parse = @import("parse.zig");

const Allocator = std.mem.Allocator;
const Io = std.Io;

pub const default_did_cache_size: usize = 100_000;
pub const default_open_file_limit: usize = 256;
pub const offset_record_size = 8;

pub const Stats = struct {
    rows_routed: u64 = 0,
    rows_no_candidate: u64 = 0,
    offsets_written: u64 = 0,
    did_cache_hits: u64 = 0,
    did_cache_misses: u64 = 0,
    segments_touched: usize = 0,
    selector_calls: u64 = 0,
};

pub const SegmentRef = struct {
    index: u64,
    name: []u8,
    bloom: gloom.Filter,
};

/// Immutable point-in-time view of sealed segments. Opening fails on an
/// unreadable resident bloom: treating corruption as a miss could silently
/// strand timestamp intent.
pub const Catalog = struct {
    allocator: Allocator,
    io: Io,
    segments: std.ArrayList(SegmentRef) = .empty,

    pub fn open(allocator: Allocator, io: Io, segments_dir: Io.Dir) !Catalog {
        var self: Catalog = .{ .allocator = allocator, .io = io };
        errdefer self.deinit();
        var iterator = segments_dir.iterate();
        while (try iterator.next(io)) |entry| {
            const index = parseSegmentName(entry.name) orelse continue;
            var file = try segments_dir.openFile(io, entry.name, .{});
            defer file.close(io);
            var header_bytes: [segment.header_size]u8 = undefined;
            if (try file.readPositionalAll(io, &header_bytes, 0) != header_bytes.len)
                return error.TruncatedSegmentHeader;
            const header = segment.Header.decode(&header_bytes) catch |err| switch (err) {
                error.ActiveSegment => continue,
                else => return err,
            };
            if (header.block_did_bloom_offset < header.did_bloom_offset)
                return error.BadSegmentBloomOffsets;
            const bloom_len = header.block_did_bloom_offset - header.did_bloom_offset;
            if (bloom_len > 1 << 30) return error.BadSegmentBloomOffsets;
            const bloom_bytes = try allocator.alloc(u8, @intCast(bloom_len));
            defer allocator.free(bloom_bytes);
            if (try file.readPositionalAll(io, bloom_bytes, header.did_bloom_offset) != bloom_bytes.len)
                return error.TruncatedSegmentBloom;
            const filter = try gloom.unmarshal(allocator, bloom_bytes);
            errdefer {
                var owned = filter;
                owned.deinit(allocator);
            }
            const name = try allocator.dupe(u8, entry.name);
            errdefer allocator.free(name);
            try self.segments.append(allocator, .{
                .index = index,
                .name = name,
                .bloom = filter,
            });
        }
        std.mem.sort(SegmentRef, self.segments.items, {}, struct {
            fn less(_: void, a: SegmentRef, b: SegmentRef) bool {
                return a.index < b.index;
            }
        }.less);
        return self;
    }

    pub fn deinit(self: *Catalog) void {
        for (self.segments.items) |*item| {
            self.allocator.free(item.name);
            item.bloom.deinit(self.allocator);
        }
        self.segments.deinit(self.allocator);
        self.* = undefined;
    }

    fn select(self: *const Catalog, allocator: Allocator, did: []const u8) ![]usize {
        var candidates: std.ArrayList(usize) = .empty;
        errdefer candidates.deinit(allocator);
        for (self.segments.items, 0..) |*item, index| {
            if (item.bloom.contains(did)) try candidates.append(allocator, index);
        }
        return candidates.toOwnedSlice(allocator);
    }
};

const CacheEntry = struct {
    candidates: []usize,
    used: u64,
};

const OpenOffset = struct {
    segment_index: u64,
    file: Io.File,
    size: u64,
    used: u64,
};

pub const Config = struct {
    did_cache_size: usize = 0,
    open_file_limit: usize = 0,
};

pub const Bucketer = struct {
    allocator: Allocator,
    io: Io,
    catalog: *const Catalog,
    job_path: []u8,
    job_dir: Io.Dir,
    did_cache_size: usize,
    open_file_limit: usize,
    cache: std.StringHashMapUnmanaged(CacheEntry) = .empty,
    files: std.ArrayList(OpenOffset) = .empty,
    touched: std.AutoHashMapUnmanaged(u64, void) = .empty,
    clock: u64 = 0,
    stats: Stats = .{},
    closed: bool = false,

    pub fn init(
        allocator: Allocator,
        io: Io,
        catalog: *const Catalog,
        job_path: []const u8,
        config: Config,
    ) !Bucketer {
        if (job_path.len == 0) return error.MissingJobPath;
        try Io.Dir.cwd().createDirPath(io, job_path);
        const dir = try Io.Dir.cwd().openDir(io, job_path, .{ .iterate = true });
        errdefer dir.close(io);
        return .{
            .allocator = allocator,
            .io = io,
            .catalog = catalog,
            .job_path = try allocator.dupe(u8, job_path),
            .job_dir = dir,
            .did_cache_size = if (config.did_cache_size == 0) default_did_cache_size else config.did_cache_size,
            .open_file_limit = if (config.open_file_limit == 0) default_open_file_limit else config.open_file_limit,
        };
    }

    pub fn deinit(self: *Bucketer) void {
        if (!self.closed) self.close() catch {};
        var iterator = self.cache.iterator();
        while (iterator.next()) |entry| {
            self.allocator.free(entry.key_ptr.*);
            self.allocator.free(entry.value_ptr.candidates);
        }
        self.cache.deinit(self.allocator);
        self.files.deinit(self.allocator);
        self.touched.deinit(self.allocator);
        self.job_dir.close(self.io);
        self.allocator.free(self.job_path);
        self.* = undefined;
    }

    pub fn routeOpaque(raw: ?*anyopaque, row: parse.Row) anyerror!void {
        const self: *Bucketer = @ptrCast(@alignCast(raw.?));
        try self.route(row);
    }

    pub fn route(self: *Bucketer, row: parse.Row) !void {
        if (self.closed) return error.BucketerClosed;
        const candidates = try self.select(row.did);
        if (candidates.len == 0) {
            self.stats.rows_no_candidate += 1;
            return;
        }
        for (candidates) |candidate| {
            try self.appendOffset(self.catalog.segments.items[candidate].index, row.offset);
            self.stats.offsets_written += 1;
        }
        self.stats.rows_routed += 1;
    }

    fn select(self: *Bucketer, did: []const u8) ![]const usize {
        self.clock += 1;
        if (self.cache.getPtr(did)) |entry| {
            entry.used = self.clock;
            self.stats.did_cache_hits += 1;
            return entry.candidates;
        }
        self.stats.did_cache_misses += 1;
        self.stats.selector_calls += 1;
        if (self.cache.count() >= self.did_cache_size) self.evictDID();
        const owned_did = try self.allocator.dupe(u8, did);
        errdefer self.allocator.free(owned_did);
        const candidates = try self.catalog.select(self.allocator, did);
        errdefer self.allocator.free(candidates);
        try self.cache.put(self.allocator, owned_did, .{ .candidates = candidates, .used = self.clock });
        return self.cache.get(did).?.candidates;
    }

    fn evictDID(self: *Bucketer) void {
        var oldest_key: ?[]const u8 = null;
        var oldest_tick: u64 = std.math.maxInt(u64);
        var iterator = self.cache.iterator();
        while (iterator.next()) |entry| {
            if (entry.value_ptr.used < oldest_tick) {
                oldest_tick = entry.value_ptr.used;
                oldest_key = entry.key_ptr.*;
            }
        }
        if (oldest_key) |key| {
            const removed = self.cache.fetchRemove(key).?;
            self.allocator.free(removed.key);
            self.allocator.free(removed.value.candidates);
        }
    }

    fn appendOffset(self: *Bucketer, segment_index: u64, offset: u64) !void {
        self.clock += 1;
        var slot: ?usize = null;
        for (self.files.items, 0..) |item, index| {
            if (item.segment_index == segment_index) {
                slot = index;
                break;
            }
        }
        if (slot == null) {
            if (self.files.items.len >= self.open_file_limit) try self.evictFile();
            var name_buffer: [64]u8 = undefined;
            const name = try offsetFileName(&name_buffer, segment_index);
            const file = try self.job_dir.createFile(self.io, name, .{ .read = true, .truncate = false });
            errdefer file.close(self.io);
            const stat = try file.stat(self.io);
            try self.files.append(self.allocator, .{
                .segment_index = segment_index,
                .file = file,
                .size = stat.size,
                .used = self.clock,
            });
            slot = self.files.items.len - 1;
        }
        const open = &self.files.items[slot.?];
        open.used = self.clock;
        var encoded: [offset_record_size]u8 = undefined;
        std.mem.writeInt(u64, &encoded, offset, .little);
        try open.file.writePositionalAll(self.io, &encoded, open.size);
        open.size += encoded.len;
        try self.touched.put(self.allocator, segment_index, {});
    }

    fn evictFile(self: *Bucketer) !void {
        var oldest: usize = 0;
        for (self.files.items, 0..) |item, index| {
            if (item.used < self.files.items[oldest].used) oldest = index;
        }
        const item = self.files.items[oldest];
        try item.file.sync(self.io);
        item.file.close(self.io);
        _ = self.files.orderedRemove(oldest);
    }

    /// Sync all offset contents, close them, then sync the job directory and
    /// its parent before the lifecycle may persist `bucketed=true`.
    pub fn close(self: *Bucketer) !void {
        if (self.closed) return;
        var first_error: ?anyerror = null;
        for (self.files.items) |item| {
            item.file.sync(self.io) catch |err| if (first_error == null) {
                first_error = err;
            };
            item.file.close(self.io);
        }
        self.files.clearRetainingCapacity();
        if (first_error) |err| return err;
        try syncDir(self.io, self.job_dir);
        const parent_path = std.fs.path.dirname(self.job_path) orelse ".";
        // On Linux Zig's default directory handle may be O_PATH, which cannot
        // be fsynced (EBADF/Unexpected).  Request an iterable directory so the
        // parent-dir durability barrier uses a real readable fd.
        var parent = try Io.Dir.cwd().openDir(self.io, parent_path, .{ .iterate = true });
        defer parent.close(self.io);
        try syncDir(self.io, parent);
        self.closed = true;
    }

    pub fn statsSnapshot(self: *const Bucketer) Stats {
        var result = self.stats;
        result.segments_touched = self.touched.count();
        return result;
    }
};

pub fn offsetFileName(buffer: *[64]u8, index: u64) ![]const u8 {
    return std.fmt.bufPrint(buffer, "offsets_{d:0>10}.bin", .{index});
}

pub fn parseOffsetFileName(name: []const u8) ?u64 {
    if (!std.mem.startsWith(u8, name, "offsets_") or !std.mem.endsWith(u8, name, ".bin")) return null;
    const digits = name[8 .. name.len - 4];
    if (digits.len != 10) return null;
    const index = std.fmt.parseInt(u64, digits, 10) catch return null;
    var buffer: [64]u8 = undefined;
    const canonical = offsetFileName(&buffer, index) catch return null;
    return if (std.mem.eql(u8, name, canonical)) index else null;
}

fn parseSegmentName(name: []const u8) ?u64 {
    if (!std.mem.startsWith(u8, name, "seg_") or !std.mem.endsWith(u8, name, ".jss")) return null;
    const digits = name[4 .. name.len - 4];
    if (digits.len != 10) return null;
    // Archive segment ordinals are fixed-width lowercase base-36. Decimal
    // parsing happens to work for the first ten files, then silently drops
    // every filename containing a-z from the timestamp-import catalog.
    const index = std.fmt.parseInt(u64, digits, 36) catch return null;
    var canonical_digits: [10]u8 = @splat('0');
    var value = index;
    var position: usize = canonical_digits.len;
    while (value > 0) : (value /= 36) {
        if (position == 0) return null;
        position -= 1;
        canonical_digits[position] = std.fmt.digitToChar(@intCast(value % 36), .lower);
    }
    return if (std.mem.eql(u8, digits, &canonical_digits)) index else null;
}

fn syncDir(io: Io, dir: Io.Dir) !void {
    try segment_io.syncDir(dir, io);
}

test "catalog discovers real base-36 segment names beyond nine" {
    const testing = std.testing;
    const writer = @import("../storage/segment_writer.zig");
    var threaded: Io.Threaded = .init(testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var segments_dir = try tmp.dir.createDirPathOpen(io, "segments", .{
        .open_options = .{ .iterate = true },
    });
    defer segments_dir.close(io);

    var active = try writer.ActiveWriter.init(testing.allocator);
    defer active.deinit();
    try active.append(.{
        .seq = 1,
        .witnessed_at = 1,
        .indexed_at = 0,
        .kind = .create,
        .did = "did:plc:base36",
        .collection = "app.bsky.feed.post",
        .rkey = "one",
        .rev = "3lbase36",
        .payload = "payload",
    });
    try active.seal();
    var file = try segments_dir.createFile(io, "seg_000000000a.jss", .{});
    defer file.close(io);
    try file.writeStreamingAll(io, active.bytes());
    try file.sync(io);

    var catalog = try Catalog.open(testing.allocator, io, segments_dir);
    defer catalog.deinit();
    try testing.expectEqual(@as(usize, 1), catalog.segments.items.len);
    try testing.expectEqual(@as(u64, 10), catalog.segments.items[0].index);
    try testing.expect(catalog.segments.items[0].bloom.contains("did:plc:base36"));
}