diff --git a/src/client.zig b/src/client.zig index 56d836d..e54998b 100644 --- a/src/client.zig +++ b/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 * [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
[expr...] + /// [having ] [--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
[expr...] [partition + /// [,field...]] [order [,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...]`: 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 [,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); + } +