From 5d433ed37bae065f0ff43e5618c8a3efd65afa4d Mon Sep 17 00:00:00 2001 From: DualFroz Date: Mon, 7 Sep 2026 21:21:12 +0200 Subject: [PATCH] feat: add PubSubMessage and deinit to protocol --- src/protocol.zig | 159 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 159 insertions(+) diff --git a/src/protocol.zig b/src/protocol.zig index f85ed71..4596070 100644 --- a/src/protocol.zig +++ b/src/protocol.zig @@ -1877,3 +1877,162 @@ pub fn decodeUnsubscribeResult(payload: []const u8) !u32 { return c.takeU32(); } +/// One pushed message on an already-open `subscribe` connection: which +/// channel it arrived on (a client can subscribe to several at once, so the +/// tag travels with every message) plus the payload bytes. Every frame a +/// subscribed connection receives after its initial ack uses this shape, +/// always with status `.ok` - there is no separate framing marker, because +/// once a client has sent `subscribe` it never sends another request on +/// that connection and so never has to disambiguate a push from a reply to +/// something else it asked (see `server.zig`'s `handleSubscribe`). +pub const PubSubMessage = struct { + channel: []u8, + payload: []u8, + + pub fn deinit(self: PubSubMessage, allocator: Allocator) void { + allocator.free(self.channel); + allocator.free(self.payload); + } +}; + +pub fn encodePubSubMessage(allocator: Allocator, channel: []const u8, payload: []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(payload); + return buf.toOwnedSlice(); +} + +/// Decodes a pushed message into independently owned copies (the frame +/// they arrived in is freed right after decoding, in `Client.nextMessage` - +/// see its doc comment) - free with `PubSubMessage.deinit`. +pub fn decodePubSubMessage(allocator: Allocator, payload: []const u8) !PubSubMessage { + var c = Cursor{ .data = payload }; + const channel = try c.takeLenPrefixed(); + const message = c.data[c.pos..]; + const channelCopy = try allocator.dupe(u8, channel); + errdefer allocator.free(channelCopy); + const payloadCopy = try allocator.dupe(u8, message); + return .{ .channel = channelCopy, .payload = payloadCopy }; +} + +// --- table explain --- + +/// Wire request for a remote `table explain`: a table name and the raw text +/// of an `expr.zig` predicate expression (empty text means "no predicate", +/// the same convention `encodeTableSelect`'s `exprText` uses). No +/// offset/limit - explain reports a plan, not rows, so there is nothing to +/// page. +pub fn encodeTableExplain(allocator: Allocator, table: []const u8, exprText: []const u8) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try putLenPrefixed(w, table); + try w.slice(exprText); + return buf.toOwnedSlice(); +} + +pub const TableExplainArgs = struct { table: []const u8, exprText: []const u8 }; + +pub fn decodeTableExplain(payload: []const u8) !TableExplainArgs { + var c = Cursor{ .data = payload }; + const table = try c.takeLenPrefixed(); + const exprText = c.data[c.pos..]; + return .{ .table = table, .exprText = exprText }; +} + +/// Encodes a `query.Plan`. `indexName` uses the empty string as its "null" +/// sentinel (an index or field name can never itself be empty - `query.zig` +/// rejects that at `createIndex`/`createCompositeIndex` time) instead of a +/// separate presence byte, the same economy `exprText`'s "empty means no +/// predicate" convention already relies on elsewhere in this file. +pub fn encodePlan(allocator: Allocator, plan: querymod.Plan) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try w.byte(@intFromEnum(plan.kind)); + try putLenPrefixed(w, plan.indexName orelse ""); + try w.putU32(@intCast(plan.matchedFields)); + try w.putU32(@intCast(plan.totalFields)); + return buf.toOwnedSlice(); +} + +/// Decodes a `query.Plan` - `indexName`, when present, is an owned copy the +/// caller must free with `Plan.deinit` (see `encodePlan`'s doc comment for +/// why an empty string decodes back to `null` instead of `""`). +pub fn decodePlan(allocator: Allocator, payload: []const u8) !querymod.Plan { + var c = Cursor{ .data = payload }; + const kindByte = try c.takeByte(); + const kind = std.meta.intToEnum(querymod.PlanKind, kindByte) catch return error.Malformed; + const rawName = try c.takeLenPrefixed(); + const indexName = if (rawName.len == 0) null else try allocator.dupe(u8, rawName); + errdefer if (indexName) |n| allocator.free(n); + const matchedFields = try c.takeU32(); + const totalFields = try c.takeU32(); + return .{ .kind = kind, .indexName = indexName, .matchedFields = matchedFields, .totalFields = totalFields }; +} + +// --- table update --- + +/// Wire request for a remote `table update`: a table name, the `SET` +/// clause's field list, and the raw text of an `expr.zig` predicate +/// expression (empty text updates every row, the same convention +/// `encodeTableSelect`'s `exprText` uses). The field list reuses +/// `Row.encode` exactly like `encodeTxnOp`'s per-entry `fields` does, +/// instead of inventing a second name+value framing just for this opcode - +/// and, like `encodeTableSelect`'s `orderBy`, sits BEFORE `exprText` +/// because `exprText` is decoded as "everything left in the payload", +/// so anything meant to survive decoding has to come first. +pub fn encodeTableUpdate(allocator: Allocator, table: []const u8, sets: []const rowmod.Field, exprText: []const u8) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try putLenPrefixed(w, table); + + var setsRow = rowmod.Row.init(allocator); + defer setsRow.deinit(); + for (sets) |f| try setsRow.set(f.name, f.value); + const setsBytes = try setsRow.encode(allocator); + defer allocator.free(setsBytes); + try putLenPrefixed(w, setsBytes); + + try w.slice(exprText); + return buf.toOwnedSlice(); +} + +pub const TableUpdateArgs = struct { table: []const u8, setsBytes: []const u8, exprText: []const u8 }; + +/// `table` and `exprText` are borrowed slices into `payload`, the same +/// convention `decodeTableExplain` uses. `setsBytes` is also borrowed and +/// still needs `rowmod.Row.decode` to become a real field list - kept +/// undecoded here so a malformed table name fails fast without touching +/// row-decoding at all, the same ownership split `DecodedTxnOpEntry` uses +/// for its own `fieldsBytes`. +pub fn decodeTableUpdate(payload: []const u8) !TableUpdateArgs { + var c = Cursor{ .data = payload }; + const table = try c.takeLenPrefixed(); + const setsBytes = try c.takeLenPrefixed(); + const exprText = c.data[c.pos..]; + return .{ .table = table, .setsBytes = setsBytes, .exprText = exprText }; +} + +// --- table delete --- + +/// Wire request for a remote `table delete`: a table name and the raw text +/// of an `expr.zig` predicate expression (empty text deletes every row, the +/// same convention `encodeTableSelect`/`encodeTableExplain`'s `exprText` +/// uses). Identical wire shape to `encodeTableExplain` - both are just +/// "table + predicate", no `SET` clause like `encodeTableUpdate` needs - but +/// kept as its own named pair rather than reused directly, the same +/// per-opcode-gets-its-own-function precedent every other opcode here +/// follows even when two shapes coincide. +pub fn encodeTableDelete(allocator: Allocator, table: []const u8, exprText: []const u8) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try putLenPrefixed(w, table); + try w.slice(exprText); + return buf.toOwnedSlice(); +} +