From 422f7c16397eaa214bdc5ff89b1c3eff93878373 Mon Sep 17 00:00:00 2001 From: DualFroz Date: Mon, 7 Sep 2026 16:04:41 +0200 Subject: [PATCH] feat: extend main with runRemoteUnsubscribe and runRemoteWindow --- src/main.zig | 173 +++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 173 insertions(+) diff --git a/src/main.zig b/src/main.zig index be4a782..4a24cbf 100644 --- a/src/main.zig +++ b/src/main.zig @@ -2906,6 +2906,179 @@ fn runRemoteSubscribe(allocator: std.mem.Allocator, stdout: anytype, stderr: any /// `` its `remote-subscribe` printed - the client-side /// counterpart of `server.zig`'s `unsubscribe` opcode. Unlike /// `remote-subscribe`, this connection is an ordinary request/response +/// round trip: it prints how many channels were actually removed and +/// exits, never blocking. `` must parse as a `u64` - a +/// malformed one is reported the same way a malformed `--limit` is +/// elsewhere in this file, before ever reaching for the network. +fn runRemoteUnsubscribe(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void { + if (rest.len < 2) return fail(stderr, "remote-unsubscribe needs [,channel...] [--user NAME --pass PASS] [--token TOKEN]"); + const address = try parseHostPort(stdout, addressSpec); + + const subscriberId = std.fmt.parseInt(u64, rest[0], 10) catch return fail(stderr, "invalid (expected a non-negative integer)"); + + var client = cobalt.Client.connect(allocator, address) catch |err| { + try stderr.print("error: could not connect to {s}: {s}\n", .{ addressSpec, @errorName(err) }); + std.process.exit(1); + }; + defer client.close(); + + const channels = parseChannelList(allocator, rest[1]) catch return fail(stderr, "invalid channel list (empty channel name)"); + defer allocator.free(channels); + + const authFlags = extractAuthFlags(allocator, rest[2..]) catch return fail(stderr, "invalid --user/--pass/--token value"); + defer allocator.free(authFlags.rest); + + try authenticateIfRequested(&client, stderr, authFlags.auth); + + const removed = client.unsubscribe(subscriberId, channels) catch |err| { + try stderr.print("error: {s}\n", .{@errorName(err)}); + std.process.exit(1); + }; + try stdout.print("{d}\n", .{removed}); +} + +fn runRemoteWindow(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void { + if (rest.len < 2) return fail(stderr, "remote-window needs [,...] [expr...] [partition [,field...]] [order [,field...]] [--limit N] [--offset N] [--range]"); + const address = try parseHostPort(stdout, addressSpec); + + var client = cobalt.Client.connect(allocator, address) catch |err| { + try stderr.print("error: could not connect to {s}: {s}\n", .{ addressSpec, @errorName(err) }); + std.process.exit(1); + }; + defer client.close(); + + const table = rest[0]; + const specs = parseWindowSpecs(allocator, rest[1]) catch return fail(stderr, "invalid window spec (expected row_number | rank | dense_rank | sum:field | avg:field | count | count:field | count_distinct:field | min:field | max:field | first_value:field | last_value:field | nth_value:field:n[:default] | lag:field[:offset[:default]] | lead:field[:offset[:default]], comma-separated)"); + defer allocator.free(specs); + + const authFlags = extractAuthFlags(allocator, rest[2..]) catch return fail(stderr, "invalid --user/--pass/--token value"); + defer allocator.free(authFlags.rest); + + try authenticateIfRequested(&client, stderr, authFlags.auth); + + const flags = extractPageFlags(allocator, authFlags.rest) catch return fail(stderr, "invalid --limit/--offset value"); + defer allocator.free(flags.rest); + + const rangeFlag = try extractRangeFlag(allocator, flags.rest); + defer allocator.free(rangeFlag.rest); + const frame: cobalt.window.FrameMode = if (rangeFlag.range) .range else .rows; + + const split = splitWindowArgs(rangeFlag.rest); + + const exprText = try std.mem.join(allocator, " ", split.whereArgs); + defer allocator.free(exprText); + + var partitionBy: [][]const u8 = &.{}; + defer allocator.free(partitionBy); + if (split.partitionArgs) |partitionArgs| { + if (partitionArgs.len != 1) return fail(stderr, "partition requires exactly one comma-separated field list"); + partitionBy = parsePartitionList(allocator, partitionArgs[0]) catch return fail(stderr, "invalid partition field list"); + } + + var orderBy: []cobalt.query.OrderBy = &.{}; + defer allocator.free(orderBy); + if (split.orderArgs) |orderArgs| { + if (orderArgs.len != 1) return fail(stderr, "order requires exactly one comma-separated field list"); + orderBy = parseOrderByList(allocator, orderArgs[0]) catch return fail(stderr, "invalid order field list"); + } + + const results = client.tableWindow(allocator, table, specs, partitionBy, orderBy, flags.page.offset, flags.page.limit, exprText, frame) catch |err| { + try stderr.print("error: {s}\n", .{@errorName(err)}); + std.process.exit(1); + }; + defer cobalt.protocol.freeWindowRows(allocator, results); + + for (results) |wr| { + for (wr.row.fields.items, 0..) |f, j| { + if (j > 0) try stdout.writeAll("\t"); + try stdout.print("{s}=", .{f.name}); + try printValue(stdout, f.value); + } + if (wr.row.fields.items.len > 0) try stdout.writeAll("\t"); + for (specs, 0..) |spec, specIdx| { + if (specIdx > 0) try stdout.writeAll("\t"); + try printWindowLabel(stdout, spec); + try stdout.writeAll("="); + try printValue(stdout, wr.values[specIdx]); + } + try stdout.writeAll("\n"); + } +} + +/// The network counterpart of `runTxn`: parses the exact same +/// `put|del
[field=value ...]` script format as the local +/// `txn ` command, but builds a list of +/// `protocol.TxnOpEntry` and commits it in one round trip through +/// `client.zig`'s `txnOp` (the `txn_op` opcode) instead of staging against +/// a local `TableTxn` directly - so a script that used to require a +/// process holding the `Db` can now run against a remote `cobalt serve`. +fn runRemoteTxn(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void { + if (rest.len < 1) return fail(stderr, "remote-txn needs [--user NAME --pass PASS] [--token TOKEN]"); + const address = try parseHostPort(stdout, addressSpec); + + const authFlags = extractAuthFlags(allocator, rest[1..]) catch return fail(stderr, "invalid --user/--pass/--token value"); + defer allocator.free(authFlags.rest); + + const contents = std.fs.cwd().readFileAlloc(allocator, rest[0], 16 * 1024 * 1024) catch |err| { + try stderr.print("failed to read script '{s}': {s}\n", .{ rest[0], @errorName(err) }); + std.process.exit(1); + }; + defer allocator.free(contents); + + var entries = std.ArrayList(cobalt.protocol.TxnOpEntry).init(allocator); + defer { + for (entries.items) |e| allocator.free(e.fields); + entries.deinit(); + } + + var lineNo: usize = 0; + var lines = std.mem.splitScalar(u8, contents, '\n'); + while (lines.next()) |rawLine| { + lineNo += 1; + const line = std.mem.trim(u8, rawLine, " \t\r"); + if (line.len == 0 or line[0] == '#') continue; + + var tok = std.mem.tokenizeScalar(u8, line, ' '); + const verb = tok.next() orelse continue; + const tableName = tok.next() orelse { + try stderr.print("line {d}: missing table name\n", .{lineNo}); + std.process.exit(2); + }; + const pkText = tok.next() orelse { + try stderr.print("line {d}: missing primary key\n", .{lineNo}); + std.process.exit(2); + }; + const pk = parseValue(pkText) catch { + try stderr.print("line {d}: invalid primary key '{s}'\n", .{ lineNo, pkText }); + std.process.exit(2); + }; + + if (std.mem.eql(u8, verb, "put")) { + var fields = std.ArrayList(Field).init(allocator); + errdefer fields.deinit(); + while (tok.next()) |raw| { + const f = parseField(raw) catch { + try stderr.print("line {d}: invalid field '{s}'\n", .{ lineNo, raw }); + std.process.exit(2); + }; + try fields.append(f); + } + try entries.append(.{ .verb = .put, .table = tableName, .pk = pk, .fields = try fields.toOwnedSlice() }); + } else if (std.mem.eql(u8, verb, "del")) { + try entries.append(.{ .verb = .del, .table = tableName, .pk = pk, .fields = &.{} }); + } else { + try stderr.print("line {d}: unknown verb '{s}' (expected put/del)\n", .{ lineNo, verb }); + std.process.exit(2); + } + } + + if (entries.items.len == 0) return fail(stderr, "txn script had no operations"); + + var client = cobalt.Client.connect(allocator, address) catch |err| { + try stderr.print("error: could not connect to {s}: {s}\n", .{ addressSpec, @errorName(err) }); + std.process.exit(1); + }; + defer client.close(); } fn runVerify(allocator: std.mem.Allocator, stdout: anytype, dir: []const u8, count: usize) !void {