/** * SSE endpoint for in-app notifications. * * Each connected browser tab opens one EventSource here. We extract the user's * AD_User_ID / AD_Role_ID from the JWT in the logship_it cookie, register a * listener on the in-process notificationBus, and forward any event whose * recipientRoleIds includes our role to this connection as an SSE event. * * The client side is in composables/useNotificationStream.ts, mounted at app * load by plugins/03.notifications-stream.client.ts. * * No token refresh is needed here — we only use the token once at handshake * to identify the user. The bus payload is fully self-contained. */ import { notificationBus, type NotificationEvent } from '../../utils/notificationBus' interface NotifClient { res: any roleId: number userId: number listener: (evt: NotificationEvent) => void } // Registry of active connections (per process). Used only for the shared // heartbeat — the bus listener is held per-client via a closure. const activeClients = new Map() const sendSSE = (res: any, payload: any, eventName?: string) => { try { if (eventName) res.write(`event: ${eventName}\n`) res.write(`data: ${JSON.stringify(payload)}\n\n`) } catch (err: any) { console.error('[NotifStream] write failed:', err?.message) } } // One shared heartbeat for all clients on this instance — keeps proxies / // load balancers from closing idle connections (nginx default is 60s). let heartbeatInterval: NodeJS.Timeout | null = null const startHeartbeat = () => { if (heartbeatInterval) return heartbeatInterval = setInterval(() => { activeClients.forEach((c) => { try { c.res.write(`: heartbeat ${Date.now()}\n\n`) } catch {} }) }, 25000) } const stopHeartbeat = () => { if (heartbeatInterval) { clearInterval(heartbeatInterval) heartbeatInterval = null } } export default defineEventHandler(async (event) => { // ── Auth ──────────────────────────────────────────────────────────────── // Cookies are sent automatically by EventSource; same scheme as the rest // of the API routes. Decode the JWT in-place rather than calling iDempiere. let userId: number let roleId: number try { const token = await getTokenHelper(event) if (!token) throw new Error('No token in cookies') const payload = JSON.parse(Buffer.from(token.split('.')[1], 'base64').toString()) userId = payload.AD_User_ID roleId = payload.AD_Role_ID if (!userId || !roleId) throw new Error('AD_User_ID or AD_Role_ID missing from token') } catch (err: any) { console.warn('[NotifStream] auth failed:', err?.message) setResponseStatus(event, 401) return { error: 'Unauthorized', detail: err?.message } } // ── SSE headers ───────────────────────────────────────────────────────── setResponseHeaders(event, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache, no-transform', 'Connection': 'keep-alive', 'X-Accel-Buffering': 'no' // tell nginx not to buffer this stream }) const res = event.node.res const clientId = `notif_${userId}_${Date.now()}_${Math.random().toString(36).slice(2, 8)}` // Per-client bus subscriber. Filters by role membership. const listener = (evt: NotificationEvent) => { if (!evt.recipientRoleIds?.includes(roleId)) return sendSSE(res, evt.payload, 'notification') } notificationBus.on('notification', listener) activeClients.set(clientId, { res, roleId, userId, listener }) console.log(`[NotifStream] connected client=${clientId} user=${userId} role=${roleId} (total=${activeClients.size})`) // Tell the client the handshake completed sendSSE(res, { type: 'connected', clientId, userId, roleId }, 'connected') if (activeClients.size === 1) startHeartbeat() // ── Cleanup ──────────────────────────────────────────────────────────── event.node.req.on('close', () => { notificationBus.off('notification', listener) activeClients.delete(clientId) console.log(`[NotifStream] disconnected client=${clientId} (remaining=${activeClients.size})`) if (activeClients.size === 0) stopHeartbeat() }) // Keep the handler open until the socket closes. return new Promise(() => {}) })