// SSE endpoint for the backlog 3-pane (PBI / story / task changes). // Simpler than /api/realtime/solo — no sprint or user scoping, just product_id filter. // Auth: iron-session cookie. Demo users may read (no 403 for demo). import { NextRequest } from 'next/server' import { Client } from 'pg' import { getSession } from '@/lib/auth' import { getAccessibleProduct } from '@/lib/product-access' import { closePgClientSafely } from '@/lib/realtime/pg-client-cleanup' export const runtime = 'nodejs' export const dynamic = 'force-dynamic' export const maxDuration = 300 const CHANNEL = 'scrum4me_changes' const HEARTBEAT_MS = 25_000 const HARD_CLOSE_MS = 240_000 type NotifyPayload = Record function shouldEmit(payload: NotifyPayload, productId: string): boolean { if ('type' in payload) return false // job / worker events — not relevant here const entity = payload.entity as string | undefined if (!entity || !['pbi', 'story', 'task'].includes(entity)) return false return payload.product_id === productId } export async function GET(request: NextRequest) { const session = await getSession() if (!session.userId) { return Response.json({ error: 'Niet ingelogd' }, { status: 401 }) } const productId = request.nextUrl.searchParams.get('product_id') if (!productId) { return Response.json({ error: 'product_id is verplicht' }, { status: 400 }) } const product = await getAccessibleProduct(productId, session.userId) if (!product) { return Response.json({ error: 'Geen toegang tot dit product' }, { status: 403 }) } const directUrl = process.env.DIRECT_URL ?? process.env.DATABASE_URL if (!directUrl) { return Response.json({ error: 'DIRECT_URL/DATABASE_URL niet geconfigureerd' }, { status: 500 }) } const encoder = new TextEncoder() const pgClient = new Client({ connectionString: directUrl }) let heartbeatTimer: ReturnType | null = null let hardCloseTimer: ReturnType | null = null let closed = false const stream = new ReadableStream({ async start(controller) { const enqueue = (chunk: string) => { if (closed) return try { controller.enqueue(encoder.encode(chunk)) } catch { // stream already closed } } const cleanup = async (reason: string) => { if (closed) return closed = true if (heartbeatTimer) clearInterval(heartbeatTimer) if (hardCloseTimer) clearTimeout(hardCloseTimer) await closePgClientSafely(pgClient, 'realtime/backlog') try { controller.close() } catch { /* already closed */ } if (process.env.NODE_ENV !== 'production') { console.log(`[realtime/backlog] closed: ${reason}`) } } try { await pgClient.connect() await pgClient.query(`LISTEN ${CHANNEL}`) } catch (err) { console.error('[realtime/backlog] pg connect/listen failed:', err) enqueue(`event: error\ndata: ${JSON.stringify({ message: 'pg connect failed' })}\n\n`) await cleanup('pg connect failed') return } pgClient.on('notification', (msg) => { if (!msg.payload) return let payload: NotifyPayload try { payload = JSON.parse(msg.payload) as NotifyPayload } catch { return } if (!shouldEmit(payload, productId)) return enqueue(`data: ${msg.payload}\n\n`) }) pgClient.on('error', async (err) => { console.error('[realtime/backlog] pg client error:', err) await cleanup('pg error') }) enqueue(`event: ready\ndata: ${JSON.stringify({ product_id: productId })}\n\n`) heartbeatTimer = setInterval(() => { enqueue(`: heartbeat\n\n`) }, HEARTBEAT_MS) hardCloseTimer = setTimeout(() => { cleanup('hard close 240s') }, HARD_CLOSE_MS) request.signal.addEventListener('abort', () => { cleanup('client aborted') }) }, }) return new Response(stream, { headers: { 'Content-Type': 'text/event-stream; charset=utf-8', 'Cache-Control': 'no-cache, no-transform', Connection: 'keep-alive', 'X-Accel-Buffering': 'no', }, }) }