//! Offline operator commands shared by the executable and unit receipts.
//!
//! `inspect-segment` intentionally reads only local JSS bytes. It follows the
//! upstream inspection contract for sealed, checksum-corrupt, and active
//! files; it never opens RocksDB or starts the network service.

const std = @import("std");
const build_options = @import("build_options");
const segment = @import("../storage/segment.zig");
const segment_writer = @import("../storage/segment_writer.zig");

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

pub const BlocksMode = enum { summary, table, full };

const Collection = struct {
    nsid: []const u8,
    event_count: u32,
};

pub const Inspection = struct {
    arena: std.heap.ArenaAllocator,
    path: []const u8,
    file_size: usize,
    sealed: bool,
    header: ?segment.Header,
    checksum_valid: bool = false,
    blocks: []const segment.BlockIndexEntry,
    collections: []const Collection,
    block_collections: []const []const u32,
    total_events: u64,
    unique_did_count: u32,
    min_seq: u64,
    max_seq: u64,
    min_witnessed_at: i64,
    max_witnessed_at: i64,
    block_index_bytes: u64 = 0,
    segment_bloom_bytes: u64 = 0,
    block_blooms_bytes: u64 = 0,
    collection_index_bytes: u64 = 0,
    per_block_bloom_bytes: u32 = 0,
    partial_error: ?anyerror = null,

    pub fn deinit(self: *Inspection) void {
        self.arena.deinit();
    }
};

pub fn renderVersion(writer: *Io.Writer) !void {
    try writer.print("jetstream version {s} (commit {s}, built {s})\n", .{
        build_options.version,
        build_options.git_sha,
        build_options.build_date,
    });
}

pub fn inspectSegment(allocator: Allocator, io: Io, path: []const u8) !Inspection {
    const bytes = try Io.Dir.cwd().readFileAlloc(io, path, allocator, .limited(1 << 31));
    defer allocator.free(bytes);
    if (bytes.len < segment.header_size) return error.Truncated;
    if (!std.mem.eql(u8, bytes[0..4], segment.magic)) return error.BadMagic;

    var arena = std.heap.ArenaAllocator.init(allocator);
    errdefer arena.deinit();
    const alloc = arena.allocator();
    const owned_path = try alloc.dupe(u8, path);
    const checksum = std.mem.readInt(u64, bytes[4..12], .little);
    if (checksum == 0) return inspectActive(&arena, owned_path, bytes);
    return inspectSealed(&arena, owned_path, bytes);
}

fn inspectSealed(
    arena: *std.heap.ArenaAllocator,
    path: []const u8,
    bytes: []const u8,
) !Inspection {
    const alloc = arena.allocator();
    var reader = try segment.Sealed.parseUnchecked(alloc, bytes);
    defer reader.deinit(alloc);
    var collections = try reader.readCollectionIndex(alloc);
    defer collections.deinit(alloc);

    const blocks = try alloc.dupe(segment.BlockIndexEntry, reader.block_index);
    const collection_rows = try alloc.alloc(Collection, collections.entries.len);
    for (collections.entries, collection_rows) |entry, *row| {
        row.* = .{
            .nsid = try alloc.dupe(u8, entry.nsid),
            .event_count = entry.count,
        };
    }
    const block_collections = try alloc.alloc([]const u32, blocks.len);
    for (block_collections, 0..) |*ids, block_index| {
        var found: std.ArrayList(u32) = .empty;
        for (collections.entries, 0..) |_, collection_id| {
            if (collections.blockHasCollection(block_index, collection_id))
                try found.append(alloc, @intCast(collection_id));
        }
        ids.* = try found.toOwnedSlice(alloc);
    }

    const header = reader.header;
    const block_index_bytes = header.did_bloom_offset - header.block_index_offset;
    const segment_bloom_bytes = header.block_did_bloom_offset - header.did_bloom_offset;
    const block_blooms_bytes = header.collection_index_offset - header.block_did_bloom_offset;
    const collection_index_bytes = bytes.len - header.collection_index_offset;
    var per_block_bloom_bytes: u32 = 0;
    if (header.block_count > 0 and block_blooms_bytes >= 8)
        per_block_bloom_bytes = @intCast((block_blooms_bytes - 8) / header.block_count);

    return .{
        .arena = arena.*,
        .path = path,
        .file_size = bytes.len,
        .sealed = true,
        .header = header,
        .checksum_valid = segment.checksumValid(bytes, header),
        .blocks = blocks,
        .collections = collection_rows,
        .block_collections = block_collections,
        .total_events = header.event_count,
        .unique_did_count = header.unique_did_count,
        .min_seq = header.min_seq,
        .max_seq = header.max_seq,
        .min_witnessed_at = header.min_witnessed_at,
        .max_witnessed_at = header.max_witnessed_at,
        .block_index_bytes = block_index_bytes,
        .segment_bloom_bytes = segment_bloom_bytes,
        .block_blooms_bytes = block_blooms_bytes,
        .collection_index_bytes = collection_index_bytes,
        .per_block_bloom_bytes = per_block_bloom_bytes,
    };
}

fn inspectActive(
    arena: *std.heap.ArenaAllocator,
    path: []const u8,
    bytes: []const u8,
) !Inspection {
    const alloc = arena.allocator();
    var blocks: std.ArrayList(segment.BlockIndexEntry) = .empty;
    var collection_rows: std.ArrayList(Collection) = .empty;
    var block_collections: std.ArrayList([]const u32) = .empty;
    var collection_ids: std.StringHashMapUnmanaged(u32) = .empty;
    var unique_dids: std.StringHashMapUnmanaged(void) = .empty;
    var total_events: u64 = 0;
    var min_seq: u64 = 0;
    var max_seq: u64 = 0;
    var min_witnessed_at: i64 = 0;
    var max_witnessed_at: i64 = 0;
    var saw_any = false;
    var partial_error: ?anyerror = null;

    var it: segment_writer.ActiveIterator = .{ .bytes = bytes };
    while (true) {
        const before = it.offset;
        var block = (it.next(alloc) catch |err| {
            partial_error = err;
            break;
        }) orelse break;
        defer block.deinit(alloc);
        if (block.events.len == 0) break;

        var info: segment.BlockIndexEntry = .{
            .offset = before,
            .compressed_size = @intCast(it.offset - before - 8),
            .uncompressed_size = @intCast(block.buffer.len),
            .event_count = @intCast(block.events.len),
            .min_seq = block.events[0].seq,
            .max_seq = block.events[0].seq,
            .min_witnessed_at = block.events[0].witnessed_at,
            .max_witnessed_at = block.events[0].witnessed_at,
        };
        var ids_in_block: std.AutoHashMapUnmanaged(u32, void) = .empty;
        for (block.events) |event| {
            info.min_seq = @min(info.min_seq, event.seq);
            info.max_seq = @max(info.max_seq, event.seq);
            info.min_witnessed_at = @min(info.min_witnessed_at, event.witnessed_at);
            info.max_witnessed_at = @max(info.max_witnessed_at, event.witnessed_at);
            if (!saw_any) {
                min_seq = event.seq;
                max_seq = event.seq;
                min_witnessed_at = event.witnessed_at;
                max_witnessed_at = event.witnessed_at;
                saw_any = true;
            } else {
                min_seq = @min(min_seq, event.seq);
                max_seq = @max(max_seq, event.seq);
                min_witnessed_at = @min(min_witnessed_at, event.witnessed_at);
                max_witnessed_at = @max(max_witnessed_at, event.witnessed_at);
            }
            if (event.did.len > 0 and !unique_dids.contains(event.did))
                try unique_dids.put(alloc, try alloc.dupe(u8, event.did), {});
            if (event.collection.len > 0)
                try internCollection(alloc, &collection_rows, &collection_ids, &ids_in_block, event.collection, true);
            if (sentinelFor(event.kind)) |sentinel|
                try internCollection(alloc, &collection_rows, &collection_ids, &ids_in_block, sentinel, false);
        }

        const ids = try alloc.alloc(u32, ids_in_block.count());
        var id_it = ids_in_block.keyIterator();
        var id_at: usize = 0;
        while (id_it.next()) |id| : (id_at += 1) ids[id_at] = id.*;
        std.mem.sort(u32, ids, {}, std.sort.asc(u32));
        try block_collections.append(alloc, ids);
        try blocks.append(alloc, info);
        total_events += block.events.len;
    }

    return .{
        .arena = arena.*,
        .path = path,
        .file_size = bytes.len,
        .sealed = false,
        .header = null,
        .blocks = try blocks.toOwnedSlice(alloc),
        .collections = try collection_rows.toOwnedSlice(alloc),
        .block_collections = try block_collections.toOwnedSlice(alloc),
        .total_events = total_events,
        .unique_did_count = @intCast(unique_dids.count()),
        .min_seq = min_seq,
        .max_seq = max_seq,
        .min_witnessed_at = min_witnessed_at,
        .max_witnessed_at = max_witnessed_at,
        .partial_error = partial_error,
    };
}

fn sentinelFor(kind: segment.Kind) ?[]const u8 {
    return switch (kind) {
        .account => "$account",
        .identity => "$identity",
        .sync => "$sync",
        else => null,
    };
}

fn internCollection(
    allocator: Allocator,
    rows: *std.ArrayList(Collection),
    ids: *std.StringHashMapUnmanaged(u32),
    ids_in_block: *std.AutoHashMapUnmanaged(u32, void),
    nsid: []const u8,
    count_event: bool,
) !void {
    const id = ids.get(nsid) orelse id: {
        const owned = try allocator.dupe(u8, nsid);
        const next: u32 = @intCast(rows.items.len);
        try rows.append(allocator, .{ .nsid = owned, .event_count = 0 });
        try ids.put(allocator, owned, next);
        break :id next;
    };
    if (count_event) rows.items[id].event_count += 1;
    try ids_in_block.put(allocator, id, {});
}

pub fn renderInspection(writer: *Io.Writer, ins: *Inspection, mode: BlocksMode, truncate: usize) !void {
    try writer.print("file: {s}\n", .{ins.path});
    try writer.print("size: {d} bytes\n", .{ins.file_size});
    try writer.writeAll(if (ins.sealed) "state: sealed\n" else "state: active (unsealed; block walk)\n");
    try writer.writeAll("magic: jss0\n");
    if (ins.header) |header| {
        try writer.print("version: {d}\n", .{segment.version});
        try writer.print("checksum: 0x{x:0>16} ({s})\n", .{ header.checksum, if (ins.checksum_valid) "valid" else "invalid" });
    } else {
        try writer.writeAll("version: -\nchecksum: 0x0 (active)\n");
    }

    if (ins.partial_error) |err|
        try writer.print("\nWARNING: partial inspection — {s}\n", .{@errorName(err)});

    try writer.writeAll("\nheader summary:\n");
    if (ins.sealed) {
        try writer.print("  block_count:       {d}\n", .{ins.blocks.len});
        try writer.print("  event_count:       {d}\n", .{ins.total_events});
        try writer.print("  unique_did_count:  {d}\n", .{ins.unique_did_count});
    } else {
        try writer.print("  block_count:       {d} (discovered via block walk)\n", .{ins.blocks.len});
        try writer.print("  event_count:       {d} (from walk)\n", .{ins.total_events});
        try writer.print("  unique_did_count:  {d} (from walk; not durable until seal)\n", .{ins.unique_did_count});
    }
    try writer.print("  seq range:         [{d}, {d}]\n", .{ ins.min_seq, ins.max_seq });
    var min_buf: [40]u8 = undefined;
    var max_buf: [40]u8 = undefined;
    try writer.print("  witnessed_at range:  [{s}, {s}]\n", .{
        formatMicros(&min_buf, ins.min_witnessed_at),
        formatMicros(&max_buf, ins.max_witnessed_at),
    });

    if (ins.header) |header| {
        try writer.writeAll("\nfooter layout (all offsets absolute; block_index_offset is also the footer start):\n");
        try writer.print("  block_index_offset:      0x{x:0>16}  block_index_size:       {d} bytes\n", .{ header.block_index_offset, ins.block_index_bytes });
        try writer.print("  did_bloom_offset:        0x{x:0>16}  segment_bloom_size:     {d} bytes\n", .{ header.did_bloom_offset, ins.segment_bloom_bytes });
        try writer.print("  block_did_bloom_offset:  0x{x:0>16}  per_block_blooms:       {d} x {d} bytes (incl. 8B region header)\n", .{ header.block_did_bloom_offset, ins.blocks.len, ins.per_block_bloom_bytes });
        try writer.print("  collection_index_offset: 0x{x:0>16}  collection_index_size:  {d} bytes\n", .{ header.collection_index_offset, ins.collection_index_bytes });
        try writer.print("  end_of_file:             0x{x:0>16}\n", .{ins.file_size});
    } else {
        try writer.writeAll("\nfooter layout: not present (active file)\n");
    }

    try renderCollections(writer, ins);
    if (mode == .summary) return;
    try renderBlocks(writer, ins, mode, truncate);
}

fn renderCollections(writer: *Io.Writer, ins: *Inspection) !void {
    try writer.print("\ncollections ({d} distinct NSIDs):\n", .{ins.collections.len});
    if (ins.collections.len == 0) return writer.writeAll("  (none)\n");

    const allocator = ins.arena.allocator();
    const block_counts = try allocator.alloc(usize, ins.collections.len);
    @memset(block_counts, 0);
    for (ins.block_collections) |ids| for (ids) |id| {
        if (id < block_counts.len) block_counts[id] += 1;
    };
    const order = try allocator.alloc(usize, ins.collections.len);
    for (order, 0..) |*id, i| id.* = i;
    std.mem.sort(usize, order, ins.collections, struct {
        fn lessThan(collections: []const Collection, a: usize, b: usize) bool {
            const ac = collections[a].event_count;
            const bc = collections[b].event_count;
            return ac > bc or (ac == bc and a < b);
        }
    }.lessThan);

    var nsid_width: usize = 0;
    var events_width: usize = "events".len;
    var blocks_width: usize = "blocks".len;
    for (ins.collections, block_counts) |collection, count| {
        nsid_width = @max(nsid_width, collection.nsid.len);
        events_width = @max(events_width, decimalWidth(collection.event_count));
        blocks_width = @max(blocks_width, decimalWidth(count));
    }
    for (order) |id| {
        const collection = ins.collections[id];
        try writer.print("  [{d:>3}] {s:<[4]}  events: {d:>[5]}  blocks: {d:>[6]}\n", .{
            id,
            collection.nsid,
            collection.event_count,
            block_counts[id],
            nsid_width,
            events_width,
            blocks_width,
        });
    }
}

fn renderBlocks(writer: *Io.Writer, ins: *const Inspection, mode: BlocksMode, truncate: usize) !void {
    try writer.print("\nblocks ({d} total):\n", .{ins.blocks.len});
    try writer.writeAll("  idx       offset  comp_size  uncomp_size  events     min_seq     max_seq                  min_witnessed_at                  max_witnessed_at  cols\n");

    const n = ins.blocks.len;
    if (truncate == 0 or mode == .full or n <= truncate) {
        for (0..n) |i| try renderBlock(writer, ins, mode, i);
        return;
    }
    const half = truncate / 2;
    for (0..half) |i| try renderBlock(writer, ins, mode, i);
    try writer.print("  ... ({d} rows omitted) ...\n", .{n - 2 * half});
    for (n - half..n) |i| try renderBlock(writer, ins, mode, i);
}

fn renderBlock(writer: *Io.Writer, ins: *const Inspection, mode: BlocksMode, i: usize) !void {
    const block = ins.blocks[i];
    const ids = if (i < ins.block_collections.len) ins.block_collections[i] else &.{};
    var min_buf: [40]u8 = undefined;
    var max_buf: [40]u8 = undefined;
    try writer.print("  {d:>3}  0x{x:0>10}  {d:>9}  {d:>11}  {d:>6}  {d:>10}  {d:>10}  {s:>30}  {s:>30}  {d:>4}\n", .{
        i,
        block.offset,
        block.compressed_size,
        block.uncompressed_size,
        block.event_count,
        block.min_seq,
        block.max_seq,
        formatMicros(&min_buf, block.min_witnessed_at),
        formatMicros(&max_buf, block.max_witnessed_at),
        ids.len,
    });
    if (mode == .full and ids.len > 0) {
        try writer.writeAll("       collections: ");
        for (ids, 0..) |id, at| {
            if (id >= ins.collections.len) continue;
            if (at > 0) try writer.writeAll(", ");
            try writer.writeAll(ins.collections[id].nsid);
        }
        try writer.writeByte('\n');
    }
}

fn decimalWidth(value: anytype) usize {
    var n = value;
    var width: usize = 1;
    while (n >= 10) : (n /= 10) width += 1;
    return width;
}

fn formatMicros(buffer: []u8, micros: i64) []const u8 {
    if (micros == 0) return "0";
    const secs: u64 = @intCast(@divFloor(micros, std.time.us_per_s));
    const fraction: u64 = @intCast(@mod(micros, std.time.us_per_s));
    const epoch_seconds: std.time.epoch.EpochSeconds = .{ .secs = secs };
    const year_day = epoch_seconds.getEpochDay().calculateYearDay();
    const month_day = year_day.calculateMonthDay();
    const day_seconds = epoch_seconds.getDaySeconds();
    return std.fmt.bufPrint(buffer, "{d:0>4}-{d:0>2}-{d:0>2}T{d:0>2}:{d:0>2}:{d:0>2}.{d:0>6}Z", .{
        year_day.year,
        month_day.month.numeric(),
        month_day.day_index + 1,
        day_seconds.getHoursIntoDay(),
        day_seconds.getMinutesIntoHour(),
        day_seconds.getSecondsIntoMinute(),
        fraction,
    }) catch unreachable;
}

test "version command follows upstream text contract" {
    var out: Io.Writer.Allocating = .init(std.testing.allocator);
    defer out.deinit();
    try renderVersion(&out.writer);
    try std.testing.expect(std.mem.startsWith(u8, out.written(), "jetstream version "));
    try std.testing.expect(std.mem.indexOf(u8, out.written(), " (commit ") != null);
    try std.testing.expect(std.mem.endsWith(u8, out.written(), ", built unknown)\n"));
}

test "sealed fixture inspection matches upstream output byte for byte after path" {
    var threaded: Io.Threaded = .init(std.testing.allocator, .{});
    defer threaded.deinit();
    const io = threaded.io();
    const fixture = "tests/fixtures/seg_0000000000.jss";
    var inspection = try inspectSegment(std.testing.allocator, io, fixture);
    defer inspection.deinit();
    var out: Io.Writer.Allocating = .init(std.testing.allocator);
    defer out.deinit();
    try renderInspection(&out.writer, &inspection, .table, 100);

    const expected = try Io.Dir.cwd().readFileAlloc(io, "tests/fixtures/seg_0000000000.inspect.txt", std.testing.allocator, .limited(1 << 20));
    defer std.testing.allocator.free(expected);
    const expected_body = expected[std.mem.indexOfScalar(u8, expected, '\n').? + 1 ..];
    const actual_body = out.written()[std.mem.indexOfScalar(u8, out.written(), '\n').? + 1 ..];
    try std.testing.expectEqualStrings(expected_body, actual_body);
}

test "active inspection walks real frames and labels torn tails as partial" {
    var active = try segment_writer.ActiveWriter.init(std.testing.allocator);
    defer active.deinit();
    active.max_events_per_block = 1;
    try active.append(.{
        .seq = 7,
        .witnessed_at = 1_700_000_000_000_007,
        .indexed_at = 0,
        .kind = .identity,
        .did = "did:plc:test",
        .collection = "",
        .rkey = "",
        .rev = "",
        .payload = "{}",
    });
    try active.buf.appendSlice(std.testing.allocator, "torn");

    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "active.jss", .data = active.bytes() });
    const path = try tmp.dir.realPathFileAlloc(std.testing.io, "active.jss", std.testing.allocator);
    defer std.testing.allocator.free(path);
    var inspection = try inspectSegment(std.testing.allocator, std.testing.io, path);
    defer inspection.deinit();
    try std.testing.expect(!inspection.sealed);
    try std.testing.expectEqual(@as(usize, 1), inspection.blocks.len);
    try std.testing.expectEqual(@as(u64, 1), inspection.total_events);
    try std.testing.expectEqual(@as(u32, 1), inspection.unique_did_count);
    try std.testing.expectEqualStrings("$identity", inspection.collections[0].nsid);
    try std.testing.expect(inspection.partial_error != null);
}

test "inspection reports checksum mismatch without weakening normal parsing" {
    const fixture = "tests/fixtures/seg_0000000000.jss";
    const original = try Io.Dir.cwd().readFileAlloc(std.testing.io, fixture, std.testing.allocator, .limited(1 << 20));
    defer std.testing.allocator.free(original);
    const header = try segment.Header.decode(original);
    const corrupt_at = header.block_did_bloom_offset + 8 +
        (header.collection_index_offset - header.block_did_bloom_offset - 8) / 2;
    original[corrupt_at] ^= 1;
    var tmp = std.testing.tmpDir(.{});
    defer tmp.cleanup();
    try tmp.dir.writeFile(std.testing.io, .{ .sub_path = "corrupt.jss", .data = original });
    const path = try tmp.dir.realPathFileAlloc(std.testing.io, "corrupt.jss", std.testing.allocator);
    defer std.testing.allocator.free(path);
    var inspection = try inspectSegment(std.testing.allocator, std.testing.io, path);
    defer inspection.deinit();
    try std.testing.expect(!inspection.checksum_valid);
}