diff --git a/apps/db/migrations/51_typing.up.sql b/apps/db/migrations/51_typing.up.sql new file mode 100644 index 0000000..aab4a8c --- /dev/null +++ b/apps/db/migrations/51_typing.up.sql @@ -0,0 +1,48 @@ +CREATE TABLE IF NOT EXISTS public.typing_status +( + chatid uuid NOT NULL, + userid uuid NOT NULL, + username text NOT NULL, + last_ping timestamp without time zone NOT NULL DEFAULT now() +); + +DO $$ +BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'typing_status_pkey' + AND conrelid = 'public.typing_status'::regclass + ) THEN + ALTER TABLE public.typing_status + ADD CONSTRAINT typing_status_pkey PRIMARY KEY (chatid,userid); + END IF; +END $$; + +DO $$ BEGIN IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'typing_user' + AND conrelid = 'public.typing_status'::regclass + ) THEN + ALTER TABLE public.typing_status + ADD CONSTRAINT typing_user + FOREIGN KEY (userid) REFERENCES public.users (id) + ON UPDATE CASCADE ON DELETE CASCADE; + END IF; +END $$; + +DO $$ BEGIN IF NOT EXISTS ( + SELECT 1 FROM pg_constraint + WHERE conname = 'typing_chat' + AND conrelid = 'public.typing_status'::regclass + ) THEN + ALTER TABLE public.typing_status + ADD CONSTRAINT typing_chat + FOREIGN KEY (chatid) REFERENCES public.chats (chatid) + ON UPDATE CASCADE ON DELETE CASCADE; + END IF; +END $$; + + + +CREATE INDEX IF NOT EXISTS typing_status_chat_lastping_idx +ON typing_status (chatid, last_ping DESC); \ No newline at end of file diff --git a/apps/infoalloggi/src/schemas/public/TypingStatus.ts b/apps/infoalloggi/src/schemas/public/TypingStatus.ts new file mode 100644 index 0000000..4804735 --- /dev/null +++ b/apps/infoalloggi/src/schemas/public/TypingStatus.ts @@ -0,0 +1,42 @@ +import { chatsChatid, type ChatsChatid } from './Chats'; +import { usersId, type UsersId } from './Users'; +import type { ColumnType, Selectable, Insertable, Updateable } from 'kysely'; +import { z } from 'zod'; + +/** Represents the table public.typing_status */ +export default interface TypingStatusTable { + chatid: ColumnType; + + userid: ColumnType; + + username: ColumnType; + + last_ping: ColumnType; +} + +export type TypingStatus = Selectable; + +export type NewTypingStatus = Insertable; + +export type TypingStatusUpdate = Updateable; + +export const TypingStatusSchema = z.object({ + chatid: chatsChatid, + userid: usersId, + username: z.string(), + last_ping: z.date(), +}); + +export const NewTypingStatusSchema = z.object({ + chatid: chatsChatid, + userid: usersId, + username: z.string(), + last_ping: z.union([z.date(), z.string()]).pipe(z.coerce.date()).optional(), +}); + +export const TypingStatusUpdateSchema = z.object({ + chatid: chatsChatid.optional(), + userid: usersId.optional(), + username: z.string().optional(), + last_ping: z.union([z.date(), z.string()]).pipe(z.coerce.date()).optional(), +}); \ No newline at end of file diff --git a/apps/infoalloggi/src/server/api/routers/chat_sse.ts b/apps/infoalloggi/src/server/api/routers/chat_sse.ts index 5fab203..240f711 100644 --- a/apps/infoalloggi/src/server/api/routers/chat_sse.ts +++ b/apps/infoalloggi/src/server/api/routers/chat_sse.ts @@ -16,7 +16,7 @@ import { protectedProcedure, publicProcedure, } from "~/server/api/trpc"; -import { db } from "~/server/db"; +import { db, pgPool } from "~/server/db"; import { add } from "~/server/services/chat.service"; import { editMessage, @@ -29,15 +29,18 @@ import { SidebarUpdateSchema, TypingDataSchema, } from "~/server/utils/zod_utils"; -import { getKeydbClient, getSubClient } from "~/utils/keydb"; import { withUserColorStr } from "~/utils/kysely-helper"; - -const getChatChannel = (chatId: string) => `chat:${chatId}:messages`; -const getTypingChannel = (chatId: string) => `chat:${chatId}:typing`; -const getEventsChannel = (chatId: string) => `chat:${chatId}:events`; -const getTypingKey = (chatId: string) => `typing:${chatId}`; - -const SIDEBAR_CHANNEL = "sidebar_updates"; +import { + clearTyping, + getChatChannel, + getEventsChannel, + getTypingChannel, + getTypingList, + pgPublish, + pruneStaleTyping, + SIDEBAR_CHANNEL, + setTyping, +} from "../../services/pubsub.service"; type SidebarUpdate = z.infer; export type Message = z.infer; @@ -80,13 +83,10 @@ export const chatSSERouter = createTRPCRouter({ }); if (didUpdate) { - const kdb = getKeydbClient(); - if (kdb) { - await kdb.publish( - getEventsChannel(input.chatId), - JSON.stringify({ eventType: "read_receipt", data: null }), - ); - } + await pgPublish( + getEventsChannel(input.chatId), + JSON.stringify({ eventType: "read_receipt", data: null }), + ); } if (input.targetMsgId) { @@ -338,30 +338,15 @@ export const chatSSERouter = createTRPCRouter({ time: newMessage.time.toISOString(), }; - const kdb = getKeydbClient(); const msgChannel = getChatChannel(input.chatId); - const typingChannel = getTypingChannel(input.chatId); - const typingKey = getTypingKey(input.chatId); - if (kdb) { - await kdb.publish(msgChannel, JSON.stringify(payload)); - const member = JSON.stringify({ - userId: input.sender, - username: userInfos.sender_nome, - }); - await kdb.zrem(typingKey, member); - const typingList = await kdb.zrange(typingKey, 0, -1); - await kdb.publish( - typingChannel, - JSON.stringify(typingList.map((m) => JSON.parse(m))), - ); - const sidebarPayload: SidebarUpdate = { - chatId: input.chatId, - lastMessage: input.message, - senderName: userInfos.sender_nome, - }; - await kdb.publish(SIDEBAR_CHANNEL, JSON.stringify(sidebarPayload)); - } + await pgPublish(msgChannel, JSON.stringify(payload)); + const sidebarPayload: SidebarUpdate = { + chatId: input.chatId, + lastMessage: input.message, + senderName: userInfos.sender_nome, + }; + await pgPublish(SIDEBAR_CHANNEL, JSON.stringify(sidebarPayload)); return payload; }), @@ -374,13 +359,10 @@ export const chatSSERouter = createTRPCRouter({ .where("messageid", "=", input.messageId) .execute(); - const kdb = getKeydbClient(); - if (kdb) { - await kdb.publish( - getEventsChannel(input.chatId), - JSON.stringify({ eventType: "delete", data: input.messageId }), - ); - } + await pgPublish( + getEventsChannel(input.chatId), + JSON.stringify({ eventType: "delete", data: input.messageId }), + ); } catch (err) { console.error("Failed to delete message:", err); throw new TRPCError({ @@ -406,16 +388,13 @@ export const chatSSERouter = createTRPCRouter({ messageId: input.messageId, }); - const kdb = getKeydbClient(); - if (kdb) { - await kdb.publish( - getEventsChannel(input.chatId), - JSON.stringify({ - eventType: "edit", - data: { messageId: input.messageId, message: input.message }, - }), - ); - } + await pgPublish( + getEventsChannel(input.chatId), + JSON.stringify({ + eventType: "edit", + data: { messageId: input.messageId, message: input.message }, + }), + ); } catch (err) { console.error("Failed to edit message:", err); throw new TRPCError({ @@ -434,20 +413,20 @@ export const chatSSERouter = createTRPCRouter({ ) .subscription(async function* ({ input, signal }) { if (!input.chatId) return; - const sub = getSubClient(); - if (!sub) return; // Build-time safety + + const client = await pgPool.connect(); const channel = getEventsChannel(input.chatId); try { - await sub.subscribe(channel); - const iterable = on(sub, "message"); + await client.query(`LISTEN "${channel}"`); + const iterable = on(client, "notification"); - for await (const [chan, message] of iterable) { + for await (const [{ channel: chan, payload }] of iterable) { if (chan === channel) { if (signal?.aborted) break; // Extra safety check const parsed = MessageUpdateEventSchema.safeParse( - JSON.parse(message), + JSON.parse(payload), ); if (parsed.success) { yield parsed.data; @@ -460,8 +439,8 @@ export const chatSSERouter = createTRPCRouter({ } } } finally { - await sub.unsubscribe(channel); - await sub.quit(); + await client.query(`UNLISTEN "${channel}"`); + client.release(); } }), onSidebarUpdate: publicProcedure @@ -472,13 +451,14 @@ export const chatSSERouter = createTRPCRouter({ }), ) .subscription(async function* () { - const sub = getSubClient(); - if (!sub) return; + const client = await pgPool.connect(); try { - await sub.subscribe(SIDEBAR_CHANNEL); - const iterable = on(sub, "message"); - for await (const [_chan, message] of iterable) { - const parsed = SidebarUpdateSchema.safeParse(JSON.parse(message)); + await client.query(`LISTEN "${SIDEBAR_CHANNEL}"`); + + const iterable = on(client, "notification"); + + for await (const [{ payload }] of iterable) { + const parsed = SidebarUpdateSchema.safeParse(JSON.parse(payload)); if (parsed.success) { yield parsed.data; } else { @@ -486,8 +466,8 @@ export const chatSSERouter = createTRPCRouter({ } } } finally { - await sub.unsubscribe(SIDEBAR_CHANNEL); - await sub.quit(); + await client.query(`UNLISTEN "${SIDEBAR_CHANNEL}"`); + client.release(); } }), @@ -501,30 +481,19 @@ export const chatSSERouter = createTRPCRouter({ }), ) .mutation(async ({ input }) => { - const kdb = getKeydbClient(); - if (!kdb) return; + const { chatId, userId, username, isTyping } = input; - const key = getTypingKey(input.chatId); - const now = Date.now(); - const member = JSON.stringify({ - userId: input.userId, - username: input.username, - }); - if (input.isTyping) { - await kdb.zadd(key, now, member); + if (isTyping) { + await setTyping(chatId, userId, username); } else { - await kdb.zrem(key, member); + await clearTyping(chatId, userId); } - // Cleanup: Remove users who haven't pinged in 6 seconds - await kdb.zremrangebyscore(key, "-inf", now - 6000); + await pruneStaleTyping(chatId, 6000); - const typingList = await kdb.zrange(key, 0, -1); + const typingList = await getTypingList(chatId); - await kdb.publish( - getTypingChannel(input.chatId), - JSON.stringify(typingList.map((m) => JSON.parse(m))), - ); + await pgPublish(getTypingChannel(chatId), JSON.stringify(typingList)); return { success: true }; }), @@ -543,15 +512,13 @@ export const chatSSERouter = createTRPCRouter({ ) .subscription(async function* ({ input, signal }) { if (!input.chatId) return; - const sub = getSubClient(); - if (!sub) return; // Build-time safety + const client = await pgPool.connect(); - const channel = getChatChannel(input.chatId); + const msgChan = getChatChannel(input.chatId); try { - // ON CONNECT - await sub.subscribe(channel); - const iterable = on(sub, "message"); + await client.query(`LISTEN "${msgChan}"`); + const iterable = on(client, "notification"); let lastMessageId: MessagesMessageid | null = (() => { @@ -638,10 +605,10 @@ export const chatSSERouter = createTRPCRouter({ yield* maybeYield(msg); } - for await (const [chan, message] of iterable) { - if (chan === channel) { - if (signal?.aborted) break; // Extra safety check - const parsed = MessageDataSchema.safeParse(JSON.parse(message)); + for await (const [{ channel, payload }] of iterable) { + if (channel === msgChan) { + if (signal?.aborted) break; + const parsed = MessageDataSchema.safeParse(JSON.parse(payload)); if (parsed.success) { yield* maybeYield(parsed.data); } else { @@ -650,10 +617,11 @@ export const chatSSERouter = createTRPCRouter({ } } } finally { - await sub.unsubscribe(channel); - await sub.quit(); + await client.query(`UNLISTEN "${msgChan}"`); + client.release(); } }), + onTyping: publicProcedure .input( z.object({ @@ -669,32 +637,28 @@ export const chatSSERouter = createTRPCRouter({ }), ) .subscription(async function* ({ input, signal }) { - if (!input.chatId) return; - const sub = getSubClient(); - const kdb = getKeydbClient(); - if (!sub || !kdb) return; // Build-time safety + const { chatId, userId } = input; + if (!chatId) return; - const channel = getTypingChannel(input.chatId); - const typingKey = getTypingKey(input.chatId); + const client = await pgPool.connect(); + + const channel = getTypingChannel(chatId); try { - // ON CONNECT - await sub.subscribe(channel); - const iterable = on(sub, "message"); + await client.query(`LISTEN "${channel}"`); + const iterable = on(client, "notification"); - const current = await kdb.zrange(typingKey, 0, -1); + const current = await getTypingList(chatId); - const parsed = TypingDataSchema.safeParse( - current.map((m) => JSON.parse(m)), - ); + const parsed = TypingDataSchema.safeParse(current); if (parsed.success) { yield parsed.data; } - for await (const [chan, message] of iterable) { + for await (const [{ channel: chan, payload }] of iterable) { if (chan === channel) { if (signal?.aborted) break; // Extra safety check - const parsed = TypingDataSchema.safeParse(JSON.parse(message)); + const parsed = TypingDataSchema.safeParse(JSON.parse(payload)); if (parsed.success) { yield parsed.data; } else { @@ -703,20 +667,13 @@ export const chatSSERouter = createTRPCRouter({ } } } finally { - // ON DISCONNECT (Tab closed, refreshed, or network dropped) - const member = JSON.stringify({ - userId: input.userId, - username: input.username, - }); - await kdb.zrem(typingKey, member); - const remainingT = await kdb.zrange(typingKey, 0, -1); - await kdb.publish( - channel, - JSON.stringify(remainingT.map((m) => JSON.parse(m))), - ); + await clearTyping(chatId, userId); - await sub.unsubscribe(channel); - await sub.quit(); + const remainingT = await getTypingList(chatId); + + await pgPublish(channel, JSON.stringify(remainingT)); + await client.query(`UNLISTEN "${channel}"`); + client.release(); } }), setRead: protectedProcedure @@ -729,13 +686,10 @@ export const chatSSERouter = createTRPCRouter({ }); if (didUpdate) { - const kdb = getKeydbClient(); - if (kdb) { - await kdb.publish( - getEventsChannel(input.chatId), - JSON.stringify({ eventType: "read_receipt", data: null }), - ); - } + await pgPublish( + getEventsChannel(input.chatId), + JSON.stringify({ eventType: "read_receipt", data: null }), + ); } return true; diff --git a/apps/infoalloggi/src/server/db.ts b/apps/infoalloggi/src/server/db.ts index e7a145b..6abb34d 100644 --- a/apps/infoalloggi/src/server/db.ts +++ b/apps/infoalloggi/src/server/db.ts @@ -23,7 +23,7 @@ const pool = new Pool({ max: 10, password: env.POSTGRES_PASSWORD, port: Number.parseInt(env.PGPORT), - user: env.POSTGRES_USER, + user: env.POSTGRES_USER }); pool.on("error", (err) => { @@ -82,3 +82,5 @@ const _DbLogger = (event: LogEvent) => { //console.error("\x1b[31merror:\x1b[0m", event.error); } }; + +export const pgPool = pool \ No newline at end of file diff --git a/apps/infoalloggi/src/server/services/pubsub.service.ts b/apps/infoalloggi/src/server/services/pubsub.service.ts new file mode 100644 index 0000000..a4e4fac --- /dev/null +++ b/apps/infoalloggi/src/server/services/pubsub.service.ts @@ -0,0 +1,109 @@ +import { TRPCError } from "@trpc/server"; +import { sql } from "kysely"; +import type { ChatsChatid } from "@/src/schemas/public/Chats"; +import type { UsersId } from "@/src/schemas/public/Users"; +import { db } from "~/server/db"; + +export const getChatChannel = (chatId: string) => `chat:${chatId}:messages`; +export const getTypingChannel = (chatId: string) => `chat:${chatId}:typing`; +export const getEventsChannel = (chatId: string) => `chat:${chatId}:events`; + +export const SIDEBAR_CHANNEL = "sidebar_updates"; +export async function pgPublish(channel: string, payload: string) { + try { + await sql`SELECT pg_notify(${channel}, ${payload})`.execute(db); + } catch (err) { + throw new TRPCError({ + code: "INTERNAL_SERVER_ERROR", + message: `pgPub Error: ${(err as Error).message}`, + }); + } +} + +//pgListen +/* +const client = await pgPool.connect(); + +try { + await client.query(`LISTEN "${msgChan}"`); + + const iterable = on(client, "notification"); + + for await (const [{ channel, payload }] of iterable) { + console.log("Received:", channel, payload); + const message = JSON.parse(payload) as { title?: string }; + + if (channel === msgChan) { + if (signal?.aborted) break; // Extra safety check + yield message.title; + } + } +} finally { + await client.query(`UNLISTEN "${msgChan}"`); + client.release(); +} +*/ + +export async function setTyping( + chatId: ChatsChatid, + userId: UsersId, + username: string, +): Promise { + await db + .insertInto("typing_status") + .values({ + chatid: chatId, + userid: userId, + username, + last_ping: new Date(), + }) + .onConflict((oc) => + oc + .columns(["chatid", "userid"]) + .doUpdateSet({ last_ping: new Date(), username }), + ) + .execute(); +} + +export async function clearTyping( + chatId: ChatsChatid, + userId: UsersId, +): Promise { + await db + .deleteFrom("typing_status") + .where("chatid", "=", chatId) + .where("userid", "=", userId) + .execute(); +} + +/** + * Remove rows that haven’t pinged in the last `maxAgeMs` milliseconds. + */ +export async function pruneStaleTyping( + chatId: ChatsChatid, + maxAgeMs: number, +): Promise { + const cutoff = new Date(Date.now() - maxAgeMs); + await db + .deleteFrom("typing_status") + .where("chatid", "=", chatId) + .where("last_ping", "<", cutoff) + .execute(); +} + +/** + * Get the current typing users for a chat, ordered by most recent ping. + * The return shape mirrors what the Redis sorted‑set previously emitted. + */ +export async function getTypingList( + chatId: ChatsChatid, +): Promise> { + const rows = await db + .selectFrom("typing_status") + .select(["userid as userId", "username"]) + .where("chatid", "=", chatId) + .orderBy("last_ping", "desc") + .execute(); + + return rows; +}