const Message = @import("message_parser.zig").Message;
const std = @import("std");

pub const ClientState = struct {
    id: usize,
    connect: Message.AllocatedConnect,

    /// Messages that this client should receive.
    recv_queue: *std.Io.Queue(Message),
    recv_queue_buffer: [1024]Message = undefined,

    from_client: *std.Io.Reader,
    to_client: *std.Io.Writer,

    task: std.Io.Future(void),

    pub fn init(
        io: std.Io,
        allocator: std.mem.Allocator,
        id: usize,
        connect: Message.AllocatedConnect,
        in: *std.Io.Reader,
        out: *std.Io.Writer,
    ) !ClientState {
        var res: ClientState = .{
            .id = id,
            .connect = connect,
            .recv_queue = try allocator.create(std.Io.Queue(Message)),
            .from_client = in,
            .to_client = out,
            .task = undefined,
        };
        res.recv_queue.* = .init(&res.recv_queue_buffer);
        res.task = try io.concurrent(processWrite, .{ &res, io });

        return res;
    }

    fn processWrite(
        self: *ClientState,
        io: std.Io,
    ) void {
        while (true) {
            const message = self.recv_queue.getOne(io) catch break;
            std.debug.print("got message in write loop to send to client: {any}\n", .{message});
            switch (message) {
                .@"+ok" => {
                    writeOk(self.to_client) catch break;
                },
                .pong => {
                    writePong(self.to_client) catch break;
                },
                .info => |info| {
                    writeInfo(self.to_client, info) catch break;
                },
                .msg => |m| {
                    if (writeMsg(self.to_client, m)) |_| {
                        @branchHint(.likely);
                    } else |_| {
                        @branchHint(.unlikely);
                        break;
                    }
                },
                else => {
                    std.debug.panic("unimplemented write", .{});
                },
            }
        }
        self.task.cancel(io);
    }

    pub fn deinit(self: *ClientState, io: std.Io, allocator: std.mem.Allocator) void {
        self.task.cancel(io);
        self.connect.deinit();
        _ = allocator;
        // allocator.destroy(self.recv_queue);
    }

    /// Return true if the value was put in the clients buffer to process, else false.
    pub fn send(self: *ClientState, io: std.Io, msg: Message) (std.Io.Cancelable || std.Io.QueueClosedError)!bool {
        try self.recv_queue.putOne(io, msg);
        return true;
    }

    pub fn next(self: *ClientState, allocator: std.mem.Allocator) !Message {
        // std.debug.print("in client awaiting next message\n", .{});
        // errdefer std.debug.print("actually it was canceled\n", .{});
        // defer std.debug.print("client returning next message!\n", .{});
        return Message.next(allocator, self.from_client);
        // return self.send_queue.getOne(io);
    }
};

fn writeOk(out: *std.Io.Writer) !void {
    _ = try out.write("+OK\r\n");
    try out.flush();
}

fn writePong(out: *std.Io.Writer) !void {
    _ = try out.write("PONG\r\n");
    try out.flush();
}

pub fn writeInfo(out: *std.Io.Writer, info: Message.ServerInfo) !void {
    _ = try out.write("INFO ");
    try std.json.Stringify.value(info, .{}, out);
    _ = try out.write("\r\n");
    try out.flush();
}

fn writeMsg(out: *std.Io.Writer, msg: Message.Msg) !void {
    std.debug.print("PRINTING MESSAGE\n\n\n\n", .{});
    std.debug.print(
        "MSG {s} {s} {s} {d}\r\n{s}\r\n-\n\n\n",
        .{
            msg.subject,
            msg.sid,
            msg.reply_to orelse "",
            msg.payload.len,
            msg.payload,
        },
    );
    try out.print(
        "MSG {s} {s} {s} {d}\r\n{s}\r\n",
        .{
            msg.subject,
            msg.sid,
            msg.reply_to orelse "",
            msg.payload.len,
            msg.payload,
        },
    );
    try out.flush();
}

test {
    const io = std.testing.io;
    const gpa = std.testing.allocator;

    var from_client: std.Io.Reader = .fixed(
        "CONNECT {\"verbose\":false,\"pedantic\":false,\"tls_required\":false,\"name\":\"NATS CLI Version v0.2.4\",\"lang\":\"go\",\"version\":\"1.43.0\",\"protocol\":1,\"echo\":true,\"headers\":true,\"no_responders\":true}\r\n" ++
            "PING\r\n",
    );
    var from_client_buf: [1024]Message = undefined;
    var from_client_queue: std.Io.Queue(Message) = .init(&from_client_buf);

    {
        // Simulate stream
        while (Message.next(gpa, &from_client)) |msg| {
            switch (msg) {
                .eos => {
                    try from_client_queue.putOne(io, msg);
                    break;
                },
                else => {
                    try from_client_queue.putOne(io, msg);
                },
            }
        } else |err| switch (err) {
            error.EndOfStream => try from_client_queue.close(io),
            else => return err,
        }

        while (from_client_queue.getOne(io)) |msg| {
            switch (msg) {
                .eos => {
                    break;
                },
                .connect => |*c| {
                    std.debug.print("Message: {any}\n", .{msg});
                    c.deinit();
                },
                else => {
                    std.debug.print("Message: {any}\n", .{msg});
                },
            }
        } else |_| {}
    }

    from_client_queue = .init(&from_client_buf);
    // Reset the reader to process it again.
    from_client.seek = 0;

    {
        const SemiClient = struct {
            q: std.Io.Queue(Message),

            fn parseClientInput(self: *@This(), ioh: std.Io, in: *std.Io.Reader) void {
                defer std.debug.print("done parse\n", .{});
                while (Message.next(gpa, in)) |msg| {
                    switch (msg) {
                        .eos => {
                            self.q.putOne(ioh, msg) catch return;
                            break;
                        },
                        else => {
                            self.q.putOne(ioh, msg) catch return;
                        },
                    }
                } else |_| {}
            }

            fn next(self: *@This(), ioh: std.Io) !Message {
                return self.q.getOne(ioh);
            }

            fn printAll(self: *@This(), ioh: std.Io) void {
                defer std.debug.print("done print\n", .{});
                while (self.next(ioh)) |*msg| {
                    switch (msg.*) {
                        .eos => |_| {
                            break;
                        },
                        else => {},
                    }
                    std.debug.print("Client msg: {any}\n", .{msg});
                    switch (msg.*) {
                        .connect => |c| {
                            // c.allocator.deinit();
                            c.deinit();
                            // @constCast(c).deinit();
                        },
                        else => {},
                    }
                } else |_| {}
            }
        };

        var c: SemiClient = .{ .q = from_client_queue };
        var group: std.Io.Group = .init;
        defer group.wait(io);

        group.concurrent(io, SemiClient.printAll, .{ &c, io }) catch {
            @panic("could not start printAll\n");
        };

        group.concurrent(io, SemiClient.parseClientInput, .{ &c, io, &from_client }) catch {
            @panic("could not start printAll\n");
        };
    }

    ////////

    // const connect = (Message.next(gpa, &from_client) catch unreachable).connect;

    // var to_client_alloc: std.Io.Writer.Allocating = .init(gpa);
    // defer to_client_alloc.deinit();
    // var to_client = to_client_alloc.writer;

    // var client: ClientState = try .init(io, gpa, 0, connect, &from_client, &to_client);
    // defer client.deinit(gpa);

    // {
    //     var get_next = io.concurrent(ClientState.next, .{ &client, io }) catch unreachable;
    //     defer if (get_next.cancel(io)) |_| {} else |_| @panic("fail");

    //     var timeout = io.concurrent(std.Io.sleep, .{ io, .fromMilliseconds(1000), .awake }) catch unreachable;
    //     defer timeout.cancel(io) catch {};

    //     switch (try io.select(.{
    //         .get_next = &get_next,
    //         .timeout = &timeout,
    //     })) {
    //         .get_next => |next| {
    //             std.debug.print("next is {any}\n", .{next});
    //             try std.testing.expect((next catch |err| return err) == .ping);
    //         },
    //         .timeout => {
    //             std.debug.print("reached timeout\n", .{});
    //             return error.TestUnexpectedResult;
    //         },
    //     }
    // }
}