import { jsonArrayFrom, jsonObjectFrom } from "kysely/helpers/postgres"; import { TRPCError } from "@trpc/server"; import type { Chats, ChatsChatid } from "~/schemas/public/Chats"; import type { UsersId } from "~/schemas/public/Users"; import { getUserInfo, type UserInfoQueryType } from "~/utils/kysely-helper"; import type { Etichette } from "~/schemas/public/Etichette"; import { db } from "~/server/db"; import { add, getRaw, type ChatUserInfo } from "~/server/services/chat.service"; import { getKeydbClient } from "~/utils/keydb"; import { subSeconds } from "date-fns"; import type { WhoIsTyping } from "~/server/sse"; export type ActiveChatsType = { id: ChatsChatid; created_at: Date; userinfo: UserInfoQueryType; lastmessage: { message: string | null; time: Date; isLastSender: boolean; } | null; etichette: Etichette[]; }; export const getActiveInfos = async ({ userId, }: { userId: UsersId; }): Promise => { const chats = await db .selectFrom("chats") .select((eb) => [ "chatid", "created_at", getUserInfo(), jsonObjectFrom( eb .selectFrom("messages") .selectAll("messages") .whereRef("messages.chatid", "=", "chats.chatid") .leftJoin("users", "users.id", "messages.sender") .select([ "users.nome as sender_nome", "users.isAdmin as sender_isAdmin", ]) .orderBy("messages.time", "desc") .limit(1), ).as("lastMessage"), jsonArrayFrom( eb .selectFrom("etichette") .selectAll("etichette") .innerJoin( "chats_etichette", "etichette.id_etichetta", "chats_etichette.etichettaid", ) .whereRef("chats_etichette.chatid", "=", "chats.chatid") .select([ "etichette.id_etichetta", "etichette.title", "etichette.color_hex", ]), ).as("etichette"), ]) .orderBy("created_at", "desc") .execute(); const clean = chats .map((chat) => { if (!chat.userinfo) { return null; } return { id: chat.chatid, created_at: chat.created_at, userinfo: chat.userinfo, lastmessage: chat.lastMessage ? { message: chat.lastMessage.message, time: chat.lastMessage.time, isLastSender: chat.lastMessage.sender === userId, } : null, etichette: chat.etichette as Etichette[], }; }) .filter((chat): chat is NonNullable => chat !== null) .sort((a, b) => { const timeA = a.lastmessage ? new Date(a.lastmessage.time).getTime() : 0; const timeB = b.lastmessage ? new Date(b.lastmessage.time).getTime() : 0; return timeB - timeA; }); return clean; }; export const getChatInfo = async ({ chatId, }: { chatId: ChatsChatid; }): Promise => { return await getRaw({ db, chatId }); }; export const getChatByUser = async ({ userId }: { userId: UsersId }) => { try { const chat = await db .selectFrom("chats") .select(["chatid", getUserInfo()]) .where("chats.user_ref", "=", userId) .executeTakeFirst(); if (!chat) { await add({ db, userId: userId, }); const chat = await db .selectFrom("chats") .select(["chatid", getUserInfo()]) .where("chats.user_ref", "=", userId) .executeTakeFirst(); if (!chat) { throw new TRPCError({ code: "NOT_FOUND" }); } return chat; } return chat; } catch (e) { throw new TRPCError({ code: "INTERNAL_SERVER_ERROR", message: "Error while getting chat: " + (e as Error).message, }); } }; export const deleteChatHandler = async ({ chatId, }: { chatId: ChatsChatid; }) => { try { return await db .deleteFrom("chats") .where("chats.chatid", "=", chatId) .returning("chatid") .execute(); } catch (e) { throw new TRPCError({ code: "INTERNAL_SERVER_ERROR", message: "Error while deleting chat: " + (e as Error).message, }); } }; export const getUserByChatHandler = async ({ chatId, }: { chatId: ChatsChatid; }) => { const chat = await db .selectFrom("chats") .select(["chatid", "created_at", "user_ref", getUserInfo()]) .where("chats.chatid", "=", chatId) .executeTakeFirstOrThrow(); if (!chat.userinfo) { throw new TRPCError({ code: "NOT_FOUND" }); } return chat.userinfo as ChatUserInfo; }; export const getChatByUserIdHandler = async ({ userId, }: { userId: UsersId; }) => { const chat = await db .selectFrom("chats") .select("chatid") .where("chats.user_ref", "=", userId) .executeTakeFirst(); return chat?.chatid || null; }; // Redis key prefix for typing status const TYPING_KEY_PREFIX = "chat:typing:"; const TYPING_TTL_SECONDS = 3; // const ONLINEUSER_KEY_PREFIX = "chat:online:"; export const addTyping = async ({ chatId, userId, username, }: { chatId: ChatsChatid; userId: UsersId; username: string; }) => { try { const keydb = getKeydbClient(); const typingKey = `${TYPING_KEY_PREFIX}${chatId}`; const typingData = JSON.stringify({ userId, username, }); if (!keydb) { console.warn("KeyDB client is not initialized. Skipping typing status."); return; } // Add the user to the Redis set with a TTL await keydb.zadd(typingKey, new Date().getTime(), typingData); } catch (e) { throw new TRPCError({ code: "INTERNAL_SERVER_ERROR", message: "Error adding typing status to Redis: " + (e as Error).message, }); } }; export const removeTyping = async ({ chatId, userId, username, }: { chatId: ChatsChatid; userId: UsersId; username: string; }) => { try { const keydb = getKeydbClient(); const typingKey = `${TYPING_KEY_PREFIX}${chatId}`; if (!keydb) { console.warn("KeyDB client is not initialized. Skipping typing removal."); return; } await keydb.zrem(typingKey, JSON.stringify({ userId, username })); } catch (e) { throw new TRPCError({ code: "INTERNAL_SERVER_ERROR", message: "Error removing typing status from Redis: " + (e as Error).message, }); } }; export const removeStaleTyping = async (chatId: ChatsChatid) => { try { const keydb = getKeydbClient(); const typingKey = `${TYPING_KEY_PREFIX}${chatId}`; const now = new Date(); const cutoff = subSeconds(now, TYPING_TTL_SECONDS); if (!keydb) { console.warn( "KeyDB client is not initialized. Skipping stale typing removal.", ); return; } await keydb.zremrangebyscore(typingKey, 0, cutoff.getTime()); } catch (e) { throw new TRPCError({ code: "INTERNAL_SERVER_ERROR", message: "Error removing stale typing status from Redis: " + (e as Error).message, }); } }; export const getCurrentlyTyping = async ( chatId: ChatsChatid, ): Promise => { try { const keydb = getKeydbClient(); const typingKey = `${TYPING_KEY_PREFIX}${chatId}`; if (!keydb) { console.warn("KeyDB client is not initialized. Skipping typing status."); return []; } // Fetch all members of the Redis set const allTypers = await keydb.zrange(typingKey, 0, -1); if (!allTypers || allTypers.length === 0) { return []; } // Parse the entries into the WhoIsTyping structure const typingStatus: WhoIsTyping[] = allTypers.map((entry) => { const { userId, username } = JSON.parse(entry) as WhoIsTyping; return { userId, username }; }); return typingStatus; } catch (e) { throw new TRPCError({ code: "INTERNAL_SERVER_ERROR", message: "Error getting typing status from Redis: " + (e as Error).message, }); } };