Merge remote-tracking branch 'upstream/dev' into feat/composio

surfsense_web/hooks/use-chat-session-state.ts

@@ -0,0 +1,39 @@
+"use client";
+
+import { useShape } from "@electric-sql/react";
+import { useSetAtom } from "jotai";
+import { useEffect } from "react";
+import { chatSessionStateAtom } from "@/atoms/chat/chat-session-state.atom";
+import type { ChatSessionState } from "@/contracts/types/chat-session-state.types";
+
+const ELECTRIC_URL = process.env.NEXT_PUBLIC_ELECTRIC_URL || "http://localhost:5133";
+
+/**
+ * Syncs chat session state for a thread via Electric SQL.
+ * Call once per thread (in page.tsx). Updates global atom.
+ */
+export function useChatSessionStateSync(threadId: number | null) {
+	const setSessionState = useSetAtom(chatSessionStateAtom);
+
+	const { data } = useShape<ChatSessionState>({
+		url: `${ELECTRIC_URL}/v1/shape`,
+		params: {
+			table: "chat_session_state",
+			where: `thread_id = ${threadId ?? -1}`,
+		},
+	});
+
+	useEffect(() => {
+		if (!threadId) {
+			setSessionState(null);
+			return;
+		}
+
+		const row = data?.[0];
+		setSessionState({
+			threadId,
+			isAiResponding: !!row?.ai_responding_to_user_id,
+			respondingToUserId: row?.ai_responding_to_user_id ?? null,
+		});
+	}, [threadId, data, setSessionState]);
+}

surfsense_web/hooks/use-comments-electric.ts

@@ -0,0 +1,405 @@
+"use client";
+
+import { useQueryClient } from "@tanstack/react-query";
+import { useAtomValue } from "jotai";
+import { useCallback, useEffect, useMemo, useRef } from "react";
+import { membersAtom, myAccessAtom } from "@/atoms/members/members-query.atoms";
+import { currentUserAtom } from "@/atoms/user/user-query.atoms";
+import type { Author, Comment, CommentReply } from "@/contracts/types/chat-comments.types";
+import type { Membership } from "@/contracts/types/members.types";
+import type { SyncHandle } from "@/lib/electric/client";
+import { useElectricClient } from "@/lib/electric/context";
+import { cacheKeys } from "@/lib/query-client/cache-keys";
+
+// Debounce delay for stream updates (ms)
+const STREAM_UPDATE_DEBOUNCE_MS = 100;
+
+// Raw comment from PGlite local database
+interface RawCommentRow {
+	id: number;
+	message_id: number;
+	thread_id: number;
+	parent_id: number | null;
+	author_id: string | null;
+	content: string;
+	created_at: string;
+	updated_at: string;
+}
+
+// Regex pattern to match @[uuid] mentions (matches backend MENTION_PATTERN)
+const MENTION_PATTERN = /@\[([0-9a-fA-F-]{36})\]/g;
+
+type MemberInfo = Pick<Membership, "user_display_name" | "user_avatar_url" | "user_email">;
+
+/**
+ * Render mentions in content by replacing @[uuid] with @{DisplayName}
+ */
+function renderMentions(content: string, memberMap: Map<string, MemberInfo>): string {
+	return content.replace(MENTION_PATTERN, (match, uuid) => {
+		const member = memberMap.get(uuid);
+		if (member?.user_display_name) {
+			return `@{${member.user_display_name}}`;
+		}
+		return match;
+	});
+}
+
+/**
+ * Build member lookup map from membersData
+ */
+function buildMemberMap(membersData: Membership[] | undefined): Map<string, MemberInfo> {
+	const map = new Map<string, MemberInfo>();
+	if (membersData) {
+		for (const m of membersData) {
+			map.set(m.user_id, {
+				user_display_name: m.user_display_name,
+				user_avatar_url: m.user_avatar_url,
+				user_email: m.user_email,
+			});
+		}
+	}
+	return map;
+}
+
+/**
+ * Build author object from member data
+ */
+function buildAuthor(authorId: string | null, memberMap: Map<string, MemberInfo>): Author | null {
+	if (!authorId) return null;
+	const m = memberMap.get(authorId);
+	if (!m) return null;
+	return {
+		id: authorId,
+		display_name: m.user_display_name ?? null,
+		avatar_url: m.user_avatar_url ?? null,
+		email: m.user_email ?? "",
+	};
+}
+
+/**
+ * Check if a comment has been edited by comparing timestamps.
+ * Uses a small threshold to handle precision differences.
+ */
+function isEdited(createdAt: string, updatedAt: string): boolean {
+	const created = new Date(createdAt).getTime();
+	const updated = new Date(updatedAt).getTime();
+	// Consider edited if updated_at is more than 1 second after created_at
+	return updated - created > 1000;
+}
+
+/**
+ * Transform raw comment to CommentReply
+ */
+function transformReply(
+	raw: RawCommentRow,
+	memberMap: Map<string, MemberInfo>,
+	currentUserId: string | undefined,
+	isOwner: boolean
+): CommentReply {
+	return {
+		id: raw.id,
+		content: raw.content,
+		content_rendered: renderMentions(raw.content, memberMap),
+		author: buildAuthor(raw.author_id, memberMap),
+		created_at: raw.created_at,
+		updated_at: raw.updated_at,
+		is_edited: isEdited(raw.created_at, raw.updated_at),
+		can_edit: currentUserId === raw.author_id,
+		can_delete: currentUserId === raw.author_id || isOwner,
+	};
+}
+
+/**
+ * Transform raw comments to Comment with replies
+ */
+function transformComments(
+	rawComments: RawCommentRow[],
+	memberMap: Map<string, MemberInfo>,
+	currentUserId: string | undefined,
+	isOwner: boolean
+): Map<number, Comment[]> {
+	// Group comments by message_id
+	const byMessage = new Map<
+		number,
+		{ topLevel: RawCommentRow[]; replies: Map<number, RawCommentRow[]> }
+	>();
+
+	for (const raw of rawComments) {
+		if (!byMessage.has(raw.message_id)) {
+			byMessage.set(raw.message_id, { topLevel: [], replies: new Map() });
+		}
+		const group = byMessage.get(raw.message_id)!;
+
+		if (raw.parent_id === null) {
+			group.topLevel.push(raw);
+		} else {
+			if (!group.replies.has(raw.parent_id)) {
+				group.replies.set(raw.parent_id, []);
+			}
+			group.replies.get(raw.parent_id)!.push(raw);
+		}
+	}
+
+	// Transform to Comment objects grouped by message_id
+	const result = new Map<number, Comment[]>();
+
+	for (const [messageId, group] of byMessage) {
+		const comments: Comment[] = group.topLevel.map((raw) => {
+			const replies = (group.replies.get(raw.id) || [])
+				.sort((a, b) => new Date(a.created_at).getTime() - new Date(b.created_at).getTime())
+				.map((r) => transformReply(r, memberMap, currentUserId, isOwner));
+
+			return {
+				id: raw.id,
+				message_id: raw.message_id,
+				content: raw.content,
+				content_rendered: renderMentions(raw.content, memberMap),
+				author: buildAuthor(raw.author_id, memberMap),
+				created_at: raw.created_at,
+				updated_at: raw.updated_at,
+				is_edited: isEdited(raw.created_at, raw.updated_at),
+				can_edit: currentUserId === raw.author_id,
+				can_delete: currentUserId === raw.author_id || isOwner,
+				reply_count: replies.length,
+				replies,
+			};
+		});
+
+		// Sort by created_at
+		comments.sort((a, b) => new Date(a.created_at).getTime() - new Date(b.created_at).getTime());
+		result.set(messageId, comments);
+	}
+
+	return result;
+}
+
+/**
+ * Hook for syncing comments with Electric SQL real-time sync.
+ *
+ * Syncs ALL comments for a thread in ONE subscription, then updates
+ * React Query cache for each message. This avoids N subscriptions for N messages.
+ *
+ * @param threadId - The thread ID to sync comments for
+ */
+export function useCommentsElectric(threadId: number | null) {
+	const electricClient = useElectricClient();
+	const queryClient = useQueryClient();
+
+	const { data: membersData } = useAtomValue(membersAtom);
+	const { data: currentUser } = useAtomValue(currentUserAtom);
+	const { data: myAccess } = useAtomValue(myAccessAtom);
+
+	const memberMap = useMemo(() => buildMemberMap(membersData), [membersData]);
+	const currentUserId = currentUser?.id;
+	const isOwner = myAccess?.is_owner ?? false;
+
+	// Use refs for values needed in live query callback to avoid stale closures
+	const memberMapRef = useRef(memberMap);
+	const currentUserIdRef = useRef(currentUserId);
+	const isOwnerRef = useRef(isOwner);
+	const queryClientRef = useRef(queryClient);
+
+	// Keep refs updated
+	useEffect(() => {
+		memberMapRef.current = memberMap;
+		currentUserIdRef.current = currentUserId;
+		isOwnerRef.current = isOwner;
+		queryClientRef.current = queryClient;
+	}, [memberMap, currentUserId, isOwner, queryClient]);
+
+	const syncHandleRef = useRef<SyncHandle | null>(null);
+	const liveQueryRef = useRef<{ unsubscribe: () => void } | null>(null);
+	const syncKeyRef = useRef<string | null>(null);
+	const streamUpdateDebounceRef = useRef<ReturnType<typeof setTimeout> | null>(null);
+
+	// Stable callback that uses refs for fresh values
+	const updateReactQueryCache = useCallback((rows: RawCommentRow[]) => {
+		const commentsByMessage = transformComments(
+			rows,
+			memberMapRef.current,
+			currentUserIdRef.current,
+			isOwnerRef.current
+		);
+
+		for (const [messageId, comments] of commentsByMessage) {
+			const cacheKey = cacheKeys.comments.byMessage(messageId);
+			queryClientRef.current.setQueryData(cacheKey, {
+				comments,
+				total_count: comments.length,
+			});
+		}
+	}, []);
+
+	useEffect(() => {
+		if (!threadId || !electricClient) {
+			return;
+		}
+
+		const syncKey = `comments_${threadId}`;
+		if (syncKeyRef.current === syncKey) {
+			return;
+		}
+
+		// Capture in local variable for use in async functions
+		const client = electricClient;
+
+		let mounted = true;
+		syncKeyRef.current = syncKey;
+
+		async function startSync() {
+			try {
+				const handle = await client.syncShape({
+					table: "chat_comments",
+					where: `thread_id = ${threadId}`,
+					columns: [
+						"id",
+						"message_id",
+						"thread_id",
+						"parent_id",
+						"author_id",
+						"content",
+						"created_at",
+						"updated_at",
+					],
+					primaryKey: ["id"],
+				});
+
+				if (!handle.isUpToDate && handle.initialSyncPromise) {
+					try {
+						await Promise.race([
+							handle.initialSyncPromise,
+							new Promise((resolve) => setTimeout(resolve, 3000)),
+						]);
+					} catch {
+						// Initial sync timeout - continue anyway
+					}
+				}
+
+				if (!mounted) {
+					handle.unsubscribe();
+					return;
+				}
+
+				syncHandleRef.current = handle;
+
+				// Fetch initial comments and update cache
+				await fetchAndUpdateCache();
+
+				// Set up live query for real-time updates
+				await setupLiveQuery();
+
+				// Subscribe to the sync stream for real-time updates from Electric SQL
+				// This ensures we catch updates even if PGlite live query misses them
+				if (handle.stream) {
+					const stream = handle.stream as {
+						subscribe?: (callback: (messages: unknown[]) => void) => void;
+					};
+					if (typeof stream.subscribe === "function") {
+						stream.subscribe((messages: unknown[]) => {
+							if (!mounted) return;
+							// When Electric sync receives new data, refresh from PGlite
+							// This handles cases where live query might miss the update
+							if (messages && messages.length > 0) {
+								// Debounce the refresh to avoid excessive queries
+								if (streamUpdateDebounceRef.current) {
+									clearTimeout(streamUpdateDebounceRef.current);
+								}
+								streamUpdateDebounceRef.current = setTimeout(() => {
+									if (mounted) {
+										fetchAndUpdateCache();
+									}
+								}, STREAM_UPDATE_DEBOUNCE_MS);
+							}
+						});
+					}
+				}
+			} catch {
+				// Sync failed - will retry on next mount
+			}
+		}
+
+		async function fetchAndUpdateCache() {
+			try {
+				const result = await client.db.query<RawCommentRow>(
+					`SELECT id, message_id, thread_id, parent_id, author_id, content, created_at, updated_at 
+					 FROM chat_comments 
+					 WHERE thread_id = $1 
+					 ORDER BY created_at ASC`,
+					[threadId]
+				);
+
+				if (mounted && result.rows) {
+					updateReactQueryCache(result.rows);
+				}
+			} catch {
+				// Query failed - data will be fetched from API
+			}
+		}
+
+		async function setupLiveQuery() {
+			try {
+				// eslint-disable-next-line @typescript-eslint/no-explicit-any
+				const db = client.db as any;
+
+				if (db.live?.query && typeof db.live.query === "function") {
+					const liveQuery = await db.live.query(
+						`SELECT id, message_id, thread_id, parent_id, author_id, content, created_at, updated_at 
+						 FROM chat_comments 
+						 WHERE thread_id = $1 
+						 ORDER BY created_at ASC`,
+						[threadId]
+					);
+
+					if (!mounted) {
+						liveQuery.unsubscribe?.();
+						return;
+					}
+
+					// Set initial results
+					if (liveQuery.initialResults?.rows) {
+						updateReactQueryCache(liveQuery.initialResults.rows);
+					} else if (liveQuery.rows) {
+						updateReactQueryCache(liveQuery.rows);
+					}
+
+					// Subscribe to changes
+					if (typeof liveQuery.subscribe === "function") {
+						liveQuery.subscribe((result: { rows: RawCommentRow[] }) => {
+							if (mounted && result.rows) {
+								updateReactQueryCache(result.rows);
+							}
+						});
+					}
+
+					if (typeof liveQuery.unsubscribe === "function") {
+						liveQueryRef.current = liveQuery;
+					}
+				}
+			} catch {
+				// Live query setup failed - will use initial fetch only
+			}
+		}
+
+		startSync();
+
+		return () => {
+			mounted = false;
+			syncKeyRef.current = null;
+
+			// Clear debounce timeout
+			if (streamUpdateDebounceRef.current) {
+				clearTimeout(streamUpdateDebounceRef.current);
+				streamUpdateDebounceRef.current = null;
+			}
+
+			if (syncHandleRef.current) {
+				syncHandleRef.current.unsubscribe();
+				syncHandleRef.current = null;
+			}
+			if (liveQueryRef.current) {
+				liveQueryRef.current.unsubscribe();
+				liveQueryRef.current = null;
+			}
+		};
+	}, [threadId, electricClient, updateReactQueryCache]);
+}

surfsense_web/hooks/use-messages-electric.ts

@@ -0,0 +1,154 @@
+"use client";
+
+import { useCallback, useEffect, useRef } from "react";
+import type { RawMessage } from "@/contracts/types/chat-messages.types";
+import type { SyncHandle } from "@/lib/electric/client";
+import { useElectricClient } from "@/lib/electric/context";
+
+/**
+ * Syncs chat messages for a thread via Electric SQL.
+ * Calls onMessagesUpdate when messages change.
+ */
+export function useMessagesElectric(
+	threadId: number | null,
+	onMessagesUpdate: (messages: RawMessage[]) => void
+) {
+	const electricClient = useElectricClient();
+
+	const syncHandleRef = useRef<SyncHandle | null>(null);
+	const liveQueryRef = useRef<{ unsubscribe: () => void } | null>(null);
+	const syncKeyRef = useRef<string | null>(null);
+	const onMessagesUpdateRef = useRef(onMessagesUpdate);
+
+	useEffect(() => {
+		onMessagesUpdateRef.current = onMessagesUpdate;
+	}, [onMessagesUpdate]);
+
+	const handleMessagesUpdate = useCallback((rows: RawMessage[]) => {
+		onMessagesUpdateRef.current(rows);
+	}, []);
+
+	useEffect(() => {
+		if (!threadId || !electricClient) {
+			return;
+		}
+
+		const syncKey = `messages_${threadId}`;
+		if (syncKeyRef.current === syncKey) {
+			return;
+		}
+
+		const client = electricClient;
+		let mounted = true;
+		syncKeyRef.current = syncKey;
+
+		async function startSync() {
+			try {
+				const handle = await client.syncShape({
+					table: "new_chat_messages",
+					where: `thread_id = ${threadId}`,
+					columns: ["id", "thread_id", "role", "content", "author_id", "created_at"],
+					primaryKey: ["id"],
+				});
+
+				if (!handle.isUpToDate && handle.initialSyncPromise) {
+					try {
+						await Promise.race([
+							handle.initialSyncPromise,
+							new Promise((resolve) => setTimeout(resolve, 3000)),
+						]);
+					} catch {
+						// Timeout
+					}
+				}
+
+				if (!mounted) {
+					handle.unsubscribe();
+					return;
+				}
+
+				syncHandleRef.current = handle;
+				await fetchMessages();
+				await setupLiveQuery();
+			} catch {
+				// Sync failed
+			}
+		}
+
+		async function fetchMessages() {
+			try {
+				const result = await client.db.query<RawMessage>(
+					`SELECT id, thread_id, role, content, author_id, created_at 
+					 FROM new_chat_messages 
+					 WHERE thread_id = $1 
+					 ORDER BY created_at ASC`,
+					[threadId]
+				);
+
+				if (mounted && result.rows) {
+					handleMessagesUpdate(result.rows);
+				}
+			} catch {
+				// Query failed
+			}
+		}
+
+		async function setupLiveQuery() {
+			try {
+				// eslint-disable-next-line @typescript-eslint/no-explicit-any
+				const db = client.db as any;
+
+				if (db.live?.query && typeof db.live.query === "function") {
+					const liveQuery = await db.live.query(
+						`SELECT id, thread_id, role, content, author_id, created_at 
+						 FROM new_chat_messages 
+						 WHERE thread_id = $1 
+						 ORDER BY created_at ASC`,
+						[threadId]
+					);
+
+					if (!mounted) {
+						liveQuery.unsubscribe?.();
+						return;
+					}
+
+					if (liveQuery.initialResults?.rows) {
+						handleMessagesUpdate(liveQuery.initialResults.rows);
+					} else if (liveQuery.rows) {
+						handleMessagesUpdate(liveQuery.rows);
+					}
+
+					if (typeof liveQuery.subscribe === "function") {
+						liveQuery.subscribe((result: { rows: RawMessage[] }) => {
+							if (mounted && result.rows) {
+								handleMessagesUpdate(result.rows);
+							}
+						});
+					}
+
+					if (typeof liveQuery.unsubscribe === "function") {
+						liveQueryRef.current = liveQuery;
+					}
+				}
+			} catch {
+				// Live query failed
+			}
+		}
+
+		startSync();
+
+		return () => {
+			mounted = false;
+			syncKeyRef.current = null;
+
+			if (syncHandleRef.current) {
+				syncHandleRef.current.unsubscribe();
+				syncHandleRef.current = null;
+			}
+			if (liveQueryRef.current) {
+				liveQueryRef.current.unsubscribe();
+				liveQueryRef.current = null;
+			}
+		};
+	}, [threadId, electricClient, handleMessagesUpdate]);
+}