@@ -2,27 +2,33 @@ import type { NextRequest } from "next/server"
22import { and , eq , gt , isNull , or } from "drizzle-orm"
33import { errorResponse , requireApiOrg } from "@/server/api"
44import { SESSION_COOKIE , validateSessionToken } from "@/server/auth/session"
5- import { subscribe } from "@/server/realtime"
5+ import { subscribe , type RealtimeEvent } from "@/server/realtime"
66import { db , schema } from "@/server/db"
7+ import { getAccountAccess , visibleConversationIds } from "@/server/access"
8+ import { assertApiScope } from "@/server/conversations/context"
79
810export const dynamic = "force-dynamic"
911export const runtime = "nodejs"
1012
1113const HEARTBEAT_MS = 20_000
1214const RECHECK_MS = 5 * 60_000
15+ const VISIBILITY_TTL_MS = 30_000
1316
1417/**
1518 * GET /api/w/[slug]/events — Server-Sent Events stream of realtime events
1619 * for this workspace (`data: RealtimeEvent` JSON per message). Events with
17- * `userIds` are only delivered to those users. Clients refetch data on events;
18- * visibility is enforced by the regular API. Heartbeat comment every 20s; the
19- * session and membership are re-validated every 5 minutes.
20+ * `userIds` are only delivered to those users; events about a conversation
21+ * only to members who can see it (cached per connection for 30s). Clients
22+ * refetch data on events; the regular API enforces visibility as well.
23+ * Heartbeat comment every 20s; the session and membership are re-validated
24+ * every 5 minutes.
2025 */
2126export async function GET ( req : NextRequest , { params } : { params : Promise < { slug : string } > } ) {
2227 const { slug } = await params
2328 let ctx : Awaited < ReturnType < typeof requireApiOrg > >
2429 try {
2530 ctx = await requireApiOrg ( req , slug )
31+ assertApiScope ( ctx , "conversations.read" )
2632 } catch ( err ) {
2733 return errorResponse ( err )
2834 }
@@ -33,6 +39,31 @@ export async function GET(req: NextRequest, { params }: { params: Promise<{ slug
3339 let closed = false
3440 let cleanup = ( ) => { }
3541
42+ const visibility = new Map < string , { visible : boolean ; at : number } > ( )
43+ let accounts : { ids : string [ ] ; at : number } | null = null
44+ const canSee = async ( conversationId : string ) => {
45+ const now = Date . now ( )
46+ const hit = visibility . get ( conversationId )
47+ if ( hit && now - hit . at < VISIBILITY_TTL_MS ) return hit . visible
48+ if ( ! accounts || now - accounts . at >= VISIBILITY_TTL_MS ) accounts = { ids : [ ...( await getAccountAccess ( ctx ) ) . keys ( ) ] , at : now }
49+ const visible = ( await visibleConversationIds ( ctx , [ conversationId ] , accounts . ids ) ) . has ( conversationId )
50+ if ( visibility . size >= 5000 ) visibility . clear ( )
51+ visibility . set ( conversationId , { visible, at : now } )
52+ return visible
53+ }
54+ /** Explicitly addressed events and my own always arrive; other conversation events need visibility. */
55+ const deliverable = async ( event : RealtimeEvent ) => {
56+ if ( ! event . conversationId ) return true
57+ if ( event . userIds ) {
58+ // Addressed to me about this conversation (assigned, mentioned...): its visibility may have changed.
59+ visibility . delete ( event . conversationId )
60+ return true
61+ }
62+ if ( event . actorId === me ) return true
63+ return canSee ( event . conversationId ) . catch ( ( ) => false )
64+ }
65+ let queue = Promise . resolve ( )
66+
3667 const stream = new ReadableStream < Uint8Array > ( {
3768 async start ( controller ) {
3869 const send = ( chunk : string ) => {
@@ -49,9 +80,13 @@ export async function GET(req: NextRequest, { params }: { params: Promise<{ slug
4980 const unsubscribe = await subscribe ( ( event ) => {
5081 if ( event . orgId !== orgId ) return
5182 if ( event . userIds && ! event . userIds . includes ( me ) ) return
52- const { userIds : _recipients , ...payload } = event
53- void _recipients
54- send ( `data: ${ JSON . stringify ( payload ) } \n\n` )
83+ // Chained to keep the order of events while visibility is checked.
84+ queue = queue . then ( async ( ) => {
85+ if ( closed || ! ( await deliverable ( event ) ) ) return
86+ const { userIds : _recipients , ...payload } = event
87+ void _recipients
88+ send ( `data: ${ JSON . stringify ( payload ) } \n\n` )
89+ } )
5590 } )
5691 const heartbeat = setInterval ( ( ) => send ( `: ping ${ Date . now ( ) } \n\n` ) , HEARTBEAT_MS )
5792 const recheck = setInterval ( async ( ) => {
0 commit comments