feat: add decodeSubscribe and encodeSubscribeAck to protocol

src/protocol.zig

@@ -1717,3 +1717,163 @@ pub const SubscribeArgs = struct {
     }
 };
 
+/// Decodes a `subscribe` request into independently owned channel-name
+/// copies - free with `SubscribeArgs.deinit`. Owned (rather than borrowed
+/// slices into `payload`, as most other decoders here return) because
+/// `handleSubscribe` hands these straight to `pubsub.Hub.subscribe`, which
+/// keeps its own copies anyway - decoding once into memory that outlives
+/// the request frame avoids a second, redundant copy inside the hub for
+/// every subscribe call.
+pub fn decodeSubscribe(allocator: Allocator, payload: []const u8) !SubscribeArgs {
+    var c = Cursor{ .data = payload };
+    const count = try c.takeU32();
+    var list = try std.ArrayList([]u8).initCapacity(allocator, count);
+    errdefer {
+        for (list.items) |ch| allocator.free(ch);
+        list.deinit();
+    }
+    var i: u32 = 0;
+    while (i < count) : (i += 1) {
+        const name = try c.takeLenPrefixed();
+        const owned = try allocator.dupe(u8, name);
+        list.appendAssumeCapacity(owned);
+    }
+    return .{ .channels = try list.toOwnedSlice() };
+}
+
+/// A `subscribe` request's one and only reply frame (status `.ok`, this as
+/// its payload) - sent once the connection is actually registered in the
+/// `pubsub.Hub`, so a client that has received this ack is guaranteed not
+/// to miss a message published right after it returns. Carries the
+/// (deduplicated) channel count purely as a sanity-checkable echo, the same
+/// spirit as `encodeAuthOk` echoing a token back - plus the `pubsub.Hub`-
+/// assigned subscriber id, the only way a later `unsubscribe` request (on a
+/// SEPARATE connection - this one is receive-only from here on, see
+/// `server.zig`'s `handleSubscribe`) can ever name this subscription again.
+pub fn encodeSubscribeAck(allocator: Allocator, subscriberId: u64, channelCount: u32) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try w.putU64(subscriberId);
+    try w.putU32(channelCount);
+    return buf.toOwnedSlice();
+}
+
+pub const SubscribeAck = struct { subscriberId: u64, channelCount: u32 };
+
+pub fn decodeSubscribeAck(payload: []const u8) !SubscribeAck {
+    var c = Cursor{ .data = payload };
+    const subscriberId = try c.takeU64();
+    const channelCount = try c.takeU32();
+    return .{ .subscriberId = subscriberId, .channelCount = channelCount };
+}
+
+/// Wire request for `publish`: a channel name and the message bytes (binary
+/// safe, "rest of the payload" like `encodeSet`'s value). Any connection
+/// may publish regardless of whether it has ever subscribed to anything -
+/// publishing and subscribing are independent roles on the same opcode
+/// family, exactly like a `Db` writer and a `Db` reader are independent
+/// roles over the same store.
+pub fn encodePublish(allocator: Allocator, channel: []const u8, message: []const u8) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try putLenPrefixed(w, channel);
+    try w.slice(message);
+    return buf.toOwnedSlice();
+}
+
+pub const PublishArgs = struct { channel: []const u8, message: []const u8 };
+
+pub fn decodePublish(payload: []const u8) !PublishArgs {
+    var c = Cursor{ .data = payload };
+    const channel = try c.takeLenPrefixed();
+    const message = c.data[c.pos..];
+    return .{ .channel = channel, .message = message };
+}
+
+/// A `publish` request's reply payload: how many subscribers `pubsub.Hub`
+/// handed the message to (see `Hub.publish`'s doc comment - 0 is a normal,
+/// non-error outcome for a channel nobody is listening to).
+pub fn encodePublishResult(allocator: Allocator, subscriberCount: u64) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try w.putU64(subscriberCount);
+    return buf.toOwnedSlice();
+}
+
+pub fn decodePublishResult(payload: []const u8) !u64 {
+    var c = Cursor{ .data = payload };
+    return c.takeU64();
+}
+
+/// Wire request for `unsubscribe`: the `pubsub.Hub`-assigned subscriber id
+/// (from `encodeSubscribeAck`) plus the list of channel names to drop from
+/// that subscription - the same channel-list shape `encodeSubscribe` uses,
+/// with the id prefixed. Unlike `subscribe`, this is an ordinary
+/// request/response opcode (see `server.zig`'s `handleUnsubscribe`): it is
+/// always sent on a DIFFERENT connection than the one that subscribed,
+/// since that connection became receive-only the moment its subscribe ack
+/// arrived and can never send another request on itself again.
+pub fn encodeUnsubscribe(allocator: Allocator, subscriberId: u64, channels: []const []const u8) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try w.putU64(subscriberId);
+    try w.putU32(@intCast(channels.len));
+    for (channels) |c| try putLenPrefixed(w, c);
+    return buf.toOwnedSlice();
+}
+
+pub const UnsubscribeArgs = struct {
+    subscriberId: u64,
+    channels: [][]u8,
+
+    pub fn deinit(self: UnsubscribeArgs, allocator: Allocator) void {
+        for (self.channels) |c| allocator.free(c);
+        allocator.free(self.channels);
+    }
+};
+
+/// Decodes an `unsubscribe` request - same owned-copy rationale as
+/// `decodeSubscribe`: `handleUnsubscribe` hands the channel names straight
+/// to `pubsub.Hub.unsubscribe`, which only ever borrows them for the
+/// duration of that call, so decoding into memory this function's caller
+/// frees (via `UnsubscribeArgs.deinit`) avoids a redundant second copy.
+pub fn decodeUnsubscribe(allocator: Allocator, payload: []const u8) !UnsubscribeArgs {
+    var c = Cursor{ .data = payload };
+    const subscriberId = try c.takeU64();
+    const count = try c.takeU32();
+    var list = try std.ArrayList([]u8).initCapacity(allocator, count);
+    errdefer {
+        for (list.items) |ch| allocator.free(ch);
+        list.deinit();
+    }
+    var i: u32 = 0;
+    while (i < count) : (i += 1) {
+        const name = try c.takeLenPrefixed();
+        const owned = try allocator.dupe(u8, name);
+        list.appendAssumeCapacity(owned);
+    }
+    return .{ .subscriberId = subscriberId, .channels = try list.toOwnedSlice() };
+}
+
+/// An `unsubscribe` request's reply payload: how many of the requested
+/// channel names were actually removed from the named subscriber's set (see
+/// `pubsub.Hub.unsubscribe`'s doc comment on why naming a channel it was
+/// never on is a harmless no-op that simply does not count here, not an
+/// error).
+pub fn encodeUnsubscribeResult(allocator: Allocator, removedCount: u32) ![]u8 {
+    var buf = std.ArrayList(u8).init(allocator);
+    errdefer buf.deinit();
+    const w = encode.ByteWriter.init(&buf);
+    try w.putU32(removedCount);
+    return buf.toOwnedSlice();
+}
+
+pub fn decodeUnsubscribeResult(payload: []const u8) !u32 {
+    var c = Cursor{ .data = payload };
+    return c.takeU32();
+}
+