From aad2726772fbd0819690b7e971143fdb15bc10ed Mon Sep 17 00:00:00 2001 From: DualFroz Date: Mon, 7 Sep 2026 20:03:48 +0200 Subject: [PATCH] feat: add decodeTxnOp and encodeTxnResult to protocol --- src/protocol.zig | 152 +++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 152 insertions(+) diff --git a/src/protocol.zig b/src/protocol.zig index 1eabdbd..79e33b0 100644 --- a/src/protocol.zig +++ b/src/protocol.zig @@ -801,3 +801,155 @@ pub const DecodedTxnOpEntry = struct { fieldsBytes: []const u8, }; +pub fn decodeTxnOp(allocator: Allocator, payload: []const u8) ![]DecodedTxnOpEntry { + var c = Cursor{ .data = payload }; + const count = try c.takeU32(); + var list = try std.ArrayList(DecodedTxnOpEntry).initCapacity(allocator, count); + errdefer list.deinit(); + var i: u32 = 0; + while (i < count) : (i += 1) { + const verb = try TxnVerb.fromByte(try c.takeByte()); + const table = try c.takeLenPrefixed(); + const pkBytes = try c.takeLenPrefixed(); + const fieldsBytes = try c.takeLenPrefixed(); + list.appendAssumeCapacity(.{ .verb = verb, .table = table, .pkBytes = pkBytes, .fieldsBytes = fieldsBytes }); + } + return list.toOwnedSlice(); +} + +/// Encodes the response to a successful `txn_op`: how many operations were +/// committed - always every entry in the request, since a partial commit is +/// impossible (`error.Conflict` aborts the whole transaction instead, and +/// surfaces as `Status.conflict` with no payload, the same shape as +/// `user_admin`'s `UserExists` conflict). +pub fn encodeTxnResult(allocator: Allocator, applied: u32) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try w.putU32(applied); + return buf.toOwnedSlice(); +} + +pub fn decodeTxnResult(payload: []const u8) !u32 { + var c = Cursor{ .data = payload }; + return c.takeU32(); +} + +// --- scan / prefix --- + +pub const ScanArgs = struct { + start: ?[]const u8 = null, + end: ?[]const u8 = null, + limit: u32 = 0, + reverse: bool = false, +}; + +const scanFlagHasStart: u8 = 1 << 0; +const scanFlagHasEnd: u8 = 1 << 1; +const scanFlagReverse: u8 = 1 << 2; + +pub fn encodeScan(allocator: Allocator, args: ScanArgs) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + var flags: u8 = 0; + if (args.start != null) flags |= scanFlagHasStart; + if (args.end != null) flags |= scanFlagHasEnd; + if (args.reverse) flags |= scanFlagReverse; + try w.byte(flags); + if (args.start) |s| try putLenPrefixed(w, s); + if (args.end) |e| try putLenPrefixed(w, e); + try w.putU32(args.limit); + return buf.toOwnedSlice(); +} + +pub fn decodeScan(payload: []const u8) !ScanArgs { + var c = Cursor{ .data = payload }; + const flags = try c.takeByte(); + var args = ScanArgs{}; + if (flags & scanFlagHasStart != 0) args.start = try c.takeLenPrefixed(); + if (flags & scanFlagHasEnd != 0) args.end = try c.takeLenPrefixed(); + args.reverse = flags & scanFlagReverse != 0; + args.limit = try c.takeU32(); + return args; +} + +pub fn encodePrefix(allocator: Allocator, prefix: []const u8, limit: u32) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try w.putU32(limit); + try w.slice(prefix); + return buf.toOwnedSlice(); +} + +pub const PrefixArgs = struct { prefix: []const u8, limit: u32 }; + +pub fn decodePrefix(payload: []const u8) !PrefixArgs { + var c = Cursor{ .data = payload }; + const limit = try c.takeU32(); + const prefix = c.data[c.pos..]; + return .{ .prefix = prefix, .limit = limit }; +} + +// --- entry lists (scan / prefix responses) --- + +pub fn encodeEntries(allocator: Allocator, entries: []const dbmod.Entry) ![]u8 { + var buf = std.ArrayList(u8).init(allocator); + errdefer buf.deinit(); + const w = encode.ByteWriter.init(&buf); + try w.putU32(@intCast(entries.len)); + for (entries) |e| { + try putLenPrefixed(w, e.key); + try putLenPrefixed(w, e.value); + } + return buf.toOwnedSlice(); +} + +pub const OwnedEntry = struct { + key: []u8, + value: []u8, + + pub fn deinit(self: *OwnedEntry, allocator: Allocator) void { + allocator.free(self.key); + allocator.free(self.value); + } +}; + +/// Copies every key and value out of `payload` so the caller can free +/// the frame that carried them and keep the entries alive independently. +pub fn decodeEntries(allocator: Allocator, payload: []const u8) ![]OwnedEntry { + var c = Cursor{ .data = payload }; + const count = try c.takeU32(); + var list = try std.ArrayList(OwnedEntry).initCapacity(allocator, count); + errdefer { + for (list.items) |*e| e.deinit(allocator); + list.deinit(); + } + var i: u32 = 0; + while (i < count) : (i += 1) { + const key = try c.takeLenPrefixed(); + const value = try c.takeLenPrefixed(); + const ownedKey = try allocator.dupe(u8, key); + errdefer allocator.free(ownedKey); + const ownedValue = try allocator.dupe(u8, value); + list.appendAssumeCapacity(.{ .key = ownedKey, .value = ownedValue }); + } + return list.toOwnedSlice(); +} + +pub fn freeOwnedEntries(allocator: Allocator, entries: []OwnedEntry) void { + for (entries) |*e| e.deinit(allocator); + allocator.free(entries); +} + +// --- batch --- + +pub const BatchOp = union(enum) { + put: struct { key: []const u8, value: []const u8 }, + del: []const u8, +}; + +const batchKindPut: u8 = 0; +const batchKindDel: u8 = 1; +