infoalloggi-monorepo/apps/infoalloggi/src/server/controllers/chat.controller.ts

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<ActiveChatsType[]> => {
  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: new Date(chat.lastMessage.time),
              isLastSender: chat.lastMessage.sender === userId,
            }
          : null,
        etichette: chat.etichette as Etichette[],
      };
    })
    .filter((chat): chat is NonNullable<typeof chat> => 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<Chats> => {
  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<WhoIsTyping[]> => {
  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,
    });
  }
};