infoalloggi-monorepo/apps/infoalloggi/src/server/api/routers/chat_sse.ts

import { TRPCError, tracked } from "@trpc/server";
import { z } from "zod/v4";
import {
	createTRPCRouter,
	protectedProcedure,
	publicProcedure,
} from "~/server/api/trpc";
import {
	addTyping,
	getCurrentlyTyping,
	removeStaleTyping,
	removeTyping,
} from "~/server/controllers/chat.controller";
import { db } from "~/server/db";
import { add } from "~/server/services/chat.service";
import { ee, type OnlineStatus } from "~/server/sse";
import { zChatId, zUserId } from "~/server/utils/zod_types";
import { getKeydbClient } from "~/utils/keydb";
import { getMessagesArray } from "~/utils/kysely-helper";

export const chatSSERouter = createTRPCRouter({
	create: protectedProcedure
		.input(z.object({ userId: zUserId }))
		.mutation(async ({ input }) => {
			const newChat = await add({
				db,
				userId: input.userId,
			});

			return newChat.chatid;
		}),

	enterChat: protectedProcedure
		.input(z.object({ chatId: zChatId }))
		.mutation(async ({ input, ctx }) => {
			const chatKey = `chat:${input.chatId}:onlineUsers`;
			const keydb = getKeydbClient();

			if (!keydb) {
				console.warn(
					"KeyDB client is not initialized. Skipping online status.",
				);
				return [];
			}
			// Add the user to the Redis set
			await keydb.sadd(
				chatKey,
				JSON.stringify({
					isAdmin: ctx.session.isAdmin,
					nome: ctx.session.nome,
					userId: ctx.session.id,
				}),
			);

			const onlineUsers = await keydb.smembers(chatKey);
			const parsedUsers = onlineUsers.map(
				(user) => JSON.parse(user) as OnlineStatus,
			);
			ee.emit("chatUpdates", input.chatId, {
				data: parsedUsers,
				eventType: "online",
			});
			return parsedUsers;
		}),

	leaveChat: protectedProcedure
		.input(z.object({ chatId: zChatId }))
		.mutation(async ({ input, ctx }) => {
			const chatKey = `chat:${input.chatId}:onlineUsers`;
			const keydb = getKeydbClient();
			if (!keydb) {
				console.warn(
					"KeyDB client is not initialized. Skipping online status.",
				);
				return [];
			}
			// Remove the user from the Redis set
			await keydb.srem(
				chatKey,
				JSON.stringify({
					isAdmin: ctx.session.isAdmin,
					nome: ctx.session.nome,
					userId: ctx.session.id,
				}),
			);

			const onlineUsers = await keydb.smembers(chatKey);
			const parsedUsers = onlineUsers.map(
				(user) => JSON.parse(user) as OnlineStatus,
			);
			ee.emit("chatUpdates", input.chatId, {
				data: parsedUsers,
				eventType: "online",
			});
			return parsedUsers;
		}),
	list: publicProcedure.query(async () => {
		return await db
			.selectFrom("chats")
			.select([
				"chats.chatid",
				"chats.created_at",
				"chats.user_ref",
				getMessagesArray(),
			])
			.innerJoin("users", "users.id", "chats.user_ref")
			.select(["users.username", "users.email", "users.nome"])
			.orderBy("chats.created_at", "desc")
			.execute();
	}),

	onChatEvent: protectedProcedure
		.input(
			z.object({
				chatId: zChatId,
			}),
		)
		.subscription(async function* ({ input, signal }) {
			try {
				const iterable = ee.toIterable("chatUpdates", {
					signal,
				});

				for await (const [chatId, data] of iterable) {
					if (chatId === input.chatId) {
						yield tracked(chatId, data);
					}
				}
			} catch (e) {
				if (e instanceof Error && e.name === "AbortError") return;
				console.error("Error in onChatEvent subscription:", e);
			}
		}),

	type: protectedProcedure
		.input(
			z.object({ chatId: zChatId, typing: z.boolean(), username: z.string() }),
		)
		.mutation(async ({ ctx, input }) => {
			try {
				await removeStaleTyping(input.chatId);

				if (input.typing) {
					// Add the user to the set
					await addTyping({
						chatId: input.chatId,
						userId: ctx.session.id,
						username: input.username,
					});
				} else {
					await removeTyping({
						chatId: input.chatId,
						userId: ctx.session.id,
						username: input.username,
					});
				}

				ee.emit("chatUpdates", input.chatId, {
					data: await getCurrentlyTyping(input.chatId),
					eventType: "typing",
				});
				return true;
			} catch (e) {
				throw new TRPCError({
					code: "INTERNAL_SERVER_ERROR",
					message:
						"An error occurred while processing the typing event: " +
						(e as Error).message,
				});
			}
		}),
});