const std = @import("std");
const zio = @import("zio");
const zat = @import("zat");
const index = @import("internal/collection_index/index.zig");
const Io = std.Io;
const Probe = struct {
    ci: *index.CollectionIndex,
    io: Io,
    heartbeat_at: i96 = 0,
    outcome: anyerror!bool = false,
    write_outcome: anyerror!void = {},
    write_at: i96 = 0,
    fn heartbeat(self: *Probe) void {
        Io.sleep(self.io, .fromMilliseconds(10), .awake) catch return;
        self.heartbeat_at = Io.Clock.awake.now(self.io).nanoseconds;
    }
    fn read(self: *Probe) void {
        self.outcome = self.readPage();
    }
    fn readPage(self: *Probe) !bool {
        var bytes: [4096]u8 = undefined;
        var spans: [100]index.Span = undefined;
        const page = try self.ci.listReposByCollection("app.bsky.feed.post", 100, null, &bytes, &spans);
        if (page.count == 0) return false;
        return std.mem.eql(u8, page.getDid(0), "did:plc:alice");
    }
    fn write(self: *Probe) void {
        self.write_outcome = self.writeFrame();
        self.write_at = Io.Clock.awake.now(self.io).nanoseconds;
    }
    fn writeFrame(self: *Probe) !void {
        const path = try std.heap.c_allocator.dupe(u8, "app.bsky.feed.post/record");
        defer std.heap.c_allocator.free(path);
        const ops: zat.cbor.Value = .{ .array = &.{.{ .map = &.{
            .{ .key = "action", .value = .{ .text = "create" } },
            .{ .key = "path", .value = .{ .text = path } },
        } }} };
        try self.ci.trackCommitOps("did:plc:bob", ops);
    }
};
const Gate = struct {
    ci: *index.CollectionIndex,
    threaded: Io,
    held: Io.Event = .unset,
    fn run(self: *Gate) void {
        self.ci.read_mutex.lockUncancelable(self.ci.read_io);
        self.held.set(self.threaded);
        Io.sleep(self.threaded, .fromMilliseconds(200), .awake) catch unreachable;
        self.ci.read_mutex.unlock(self.ci.read_io);
    }
};
pub fn main(init: std.process.Init) !void {
    const args = try init.minimal.args.toSlice(init.arena.allocator());
    if (args.len != 2) return error.ExpectedScratchPath;
    const runtime = try zio.Runtime.init(init.gpa, .{ .executors = .exact(1) });
    defer runtime.deinit();
    const io = runtime.io();
    var output = Io.File.stdout().writer(init.io, &.{});
    try output.interface.writeAll("reader_io,hold_ms,heartbeat_ms,write_ack_ms\n");
    for (0..3) |mode| {
        const path = try std.fmt.allocPrint(init.arena.allocator(), "{s}/{d}", .{ args[1], mode });
        var ci = try index.CollectionIndex.openWithReaderIo(init.gpa, init.io, if (mode == 0) init.io else io, path);
        defer ci.deinit();
        if (mode == 2) try ci.enablePageWorker(2);
        try ci.addCollection("did:plc:alice", "app.bsky.feed.post");
        try ci.enableBatching(.{});
        var gate: Gate = .{ .ci = &ci, .threaded = init.io };
        const releaser = try std.Thread.spawn(.{}, Gate.run, .{&gate});
        gate.held.waitUncancelable(init.io);
        var joined = false;
        defer if (!joined) releaser.join();
        var probe: Probe = .{ .ci = &ci, .io = io };
        const start = Io.Clock.awake.now(io).nanoseconds;
        var group: Io.Group = .init;
        defer group.cancel(io);
        try group.concurrent(io, Probe.heartbeat, .{&probe});
        try group.concurrent(io, Probe.read, .{&probe});
        try group.concurrent(io, Probe.write, .{&probe});
        try group.await(io);
        releaser.join();
        joined = true;
        if (!(try probe.outcome)) return error.MissingAcknowledgedPair;
        try probe.write_outcome;
        if (!(try ci.hasCollectionsChecked("did:plc:bob"))) return error.MissingOwnedFrame;
        const write_ms = @divTrunc(probe.write_at - start, 1_000_000);
        const elapsed_ms = @divTrunc(probe.heartbeat_at - start, 1_000_000);
        try output.interface.print("{s},200,{d},{d}\n", .{ if (mode == 0) "threaded" else if (mode == 1) "zio" else "page-worker", elapsed_ms, write_ms });
        if (mode == 0 and elapsed_ms < 150) return error.ControlDidNotBlock;
        if (mode != 0 and write_ms >= 100) return error.WriterBlockedByReader;
        if (mode != 0 and elapsed_ms >= 100) return error.ExecutorBlocked;
    }
}