From 20f56efa1a692a99690c284e30b292a715695bda Mon Sep 17 00:00:00 2001 From: DualFroz Date: Sat, 12 Sep 2026 18:45:06 +0200 Subject: [PATCH] feat: fill in server --- src/server.zig | 112 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 112 insertions(+) diff --git a/src/server.zig b/src/server.zig index 31c115f..5440b3e 100644 --- a/src/server.zig +++ b/src/server.zig @@ -8031,3 +8031,115 @@ test "server unsubscribe from every remaining channel ends the subscription like 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 sub = try client_mod.Client.connect(allocator, server.boundAddress()); + defer sub.close(); + const subId = try sub.subscribe(&.{"ch"}); + + var ctrl = try client_mod.Client.connect(allocator, server.boundAddress()); + defer ctrl.close(); + try std.testing.expectEqual(@as(u32, 1), try ctrl.unsubscribe(subId, &.{"ch"})); + + // The push loop must notice the close exactly like a real disconnect - + // `nextMessage` returns `null`, never blocks forever or errors. + try std.testing.expect((try sub.nextMessage(allocator)) == null); + + const startedAt = std.time.nanoTimestamp(); + while (server.pubsub.channelSubscriberCount("ch") != 0) { + if (std.time.nanoTimestamp() - startedAt > 2 * std.time.ns_per_s) break; + std.time.sleep(5 * std.time.ns_per_ms); + } + try std.testing.expectEqual(@as(usize, 0), server.pubsub.channelSubscriberCount("ch")); +} + +test "server rejects unsubscribe with an unknown subscriber id as bad_request" { + 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(); + + try std.testing.expectError(client_mod.ClientError.ServerBadRequest, client.unsubscribe(999999, &.{"ch"})); + // A rejected unsubscribe must not leave the connection in some + // half-dispatched state - `ping` still works right after. + try client.ping(); +} + +test "server rejects an empty unsubscribe channel list as bad_request" { + 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 sub = try client_mod.Client.connect(allocator, server.boundAddress()); + defer sub.close(); + const subId = try sub.subscribe(&.{"ch"}); + + var ctrl = try client_mod.Client.connect(allocator, server.boundAddress()); + defer ctrl.close(); + try std.testing.expectError(client_mod.ClientError.ServerBadRequest, ctrl.unsubscribe(subId, &.{})); +} + +test "server with auth enabled requires authentication before unsubscribe, readonly role suffices" { + 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 store = authmod.Store.init(&db); + try store.createUser(allocator, "viewer", "hunter2", .readonly, 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 sub = try client_mod.Client.connect(allocator, server.boundAddress()); + defer sub.close(); + try sub.authenticate("viewer", "hunter2"); + const subId = try sub.subscribe(&.{"ch"}); + + var anon = try client_mod.Client.connect(allocator, server.boundAddress()); + defer anon.close(); + try std.testing.expectError(client_mod.ClientError.ServerUnauthorized, anon.unsubscribe(subId, &.{"ch"})); + + var ctrl = try client_mod.Client.connect(allocator, server.boundAddress()); + defer ctrl.close(); + try ctrl.authenticate("viewer", "hunter2"); + try std.testing.expectEqual(@as(u32, 1), try ctrl.unsubscribe(subId, &.{"ch"})); +}