const std = @import("std");
const ArenaAllocator = std.heap.ArenaAllocator;
const Allocator = std.mem.Allocator;
const ArrayList = std.ArrayList;
const Reader = std.Io.Reader;
const Writer = std.Io.Writer;
const AllocatingWriter = std.Io.Writer.Allocating;
const StaticStringMap = std.StaticStringMap;

const log = std.log.scoped(.zits);

const isDigit = std.ascii.isDigit;
const isUpper = std.ascii.isUpper;
const isWhitespace = std.ascii.isWhitespace;

const parseUnsigned = std.fmt.parseUnsigned;

const message = @import("./message.zig");
const Message = message.Message;
const Payload = @import("./Payload.zig");

const client_control = StaticStringMap(message.Control).initComptime(
    .{
        // {"INFO", .info},
        .{ @tagName(.CONNECT), .CONNECT },
        .{ @tagName(.PUB), .PUB },
        .{ @tagName(.HPUB), .HPUB },
        .{ @tagName(.SUB), .SUB },
        .{ @tagName(.UNSUB), .UNSUB },
        // {"MSG", .msg},
        // {"HMSG", .hmsg},
        .{ @tagName(.PING), .PING },
        .{ @tagName(.PONG), .PONG },
        // {"+OK", .@"+ok"},
        // {"-ERR", .@"-err"},
    },
);

pub fn control(in: *Reader) !message.Control {
    const longest_ctrl = comptime blk: {
        var min_len = 0;
        for (client_control.keys()) |ctrl| {
            min_len = @max(ctrl.len, min_len);
        }
        break :blk min_len;
    };
    // log.debug("buffered: '{s}'", .{in.buffered()});
    std.debug.assert(in.buffer.len >= longest_ctrl);
    while (true) {
        // log.debug("buffered l: '{s}'", .{in.buffered()});
        var iter = std.mem.tokenizeAny(u8, in.buffered(), " \t\r");
        if (iter.next()) |str| {
            if (client_control.get(str)) |ctrl| {
                in.toss(str.len);
                return ctrl;
            } else if (str.len >= longest_ctrl) {
                return error.InvalidControl;
            }
        }
        try in.fillMore();
    }
}

test control {
    {
        var buf: [7]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "PUB " },
        });
        try std.testing.expectEqual(.PUB, try control(&in.interface));
        try std.testing.expectEqualSlices(u8, " ", in.interface.buffered());
    }
    {
        var buf: [7]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "PUB" },
        });
        try std.testing.expectEqual(.PUB, try control(&in.interface));
        try std.testing.expectEqualSlices(u8, "", in.interface.buffered());
    }
    {
        var buf: [7]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "PU" },
            .{ .buffer = "B" },
        });
        try std.testing.expectEqual(.PUB, try control(&in.interface));
        try std.testing.expectEqualSlices(u8, "", in.interface.buffered());
    }
    {
        var buf: [7]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "PIN" },
            .{ .buffer = "G\r\n" },
        });
        try std.testing.expectEqual(.PING, try control(&in.interface));
        try std.testing.expectEqualSlices(u8, "\r\n", in.interface.buffered());
    }
    {
        var buf: [7]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "CONNECT" },
        });
        try std.testing.expectEqual(.CONNECT, try control(&in.interface));
    }
    {
        var buf: [7]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "CONNECC" },
        });
        try std.testing.expectError(error.InvalidControl, control(&in.interface));
    }
}

/// The return value is owned by the reader passed to this function.
/// Operations that modify the readers buffer invalidates this value.
pub fn @"pub"(in: *Reader) !Message.Pub {
    // TODO: Add pedantic option.
    // See: https://docs.nats.io/reference/reference-protocols/nats-protocol#syntax-1

    while (true) {
        var iter = std.mem.tokenizeAny(u8, in.buffered(), " \t\r");

        if (iter.next()) |subject| {
            if (iter.next()) |second| {
                if (in.buffered().len > iter.index) {
                    if (in.buffered()[iter.index] == '\r') {
                        const bytes_str = second;
                        const bytes = try parseUnsigned(usize, bytes_str, 10);

                        // 4 bytes for CRLF on either side of the payload.
                        _ = try in.take(iter.index + 2);
                        defer {
                            _ = in.take(2) catch {
                                log.warn("very bad parsing issue", .{});
                            };
                        }
                        return .{
                            .subject = subject,
                            .reply_to = null,
                            .payload = try in.take(bytes),
                        };
                    }

                    const reply_to = second;
                    if (iter.next()) |bytes_str| {
                        const bytes = try parseUnsigned(usize, bytes_str, 10);

                        if (in.buffered()[iter.index] == '\r') {
                            if (iter.rest().len > bytes + 2) {
                                // 4 bytes for CRLF on either side of the payload.
                                _ = try in.take(iter.index + bytes + 4);
                                return .{
                                    .subject = subject,
                                    .reply_to = reply_to,
                                    .payload = iter.rest()[1 .. 1 + bytes],
                                };
                            }
                        }
                    }
                }
            }
        }

        try in.fillMore();
    }
}

test @"pub" {
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo bar 2\r\nhi\r\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Pub{
                .subject = "foo",
                .reply_to = "bar",
                .payload = "hi",
            },
            try @"pub"(&in.interface),
        );
        try std.testing.expectEqualSlices(u8, "", in.interface.buffered());
    }
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo 2\r\nhi\r\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Pub{
                .subject = "foo",
                .reply_to = null,
                .payload = "hi",
            },
            try @"pub"(&in.interface),
        );
        try std.testing.expectEqualSlices(u8, "", in.interface.buffered());
    }
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo " },
            .{ .buffer = "2\r\nhi\r\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Pub{
                .subject = "foo",
                .reply_to = null,
                .payload = "hi",
            },
            try @"pub"(&in.interface),
        );
        try std.testing.expectEqualSlices(u8, "", in.interface.buffered());
    }
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo " },
            .{ .buffer = "2" },
            .{ .buffer = "\r\nhi\r\n " },
        });
        try std.testing.expectEqualDeep(
            Message.Pub{
                .subject = "foo",
                .reply_to = null,
                .payload = "hi",
            },
            try @"pub"(&in.interface),
        );
        try std.testing.expectEqualSlices(u8, " ", in.interface.buffered());
    }
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo " },
            .{ .buffer = "2" },
            .{ .buffer = "\r\nhi\r" },
            .{ .buffer = "\n " },
        });
        try std.testing.expectEqualDeep(
            Message.Pub{
                .subject = "foo",
                .reply_to = null,
                .payload = "hi",
            },
            try @"pub"(&in.interface),
        );
        try std.testing.expectEqualSlices(u8, " ", in.interface.buffered());
    }
}

/// The return value is owned by the reader passed to this function.
/// Operations that modify the readers buffer invalidates this value.
pub fn sub(in: *Reader) !Message.Sub {
    // TODO: Add pedantic option.
    // See: https://docs.nats.io/reference/reference-protocols/nats-protocol#syntax-1

    while (true) {
        var iter = std.mem.tokenizeAny(u8, in.buffered(), " \t\r");

        if (iter.next()) |subject| {
            if (iter.next()) |second| {
                if (in.buffered().len > iter.index) {
                    if (in.buffered()[iter.index] == '\r') {
                        const sid = second;

                        // 2 bytes for CRLF at the end.
                        _ = try in.take(iter.index + 2);
                        return .{
                            .subject = subject,
                            .queue_group = null,
                            .sid = sid,
                        };
                    }

                    const queue_group = second;
                    if (iter.next()) |sid| {
                        if (in.buffered()[iter.index] == '\r') {
                            // 2 bytes for CRLF at the end.
                            _ = try in.take(iter.index + 2);
                            return .{
                                .subject = subject,
                                .queue_group = queue_group,
                                .sid = sid,
                            };
                        }
                    }
                }
            }
        }

        try in.fillMore();
    }
}

test sub {
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo q 1\r\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Sub{
                .subject = "foo",
                .queue_group = "q",
                .sid = "1",
            },
            try sub(&in.interface),
        );
    }
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo  1\r\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Sub{
                .subject = "foo",
                .queue_group = null,
                .sid = "1",
            },
            try sub(&in.interface),
        );
    }
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo " },
            .{ .buffer = " 1\r\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Sub{
                .subject = "foo",
                .queue_group = null,
                .sid = "1",
            },
            try sub(&in.interface),
        );
    }
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo " },
            .{ .buffer = " 1\r" },
            .{ .buffer = "\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Sub{
                .subject = "foo",
                .queue_group = null,
                .sid = "1",
            },
            try sub(&in.interface),
        );
    }
}

/// The return value is owned by the reader passed to this function.
/// Operations that modify the readers buffer invalidates this value.
pub fn unsub(in: *Reader) !Message.Unsub {
    // TODO: Add pedantic option.
    // See: https://docs.nats.io/reference/reference-protocols/nats-protocol#syntax-1

    while (true) {
        var iter = std.mem.tokenizeAny(u8, in.buffered(), " \t\r");

        if (iter.next()) |sid| {
            if (in.buffered()[iter.index] == '\r') {
                // 2 bytes for CRLF at the end.
                _ = try in.take(iter.index + 2);
                return .{
                    .sid = sid,
                    .max_msgs = null,
                };
            }
            if (iter.next()) |max_msgs_str| {
                if (in.buffered()[iter.index] == '\r') {
                    const max_msgs = try parseUnsigned(usize, max_msgs_str, 10);
                    // 2 bytes for CRLF at the end.
                    _ = try in.take(iter.index + 2);
                    return .{
                        .sid = sid,
                        .max_msgs = max_msgs,
                    };
                }
            }
        }

        try in.fillMore();
    }
}

test unsub {
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo 1\r\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Unsub{
                .sid = "foo",
                .max_msgs = 1,
            },
            try unsub(&in.interface),
        );
    }
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo\r\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Unsub{
                .sid = "foo",
                .max_msgs = null,
            },
            try unsub(&in.interface),
        );
    }
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo " },
            .{ .buffer = " 1\r\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Unsub{
                .sid = "foo",
                .max_msgs = 1,
            },
            try unsub(&in.interface),
        );
    }
    {
        var buf: [64]u8 = undefined;
        var in: std.testing.Reader = .init(&buf, &.{
            .{ .buffer = "foo " },
            .{ .buffer = " 1\r" },
            .{ .buffer = "\n" },
        });
        try std.testing.expectEqualDeep(
            Message.Unsub{
                .sid = "foo",
                .max_msgs = 1,
            },
            try unsub(&in.interface),
        );
    }
}

/// The return value is owned by the reader passed to this function.
/// Operations that modify the readers buffer invalidates this value.
pub fn hpub(in: *Reader) !Message.HPub {
    // TODO: Add pedantic option.
    // See: https://docs.nats.io/reference/reference-protocols/nats-protocol#syntax-1
    _ = in;
    @compileError("TODO");
}

test hpub {}

// /// Get the next Message from the input stream.
// pub fn next(alloc: Allocator, in: *Reader) !Message {
//     errdefer log.err("Failed to parse {s}", .{operation_string.items});

//     switch (operation) {
//         .CONNECT => return connect(alloc, in),
//         .PUB => {
//             @branchHint(.likely);
//             return @"pub"(alloc, in);
//         },
//         .HPUB => {
//             @branchHint(.likely);
//             return hpub(alloc, in);
//         },
//         .PING => {
//             try expectStreamBytes(in, "\r\n");
//             return .PING;
//         },
//         .PONG => {
//             try expectStreamBytes(in, "\r\n");
//             return .PONG;
//         },
//         .SUB => return sub(alloc, in),
//         .UNSUB => return unsub(alloc, in),
//         else => |msg| std.debug.panic("Not implemented: {}\n", .{msg}),
//     }
// }

pub fn connect(alloc: Allocator, in: *Reader) !Message.Connect {
    // for storing the json string
    var connect_string_writer_allocating: AllocatingWriter = .init(alloc);
    defer connect_string_writer_allocating.deinit();
    var connect_string_writer = &connect_string_writer_allocating.writer;

    // for parsing the json string
    var connect_arena_allocator: ArenaAllocator = .init(alloc);
    defer connect_arena_allocator.deinit();
    const connect_allocator = connect_arena_allocator.allocator();

    try in.discardAll(1); // throw away space

    // Should read the next JSON object to the fixed buffer writer.
    _ = try in.streamDelimiter(connect_string_writer, '}');
    try connect_string_writer.writeByte('}');
    try expectStreamBytes(in, "}\r\n"); // discard '}\r\n'

    const connect_str = try connect_string_writer_allocating.toOwnedSlice();
    defer alloc.free(connect_str);
    const res = try std.json.parseFromSliceLeaky(
        Message.Connect,
        connect_allocator,
        connect_str,
        .{ .allocate = .alloc_always },
    );

    return res.dupe(alloc);
}

// pub fn sub(alloc: Allocator, in: *Reader) !Message {
//     try in.discardAll(1); // throw away space
//     const subject = try readSubject(alloc, in, .sub);

//     const States = enum {
//         before_second,
//         in_second,
//         after_second,
//         in_third,
//         in_end,
//     };

//     var second: ArrayList(u8) = .empty;
//     errdefer second.deinit(alloc);
//     var third: ?ArrayList(u8) = null;
//     errdefer if (third) |*t| t.deinit(alloc);

//     sw: switch (@as(States, .before_second)) {
//         .before_second => {
//             const byte = try in.peekByte();
//             if (isWhitespace(byte)) {
//                 in.toss(1);
//                 continue :sw .before_second;
//             }
//             continue :sw .in_second;
//         },
//         .in_second => {
//             for (1..in.buffer.len) |i| {
//                 try in.fill(i + 1);
//                 if (isWhitespace(in.buffered()[i])) {
//                     @memcpy(try second.addManyAsSlice(alloc, i), in.buffered()[0..i]);
//                     in.toss(i);
//                     break;
//                 }
//             } else return error.EndOfStream;
//             continue :sw .after_second;
//         },
//         .after_second => {
//             const byte = try in.peekByte();
//             if (byte == '\r') {
//                 continue :sw .in_end;
//             } else if (isWhitespace(byte)) {
//                 in.toss(1);
//                 continue :sw .after_second;
//             }
//             third = .empty;
//             continue :sw .in_third;
//         },
//         .in_third => {
//             for (1..in.buffer.len) |i| {
//                 try in.fill(i + 1);
//                 if (isWhitespace(in.buffered()[i])) {
//                     @memcpy(try third.?.addManyAsSlice(alloc, i), in.buffered()[0..i]);
//                     in.toss(i);
//                     break;
//                 }
//             } else return error.EndOfStream;
//             continue :sw .in_end;
//         },
//         .in_end => {
//             try expectStreamBytes(in, "\r\n");
//         },
//     }

//     return .{
//         .SUB = .{
//             .subject = subject,
//             .queue_group = if (third) |_| try second.toOwnedSlice(alloc) else null,
//             .sid = if (third) |*t| try t.toOwnedSlice(alloc) else try second.toOwnedSlice(alloc),
//         },
//     };
// }

// pub fn unsub(alloc: Allocator, in: *Reader) !Message {
//     const States = enum {
//         before_first,
//         in_first,
//         after_first,
//         in_second,
//         in_end,
//     };

//     var first: ArrayList(u8) = .empty;
//     errdefer first.deinit(alloc);
//     var second: ?ArrayList(u8) = null;
//     defer if (second) |*s| s.deinit(alloc);

//     sw: switch (@as(States, .before_first)) {
//         .before_first => {
//             const byte = try in.peekByte();
//             if (isWhitespace(byte)) {
//                 in.toss(1);
//                 continue :sw .before_first;
//             }
//             continue :sw .in_first;
//         },
//         .in_first => {
//             const byte = try in.peekByte();
//             if (!isWhitespace(byte)) {
//                 try first.append(alloc, byte);
//                 in.toss(1);
//                 continue :sw .in_first;
//             }
//             continue :sw .after_first;
//         },
//         .after_first => {
//             const byte = try in.peekByte();
//             if (byte == '\r') {
//                 continue :sw .in_end;
//             } else if (isWhitespace(byte)) {
//                 in.toss(1);
//                 continue :sw .after_first;
//             }
//             second = .empty;
//             continue :sw .in_second;
//         },
//         .in_second => {
//             const byte = try in.peekByte();
//             if (byte == '\r') {
//                 continue :sw .in_end;
//             }
//             try second.?.append(alloc, byte);
//             in.toss(1);
//             continue :sw .in_second;
//         },
//         .in_end => {
//             try expectStreamBytes(in, "\r\n");
//         },
//     }

//     return .{
//         .UNSUB = .{
//             .sid = try first.toOwnedSlice(alloc),
//             .max_msgs = if (second) |s| try parseUnsigned(usize, s.items, 10) else null,
//         },
//     };
// }

// pub fn @"pub"(alloc: Allocator, in: *Reader) !Message {
//     try in.discardAll(1); // throw away space

//     // Parse subject
//     const subject: []const u8 = try readSubject(alloc, in, .@"pub");
//     errdefer alloc.free(subject);

//     const States = enum {
//         before_second,
//         in_second,
//         after_second,
//         in_third,
//         in_end,
//     };

//     var second: ArrayList(u8) = .empty;
//     defer second.deinit(alloc);
//     var third: ?ArrayList(u8) = null;
//     defer if (third) |*t| t.deinit(alloc);

//     sw: switch (@as(States, .before_second)) {
//         .before_second => {
//             // Drop whitespace
//             const byte = try in.peekByte();
//             if (isWhitespace(byte)) {
//                 in.toss(1);
//                 continue :sw .before_second;
//             }
//             continue :sw .in_second;
//         },
//         .in_second => {
//             for (1..in.buffer.len) |i| {
//                 try in.fill(i + 1);
//                 if (isWhitespace(in.buffered()[i])) {
//                     @memcpy(try second.addManyAsSlice(alloc, i), in.buffered()[0..i]);
//                     in.toss(i);
//                     break;
//                 }
//             } else return error.EndOfStream;
//             continue :sw .after_second;
//         },
//         .after_second => {
//             const byte = try in.peekByte();
//             if (byte == '\r') {
//                 continue :sw .in_end;
//             } else if (isWhitespace(byte)) {
//                 in.toss(1);
//                 continue :sw .after_second;
//             }
//             third = .empty;
//             continue :sw .in_third;
//         },
//         .in_third => {
//             for (1..in.buffer.len) |i| {
//                 try in.fill(i + 1);
//                 if (isWhitespace(in.buffered()[i])) {
//                     @memcpy(try third.?.addManyAsSlice(alloc, i), in.buffered()[0..i]);
//                     in.toss(i);
//                     break;
//                 }
//             } else return error.EndOfStream;
//             continue :sw .in_end;
//         },
//         .in_end => {
//             try expectStreamBytes(in, "\r\n");
//         },
//     }

//     const reply_to: ?[]const u8, const bytes: usize =
//         if (third) |t| .{
//             try alloc.dupe(u8, second.items),
//             try parseUnsigned(usize, t.items, 10),
//         } else .{
//             null,
//             try parseUnsigned(usize, second.items, 10),
//         };

//     const payload: Payload = try .read(alloc, in, bytes);
//     errdefer payload.deinit(alloc);
//     try expectStreamBytes(in, "\r\n");

//     return .{
//         .PUB = .{
//             .subject = subject,
//             .payload = payload,
//             .reply_to = reply_to,
//         },
//     };
// }

// pub fn hpub(alloc: Allocator, in: *Reader) !Message {
//     try in.discardAll(1); // throw away space

//     // Parse subject
//     const subject: []const u8 = try readSubject(alloc, in, .@"pub");
//     errdefer alloc.free(subject);

//     const States = enum {
//         before_second,
//         in_second,
//         after_second,
//         in_third,
//         after_third,
//         in_fourth,
//         in_end,
//     };

//     var second: ArrayList(u8) = .empty;
//     defer second.deinit(alloc);
//     var third: ArrayList(u8) = .empty;
//     defer third.deinit(alloc);
//     var fourth: ?ArrayList(u8) = null;
//     defer if (fourth) |*f| f.deinit(alloc);

//     sw: switch (@as(States, .before_second)) {
//         .before_second => {
//             // Drop whitespace
//             const byte = try in.peekByte();
//             if (isWhitespace(byte)) {
//                 in.toss(1);
//                 continue :sw .before_second;
//             }
//             continue :sw .in_second;
//         },
//         .in_second => {
//             for (1..in.buffer.len) |i| {
//                 try in.fill(i + 1);
//                 if (isWhitespace(in.buffered()[i])) {
//                     @memcpy(try second.addManyAsSlice(alloc, i), in.buffered()[0..i]);
//                     in.toss(i);
//                     break;
//                 }
//             } else return error.EndOfStream;
//             continue :sw .after_second;
//         },
//         .after_second => {
//             const byte = try in.peekByte();
//             if (byte == '\r') {
//                 continue :sw .in_end;
//             } else if (isWhitespace(byte)) {
//                 in.toss(1);
//                 continue :sw .after_second;
//             }
//             third = .empty;
//             continue :sw .in_third;
//         },
//         .in_third => {
//             for (1..in.buffer.len) |i| {
//                 try in.fill(i + 1);
//                 if (isWhitespace(in.buffered()[i])) {
//                     @memcpy(try third.addManyAsSlice(alloc, i), in.buffered()[0..i]);
//                     in.toss(i);
//                     break;
//                 }
//             } else return error.EndOfStream;
//             continue :sw .after_third;
//         },
//         .after_third => {
//             const byte = try in.peekByte();
//             if (byte == '\r') {
//                 continue :sw .in_end;
//             } else if (isWhitespace(byte)) {
//                 in.toss(1);
//                 continue :sw .after_third;
//             }
//             fourth = .empty;
//             continue :sw .in_fourth;
//         },
//         .in_fourth => {
//             for (1..in.buffer.len) |i| {
//                 try in.fill(i + 1);
//                 if (isWhitespace(in.buffered()[i])) {
//                     @memcpy(try fourth.?.addManyAsSlice(alloc, i), in.buffered()[0..i]);
//                     in.toss(i);
//                     break;
//                 }
//             } else return error.EndOfStream;
//             continue :sw .in_end;
//         },
//         .in_end => {
//             try expectStreamBytes(in, "\r\n");
//         },
//     }

//     const reply_to: ?[]const u8, const header_bytes: usize, const total_bytes: usize =
//         if (fourth) |f| .{
//             try alloc.dupe(u8, second.items),
//             try parseUnsigned(usize, third.items, 10),
//             try parseUnsigned(usize, f.items, 10),
//         } else .{
//             null,
//             try parseUnsigned(usize, second.items, 10),
//             try parseUnsigned(usize, third.items, 10),
//         };

//     const payload: Payload = try .read(alloc, in, total_bytes);
//     errdefer payload.deinit(alloc);
//     try expectStreamBytes(in, "\r\n");

//     return .{
//         .HPUB = .{
//             .header_bytes = header_bytes,
//             .@"pub" = .{
//                 .subject = subject,
//                 .payload = payload,
//                 .reply_to = reply_to,
//             },
//         },
//     };
// }

// fn readSubject(alloc: Allocator, in: *Reader, comptime pub_or_sub: enum { @"pub", sub }) ![]const u8 {
//     var subject_list: ArrayList(u8) = .empty;
//     errdefer subject_list.deinit(alloc);

//     // Handle the first character
//     {
//         const byte = try in.takeByte();
//         if (isWhitespace(byte) or byte == '.' or (pub_or_sub == .@"pub" and (byte == '*' or byte == '>')))
//             return error.InvalidStream;

//         try subject_list.append(alloc, byte);
//     }

//     switch (pub_or_sub) {
//         .sub => {
//             while (in.takeByte()) |byte| {
//                 if (isWhitespace(byte)) break;
//                 if (byte == '.') {
//                     const next_byte = try in.peekByte();
//                     if (next_byte == '.' or isWhitespace(next_byte))
//                         return error.InvalidStream;
//                 } else if (byte == '>') {
//                     const next_byte = try in.takeByte();
//                     if (!isWhitespace(next_byte))
//                         return error.InvalidStream;
//                 } else if (byte == '*') {
//                     const next_byte = try in.peekByte();
//                     if (next_byte != '.' and !isWhitespace(next_byte))
//                         return error.InvalidStream;
//                 }
//                 try subject_list.append(alloc, byte);
//             } else |err| return err;
//         },
//         .@"pub" => {
//             while (in.takeByte()) |byte| {
//                 if (isWhitespace(byte)) break;
//                 if (byte == '*' or byte == '>') return error.InvalidStream;
//                 if (byte == '.') {
//                     const next_byte = try in.peekByte();
//                     if (next_byte == '.' or isWhitespace(next_byte))
//                         return error.InvalidStream;
//                 }
//                 try subject_list.append(alloc, byte);
//             } else |err| return err;
//         },
//     }

//     return subject_list.toOwnedSlice(alloc);
// }

inline fn expectStreamBytes(reader: *Reader, expected: []const u8) !void {
    if (!std.mem.eql(u8, try reader.take(expected.len), expected)) {
        @branchHint(.unlikely);
        return error.InvalidStream;
    }
}

// test sub {
//     const alloc = std.testing.allocator;
//     const expectEqualDeep = std.testing.expectEqualDeep;
//     {
//         var in: Reader = .fixed(" foo 1\r\n");
//         var res = try sub(alloc, &in);
//         defer res.SUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .SUB = .{
//                     .subject = "foo",
//                     .queue_group = null,
//                     .sid = "1",
//                 },
//             },
//             res,
//         );
//     }
//     {
//         var in: Reader = .fixed(" foo  1\r\n");
//         var res = try sub(alloc, &in);
//         defer res.SUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .SUB = .{
//                     .subject = "foo",
//                     .queue_group = null,
//                     .sid = "1",
//                 },
//             },
//             res,
//         );
//     }
//     {
//         var in: Reader = .fixed(" foo q 1\r\n");
//         var res = try sub(alloc, &in);
//         defer res.SUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .SUB = .{
//                     .subject = "foo",
//                     .queue_group = "q",
//                     .sid = "1",
//                 },
//             },
//             res,
//         );
//     }
//     {
//         var in: Reader = .fixed(" 1 q 1\r\n");
//         var res = try sub(alloc, &in);
//         defer res.SUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .SUB = .{
//                     .subject = "1",
//                     .queue_group = "q",
//                     .sid = "1",
//                 },
//             },
//             res,
//         );
//     }
//     {
//         var in: Reader = .fixed(" $SRV.PING  4\r\n");
//         var res = try sub(alloc, &in);
//         defer res.SUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .SUB = .{
//                     .subject = "$SRV.PING",
//                     .queue_group = null,
//                     .sid = "4",
//                 },
//             },
//             res,
//         );
//     }
//     {
//         var in: Reader = .fixed(" foo.echo q 10\r\n");
//         var res = try sub(alloc, &in);
//         defer res.SUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .SUB = .{
//                     .subject = "foo.echo",
//                     .queue_group = "q",
//                     .sid = "10",
//                 },
//             },
//             res,
//         );
//     }
// }

// test unsub {
//     const alloc = std.testing.allocator;
//     const expectEqualDeep = std.testing.expectEqualDeep;
//     const expectEqual = std.testing.expectEqual;
//     {
//         var in: Reader = .fixed(" 1\r\n");
//         var res = try unsub(alloc, &in);
//         defer res.UNSUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .UNSUB = .{
//                     .sid = "1",
//                     .max_msgs = null,
//                 },
//             },
//             res,
//         );
//         try expectEqual(0, in.buffered().len);
//     }

//     {
//         var in: Reader = .fixed(" 1 1\r\n");
//         var res = try unsub(alloc, &in);
//         defer res.UNSUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .UNSUB = .{
//                     .sid = "1",
//                     .max_msgs = 1,
//                 },
//             },
//             res,
//         );
//         try expectEqual(0, in.buffered().len);
//     }
// }

// test @"pub" {
//     const alloc = std.testing.allocator;
//     const expectEqualDeep = std.testing.expectEqualDeep;
//     const expectEqual = std.testing.expectEqual;
//     {
//         var in: Reader = .fixed(" foo 3\r\nbar\r\n");
//         var res = try @"pub"(alloc, &in);
//         defer res.PUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .PUB = .{
//                     .subject = "foo",
//                     .reply_to = null,
//                     .payload = .{
//                         .len = 3,
//                         .short = blk: {
//                             var s: [128]u8 = undefined;
//                             @memcpy(s[0..3], "bar");
//                             break :blk s;
//                         },
//                         .long = null,
//                     },
//                 },
//             },
//             res,
//         );
//         try expectEqual(0, in.buffered().len);
//     }

//     {
//         var in: Reader = .fixed(" foo reply.to 3\r\nbar\r\n");
//         var res = try @"pub"(alloc, &in);
//         defer res.PUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .PUB = .{
//                     .subject = "foo",
//                     .reply_to = "reply.to",
//                     .payload = .{
//                         .len = 3,
//                         .short = blk: {
//                             var s: [128]u8 = undefined;
//                             @memcpy(s[0..3], "bar");
//                             break :blk s;
//                         },
//                         .long = null,
//                     },
//                 },
//             },
//             res,
//         );
//         try expectEqual(0, in.buffered().len);
//     }

//     // numeric reply subject
//     {
//         var in: Reader = .fixed(" foo 5 3\r\nbar\r\n");
//         var res = try @"pub"(alloc, &in);
//         defer res.PUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .PUB = .{
//                     .subject = "foo",
//                     .reply_to = "5",
//                     .payload = .{
//                         .len = 3,
//                         .short = blk: {
//                             var s: [128]u8 = undefined;
//                             @memcpy(s[0..3], "bar");
//                             break :blk s;
//                         },
//                         .long = null,
//                     },
//                 },
//             },
//             res,
//         );
//         try expectEqual(0, in.buffered().len);
//     }
// }

// test hpub {
//     const alloc = std.testing.allocator;
//     const expectEqualDeep = std.testing.expectEqualDeep;
//     const expectEqual = std.testing.expectEqual;
//     {
//         var in: Reader = .fixed(" foo 22 33\r\nNATS/1.0\r\nBar: Baz\r\n\r\nHello NATS!\r\n");
//         var res = try hpub(alloc, &in);
//         defer res.HPUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .HPUB = .{
//                     .header_bytes = 22,
//                     .@"pub" = .{
//                         .subject = "foo",
//                         .reply_to = null,
//                         .payload = .{
//                             .len = "NATS/1.0\r\nBar: Baz\r\n\r\nHello NATS!".len,
//                             .short = blk: {
//                                 var s: [128]u8 = undefined;
//                                 const str = "NATS/1.0\r\nBar: Baz\r\n\r\nHello NATS!";
//                                 @memcpy(s[0..str.len], str);
//                                 break :blk s;
//                             },
//                             .long = null,
//                         },
//                     },
//                 },
//             },
//             res,
//         );
//         try expectEqual(0, in.buffered().len);
//     }

//     {
//         var in: Reader = .fixed(" foo reply.to 22 33\r\nNATS/1.0\r\nBar: Baz\r\n\r\nHello NATS!\r\n");
//         var res = try hpub(alloc, &in);
//         defer res.HPUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .HPUB = .{
//                     .header_bytes = 22,
//                     .@"pub" = .{
//                         .subject = "foo",
//                         .reply_to = "reply.to",
//                         .payload = .{
//                             .len = "NATS/1.0\r\nBar: Baz\r\n\r\nHello NATS!".len,
//                             .short = blk: {
//                                 var s: [128]u8 = undefined;
//                                 const str = "NATS/1.0\r\nBar: Baz\r\n\r\nHello NATS!";
//                                 @memcpy(s[0..str.len], str);
//                                 break :blk s;
//                             },
//                             .long = null,
//                         },
//                     },
//                 },
//             },
//             res,
//         );
//         try expectEqual(0, in.buffered().len);
//     }

//     {
//         var in: Reader = .fixed(" foo 6 22 33\r\nNATS/1.0\r\nBar: Baz\r\n\r\nHello NATS!\r\n");
//         var res = try hpub(alloc, &in);
//         defer res.HPUB.deinit(alloc);
//         try expectEqualDeep(
//             Message{
//                 .HPUB = .{
//                     .header_bytes = 22,
//                     .@"pub" = .{
//                         .subject = "foo",
//                         .reply_to = "6",
//                         .payload = .{
//                             .len = "NATS/1.0\r\nBar: Baz\r\n\r\nHello NATS!".len,
//                             .short = blk: {
//                                 var s: [128]u8 = undefined;
//                                 const str = "NATS/1.0\r\nBar: Baz\r\n\r\nHello NATS!";
//                                 @memcpy(s[0..str.len], str);
//                                 break :blk s;
//                             },
//                             .long = null,
//                         },
//                     },
//                 },
//             },
//             res,
//         );
//         try expectEqual(0, in.buffered().len);
//     }
// }