fix(server): a revoked or expired session's preview stream closes

apps/server/src/preview/ServerBrowserStream.test.ts

@@ -12,6 +12,7 @@ import {
   type AuthEnvironmentScope,
 } from "@t3tools/contracts";
 import * as Context from "effect/Context";
+import * as Deferred from "effect/Deferred";
 import * as Effect from "effect/Effect";
 import * as FileSystem from "effect/FileSystem";
 import * as Layer from "effect/Layer";
@@ -21,6 +22,7 @@ import * as Schema from "effect/Schema";
 import { HttpRouter, HttpServer } from "effect/http";
 
 import * as EnvironmentAuth from "../auth/EnvironmentAuth.ts";
+import * as SessionStore from "../auth/SessionStore.ts";
 import * as PreviewBrowserHost from "./PreviewBrowserHost.ts";
 import * as ServerBrowser from "./ServerBrowser.ts";
 import { routeLayer } from "./ServerBrowserStream.ts";
@@ -30,6 +32,7 @@ const platformLayer = NodeHttpPlatform.layer.pipe(Layer.provideMerge(NodeService
 const makeAuth = (
   scopes: ReadonlyArray<AuthEnvironmentScope>,
   error?: EnvironmentAuth.ServerAuthCredentialError,
+  invalidation: Effect.Effect<void> = Effect.never,
 ) => {
   const requests: string[] = [];
   const layer = Layer.mock(EnvironmentAuth.EnvironmentAuth, {
@@ -45,7 +48,17 @@ const makeAuth = (
           });
     },
   });
-  return { layer, requests };
+  return {
+    layer: Layer.merge(
+      layer,
+      Layer.mock(SessionStore.SessionStore, {
+        cookieName: "test-session",
+        legacyCookieName: undefined,
+        awaitInvalidation: () => invalidation,
+      }),
+    ),
+    requests,
+  };
 };
 
 const mutations = [
@@ -152,6 +165,86 @@ it.effect.each([
   }).pipe(Effect.scoped),
 );
 
+it.effect.each([false, true])(
+  "closes an established viewer when its session is invalidated (operate=%s)",
+  (operate) =>
+    Effect.gen(function* () {
+      const invalidated = yield* Deferred.make<void>();
+      const detached = yield* Deferred.make<void>();
+      const inputStarted = yield* Deferred.make<void>();
+      const inputInterrupted = yield* Deferred.make<void>();
+      const output = yield* Queue.make<ServerBrowser.ServerBrowserViewerOutput>();
+      const auth = makeAuth(
+        operate
+          ? [AuthOrchestrationReadScope, AuthPreviewOperateScope]
+          : [AuthOrchestrationReadScope],
+        undefined,
+        Deferred.await(invalidated),
+      );
+      const browser = ServerBrowser.ServerBrowser.of({
+        clearProfile: () => Effect.void,
+        profileReporter: Effect.succeed(() => Effect.void),
+        openDownload: () => Effect.succeedNone,
+        answerFileChooser: () => Effect.succeed(false),
+        attachViewer: () =>
+          Effect.gen(function* () {
+            yield* Effect.addFinalizer(() => Deferred.succeed(detached, undefined));
+            return {
+              output,
+              input: () =>
+                Deferred.succeed(inputStarted, undefined).pipe(
+                  Effect.andThen(Effect.never),
+                  Effect.onInterrupt(() => Deferred.succeed(inputInterrupted, undefined)),
+                ),
+            };
+          }),
+      });
+      const services = yield* Layer.build(
+        HttpRouter.serve(
+          routeLayer.pipe(
+            Layer.provide(Layer.succeed(ServerBrowser.ServerBrowser, browser)),
+            Layer.provide(platformLayer),
+          ),
+          { disableListenLog: true },
+        ).pipe(Layer.provideMerge(NodeHttpServer.layerTest), Layer.provide(auth.layer)),
+      );
+      const server = Context.get(services, HttpServer.HttpServer);
+      const origin = HttpServer.formatAddress(server.address).replace(/^http/, "ws");
+      const received = Promise.withResolvers<void>();
+      const closed = Promise.withResolvers<void>();
+      const frames: unknown[] = [];
+      const socket = yield* Effect.acquireRelease(
+        Effect.sync(() => {
+          const socket = new WebSocket(
+            `${origin}/api/preview-stream/ws?threadId=thread&tabId=tab&wsTicket=one-use-ticket`,
+          );
+          socket.addEventListener("message", (event) => {
+            frames.push(event.data);
+            received.resolve();
+          });
+          socket.addEventListener("close", () => closed.resolve());
+          socket.addEventListener("error", () => received.reject(new Error("stream failed")));
+          return socket;
+        }),
+        (socket) => Effect.sync(() => socket.close()),
+      );
+      yield* Queue.offer(output, { _tag: "frame", data: new Uint8Array([1]), ack: Effect.void });
+      yield* Effect.promise(() => received.promise);
+      if (operate) {
+        socket.send(JSON.stringify({ type: "takeControl" }));
+        yield* Deferred.await(inputStarted);
+      }
+      yield* Deferred.succeed(invalidated, undefined);
+      // The viewer detaches, any running input is interrupted, and the socket closes.
+      yield* Deferred.await(detached);
+      yield* Effect.promise(() => closed.promise);
+      if (operate) yield* Deferred.await(inputInterrupted);
+      yield* Queue.offer(output, { _tag: "frame", data: new Uint8Array([2]), ack: Effect.void });
+      expect(frames).toHaveLength(1);
+      expect(socket.readyState).toBe(WebSocket.CLOSED);
+    }).pipe(Effect.scoped),
+);
+
 it.effect("sends replies ahead of stale frames and only the newest frame after an ack", () =>
   Effect.gen(function* () {
     const output = yield* Queue.make<ServerBrowser.ServerBrowserViewerOutput>();

apps/server/src/preview/ServerBrowserStream.ts

@@ -20,6 +20,7 @@ import {
 } from "effect/http";
 import * as Socket from "effect/socket/Socket";
 
+import * as SessionStore from "../auth/SessionStore.ts";
 import { authenticateMediaRequest } from "../auth/http.ts";
 import { assetResponseHeaders } from "../http.ts";
 import * as PreviewBrowserHost from "./PreviewBrowserHost.ts";
@@ -71,6 +72,7 @@ const makeHandler = Effect.fnUntraced(function* (browser: ServerBrowser.ServerBr
     return HttpServerResponse.text("Not Found", { status: 404 });
   }
   const session = yield* authenticateMediaRequest(AuthOrchestrationReadScope);
+  const sessions = yield* SessionStore.SessionStore;
   const params = url.value.searchParams;
   const canOperate =
     session.scopes.includes(AuthPreviewOperateScope) && params.get("interactive") !== "false";
@@ -201,6 +203,12 @@ const makeHandler = Effect.fnUntraced(function* (browser: ServerBrowser.ServerBr
       return yield* Effect.raceFirst(Effect.forever(sendOutput), Effect.forever(receiveInput));
     }),
   ).pipe(
+    // Like the RPC socket: a revoked or expired session ends the stream and its input.
+    Effect.raceFirst(
+      sessions
+        .awaitInvalidation(session.sessionId, session.proofKeyThumbprint)
+        .pipe(Effect.as(HttpServerResponse.empty())),
+    ),
     Effect.catchTags({
       ServerBrowserLaunchError: (error) =>
         Effect.logWarning("server preview browser failed to start", { cause: error.cause }).pipe(