feat: update PostgreSQL image to version 18-bookworm and adjust volume paths; add chat_sse2 router and related schemas; implement AdminChat and GuestChat components

apps/db/dev-docker-compose.yml

@@ -1,7 +1,7 @@
 
 services:
   db:
-    image: postgres:16
+    image: postgres:18-bookworm
     environment:
       POSTGRES_USER: ${POSTGRES_USER}
       POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
@@ -12,7 +12,7 @@ services:
     ports:
       - "5433:5432"
     volumes:
-      - dbdata:/var/lib/postgresql/data
+      - dbdata_v18:/var/lib/postgresql
     healthcheck:
       test: ["CMD-SHELL", "pg_isready -d $${POSTGRES_DB} -U $${POSTGRES_USER}"]
       interval: 1s
@@ -42,4 +42,4 @@ networks:
     external: true
       
 volumes:
-  dbdata:
+  dbdata_v18:

apps/infoalloggi/.vscode/settings.json

@@ -3,6 +3,7 @@
 		"source.organizeImports.biome": "explicit",
 		"source.fixAll.biome": "explicit"
 	},
+	"files.autoSave": "off",
 	"editor.formatOnPaste": false,
 	"editor.formatOnType": false,
 	"editor.formatOnSave": true,

apps/infoalloggi/TODO

@@ -1,5 +1,7 @@
 TODOS:
   - Log email fallite e retry
+  - refresh token rotation
+  - Annunci compatibili filtri
 
 AFTER MVP:
   - TODO migrazione app router https://nextjs.org/docs/pages/building-your-application/upgrading/app-router-migration#migrating-from-pages-to-app

apps/infoalloggi/src/pages/test/chat-admin.tsx

@@ -0,0 +1,277 @@
+import { useEffect, useRef, useState } from "react";
+import { useSession } from "~/providers/SessionProvider";
+import type { ChatsChatid } from "~/schemas/public/Chats";
+import type {
+	ChatEvent,
+	Message,
+	SidebarUpdate,
+} from "~/server/api/routers/chat_sse2";
+import type { Session } from "~/server/api/trpc";
+import { MessageDataSchema } from "~/server/utils/zod_types";
+import { api } from "~/utils/api";
+
+const TEST_CHAT_ID = "123e4567-e89b-12d3-a456-426614174000";
+
+export default function AdminChat() {
+	const { status, user } = useSession();
+	if (status === "LOADING") return <div>Loading...</div>;
+	if (status !== "AUTHENTICATED") return null;
+	return <Content session={user} />;
+}
+
+const Content = ({ session }: { session: Session }) => {
+	const [selectedChat, setSelectedChat] = useState<ChatsChatid | null>(null);
+	const [messages, setMessages] = useState<Message[]>([]);
+	const [inputText, setInputText] = useState("");
+
+	const [sidebarEvents, setSidebarEvents] = useState<SidebarUpdate[]>([]);
+	const [chatEvents, setChatEvents] = useState<ChatEvent[]>([]);
+	const [activeTypers, setActiveTypers] = useState<string[]>([]);
+
+	const typingTimeoutRef = useRef<NodeJS.Timeout | null>(null);
+	const { data: availableChats } = api.chatSSE2.listChats.useQuery(undefined, {
+		refetchOnWindowFocus: false,
+	});
+	// 1. Sidebar Subscription (Global)
+	api.chatSSE2.onSidebarUpdate.useSubscription(undefined, {
+		onData(update: SidebarUpdate) {
+			setSidebarEvents((prev) => [update, ...prev].slice(0, 5));
+		},
+	});
+
+	// 2. Fetch History (Only runs when a chat is selected)
+	const { data: chatData, isSuccess } = api.chatSSE2.getChat.useQuery(
+		{ chatId: selectedChat },
+		{
+			enabled: !!selectedChat,
+			refetchOnWindowFocus: false,
+		},
+	);
+
+	useEffect(() => {
+		if (selectedChat && isSuccess && chatData) {
+			setMessages(chatData);
+		}
+	}, [selectedChat, isSuccess, chatData]);
+
+	// 3. Chat Specific Subscription
+	api.chatSSE2.onChatUpdate.useSubscription(
+		{ chatId: selectedChat, username: session.username },
+		{
+			enabled: !!selectedChat,
+			onData(update: ChatEvent) {
+				setChatEvents((prev) => [update, ...prev].slice(0, 5));
+
+				if (update.eventType === "message") {
+					const parsedMessage = MessageDataSchema.safeParse(update.data);
+					if (parsedMessage.success) {
+						setMessages((prev) => [...prev, parsedMessage.data]);
+					} else {
+						console.warn("Received invalid message data:", parsedMessage.error);
+					}
+				}
+
+				if (update.eventType === "typing_update")
+					setActiveTypers(
+						update.data.filter((n: string) => n !== session.username),
+					);
+			},
+		},
+	);
+
+	const sendMessage = api.chatSSE2.sendMessage.useMutation();
+	const toggleTyping = api.chatSSE2.toggleTyping.useMutation();
+
+	const handleSend = (e: React.FormEvent) => {
+		e.preventDefault();
+		if (!inputText || !selectedChat) return;
+
+		sendMessage.mutate({
+			chatId: selectedChat,
+			message: inputText,
+			sender: session.id, // Prod: Admin ID
+		});
+
+		setInputText("");
+		if (typingTimeoutRef.current) clearTimeout(typingTimeoutRef.current);
+		toggleTyping.mutate({
+			chatId: selectedChat,
+			username: session.username,
+			isTyping: false,
+		});
+	};
+
+	const handleInputChange = (e: React.ChangeEvent<HTMLInputElement>) => {
+		setInputText(e.target.value);
+		if (!selectedChat) return;
+
+		toggleTyping.mutate({
+			chatId: selectedChat,
+			username: session.username,
+			isTyping: true,
+		});
+		if (typingTimeoutRef.current) clearTimeout(typingTimeoutRef.current);
+		typingTimeoutRef.current = setTimeout(() => {
+			toggleTyping.mutate({
+				chatId: selectedChat,
+				username: session.username,
+				isTyping: false,
+			});
+		}, 2000);
+	};
+
+	return (
+		<div
+			style={{
+				display: "flex",
+				gap: "20px",
+				padding: "20px",
+				fontFamily: "sans-serif",
+				height: "90vh",
+			}}
+		>
+			{/* Sidebar View */}
+			<div
+				style={{
+					width: "250px",
+					borderRight: "1px solid #ccc",
+					paddingRight: "20px",
+				}}
+			>
+				<h2>Admin Inbox</h2>
+
+				{availableChats?.length === 0 && <div>No active chats</div>}
+				{availableChats?.map((chat) => (
+					<button
+						key={chat.chatid}
+						onClick={() => setSelectedChat(chat.chatid)}
+						style={{
+							width: "100%",
+							padding: "10px",
+							backgroundColor:
+								selectedChat === chat.chatid ? "#e0f7fa" : "#fff",
+						}}
+						type="button"
+					>
+						{chat.chatid}
+					</button>
+				))}
+
+				<div
+					style={{
+						marginTop: "40px",
+						backgroundColor: "#f9f9f9",
+						padding: "10px",
+						fontSize: "12px",
+					}}
+				>
+					<h4>Global Sidebar Events:</h4>
+					{sidebarEvents.map((ev, i) => (
+						<div
+							// biome-ignore lint/suspicious/noArrayIndexKey: <ok>
+							key={i}
+							style={{ borderBottom: "1px solid #ddd", marginBottom: "5px" }}
+						>
+							<strong>{ev.senderName}:</strong> {ev.lastMessage}
+						</div>
+					))}
+				</div>
+			</div>
+
+			{/* Main Chat View */}
+			<div style={{ flex: 1, display: "flex", flexDirection: "column" }}>
+				{!selectedChat ? (
+					<div style={{ margin: "auto", color: "gray" }}>
+						Select a chat to begin
+					</div>
+				) : (
+					<>
+						<div
+							style={{
+								display: "flex",
+								justifyContent: "space-between",
+								alignItems: "center",
+							}}
+						>
+							<h3>Chatting with: {TEST_CHAT_ID.split("-")[0]}...</h3>
+						</div>
+
+						<div
+							style={{
+								flex: 1,
+								overflowY: "auto",
+								border: "1px solid #eee",
+								padding: "10px",
+								marginBottom: "10px",
+							}}
+						>
+							{messages.map((m) => (
+								<div
+									key={m.messageid}
+									style={{
+										margin: "5px 0",
+										textAlign: m.sender_isAdmin ? "right" : "left",
+									}}
+								>
+									<span
+										style={{
+											backgroundColor: m.sender_isAdmin ? "#007bff" : "#e9ecef",
+											color: m.sender_isAdmin ? "white" : "black",
+											padding: "5px 10px",
+											borderRadius: "15px",
+											display: "inline-block",
+										}}
+									>
+										{m.message}
+									</span>
+								</div>
+							))}
+							{activeTypers.length > 0 && (
+								<div
+									style={{
+										color: "gray",
+										fontSize: "0.8em",
+										fontStyle: "italic",
+									}}
+								>
+									{activeTypers.join(", ")} is typing...
+								</div>
+							)}
+						</div>
+
+						<form
+							onSubmit={handleSend}
+							style={{ display: "flex", gap: "10px" }}
+						>
+							<input
+								onChange={handleInputChange}
+								placeholder="Admin reply..."
+								style={{ flex: 1, padding: "8px" }}
+								value={inputText}
+							/>
+							<button type="submit">Reply</button>
+						</form>
+					</>
+				)}
+			</div>
+			<div
+				style={{
+					width: "380px",
+					marginTop: "20px",
+					backgroundColor: "#1e1e1e",
+					color: "#00ff00",
+					padding: "10px",
+					borderRadius: "8px",
+				}}
+			>
+				<h4>Raw Chat Events (KeyDB)</h4>
+				<button onClick={() => setChatEvents([])} type="button">
+					Clear
+				</button>
+				<pre style={{ fontSize: "11px" }}>
+					{JSON.stringify(chatEvents, null, 2)}
+				</pre>
+			</div>
+		</div>
+	);
+};

apps/infoalloggi/src/pages/test/chat.tsx

@@ -0,0 +1,196 @@
+import { useEffect, useRef, useState } from "react";
+import { useSession } from "~/providers/SessionProvider";
+import type { ChatsChatid } from "~/schemas/public/Chats";
+import type { ChatEvent, Message } from "~/server/api/routers/chat_sse2";
+import type { Session } from "~/server/api/trpc";
+import { MessageDataSchema } from "~/server/utils/zod_types";
+import { api } from "~/utils/api";
+
+// Hardcoded for testing the connection
+const TEST_CHAT_ID = "85b6c0fb-2997-4b9a-9cac-ebfc1886927b" as ChatsChatid;
+
+export default function GuestChat() {
+	const { status, user } = useSession();
+	if (status === "LOADING") return <div>Loading...</div>;
+	if (status !== "AUTHENTICATED") return null;
+	return <Content session={user} />;
+}
+
+const Content = ({ session }: { session: Session }) => {
+	const [messages, setMessages] = useState<Message[]>([]);
+	const [inputText, setInputText] = useState("");
+	const [activeTypers, setActiveTypers] = useState<string[]>([]);
+	const [debugLog, setDebugLog] = useState<ChatEvent[]>([]);
+
+	const typingTimeoutRef = useRef<NodeJS.Timeout | null>(null);
+	const { data: historyData, isSuccess } = api.chatSSE2.getChat.useQuery(
+		{ chatId: TEST_CHAT_ID },
+		{
+			refetchOnWindowFocus: false,
+		},
+	);
+
+	useEffect(() => {
+		if (isSuccess && historyData) {
+			setMessages(historyData);
+		}
+	}, [isSuccess, historyData]);
+
+	// 2. Mutations
+	const sendMessage = api.chatSSE2.sendMessage.useMutation();
+	const toggleTyping = api.chatSSE2.toggleTyping.useMutation();
+
+	// 3. The Live Subscription
+	api.chatSSE2.onChatUpdate.useSubscription(
+		{ chatId: TEST_CHAT_ID, username: session.username },
+		{
+			onData(update: ChatEvent) {
+				// Log raw event for debugging
+				setDebugLog((prev) => [update, ...prev].slice(0, 10));
+
+				if (update.eventType === "message") {
+					const parsedMessage = MessageDataSchema.safeParse(update.data);
+					if (parsedMessage.success) {
+						setMessages((prev) => [...prev, parsedMessage.data]);
+					} else {
+						console.warn("Received invalid message data:", parsedMessage.error);
+					}
+				}
+				if (update.eventType === "typing_update") {
+					setActiveTypers(
+						update.data.filter((name: string) => name !== session.username),
+					);
+				}
+			},
+			onError(err) {
+				console.error("Subscription Error:", err);
+			},
+		},
+	);
+
+	// Typing logic (Heartbeat)
+	const handleInputChange = (e: React.ChangeEvent<HTMLInputElement>) => {
+		setInputText(e.target.value);
+
+		// Ping typing status
+		toggleTyping.mutate({
+			chatId: TEST_CHAT_ID,
+			username: session.username,
+			isTyping: true,
+		});
+
+		// Clear previous timeout
+		if (typingTimeoutRef.current) clearTimeout(typingTimeoutRef.current);
+
+		// If user stops typing for 2 seconds, send isTyping: false
+		typingTimeoutRef.current = setTimeout(() => {
+			toggleTyping.mutate({
+				chatId: TEST_CHAT_ID,
+				username: session.username,
+				isTyping: false,
+			});
+		}, 2000);
+	};
+
+	const handleSend = (e: React.FormEvent) => {
+		e.preventDefault();
+		if (!inputText) return;
+
+		sendMessage.mutate({
+			chatId: TEST_CHAT_ID,
+			message: inputText,
+			sender: session.id,
+		});
+
+		setInputText("");
+		if (typingTimeoutRef.current) clearTimeout(typingTimeoutRef.current);
+		toggleTyping.mutate({
+			chatId: TEST_CHAT_ID,
+			username: session.username,
+			isTyping: false,
+		});
+	};
+
+	return (
+		<div
+			style={{
+				display: "flex",
+				gap: "20px",
+				padding: "20px",
+				fontFamily: "sans-serif",
+			}}
+		>
+			{/* UI Panel */}
+			<div
+				style={{
+					flex: 1,
+					border: "1px solid #ccc",
+					padding: "20px",
+					borderRadius: "8px",
+				}}
+			>
+				<h2>Guest View (Logged in as: {session.username})</h2>
+				<div
+					style={{
+						height: "80vh",
+						overflowY: "auto",
+						border: "1px solid #eee",
+						padding: "10px",
+						marginBottom: "10px",
+					}}
+				>
+					{messages.map((m) => (
+						<div
+							key={m.messageid}
+							style={{
+								margin: "5px 0",
+								color: m.sender_isAdmin ? "blue" : "black",
+							}}
+						>
+							<strong>{m.sender_nome || "You"}:</strong> {m.message}
+						</div>
+					))}
+					{activeTypers.length > 0 && (
+						<div
+							style={{ color: "gray", fontSize: "0.8em", fontStyle: "italic" }}
+						>
+							{activeTypers.join(", ")}{" "}
+							{activeTypers.length === 1 ? "is" : "are"} typing...
+						</div>
+					)}
+				</div>
+
+				<form onSubmit={handleSend} style={{ display: "flex", gap: "10px" }}>
+					<input
+						onChange={handleInputChange}
+						placeholder="Type a message..."
+						style={{ flex: 1, padding: "8px" }}
+						value={inputText}
+					/>
+					<button type="submit">Send</button>
+				</form>
+			</div>
+
+			<div
+				style={{
+					flex: 1,
+					backgroundColor: "#1e1e1e",
+					color: "#00ff00",
+					padding: "20px",
+					borderRadius: "8px",
+					overflow: "hidden",
+				}}
+			>
+				<h3>📡 KeyDB Debug Panel</h3>
+				<button onClick={() => setDebugLog([])} type="button">
+					Clear
+				</button>
+				<hr style={{ borderColor: "#333" }} />
+				<h4>Last 10 Events:</h4>
+				<pre style={{ fontSize: "12px", whiteSpace: "pre-wrap" }}>
+					{JSON.stringify(debugLog, null, 2)}
+				</pre>
+			</div>
+		</div>
+	);
+};

apps/infoalloggi/src/schemas/public/PublicSchema.ts

@@ -22,8 +22,8 @@ import type { default as RatelimiterTable } from './Ratelimiter';
 import type { default as ChatsTable } from './Chats';
 import type { default as EtichetteTable } from './Etichette';
 import type { default as MessagesTable } from './Messages';
-import type { default as ProvincieTable } from './Provincie';
 import type { default as UsersAnagraficaTable } from './UsersAnagrafica';
+import type { default as ProvincieTable } from './Provincie';
 import type { default as PotenzialiTable } from './Potenziali';
 import type { default as VideosRefsTable } from './VideosRefs';
 import type { default as BanlistTable } from './Banlist';
@@ -75,10 +75,10 @@ export default interface PublicSchema {
 
   messages: MessagesTable;
 
-  provincie: ProvincieTable;
-
   users_anagrafica: UsersAnagraficaTable;
 
+  provincie: ProvincieTable;
+
   potenziali: PotenzialiTable;
 
   videos_refs: VideosRefsTable;

apps/infoalloggi/src/server/api/root.ts

@@ -3,6 +3,7 @@ import { authRouter } from "~/server/api/routers/auth";
 import { banlistRouter } from "~/server/api/routers/banlist";
 import { chatRouter } from "~/server/api/routers/chat";
 import { chatSSERouter } from "~/server/api/routers/chat_sse";
+import { chatSSE2Router } from "~/server/api/routers/chat_sse2";
 import { comunicazioniRouter } from "~/server/api/routers/comunicazioni";
 import { eventQueueRouter } from "~/server/api/routers/event_queue";
 import { fattureRouter } from "~/server/api/routers/fatture";
@@ -35,6 +36,7 @@ export const appRouter = createTRPCRouter({
 	banlist: banlistRouter,
 	chat: chatRouter,
 	chatSSE: chatSSERouter,
+	chatSSE2: chatSSE2Router,
 	comunicazioni: comunicazioniRouter,
 	eventQueue: eventQueueRouter,
 	fatture: fattureRouter,

apps/infoalloggi/src/server/api/routers/chat_sse2.ts

@@ -0,0 +1,246 @@
+import { on } from "node:events";
+import { TRPCError } from "@trpc/server";
+import { z } from "zod/v4";
+import {
+	createTRPCRouter,
+	protectedProcedure,
+	publicProcedure,
+} from "~/server/api/trpc";
+import { db } from "~/server/db";
+import { add } from "~/server/services/chat.service";
+import { zAsyncIterable } from "~/server/utils/zAsyncIterable";
+import {
+	ChatEventSchema,
+	type MessageDataSchema,
+	type MessageEventSchema,
+	SidebarUpdateSchema,
+	zChatId,
+	zUserId,
+} from "~/server/utils/zod_types";
+import { getKeydbClient, getSubClient } from "~/utils/keydb";
+
+const getChatChannel = (chatId: string) => `chat:${chatId}`;
+const getTypingKey = (chatId: string) => `typing:${chatId}`;
+const SIDEBAR_CHANNEL = "sidebar_updates";
+
+export type SidebarUpdate = z.infer<typeof SidebarUpdateSchema>;
+export type Message = z.infer<typeof MessageDataSchema>;
+type MessageEvent = z.infer<typeof MessageEventSchema>;
+export type ChatEvent = z.infer<typeof ChatEventSchema>;
+
+export const chatSSE2Router = createTRPCRouter({
+	listChats: protectedProcedure.query(async () => {
+		return await db.selectFrom("chats").selectAll().execute();
+	}),
+	createChat: protectedProcedure
+		.input(z.object({ userId: zUserId }))
+		.mutation(async ({ input }) => {
+			const newChat = await add({
+				db,
+				userId: input.userId,
+			});
+
+			return newChat.chatid;
+		}),
+	getChat: publicProcedure
+		.input(z.object({ chatId: zChatId.nullable() }))
+		.query(async ({ input }) => {
+			if (!input.chatId) return;
+			const data = await db
+				.selectFrom("messages")
+				.innerJoin("users", "users.id", "messages.sender")
+				.select([
+					"messages.message",
+					"messages.messageid",
+					"messages.time",
+					"users.nome as sender_nome",
+					"users.isAdmin as sender_isAdmin",
+				])
+				.where("messages.chatid", "=", input.chatId)
+				.orderBy("messages.time", "asc")
+				.execute();
+			return data.map((item) => ({
+				...item,
+				time: item.time.toISOString(),
+			}));
+		}),
+	sendMessage: protectedProcedure
+		.input(
+			z.object({
+				chatId: zChatId,
+				message: z.string(),
+				sender: zUserId,
+			}),
+		)
+		.mutation(async ({ input }) => {
+			const newMessage = await db
+				.insertInto("messages")
+				.values({
+					chatid: input.chatId,
+					message: input.message,
+					sender: input.sender,
+					time: new Date(),
+				})
+				.returningAll()
+				.executeTakeFirstOrThrow(
+					() => new Error("Failed to insert message into database"),
+				);
+
+			const userInfos = await db
+				.selectFrom("users")
+				.select(["nome as sender_nome", "isAdmin as sender_isAdmin"])
+				.where("id", "=", input.sender)
+				.executeTakeFirst();
+
+			if (!userInfos) {
+				throw new TRPCError({
+					code: "NOT_FOUND",
+					message: "User not found or no name",
+				});
+			}
+			const payload: MessageEvent = {
+				data: {
+					...newMessage,
+					...userInfos,
+					time: newMessage.time.toISOString(),
+				},
+				eventType: "message",
+			};
+
+			const kdb = getKeydbClient();
+			if (kdb) {
+				await kdb.publish(
+					getChatChannel(input.chatId),
+					JSON.stringify(payload),
+				);
+
+				await kdb.zrem(getTypingKey(input.chatId), userInfos.sender_nome);
+				const typingList = await kdb.zrange(getTypingKey(input.chatId), 0, -1);
+
+				await kdb.publish(
+					getChatChannel(input.chatId),
+					JSON.stringify({
+						eventType: "typing_update",
+						data: typingList,
+					}),
+				);
+				const sidebarPayload: SidebarUpdate = {
+					chatId: input.chatId,
+					lastMessage: input.message,
+					senderName: userInfos.sender_nome,
+				};
+				await kdb.publish(SIDEBAR_CHANNEL, JSON.stringify(sidebarPayload));
+			}
+
+			return payload;
+		}),
+	onSidebarUpdate: publicProcedure
+		.output(
+			zAsyncIterable({
+				yield: SidebarUpdateSchema,
+				tracked: false,
+			}),
+		)
+		.subscription(async function* () {
+			const sub = getSubClient();
+			if (!sub) return;
+			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));
+					if (parsed.success) {
+						yield parsed.data;
+					} else {
+						console.warn("Received invalid message event:", parsed.error);
+					}
+				}
+			} finally {
+				await sub.unsubscribe(SIDEBAR_CHANNEL);
+				await sub.quit();
+			}
+		}),
+
+	toggleTyping: publicProcedure
+		.input(
+			z.object({
+				chatId: zChatId,
+				isTyping: z.boolean(),
+				username: z.string(),
+			}),
+		)
+		.mutation(async ({ input }) => {
+			const kdb = getKeydbClient();
+			if (!kdb) return;
+
+			const key = getTypingKey(input.chatId);
+			const now = Date.now();
+
+			if (input.isTyping) {
+				await kdb.zadd(key, now, input.username);
+			} else {
+				await kdb.zrem(key, input.username);
+			}
+
+			// Cleanup: Remove users who haven't pinged in 6 seconds
+			await kdb.zremrangebyscore(key, "-inf", now - 6000);
+
+			const typingList = await kdb.zrange(key, 0, -1);
+
+			await kdb.publish(
+				getChatChannel(input.chatId),
+				JSON.stringify({
+					eventType: "typing_update",
+					data: typingList,
+				}),
+			);
+
+			return { success: true };
+		}),
+	onChatUpdate: publicProcedure
+		.input(z.object({ chatId: zChatId.nullable(), username: z.string() }))
+		.output(
+			zAsyncIterable({
+				yield: ChatEventSchema,
+				tracked: false,
+			}),
+		)
+		.subscription(async function* ({ input, signal }) {
+			if (!input.chatId) return;
+			const sub = getSubClient();
+			const kdb = getKeydbClient();
+			if (!sub || !kdb) return; // Build-time safety
+
+			const channel = getChatChannel(input.chatId);
+			const typingKey = getTypingKey(input.chatId);
+
+			try {
+				// ON CONNECT
+				await sub.subscribe(channel);
+				const iterable = on(sub, "message");
+
+				for await (const [chan, message] of iterable) {
+					if (chan === channel) {
+						if (signal?.aborted) break; // Extra safety check
+						const parsed = ChatEventSchema.safeParse(JSON.parse(message));
+						if (parsed.success) {
+							yield parsed.data;
+						} else {
+							console.warn("Received invalid chat event:", parsed.error);
+						}
+					}
+				}
+			} finally {
+				// ON DISCONNECT (Tab closed, refreshed, or network dropped)
+				await kdb.zrem(typingKey, input.username);
+				const remainingT = await kdb.zrange(typingKey, 0, -1);
+				await kdb.publish(
+					channel,
+					JSON.stringify({ eventType: "typing_update", data: remainingT }),
+				);
+
+				await sub.unsubscribe(channel);
+				await sub.quit();
+			}
+		}),
+});

apps/infoalloggi/src/server/utils/zAsyncIterable.ts

@@ -0,0 +1,81 @@
+import type { TrackedEnvelope } from "@trpc/server";
+import { isTrackedEnvelope, tracked } from "@trpc/server";
+import { z } from "zod";
+
+function isAsyncIterable<TValue, TReturn = unknown>(
+	value: unknown,
+): value is AsyncIterable<TValue, TReturn> {
+	return !!value && typeof value === "object" && Symbol.asyncIterator in value;
+}
+const trackedEnvelopeSchema =
+	z.custom<TrackedEnvelope<unknown>>(isTrackedEnvelope);
+
+/**
+ * A Zod schema helper designed specifically for validating async iterables. This schema ensures that:
+ * 1. The value being validated is an async iterable.
+ * 2. Each item yielded by the async iterable conforms to a specified type.
+ * 3. The return value of the async iterable, if any, also conforms to a specified type.
+ */
+export function zAsyncIterable<
+	TYieldIn,
+	TYieldOut,
+	TReturnIn = void,
+	TReturnOut = void,
+	Tracked extends boolean = false,
+>(opts: {
+	/**
+	 * Validate the value yielded by the async generator
+	 */
+	yield: z.ZodType<TYieldOut, TYieldIn>;
+	/**
+	 * Validate the return value of the async generator
+	 * @remarks not applicable for subscriptions
+	 */
+	return?: z.ZodType<TReturnOut, TReturnIn>;
+	/**
+	 * Whether the yielded values are tracked
+	 * @remarks only applicable for subscriptions
+	 */
+	tracked?: Tracked;
+}) {
+	return z
+		.custom<
+			AsyncIterable<
+				Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,
+				TReturnIn
+			>
+		>((val) => isAsyncIterable(val))
+		.transform(async function* (iter) {
+			const iterator = iter[Symbol.asyncIterator]();
+			type TYield = Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn;
+			try {
+				let next: IteratorResult<TYield, TReturnIn>;
+				// biome-ignore lint/suspicious/noAssignInExpressions: <ok>
+				while ((next = await iterator.next()) && !next.done) {
+					if (opts.tracked) {
+						const [id, data] = trackedEnvelopeSchema.parse(next.value);
+						yield tracked(id, await opts.yield.parseAsync(data));
+						continue;
+					}
+					yield opts.yield.parseAsync(next.value);
+				}
+				if (opts.return) {
+					return await opts.return.parseAsync(next.value);
+				}
+				return;
+			} finally {
+				await iterator.return?.();
+			}
+		}) as z.ZodType<
+		AsyncIterable<
+			Tracked extends true ? TrackedEnvelope<TYieldIn> : TYieldIn,
+			TReturnIn,
+			unknown
+		>,
+		AsyncIterable<
+			Tracked extends true ? TrackedEnvelope<TYieldOut> : TYieldOut,
+			TReturnOut,
+			unknown
+		>
+	>;
+}

apps/infoalloggi/src/server/utils/zod_types.ts

@@ -97,3 +97,32 @@ export const zEventId = z.custom<EventQueueEventId>(
 	(val) => typeof val === "number",
 	"falied to validate EventId",
 );
+
+export const MessageDataSchema = z.object({
+	messageid: z.custom<MessagesMessageid>((val) => typeof val === "string"),
+	message: z.string().nullable(),
+	time: z.string(),
+	sender_nome: z.string(),
+	sender_isAdmin: z.boolean(),
+});
+export const MessageEventSchema = z.object({
+	eventType: z.literal("message"),
+	data: MessageDataSchema,
+});
+
+const TypingUpdateEventSchema = z.object({
+	eventType: z.literal("typing_update"),
+	data: z.array(z.string()), // List of users currently typing
+});
+
+export const ChatEventSchema = z.discriminatedUnion("eventType", [
+	MessageEventSchema,
+	TypingUpdateEventSchema,
+	
+]);
+
+export const SidebarUpdateSchema = z.object({
+	chatId: zChatId,
+	lastMessage: z.string(),
+	senderName: z.string(),
+});

apps/infoalloggi/src/utils/keydb.ts

@@ -1,14 +1,19 @@
 import Redis from "ioredis";
 import { env } from "~/env";
 
+const keydbConfig = `keydb://${env.KEYDB_URL}`;
+
 let keydb: Redis | null = null;
 
 export function getKeydbClient() {
-	if (env.SKIP_REDIS === "true") {
+	if (env.SKIP_REDIS === "true") return null;
-		return null;
-	}
 	if (!keydb) {
-		keydb = new Redis(`keydb://${env.KEYDB_URL}`);
+		keydb = new Redis(keydbConfig);
 	}
 	return keydb;
 }
+
+export function getSubClient() {
+	if (env.SKIP_REDIS === "true") return null;
+	return new Redis(keydbConfig, { enableReadyCheck: false });
+}