fix(events): authorize the events queued before a stream opens once (#8718)

* fix(events): authorize the events queued before a stream opens once

Events that arrived before the stream opened each queued their own
authorization behind the write chain, so ten of them made ten serial checks.
They now share one that starts when the stream opens, while each is still
written in order.

- Comments explain the two defensive guards no test can reach: the fast
  path's empty-queue check and the rejection handler on each authorization.
- The mothership and MCP event route tests reset their ready mocks between
  tests.

* test(events): prove the shared pre-open authorization through the stream's writes

Author: waleedlatif1

apps/sim/app/api/mcp/events/route.test.ts

@@ -17,6 +17,7 @@ describe('MCP tool-change event stream', () => {
     vi.useFakeTimers()
     authMockFns.mockGetSession.mockResolvedValue({ user: { id: 'user-1' } })
     permissionsMockFns.mockGetUserEntityPermissions.mockResolvedValue('read')
+    mcpPubsubMockFns.mockReady.mockReset().mockResolvedValue(undefined)
   })
 
   afterEach(() => {

apps/sim/app/api/mothership/events/route.test.ts

@@ -64,6 +64,7 @@ describe('Mothership owner-scoped event stream', () => {
     permissionsMockFns.mockGetUserEntityPermissions.mockResolvedValue('read')
     authorize.mockResolvedValue({ organizationId: 'org-1', userId: 'user-1', role: 'member' })
     subscribe.mockReturnValue(unsubscribe)
+    mothershipChatStatusMockFns.mockReady.mockReset().mockResolvedValue(undefined)
   })
   afterEach(() => {
     vi.useRealTimers()

apps/sim/lib/events/sse-endpoint.test.ts

@@ -291,6 +291,44 @@ describe('createSSEStream', () => {
     ])
   })
 
+  it('writes every event queued before it opened once one authorization passes', async () => {
+    let live: () => void = () => {}
+    let publish: (eventName: string, data: Record<string, unknown>) => void = () => {}
+    const authorizations: Array<() => void> = []
+    const response = createSSEStream(new NextRequest(new URL('https://sim.test/api/test/stream')), {
+      label: 'test',
+      revalidate: () =>
+        new Promise<void>((resolve) => {
+          authorizations.push(resolve)
+        }),
+      subscriptions: [
+        {
+          subscribe: (send) => {
+            publish = send
+            return () => {}
+          },
+          ready: () =>
+            new Promise<void>((resolve) => {
+              live = resolve
+            }),
+        },
+      ],
+    })
+    const chunks: string[] = []
+    void collect(response.body as ReadableStream<Uint8Array>, chunks)
+    for (let n = 1; n <= 10; n += 1) publish('changed', { n })
+    live()
+    await vi.advanceTimersByTimeAsync(0)
+
+    authorizations[0]()
+    await vi.advanceTimersByTimeAsync(0)
+
+    expect(chunks).toEqual([
+      OPENED_COMMENT,
+      ...Array.from({ length: 10 }, (_, index) => `event: changed\ndata: {"n":${index + 1}}\n\n`),
+    ])
+  })
+
   it('clears its timers when it closes before it opens', async () => {
     const controller = new AbortController()
     createSSEStream(

apps/sim/lib/events/sse-endpoint.ts

@@ -170,11 +170,17 @@ export function createSSEStream(request: NextRequest, config: SSEStreamConfig):
        * waits here while the stream has not opened or an earlier event is still being written.
        */
       let writes: Promise<void> = Promise.resolve()
+      /** Settles when the stream opens, or when it closes first. */
+      let opening: Promise<void> = Promise.resolve()
+      /** The one authorization every event that arrived before the stream opened waits for. */
+      let preOpenAuthorization: Promise<void> | undefined
       let opened = false
       let pendingEvents = 0
       const send = (eventName: string, data: Record<string, unknown>) => {
         if (cleaned) return
         const payload = `event: ${eventName}\ndata: ${JSON.stringify(data)}\n\n`
+        // Defensive: the opening and the writes queued before it settle in the same run of
+        // microtasks, so an event sent from a microtask in between must still queue behind them.
         if (opened && pendingEvents === 0 && !config.revalidate) {
           enqueue(payload)
           return
@@ -184,11 +190,14 @@ export function createSSEStream(request: NextRequest, config: SSEStreamConfig):
           return
         }
         pendingEvents += 1
-        // Authorized as soon as it arrives, so concurrent events share one check, but written only
-        // after every event before it.
+        // Authorized as soon as the stream is open, so events in flight together share one check:
+        // those that arrived before it opened share one that starts when it opens, and later ones
+        // share whichever check is in flight. Each is written only after every event before it.
         const authorized = opened
           ? revalidate()
-          : writes.then(() => (cleaned ? undefined : revalidate()))
+          : (preOpenAuthorization ??= opening.then(() => (cleaned ? undefined : revalidate())))
+        // Defensive: a rejection is handled once the write chain reaches this event, which can be
+        // after the authorization settles; this keeps it from surfacing as unhandled meanwhile.
         authorized.catch(noop)
         writes = writes
           .then(() => authorized)
@@ -211,7 +220,7 @@ export function createSSEStream(request: NextRequest, config: SSEStreamConfig):
         // heartbeat cannot open it first: the deadline is shorter than the heartbeat interval.
         // Closing settles it too, so a stream closed before it opened is not kept alive by a
         // subscription that never becomes ready.
-        writes = Promise.race([
+        opening = Promise.race([
           Promise.all(config.subscriptions.map((subscription) => subscription.ready?.())),
           new Promise<void>((resolve) => {
             const deadline = setTimeout(resolve, OPEN_DEADLINE_MS)
@@ -227,6 +236,7 @@ export function createSSEStream(request: NextRequest, config: SSEStreamConfig):
           },
           () => close('subscription_failed')
         )
+        writes = opening
         for (const subscription of config.subscriptions) {
           teardowns.push(subscription.subscribe(send))
         }