konnosDashboard
subscription-registry.ts82 lines · main
1/**
2 * Ref-counted map of LISTEN channels to subscriber ids.
3 *
4 * Phase 1 §1.1: factored out of `index.ts` so the add/remove invariants
5 * are unit-testable in isolation. The realtime service owns one of these
6 * per process. Channel names are project- and table-scoped strings
7 * (`briven_<schema>_<table>`) — the registry treats them as opaque keys.
8 *
9 * Semantics
10 * - `attach(subId, channel)` → adds the sub to the channel's set;
11 * returns true iff this attach took the
12 * set from empty → non-empty (the caller
13 * uses that signal to issue `LISTEN`).
14 * - `detach(subId, channel)` → removes the sub; returns true iff this
15 * detach drained the set (caller issues
16 * `UNLISTEN`). Detaching an unknown sub
17 * from a known channel is a no-op false.
18 * - Idempotency: attach is idempotent (Set semantics); detach is
19 * idempotent against an already-empty channel.
20 * - `subsForChannel` returns a snapshot — callers can mutate the
21 * registry while iterating without skipping or revisiting.
22 */
23export class SubscriptionRegistry {
24 private channelToSubs = new Map<string, Set<string>>();
25
26 attach(subId: string, channel: string): boolean {
27 let set = this.channelToSubs.get(channel);
28 if (!set) {
29 set = new Set();
30 this.channelToSubs.set(channel, set);
31 }
32 const wasEmpty = set.size === 0;
33 set.add(subId);
34 return wasEmpty;
35 }
36
37 detach(subId: string, channel: string): boolean {
38 const set = this.channelToSubs.get(channel);
39 if (!set) return false;
40 if (!set.delete(subId)) return false;
41 if (set.size === 0) {
42 this.channelToSubs.delete(channel);
43 return true;
44 }
45 return false;
46 }
47
48 subsForChannel(channel: string): readonly string[] {
49 const set = this.channelToSubs.get(channel);
50 return set ? [...set] : [];
51 }
52
53 hasChannel(channel: string): boolean {
54 return this.channelToSubs.has(channel);
55 }
56
57 get channelCount(): number {
58 return this.channelToSubs.size;
59 }
60
61 /** Return every channel that belongs to a given project. Channels are
62 * named `briven_proj_<id>_<table>` — we prefix-match the project id. */
63 channelsForProject(projectId: string): string[] {
64 const prefix = `briven_${dbNameFor(projectId)}_`;
65 return [...this.channelToSubs.keys()].filter((c) => c.startsWith(prefix));
66 }
67
68 /** Snapshot of all channels and their current sub counts. Used by the
69 * `/v1/realtime/stats` operator endpoint; not on any hot path. */
70 channelCounts(): { channel: string; subscriptions: number }[] {
71 const out: { channel: string; subscriptions: number }[] = [];
72 for (const [channel, set] of this.channelToSubs) {
73 out.push({ channel, subscriptions: set.size });
74 }
75 return out;
76 }
77}
78
79function dbNameFor(projectId: string): string {
80 const safe = projectId.replace(/[^a-zA-Z0-9]/g, '_').toLowerCase();
81 return `proj_${safe}`;
82}