feat: add Hub and deinit to pubsub

src/pubsub.zig

@@ -151,3 +151,161 @@ pub const Subscriber = struct {
     }
 };
 
+/// The channel registry every `Subscriber` is filed under. Completely
+/// independent of `Db` - publish/subscribe never touches storage, so a
+/// `Hub`'s own mutex is never held anywhere near `Db`'s (see `server.zig`'s
+/// `dispatch`, which handles the `subscribe`/`publish` opcodes before it
+/// ever reaches for `Server.mutex`, so a slow or permanently-open
+/// subscriber connection can never stall a read or write against the
+/// store).
+pub const Hub = struct {
+    allocator: Allocator,
+    mutex: std.Thread.Mutex = .{},
+    channels: std.StringHashMap(std.ArrayList(*Subscriber)),
+    /// Every currently-live subscriber, keyed by the `id` `subscribe`
+    /// assigned it - lets `unsubscribe` (see its doc comment) look one up by
+    /// the bare number a client saw in its subscribe ack, without needing
+    /// the original `*Subscriber` pointer, which never crosses a second TCP
+    /// connection. Entries are removed exactly when `closeSubscriber` runs
+    /// (or `unsubscribe` empties a subscriber's channel set and closes it
+    /// itself), so a stale or already-closed id reliably misses here rather
+    /// than risking a dangling pointer.
+    byId: std.AutoHashMap(u64, *Subscriber),
+    nextId: u64 = 0,
+
+    pub fn init(allocator: Allocator) Hub {
+        return .{
+            .allocator = allocator,
+            .channels = std.StringHashMap(std.ArrayList(*Subscriber)).init(allocator),
+            .byId = std.AutoHashMap(u64, *Subscriber).init(allocator),
+        };
+    }
+
+    /// Frees every remaining channel's subscriber list and key copy, and the
+    /// id registry. Does NOT touch any `Subscriber` still registered -
+    /// callers must `closeSubscriber` (which also unregisters, so in an
+    /// orderly shutdown this map is already empty by the time `deinit` runs)
+    /// and `destroySubscriber` every subscriber they created first, the same
+    /// "caller drains, then frees the container" ordering `Db`/`Server`
+    /// already use elsewhere in this codebase.
+    pub fn deinit(self: *Hub) void {
+        var it = self.channels.iterator();
+        while (it.next()) |entry| {
+            entry.value_ptr.deinit();
+            self.allocator.free(entry.key_ptr.*);
+        }
+        self.channels.deinit();
+        self.byId.deinit();
+    }
+
+    /// Creates a new subscriber registered under every name in
+    /// `channelNames` (duplicate names collapse to one registration, so a
+    /// message published on a channel a caller listed twice is still only
+    /// ever delivered once). `capacity` bounds its mailbox (see
+    /// `Subscriber.push`). Returns a pointer the caller must eventually
+    /// pass to `closeSubscriber` then `destroySubscriber`, in that order.
+    pub fn subscribe(self: *Hub, channelNames: []const []const u8, capacity: usize) !*Subscriber {
+        var unique = std.ArrayList([]const u8).init(self.allocator);
+        defer unique.deinit();
+        for (channelNames) |name| {
+            var seen = false;
+            for (unique.items) |u| {
+                if (std.mem.eql(u8, u, name)) {
+                    seen = true;
+                    break;
+                }
+            }
+            if (!seen) try unique.append(name);
+        }
+
+        const owned = try self.allocator.alloc([]u8, unique.items.len);
+        var ownedCount: usize = 0;
+        errdefer {
+            for (owned[0..ownedCount]) |c| self.allocator.free(c);
+            self.allocator.free(owned);
+        }
+        for (unique.items, 0..) |name, i| {
+            owned[i] = try self.allocator.dupe(u8, name);
+            ownedCount += 1;
+        }
+
+        const sub = try self.allocator.create(Subscriber);
+        errdefer self.allocator.destroy(sub);
+        sub.* = Subscriber.init(self.allocator, capacity, owned);
+
+        self.mutex.lock();
+        defer self.mutex.unlock();
+
+        const id = self.nextId;
+        self.nextId += 1;
+        sub.id = id;
+        try self.byId.put(id, sub);
+        errdefer _ = self.byId.remove(id);
+
+        var registered: usize = 0;
+        errdefer {
+            for (owned[0..registered]) |name| {
+                if (self.channels.getPtr(name)) |list| {
+                    for (list.items, 0..) |s, idx| {
+                        if (s == sub) {
+                            _ = list.swapRemove(idx);
+                            break;
+                        }
+                    }
+                }
+            }
+        }
+        for (owned) |name| {
+            const gop = try self.channels.getOrPut(name);
+            if (!gop.found_existing) {
+                gop.key_ptr.* = self.allocator.dupe(u8, name) catch |err| {
+                    // The map entry was already inserted with `name`'s
+                    // pointer as a placeholder key by `getOrPut` itself;
+                    // remove it before propagating so `deinit`/`subscribe`
+                    // never iterate over a key pointing into memory this
+                    // function is about to free via `owned`'s errdefer.
+                    _ = self.channels.remove(name);
+                    return err;
+                };
+                gop.value_ptr.* = std.ArrayList(*Subscriber).init(self.allocator);
+            }
+            try gop.value_ptr.append(sub);
+            registered += 1;
+        }
+
+        return sub;
+    }
+
+    /// Stops delivering new messages to `sub` and unregisters it from every
+    /// channel it was subscribed to, wakes any thread blocked in its
+    /// `waitMessage`, and drops a channel's map entry entirely once its
+    /// subscriber list becomes empty (an unsubscribed-from channel is a
+    /// completely normal, permanent steady state, not a reason to keep an
+    /// empty list around forever). Safe to call more than once (from a
+    /// watcher thread and a failed publish both noticing the same dead
+    /// connection, say) - both `Subscriber.close` and the unregister loop
+    /// below are no-ops the second time. Does NOT free `sub` itself - see
+    /// `destroySubscriber`.
+    pub fn closeSubscriber(self: *Hub, sub: *Subscriber) void {
+        sub.close();
+        self.mutex.lock();
+        defer self.mutex.unlock();
+        _ = self.byId.remove(sub.id);
+        for (sub.channels) |name| {
+            if (self.channels.getPtr(name)) |list| {
+                for (list.items, 0..) |s, idx| {
+                    if (s == sub) {
+                        _ = list.swapRemove(idx);
+                        break;
+                    }
+                }
+                if (list.items.len == 0) {
+                    if (self.channels.fetchRemove(name)) |kv| {
+                        kv.value.deinit();
+                        self.allocator.free(kv.key);
+                    }
+                }
+            }
+        }
+    }
+