feat: add deployment reload controls

lib/realtime/deployment-refresh.test.ts

@@ -0,0 +1,37 @@
+import { describe, expect, test } from "bun:test";
+import {
+  checkDeploymentVersion,
+  normalizeReloadPrefix,
+  shouldReloadPath,
+} from "@/lib/realtime/deployment-refresh";
+
+describe("deployment refresh policy", () => {
+  test("uses the first version as a baseline and detects later changes", () => {
+    expect(checkDeploymentVersion(null, "build-a")).toEqual({
+      version: "build-a",
+      hasUpdate: false,
+    });
+    expect(checkDeploymentVersion("build-a", "build-a")).toEqual({
+      version: "build-a",
+      hasUpdate: false,
+    });
+    expect(checkDeploymentVersion("build-a", "build-b")).toEqual({
+      version: "build-b",
+      hasUpdate: true,
+    });
+  });
+
+  test("matches global and pathname-prefix reloads", () => {
+    expect(shouldReloadPath(null, "/profile")).toBeTrue();
+    expect(shouldReloadPath("/admin", "/admin/backend")).toBeTrue();
+    expect(shouldReloadPath("/form", "/profile")).toBeFalse();
+  });
+
+  test("normalizes valid prefixes and rejects invalid ones", () => {
+    expect(normalizeReloadPrefix(null)).toBeNull();
+    expect(normalizeReloadPrefix("  /admin  ")).toBe("/admin");
+    expect(() => normalizeReloadPrefix("admin")).toThrow(
+      "Reload path must start with /",
+    );
+  });
+});

lib/realtime/deployment-refresh.ts

@@ -0,0 +1,27 @@
+export const UPDATE_HASH = "#update";
+
+export function checkDeploymentVersion(
+  currentVersion: string | null,
+  incomingVersion: string,
+) {
+  return {
+    version: incomingVersion,
+    hasUpdate:
+      currentVersion !== null && currentVersion !== incomingVersion,
+  };
+}
+
+export function shouldReloadPath(prefix: string | null, pathname: string) {
+  return prefix === null || pathname.startsWith(prefix);
+}
+
+export function normalizeReloadPrefix(prefix: string | null) {
+  if (prefix === null) return null;
+
+  const normalized = prefix.trim();
+  if (!normalized.startsWith("/")) {
+    throw new Error("Reload path must start with /");
+  }
+
+  return normalized;
+}

lib/realtime/sse.test.ts

@@ -1,6 +1,31 @@
 import { describe, expect, test } from "bun:test";
 import { isSseTopic, sse } from "@/lib/realtime/sse";
 
+describe("active SSE contract", () => {
+  test("registers public version and refresh events", () => {
+    expect(isSseTopic("active")).toBeTrue();
+    expect(sse.active.adminOnly).toBeFalse();
+    expect(sse.active.parseRedisMessage(
+      JSON.stringify({ event: "version", data: "build-a" }),
+    )).toEqual({ event: "version", data: "build-a" });
+    expect(sse.active.parseRedisMessage(
+      JSON.stringify({ event: "refresh", data: "/admin" }),
+    )).toEqual({ event: "refresh", data: "/admin" });
+    expect(sse.active.parseRedisMessage(
+      JSON.stringify({ event: "refresh", data: null }),
+    )).toEqual({ event: "refresh", data: null });
+  });
+
+  test("rejects empty active-event values", () => {
+    expect(sse.active.parseRedisMessage(
+      JSON.stringify({ event: "version", data: "" }),
+    )).toBeNull();
+    expect(sse.active.parseRedisMessage(
+      JSON.stringify({ event: "refresh", data: "" }),
+    )).toBeNull();
+  });
+});
+
 describe("leaderboard SSE contract", () => {
   test("registers the public leaderboards topic", () => {
     expect(isSseTopic("leaderboards")).toBeTrue();

lib/realtime/sse.ts

@@ -19,8 +19,16 @@ type SubscriptionOptions = {
   onopen?: (event: Event) => void;
 };
 
-type StreamOptions = {
+type StreamMessage<T extends EventMap> = {
+  [K in Extract<keyof T, string>]: {
+    event: K;
+    data: z.input<T[K]>;
+  };
+}[Extract<keyof T, string>];
+
+type StreamOptions<T extends EventMap> = {
   signal?: AbortSignal;
+  motd?: StreamMessage<T> | StreamMessage<T>[];
 };
 
 const redisEnvelopeSchema = z.object({
@@ -154,7 +162,11 @@ export class SseEndpoint<T extends EventMap> {
     }
   }
 
-  async stream({ signal }: StreamOptions = {}) {
+  async stream({ signal, motd }: StreamOptions<T> = {}) {
+    const initialMessages = (Array.isArray(motd) ? motd : motd ? [motd] : [])
+      .map(({ event, data }) =>
+        encodeEvent(event, this.events[event].parse(data)),
+      );
     const subscriber = await getRedisSubscriber();
     const topic = this.topic;
     const channel = channelName(this.topic);
@@ -223,6 +235,7 @@ export class SseEndpoint<T extends EventMap> {
       start(streamController) {
         controller = streamController;
         write(`retry: ${DEFAULT_RETRY_MS}\n\n: connected ${topic}\n\n`);
+        for (const message of initialMessages) write(message);
 
         receive = handleMessage;
         for (const message of pendingMessages.splice(0)) {
@@ -287,6 +300,12 @@ export function createSseEndpoints<
 const updateActionSchema = z.enum(["created", "updated", "deleted"]);
 
 export const sse = createSseEndpoints({
+  active: {
+    events: {
+      version: z.string().min(1),
+      refresh: z.string().min(1).nullable(),
+    },
+  },
   forms: {
     events: {
       update: z.object({