/// Child binary for the endpoint test.
///
/// Runs a full Thespian context with actors registered in a StdioEndpoint:
///
///   "echo" (wire ID 2): receives {from_id, payload}, replies via send_named
///     to "test_receiver" on the parent side.
///
///   "echo_id" (wire ID 3): receives {from_id, payload}.
///     - If from_id == 0: trigger mode — sends send_named to "test_receiver"
///       so the parent can establish a proxy and reply by ID.
///     - If from_id != 0: reply mode — sends wire send to from_id with ["done"].
///
/// StdioEndpoint dispatches inbound wire messages to actors, passing
/// {from_id, payload} so actors can reply by ID when needed.
const std = @import("std");
const tp = @import("thespian");
const cbor = @import("cbor");
const framing = @import("framing");
const protocol = @import("protocol");

var trace_file: ?std.fs.File = null;
var trace_buf: [4096]u8 = undefined;
var trace_file_writer: std.fs.File.Writer = undefined;

fn trace_handler(buf: tp.message.c_buffer_type) callconv(.c) void {
    if (trace_file == null) return;
    cbor.toJsonWriter(buf.base[0..buf.len], &trace_file_writer.interface, .{}) catch return;
    trace_file_writer.interface.writeByte('\n') catch return;
}

// ---------------------------------------------------------------------------
// EchoActor (wire ID 2)
// ---------------------------------------------------------------------------

const EchoActor = struct {
    allocator: std.mem.Allocator,
    receiver: tp.Receiver(*@This()),

    const Args = struct {
        allocator: std.mem.Allocator,
        /// Owned pid of the StdioEndpoint to notify when ready.
        parent: tp.pid,
    };

    fn start(args: Args) tp.result {
        return init(args) catch |e| return tp.exit_error(e, @errorReturnTrace());
    }

    fn init(args: Args) !void {
        defer args.parent.deinit();
        const self = try args.allocator.create(@This());
        self.* = .{
            .allocator = args.allocator,
            .receiver = .init(receive_fn, deinit, self),
        };
        errdefer self.deinit();
        try args.parent.send(.{"echo_ready"});
        tp.receive(&self.receiver);
    }

    fn deinit(self: *@This()) void {
        self.allocator.destroy(self);
    }

    fn receive_fn(self: *@This(), from: tp.pid_ref, m: tp.message) tp.result {
        return self.receive(from, m) catch |e| return tp.exit_error(e, @errorReturnTrace());
    }

    fn receive(self: *@This(), from: tp.pid_ref, m: tp.message) !void {
        var from_id: u64 = 0;
        var payload: []const u8 = "";
        if (try m.match(.{ tp.extract(&from_id), cbor.extract_cbor(&payload) })) {
            // Reply via send_named to "test_receiver" on the parent side.
            try from.send(.{ "send", @as(u64, 2), "test_receiver", protocol.RawCbor{ .bytes = payload } });
        } else {
            return tp.unexpected(m);
        }
        _ = self;
    }
};

// ---------------------------------------------------------------------------
// EchoIdActor (wire ID 3)
// ---------------------------------------------------------------------------

const EchoIdActor = struct {
    allocator: std.mem.Allocator,
    receiver: tp.Receiver(*@This()),

    const Args = struct {
        allocator: std.mem.Allocator,
        /// Owned pid of the StdioEndpoint to notify when ready.
        parent: tp.pid,
    };

    fn start(args: Args) tp.result {
        return init(args) catch |e| return tp.exit_error(e, @errorReturnTrace());
    }

    fn init(args: Args) !void {
        defer args.parent.deinit();
        const self = try args.allocator.create(@This());
        self.* = .{
            .allocator = args.allocator,
            .receiver = .init(receive_fn, deinit, self),
        };
        errdefer self.deinit();
        try args.parent.send(.{"echo_id_ready"});
        tp.receive(&self.receiver);
    }

    fn deinit(self: *@This()) void {
        self.allocator.destroy(self);
    }

    fn receive_fn(self: *@This(), from: tp.pid_ref, m: tp.message) tp.result {
        return self.receive(from, m) catch |e| return tp.exit_error(e, @errorReturnTrace());
    }

    fn receive(self: *@This(), from: tp.pid_ref, m: tp.message) !void {
        var from_id: u64 = 0;
        var payload: []const u8 = "";
        if (try m.match(.{ tp.extract(&from_id), cbor.extract_cbor(&payload) })) {
            if (from_id == 0) {
                // Trigger mode: send send_named so the parent can establish
                // a proxy and will reply by ID.
                try from.send(.{ "send", @as(u64, 3), "test_receiver", protocol.RawCbor{ .bytes = payload } });
            } else {
                // Reply mode: send ["done"] back to the parent actor by wire ID.
                var done_buf: [8]u8 = undefined;
                const done_msg = try tp.message.fmtbuf(&done_buf, .{"done"});
                try from.send(.{ "send", @as(u64, 3), from_id, protocol.RawCbor{ .bytes = done_msg.buf } });
            }
        } else {
            return tp.unexpected(m);
        }
        _ = self;
    }
};

// ---------------------------------------------------------------------------
// StdioEndpoint
// ---------------------------------------------------------------------------

const StdioEndpoint = struct {
    allocator: std.mem.Allocator,
    fd_stdin: tp.file_descriptor,
    accumulator: framing.Accumulator,
    read_buf: [4096]u8,
    /// Maps wire ID → local actor pid for routing inbound send-by-ID.
    wire_ids: std.AutoHashMap(u64, tp.pid),
    /// Number of child actors that have reported ready; arm stdin when all ready.
    ready_count: u8,
    receiver: tp.Receiver(*@This()),

    const Args = struct { allocator: std.mem.Allocator };

    fn start(args: Args) tp.result {
        return init(args) catch |e| return tp.exit_error(e, @errorReturnTrace());
    }

    fn init(args: Args) !void {
        const fd_stdin = try tp.file_descriptor.init("stdin", 0);
        const self = try args.allocator.create(@This());
        self.* = .{
            .allocator = args.allocator,
            .fd_stdin = fd_stdin,
            .accumulator = .{},
            .read_buf = undefined,
            .wire_ids = std.AutoHashMap(u64, tp.pid).init(args.allocator),
            .ready_count = 0,
            .receiver = .init(receive_fn, deinit, self),
        };
        errdefer self.deinit();

        _ = try tp.spawn_link(args.allocator, EchoActor.Args{
            .allocator = args.allocator,
            .parent = tp.self_pid().clone(),
        }, EchoActor.start, "echo");

        _ = try tp.spawn_link(args.allocator, EchoIdActor.Args{
            .allocator = args.allocator,
            .parent = tp.self_pid().clone(),
        }, EchoIdActor.start, "echo_id");

        tp.receive(&self.receiver);
    }

    fn deinit(self: *@This()) void {
        var it = self.wire_ids.valueIterator();
        while (it.next()) |p| p.deinit();
        self.wire_ids.deinit();
        self.fd_stdin.deinit();
        self.allocator.destroy(self);
    }

    fn receive_fn(self: *@This(), from: tp.pid_ref, m: tp.message) tp.result {
        return self.receive(from, m) catch |e| return tp.exit_error(e, @errorReturnTrace());
    }

    fn receive(self: *@This(), from: tp.pid_ref, m: tp.message) !void {
        var from_id: u64 = 0;
        var to_id_wire: u64 = 0;
        var to_name: []const u8 = "";
        var payload: []const u8 = "";

        if (try m.match(.{"echo_ready"})) {
            tp.env.get().proc_set("echo", from);
            try self.wire_ids.put(2, from.clone());
            self.ready_count += 1;
            if (self.ready_count == 2) try self.fd_stdin.wait_read();
        } else if (try m.match(.{"echo_id_ready"})) {
            tp.env.get().proc_set("echo_id", from);
            try self.wire_ids.put(3, from.clone());
            self.ready_count += 1;
            if (self.ready_count == 2) try self.fd_stdin.wait_read();
        } else if (try m.match(.{ "fd", "stdin", "read_ready" })) {
            try self.dispatch_stdin();
            try self.fd_stdin.wait_read();
        } else if (try m.match(.{ "fd", "stdin", "read_error", tp.any, tp.any })) {
            return tp.exit("stdin_closed");
        } else if (try m.match(.{ "send", tp.extract(&from_id), tp.extract(&to_id_wire), cbor.extract_cbor(&payload) })) {
            try self.send_wire_by_id(from_id, to_id_wire, payload);
        } else if (try m.match(.{ "send", tp.extract(&from_id), tp.extract(&to_name), cbor.extract_cbor(&payload) })) {
            try self.send_wire(from_id, to_name, payload);
        } else {
            return tp.unexpected(m);
        }
    }

    fn dispatch_stdin(self: *@This()) !void {
        const n = std.fs.File.stdin().read(&self.read_buf) catch |e| switch (e) {
            error.WouldBlock => return,
            else => return tp.exit_error(e, @errorReturnTrace()),
        };
        if (n == 0) return tp.exit("stdin_closed");
        if (self.accumulator.feed(self.read_buf[0..n])) |frame| {
            try self.dispatch_frame(frame);
        }
    }

    fn dispatch_frame(self: *@This(), frame: []const u8) !void {
        const msg = try protocol.decode(frame);
        switch (msg) {
            .send_named => |s| {
                const actor = tp.env.get().proc(s.to_name);
                // Include from_id so the actor can reply by ID if needed.
                try actor.send(.{ s.from_id, protocol.RawCbor{ .bytes = s.payload } });
            },
            .send => |s| {
                // Route by wire ID.
                if (self.wire_ids.get(s.to_id)) |actor|
                    try actor.send(.{ s.from_id, protocol.RawCbor{ .bytes = s.payload } })
                else
                    return tp.exit_error(error.UnknownWireId, null);
            },
            .link, .exit, .proxy_id, .transport_error => return tp.exit_error(error.UnexpectedMessage, null),
        }
    }

    fn send_wire_by_id(_: *@This(), from_id: u64, to_id: u64, payload: []const u8) !void {
        var msg_buf: [framing.max_frame_size]u8 = undefined;
        var msg_stream: std.Io.Writer = .fixed(&msg_buf);
        try protocol.encode_send(&msg_stream, from_id, to_id, payload);

        var frame_buf: [framing.max_frame_size + 4]u8 = undefined;
        var frame_stream: std.Io.Writer = .fixed(&frame_buf);
        try framing.write_frame(&frame_stream, msg_stream.buffered());
        try std.fs.File.stdout().writeAll(frame_stream.buffered());
    }

    fn send_wire(_: *@This(), from_id: u64, to_name: []const u8, payload: []const u8) !void {
        var msg_buf: [framing.max_frame_size]u8 = undefined;
        var msg_stream: std.Io.Writer = .fixed(&msg_buf);
        try protocol.encode_send_named(&msg_stream, from_id, to_name, payload);

        var frame_buf: [framing.max_frame_size + 4]u8 = undefined;
        var frame_stream: std.Io.Writer = .fixed(&frame_buf);
        try framing.write_frame(&frame_stream, msg_stream.buffered());
        try std.fs.File.stdout().writeAll(frame_stream.buffered());
    }
};

// ---------------------------------------------------------------------------
// main
// ---------------------------------------------------------------------------

pub fn main() !void {
    var gpa: std.heap.GeneralPurposeAllocator(.{}) = .{};
    defer _ = gpa.deinit();
    const allocator = gpa.allocator();

    var initial_env: ?tp.env = null;
    if (std.posix.getenv("TRACE") != null) {
        const f = try std.fs.cwd().createFile("remote_child_endpoint_trace.json", .{});
        trace_file = f;
        trace_file_writer = f.writer(&trace_buf);
        var e = tp.env.init();
        e.on_trace(&trace_handler);
        e.enable_all_channels();
        initial_env = e;
    }

    var ctx = try tp.context.init(allocator);
    defer ctx.deinit();

    var exit_ok = true;
    var exit_handler = tp.make_exit_handler(&exit_ok, struct {
        fn handle(ok: *bool, status: []const u8) void {
            if (!std.mem.eql(u8, status, "stdin_closed") and
                !std.mem.eql(u8, status, "normal"))
                ok.* = false;
        }
    }.handle);

    _ = try ctx.spawn_link(
        StdioEndpoint.Args{ .allocator = allocator },
        StdioEndpoint.start,
        "stdio_endpoint",
        &exit_handler,
        if (initial_env) |*e| e else null,
    );

    ctx.run();

    if (initial_env) |e| {
        trace_file_writer.interface.flush() catch {};
        trace_file.?.close();
        trace_file = null;
        e.deinit();
    }

    std.process.exit(if (exit_ok) 0 else 1);
}