feat: add tableJoin and tableAggregateGlobal to client

src/client.zig

@@ -320,3 +320,164 @@ pub const Client = struct {
         return protocol.decodeRows(allocator, frame.payload);
     }
 
+    /// Remote `table join`: the wire counterpart of the CLI's local `table
+    /// join [--left|--right|--full]`, running whichever of `join.zig`'s four
+    /// equi-join variants `mode` names against the server's `*Db` instead of
+    /// a locally opened one. `leftExprText`/`rightExprText` are the same
+    /// `expr.zig` predicate grammar as `tableSelect`'s `exprText` (empty
+    /// selects every row on that side), filtering each table before the
+    /// join runs - the network counterpart of `innerJoin`'s
+    /// `leftPred`/`rightPred` parameters. `offset`/`limit` window the
+    /// already-joined result server-side (0/0 = unlimited), the same
+    /// convention `tableSelect` uses, via `join.applyPage`. Returned pairs
+    /// are independently owned - free with `protocol.freeJoinPairs`.
+    pub fn tableJoin(
+        self: *Client,
+        allocator: Allocator,
+        mode: protocol.JoinMode,
+        leftTable: []const u8,
+        leftField: []const u8,
+        rightTable: []const u8,
+        rightField: []const u8,
+        offset: u32,
+        limit: u32,
+        leftExprText: []const u8,
+        rightExprText: []const u8,
+    ) ![]protocol.JoinPair {
+        const payload = try protocol.encodeTableJoin(self.allocator, mode, leftTable, leftField, rightTable, rightField, offset, limit, leftExprText, rightExprText);
+        defer self.allocator.free(payload);
+        var frame = try self.roundTrip(.table_join, payload);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeJoinedRows(allocator, frame.payload);
+    }
+
+    /// Remote `table aggregate <table> * <specs> [expr...]` (no `GROUP BY`):
+    /// the wire counterpart of the CLI's local `table aggregate` with a `*`
+    /// group field, calling `aggregate.runGlobal` server-side over rows
+    /// filtered by `exprText` (same `expr.zig` grammar, empty selects every
+    /// row, as `tableSelect`). `havingText` (empty = no HAVING) filters the
+    /// single result server-side via `aggregate.globalMatchesHaving` -
+    /// returns zero values (not an error) when it excludes the only row.
+    /// Otherwise always returns exactly `specs.len` values in the same order
+    /// as `specs` - free with `protocol.freeAggValues`.
+    pub fn tableAggregateGlobal(self: *Client, allocator: Allocator, table: []const u8, specs: []const aggregatemod.AggSpec, havingText: []const u8, exprText: []const u8) ![]rowmod.Value {
+        const payload = try protocol.encodeTableAggregate(self.allocator, table, "", specs, 0, 0, &.{}, havingText, exprText);
+        defer self.allocator.free(payload);
+        var frame = try self.roundTrip(.table_aggregate, payload);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeAggGlobalResult(allocator, frame.payload);
+    }
+
+    /// Remote `table aggregate <table> <groupField> <specs> [expr...]
+    /// [having <expr>] [--limit N] [--offset N]`: the wire counterpart of
+    /// the CLI's local `table aggregate` with a real GROUP BY field, calling
+    /// `aggregate.run` server-side, then (when `havingText` is non-empty)
+    /// `aggregate.filterHaving` to narrow the groups, then windowing what
+    /// remains server-side with `aggregate.applyPageGroups` (0/0 =
+    /// unlimited, the same convention `tableJoin`'s `offset`/`limit` use -
+    /// HAVING always runs before the window, same as SQL). `orderBy`
+    /// (possibly empty) sorts the HAVING-narrowed groups server-side via
+    /// `aggregate.sortGroupsBy` before that window, the same
+    /// HAVING-then-ORDER-BY-then-LIMIT clause order SQL evaluates - each
+    /// `orderBy[i].field` must name either `groupField` or one of `specs`'
+    /// result labels (see `aggregate.specLabel`). One `GroupRow` per
+    /// distinct group value kept by HAVING and the window, in `orderBy`
+    /// order when non-empty, otherwise first-seen order (not sorted), same
+    /// as the local library call - free with `protocol.freeAggGroupRows`.
+    pub fn tableAggregateGroups(self: *Client, allocator: Allocator, table: []const u8, groupField: []const u8, specs: []const aggregatemod.AggSpec, offset: u32, limit: u32, orderBy: []const querymod.OrderBy, havingText: []const u8, exprText: []const u8) ![]aggregatemod.GroupRow {
+        const payload = try protocol.encodeTableAggregate(self.allocator, table, groupField, specs, offset, limit, orderBy, havingText, exprText);
+        defer self.allocator.free(payload);
+        var frame = try self.roundTrip(.table_aggregate, payload);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeAggGroupRows(allocator, frame.payload);
+    }
+
+    /// Remote `table window <table> <kind[:field]> [expr...] [partition
+    /// <field>[,field...]] [order <field[:desc]>[,field...]] [--limit N]
+    /// [--offset N]`: the wire counterpart of the CLI's local `table
+    /// window`, calling `window.compute` server-side over rows filtered by
+    /// `exprText` (same `expr.zig` grammar, empty selects every row).
+    /// Unlike `tableAggregateGroups`, this never collapses rows - every
+    /// matching row keeps its own `protocol.WindowRow` - but `offset`/`limit`
+    /// still window WHICH rows come back, walked in the same
+    /// partition-then-order display sequence `window.orderedIndices`
+    /// produces server-side (0/0 means unlimited, same convention as every
+    /// other paginated opcode). Returns one `WindowRow` per surviving row, in
+    /// that display order - free with `protocol.freeWindowRows`.
+    pub fn tableWindow(self: *Client, allocator: Allocator, table: []const u8, specs: []const windowmod.WindowSpec, partitionBy: []const []const u8, orderBy: []const querymod.OrderBy, offset: u32, limit: u32, exprText: []const u8, windowFrame: windowmod.FrameMode) ![]protocol.WindowRow {
+        const payload = try protocol.encodeTableWindow(self.allocator, table, specs, partitionBy, orderBy, offset, limit, exprText, windowFrame);
+        defer self.allocator.free(payload);
+        var frame = try self.roundTrip(.table_window, payload);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeWindowRows(allocator, frame.payload);
+    }
+
+    /// Remote `subscribe <channel>[,channel...]`: registers this connection
+    /// with the server's `pubsub.Hub` and blocks until the server's ack
+    /// frame comes back, so a `publish` issued on a different connection
+    /// right after this returns is guaranteed to reach this subscription -
+    /// see `server.zig`'s `handleSubscribe` doc comment for why the ack
+    /// exists at all. After this call, `self` must only ever be used with
+    /// `nextMessage` - the server never answers another request opcode on
+    /// this same connection again (a real, documented protocol
+    /// simplification, not an oversight: see `handleSubscribe`'s doc
+    /// comment on why a subscribed connection is receive-only). Returns the
+    /// `pubsub.Hub`-assigned subscriber id, the only handle a later
+    /// `unsubscribe` call - necessarily made through a DIFFERENT `Client`,
+    /// since this one can never send another request - can use to name this
+    /// subscription.
+    pub fn subscribe(self: *Client, channels: []const []const u8) !u64 {
+        const payload = try protocol.encodeSubscribe(self.allocator, channels);
+        defer self.allocator.free(payload);
+        var frame = try self.roundTrip(.subscribe, payload);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        const ack = try protocol.decodeSubscribeAck(frame.payload);
+        return ack.subscriberId;
+    }
+
+    /// Remote `unsubscribe <subscriber-id> <channel>[,channel...]`: the
+    /// client-side counterpart of `server.zig`'s `unsubscribe` opcode. Sent
+    /// over an ordinary connection (this one may be reused for other
+    /// requests afterward, unlike the subscribe connection it targets) to
+    /// drop a subset of `subscriberId`'s channels - or, if it names every
+    /// channel that subscriber still has left, to end it outright (see
+    /// `pubsub.Hub.unsubscribe`'s doc comment). Returns how many of
+    /// `channels` were actually removed.
+    pub fn unsubscribe(self: *Client, subscriberId: u64, channels: []const []const u8) !u32 {
+        const payload = try protocol.encodeUnsubscribe(self.allocator, subscriberId, channels);
+        defer self.allocator.free(payload);
+        var frame = try self.roundTrip(.unsubscribe, payload);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeUnsubscribeResult(frame.payload);
+    }
+
+    /// Reads the next pushed message on a connection that already called
+    /// `subscribe`, or `null` once the server closes the connection (the
+    /// same "no more frames coming" signal `roundTrip` treats as an error
+    /// everywhere else, but is the routine end of a subscription here).
+    /// Blocks indefinitely otherwise - there is no polling variant, exactly
+    /// like `Subscriber.waitMessage` server-side. Returns an independently
+    /// owned `protocol.PubSubMessage` - free with its own `deinit`.
+    pub fn nextMessage(self: *Client, allocator: Allocator) !?protocol.PubSubMessage {
+        var frame = protocol.readFrame(self.allocator, self.stream.reader()) catch |err| switch (err) {
+            error.EndOfStream => return null,
+            else => return err,
+        };
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return try protocol.decodePubSubMessage(allocator, frame.payload);
+    }
+