diff --git a/src/server.zig b/src/server.zig index 56f63b2..8a30e44 100644 --- a/src/server.zig +++ b/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); +