Marco/infoalloggi-monorepo
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages 6 Wiki Activity Actions

typing

Browse source
This commit is contained in:
Marco Pedone 2026-05-06 14:59:31 +02:00
parent 7eb2e8637a
commit 14343ed611
5 changed files with 292 additions and 137 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

48
apps/db/migrations/51_typing.up.sql Normal file
View file

@ -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);

42
apps/infoalloggi/src/schemas/public/TypingStatus.ts Normal file
View file

@ -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<ChatsChatid, ChatsChatid, ChatsChatid>;
userid: ColumnType<UsersId, UsersId, UsersId>;
username: ColumnType<string, string, string>;
last_ping: ColumnType<Date, Date | string | undefined, Date | string>;
}
export type TypingStatus = Selectable<TypingStatusTable>;
export type NewTypingStatus = Insertable<TypingStatusTable>;
export type TypingStatusUpdate = Updateable<TypingStatusTable>;
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(),
});

186
apps/infoalloggi/src/server/api/routers/chat_sse.ts
View file

@ -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<typeof SidebarUpdateSchema>;
export type Message = z.infer<typeof MessageDataSchema>;
@ -80,14 +83,11 @@ export const chatSSERouter = createTRPCRouter({
});
if (didUpdate) {
const kdb = getKeydbClient();
if (kdb) {
await kdb.publish(
await pgPublish(
getEventsChannel(input.chatId),
JSON.stringify({ eventType: "read_receipt", data: null }),
);
}
}
if (input.targetMsgId) {
// If a targetMsgId is provided, we want to fetch messages around that message
@ -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))),
);
await pgPublish(msgChannel, JSON.stringify(payload));
const sidebarPayload: SidebarUpdate = {
chatId: input.chatId,
lastMessage: input.message,
senderName: userInfos.sender_nome,
};
await kdb.publish(SIDEBAR_CHANNEL, JSON.stringify(sidebarPayload));
}
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(
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(
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,14 +686,11 @@ export const chatSSERouter = createTRPCRouter({
});
if (didUpdate) {
const kdb = getKeydbClient();
if (kdb) {
await kdb.publish(
await pgPublish(
getEventsChannel(input.chatId),
JSON.stringify({ eventType: "read_receipt", data: null }),
);
}
}
return true;
}),

4
apps/infoalloggi/src/server/db.ts
View file

@ -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

109
apps/infoalloggi/src/server/services/pubsub.service.ts Normal file
View file

@ -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<void> {
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<void> {
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<void> {
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<Array<{ userId: UsersId; username: string }>> {
const rows = await db
.selectFrom("typing_status")
.select(["userid as userId", "username"])
.where("chatid", "=", chatId)
.orderBy("last_ping", "desc")
.execute();
return rows;
}