feat: add handlePublish and handleUnsubscribe to server

src/server.zig

@@ -1541,3 +1541,161 @@ pub const Server = struct {
         hub.closeSubscriber(sub);
     }
 
+    /// Executes a `publish` request: any connection may call this
+    /// regardless of whether it has ever subscribed to anything (see
+    /// `protocol.encodePublish`'s doc comment). Delivery is fire-and-forget
+    /// from this handler's point of view - `pubsub.Hub.publish` never
+    /// blocks on a slow subscriber, so this returns as soon as every
+    /// currently-registered subscriber's mailbox has been handed its own
+    /// copy (or dropped one, under `Subscriber.push`'s backpressure policy).
+    fn handlePublish(self: *Server, payload: []const u8, writer: anytype) !void {
+        const args = protocol.decodePublish(payload) catch {
+            return writeErr(writer, .bad_request, "malformed publish payload");
+        };
+        const count = self.pubsub.publish(self.allocator, args.channel, args.message) catch {
+            return writeErr(writer, .internal, "failed to publish message");
+        };
+        if (count > 0) self.messagesPublishedTotal.incrBy(@intCast(count));
+        const resp = try protocol.encodePublishResult(self.allocator, @intCast(count));
+        defer self.allocator.free(resp);
+        try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), resp);
+    }
+
+    /// Executes an `unsubscribe` request: unlike `subscribe`, this is a
+    /// plain request/response opcode that returns like any other - it never
+    /// touches the connection that originally subscribed, because that
+    /// connection became receive-only the moment its ack arrived (see
+    /// `handleSubscribe`'s doc comment) and this one is always a separate
+    /// connection entirely (see `protocol.encodeUnsubscribe`'s doc
+    /// comment). `pubsub.Hub.unsubscribe`'s `error.UnknownSubscriber` (a
+    /// stale id, a typo, or a subscriber that already disconnected on its
+    /// own - all indistinguishable from here, see that function's doc
+    /// comment) is the caller's mistake, `bad_request`, the same treatment
+    /// `table_window` gives a `MissingField` from a bad field name.
+    fn handleUnsubscribe(self: *Server, payload: []const u8, writer: anytype) !void {
+        const args = protocol.decodeUnsubscribe(self.allocator, payload) catch {
+            return writeErr(writer, .bad_request, "malformed unsubscribe payload");
+        };
+        defer args.deinit(self.allocator);
+        if (args.channels.len == 0) {
+            return writeErr(writer, .bad_request, "unsubscribe needs at least one channel");
+        }
+
+        const channelNames = try self.allocator.alloc([]const u8, args.channels.len);
+        defer self.allocator.free(channelNames);
+        for (args.channels, 0..) |c, i| channelNames[i] = c;
+
+        const removed = self.pubsub.unsubscribe(args.subscriberId, channelNames) catch |err| {
+            return switch (err) {
+                error.UnknownSubscriber => writeErr(writer, .bad_request, "unknown subscriber id"),
+                else => writeErr(writer, .internal, @errorName(err)),
+            };
+        };
+
+        const resp = try protocol.encodeUnsubscribeResult(self.allocator, @intCast(removed));
+        defer self.allocator.free(resp);
+        try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), resp);
+    }
+
+    fn handleTableWindow(self: *Server, payload: []const u8, writer: anytype) !void {
+        const args = protocol.decodeTableWindow(self.allocator, payload) catch {
+            return writeErr(writer, .bad_request, "malformed table_window payload");
+        };
+        defer self.allocator.free(args.specs);
+        defer self.allocator.free(args.partitionBy);
+        defer self.allocator.free(args.orderBy);
+
+        var table = querymod.Table.open(self.db, self.allocator, args.table) catch |err| {
+            return writeErr(writer, .bad_request, @errorName(err));
+        };
+        defer table.deinit();
+
+        var parsed: ?exprmod.Parsed = null;
+        defer if (parsed) |*p| p.deinit();
+        if (args.exprText.len > 0) {
+            parsed = exprmod.parse(self.allocator, args.exprText) catch |err| {
+                return writeErr(writer, .bad_request, @errorName(err));
+            };
+        }
+        const pred: ?querymod.Predicate = if (parsed) |p| p.predicate else null;
+
+        const rows = table.select(pred) catch |err| {
+            return writeErr(writer, .internal, @errorName(err));
+        };
+        defer querymod.freeRows(self.allocator, rows);
+
+        const results = windowmod.computeMany(self.allocator, rows, args.partitionBy, args.orderBy, args.specs, args.frame) catch |err| {
+            return switch (err) {
+                error.NonNumericField, error.Overflow, error.MissingField, error.RangeNotApplicable => writeErr(writer, .bad_request, @errorName(err)),
+                else => writeErr(writer, .internal, @errorName(err)),
+            };
+        };
+        defer windowmod.freeManyResults(self.allocator, results);
+
+        const displayOrder = windowmod.orderedIndices(self.allocator, rows, args.partitionBy, args.orderBy) catch |err| {
+            return writeErr(writer, .internal, @errorName(err));
+        };
+        defer self.allocator.free(displayOrder);
+
+        const total = displayOrder.len;
+        const pageStart = @min(args.offset, total);
+        const end = if (args.limit == 0) total else @min(pageStart + args.limit, total);
+
+        // `out`'s WindowRow.values arrays hold SHALLOW copies borrowed from
+        // `results` (safe: `encodeWindowRows` below only reads them to
+        // serialize, never frees them) - `results` stays the true owner,
+        // freed by the `defer` above; only the borrowing arrays themselves
+        // get freed here, same "shallow copy into a throwaway wire-shape
+        // struct, one true owner frees the contents" pattern
+        // `handleTableAggregate`'s `GroupRow`-to-wire path already relies on.
+        var out = std.ArrayList(protocol.WindowRow).init(self.allocator);
+        defer {
+            for (out.items) |wr| self.allocator.free(wr.values);
+            out.deinit();
+        }
+        var i: usize = pageStart;
+        while (i < end) : (i += 1) {
+            const idx = displayOrder[i];
+            const values = try self.allocator.alloc(rowmod.Value, args.specs.len);
+            @memcpy(values, results[idx]);
+            try out.append(.{ .row = rows[idx], .values = values });
+        }
+
+        const encoded = protocol.encodeWindowRows(self.allocator, out.items) catch |err| {
+            return writeErr(writer, .internal, @errorName(err));
+        };
+        defer self.allocator.free(encoded);
+        try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), encoded);
+    }
+
+    /// Executes a `table_explain` request: opens the named table over the
+    /// same `*Db` the rest of the server serves, parses the expr text with
+    /// the same grammar `table_select`/`table_join`/`table_aggregate`
+    /// already use, then reports which strategy `querymod.Table.explain`
+    /// says a matching `table select`/`table_select` would take - without
+    /// running the query or touching any row. A bad table name or malformed
+    /// expression is a client mistake (`bad_request`), not a server fault.
+    fn handleTableExplain(self: *Server, payload: []const u8, writer: anytype) !void {
+        const args = protocol.decodeTableExplain(payload) catch {
+            return writeErr(writer, .bad_request, "malformed table_explain payload");
+        };
+
+        var table = querymod.Table.open(self.db, self.allocator, args.table) catch |err| {
+            return writeErr(writer, .bad_request, @errorName(err));
+        };
+        defer table.deinit();
+
+        var parsed: ?exprmod.Parsed = null;
+        defer if (parsed) |*p| p.deinit();
+        if (args.exprText.len > 0) {
+            parsed = exprmod.parse(self.allocator, args.exprText) catch |err| {
+                return writeErr(writer, .bad_request, @errorName(err));
+            };
+        }
+        const pred: ?querymod.Predicate = if (parsed) |p| p.predicate else null;
+
+        var plan = table.explain(self.allocator, pred) catch |err| {
+            return writeErr(writer, .internal, @errorName(err));
+        };
+        defer plan.deinit(self.allocator);
+