/// Child binary for the endpoint test.
///
/// Runs a full Thespian context with two actors:
///   - EchoActor: registered as "echo" in env; forwards any received message
///     back through the endpoint to "test_receiver" on the parent side.
///   - StdioEndpoint: reads framed CBOR from stdin, dispatches send_named
///     messages to local actors via env, and writes outbound frames to stdout.
const std = @import("std");
const tp = @import("thespian");
const cbor = @import("cbor");
const framing = @import("framing");
const protocol = @import("protocol");

// ---------------------------------------------------------------------------
// EchoActor
// ---------------------------------------------------------------------------

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();
        // Notify StdioEndpoint so it can register us (via `from`) in its env.
        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 {
        // Reply to sender (the StdioEndpoint) with {"send", 2, "test_receiver", <payload>}
        try from.send(.{ "send", @as(u64, 2), "test_receiver", protocol.RawCbor{ .bytes = m.buf } });
        _ = self;
    }
};

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

const StdioEndpoint = struct {
    allocator: std.mem.Allocator,
    fd_stdin: tp.file_descriptor,
    accumulator: framing.Accumulator,
    read_buf: [4096]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,
            .receiver = .init(receive_fn, deinit, self),
        };
        errdefer self.deinit();

        // Spawn echo actor linked to this endpoint. Pass our pid so EchoActor
        // can notify us when "echo" is registered in the env. We arm wait_read
        // only after that notification to avoid a race where read_ready fires
        // before EchoActor has finished registering.
        _ = try tp.spawn_link(args.allocator, EchoActor.Args{
            .allocator = args.allocator,
            .parent = tp.self_pid().clone(),
        }, EchoActor.start, "echo");

        tp.receive(&self.receiver);
    }

    fn deinit(self: *@This()) void {
        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_name: []const u8 = "";
        var payload: []const u8 = "";

        if (try m.match(.{"echo_ready"})) {
            // EchoActor is ready; register its pid in our env so dispatch_frame
            // can route to it, then arm stdin reads.
            tp.env.get().proc_set("echo", from);
            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_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 => |s| {
                _ = s;
                _ = self;
                return tp.exit_error(error.UnexpectedSend, null);
            },
            .send_named => |s| {
                const actor = tp.env.get().proc(s.to_name);
                try actor.send_raw(tp.message{ .buf = s.payload });
                _ = self;
            },
            .link, .exit, .proxy_id, .transport_error => return tp.exit_error(error.UnexpectedMessage, null),
        }
    }

    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 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,
        null,
    );

    ctx.run();

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