feat: extend main with parseChannelList and runRemotePublish

src/main.zig

@@ -2733,6 +2733,179 @@ fn runRemoteAggregate(allocator: std.mem.Allocator, stdout: anytype, stderr: any
     const specs = parseAggSpecs(allocator, rest[2]) catch return fail(stderr, "invalid aggregate spec (expected count | count:field | sum:field | avg:field | min:field | max:field | distinct:field, comma-separated)");
     defer allocator.free(specs);
 
+    const authFlags = extractAuthFlags(allocator, rest[3..]) 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 split = splitAggFilterArgs(flags.rest);
+
+    const exprText = try std.mem.join(allocator, " ", split.whereArgs);
+    defer allocator.free(exprText);
+
+    const havingText = if (split.havingArgs) |ha| try std.mem.join(allocator, " ", ha) else try allocator.dupe(u8, "");
+    defer allocator.free(havingText);
+    if (split.havingArgs != null and havingText.len == 0) return fail(stderr, "having requires an expression");
+
+    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");
+    }
+
+    if (std.mem.eql(u8, groupField, "*")) {
+        const values = client.tableAggregateGlobal(allocator, table, specs, havingText, exprText) catch |err| {
+            try stderr.print("error: {s}\n", .{@errorName(err)});
+            std.process.exit(1);
+        };
+        defer cobalt.protocol.freeAggValues(allocator, values);
+        // `values.len == 0` means the server's HAVING excluded the only
+        // global result (see `Client.tableAggregateGlobal`'s doc) - print
+        // nothing rather than indexing `specs` against a shorter `values`,
+        // which a naive `for (specs, values, ...)` would panic on.
+        if (values.len > 0) {
+            for (specs, values, 0..) |spec, v, i| {
+                if (i > 0) try stdout.writeAll("\t");
+                try printAggLabel(stdout, spec);
+                try stdout.writeAll("=");
+                try printValue(stdout, v);
+            }
+            try stdout.writeAll("\n");
+        }
+    } else {
+        const groups = client.tableAggregateGroups(allocator, table, groupField, specs, flags.page.offset, flags.page.limit, orderBy, havingText, exprText) catch |err| {
+            try stderr.print("error: {s}\n", .{@errorName(err)});
+            std.process.exit(1);
+        };
+        defer cobalt.protocol.freeAggGroupRows(allocator, groups);
+        for (groups) |g| {
+            try stdout.writeAll("group=");
+            try printValue(stdout, g.group);
+            for (specs, g.values) |spec, v| {
+                try stdout.writeAll("\t");
+                try printAggLabel(stdout, spec);
+                try stdout.writeAll("=");
+                try printValue(stdout, v);
+            }
+            try stdout.writeAll("\n");
+        }
+    }
+}
+
+/// Connects to a remote `cobalt serve` instance and runs `table window`
+/// there instead of against a local Db - the client-side counterpart of
+/// `server.zig`'s `table_window` opcode, mirroring `runRemoteAggregate`'s
+/// shape. Same comma-separated spec-list grammar as local `table window`
+/// (`parseWindowSpecs`), same `partition <field>[,field...]`/`order
+/// <field[:desc]>[,field...]` clause split (`splitWindowArgs`), and the same
+/// per-row output format as the local command (row fields, then
+/// `<label>=<value>`). Unlike `runRemoteAggregate`, there is no `*`
+/// sentinel or global/grouped branch - a window annotation never collapses
+/// rows, so there is exactly one request/response shape
+/// (`Client.tableWindow`). `--limit`/`--offset` window which rows come back
+/// (server-side, in `window.orderedIndices`' partition-then-order display
+/// sequence, not raw scan order - see `Client.tableWindow`'s doc).
+/// `--user`/`--pass`/`--token` authenticate first, needed only against a
+/// server started with `serve --auth`.
+/// Splits `text` on commas into channel names for `remote-subscribe` - no
+/// identifier validation, unlike `parsePartitionList`, since a pub/sub
+/// channel is an arbitrary string (`pubsub.zig`'s `Hub` keys purely on
+/// byte equality), not a table field name.
+fn parseChannelList(allocator: std.mem.Allocator, text: []const u8) ![][]const u8 {
+    var list = std.ArrayList([]const u8).init(allocator);
+    errdefer list.deinit();
+    var it = std.mem.splitScalar(u8, text, ',');
+    while (it.next()) |ch| {
+        if (ch.len == 0) return error.EmptyChannelName;
+        try list.append(ch);
+    }
+    return list.toOwnedSlice();
+}
+
+/// Connects to a running `cobalt serve` and delivers a message to every
+/// connection currently subscribed to `<channel>` - the client-side
+/// counterpart of `server.zig`'s `publish` opcode. `<message...>` is every
+/// remaining non-flag argument joined with single spaces, the same
+/// "trailing free text" convention `runRemoteWindow`'s `exprText` uses, so a
+/// multi-word message does not need manual quoting tricks beyond the
+/// shell's own.
+fn runRemotePublish(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void {
+    if (rest.len < 2) return fail(stderr, "remote-publish needs <channel> <message...> [--user NAME --pass PASS] [--token TOKEN]");
+    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 channel = rest[0];
+
+    const authFlags = extractAuthFlags(allocator, rest[1..]) catch return fail(stderr, "invalid --user/--pass/--token value");
+    defer allocator.free(authFlags.rest);
+
+    try authenticateIfRequested(&client, stderr, authFlags.auth);
+
+    if (authFlags.rest.len == 0) return fail(stderr, "remote-publish needs a message");
+    const message = try std.mem.join(allocator, " ", authFlags.rest);
+    defer allocator.free(message);
+
+    const count = client.publish(channel, message) catch |err| {
+        try stderr.print("error: {s}\n", .{@errorName(err)});
+        std.process.exit(1);
+    };
+    try stdout.print("{d}\n", .{count});
+}
+
+/// Connects to a running `cobalt serve`, subscribes to every channel in the
+/// comma-separated `<channel>[,channel...]` list, and blocks printing
+/// `<channel>: <message>` for each message that arrives until the
+/// connection ends - the client-side counterpart of `server.zig`'s
+/// `subscribe` opcode and `Client.nextMessage`'s doc comment on why this
+/// connection never sends another request afterward. Prints "subscribed as
+/// <id>" first (the id `remote-unsubscribe`, run from a different terminal
+/// against a different connection, needs to ever name this subscription
+/// again) - closing this process (or the connection) still ends the whole
+/// subscription outright, same as before `remote-unsubscribe` existed.
+fn runRemoteSubscribe(allocator: std.mem.Allocator, stdout: anytype, stderr: anytype, addressSpec: []const u8, rest: []const []const u8) !void {
+    if (rest.len < 1) return fail(stderr, "remote-subscribe needs <channel>[,channel...] [--user NAME --pass PASS] [--token TOKEN]");
+    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 channels = parseChannelList(allocator, rest[0]) catch return fail(stderr, "invalid channel list (empty channel name)");
+    defer allocator.free(channels);
+
+    const authFlags = extractAuthFlags(allocator, rest[1..]) catch return fail(stderr, "invalid --user/--pass/--token value");
+    defer allocator.free(authFlags.rest);
+
+    try authenticateIfRequested(&client, stderr, authFlags.auth);
+
+    const subscriberId = client.subscribe(channels) catch |err| {
+        try stderr.print("error: {s}\n", .{@errorName(err)});
+        std.process.exit(1);
+    };
+    try stdout.print("subscribed as {d}\n", .{subscriberId});
+
+    while (try client.nextMessage(allocator)) |msg| {
+        defer msg.deinit(allocator);
+        try stdout.print("{s}: {s}\n", .{ msg.channel, msg.payload });
+    }
+}
+
+/// Connects to a running `cobalt serve` and drops the listed channels from
+/// an already-subscribed connection's set, identified by the
+/// `<subscriber-id>` its `remote-subscribe` printed - the client-side
+/// counterpart of `server.zig`'s `unsubscribe` opcode. Unlike
+/// `remote-subscribe`, this connection is an ordinary request/response
 }
 
 fn runVerify(allocator: std.mem.Allocator, stdout: anytype, dir: []const u8, count: usize) !void {