apunkt/SurfSense
1
0
Fork 0
mirror of https://github.com/MODSetter/SurfSense.git synced 2026-05-05 13:52:40 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

feat: Reduce memory footprint for PGlite, Implement user-specific Electric SQL database management

Browse source 
- Added cleanup logic for other users' databases on login to ensure data isolation.
- Introduced best-effort cleanup on logout to remove stale databases.
- Updated Electric client initialization to support user-specific databases.
- Refactored hooks to utilize the Electric context for better state management and prevent race conditions.
- Enhanced error handling and logging for Electric SQL operations.
This commit is contained in:
Anish Sarkar 2026-01-15 15:22:23 +05:30
parent eb1ddf0c92
commit 703ec08d19
9 changed files with 752 additions and 489 deletions
Download patch file Download diff file Expand all files Collapse all files

13
surfsense_web/components/UserDropdown.tsx
View file

@ -13,6 +13,7 @@ import {
DropdownMenuSeparator,
DropdownMenuTrigger,
} from "@/components/ui/dropdown-menu";
import { cleanupElectric } from "@/lib/electric/client";
import { resetUser, trackLogout } from "@/lib/posthog/events";
export function UserDropdown({
@ -26,12 +27,20 @@ export function UserDropdown({
}) {
const router = useRouter();
const handleLogout = () => {
const handleLogout = async () => {
try {
// Track logout event and reset PostHog identity
trackLogout();
resetUser();
// Best-effort cleanup of Electric SQL / PGlite
// Even if this fails, login-time cleanup will handle it
try {
await cleanupElectric();
} catch (err) {
console.warn("[Logout] Electric cleanup failed (will be handled on next login):", err);
}
if (typeof window !== "undefined") {
localStorage.removeItem("surfsense_bearer_token");
window.location.href = "/";
@ -40,7 +49,7 @@ export function UserDropdown({
console.error("Error during logout:", error);
// Optionally, provide user feedback
if (typeof window !== "undefined") {
alert("Logout failed. Please try again.");
localStorage.removeItem("surfsense_bearer_token");
window.location.href = "/";
}
}

12
surfsense_web/components/layout/providers/LayoutDataProvider.tsx
View file

@ -21,6 +21,7 @@ import {
} from "@/components/ui/dialog";
import { searchSpacesApiService } from "@/lib/apis/search-spaces-api.service";
import { deleteThread, fetchThreads } from "@/lib/chat/thread-persistence";
import { cleanupElectric } from "@/lib/electric/client";
import { resetUser, trackLogout } from "@/lib/posthog/events";
import { cacheKeys } from "@/lib/query-client/cache-keys";
import type { ChatItem, NavItem, SearchSpace } from "../types/layout.types";
@ -278,10 +279,19 @@ export function LayoutDataProvider({
router.push(`/dashboard/${searchSpaceId}/team`);
}, [router, searchSpaceId]);
const handleLogout = useCallback(() => {
const handleLogout = useCallback(async () => {
try {
trackLogout();
resetUser();
// Best-effort cleanup of Electric SQL / PGlite
// Even if this fails, login-time cleanup will handle it
try {
await cleanupElectric();
} catch (err) {
console.warn("[Logout] Electric cleanup failed (will be handled on next login):", err);
}
if (typeof window !== "undefined") {
localStorage.removeItem("surfsense_bearer_token");
router.push("/");

10
surfsense_web/components/notifications/NotificationButton.tsx
View file

@ -9,12 +9,20 @@ import { useAtomValue } from "jotai";
import { currentUserAtom } from "@/atoms/user/user-query.atoms";
import { NotificationPopup } from "./NotificationPopup";
import { cn } from "@/lib/utils";
import { useParams } from "next/navigation";
export function NotificationButton() {
const { data: user } = useAtomValue(currentUserAtom);
const params = useParams();
const userId = user?.id ? String(user.id) : null;
// Get searchSpaceId from URL params - the component is rendered within /dashboard/[search_space_id]/
const searchSpaceId = params?.search_space_id
? Number(params.search_space_id)
: null;
const { notifications, unreadCount, loading, markAsRead, markAllAsRead } =
useNotifications(userId);
useNotifications(userId, searchSpaceId);
return (
<Popover>

117
surfsense_web/components/providers/ElectricProvider.tsx
View file

@ -1,44 +1,95 @@
"use client";
import { useEffect, useState } from "react";
import { initElectric, isElectricInitialized } from "@/lib/electric/client";
import { useEffect, useState, useRef } from "react";
import { useAtomValue } from "jotai";
import { currentUserAtom } from "@/atoms/user/user-query.atoms";
import {
initElectric,
cleanupElectric,
isElectricInitialized,
type ElectricClient,
} from "@/lib/electric/client";
import { ElectricContext } from "@/lib/electric/context";
interface ElectricProviderProps {
children: React.ReactNode;
}
/**
* ElectricProvider initializes the Electric SQL client with PGlite
* ElectricProvider initializes the Electric SQL client with user-specific PGlite database
* and provides it to children via context.
*
* This provider ensures Electric is initialized before rendering children,
* but doesn't block if initialization fails (app can still work without real-time sync)
* KEY BEHAVIORS:
* 1. Single initialization point - only this provider creates the Electric client
* 2. Creates user-specific database (isolated per user)
* 3. Cleans up other users' databases on login
* 4. Re-initializes when user changes
* 5. Provides client via context - hooks should use useElectricClient()
*/
export function ElectricProvider({ children }: ElectricProviderProps) {
const [initialized, setInitialized] = useState(false);
const [electricClient, setElectricClient] = useState<ElectricClient | null>(null);
const [error, setError] = useState<Error | null>(null);
const { data: user, isSuccess: isUserLoaded, isError: isUserError } = useAtomValue(currentUserAtom);
const previousUserIdRef = useRef<string | null>(null);
const initializingRef = useRef(false);
useEffect(() => {
// Skip if already initialized
if (isElectricInitialized()) {
setInitialized(true);
// Skip on server side
if (typeof window === "undefined") return;
// If no user is logged in, don't initialize Electric
// The app can still function without real-time sync for non-authenticated pages
if (!isUserLoaded || !user?.id) {
// If we had a previous user and now logged out, cleanup
if (previousUserIdRef.current && isElectricInitialized()) {
console.log("[ElectricProvider] User logged out, cleaning up...");
cleanupElectric().then(() => {
previousUserIdRef.current = null;
setElectricClient(null);
});
}
return;
}
const userId = String(user.id);
// If already initialized for THIS user, skip
if (electricClient && previousUserIdRef.current === userId) {
return;
}
// Prevent concurrent initialization attempts
if (initializingRef.current) {
return;
}
// User changed or first initialization
initializingRef.current = true;
let mounted = true;
async function init() {
try {
await initElectric();
console.log(`[ElectricProvider] Initializing for user: ${userId}`);
// If different user was previously initialized, cleanup will happen inside initElectric
const client = await initElectric(userId);
if (mounted) {
setInitialized(true);
previousUserIdRef.current = userId;
setElectricClient(client);
setError(null);
console.log(`[ElectricProvider] ✅ Ready for user: ${userId}`);
}
} catch (err) {
console.error("Failed to initialize Electric SQL:", err);
console.error("[ElectricProvider] Failed to initialize:", err);
if (mounted) {
setError(err instanceof Error ? err : new Error("Failed to initialize Electric SQL"));
// Don't block rendering if Electric SQL fails - app can still work
setInitialized(true);
// Set client to null so hooks know initialization failed
setElectricClient(null);
}
} finally {
if (mounted) {
initializingRef.current = false;
}
}
}
@ -48,22 +99,38 @@ export function ElectricProvider({ children }: ElectricProviderProps) {
return () => {
mounted = false;
};
}, []);
}, [user?.id, isUserLoaded, electricClient]);
// Show loading state only briefly, then render children
// Electric SQL will sync in the background
if (!initialized) {
// For non-authenticated pages (like landing page), render immediately with null context
// Also render immediately if user query failed (e.g., token expired)
if (!isUserLoaded || !user?.id || isUserError) {
return (
<div className="flex items-center justify-center min-h-screen">
<div className="text-muted-foreground">Initializing...</div>
</div>
<ElectricContext.Provider value={null}>
{children}
</ElectricContext.Provider>
);
}
// If there's an error, still render children but log the error
if (error) {
console.warn("Electric SQL initialization failed, notifications may not sync:", error.message);
// Show loading state while initializing for authenticated users
if (!electricClient && !error) {
return (
<ElectricContext.Provider value={null}>
<div className="flex items-center justify-center min-h-screen">
<div className="text-muted-foreground">Initializing...</div>
</div>
</ElectricContext.Provider>
);
}
return <>{children}</>;
// If there's an error, still render but warn
if (error) {
console.warn("[ElectricProvider] Initialization failed, sync may not work:", error.message);
}
// Provide the Electric client to children
return (
<ElectricContext.Provider value={electricClient}>
{children}
</ElectricContext.Provider>
);
}

254
surfsense_web/hooks/use-connectors-electric.ts
View file

@ -1,24 +1,28 @@
"use client";
import { useEffect, useState, useCallback, useRef } from "react";
import {
initElectric,
isElectricInitialized,
type ElectricClient,
type SyncHandle,
} from "@/lib/electric/client";
import { useElectricClient } from "@/lib/electric/context";
import type { SyncHandle } from "@/lib/electric/client";
import type { SearchSourceConnector } from "@/contracts/types/connector.types";
/**
* Hook for managing connectors 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 useConnectorsElectric(searchSpaceId: number | string | null) {
const [electric, setElectric] = useState<ElectricClient | null>(null);
// Get Electric client from context - ElectricProvider handles initialization
const electricClient = useElectricClient();
const [connectors, setConnectors] = useState<SearchSourceConnector[]>([]);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<Error | null>(null);
const syncHandleRef = useRef<SyncHandle | null>(null);
const liveQueryRef = useRef<{ unsubscribe: () => void } | null>(null);
const syncKeyRef = useRef<string | null>(null);
// Transform connector data from Electric SQL/PGlite to match expected format
// Converts Date objects to ISO strings as expected by Zod schema
function transformConnector(connector: any): SearchSourceConnector {
return {
...connector,
@ -36,69 +40,57 @@ export function useConnectorsElectric(searchSpaceId: number | string | null) {
? typeof connector.created_at === "string"
? connector.created_at
: new Date(connector.created_at).toISOString()
: new Date().toISOString(), // fallback
: new Date().toISOString(),
};
}
// Initialize Electric SQL and start syncing with real-time updates
// Start syncing when Electric client is available
useEffect(() => {
if (!searchSpaceId) {
setLoading(false);
setConnectors([]);
// Wait for both searchSpaceId and Electric client to be available
if (!searchSpaceId || !electricClient) {
setLoading(!electricClient); // Still loading if waiting for Electric
if (!searchSpaceId) {
setConnectors([]);
}
return;
}
// Create a unique key for this sync to prevent duplicate subscriptions
const syncKey = `connectors_${searchSpaceId}`;
if (syncKeyRef.current === syncKey) {
// Already syncing for this search space
return;
}
let mounted = true;
syncKeyRef.current = syncKey;
async function init() {
async function startSync() {
try {
const electricClient = await initElectric();
if (!mounted) return;
console.log("[useConnectorsElectric] Starting sync for search space:", searchSpaceId);
setElectric(electricClient);
// Start syncing connectors for this search space via Electric SQL
console.log("Starting Electric SQL sync for connectors, search_space_id:", searchSpaceId);
// Use numeric format for WHERE clause (PGlite sync plugin expects this format)
const handle = await electricClient.syncShape({
table: "search_source_connectors",
where: `search_space_id = ${searchSpaceId}`,
primaryKey: ["id"],
});
console.log("Electric SQL sync started for connectors:", {
console.log("[useConnectorsElectric] Sync started:", {
isUpToDate: handle.isUpToDate,
hasStream: !!handle.stream,
hasInitialSyncPromise: !!handle.initialSyncPromise,
});
// Optimized: Check if already up-to-date before waiting
if (handle.isUpToDate) {
console.log("Connectors sync already up-to-date, skipping wait");
} else if (handle.initialSyncPromise) {
// Only wait if not already up-to-date
console.log("Waiting for initial connectors sync to complete...");
// Wait for initial sync with timeout
if (!handle.isUpToDate && handle.initialSyncPromise) {
try {
// Use Promise.race with a shorter timeout to avoid long waits
await Promise.race([
handle.initialSyncPromise,
new Promise((resolve) => setTimeout(resolve, 2000)), // Max 2s wait
new Promise((resolve) => setTimeout(resolve, 2000)),
]);
console.log("Initial connectors sync promise resolved or timed out, checking status:", {
isUpToDate: handle.isUpToDate,
});
} catch (syncErr) {
console.error("Initial connectors sync failed:", syncErr);
console.error("[useConnectorsElectric] Initial sync failed:", syncErr);
}
}
// Check status after waiting
console.log("Connectors sync status after waiting:", {
isUpToDate: handle.isUpToDate,
hasStream: !!handle.stream,
});
if (!mounted) {
handle.unsubscribe();
return;
@ -108,108 +100,104 @@ export function useConnectorsElectric(searchSpaceId: number | string | null) {
setLoading(false);
setError(null);
// Fetch connectors after sync is complete (we already waited above)
await fetchConnectors(electricClient.db);
// Fetch initial connectors
await fetchConnectors();
// Set up real-time updates using PGlite live queries
// Electric SQL syncs data to PGlite in real-time via HTTP streaming
// PGlite live queries detect when the synced data changes and trigger callbacks
try {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const db = electricClient.db as any;
// Use PGlite's live query API for real-time updates
// CORRECT API: await db.live.query() then use .subscribe()
if (db.live?.query && typeof db.live.query === "function") {
// IMPORTANT: db.live.query() returns a Promise - must await it!
const liveQuery = await db.live.query(
`SELECT * FROM search_source_connectors WHERE search_space_id = $1 ORDER BY created_at DESC`,
[searchSpaceId]
);
if (!mounted) {
liveQuery.unsubscribe?.();
return;
}
// Set initial results immediately from the resolved query
if (liveQuery.initialResults?.rows) {
console.log(
"📋 Initial live query results for connectors:",
liveQuery.initialResults.rows.length
);
setConnectors(liveQuery.initialResults.rows.map(transformConnector));
} else if (liveQuery.rows) {
// Some versions have rows directly on the result
console.log(
"📋 Initial live query results for connectors (direct):",
liveQuery.rows.length
);
setConnectors(liveQuery.rows.map(transformConnector));
}
// Subscribe to changes - this is the correct API!
// The callback fires automatically when Electric SQL syncs new data to PGlite
if (typeof liveQuery.subscribe === "function") {
liveQuery.subscribe((result: { rows: any[] }) => {
if (mounted && result.rows) {
console.log("🔄 Connectors updated via live query:", result.rows.length);
setConnectors(result.rows.map(transformConnector));
}
});
// Store unsubscribe function for cleanup
liveQueryRef.current = liveQuery;
}
} else {
console.warn("PGlite live query API not available, falling back to polling");
}
} catch (liveQueryErr) {
console.error("Failed to set up live query for connectors:", liveQueryErr);
// Don't fail completely - we still have the initial fetch
}
// Set up live query for real-time updates
await setupLiveQuery();
} catch (err) {
console.error("Failed to initialize Electric SQL for connectors:", err);
if (mounted) {
setError(
err instanceof Error
? err
: new Error("Failed to initialize Electric SQL for connectors")
);
setLoading(false);
}
if (!mounted) return;
console.error("[useConnectorsElectric] Failed to start sync:", err);
setError(err instanceof Error ? err : new Error("Failed to sync connectors"));
setLoading(false);
}
}
init();
async function fetchConnectors() {
try {
const result = await electricClient.db.query(
`SELECT * FROM search_source_connectors WHERE search_space_id = $1 ORDER BY created_at DESC`,
[searchSpaceId]
);
if (mounted) {
setConnectors((result.rows || []).map(transformConnector));
}
} catch (err) {
console.error("[useConnectorsElectric] 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 * FROM search_source_connectors WHERE search_space_id = $1 ORDER BY created_at DESC`,
[searchSpaceId]
);
if (!mounted) {
liveQuery.unsubscribe?.();
return;
}
// Set initial results
if (liveQuery.initialResults?.rows) {
setConnectors(liveQuery.initialResults.rows.map(transformConnector));
} else if (liveQuery.rows) {
setConnectors(liveQuery.rows.map(transformConnector));
}
// Subscribe to changes
if (typeof liveQuery.subscribe === "function") {
liveQuery.subscribe((result: { rows: any[] }) => {
if (mounted && result.rows) {
setConnectors(result.rows.map(transformConnector));
}
});
}
if (typeof liveQuery.unsubscribe === "function") {
liveQueryRef.current = liveQuery;
}
}
} catch (liveErr) {
console.error("[useConnectorsElectric] Failed to set up live query:", liveErr);
}
}
startSync();
return () => {
mounted = false;
syncHandleRef.current?.unsubscribe?.();
liveQueryRef.current?.unsubscribe?.();
syncHandleRef.current = null;
liveQueryRef.current = null;
syncKeyRef.current = null;
if (syncHandleRef.current) {
syncHandleRef.current.unsubscribe();
syncHandleRef.current = null;
}
if (liveQueryRef.current) {
liveQueryRef.current.unsubscribe();
liveQueryRef.current = null;
}
};
}, [searchSpaceId]);
async function fetchConnectors(db: any) {
try {
const result = await db.query(
`SELECT * FROM search_source_connectors WHERE search_space_id = $1 ORDER BY created_at DESC`,
[searchSpaceId]
);
console.log("📋 Fetched connectors from PGlite:", result.rows?.length || 0);
setConnectors((result.rows || []).map(transformConnector));
} catch (err) {
console.error("Failed to fetch connectors from PGlite:", err);
}
}
}, [searchSpaceId, electricClient]);
// Manual refresh function (optional, for fallback)
const refreshConnectors = useCallback(async () => {
if (!electric) return;
await fetchConnectors(electric.db);
}, [electric]);
if (!electricClient) return;
try {
const result = await electricClient.db.query(
`SELECT * FROM search_source_connectors WHERE search_space_id = $1 ORDER BY created_at DESC`,
[searchSpaceId]
);
setConnectors((result.rows || []).map(transformConnector));
} catch (err) {
console.error("[useConnectorsElectric] Failed to refresh:", err);
}
}, [electricClient, searchSpaceId]);
return { connectors, loading, error, refreshConnectors };
}

234
surfsense_web/hooks/use-documents-electric.ts
View file

@ -1,7 +1,8 @@
"use client";
import { useEffect, useState, useRef, useMemo } from "react";
import { initElectric, type ElectricClient, type SyncHandle } from "@/lib/electric/client";
import { useElectricClient } from "@/lib/electric/context";
import type { SyncHandle } from "@/lib/electric/client";
interface Document {
id: number;
@ -10,13 +11,22 @@ interface Document {
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) {
const [electric, setElectric] = useState<ElectricClient | null>(null);
// Get Electric client from context - ElectricProvider handles initialization
const electricClient = useElectricClient();
const [documents, setDocuments] = useState<Document[]>([]);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<Error | null>(null);
const syncHandleRef = useRef<SyncHandle | null>(null);
const liveQueryRef = useRef<{ unsubscribe: () => void } | null>(null);
const syncKeyRef = useRef<string | null>(null);
// Calculate document type counts from synced documents
const documentTypeCounts = useMemo(() => {
@ -29,26 +39,30 @@ export function useDocumentsElectric(searchSpaceId: number | string | null) {
return counts;
}, [documents]);
// Initialize Electric SQL and start syncing with real-time updates
// Start syncing when Electric client is available
useEffect(() => {
if (!searchSpaceId) {
setLoading(false);
setDocuments([]);
// 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 init() {
async function startSync() {
try {
const electricClient = await initElectric();
if (!mounted) return;
setElectric(electricClient);
// Start syncing documents for this search space via Electric SQL
// Only sync id, document_type, search_space_id columns for efficiency
console.log("Starting Electric SQL sync for documents, search_space_id:", searchSpaceId);
console.log("[useDocumentsElectric] Starting sync for search space:", searchSpaceId);
const handle = await electricClient.syncShape({
table: "documents",
@ -57,38 +71,22 @@ export function useDocumentsElectric(searchSpaceId: number | string | null) {
primaryKey: ["id"],
});
console.log("Electric SQL sync started for documents:", {
console.log("[useDocumentsElectric] Sync started:", {
isUpToDate: handle.isUpToDate,
hasStream: !!handle.stream,
hasInitialSyncPromise: !!handle.initialSyncPromise,
});
// Optimized: Check if already up-to-date before waiting
if (handle.isUpToDate) {
console.log("Documents sync already up-to-date, skipping wait");
} else if (handle.initialSyncPromise) {
// Only wait if not already up-to-date
console.log("Waiting for initial documents sync to complete...");
// Wait for initial sync with timeout
if (!handle.isUpToDate && handle.initialSyncPromise) {
try {
// Use Promise.race with a shorter timeout to avoid long waits
await Promise.race([
handle.initialSyncPromise,
new Promise((resolve) => setTimeout(resolve, 2000)), // Max 2s wait
new Promise((resolve) => setTimeout(resolve, 2000)),
]);
console.log("Initial documents sync promise resolved or timed out, checking status:", {
isUpToDate: handle.isUpToDate,
});
} catch (syncErr) {
console.error("Initial documents sync failed:", syncErr);
console.error("[useDocumentsElectric] Initial sync failed:", syncErr);
}
}
// Check status after waiting
console.log("Documents sync status after waiting:", {
isUpToDate: handle.isUpToDate,
hasStream: !!handle.stream,
});
if (!mounted) {
handle.unsubscribe();
return;
@ -98,104 +96,90 @@ export function useDocumentsElectric(searchSpaceId: number | string | null) {
setLoading(false);
setError(null);
// Fetch documents after sync is complete (we already waited above)
await fetchDocuments(electricClient.db);
// Fetch initial documents
await fetchDocuments();
// Set up real-time updates using PGlite live queries
// Electric SQL syncs data to PGlite in real-time via HTTP streaming
// PGlite live queries detect when the synced data changes and trigger callbacks
try {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const db = electricClient.db as any;
// Use PGlite's live query API for real-time updates
// CORRECT API: await db.live.query() then use .subscribe()
if (db.live?.query && typeof db.live.query === "function") {
// IMPORTANT: db.live.query() returns a Promise - must await it!
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 immediately from the resolved query
if (liveQuery.initialResults?.rows) {
console.log(
"📋 Initial live query results for documents:",
liveQuery.initialResults.rows.length
);
setDocuments(liveQuery.initialResults.rows);
} else if (liveQuery.rows) {
// Some versions have rows directly on the result
console.log(
"📋 Initial live query results for documents (direct):",
liveQuery.rows.length
);
setDocuments(liveQuery.rows);
}
// Subscribe to changes - this is the correct API!
// The callback fires automatically when Electric SQL syncs new data to PGlite
if (typeof liveQuery.subscribe === "function") {
liveQuery.subscribe((result: { rows: Document[] }) => {
if (mounted && result.rows) {
console.log("🔄 Documents updated via live query:", result.rows.length);
setDocuments(result.rows);
}
});
// Store unsubscribe function for cleanup
liveQueryRef.current = liveQuery;
}
} else {
console.warn(
"PGlite live query API not available for documents, falling back to polling"
);
}
} catch (liveQueryErr) {
console.error("Failed to set up live query for documents:", liveQueryErr);
// Don't fail completely - we still have the initial fetch
}
// Set up live query for real-time updates
await setupLiveQuery();
} catch (err) {
console.error("Failed to initialize Electric SQL for documents:", err);
if (mounted) {
setError(
err instanceof Error
? err
: new Error("Failed to initialize Electric SQL for documents")
);
setLoading(false);
}
if (!mounted) return;
console.error("[useDocumentsElectric] Failed to start sync:", err);
setError(err instanceof Error ? err : new Error("Failed to sync documents"));
setLoading(false);
}
}
init();
async function fetchDocuments() {
try {
const result = await electricClient.db.query<Document>(
`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;
syncHandleRef.current?.unsubscribe?.();
liveQueryRef.current?.unsubscribe?.();
syncHandleRef.current = null;
liveQueryRef.current = null;
syncKeyRef.current = null;
if (syncHandleRef.current) {
syncHandleRef.current.unsubscribe();
syncHandleRef.current = null;
}
if (liveQueryRef.current) {
liveQueryRef.current.unsubscribe();
liveQueryRef.current = null;
}
};
}, [searchSpaceId]);
async function fetchDocuments(db: any) {
try {
const result = await db.query(
`SELECT id, document_type, search_space_id, created_at FROM documents WHERE search_space_id = $1 ORDER BY created_at DESC`,
[searchSpaceId]
);
console.log("📋 Fetched documents from PGlite:", result.rows?.length || 0);
setDocuments(result.rows || []);
} catch (err) {
console.error("Failed to fetch documents from PGlite:", err);
}
}
}, [searchSpaceId, electricClient]);
return { documentTypeCounts, loading, error };
}

243
surfsense_web/hooks/use-notifications.ts
View file

@ -1,81 +1,81 @@
"use client";
import { useEffect, useState, useCallback, useRef } from "react";
import { initElectric, type ElectricClient, type SyncHandle } from "@/lib/electric/client";
import { useElectricClient } from "@/lib/electric/context";
import type { SyncHandle } from "@/lib/electric/client";
import type { Notification } from "@/contracts/types/notification.types";
import { authenticatedFetch } from "@/lib/auth-utils";
export type { Notification } from "@/contracts/types/notification.types";
export function useNotifications(userId: string | null) {
const [electric, setElectric] = useState<ElectricClient | null>(null);
/**
* Hook for managing notifications 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
*
* @param userId - The user ID to fetch notifications for
* @param searchSpaceId - The search space ID to filter notifications (null shows global notifications only)
*/
export function useNotifications(userId: string | null, searchSpaceId: number | null) {
// Get Electric client from context - ElectricProvider handles initialization
const electricClient = useElectricClient();
const [notifications, setNotifications] = useState<Notification[]>([]);
const [loading, setLoading] = useState(true);
const [error, setError] = useState<Error | null>(null);
const syncHandleRef = useRef<SyncHandle | null>(null);
const liveQueryRef = useRef<{ unsubscribe: () => void } | null>(null);
// Use ref instead of state to track initialization - prevents cleanup from running when set
const initializedRef = useRef(false);
const syncKeyRef = useRef<string | null>(null);
// Initialize Electric SQL and start syncing with real-time updates
// Start syncing when Electric client is available
useEffect(() => {
// Use ref to prevent re-initialization without triggering cleanup
if (!userId || initializedRef.current) return;
initializedRef.current = true;
// Wait for both userId and Electric client to be available
if (!userId || !electricClient) {
setLoading(!electricClient); // Still loading if waiting for Electric
return;
}
// Create a unique key for this sync - includes searchSpaceId for proper tracking
// Note: We sync ALL user notifications but filter by searchSpaceId in queries (memory efficient)
const syncKey = `notifications_${userId}_space_${searchSpaceId ?? "global"}`;
if (syncKeyRef.current === syncKey) {
// Already syncing for this user/searchSpace combo
return;
}
let mounted = true;
syncKeyRef.current = syncKey;
async function init() {
async function startSync() {
try {
const electricClient = await initElectric();
if (!mounted) return;
console.log("[useNotifications] Starting sync for user:", userId, "searchSpace:", searchSpaceId);
setElectric(electricClient);
// Start syncing notifications for this user via Electric SQL
// Note: user_id is stored as TEXT in PGlite (UUID from backend is converted)
console.log("Starting Electric SQL sync for user:", userId);
// Use string format for WHERE clause (PGlite sync plugin expects this format)
// The user_id is a UUID string, so we need to quote it properly
// Sync ALL notifications for this user (one subscription for all search spaces)
// This is memory efficient - we filter by searchSpaceId in queries only
const handle = await electricClient.syncShape({
table: "notifications",
where: `user_id = '${userId}'`,
primaryKey: ["id"],
});
console.log("Electric SQL sync started:", {
console.log("[useNotifications] Sync started:", {
isUpToDate: handle.isUpToDate,
hasStream: !!handle.stream,
hasInitialSyncPromise: !!handle.initialSyncPromise,
});
// Optimized: Check if already up-to-date before waiting
if (handle.isUpToDate) {
console.log("Sync already up-to-date, skipping wait");
} else if (handle.initialSyncPromise) {
// Only wait if not already up-to-date
console.log("Waiting for initial sync to complete...");
// Wait for initial sync with timeout
if (!handle.isUpToDate && handle.initialSyncPromise) {
try {
// Use Promise.race with a shorter timeout to avoid long waits
await Promise.race([
handle.initialSyncPromise,
new Promise((resolve) => setTimeout(resolve, 2000)), // Max 2s wait
new Promise((resolve) => setTimeout(resolve, 2000)),
]);
console.log("Initial sync promise resolved or timed out, checking status:", {
isUpToDate: handle.isUpToDate,
});
} catch (syncErr) {
console.error("Initial sync failed:", syncErr);
console.error("[useNotifications] Initial sync failed:", syncErr);
}
}
// Check status after waiting
console.log("Sync status after waiting:", {
isUpToDate: handle.isUpToDate,
hasStream: !!handle.stream,
});
if (!mounted) {
handle.unsubscribe();
return;
@ -85,119 +85,88 @@ export function useNotifications(userId: string | null) {
setLoading(false);
setError(null);
// Fetch notifications after sync is complete (we already waited above)
await fetchNotifications(electricClient.db);
// Fetch initial notifications
await fetchNotifications();
// Set up real-time updates using PGlite live queries
// Electric SQL syncs data to PGlite in real-time via HTTP streaming
// PGlite live queries detect when the synced data changes and trigger callbacks
try {
// eslint-disable-next-line @typescript-eslint/no-explicit-any
const db = electricClient.db as any;
// Use PGlite's live query API for real-time updates
// CORRECT API: await db.live.query() then use .subscribe()
if (db.live?.query && typeof db.live.query === "function") {
// IMPORTANT: db.live.query() returns a Promise - must await it!
const liveQuery = await db.live.query(
`SELECT * FROM notifications WHERE user_id = $1 ORDER BY created_at DESC`,
[userId]
);
if (!mounted) {
liveQuery.unsubscribe?.();
return;
}
// Set initial results immediately from the resolved query
if (liveQuery.initialResults?.rows) {
console.log("📋 Initial live query results:", liveQuery.initialResults.rows.length);
setNotifications(liveQuery.initialResults.rows);
} else if (liveQuery.rows) {
// Some versions have rows directly on the result
console.log("📋 Initial live query results (direct):", liveQuery.rows.length);
setNotifications(liveQuery.rows);
}
// Subscribe to changes - this is the correct API!
// The callback fires automatically when Electric SQL syncs new data to PGlite
if (typeof liveQuery.subscribe === "function") {
liveQuery.subscribe((result: { rows: Notification[] }) => {
console.log(
"🔔 Live query update received:",
result.rows?.length || 0,
"notifications"
);
if (mounted && result.rows) {
setNotifications(result.rows);
}
});
console.log("✅ Real-time notifications enabled via PGlite live queries");
} else {
console.warn("⚠️ Live query subscribe method not available");
}
// Store for cleanup
if (typeof liveQuery.unsubscribe === "function") {
liveQueryRef.current = liveQuery;
}
} else {
console.error("❌ PGlite live queries not available - db.live.query is not a function");
console.log("db.live:", db.live);
}
} catch (liveErr) {
console.error("❌ Failed to set up real-time updates:", liveErr);
}
// Set up live query for real-time updates
await setupLiveQuery();
} catch (err) {
if (!mounted) return;
console.error("Failed to initialize Electric SQL:", err);
setError(err instanceof Error ? err : new Error("Failed to initialize Electric SQL"));
// Still mark as loaded so the UI doesn't block
console.error("[useNotifications] Failed to start sync:", err);
setError(err instanceof Error ? err : new Error("Failed to sync notifications"));
setLoading(false);
}
}
async function fetchNotifications(
db: InstanceType<typeof import("@electric-sql/pglite").PGlite>
) {
async function fetchNotifications() {
try {
// Debug: Check all notifications first
const allNotifications = await db.query<Notification>(
`SELECT * FROM notifications ORDER BY created_at DESC`
);
console.log(
"All notifications in PGlite:",
allNotifications.rows?.length || 0,
allNotifications.rows
);
// Use PGlite's query method (not exec for SELECT queries)
const result = await db.query<Notification>(
// Filter by user_id AND searchSpaceId (or global notifications where search_space_id IS NULL)
const result = await electricClient.db.query<Notification>(
`SELECT * FROM notifications
WHERE user_id = $1
AND (search_space_id = $2 OR search_space_id IS NULL)
ORDER BY created_at DESC`,
[userId]
[userId, searchSpaceId]
);
console.log(`Notifications for user ${userId}:`, result.rows?.length || 0, result.rows);
if (mounted) {
// PGlite query returns { rows: [] } format
setNotifications(result.rows || []);
}
} catch (err) {
console.error("Failed to fetch notifications:", err);
// Log more details for debugging
console.error("Error details:", err);
console.error("[useNotifications] Failed to fetch:", err);
}
}
init();
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") {
// Filter by user_id AND searchSpaceId (or global notifications)
const liveQuery = await db.live.query(
`SELECT * FROM notifications
WHERE user_id = $1
AND (search_space_id = $2 OR search_space_id IS NULL)
ORDER BY created_at DESC`,
[userId, searchSpaceId]
);
if (!mounted) {
liveQuery.unsubscribe?.();
return;
}
// Set initial results
if (liveQuery.initialResults?.rows) {
setNotifications(liveQuery.initialResults.rows);
} else if (liveQuery.rows) {
setNotifications(liveQuery.rows);
}
// Subscribe to changes
if (typeof liveQuery.subscribe === "function") {
liveQuery.subscribe((result: { rows: Notification[] }) => {
if (mounted && result.rows) {
setNotifications(result.rows);
}
});
}
if (typeof liveQuery.unsubscribe === "function") {
liveQueryRef.current = liveQuery;
}
}
} catch (liveErr) {
console.error("[useNotifications] Failed to set up live query:", liveErr);
}
}
startSync();
return () => {
mounted = false;
// Reset initialization state so we can reinitialize with a new userId
initializedRef.current = false;
setLoading(true);
syncKeyRef.current = null;
if (syncHandleRef.current) {
syncHandleRef.current.unsubscribe();
syncHandleRef.current = null;
@ -207,15 +176,11 @@ export function useNotifications(userId: string | null) {
liveQueryRef.current = null;
}
};
// Only depend on userId - using ref for initialization tracking to prevent cleanup issues
// eslint-disable-next-line react-hooks/exhaustive-deps
}, [userId]);
}, [userId, searchSpaceId, electricClient]);
// Mark notification as read via backend API
// Electric SQL will automatically sync the change to all clients
const markAsRead = useCallback(async (notificationId: number) => {
try {
// Call backend API - Electric SQL will sync the change automatically
const response = await authenticatedFetch(
`${process.env.NEXT_PUBLIC_FASTAPI_BACKEND_URL}/api/v1/notifications/${notificationId}/read`,
{ method: "PATCH" }
@ -226,8 +191,6 @@ export function useNotifications(userId: string | null) {
throw new Error(error.detail || "Failed to mark notification as read");
}
// Electric SQL will sync the change from PostgreSQL to PGlite automatically
// The live query subscription will update the UI
return true;
} catch (err) {
console.error("Failed to mark notification as read:", err);
@ -236,10 +199,8 @@ export function useNotifications(userId: string | null) {
}, []);
// Mark all notifications as read via backend API
// Electric SQL will automatically sync the changes to all clients
const markAllAsRead = useCallback(async () => {
try {
// Call backend API - Electric SQL will sync all changes automatically
const response = await authenticatedFetch(
`${process.env.NEXT_PUBLIC_FASTAPI_BACKEND_URL}/api/v1/notifications/read-all`,
{ method: "PATCH" }
@ -250,8 +211,6 @@ export function useNotifications(userId: string | null) {
throw new Error(error.detail || "Failed to mark all notifications as read");
}
// Electric SQL will sync the changes from PostgreSQL to PGlite automatically
// The live query subscription will update the UI
return true;
} catch (err) {
console.error("Failed to mark all notifications as read:", err);

321
surfsense_web/lib/electric/client.ts
View file

@ -1,10 +1,15 @@
/**
* Electric SQL client setup for ElectricSQL 1.x with PGlite
*
* This uses the new ElectricSQL 1.x architecture:
* - PGlite: In-browser PostgreSQL database (local storage)
* - @electric-sql/pglite-sync: Sync plugin to sync Electric shapes into PGlite
* - @electric-sql/client: HTTP client for subscribing to shapes
* USER-SPECIFIC DATABASE ARCHITECTURE:
* - Each user gets their own IndexedDB database: idb://surfsense-{userId}-v{version}
* - On login: cleanup databases from other users, then initialize current user's DB
* - On logout: best-effort cleanup (not relied upon)
*
* This ensures:
* 1. Complete user isolation (data can never leak between users)
* 2. Self-healing on login (stale databases are cleaned up)
* 3. Works even if logout cleanup fails
*/
import { PGlite } from "@electric-sql/pglite";
@ -14,6 +19,7 @@ import { live } from "@electric-sql/pglite/live";
// Types
export interface ElectricClient {
db: PGlite;
userId: string;
syncShape: (options: SyncShapeOptions) => Promise<SyncHandle>;
}
@ -33,14 +39,21 @@ export interface SyncHandle {
initialSyncPromise?: Promise<void>;
}
// Singleton instance
// Singleton state - now tracks the user ID
let electricClient: ElectricClient | null = null;
let currentUserId: string | null = null;
let isInitializing = false;
let initPromise: Promise<ElectricClient> | null = null;
// Cache for sync handles to prevent duplicate subscriptions (memory optimization)
const activeSyncHandles = new Map<string, SyncHandle>();
// Version for sync state - increment this to force fresh sync when Electric config changes
// Incremented to v4 to fix sync completion issues
const SYNC_VERSION = 4;
// Incremented to v5 for user-specific database architecture
const SYNC_VERSION = 5;
// Database name prefix for identifying SurfSense databases
const DB_PREFIX = "surfsense-";
// Get Electric URL from environment
function getElectricUrl(): string {
@ -51,24 +64,100 @@ function getElectricUrl(): string {
}
/**
* Initialize the Electric SQL client with PGlite and sync plugin
* Get the database name for a specific user
*/
export async function initElectric(): Promise<ElectricClient> {
if (electricClient) {
function getDbName(userId: string): string {
return `idb://${DB_PREFIX}${userId}-v${SYNC_VERSION}`;
}
/**
* Clean up databases from OTHER users (not the current user)
* This is called on login to ensure clean state
*/
async function cleanupOtherUserDatabases(currentUserId: string): Promise<void> {
if (typeof window === "undefined" || !window.indexedDB) {
return;
}
try {
// Try to list all databases (not supported in all browsers)
if (typeof window.indexedDB.databases === "function") {
const databases = await window.indexedDB.databases();
for (const dbInfo of databases) {
const dbName = dbInfo.name;
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}`);
continue;
}
// Delete databases from other users
try {
console.log(`[Electric] Deleting stale database: ${dbName}`);
window.indexedDB.deleteDatabase(dbName);
} catch (deleteErr) {
console.warn(`[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);
}
}
/**
* Initialize the Electric SQL client for a specific user
*
* KEY BEHAVIORS:
* 1. If already initialized for the SAME user, returns existing client
* 2. If initialized for a DIFFERENT user, closes old client and creates new one
* 3. On first init, cleans up databases from other users
*
* @param userId - The current user's ID (required)
*/
export async function initElectric(userId: string): Promise<ElectricClient> {
if (!userId) {
throw new Error("userId is required for Electric initialization");
}
// If already initialized for this user, return existing client
if (electricClient && currentUserId === userId) {
return electricClient;
}
// 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...`);
await cleanupElectric();
}
// If already initializing, wait for it
if (isInitializing && initPromise) {
return initPromise;
}
isInitializing = true;
currentUserId = userId;
initPromise = (async () => {
try {
// Create PGlite instance with Electric sync plugin and live queries
// Include version in database name to force fresh sync when Electric config changes
// STEP 1: Clean up databases from other users (login-time cleanup)
console.log("[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}`);
const db = await PGlite.create({
dataDir: `idb://surfsense-notifications-v${SYNC_VERSION}`,
dataDir: dbName,
relaxedDurability: true,
extensions: {
// Enable debug mode in electricSync to see detailed sync logs
@ -77,7 +166,7 @@ export async function initElectric(): Promise<ElectricClient> {
},
});
// Create the notifications table schema in PGlite
// STEP 3: Create the notifications table schema in PGlite
// This matches the backend schema
await db.exec(`
CREATE TABLE IF NOT EXISTS notifications (
@ -137,12 +226,23 @@ export async function initElectric(): Promise<ElectricClient> {
const electricUrl = getElectricUrl();
// Create the client wrapper
// STEP 4: Create the client wrapper
electricClient = {
db,
userId,
syncShape: async (options: SyncShapeOptions): Promise<SyncHandle> => {
const { table, where, columns, primaryKey = ["id"] } = options;
// Create cache key for this sync shape
const cacheKey = `${table}_${where || "all"}_${columns?.join(",") || "all"}`;
// 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}`);
return existingHandle;
}
// Build params for the shape request
// Electric SQL expects params as URL query parameters
const params: Record<string, string> = { table };
@ -177,15 +277,15 @@ export async function initElectric(): Promise<ElectricClient> {
if (columns) params.columns = columns.join(",");
console.log("Syncing shape with params:", params);
console.log("Electric URL:", `${electricUrl}/v1/shape`);
console.log("Where clause:", where, "Validated:", validatedWhere);
console.log("[Electric] Syncing shape with params:", params);
console.log("[Electric] Electric URL:", `${electricUrl}/v1/shape`);
console.log("[Electric] Where clause:", where, "Validated:", validatedWhere);
try {
// Debug: Test Electric SQL connection directly first
// Use validatedWhere to ensure proper URL encoding
const testUrl = `${electricUrl}/v1/shape?table=${table}&offset=-1${validatedWhere ? `&where=${encodeURIComponent(validatedWhere)}` : ""}`;
console.log("Testing Electric SQL directly:", testUrl);
console.log("[Electric] Testing Electric SQL directly:", testUrl);
try {
const testResponse = await fetch(testUrl);
const testHeaders = {
@ -193,15 +293,15 @@ export async function initElectric(): Promise<ElectricClient> {
offset: testResponse.headers.get("electric-offset"),
upToDate: testResponse.headers.get("electric-up-to-date"),
};
console.log("Direct Electric SQL response headers:", testHeaders);
console.log("[Electric] Direct Electric SQL response headers:", testHeaders);
const testData = await testResponse.json();
console.log(
"Direct Electric SQL data count:",
"[Electric] Direct Electric SQL data count:",
Array.isArray(testData) ? testData.length : "not array",
testData
);
} catch (testErr) {
console.error("Direct Electric SQL test failed:", testErr);
console.error("[Electric] Direct Electric SQL test failed:", testErr);
}
// Use PGlite's electric sync plugin to sync the shape
@ -211,9 +311,10 @@ export async function initElectric(): Promise<ElectricClient> {
// Create a promise that resolves when initial sync is complete
// Using recommended approach: check isUpToDate immediately, watch stream, shorter timeout
// IMPORTANT: We don't unsubscribe from the stream - it must stay active for real-time updates
let resolveInitialSync: () => void;
let rejectInitialSync: (error: Error) => void;
let syncResolved = false;
// Initialize with no-op functions to satisfy TypeScript
let resolveInitialSync: () => void = () => {};
let rejectInitialSync: (error: Error) => void = () => {};
const initialSyncPromise = new Promise<void>((resolve, reject) => {
resolveInitialSync = () => {
@ -232,26 +333,26 @@ export async function initElectric(): Promise<ElectricClient> {
};
// Shorter timeout (5 seconds) as fallback
const timeoutId = setTimeout(() => {
setTimeout(() => {
if (!syncResolved) {
console.warn(
`⚠️ Sync timeout for ${table} - checking isUpToDate one more time...`
`[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(`⚠️ Sync timeout for ${table} - resolving anyway after 5s`);
console.warn(
`[Electric] ⚠️ Sync timeout for ${table} - resolving anyway after 5s`
);
resolveInitialSync();
}
}, 100);
}
}, 5000);
// Store timeout ID for cleanup if needed
// Note: timeout will be cleared if sync completes early
});
// Include userId in shapeKey for user-specific sync state
const shapeConfig = {
shape: {
url: `${electricUrl}/v1/shape`,
@ -263,22 +364,24 @@ export async function initElectric(): Promise<ElectricClient> {
},
table,
primaryKey,
shapeKey: `v${SYNC_VERSION}_${table}_${where?.replace(/[^a-zA-Z0-9]/g, "_") || "all"}`, // Versioned key to force fresh sync when needed
shapeKey: `${userId}_v${SYNC_VERSION}_${table}_${where?.replace(/[^a-zA-Z0-9]/g, "_") || "all"}`, // User-specific versioned key
onInitialSync: () => {
console.log(`✅ Initial sync complete for ${table} - data should now be in PGlite`);
console.log(
`[Electric] ✅ Initial sync complete for ${table} - data should now be in PGlite`
);
resolveInitialSync();
},
onError: (error: Error) => {
console.error(`❌ Shape sync error for ${table}:`, error);
console.error(`[Electric] ❌ Shape sync error for ${table}:`, error);
console.error(
"Error details:",
"[Electric] Error details:",
JSON.stringify(error, Object.getOwnPropertyNames(error))
);
rejectInitialSync(error);
},
};
console.log("syncShapeToTable config:", JSON.stringify(shapeConfig, null, 2));
console.log("[Electric] syncShapeToTable config:", JSON.stringify(shapeConfig, null, 2));
// Type assertion to PGlite with electric extension
const pgWithElectric = db as PGlite & {
@ -295,7 +398,7 @@ export async function initElectric(): Promise<ElectricClient> {
}
// Log the actual shape result structure
console.log("Shape sync result (initial):", {
console.log("[Electric] Shape sync result (initial):", {
hasUnsubscribe: typeof shape?.unsubscribe === "function",
isUpToDate: shape?.isUpToDate,
hasStream: !!shape?.stream,
@ -304,13 +407,15 @@ export async function initElectric(): Promise<ElectricClient> {
// Recommended Approach Step 1: Check isUpToDate immediately
if (shape.isUpToDate) {
console.log(`✅ Sync already up-to-date for ${table} (resuming from previous state)`);
console.log(
`[Electric] ✅ Sync already up-to-date for ${table} (resuming from previous state)`
);
resolveInitialSync();
} else {
// 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("Shape stream details:", {
console.log("[Electric] Shape stream details:", {
shapeHandle: stream?.shapeHandle,
lastOffset: stream?.lastOffset,
isUpToDate: stream?.isUpToDate,
@ -323,12 +428,17 @@ export async function initElectric(): Promise<ElectricClient> {
// 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("Subscribing to shape stream to watch for up-to-date message...");
console.log(
"[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("🔵 Shape stream received messages:", messages?.length || 0);
console.log(
"[Electric] 🔵 Shape stream received messages:",
messages?.length || 0
);
}
// Check if any message indicates sync is complete
@ -342,27 +452,34 @@ export async function initElectric(): Promise<ElectricClient> {
(typeof msg === "object" && "up-to-date" in msg)
) {
if (!syncResolved) {
console.log(`✅ Received up-to-date message for ${table}`);
console.log(
`[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("First message:", JSON.stringify(messages[0], null, 2));
console.log(
"[Electric] First message:",
JSON.stringify(messages[0], null, 2)
);
}
}
// Also check stream's isUpToDate property after receiving messages
if (!syncResolved && stream?.isUpToDate) {
console.log(`✅ Stream isUpToDate is true for ${table}`);
console.log(`[Electric] ✅ Stream isUpToDate is true for ${table}`);
resolveInitialSync();
}
});
// Also check stream's isUpToDate property immediately
if (stream?.isUpToDate) {
console.log(`✅ Stream isUpToDate is true immediately for ${table}`);
console.log(
`[Electric] ✅ Stream isUpToDate is true immediately for ${table}`
);
resolveInitialSync();
}
}
@ -375,7 +492,7 @@ export async function initElectric(): Promise<ElectricClient> {
}
if (shape.isUpToDate || stream?.isUpToDate) {
console.log(`✅ Sync completed (detected via polling) for ${table}`);
console.log(`[Electric] ✅ Sync completed (detected via polling) for ${table}`);
clearInterval(pollInterval);
resolveInitialSync();
}
@ -386,14 +503,19 @@ export async function initElectric(): Promise<ElectricClient> {
clearInterval(pollInterval);
});
} else {
console.warn(`⚠️ No stream available for ${table}, relying on callback and timeout`);
console.warn(
`[Electric] ⚠️ No stream available for ${table}, relying on callback and timeout`
);
}
}
// Return the shape handle - isUpToDate is a getter that reflects current state
return {
// Create the sync handle with proper cleanup
const syncHandle: SyncHandle = {
unsubscribe: () => {
console.log("unsubscribing");
console.log(`[Electric] Unsubscribing from: ${cacheKey}`);
// Remove from cache first
activeSyncHandles.delete(cacheKey);
// Then unsubscribe from the shape
if (shape && typeof shape.unsubscribe === "function") {
shape.unsubscribe();
}
@ -405,30 +527,43 @@ export async function initElectric(): Promise<ElectricClient> {
stream: shape?.stream,
initialSyncPromise, // Expose promise so callers can wait for sync
};
// Cache the sync handle for reuse (memory optimization)
activeSyncHandles.set(cacheKey, syncHandle);
console.log(`[Electric] Cached sync handle for: ${cacheKey} (total cached: ${activeSyncHandles.size})`);
return syncHandle;
} catch (error) {
console.error("Failed to sync shape:", error);
console.error("[Electric] Failed to sync shape:", error);
// Check if Electric SQL server is reachable
try {
const response = await fetch(`${electricUrl}/v1/shape?table=${table}&offset=-1`, {
method: "GET",
});
console.log("Electric SQL server response:", response.status, response.statusText);
console.log(
"[Electric] Electric SQL server response:",
response.status,
response.statusText
);
if (!response.ok) {
console.error("Electric SQL server error:", await response.text());
console.error("[Electric] Electric SQL server error:", await response.text());
}
} catch (fetchError) {
console.error("Cannot reach Electric SQL server:", fetchError);
console.error("Make sure Electric SQL is running at:", electricUrl);
console.error("[Electric] Cannot reach Electric SQL server:", fetchError);
console.error("[Electric] Make sure Electric SQL is running at:", electricUrl);
}
throw error;
}
},
};
console.log("Electric SQL initialized successfully with PGlite");
console.log(`[Electric] ✅ Initialized successfully for user: ${userId}`);
return electricClient;
} catch (error) {
console.error("Failed to initialize Electric SQL:", error);
console.error("[Electric] Failed to initialize:", error);
// Reset state on failure
electricClient = null;
currentUserId = null;
throw error;
} finally {
isInitializing = false;
@ -438,21 +573,87 @@ export async function initElectric(): Promise<ElectricClient> {
return initPromise;
}
/**
* Cleanup Electric SQL - close database and reset singleton
* Called on logout (best-effort) and when switching users
*/
export async function cleanupElectric(): Promise<void> {
if (!electricClient) {
return;
}
const userIdToClean = currentUserId;
console.log(`[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`);
// Copy keys to array to avoid mutation during iteration
const handleKeys = Array.from(activeSyncHandles.keys());
for (const key of handleKeys) {
const handle = activeSyncHandles.get(key);
if (handle) {
try {
handle.unsubscribe();
} catch (err) {
console.warn(`[Electric] Failed to unsubscribe from ${key}:`, err);
}
}
}
// Ensure cache is empty
activeSyncHandles.clear();
try {
// Close the PGlite database connection
await electricClient.db.close();
console.log("[Electric] Database closed");
} catch (error) {
console.error("[Electric] Error closing database:", error);
}
// Reset singleton state
electricClient = null;
currentUserId = null;
isInitializing = false;
initPromise = null;
// Delete the user's IndexedDB database (best-effort cleanup on logout)
if (typeof window !== "undefined" && window.indexedDB && userIdToClean) {
try {
const dbName = `${DB_PREFIX}${userIdToClean}-v${SYNC_VERSION}`;
window.indexedDB.deleteDatabase(dbName);
console.log(`[Electric] Deleted database: ${dbName}`);
} catch (err) {
console.warn("[Electric] Failed to delete database:", err);
}
}
console.log("[Electric] Cleanup complete");
}
/**
* Get the Electric client (throws if not initialized)
*/
export function getElectric(): ElectricClient {
if (!electricClient) {
throw new Error("Electric not initialized. Call initElectric() first.");
throw new Error("Electric not initialized. Call initElectric(userId) first.");
}
return electricClient;
}
/**
* Check if Electric is initialized
* Check if Electric is initialized for a specific user
*/
export function isElectricInitialized(): boolean {
return electricClient !== null;
export function isElectricInitialized(userId?: string): boolean {
if (!electricClient) return false;
if (userId && currentUserId !== userId) return false;
return true;
}
/**
* Get the current user ID that Electric is initialized for
*/
export function getCurrentElectricUserId(): string | null {
return currentUserId;
}
/**

37
surfsense_web/lib/electric/context.ts Normal file
View file

@ -0,0 +1,37 @@
"use client";
import { createContext, useContext } from "react";
import type { ElectricClient } from "./client";
/**
* Context for sharing the Electric SQL client across the app
*
* This ensures:
* 1. Single initialization point (ElectricProvider only)
* 2. No race conditions (hooks wait for context)
* 3. Clean cleanup (ElectricProvider manages lifecycle)
*/
export const ElectricContext = createContext<ElectricClient | null>(null);
/**
* Hook to get the Electric client from context
* Returns null if Electric is not initialized yet
*/
export function useElectricClient(): ElectricClient | null {
return useContext(ElectricContext);
}
/**
* Hook to get the Electric client, throwing if not available
* Use this when you're sure Electric should be initialized
*/
export function useElectricClientOrThrow(): ElectricClient {
const client = useContext(ElectricContext);
if (!client) {
throw new Error(
"Electric client not available. Make sure you're inside ElectricProvider and user is authenticated."
);
}
return client;
}