feat: add handleUserAdmin and stop to server

src/server.zig

@@ -1851,3 +1851,163 @@ pub const Server = struct {
         try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), out);
     }
 
+    /// Executes a `user_admin` request (create/set_password/set_role/delete/
+    /// list) against a fresh `authmod.Store` over the same `*Db` the rest of
+    /// the server serves. Deliberately does NOT consult `self.authStore` -
+    /// that field only controls whether *login enforcement* is turned on
+    /// for this server, not whether the user table exists, so remote user
+    /// management works the same way whether or not `--auth` is active.
+    /// `requiredRole` already gates this opcode behind `admin` whenever
+    /// auth enforcement IS enabled; a server with auth disabled enforces
+    /// nothing on any opcode, this one included, matching every other op's
+    /// documented behavior.
+    fn handleUserAdmin(self: *Server, payload: []const u8, writer: anytype) !void {
+        const args = protocol.decodeUserAdmin(payload) catch {
+            return writeErr(writer, .bad_request, "malformed user_admin payload");
+        };
+        const store = authmod.Store.init(self.db);
+
+        switch (args.action) {
+            .create => {
+                const role = authmod.Role.fromByte(args.roleByte) catch {
+                    return writeErr(writer, .bad_request, "invalid role");
+                };
+                const rounds = if (args.rounds != 0) args.rounds else authmod.recommendedRounds;
+                store.createUser(self.allocator, args.username, args.password, role, rounds) catch |err| {
+                    return switch (err) {
+                        error.UserExists => writeErr(writer, .conflict, "user already exists"),
+                        else => writeErr(writer, .internal, @errorName(err)),
+                    };
+                };
+                try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), "");
+            },
+
+            .set_password => {
+                const rounds = if (args.rounds != 0) args.rounds else authmod.recommendedRounds;
+                store.setPassword(self.allocator, args.username, args.password, rounds) catch |err| {
+                    return switch (err) {
+                        error.UserNotFound => writeErr(writer, .not_found, "user not found"),
+                        else => writeErr(writer, .internal, @errorName(err)),
+                    };
+                };
+                try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), "");
+            },
+
+            .set_role => {
+                const role = authmod.Role.fromByte(args.roleByte) catch {
+                    return writeErr(writer, .bad_request, "invalid role");
+                };
+                store.setRole(self.allocator, args.username, role) catch |err| {
+                    return switch (err) {
+                        error.UserNotFound => writeErr(writer, .not_found, "user not found"),
+                        else => writeErr(writer, .internal, @errorName(err)),
+                    };
+                };
+                try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), "");
+            },
+
+            .delete => {
+                store.deleteUser(self.allocator, args.username) catch |err| {
+                    return switch (err) {
+                        error.UserNotFound => writeErr(writer, .not_found, "user not found"),
+                        else => writeErr(writer, .internal, @errorName(err)),
+                    };
+                };
+                try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), "");
+            },
+
+            .list => {
+                const users = store.listUsers(self.allocator) catch |err| {
+                    return writeErr(writer, .internal, @errorName(err));
+                };
+                defer authmod.freeUserList(self.allocator, users);
+                const out = protocol.encodeUserList(self.allocator, users) catch |err| {
+                    return writeErr(writer, .internal, @errorName(err));
+                };
+                defer self.allocator.free(out);
+                try protocol.writeFrame(writer, @intFromEnum(protocol.Status.ok), out);
+            },
+        }
+    }
+
+    /// Stops accepting new connections and waits for the accept loop to
+    /// exit. Closing the listening socket while another thread is
+    /// blocked inside `accept()` on that same fd is a race the standard
+    /// library treats as a bug (it asserts on EBADF), so instead we flip
+    /// `stopping` and connect a throwaway client to our own listen
+    /// address: that wakes the pending `accept()` with a real, briefly
+    /// alive connection, which the loop then discards and exits on. Only
+    /// once the accept thread has rejoined is it safe to close the
+    /// listening socket itself.
+    ///
+    /// Already-accepted connections are `.detach()`ed worker threads this
+    /// struct holds no `std.Thread` handle for, so they cannot be `join`ed
+    /// directly - `drainConnections` below waits on the `active` counter
+    /// instead. This used to be entirely the caller's problem ("leave them
+    /// to finish on their own, call `activeConnections()` first if you need
+    /// to wait"), but every real caller in this codebase - every test in
+    /// this file included - called `stop()` and then immediately tore down
+    /// the `Db`/allocator the detached worker's `dispatch()` still reads
+    /// from, with nothing enforcing that the worker had actually noticed
+    /// its connection closed and returned first. Most of the time the
+    /// worker's blocking read already sees EOF and exits before the caller
+    /// gets around to teardown, so the race stayed invisible for a long
+    /// time; occasionally it does not, and a worker thread touches a freed
+    /// `Db`/`Allocator.VTable`, corrupting the heap in a way that only
+    /// surfaces later as an unrelated test's leak/double-free abort - this
+    /// is exactly the flaky `zig build test` failure the gate re-run in
+    /// this session caught (roughly 1 run in 4-6, always attributed to
+    /// whichever unrelated test happened to run next). Waiting here by
+    /// default, bounded so a genuinely long-lived idle connection (still
+    /// blocked in `stream.reader()` with nothing sent and no EOF) cannot
+    /// hang shutdown forever, closes that gap for every caller instead of
+    /// relying on each one to remember `activeConnections()`.
+    pub fn stop(self: *Server) void {
+        if (self.acceptThread == null) return;
+        self.stopping.store(true, .release);
+        const listenAddr = self.net.listen_address;
+        if (std.net.tcpConnectToAddress(listenAddr)) |s| {
+            s.close();
+        } else |_| {}
+        self.acceptThread.?.join();
+        self.acceptThread = null;
+        if (self.sweeperThread) |t| {
+            t.join();
+            self.sweeperThread = null;
+        }
+        if (self.ttlSweeperThread) |t| {
+            t.join();
+            self.ttlSweeperThread = null;
+        }
+        if (self.collTtlSweeperThread) |t| {
+            t.join();
+            self.collTtlSweeperThread = null;
+        }
+        self.drainConnections(500 * std.time.ns_per_ms);
+        self.net.deinit();
+        // Best-effort, same as `drainConnections` above: a subscriber
+        // connection that never disconnects (client vanished without
+        // closing its socket) is still `active` past the drain timeout, so
+        // its `Subscriber` is never `destroySubscriber`'d and leaks for the
+        // remaining life of the process - this `deinit` only frees `Hub`'s
+        // own channel map, which by then holds no dangling entries for any
+        // subscriber that WAS properly torn down.
+        self.pubsub.deinit();
+    }
+
+    /// Polls `activeConnections()` until it reaches zero or `timeoutNanos`
+    /// elapses, whichever comes first. See the doc comment on `stop()` for
+    /// why this exists and why it is bounded rather than unconditional.
+    fn drainConnections(self: *Server, timeoutNanos: u64) void {
+        const startedAt = std.time.nanoTimestamp();
+        while (self.activeConnections() > 0) {
+            if (std.time.nanoTimestamp() - startedAt >= timeoutNanos) return;
+            std.time.sleep(1 * std.time.ns_per_ms);
+        }
+    }
+};
+
+fn testDb(allocator: Allocator, dir: []const u8) !Db {
+    return Db.open(allocator, dir, .{});
+}
+