feat(core-realtime): SocketIORealtimeBroadcaster

packages/core-realtime/src/socket-io-realtime-broadcaster.test.ts

@@ -0,0 +1,33 @@
+import { describe, it, expect, vi } from "vitest";
+import { z } from "zod";
+import { SocketIORealtimeBroadcaster } from "@/socket-io-realtime-broadcaster";
+import { defineRealtimeChannel } from "@/realtime-channel";
+
+const ch = defineRealtimeChannel(
+  "a.b",
+  z.object({ x: z.number() }).strict(),
+  { scope: "public" },
+);
+
+describe("SocketIORealtimeBroadcaster", () => {
+  it("emits to the channel's room with the channel name as event", async () => {
+    const emit = vi.fn();
+    const to = vi.fn(() => ({ emit }));
+    const io = { to } as never;
+    const b = new SocketIORealtimeBroadcaster(io);
+    await b.broadcast(ch, { x: 1 });
+    expect(to).toHaveBeenCalledWith("ch:a.b");
+    expect(emit).toHaveBeenCalledWith("a.b", { x: 1 });
+  });
+
+  it("validates payload before emitting", async () => {
+    const emit = vi.fn();
+    const to = vi.fn(() => ({ emit }));
+    const io = { to } as never;
+    const b = new SocketIORealtimeBroadcaster(io);
+    await expect(
+      b.broadcast(ch, { x: "not a number" } as never),
+    ).rejects.toThrow();
+    expect(emit).not.toHaveBeenCalled();
+  });
+});

packages/core-realtime/src/socket-io-realtime-broadcaster.ts

@@ -0,0 +1,16 @@
+import type { Server as IOServer } from "socket.io";
+import type { z } from "zod";
+import type { IRealtimeBroadcaster } from "./realtime-broadcaster.interface";
+import type { RealtimeChannelDescriptor } from "./realtime-channel";
+
+export class SocketIORealtimeBroadcaster implements IRealtimeBroadcaster {
+  constructor(private readonly io: IOServer) {}
+
+  async broadcast<T>(
+    descriptor: RealtimeChannelDescriptor<string, z.ZodType<T>>,
+    payload: T,
+  ): Promise<void> {
+    descriptor.schema.parse(payload);
+    this.io.to(`ch:${descriptor.name}`).emit(descriptor.name, payload);
+  }
+}