fix+feat(realtime): connId-keyed subs + event-typed topics

Two fixes in v0.10.2:

1. Subscriptions keyed by stable connId in ws.data, not ws object
   identity. Bun/Elysia hands different wrapper objects to open vs
   message handlers — subscribe stored wrapper-A, unsubscribe looked
   up wrapper-B, Set.delete returned false, broadcasts kept firing.
   Reproduced from a Postman log: subscribe acked, unsubscribe replied
   topics:[] on the same connection.

2. Event-typed topic syntax. posts.create / posts.update / posts.delete
   filter to that event type. *.create etc. catches globally.
   Previously stored as literal keys nothing broadcasted to.

Also: {type:"list-subs"} debug message returns the active topic set
for this connection; subscribe/unsubscribe acks now carry the
canonical list; SSE adapters auto-attach data.connId at register time.

Topic grammar after this:

| Topic                                  | Receives                            |
|----------------------------------------|-------------------------------------|
| <col>                                  | every event for collection          |
| <col>/<id>                             | events for one record               |
| <col>.create / .update / .delete       | only that event type per collection |
| *                                      | every event everywhere              |
| *.create / .update / .delete           | that event type globally            |
| <col>.* / <col>/*                      | normalised to <col>                 |

Tests: +13 (event-typed 5, wrapper-identity 1, normalise+list-subs+ack 7).
Full suite: 831 pass / 2 pre-existing GIF flakes.

Author: khalidsheet

admin/package.json

@@ -1,7 +1,7 @@
 {
   "name": "vaultbase-admin",
   "private": true,
-  "version": "0.10.1",
+  "version": "0.10.2",
   "type": "module",
   "scripts": {
     "dev": "vite",

package.json

@@ -1,6 +1,6 @@
 {
   "name": "vaultbase",
-  "version": "0.10.1",
+  "version": "0.10.2",
   "type": "module",
   "scripts": {
     "dev": "bun --watch src/index.ts",

src/__tests__/realtime-rules.test.ts

@@ -4,10 +4,16 @@ import { subscribe, broadcast, setWSAuth, _reset } from "../realtime/manager.ts"
 interface MockWS {
   sent: string[];
   send(data: string): void;
+  data: { connId: string };
 }
 
+let _mockId = 0;
 function mockWs(): MockWS {
-  return { sent: [], send(data) { this.sent.push(data); } };
+  return {
+    sent: [],
+    send(data) { this.sent.push(data); },
+    data: { connId: `rules-${++_mockId}` },
+  };
 }
 
 function rec(extra: Record<string, unknown> = {}): { record: Parameters<typeof broadcast>[1] extends infer E ? (E extends { record: infer R } ? R : never) : never; raw: Record<string, unknown> } {

src/__tests__/realtime.test.ts

@@ -4,10 +4,16 @@ import { subscribe, unsubscribe, disconnectAll, broadcast, normalizeTopic, _rese
 interface MockWS {
   sent: string[];
   send(data: string): void;
+  data: { connId: string };
 }
 
+let _mockId = 0;
 function mockWs(): MockWS {
-  const ws: MockWS = { sent: [], send(data) { this.sent.push(data); } };
+  const ws: MockWS = {
+    sent: [],
+    send(data) { this.sent.push(data); },
+    data: { connId: `mock-${++_mockId}` },
+  };
   return ws;
 }
 
@@ -51,6 +57,7 @@ describe("RealtimeManager", () => {
     const dead: MockWS = {
       sent: [],
       send() { throw new Error("WebSocket is closed"); },
+      data: { connId: `dead-${++_mockId}` },
     };
     subscribe(dead, ["posts"]);
     expect(() =>
@@ -170,4 +177,85 @@ describe("RealtimeManager", () => {
     const removed = unsubscribe(ws, ["posts.*", "users.*"]);
     expect(removed).toEqual(["posts"]);
   });
+
+  // ── Event-typed topics ────────────────────────────────────────────────
+
+  it("posts.create only fires on create events, not update/delete", () => {
+    const ws = mockWs();
+    subscribe(ws, ["posts.create"]);
+    broadcast("posts", { type: "create", collection: "posts", record: { id: "1", collectionId: "c", collectionName: "posts", created: 0, updated: 0 } });
+    broadcast("posts", { type: "update", collection: "posts", record: { id: "1", collectionId: "c", collectionName: "posts", created: 0, updated: 0 } });
+    broadcast("posts", { type: "delete", collection: "posts", id: "1" });
+    expect(ws.sent).toHaveLength(1);
+    expect(JSON.parse(ws.sent[0]!).type).toBe("create");
+  });
+
+  it("posts.update + posts.delete topics filter independently", () => {
+    const upd = mockWs();
+    const del = mockWs();
+    subscribe(upd, ["posts.update"]);
+    subscribe(del, ["posts.delete"]);
+    broadcast("posts", { type: "create", collection: "posts", record: { id: "1", collectionId: "c", collectionName: "posts", created: 0, updated: 0 } });
+    broadcast("posts", { type: "update", collection: "posts", record: { id: "1", collectionId: "c", collectionName: "posts", created: 0, updated: 0 } });
+    broadcast("posts", { type: "delete", collection: "posts", id: "1" });
+    expect(upd.sent).toHaveLength(1);
+    expect(del.sent).toHaveLength(1);
+    expect(JSON.parse(upd.sent[0]!).type).toBe("update");
+    expect(JSON.parse(del.sent[0]!).type).toBe("delete");
+  });
+
+  it("*.create catches creates from every collection", () => {
+    const ws = mockWs();
+    subscribe(ws, ["*.create"]);
+    broadcast("posts",  { type: "create", collection: "posts",  record: { id: "1", collectionId: "c", collectionName: "posts",  created: 0, updated: 0 } });
+    broadcast("orders", { type: "create", collection: "orders", record: { id: "2", collectionId: "c", collectionName: "orders", created: 0, updated: 0 } });
+    broadcast("posts",  { type: "delete", collection: "posts",  id: "3" });
+    expect(ws.sent).toHaveLength(2);
+  });
+
+  it("collection-level + event-typed sub on same connection still receives once per event", () => {
+    const ws = mockWs();
+    subscribe(ws, ["posts", "posts.create"]);
+    broadcast("posts", { type: "create", collection: "posts", record: { id: "1", collectionId: "c", collectionName: "posts", created: 0, updated: 0 } });
+    expect(ws.sent).toHaveLength(1);
+  });
+
+  it("normalizeTopic preserves event-typed canonical forms", () => {
+    expect(normalizeTopic("posts.create")).toBe("posts.create");
+    expect(normalizeTopic("posts.update")).toBe("posts.update");
+    expect(normalizeTopic("posts.delete")).toBe("posts.delete");
+    expect(normalizeTopic("*.create")).toBe("*.create");
+    // Unknown event suffix kept verbatim — won't match any broadcast,
+    // but doesn't surprise legacy callers either.
+    expect(normalizeTopic("posts.foobar")).toBe("posts.foobar");
+  });
+
+  // ── Bun/Elysia wrapper-identity quirk ──────────────────────────────────
+  // Bun hands `open(ws)` and `message(ws, ...)` different wrapper objects
+  // backing the same socket. The manager keys by `ws.data.connId`, so
+  // wrapper identity doesn't matter — both wrappers carry the same connId
+  // via Bun's persistent `data` slot. Reproduce here with two distinct
+  // mock objects that share a connId, the way Elysia + Bun appear to
+  // present them to handlers.
+  it("subscribe + unsubscribe work across distinct wrappers sharing connId", () => {
+    const sharedId = "conn-A";
+    const wsAtOpen: MockWS = {
+      sent: [], send(d) { this.sent.push(d); },
+      data: { connId: sharedId },
+    };
+    const wsAtMessage: MockWS = {
+      sent: [], send(d) { this.sent.push(d); },
+      data: { connId: sharedId },
+    };
+
+    subscribe(wsAtOpen, ["posts"]);
+    // Unsubscribe via the second wrapper — must still find + delete the sub.
+    const removed = unsubscribe(wsAtMessage, ["posts"]);
+    expect(removed).toEqual(["posts"]);
+
+    // After unsubscribe, broadcasts must not deliver to either wrapper.
+    broadcast("posts", { type: "delete", collection: "posts", id: "x" });
+    expect(wsAtOpen.sent).toHaveLength(0);
+    expect(wsAtMessage.sent).toHaveLength(0);
+  });
 });

src/core/version.ts

@@ -4,4 +4,4 @@
  *
  * Bump in lockstep with `package.json` version + the git tag.
  */
-export const VAULTBASE_VERSION = "0.10.1";
+export const VAULTBASE_VERSION = "0.10.2";

src/realtime/manager.ts

@@ -43,73 +43,116 @@ const WILDCARD = "*";
  *   - "<collection>"          → all events for the collection
  *   - "<collection>/<id>"     → events for one specific record
  *   - "*"                     → every event everywhere
+ *
+ * Storage is keyed by **connection id** (string), not by `WSLike` object
+ * identity. Bun/Elysia can hand you a different wrapper per handler call
+ * (one for `open`, another for `message`); using `===` for membership
+ * misbehaves — subscribe stored wrapper A, unsubscribe looked up wrapper B,
+ * cross-call mutation silently dropped. The id is minted at connect time
+ * and stashed in Bun's persistent `ws.data` slot.
+ *
+ * The inner Map maps connId → adapter so broadcast can still call .send()
+ * via the wrapper that's currently live. Whichever wrapper subscribed last
+ * "wins" — the most recent send target is what fires.
  */
-const subs = new Map<string, Set<WSLike>>();
-const wsAuth = new WeakMap<WSLike, WSAuth>();
+const subs = new Map<string, Map<string, WSLike>>();
+const wsAuth = new Map<string, WSAuth>();
+
+/** Pull the persistent connection id off `ws.data` (set by the WS open handler). */
+function connId(ws: WSLike): string {
+  const id = (ws as unknown as { data?: { connId?: string } }).data?.connId;
+  if (typeof id !== "string") throw new Error("realtime: ws.data.connId missing — open handler must mint one");
+  return id;
+}
 
 export function setWSAuth(ws: WSLike, auth: WSAuth | null): void {
-  if (auth) wsAuth.set(ws, auth);
-  else wsAuth.delete(ws);
+  const id = connId(ws);
+  if (auth) wsAuth.set(id, auth);
+  else wsAuth.delete(id);
 }
 
 export function getWSAuth(ws: WSLike): WSAuth | undefined {
-  return wsAuth.get(ws);
+  return wsAuth.get(connId(ws));
 }
 
 /**
  * Canonicalise a topic string. The internal store keys are:
  *
- *   <collection>            collection-level
- *   <collection>/<id>       single-record
- *   *                       global wildcard
+ *   <collection>                 every event for the collection
+ *   <collection>/<id>            events for one specific record
+ *   <collection>.<event-type>    only that event-type (create / update / delete)
+ *   *                            every event everywhere
+ *   *.<event-type>               that event-type globally
  *
- * For ergonomic + PB-compat reasons we accept these synonyms:
+ * Ergonomic synonyms we collapse:
  *
- *   <collection>.*          → <collection>     (dotted-wildcard)
- *   <collection>/*          → <collection>     (slashed-wildcard)
- *   <collection>            → <collection>     (no-op)
+ *   <collection>.*  → <collection>     (dotted-wildcard)
+ *   <collection>/*  → <collection>     (slashed-wildcard)
  *
- * Returns the canonical form, or `null` if the topic is empty/malformed.
- * Symmetric — used by both subscribe + unsubscribe so the two halves
- * always agree on the storage key.
+ * Symmetric — applied by both subscribe + unsubscribe so the two halves
+ * always agree on the storage key. Returns `null` on empty input.
  */
+const EVENT_KINDS = new Set(["create", "update", "delete"]);
+
 export function normalizeTopic(raw: string): string | null {
   if (typeof raw !== "string") return null;
   const t = raw.trim();
   if (!t) return null;
   if (t === "*") return "*";
   if (t.endsWith(".*")) return t.slice(0, -2) || null;
   if (t.endsWith("/*")) return t.slice(0, -2) || null;
+  // `<base>.<event-type>` — keep verbatim only when the suffix is a
+  // known event kind. Anything else stays as-is for legacy callers.
+  const dot = t.lastIndexOf(".");
+  if (dot > 0) {
+    const suffix = t.slice(dot + 1);
+    if (EVENT_KINDS.has(suffix)) return t; // canonical event-typed form
+  }
   return t;
 }
 
 export function subscribe(ws: WSLike, topics: string[]): string[] {
+  const id = connId(ws);
   const accepted: string[] = [];
   for (const raw of topics) {
     const t = normalizeTopic(raw);
     if (!t) continue;
-    if (!subs.has(t)) subs.set(t, new Set());
-    subs.get(t)!.add(ws);
+    let inner = subs.get(t);
+    if (!inner) { inner = new Map(); subs.set(t, inner); }
+    inner.set(id, ws);
     accepted.push(t);
   }
   return accepted;
 }
 
 export function unsubscribe(ws: WSLike, topics: string[]): string[] {
+  const id = connId(ws);
   const removed: string[] = [];
   for (const raw of topics) {
     const t = normalizeTopic(raw);
     if (!t) continue;
-    if (subs.get(t)?.delete(ws)) removed.push(t);
+    if (subs.get(t)?.delete(id)) removed.push(t);
   }
   return removed;
 }
 
+/** Every topic this WS is currently subscribed to. Cheap introspection for debugging. */
+export function listSubsFor(ws: WSLike): string[] {
+  const id = connId(ws);
+  const out: string[] = [];
+  for (const [topic, inner] of subs.entries()) {
+    if (inner.has(id)) out.push(topic);
+  }
+  out.sort();
+  return out;
+}
+
 export function disconnectAll(ws: WSLike): void {
-  for (const set of subs.values()) {
-    set.delete(ws);
+  const id = connId(ws);
+  for (const inner of subs.values()) {
+    inner.delete(id);
   }
-  wsAuth.delete(ws);
+  wsAuth.delete(id);
 }
 
 /**
@@ -118,9 +161,9 @@ export function disconnectAll(ws: WSLike): void {
  * supplied, everyone passes (back-compat). When supplied, behavior matches
  * the records HTTP `view_rule` semantics.
  */
-function shouldSendTo(ws: WSLike, opts?: BroadcastOpts): boolean {
+function shouldSendTo(id: string, opts?: BroadcastOpts): boolean {
   if (!opts || opts.viewRule === undefined) return true;
-  const auth = wsAuth.get(ws);
+  const auth = wsAuth.get(id);
   if (auth?.type === "admin") return true;
   const rule = opts.viewRule;
   if (rule === null) return true;       // public
@@ -133,34 +176,39 @@ function shouldSendTo(ws: WSLike, opts?: BroadcastOpts): boolean {
 
 /**
  * Send to subscribers of `<collection>`, `<collection>/<id>` (when the event has
- * a record id), and the wildcard `*` topic — fans out with per-ws dedup. When
+ * a record id), and the wildcard `*` topic — fans out with per-id dedup. When
  * the caller passes `opts.viewRule` (and `opts.record` for the eval target),
  * each subscriber's auth is checked against the rule and non-matching connections
  * are skipped silently.
  */
 export function broadcast(collection: string, event: RealtimeEvent, opts?: BroadcastOpts): void {
-  const targets: (string | undefined)[] = [collection, WILDCARD];
+  const targets: (string | undefined)[] = [
+    collection,                          // collection-level
+    WILDCARD,                            // global
+    `${collection}.${event.type}`,       // event-typed per collection
+    `${WILDCARD}.${event.type}`,         // event-typed global
+  ];
   if (event.type === "create" || event.type === "update") {
     targets.push(`${collection}/${event.record.id}`);
   } else if (event.type === "delete") {
     targets.push(`${collection}/${event.id}`);
   }
   const payload = JSON.stringify(event);
-  // Dedup: a ws subscribed to both "posts" and "*" should still receive the
-  // event once. WeakSet doesn't support iteration, so use a regular Set.
-  const sent = new Set<WSLike>();
+  // Dedup: a connection subscribed to both "posts" and "*" should still receive
+  // the event once.
+  const sent = new Set<string>();
   for (const topic of targets) {
     if (!topic) continue;
-    const set = subs.get(topic);
-    if (!set) continue;
-    for (const ws of set) {
-      if (sent.has(ws)) continue;
-      sent.add(ws);
-      if (!shouldSendTo(ws, opts)) continue;
+    const inner = subs.get(topic);
+    if (!inner) continue;
+    for (const [id, ws] of inner) {
+      if (sent.has(id)) continue;
+      sent.add(id);
+      if (!shouldSendTo(id, opts)) continue;
       try {
         ws.send(payload);
       } catch {
-        set.delete(ws);
+        inner.delete(id);
       }
     }
   }
@@ -174,12 +222,12 @@ export function broadcast(collection: string, event: RealtimeEvent, opts?: Broad
  * collide with a user-defined collection.
  */
 export function broadcastSystem(topic: string, message: object): void {
-  const set = subs.get(topic);
-  if (!set) return;
+  const inner = subs.get(topic);
+  if (!inner) return;
   const payload = JSON.stringify(message);
-  for (const ws of set) {
+  for (const [id, ws] of inner) {
     try { ws.send(payload); }
-    catch { set.delete(ws); }
+    catch { inner.delete(id); }
   }
 }
 
@@ -193,6 +241,12 @@ export function broadcastSystem(topic: string, message: object): void {
 const sseClients = new Map<string, WSLike>();
 
 export function registerSSEClient(clientId: string, adapter: WSLike): void {
+  // Mirror the WS contract: every adapter must carry a stable `data.connId`
+  // so subscribe / unsubscribe / disconnectAll have a real key. SSE adapters
+  // typically don't carry `data`, so we attach it here.
+  const a = adapter as unknown as { data?: { connId?: string } };
+  if (!a.data || typeof a.data !== "object") a.data = { connId: clientId };
+  else if (typeof a.data.connId !== "string") a.data.connId = clientId;
   sseClients.set(clientId, adapter);
 }
 
@@ -212,8 +266,9 @@ export function unregisterSSEClient(clientId: string): void {
 export function setSSESubscriptions(clientId: string, topics: string[]): boolean {
   const adapter = sseClients.get(clientId);
   if (!adapter) return false;
+  const id = connId(adapter);
   // Remove from every topic, then re-add the new set.
-  for (const set of subs.values()) set.delete(adapter);
+  for (const inner of subs.values()) inner.delete(id);
   subscribe(adapter, topics);
   return true;
 }