feat: build out server

src/server.zig

@@ -6282,3 +6282,161 @@ test "a server with no ttl sweep interval never runs the ttl sweeper" {
     var client = try client_mod.Client.connect(allocator, server.boundAddress());
     defer client.close();
 
+    try client.set("a", "1");
+    try client.ttlExpire("a", 0);
+    std.time.sleep(100 * std.time.ns_per_ms);
+
+    // Nothing swept it in the background, but the key IS logically due -
+    // client.get goes straight through db.get (not the lazy ttl path), so
+    // the underlying value is still physically present until something
+    // sweeps it.
+    try std.testing.expectEqual(@as(u64, 0), server.ttlSweptTotal.get());
+    const v = (try client.get(allocator, "a")).?;
+    allocator.free(v);
+}
+
+test "coll_ttl_op over the wire: expire, query, persist, and sweep round trip against a real server" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const path = try tmp.dir.realpathAlloc(allocator, ".");
+    defer allocator.free(path);
+
+    var db = try testDb(allocator, path);
+    defer db.close();
+
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    try server.start();
+    defer server.stop();
+
+    const client_mod = @import("client.zig");
+    var client = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer client.close();
+
+    // A hash with no ttl at all reads back .noExpiry, not .noKey - the
+    // collection itself has to actually exist server-side first, created
+    // directly through `db` here for brevity (the `coll_op` opcode below
+    // is this same test file's own coverage for reaching it remotely).
+    _ = try collmod.Hash.set(&db, allocator, "profile", "name", "ada");
+    try std.testing.expectEqual(ttlmod.TtlResult.noExpiry, try client.collTtlQuery(.hash, "profile"));
+
+    // expire on a collection that does not exist is a real, distinct failure.
+    try std.testing.expectError(client_mod.ClientError.ServerCollectionNotFound, client.collTtlExpire(.hash, "ghost", 60));
+
+    // expire on an existing collection sets a marker the query action can
+    // see. Real wall-clock time can shave a second off by the time the
+    // second round trip lands, so accept 59 or 60 rather than an exact value.
+    try client.collTtlExpire(.hash, "profile", 60);
+    const withTtl = try client.collTtlQuery(.hash, "profile");
+    try std.testing.expect(withTtl.seconds == 59 or withTtl.seconds == 60);
+
+    // persist removes the marker again, back to .noExpiry.
+    try client.collTtlPersist(.hash, "profile");
+    try std.testing.expectEqual(ttlmod.TtlResult.noExpiry, try client.collTtlQuery(.hash, "profile"));
+
+    // sweep over a collection that is already due for expiry reaps exactly
+    // it, and the whole hash (not just one field) is really gone afterwards.
+    _ = try collmod.Hash.set(&db, allocator, "doomed", "k", "v");
+    try client.collTtlExpire(.hash, "doomed", 0);
+    std.time.sleep(5 * std.time.ns_per_ms);
+    const reaped = try client.collTtlSweep();
+    try std.testing.expectEqual(@as(u64, 1), reaped);
+    try std.testing.expect(!try collmod.collectionExists(&db, allocator, .hash, "doomed"));
+
+    // querying a collection that was never there at all (not merely
+    // expired) is .noKey, distinct from .noExpiry.
+    try std.testing.expectEqual(ttlmod.TtlResult.noKey, try client.collTtlQuery(.set, "never-existed"));
+}
+
+test "coll_ttl_op gates query as readonly and expire as readwrite" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const path = try tmp.dir.realpathAlloc(allocator, ".");
+    defer allocator.free(path);
+
+    var db = try testDb(allocator, path);
+    defer db.close();
+    _ = try collmod.Hash.set(&db, allocator, "k", "f", "v");
+
+    const store = authmod.Store.init(&db);
+    try store.createUser(allocator, "reader", "pw", .readonly, 16);
+    try store.createUser(allocator, "writer", "pw", .readwrite, 16);
+
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    server.enableAuth(store);
+    try server.start();
+    defer server.stop();
+
+    const client_mod = @import("client.zig");
+
+    var writer = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer writer.close();
+    try writer.authenticate("writer", "pw");
+
+    var reader = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer reader.close();
+    try reader.authenticate("reader", "pw");
+
+    // query is readonly: a readonly connection can run it.
+    try std.testing.expectEqual(ttlmod.TtlResult.noExpiry, try reader.collTtlQuery(.hash, "k"));
+
+    // expire/persist/sweep all mutate: forbidden for a readonly connection.
+    try std.testing.expectError(client_mod.ClientError.ServerForbidden, reader.collTtlExpire(.hash, "k", 30));
+    try std.testing.expectError(client_mod.ClientError.ServerForbidden, reader.collTtlPersist(.hash, "k"));
+    try std.testing.expectError(client_mod.ClientError.ServerForbidden, reader.collTtlSweep());
+
+    // The readwrite connection can do all four.
+    try writer.collTtlExpire(.hash, "k", 30);
+    try writer.collTtlPersist(.hash, "k");
+    _ = try writer.collTtlSweep();
+}
+
+test "server rejects a malformed coll_ttl_op frame with bad_request without crashing" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const path = try tmp.dir.realpathAlloc(allocator, ".");
+    defer allocator.free(path);
+
+    var db = try testDb(allocator, path);
+    defer db.close();
+
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    try server.start();
+    defer server.stop();
+
+    const stream = try std.net.tcpConnectToAddress(server.boundAddress());
+    defer stream.close();
+
+    // A single byte is enough for an action but not the kind byte that
+    // must follow it, so `decodeCollTtlOp` fails on the second `takeByte`.
+    try protocol.writeFrame(stream.writer(), @intFromEnum(protocol.Op.coll_ttl_op), "\x02");
+    var frame = try protocol.readFrame(allocator, stream.reader());
+    defer frame.deinit();
+    try std.testing.expectEqual(@intFromEnum(protocol.Status.bad_request), frame.tag);
+}
+
+test "enableCollTtlSweep periodically reaps due collections nobody ever touches again" {
+    const allocator = std.testing.allocator;
+    var tmp = std.testing.tmpDir(.{});
+    defer tmp.cleanup();
+    const path = try tmp.dir.realpathAlloc(allocator, ".");
+    defer allocator.free(path);
+
+    var db = try testDb(allocator, path);
+    defer db.close();
+
+    const address = try std.net.Address.parseIp("127.0.0.1", 0);
+    var server = try Server.init(allocator, &db, address);
+    server.enableCollTtlSweep(50 * std.time.ns_per_ms);
+    try server.start();
+    defer server.stop();
+
+    const client_mod = @import("client.zig");
+    var client = try client_mod.Client.connect(allocator, server.boundAddress());
+    defer client.close();
+