feat: add publish and txnOp to client

src/client.zig

@@ -481,3 +481,223 @@ pub const Client = struct {
         return try protocol.decodePubSubMessage(allocator, frame.payload);
     }
 
+    /// Remote `publish <channel> <message>`: delivers `message` to every
+    /// subscriber currently registered on `channel` across every connection
+    /// the server has, returning how many subscribers it reached (0 is a
+    /// normal outcome for a channel nobody is listening to, not an error -
+    /// see `pubsub.Hub.publish`'s doc comment). Any connection may publish,
+    /// whether or not it has ever subscribed to anything itself.
+    pub fn publish(self: *Client, channel: []const u8, message: []const u8) !u64 {
+        const payload = try protocol.encodePublish(self.allocator, channel, message);
+        defer self.allocator.free(payload);
+        var frame = try self.roundTrip(.publish, payload);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodePublishResult(frame.payload);
+    }
+
+    /// Remote `txn <script>`: applies every entry in `entries` as one atomic
+    /// `Transaction` commit across however many tables they touch - the wire
+    /// counterpart of the CLI's local `txn <script-file>` (main.zig's
+    /// `runTxn`). Returns the number of operations committed, always
+    /// `entries.len` on success since a partial commit is impossible; a
+    /// conflicting transaction surfaces as `error.ServerConflict` with
+    /// nothing applied, matching `error.Conflict` from a local `Transaction.
+    /// commit()`.
+    pub fn txnOp(self: *Client, entries: []const protocol.TxnOpEntry) !u32 {
+        const payload = try protocol.encodeTxnOp(self.allocator, entries);
+        defer self.allocator.free(payload);
+        var frame = try self.roundTrip(.txn_op, payload);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeTxnResult(frame.payload);
+    }
+
+    fn userAdminRoundTrip(self: *Client, action: protocol.UserAdminAction, username: []const u8, password: []const u8, role: authmod.Role, rounds: u32) !protocol.Frame {
+        const payload = try protocol.encodeUserAdmin(self.allocator, action, username, password, role, rounds);
+        defer self.allocator.free(payload);
+        return self.roundTrip(.user_admin, payload);
+    }
+
+    /// The wire counterpart of `auth.Store.createUser`, gated server-side
+    /// behind `admin` role whenever the server has auth enabled (see
+    /// `server.zig`'s `requiredRole`). `rounds` of 0 asks the server to use
+    /// its own `auth.recommendedRounds` default.
+    pub fn adminCreateUser(self: *Client, username: []const u8, password: []const u8, role: authmod.Role, rounds: u32) !void {
+        var frame = try self.userAdminRoundTrip(.create, username, password, role, rounds);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+    }
+
+    /// The wire counterpart of `auth.Store.setPassword`. Reports
+    /// `error.ServerUserNotFound` (not the generic `UnexpectedStatus`) when
+    /// the account does not exist, since that is a normal, expected outcome
+    /// worth distinguishing from every other non-ok status.
+    pub fn adminSetPassword(self: *Client, username: []const u8, password: []const u8, rounds: u32) !void {
+        var frame = try self.userAdminRoundTrip(.set_password, username, password, .readonly, rounds);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status == .not_found) return error.ServerUserNotFound;
+        if (status != .ok) return statusToError(status);
+    }
+
+    /// The wire counterpart of `auth.Store.setRole`.
+    pub fn adminSetRole(self: *Client, username: []const u8, role: authmod.Role) !void {
+        var frame = try self.userAdminRoundTrip(.set_role, username, "", role, 0);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status == .not_found) return error.ServerUserNotFound;
+        if (status != .ok) return statusToError(status);
+    }
+
+    /// The wire counterpart of `auth.Store.deleteUser`.
+    pub fn adminDeleteUser(self: *Client, username: []const u8) !void {
+        var frame = try self.userAdminRoundTrip(.delete, username, "", .readonly, 0);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status == .not_found) return error.ServerUserNotFound;
+        if (status != .ok) return statusToError(status);
+    }
+
+    /// The wire counterpart of `auth.Store.listUsers`. Returned entries are
+    /// independently owned - free with `authmod.freeUserList`.
+    pub fn adminListUsers(self: *Client, allocator: Allocator) ![]authmod.UserInfo {
+        var frame = try self.userAdminRoundTrip(.list, "", "", .readonly, 0);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeUserList(allocator, frame.payload);
+    }
+
+    fn ttlOpRoundTrip(self: *Client, action: protocol.TtlAction, key: []const u8, seconds: u64) !protocol.Frame {
+        const payload = try protocol.encodeTtlOp(self.allocator, action, key, seconds);
+        defer self.allocator.free(payload);
+        return self.roundTrip(.ttl_op, payload);
+    }
+
+    /// The wire counterpart of `ttl.setExpireSeconds`, gated server-side the
+    /// same way the CLI's local `expire` is: `error.ServerKeyNotFound` (not
+    /// the generic `UnexpectedStatus`) when `key` does not currently exist,
+    /// since that is a normal, expected outcome worth distinguishing from
+    /// every other non-ok status - matching `adminSetPassword`'s treatment
+    /// of an unknown username.
+    pub fn ttlExpire(self: *Client, key: []const u8, seconds: u64) !void {
+        var frame = try self.ttlOpRoundTrip(.expire, key, seconds);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status == .not_found) return error.ServerKeyNotFound;
+        if (status != .ok) return statusToError(status);
+    }
+
+    /// The wire counterpart of `ttl.clearExpire`.
+    pub fn ttlPersist(self: *Client, key: []const u8) !void {
+        var frame = try self.ttlOpRoundTrip(.persist, key, 0);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+    }
+
+    /// The wire counterpart of `ttl.remaining`.
+    pub fn ttlQuery(self: *Client, key: []const u8) !ttlmod.TtlResult {
+        var frame = try self.ttlOpRoundTrip(.query, key, 0);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeTtlResult(frame.payload);
+    }
+
+    /// The wire counterpart of `ttl.sweepExpired`: reaps every due key on
+    /// the server and reports how many were removed.
+    pub fn ttlSweep(self: *Client) !u64 {
+        var frame = try self.ttlOpRoundTrip(.sweep, "", 0);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeTtlSweepCount(frame.payload);
+    }
+
+    fn collTtlOpRoundTrip(self: *Client, action: protocol.TtlAction, kind: collmod.Kind, name: []const u8, seconds: u64) !protocol.Frame {
+        const payload = try protocol.encodeCollTtlOp(self.allocator, action, kind, name, seconds);
+        defer self.allocator.free(payload);
+        return self.roundTrip(.coll_ttl_op, payload);
+    }
+
+    /// The wire counterpart of `collttl.setExpireSeconds`, gated server-side
+    /// the same way the CLI's local `coll <kind> expire` is:
+    /// `error.ServerCollectionNotFound` (not the generic `UnexpectedStatus`)
+    /// when the collection does not currently exist, matching `ttlExpire`'s
+    /// treatment of an unknown flat key.
+    pub fn collTtlExpire(self: *Client, kind: collmod.Kind, name: []const u8, seconds: u64) !void {
+        var frame = try self.collTtlOpRoundTrip(.expire, kind, name, seconds);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status == .not_found) return error.ServerCollectionNotFound;
+        if (status != .ok) return statusToError(status);
+    }
+
+    /// The wire counterpart of `collttl.clearExpire`.
+    pub fn collTtlPersist(self: *Client, kind: collmod.Kind, name: []const u8) !void {
+        var frame = try self.collTtlOpRoundTrip(.persist, kind, name, 0);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+    }
+
+    /// The wire counterpart of `collttl.remaining`.
+    pub fn collTtlQuery(self: *Client, kind: collmod.Kind, name: []const u8) !ttlmod.TtlResult {
+        var frame = try self.collTtlOpRoundTrip(.query, kind, name, 0);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeTtlResult(frame.payload);
+    }
+
+    /// The wire counterpart of `collttl.sweepExpired`: reaps every due
+    /// collection of every kind on the server and reports how many were
+    /// removed. The `kind` argument to `collTtlOpRoundTrip` is a
+    /// placeholder here (see `protocol.encodeCollTtlOp`'s doc comment) -
+    /// `.counter` is chosen arbitrarily, same convention as `ttlSweep`
+    /// passing an empty key.
+    pub fn collTtlSweep(self: *Client) !u64 {
+        var frame = try self.collTtlOpRoundTrip(.sweep, .counter, "", 0);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeTtlSweepCount(frame.payload);
+    }
+
+    /// The wire counterpart of `main.zig`'s local `coll <kind> <op> ...`
+    /// (backed by `collections.zig`) - one generic round trip for every
+    /// `protocol.CollAction`, mirroring that opcode's own "one opcode, many
+    /// actions" shape rather than exposing thirty near-identical wrapper
+    /// methods here. `args` are the same CLI-text-form string arguments
+    /// (values/members/indices/scores) `runColl` already parses locally, so
+    /// the server decodes exactly the same numbers with exactly the same
+    /// `std.fmt.parseInt`/`parseFloat` calls the local command uses. Unlike
+    /// `collTtlExpire`, a missing hash field/list element/zset member is not
+    /// an error here - it comes back as `OwnedCollReply.nil`, exactly like
+    /// the local CLI prints `(nil)` rather than failing. Returned reply is
+    /// independently owned; free with `reply.deinit(allocator)`.
+    pub fn collOp(self: *Client, allocator: Allocator, action: protocol.CollAction, name: []const u8, args: []const []const u8) !protocol.OwnedCollReply {
+        const payload = try protocol.encodeCollOp(self.allocator, action, name, args);
+        defer self.allocator.free(payload);
+        var frame = try self.roundTrip(.coll_op, payload);
+        defer frame.deinit();
+        const status = try protocol.Status.fromByte(frame.tag);
+        if (status != .ok) return statusToError(status);
+        return protocol.decodeCollReply(allocator, frame.payload);
+    }
+};
+
+test "statusToError maps every non-ok status to a distinct error" {
+    try std.testing.expectEqual(error.ServerBadRequest, statusToError(.bad_request));
+    try std.testing.expectEqual(error.ServerInternal, statusToError(.internal));
+    try std.testing.expectEqual(error.ServerConflict, statusToError(.conflict));
+    try std.testing.expectEqual(error.ServerRateLimited, statusToError(.rate_limited));
+    try std.testing.expectEqual(error.ServerUnauthorized, statusToError(.unauthorized));
+    try std.testing.expectEqual(error.ServerForbidden, statusToError(.forbidden));
+    try std.testing.expectEqual(error.UnexpectedStatus, statusToError(.not_found));
+}