4 ? "pb-8" : ""}`}
@@ -206,7 +226,15 @@ export function Sidebar({
) : undefined
}
>
- {chats.length > 0 ? (
+ {isLoadingChats ? (
+
4 ? "pb-8" : ""}`}
diff --git a/surfsense_web/contracts/enums/connectorIcons.tsx b/surfsense_web/contracts/enums/connectorIcons.tsx
index aaf476215..18a872d94 100644
--- a/surfsense_web/contracts/enums/connectorIcons.tsx
+++ b/surfsense_web/contracts/enums/connectorIcons.tsx
@@ -92,7 +92,7 @@ export const getConnectorIcon = (connectorType: EnumConnectorName | string, clas
case "FILE":
return ;
case "GOOGLE_DRIVE_FILE":
- return ;
+ return ;
case "COMPOSIO_GOOGLE_DRIVE_CONNECTOR":
return ;
case "COMPOSIO_GMAIL_CONNECTOR":
diff --git a/surfsense_web/contracts/types/document.types.ts b/surfsense_web/contracts/types/document.types.ts
index 01a58173e..b7a2d2cf8 100644
--- a/surfsense_web/contracts/types/document.types.ts
+++ b/surfsense_web/contracts/types/document.types.ts
@@ -23,6 +23,7 @@ export const documentTypeEnum = z.enum([
"ELASTICSEARCH_CONNECTOR",
"BOOKSTACK_CONNECTOR",
"CIRCLEBACK",
+ "OBSIDIAN_CONNECTOR",
"SURFSENSE_DOCS",
"NOTE",
"COMPOSIO_GOOGLE_DRIVE_CONNECTOR",
@@ -41,6 +42,8 @@ export const document = z.object({
created_at: z.string(),
updated_at: z.string().nullable(),
search_space_id: z.number(),
+ created_by_id: z.string().nullable().optional(),
+ created_by_name: z.string().nullable().optional(),
});
export const extensionDocumentContent = z.object({
diff --git a/surfsense_web/hooks/use-documents-electric.ts b/surfsense_web/hooks/use-documents-electric.ts
deleted file mode 100644
index 43809499e..000000000
--- a/surfsense_web/hooks/use-documents-electric.ts
+++ /dev/null
@@ -1,185 +0,0 @@
-"use client";
-
-import { useEffect, useMemo, useRef, useState } from "react";
-import type { SyncHandle } from "@/lib/electric/client";
-import { useElectricClient } from "@/lib/electric/context";
-
-interface Document {
- id: number;
- search_space_id: number;
- document_type: string;
- created_at: string;
-}
-
-/**
- * Hook for managing documents with Electric SQL real-time sync
- *
- * Uses the Electric client from context (provided by ElectricProvider)
- * instead of initializing its own - prevents race conditions and memory leaks
- */
-export function useDocumentsElectric(searchSpaceId: number | string | null) {
- // Get Electric client from context - ElectricProvider handles initialization
- const electricClient = useElectricClient();
-
- const [documents, setDocuments] = useState([]);
- const [loading, setLoading] = useState(true);
- const [error, setError] = useState(null);
- const syncHandleRef = useRef(null);
- const liveQueryRef = useRef<{ unsubscribe: () => void } | null>(null);
- const syncKeyRef = useRef(null);
-
- // Calculate document type counts from synced documents
- const documentTypeCounts = useMemo(() => {
- if (!documents.length) return {};
-
- const counts: Record = {};
- for (const doc of documents) {
- counts[doc.document_type] = (counts[doc.document_type] || 0) + 1;
- }
- return counts;
- }, [documents]);
-
- // Start syncing when Electric client is available
- useEffect(() => {
- // Wait for both searchSpaceId and Electric client to be available
- if (!searchSpaceId || !electricClient) {
- setLoading(!electricClient); // Still loading if waiting for Electric
- if (!searchSpaceId) {
- setDocuments([]);
- }
- return;
- }
-
- // Create a unique key for this sync to prevent duplicate subscriptions
- const syncKey = `documents_${searchSpaceId}`;
- if (syncKeyRef.current === syncKey) {
- // Already syncing for this search space
- return;
- }
-
- let mounted = true;
- syncKeyRef.current = syncKey;
-
- async function startSync() {
- try {
- console.log("[useDocumentsElectric] Starting sync for search space:", searchSpaceId);
-
- const handle = await electricClient.syncShape({
- table: "documents",
- where: `search_space_id = ${searchSpaceId}`,
- columns: ["id", "document_type", "search_space_id", "created_at"],
- primaryKey: ["id"],
- });
-
- console.log("[useDocumentsElectric] Sync started:", {
- isUpToDate: handle.isUpToDate,
- });
-
- // Wait for initial sync with timeout
- if (!handle.isUpToDate && handle.initialSyncPromise) {
- try {
- await Promise.race([
- handle.initialSyncPromise,
- new Promise((resolve) => setTimeout(resolve, 2000)),
- ]);
- } catch (syncErr) {
- console.error("[useDocumentsElectric] Initial sync failed:", syncErr);
- }
- }
-
- if (!mounted) {
- handle.unsubscribe();
- return;
- }
-
- syncHandleRef.current = handle;
- setLoading(false);
- setError(null);
-
- // Fetch initial documents
- await fetchDocuments();
-
- // Set up live query for real-time updates
- await setupLiveQuery();
- } catch (err) {
- if (!mounted) return;
- console.error("[useDocumentsElectric] Failed to start sync:", err);
- setError(err instanceof Error ? err : new Error("Failed to sync documents"));
- setLoading(false);
- }
- }
-
- async function fetchDocuments() {
- try {
- const result = await electricClient.db.query(
- `SELECT id, document_type, search_space_id, created_at FROM documents WHERE search_space_id = $1 ORDER BY created_at DESC`,
- [searchSpaceId]
- );
- if (mounted) {
- setDocuments(result.rows || []);
- }
- } catch (err) {
- console.error("[useDocumentsElectric] Failed to fetch:", err);
- }
- }
-
- async function setupLiveQuery() {
- try {
- // eslint-disable-next-line @typescript-eslint/no-explicit-any
- const db = electricClient.db as any;
-
- if (db.live?.query && typeof db.live.query === "function") {
- const liveQuery = await db.live.query(
- `SELECT id, document_type, search_space_id, created_at FROM documents WHERE search_space_id = $1 ORDER BY created_at DESC`,
- [searchSpaceId]
- );
-
- if (!mounted) {
- liveQuery.unsubscribe?.();
- return;
- }
-
- // Set initial results
- if (liveQuery.initialResults?.rows) {
- setDocuments(liveQuery.initialResults.rows);
- } else if (liveQuery.rows) {
- setDocuments(liveQuery.rows);
- }
-
- // Subscribe to changes
- if (typeof liveQuery.subscribe === "function") {
- liveQuery.subscribe((result: { rows: Document[] }) => {
- if (mounted && result.rows) {
- setDocuments(result.rows);
- }
- });
- }
-
- if (typeof liveQuery.unsubscribe === "function") {
- liveQueryRef.current = liveQuery;
- }
- }
- } catch (liveErr) {
- console.error("[useDocumentsElectric] Failed to set up live query:", liveErr);
- }
- }
-
- 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;
- }
- };
- }, [searchSpaceId, electricClient]);
-
- return { documentTypeCounts, loading, error };
-}
diff --git a/surfsense_web/hooks/use-documents.ts b/surfsense_web/hooks/use-documents.ts
new file mode 100644
index 000000000..369cc7b41
--- /dev/null
+++ b/surfsense_web/hooks/use-documents.ts
@@ -0,0 +1,449 @@
+"use client";
+
+import { useCallback, useEffect, useMemo, useRef, useState } from "react";
+import type { DocumentTypeEnum } from "@/contracts/types/document.types";
+import { documentsApiService } from "@/lib/apis/documents-api.service";
+import type { SyncHandle } from "@/lib/electric/client";
+import { useElectricClient } from "@/lib/electric/context";
+
+// Stable empty array to prevent infinite re-renders when no typeFilter is provided
+const EMPTY_TYPE_FILTER: DocumentTypeEnum[] = [];
+
+// Document status type (matches backend DocumentStatus JSONB)
+export interface DocumentStatusType {
+ state: "ready" | "pending" | "processing" | "failed";
+ reason?: string;
+}
+
+// Document from Electric sync (lightweight table columns - NO content/metadata)
+interface DocumentElectric {
+ id: number;
+ search_space_id: number;
+ document_type: string;
+ title: string;
+ created_by_id: string | null;
+ created_at: string;
+ status: DocumentStatusType | null;
+}
+
+// Document for display (with resolved user name)
+export interface DocumentDisplay {
+ id: number;
+ search_space_id: number;
+ document_type: string;
+ title: string;
+ created_by_id: string | null;
+ created_by_name: string | null;
+ created_at: string;
+ status: DocumentStatusType;
+}
+
+/**
+ * Deduplicate by ID and sort by created_at descending (newest first)
+ */
+function deduplicateAndSort(items: T[]): T[] {
+ const seen = new Map();
+ for (const item of items) {
+ // Keep the most recent version if duplicate
+ const existing = seen.get(item.id);
+ if (!existing || new Date(item.created_at) > new Date(existing.created_at)) {
+ seen.set(item.id, item);
+ }
+ }
+ return Array.from(seen.values()).sort(
+ (a, b) => new Date(b.created_at).getTime() - new Date(a.created_at).getTime()
+ );
+}
+
+/**
+ * Check if a document has valid/complete data
+ */
+function isValidDocument(doc: DocumentElectric): boolean {
+ return doc.id != null && doc.title != null && doc.title !== "";
+}
+
+/**
+ * Real-time documents hook with Electric SQL
+ *
+ * Architecture (100% Reliable):
+ * 1. API is the PRIMARY source of truth - always loads first
+ * 2. Electric provides REAL-TIME updates for additions and deletions
+ * 3. Use syncHandle.isUpToDate to determine if deletions can be trusted
+ * 4. Handles bulk deletions correctly by checking sync state
+ *
+ * @param searchSpaceId - The search space ID to filter documents
+ * @param typeFilter - Optional document types to filter by
+ */
+export function useDocuments(
+ searchSpaceId: number | null,
+ typeFilter: DocumentTypeEnum[] = EMPTY_TYPE_FILTER
+) {
+ const electricClient = useElectricClient();
+
+ const [documents, setDocuments] = useState([]);
+ const [loading, setLoading] = useState(true);
+ const [error, setError] = useState(null);
+
+ // Track if initial API load is complete (source of truth)
+ const apiLoadedRef = useRef(false);
+
+ // User cache: userId → displayName
+ const userCacheRef = useRef>(new Map());
+
+ // Electric sync refs
+ const syncHandleRef = useRef(null);
+ const liveQueryRef = useRef<{ unsubscribe?: () => void } | null>(null);
+
+ // Real-time type counts
+ const typeCounts = useMemo(() => {
+ const counts: Record = {};
+ for (const doc of documents) {
+ counts[doc.document_type] = (counts[doc.document_type] || 0) + 1;
+ }
+ return counts;
+ }, [documents]);
+
+ // Populate user cache from API response
+ const populateUserCache = useCallback(
+ (items: Array<{ created_by_id?: string | null; created_by_name?: string | null }>) => {
+ for (const item of items) {
+ if (item.created_by_id && item.created_by_name) {
+ userCacheRef.current.set(item.created_by_id, item.created_by_name);
+ }
+ }
+ },
+ []
+ );
+
+ // Convert API item to display doc
+ const apiToDisplayDoc = useCallback(
+ (item: {
+ id: number;
+ search_space_id: number;
+ document_type: string;
+ title: string;
+ created_by_id?: string | null;
+ created_by_name?: string | null;
+ created_at: string;
+ status?: DocumentStatusType | null;
+ }): DocumentDisplay => ({
+ id: item.id,
+ search_space_id: item.search_space_id,
+ document_type: item.document_type,
+ title: item.title,
+ created_by_id: item.created_by_id ?? null,
+ created_by_name: item.created_by_name ?? null,
+ created_at: item.created_at,
+ status: item.status ?? { state: "ready" },
+ }),
+ []
+ );
+
+ // Convert Electric doc to display doc
+ const electricToDisplayDoc = useCallback(
+ (doc: DocumentElectric): DocumentDisplay => ({
+ ...doc,
+ created_by_name: doc.created_by_id
+ ? (userCacheRef.current.get(doc.created_by_id) ?? null)
+ : null,
+ status: doc.status ?? { state: "ready" },
+ }),
+ []
+ );
+
+ // EFFECT 1: Load from API (PRIMARY source of truth)
+ useEffect(() => {
+ if (!searchSpaceId) {
+ setLoading(false);
+ return;
+ }
+
+ // Capture validated value for async closure
+ const spaceId = searchSpaceId;
+ const currentTypeFilter = typeFilter;
+
+ let mounted = true;
+ apiLoadedRef.current = false;
+
+ async function loadFromApi() {
+ try {
+ setLoading(true);
+ console.log("[useDocuments] Loading from API (source of truth):", spaceId);
+
+ const response = await documentsApiService.getDocuments({
+ queryParams: {
+ search_space_id: spaceId,
+ page: 0,
+ page_size: -1, // Fetch all documents
+ ...(currentTypeFilter.length > 0 && { document_types: currentTypeFilter }),
+ },
+ });
+
+ if (!mounted) return;
+
+ populateUserCache(response.items);
+ const docs = response.items.map(apiToDisplayDoc);
+ setDocuments(docs);
+ apiLoadedRef.current = true;
+ setError(null);
+ console.log("[useDocuments] API loaded", docs.length, "documents");
+ } catch (err) {
+ if (!mounted) return;
+ console.error("[useDocuments] API load failed:", err);
+ setError(err instanceof Error ? err : new Error("Failed to load documents"));
+ } finally {
+ if (mounted) setLoading(false);
+ }
+ }
+
+ loadFromApi();
+
+ return () => {
+ mounted = false;
+ };
+ }, [searchSpaceId, typeFilter, populateUserCache, apiToDisplayDoc]);
+
+ // EFFECT 2: Start Electric sync + live query for real-time updates
+ useEffect(() => {
+ if (!searchSpaceId || !electricClient) return;
+
+ // Capture validated values for async closure
+ const spaceId = searchSpaceId;
+ const client = electricClient;
+ const currentTypeFilter = typeFilter;
+
+ let mounted = true;
+
+ async function setupElectricRealtime() {
+ // Cleanup previous subscriptions
+ if (syncHandleRef.current) {
+ syncHandleRef.current.unsubscribe();
+ syncHandleRef.current = null;
+ }
+ if (liveQueryRef.current) {
+ liveQueryRef.current.unsubscribe?.();
+ liveQueryRef.current = null;
+ }
+
+ try {
+ console.log("[useDocuments] Starting Electric sync for real-time updates");
+
+ // Start Electric sync
+ const handle = await client.syncShape({
+ table: "documents",
+ where: `search_space_id = ${spaceId}`,
+ columns: [
+ "id",
+ "document_type",
+ "search_space_id",
+ "title",
+ "created_by_id",
+ "created_at",
+ "status",
+ ],
+ primaryKey: ["id"],
+ });
+
+ if (!mounted) {
+ handle.unsubscribe();
+ return;
+ }
+
+ syncHandleRef.current = handle;
+ console.log("[useDocuments] Sync started, isUpToDate:", handle.isUpToDate);
+
+ // Wait for initial sync (with timeout)
+ if (!handle.isUpToDate && handle.initialSyncPromise) {
+ await Promise.race([
+ handle.initialSyncPromise,
+ new Promise((resolve) => setTimeout(resolve, 5000)),
+ ]);
+ console.log("[useDocuments] Initial sync complete, isUpToDate:", handle.isUpToDate);
+ }
+
+ if (!mounted) return;
+
+ // Set up live query
+ const db = client.db as {
+ live?: {
+ query: (
+ sql: string,
+ params?: (number | string)[]
+ ) => Promise<{
+ subscribe: (cb: (result: { rows: T[] }) => void) => void;
+ unsubscribe?: () => void;
+ }>;
+ };
+ };
+
+ if (!db.live?.query) {
+ console.warn("[useDocuments] Live queries not available");
+ return;
+ }
+
+ let query = `SELECT id, document_type, search_space_id, title, created_by_id, created_at, status
+ FROM documents
+ WHERE search_space_id = $1`;
+
+ const params: (number | string)[] = [spaceId];
+
+ if (currentTypeFilter.length > 0) {
+ const placeholders = currentTypeFilter.map((_, i) => `$${i + 2}`).join(", ");
+ query += ` AND document_type IN (${placeholders})`;
+ params.push(...currentTypeFilter);
+ }
+
+ query += ` ORDER BY created_at DESC`;
+
+ const liveQuery = await db.live.query(query, params);
+
+ if (!mounted) {
+ liveQuery.unsubscribe?.();
+ return;
+ }
+
+ console.log("[useDocuments] Live query subscribed");
+
+ liveQuery.subscribe((result: { rows: DocumentElectric[] }) => {
+ if (!mounted || !result.rows) return;
+
+ // DEBUG: Log first few raw documents to see what's coming from Electric
+ console.log("[useDocuments] Raw data sample:", result.rows.slice(0, 3));
+
+ const validItems = result.rows.filter(isValidDocument);
+ const isFullySynced = syncHandleRef.current?.isUpToDate ?? false;
+
+ console.log(
+ `[useDocuments] Live update: ${result.rows.length} raw, ${validItems.length} valid, synced: ${isFullySynced}`
+ );
+
+ // Fetch user names for new users (non-blocking)
+ const unknownUserIds = validItems
+ .filter(
+ (doc): doc is DocumentElectric & { created_by_id: string } =>
+ doc.created_by_id !== null && !userCacheRef.current.has(doc.created_by_id)
+ )
+ .map((doc) => doc.created_by_id);
+
+ if (unknownUserIds.length > 0) {
+ documentsApiService
+ .getDocuments({
+ queryParams: { search_space_id: spaceId, page: 0, page_size: 20 },
+ })
+ .then((response) => {
+ populateUserCache(response.items);
+ if (mounted) {
+ setDocuments((prev) =>
+ prev.map((doc) => ({
+ ...doc,
+ created_by_name: doc.created_by_id
+ ? (userCacheRef.current.get(doc.created_by_id) ?? null)
+ : null,
+ }))
+ );
+ }
+ })
+ .catch(() => {});
+ }
+
+ // Smart update logic based on sync state
+ setDocuments((prev) => {
+ // Don't process if API hasn't loaded yet
+ if (!apiLoadedRef.current) {
+ console.log("[useDocuments] Waiting for API load, skipping live update");
+ return prev;
+ }
+
+ // Case 1: Live query is empty
+ if (validItems.length === 0) {
+ if (isFullySynced && prev.length > 0) {
+ // Electric is fully synced and says 0 items - trust it (all deleted)
+ console.log("[useDocuments] All documents deleted (Electric synced)");
+ return [];
+ }
+ // Partial sync or error - keep existing
+ console.log("[useDocuments] Empty live result, keeping existing");
+ return prev;
+ }
+
+ // Case 2: Electric is fully synced - TRUST IT COMPLETELY (handles bulk deletes)
+ if (isFullySynced) {
+ const liveDocs = deduplicateAndSort(validItems.map(electricToDisplayDoc));
+ console.log(
+ `[useDocuments] Synced update: ${liveDocs.length} docs (was ${prev.length})`
+ );
+ return liveDocs;
+ }
+
+ // Case 3: Partial sync - only ADD new items, don't remove any
+ const existingIds = new Set(prev.map((d) => d.id));
+ const liveIds = new Set(validItems.map((d) => d.id));
+
+ // Find new items (in live but not in prev)
+ const newItems = validItems
+ .filter((item) => !existingIds.has(item.id))
+ .map(electricToDisplayDoc);
+
+ // Find updated items (in both, update with latest data)
+ const updatedPrev = prev.map((doc) => {
+ if (liveIds.has(doc.id)) {
+ const liveItem = validItems.find((v) => v.id === doc.id);
+ if (liveItem) {
+ return electricToDisplayDoc(liveItem);
+ }
+ }
+ return doc;
+ });
+
+ if (newItems.length > 0) {
+ console.log(`[useDocuments] Adding ${newItems.length} new items (partial sync)`);
+ return deduplicateAndSort([...newItems, ...updatedPrev]);
+ }
+
+ return updatedPrev;
+ });
+ });
+
+ liveQueryRef.current = liveQuery;
+ } catch (err) {
+ console.error("[useDocuments] Electric setup failed:", err);
+ // Don't set error - API data is already loaded
+ }
+ }
+
+ setupElectricRealtime();
+
+ return () => {
+ mounted = false;
+ if (syncHandleRef.current) {
+ syncHandleRef.current.unsubscribe();
+ syncHandleRef.current = null;
+ }
+ if (liveQueryRef.current) {
+ liveQueryRef.current.unsubscribe?.();
+ liveQueryRef.current = null;
+ }
+ };
+ }, [searchSpaceId, electricClient, typeFilter, electricToDisplayDoc, populateUserCache]);
+
+ // Track previous searchSpaceId to detect actual changes
+ const prevSearchSpaceIdRef = useRef(null);
+
+ // Reset on search space change (not on initial mount)
+ useEffect(() => {
+ if (prevSearchSpaceIdRef.current !== null && prevSearchSpaceIdRef.current !== searchSpaceId) {
+ setDocuments([]);
+ apiLoadedRef.current = false;
+ userCacheRef.current.clear();
+ }
+ prevSearchSpaceIdRef.current = searchSpaceId;
+ }, [searchSpaceId]);
+
+ return {
+ documents,
+ typeCounts,
+ total: documents.length,
+ loading,
+ error,
+ };
+}
diff --git a/surfsense_web/hooks/use-inbox.ts b/surfsense_web/hooks/use-inbox.ts
index 362feb747..56ddb46a4 100644
--- a/surfsense_web/hooks/use-inbox.ts
+++ b/surfsense_web/hooks/use-inbox.ts
@@ -38,10 +38,14 @@ function deduplicateAndSort(items: InboxItem[]): InboxItem[] {
/**
* Calculate the cutoff date for sync window
+ * IMPORTANT: Rounds to the start of the day (midnight UTC) to ensure stable values
+ * across re-renders. Without this, millisecond differences cause multiple syncs!
*/
function getSyncCutoffDate(): string {
const cutoff = new Date();
cutoff.setDate(cutoff.getDate() - SYNC_WINDOW_DAYS);
+ // Round to start of day to prevent millisecond differences causing duplicate syncs
+ cutoff.setUTCHours(0, 0, 0, 0);
return cutoff.toISOString();
}
diff --git a/surfsense_web/lib/electric/client.ts b/surfsense_web/lib/electric/client.ts
index 177a66d28..9d596a261 100644
--- a/surfsense_web/lib/electric/client.ts
+++ b/surfsense_web/lib/electric/client.ts
@@ -12,10 +12,21 @@
* 3. Works even if logout cleanup fails
*/
-import { PGlite } from "@electric-sql/pglite";
+import { PGlite, type Transaction } from "@electric-sql/pglite";
import { live } from "@electric-sql/pglite/live";
import { electricSync } from "@electric-sql/pglite-sync";
+// Debug logging - only logs in development, silent in production
+const IS_DEV = process.env.NODE_ENV === "development";
+
+function debugLog(...args: unknown[]) {
+ if (IS_DEV) console.log(...args);
+}
+
+function debugWarn(...args: unknown[]) {
+ if (IS_DEV) console.warn(...args);
+}
+
// Types
export interface ElectricClient {
db: PGlite;
@@ -56,7 +67,14 @@ const pendingSyncs = new Map>();
// v2: user-specific database architecture
// v3: consistent cutoff date for sync+queries, visibility refresh support
// v4: heartbeat-based stale notification detection with updated_at tracking
-const SYNC_VERSION = 4;
+// v5: fixed duplicate key errors (root cause: unstable cutoff dates in use-inbox.ts)
+// - added onMustRefetch handler for server-side refetch scenarios
+// - fixed getSyncCutoffDate to use stable midnight UTC timestamps
+// v6: real-time documents table - added title and created_by_id columns for live document display
+// v7: removed use-documents-electric.ts - consolidated to single documents sync to prevent conflicts
+// v8: added status column for real-time document processing status (ready/processing/failed)
+// v9: added pending state for accurate document queue visibility
+const SYNC_VERSION = 11;
// Database name prefix for identifying SurfSense databases
const DB_PREFIX = "surfsense-";
@@ -77,7 +95,7 @@ function getDbName(userId: string): string {
}
/**
- * Clean up databases from OTHER users (not the current user)
+ * Clean up databases from OTHER users AND old versions
* This is called on login to ensure clean state
*/
async function cleanupOtherUserDatabases(currentUserId: string): Promise {
@@ -85,6 +103,10 @@ async function cleanupOtherUserDatabases(currentUserId: string): Promise {
return;
}
+ // The exact database identifier we want to keep (current user + current version)
+ // Format: "surfsense-{userId}-v{version}"
+ const currentDbIdentifier = `${DB_PREFIX}${currentUserId}-v${SYNC_VERSION}`;
+
try {
// Try to list all databases (not supported in all browsers)
if (typeof window.indexedDB.databases === "function") {
@@ -95,26 +117,27 @@ async function cleanupOtherUserDatabases(currentUserId: string): Promise {
if (!dbName) continue;
// Check if this is a SurfSense database
- if (dbName.startsWith(DB_PREFIX) || dbName.includes("surfsense")) {
- // Don't delete current user's database
- if (dbName.includes(currentUserId)) {
- console.log(`[Electric] Keeping current user's database: ${dbName}`);
+ if (dbName.includes("surfsense")) {
+ // Check if this is the current database
+ // PGlite stores with "/pglite/" prefix, so we check if the name ENDS WITH our identifier
+ if (dbName.endsWith(currentDbIdentifier)) {
+ debugLog(`[Electric] Keeping current database: ${dbName}`);
continue;
}
- // Delete databases from other users
+ // Delete ALL other databases (other users OR old versions of current user)
try {
- console.log(`[Electric] Deleting stale database: ${dbName}`);
+ debugLog(`[Electric] Deleting stale database: ${dbName}`);
window.indexedDB.deleteDatabase(dbName);
} catch (deleteErr) {
- console.warn(`[Electric] Failed to delete database ${dbName}:`, deleteErr);
+ debugWarn(`[Electric] Failed to delete database ${dbName}:`, deleteErr);
}
}
}
}
} catch (err) {
// indexedDB.databases() not supported - that's okay, login cleanup is best-effort
- console.warn("[Electric] Could not enumerate databases for cleanup:", err);
+ debugWarn("[Electric] Could not enumerate databases for cleanup:", err);
}
}
@@ -140,7 +163,7 @@ export async function initElectric(userId: string): Promise {
// If initialized for a different user, close the old client first
if (electricClient && currentUserId !== userId) {
- console.log(`[Electric] User changed from ${currentUserId} to ${userId}, reinitializing...`);
+ debugLog(`[Electric] User changed from ${currentUserId} to ${userId}, reinitializing...`);
await cleanupElectric();
}
@@ -155,12 +178,12 @@ export async function initElectric(userId: string): Promise {
initPromise = (async () => {
try {
// STEP 1: Clean up databases from other users (login-time cleanup)
- console.log("[Electric] Cleaning up databases from other users...");
+ debugLog("[Electric] Cleaning up databases from other users...");
await cleanupOtherUserDatabases(userId);
// STEP 2: Create user-specific PGlite database
const dbName = getDbName(userId);
- console.log(`[Electric] Initializing database: ${dbName}`);
+ debugLog(`[Electric] Initializing database: ${dbName}`);
const db = await PGlite.create({
dataDir: dbName,
@@ -216,18 +239,22 @@ export async function initElectric(userId: string): Promise {
`);
// Create the documents table schema in PGlite
- // Only sync minimal fields needed for type counts: id, document_type, search_space_id
+ // Sync columns needed for real-time table display (lightweight - no content/metadata)
await db.exec(`
CREATE TABLE IF NOT EXISTS documents (
id INTEGER PRIMARY KEY,
search_space_id INTEGER NOT NULL,
document_type TEXT NOT NULL,
- created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+ title TEXT NOT NULL DEFAULT '',
+ created_by_id TEXT,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+ status JSONB DEFAULT '{"state": "ready"}'::jsonb
);
CREATE INDEX IF NOT EXISTS idx_documents_search_space_id ON documents(search_space_id);
CREATE INDEX IF NOT EXISTS idx_documents_type ON documents(document_type);
CREATE INDEX IF NOT EXISTS idx_documents_search_space_type ON documents(search_space_id, document_type);
+ CREATE INDEX IF NOT EXISTS idx_documents_status ON documents((status->>'state'));
`);
await db.exec(`
@@ -290,14 +317,14 @@ export async function initElectric(userId: string): Promise {
// Check if we already have an active sync for this shape (memory optimization)
const existingHandle = activeSyncHandles.get(cacheKey);
if (existingHandle) {
- console.log(`[Electric] Reusing existing sync handle for: ${cacheKey}`);
+ debugLog(`[Electric] Reusing existing sync handle for: ${cacheKey}`);
return existingHandle;
}
// Check if there's already a pending sync for this shape (prevent race condition)
const pendingSync = pendingSyncs.get(cacheKey);
if (pendingSync) {
- console.log(`[Electric] Waiting for pending sync to complete: ${cacheKey}`);
+ debugLog(`[Electric] Waiting for pending sync to complete: ${cacheKey}`);
return pendingSync;
}
@@ -323,7 +350,7 @@ export async function initElectric(userId: string): Promise {
if (singleQuoteCount % 2 !== 0) {
// Odd number of quotes means unterminated string literal
- console.warn("Where clause has unmatched quotes, fixing:", where);
+ debugWarn("Where clause has unmatched quotes, fixing:", where);
// Add closing quote at the end
validatedWhere = `${where}'`;
params.where = validatedWhere;
@@ -337,15 +364,15 @@ export async function initElectric(userId: string): Promise {
if (columns) params.columns = columns.join(",");
- console.log("[Electric] Syncing shape with params:", params);
- console.log("[Electric] Electric URL:", `${electricUrl}/v1/shape`);
- console.log("[Electric] Where clause:", where, "Validated:", validatedWhere);
+ debugLog("[Electric] Syncing shape with params:", params);
+ debugLog("[Electric] Electric URL:", `${electricUrl}/v1/shape`);
+ debugLog("[Electric] Where clause:", where, "Validated:", validatedWhere);
try {
// Debug: Test Electric SQL connection directly first (DEV ONLY - skipped in production)
if (process.env.NODE_ENV === "development") {
const testUrl = `${electricUrl}/v1/shape?table=${table}&offset=-1${validatedWhere ? `&where=${encodeURIComponent(validatedWhere)}` : ""}`;
- console.log("[Electric] Testing Electric SQL directly:", testUrl);
+ debugLog("[Electric] Testing Electric SQL directly:", testUrl);
try {
const testResponse = await fetch(testUrl);
const testHeaders = {
@@ -353,9 +380,9 @@ export async function initElectric(userId: string): Promise {
offset: testResponse.headers.get("electric-offset"),
upToDate: testResponse.headers.get("electric-up-to-date"),
};
- console.log("[Electric] Direct Electric SQL response headers:", testHeaders);
+ debugLog("[Electric] Direct Electric SQL response headers:", testHeaders);
const testData = await testResponse.json();
- console.log(
+ debugLog(
"[Electric] Direct Electric SQL data count:",
Array.isArray(testData) ? testData.length : "not array",
testData
@@ -396,14 +423,14 @@ export async function initElectric(userId: string): Promise {
// Shorter timeout (5 seconds) as fallback
setTimeout(() => {
if (!syncResolved) {
- console.warn(
+ debugWarn(
`[Electric] ⚠️ Sync timeout for ${table} - checking isUpToDate one more time...`
);
// Check isUpToDate one more time before resolving
// This will be checked after shape is created
setTimeout(() => {
if (!syncResolved) {
- console.warn(
+ debugWarn(
`[Electric] ⚠️ Sync timeout for ${table} - resolving anyway after 5s`
);
resolveInitialSync();
@@ -413,7 +440,22 @@ export async function initElectric(userId: string): Promise {
}, 5000);
});
- // Include userId in shapeKey for user-specific sync state
+ // ROOT CAUSE FIX: The duplicate key errors were caused by unstable cutoff dates
+ // in use-inbox.ts generating different sync keys on each render.
+ // That's now fixed (rounded to midnight UTC in getSyncCutoffDate).
+ // We can safely use shapeKey for fast incremental sync.
+
+ const shapeKey = `${userId}_v${SYNC_VERSION}_${table}_${where?.replace(/[^a-zA-Z0-9]/g, "_") || "all"}`;
+
+ // Type assertion to PGlite with electric extension
+ const pgWithElectric = db as unknown as {
+ electric: {
+ syncShapeToTable: (
+ config: Record
+ ) => Promise<{ unsubscribe: () => void; isUpToDate: boolean; stream: unknown }>;
+ };
+ };
+
const shapeConfig = {
shape: {
url: `${electricUrl}/v1/shape`,
@@ -425,9 +467,9 @@ export async function initElectric(userId: string): Promise {
},
table,
primaryKey,
- shapeKey: `${userId}_v${SYNC_VERSION}_${table}_${where?.replace(/[^a-zA-Z0-9]/g, "_") || "all"}`, // User-specific versioned key
+ shapeKey, // Re-enabled for fast incremental sync (root cause in use-inbox.ts is fixed)
onInitialSync: () => {
- console.log(
+ debugLog(
`[Electric] ✅ Initial sync complete for ${table} - data should now be in PGlite`
);
resolveInitialSync();
@@ -440,21 +482,37 @@ export async function initElectric(userId: string): Promise {
);
rejectInitialSync(error);
},
+ // Handle must-refetch: clear table data before Electric re-inserts from scratch
+ // This prevents "duplicate key" errors when the shape is invalidated
+ onMustRefetch: async (tx: Transaction) => {
+ debugLog(
+ `[Electric] ⚠️ Must refetch triggered for ${table} - clearing existing data`
+ );
+ try {
+ // Delete rows matching the shape's WHERE clause
+ // If no WHERE clause, delete all rows from the table
+ if (validatedWhere) {
+ // Parse the WHERE clause to build a DELETE statement
+ // The WHERE clause is already validated and formatted
+ await tx.exec(`DELETE FROM ${table} WHERE ${validatedWhere}`);
+ debugLog(`[Electric] 🗑️ Cleared ${table} rows matching: ${validatedWhere}`);
+ } else {
+ // No WHERE clause means we're syncing the entire table
+ await tx.exec(`DELETE FROM ${table}`);
+ debugLog(`[Electric] 🗑️ Cleared all rows from ${table}`);
+ }
+ } catch (cleanupError) {
+ console.error(
+ `[Electric] ❌ Failed to clear ${table} during must-refetch:`,
+ cleanupError
+ );
+ // Re-throw to let Electric handle the error
+ throw cleanupError;
+ }
+ },
};
- console.log(
- "[Electric] syncShapeToTable config:",
- JSON.stringify(shapeConfig, null, 2)
- );
-
- // Type assertion to PGlite with electric extension
- const pgWithElectric = db as PGlite & {
- electric: {
- syncShapeToTable: (
- config: typeof shapeConfig
- ) => Promise<{ unsubscribe: () => void; isUpToDate: boolean; stream: unknown }>;
- };
- };
+ debugLog("[Electric] syncShapeToTable config:", JSON.stringify(shapeConfig, null, 2));
let shape: { unsubscribe: () => void; isUpToDate: boolean; stream: unknown };
try {
@@ -464,7 +522,7 @@ export async function initElectric(userId: string): Promise {
const errorMessage =
syncError instanceof Error ? syncError.message : String(syncError);
if (errorMessage.includes("Already syncing")) {
- console.warn(
+ debugWarn(
`[Electric] Already syncing ${table}, waiting for existing sync to settle...`
);
@@ -474,12 +532,12 @@ export async function initElectric(userId: string): Promise {
// Check if an active handle now exists (another sync might have completed)
const existingHandle = activeSyncHandles.get(cacheKey);
if (existingHandle) {
- console.log(`[Electric] Found existing handle after waiting: ${cacheKey}`);
+ debugLog(`[Electric] Found existing handle after waiting: ${cacheKey}`);
return existingHandle;
}
// Retry once after waiting
- console.log(`[Electric] Retrying sync for ${table}...`);
+ debugLog(`[Electric] Retrying sync for ${table}...`);
try {
shape = await pgWithElectric.electric.syncShapeToTable(shapeConfig);
} catch (retryError) {
@@ -487,12 +545,10 @@ export async function initElectric(userId: string): Promise {
retryError instanceof Error ? retryError.message : String(retryError);
if (retryMessage.includes("Already syncing")) {
// Still syncing - create a placeholder handle that indicates the table is being synced
- console.warn(
- `[Electric] ${table} still syncing, creating placeholder handle`
- );
+ debugWarn(`[Electric] ${table} still syncing, creating placeholder handle`);
const placeholderHandle: SyncHandle = {
unsubscribe: () => {
- console.log(`[Electric] Placeholder unsubscribe for: ${cacheKey}`);
+ debugLog(`[Electric] Placeholder unsubscribe for: ${cacheKey}`);
activeSyncHandles.delete(cacheKey);
},
get isUpToDate() {
@@ -516,7 +572,7 @@ export async function initElectric(userId: string): Promise {
}
// Log the actual shape result structure
- console.log("[Electric] Shape sync result (initial):", {
+ debugLog("[Electric] Shape sync result (initial):", {
hasUnsubscribe: typeof shape?.unsubscribe === "function",
isUpToDate: shape?.isUpToDate,
hasStream: !!shape?.stream,
@@ -525,7 +581,7 @@ export async function initElectric(userId: string): Promise {
// Recommended Approach Step 1: Check isUpToDate immediately
if (shape.isUpToDate) {
- console.log(
+ debugLog(
`[Electric] ✅ Sync already up-to-date for ${table} (resuming from previous state)`
);
resolveInitialSync();
@@ -533,7 +589,7 @@ export async function initElectric(userId: string): Promise {
// Recommended Approach Step 2: Subscribe to stream and watch for "up-to-date" message
if (shape?.stream) {
const stream = shape.stream as any;
- console.log("[Electric] Shape stream details:", {
+ debugLog("[Electric] Shape stream details:", {
shapeHandle: stream?.shapeHandle,
lastOffset: stream?.lastOffset,
isUpToDate: stream?.isUpToDate,
@@ -546,14 +602,14 @@ export async function initElectric(userId: string): Promise {
// NOTE: We keep this subscription active - don't unsubscribe!
// The stream is what Electric SQL uses for real-time updates
if (typeof stream?.subscribe === "function") {
- console.log(
+ debugLog(
"[Electric] Subscribing to shape stream to watch for up-to-date message..."
);
// Subscribe but don't store unsubscribe - we want it to stay active
stream.subscribe((messages: unknown[]) => {
// Continue receiving updates even after sync is resolved
if (!syncResolved) {
- console.log(
+ debugLog(
"[Electric] 🔵 Shape stream received messages:",
messages?.length || 0
);
@@ -570,14 +626,14 @@ export async function initElectric(userId: string): Promise {
(typeof msg === "object" && "up-to-date" in msg)
) {
if (!syncResolved) {
- console.log(`[Electric] ✅ Received up-to-date message for ${table}`);
+ debugLog(`[Electric] ✅ Received up-to-date message for ${table}`);
resolveInitialSync();
}
// Continue listening for real-time updates - don't return!
}
}
if (!syncResolved && messages.length > 0) {
- console.log(
+ debugLog(
"[Electric] First message:",
JSON.stringify(messages[0], null, 2)
);
@@ -586,16 +642,14 @@ export async function initElectric(userId: string): Promise {
// Also check stream's isUpToDate property after receiving messages
if (!syncResolved && stream?.isUpToDate) {
- console.log(`[Electric] ✅ Stream isUpToDate is true for ${table}`);
+ debugLog(`[Electric] ✅ Stream isUpToDate is true for ${table}`);
resolveInitialSync();
}
});
// Also check stream's isUpToDate property immediately
if (stream?.isUpToDate) {
- console.log(
- `[Electric] ✅ Stream isUpToDate is true immediately for ${table}`
- );
+ debugLog(`[Electric] ✅ Stream isUpToDate is true immediately for ${table}`);
resolveInitialSync();
}
}
@@ -608,9 +662,7 @@ export async function initElectric(userId: string): Promise {
}
if (shape.isUpToDate || stream?.isUpToDate) {
- console.log(
- `[Electric] ✅ Sync completed (detected via polling) for ${table}`
- );
+ debugLog(`[Electric] ✅ Sync completed (detected via polling) for ${table}`);
clearInterval(pollInterval);
resolveInitialSync();
}
@@ -621,7 +673,7 @@ export async function initElectric(userId: string): Promise {
clearInterval(pollInterval);
});
} else {
- console.warn(
+ debugWarn(
`[Electric] ⚠️ No stream available for ${table}, relying on callback and timeout`
);
}
@@ -630,7 +682,7 @@ export async function initElectric(userId: string): Promise {
// Create the sync handle with proper cleanup
const syncHandle: SyncHandle = {
unsubscribe: () => {
- console.log(`[Electric] Unsubscribing from: ${cacheKey}`);
+ debugLog(`[Electric] Unsubscribing from: ${cacheKey}`);
// Remove from cache first
activeSyncHandles.delete(cacheKey);
// Then unsubscribe from the shape
@@ -648,7 +700,7 @@ export async function initElectric(userId: string): Promise {
// Cache the sync handle for reuse (memory optimization)
activeSyncHandles.set(cacheKey, syncHandle);
- console.log(
+ debugLog(
`[Electric] Cached sync handle for: ${cacheKey} (total cached: ${activeSyncHandles.size})`
);
@@ -660,7 +712,7 @@ export async function initElectric(userId: string): Promise {
const response = await fetch(`${electricUrl}/v1/shape?table=${table}&offset=-1`, {
method: "GET",
});
- console.log(
+ debugLog(
"[Electric] Electric SQL server response:",
response.status,
response.statusText
@@ -682,14 +734,14 @@ export async function initElectric(userId: string): Promise {
// Clean up the pending sync when done (whether success or failure)
syncPromise.finally(() => {
pendingSyncs.delete(cacheKey);
- console.log(`[Electric] Pending sync removed for: ${cacheKey}`);
+ debugLog(`[Electric] Pending sync removed for: ${cacheKey}`);
});
return syncPromise;
},
};
- console.log(`[Electric] ✅ Initialized successfully for user: ${userId}`);
+ debugLog(`[Electric] ✅ Initialized successfully for user: ${userId}`);
return electricClient;
} catch (error) {
console.error("[Electric] Failed to initialize:", error);
@@ -715,10 +767,10 @@ export async function cleanupElectric(): Promise {
}
const userIdToClean = currentUserId;
- console.log(`[Electric] Cleaning up for user: ${userIdToClean}`);
+ debugLog(`[Electric] Cleaning up for user: ${userIdToClean}`);
// Unsubscribe from all active sync handles first (memory cleanup)
- console.log(`[Electric] Unsubscribing from ${activeSyncHandles.size} active sync handles`);
+ debugLog(`[Electric] Unsubscribing from ${activeSyncHandles.size} active sync handles`);
// Copy keys to array to avoid mutation during iteration
const handleKeys = Array.from(activeSyncHandles.keys());
for (const key of handleKeys) {
@@ -727,7 +779,7 @@ export async function cleanupElectric(): Promise {
try {
handle.unsubscribe();
} catch (err) {
- console.warn(`[Electric] Failed to unsubscribe from ${key}:`, err);
+ debugWarn(`[Electric] Failed to unsubscribe from ${key}:`, err);
}
}
}
@@ -738,7 +790,7 @@ export async function cleanupElectric(): Promise {
try {
// Close the PGlite database connection
await electricClient.db.close();
- console.log("[Electric] Database closed");
+ debugLog("[Electric] Database closed");
} catch (error) {
console.error("[Electric] Error closing database:", error);
}
@@ -754,13 +806,13 @@ export async function cleanupElectric(): Promise {
try {
const dbName = `${DB_PREFIX}${userIdToClean}-v${SYNC_VERSION}`;
window.indexedDB.deleteDatabase(dbName);
- console.log(`[Electric] Deleted database: ${dbName}`);
+ debugLog(`[Electric] Deleted database: ${dbName}`);
} catch (err) {
- console.warn("[Electric] Failed to delete database:", err);
+ debugWarn("[Electric] Failed to delete database:", err);
}
}
- console.log("[Electric] Cleanup complete");
+ debugLog("[Electric] Cleanup complete");
}
/**
diff --git a/surfsense_web/messages/en.json b/surfsense_web/messages/en.json
index 5a18f80c3..fae4c7265 100644
--- a/surfsense_web/messages/en.json
+++ b/surfsense_web/messages/en.json
@@ -308,6 +308,7 @@
"no_rows_selected": "No rows selected",
"delete_success_count": "Successfully deleted {count} document(s)",
"delete_partial_failed": "Some documents could not be deleted",
+ "delete_success": "Document deleted successfully",
"delete_error": "Error deleting documents",
"filter_by_title": "Filter by title...",
"bulk_delete": "Delete Selected",
@@ -328,7 +329,6 @@
"filter_placeholder": "Filter by title...",
"rows_per_page": "Rows per page",
"refresh": "Refresh",
- "refresh_success": "Documents refreshed",
"upload_documents": "Upload Documents",
"create_shared_note": "Create Shared Note",
"processing_documents": "Processing documents...",
diff --git a/surfsense_web/messages/zh.json b/surfsense_web/messages/zh.json
index 1046b7296..2667a06d1 100644
--- a/surfsense_web/messages/zh.json
+++ b/surfsense_web/messages/zh.json
@@ -313,7 +313,6 @@
"filter_placeholder": "按标题筛选...",
"rows_per_page": "每页行数",
"refresh": "刷新",
- "refresh_success": "文档已刷新",
"upload_documents": "上传文档",
"create_shared_note": "创建共享笔记",
"processing_documents": "正在处理文档...",