EpicNext-Cms/src/app/api/notifications/stream/route.ts

import { auth } from "@/lib/auth";
import { redis } from "@/lib/redis";

export const dynamic = "force-dynamic";
export const runtime = "nodejs";

export async function GET(request: Request) {
	const session = await auth();
	if (!session?.user?.id) {
		return new Response("Unauthorized", { status: 401 });
	}

	const userId = Number(session.user.id);
	if (!userId) {
		return new Response("Unauthorized", { status: 401 });
	}

	// Track active connections per user — close duplicate connections
	if (redis) {
		try {
			const connKey = `stream:conn:${userId}`;
			const existing = await redis.get(connKey);
			if (existing) {
				// Signal the previous connection to close via pub/sub
				await redis.publish(
					`stream:close:${userId}`,
					JSON.stringify({ reason: "duplicate" }),
				);
			}
			await redis.set(connKey, "active", "EX", 60);
		} catch {}
	}

	const stream = new ReadableStream({
		start(controller) {
			const encoder = new TextEncoder();
			let cleanup: (() => void) | null = null;
			let closed = false;

			const sendEvent = (data: string) => {
				if (!closed) {
					controller.enqueue(encoder.encode(`data: ${data}\n\n`));
				}
			};

			sendEvent(JSON.stringify({ type: "connected" }));

			if (redis) {
				const subscriber = redis.duplicate();

				// Listen for close signals (duplicate connection)
				subscriber.subscribe(`stream:close:${userId}`, () => {
					sendEvent(
						JSON.stringify({
							type: "closed",
							message: "Duplicate connection detected",
						}),
					);
					closed = true;
					controller.close();
				});

				subscriber.subscribe(`user:${userId}`, (err) => {
					if (err) {
						sendEvent(
							JSON.stringify({
								type: "error",
								message: "Subscription failed",
							}),
						);
					}
				});

				subscriber.on("message", (_channel, message) => {
					sendEvent(message);
				});

				cleanup = () => {
					subscriber.unsubscribe();
					subscriber.quit();
				};
			}

			const keepAlive = setInterval(() => {
				sendEvent(JSON.stringify({ type: "ping" }));
			}, 30000);

			request.signal.addEventListener("abort", () => {
				clearInterval(keepAlive);
				cleanup?.();
				if (redis) {
					try {
						redis.del(`stream:conn:${userId}`);
					} catch {}
				}
			});
		},
	});

	return new Response(stream, {
		headers: {
			"Content-Type": "text/event-stream",
			"Cache-Control": "no-cache",
			Connection: "keep-alive",
		},
	});
}