import type { NextRequest } from "next/server";
import { redisClient } from "@/repo/RedisClient";
import { FlagsService } from "@/services/FlagsService";

const subscriber = redisClient.createSubscriber();

// Map: topic -> Set<{controller, connectionId}> (still pod-local)
const connections = new Map<
	string,
	Set<{ controller: ReadableStreamDefaultController; connectionId: string }>
>();

function findConnectionByController(
	controller: ReadableStreamDefaultController,
) {
	for (const [topic, connectionSet] of connections) {
		for (const connection of connectionSet) {
			if (connection.controller === controller) {
				return { connection, topic };
			}
		}
	}
	return null;
}

subscriber.subscribe("sse-broadcast");
subscriber.on("message", (channel, message) => {
	if (channel === "sse-broadcast") {
		try {
			const event = JSON.parse(message);
			broadcastToLocalConnections(event);
		} catch (error) {
			console.error("Error processing Redis message:", error);
		}
	}
});

// biome-ignore lint/suspicious/noExplicitAny: SSE data can contain anything
function broadcastToLocalConnections(event: any) {
	const {
		topic,
		broadcastToAll,
		broadcastToOtherUsers,
		excludeConnectionId,
		targetConnectionId,
	} = event;

	const eventString = `data: ${JSON.stringify(event)}\n\n`;
	const encoder = new TextEncoder();
	const eventData = encoder.encode(eventString);

	const connectionsToRemove: Array<{
		connection: {
			controller: ReadableStreamDefaultController;
			connectionId: string;
		};
		topic: string;
	}> = [];

	if (targetConnectionId) {
		// Send to specific connection
		connections.forEach((connectionSet, currentTopic) => {
			if (currentTopic === topic || topic === "*") {
				connectionSet.forEach((connection) => {
					if (connection.connectionId === targetConnectionId) {
						try {
							connection.controller.enqueue(eventData);
						} catch {
							connectionsToRemove.push({ connection, topic: currentTopic });
						}
					}
				});
			}
		});
	} else if (broadcastToOtherUsers && excludeConnectionId) {
		// Broadcast to all connections except the one with excludeConnectionId
		connections.forEach((connectionSet, currentTopic) => {
			if (currentTopic === topic || topic === "*") {
				connectionSet.forEach((connection) => {
					if (connection.connectionId !== excludeConnectionId) {
						try {
							connection.controller.enqueue(eventData);
						} catch {
							connectionsToRemove.push({ connection, topic: currentTopic });
						}
					}
				});
			}
		});
	} else if (broadcastToAll) {
		// Broadcast to all connections in the topic
		connections.forEach((connectionSet, currentTopic) => {
			if (currentTopic === topic || topic === "*") {
				connectionSet.forEach((connection) => {
					try {
						connection.controller.enqueue(eventData);
					} catch {
						connectionsToRemove.push({ connection, topic: currentTopic });
					}
				});
			}
		});
	} else {
		// Broadcast to specific topic connections
		const connectionSet = connections.get(topic);
		if (connectionSet) {
			connectionSet.forEach((connection) => {
				try {
					connection.controller.enqueue(eventData);
				} catch {
					connectionsToRemove.push({ connection, topic });
				}
			});
		}
	}

	// Clean up dead connections
	cleanupConnections(connectionsToRemove);
}

function cleanupConnections(
	connectionsToRemove: Array<{
		connection: {
			controller: ReadableStreamDefaultController;
			connectionId: string;
		};
		topic: string;
	}>,
) {
	connectionsToRemove.forEach(({ connection, topic: topicToClean }) => {
		const connectionSet = connections.get(topicToClean);
		if (connectionSet) {
			connectionSet.delete(connection);
			if (connectionSet.size === 0) {
				connections.delete(topicToClean);
			}
		}
	});
}

export async function GET(request: NextRequest) {
	// Check if SSE feature is enabled
	const isEnabled = await FlagsService.isFeatureEnabled("enable_sse");
	if (!isEnabled) {
		return new Response(
			JSON.stringify({ error: "SSE is currently disabled" }),
			{
				status: 418, // I'm a teapot - indicates SSE feature is disabled
				headers: { "Content-Type": "application/json" },
			},
		);
	}

	const { searchParams } = new URL(request.url);
	const topic = searchParams.get("topic") || "*";

	const stream = new ReadableStream({
		start(controller) {
			if (!connections.has(topic)) {
				connections.set(topic, new Set());
			}
			const connectionSet = connections.get(topic);

			const connectionId = crypto.randomUUID();
			const connection = { controller, connectionId };
			connectionSet?.add(connection);

			const connectionEvent = {
				id: crypto.randomUUID(),
				topic: "connection",
				data: {
					message: "Connected to SSE",
					connectionId,
				},
				timestamp: Date.now(),
			};

			const eventString = `data: ${JSON.stringify(connectionEvent)}\n\n`;
			const encoder = new TextEncoder();
			const eventData = encoder.encode(eventString);

			try {
				controller.enqueue(eventData);
			} catch (err) {
				console.error("Failed to send connection event:", err);
			}
		},

		cancel(controller) {
			// Find and remove the connection
			const found = findConnectionByController(controller);
			if (found) {
				const connectionSet = connections.get(found.topic);
				if (connectionSet) {
					connectionSet.delete(found.connection);
					if (connectionSet.size === 0) {
						connections.delete(found.topic);
					}
				}
			}
		},
	});

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

export async function POST(request: NextRequest) {
	try {
		// Check if SSE feature is enabled
		const isEnabled = await FlagsService.isFeatureEnabled("enable_sse");
		if (!isEnabled) {
			return Response.json(
				{ error: "SSE is currently disabled" },
				{ status: 418 }, // I'm a teapot - indicates SSE feature is disabled
			);
		}

		const {
			topic,
			data,
			targetConnectionId,
			excludeConnectionId,
			broadcastToAll,
			broadcastToOtherUsers,
		} = await request.json();

		if (!topic) {
			return Response.json({ error: "topic is required" }, { status: 400 });
		}

		const event = {
			id: crypto.randomUUID(),
			topic,
			data: data || {},
			timestamp: Date.now(),
			targetConnectionId,
			excludeConnectionId,
			broadcastToAll,
			broadcastToOtherUsers,
		};

		await redisClient.publish("sse-broadcast", JSON.stringify(event));

		return Response.json({
			success: true,
			eventId: event.id,
			message: "Event published to all connections",
		});
	} catch (error) {
		console.error("Error processing SSE event:", error);
		return Response.json({ error: "Internal server error" }, { status: 500 });
	}
}