feat: add decodeTableWindow and WindowRow to protocol

src/protocol.zig

@@ -1570,3 +1570,150 @@ pub const TableWindowArgs = struct {
     exprText: []const u8,
 };
 
+pub fn decodeTableWindow(allocator: Allocator, payload: []const u8) !TableWindowArgs {
+    var c = Cursor{ .data = payload };
+    const table = try c.takeLenPrefixed();
+    const specCount = try c.takeByte();
+    var specs = try std.ArrayList(windowmod.WindowSpec).initCapacity(allocator, specCount);
+    errdefer specs.deinit();
+    var s: u8 = 0;
+    while (s < specCount) : (s += 1) {
+        const kindByte = try c.takeByte();
+        const kind = std.meta.intToEnum(windowmod.WindowKind, kindByte) catch return error.Malformed;
+        const field = try c.takeLenPrefixed();
+        if (kind == .lag or kind == .lead or kind == .nth_value) {
+            const extra = try takeLagLeadExtra(&c);
+            specs.appendAssumeCapacity(.{ .kind = kind, .field = field, .offset = extra.offset, .default = extra.default });
+        } else {
+            specs.appendAssumeCapacity(.{ .kind = kind, .field = field });
+        }
+    }
+    const partitionCount = try c.takeByte();
+    var partitionBy = try std.ArrayList([]const u8).initCapacity(allocator, partitionCount);
+    errdefer partitionBy.deinit();
+    var i: u8 = 0;
+    while (i < partitionCount) : (i += 1) partitionBy.appendAssumeCapacity(try c.takeLenPrefixed());
+    const orderCount = try c.takeByte();
+    var orderBy = try std.ArrayList(querymod.OrderBy).initCapacity(allocator, orderCount);
+    errdefer orderBy.deinit();
+    var j: u8 = 0;
+    while (j < orderCount) : (j += 1) {
+        const descending = try c.takeByte();
+        const obField = try c.takeLenPrefixed();
+        orderBy.appendAssumeCapacity(.{ .field = obField, .descending = descending != 0 });
+    }
+    const offset = try c.takeU32();
+    const limit = try c.takeU32();
+    const frameByte = try c.takeByte();
+    const frame = std.meta.intToEnum(windowmod.FrameMode, frameByte) catch return error.Malformed;
+    const exprText = c.data[c.pos..];
+    return .{
+        .table = table,
+        .specs = try specs.toOwnedSlice(),
+        .partitionBy = try partitionBy.toOwnedSlice(),
+        .orderBy = try orderBy.toOwnedSlice(),
+        .offset = offset,
+        .limit = limit,
+        .frame = frame,
+        .exprText = exprText,
+    };
+}
+
+/// One annotated row from a `table_window` result: the original row
+/// (unmodified, same fields the request matched) paired with one computed
+/// window value PER requested spec, in the same order as the request's
+/// `specs` list - the network shape of what local `table window` prints per
+/// row (`printRow` fields, then `<label>=<value>` per spec). Already in the
+/// server-paginated partition-then-order display order (see
+/// `handleTableWindow`), so the client prints these in order with no
+/// further sorting.
+pub const WindowRow = struct { row: rowmod.Row, values: []rowmod.Value };
+
+pub fn encodeWindowRows(allocator: Allocator, rows: []const WindowRow) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try w.putU32(@intCast(rows.len));
+    for (rows) |wr| {
+        const encoded = try wr.row.encode(allocator);
+        defer allocator.free(encoded);
+        try putLenPrefixed(w, encoded);
+        try w.putU32(@intCast(wr.values.len));
+        for (wr.values) |v| try putEncodedValue(allocator, w, v);
+    }
+    return buf.toOwnedSlice();
+}
+
+/// Decodes a `table_window` result into independently owned `WindowRow`s -
+/// callers must free each (`.row.deinit()`, each `.values[i].free(allocator)`
+/// then `allocator.free(.values)`), then the slice itself (see
+/// `freeWindowRows`).
+pub fn decodeWindowRows(allocator: Allocator, payload: []const u8) ![]WindowRow {
+    var c = Cursor{ .data = payload };
+    const count = try c.takeU32();
+    var list = try std.ArrayList(WindowRow).initCapacity(allocator, count);
+    errdefer {
+        for (list.items) |*wr| {
+            wr.row.deinit();
+            for (wr.values) |v| v.free(allocator);
+            allocator.free(wr.values);
+        }
+        list.deinit();
+    }
+    var i: u32 = 0;
+    while (i < count) : (i += 1) {
+        const raw = try c.takeLenPrefixed();
+        var row = rowmod.Row.decode(allocator, raw) catch return error.Malformed;
+        errdefer row.deinit();
+        const valueCount = try c.takeU32();
+        var values = try allocator.alloc(rowmod.Value, valueCount);
+        errdefer allocator.free(values);
+        var filled: u32 = 0;
+        errdefer for (values[0..filled]) |v| v.free(allocator);
+        var j: u32 = 0;
+        while (j < valueCount) : (j += 1) {
+            values[j] = try takeDecodedValue(allocator, &c);
+            filled += 1;
+        }
+        list.appendAssumeCapacity(.{ .row = row, .values = values });
+    }
+    return list.toOwnedSlice();
+}
+
+pub fn freeWindowRows(allocator: Allocator, rows: []WindowRow) void {
+    for (rows) |*wr| {
+        wr.row.deinit();
+        for (wr.values) |v| v.free(allocator);
+        allocator.free(wr.values);
+    }
+    allocator.free(rows);
+}
+
+// --- pubsub ---
+
+/// Wire request for `subscribe`: the list of channel names to listen on (at
+/// least one - `server.zig`'s `handleSubscribe` rejects an empty list as
+/// malformed rather than opening a subscription that can never receive
+/// anything). Unlike every other opcode in this file, a `subscribe` request
+/// does not get a single reply frame back: after the initial acknowledgment
+/// (see `encodeSubscribeAck`), the connection becomes a one-way stream of
+/// `encodePubSubMessage` frames until it closes - see `pubsub.zig`'s doc
+/// comment on `Hub` and `server.zig`'s `handleSubscribe`.
+pub fn encodeSubscribe(allocator: Allocator, channels: []const []const u8) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try w.putU32(@intCast(channels.len));
+    for (channels) |c| try putLenPrefixed(w, c);
+    return buf.toOwnedSlice();
+}
+
+pub const SubscribeArgs = struct {
+    channels: [][]u8,
+
+    pub fn deinit(self: SubscribeArgs, allocator: Allocator) void {
+        for (self.channels) |c| allocator.free(c);
+        allocator.free(self.channels);
+    }
+};
+