SurfSense/surfsense_web/hooks/use-connectors-electric.ts

"use client";

import { useEffect, useState, useCallback, useRef } from "react";
import {
	initElectric,
	isElectricInitialized,
	type ElectricClient,
	type SyncHandle,
} from "@/lib/electric/client";
import type { SearchSourceConnector } from "@/contracts/types/connector.types";

export function useConnectorsElectric(searchSpaceId: number | string | null) {
	const [electric, setElectric] = useState<ElectricClient | null>(null);
	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);

	// 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,
			last_indexed_at: connector.last_indexed_at
				? typeof connector.last_indexed_at === "string"
					? connector.last_indexed_at
					: new Date(connector.last_indexed_at).toISOString()
				: null,
			next_scheduled_at: connector.next_scheduled_at
				? typeof connector.next_scheduled_at === "string"
					? connector.next_scheduled_at
					: new Date(connector.next_scheduled_at).toISOString()
				: null,
			created_at: connector.created_at
				? typeof connector.created_at === "string"
					? connector.created_at
					: new Date(connector.created_at).toISOString()
				: new Date().toISOString(), // fallback
		};
	}

	// Initialize Electric SQL and start syncing with real-time updates
	useEffect(() => {
		if (!searchSpaceId) {
			setLoading(false);
			setConnectors([]);
			return;
		}

		let mounted = true;

		async function init() {
			try {
				const electricClient = await initElectric();
				if (!mounted) return;

				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:", {
					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...");
					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
						]);
						console.log("Initial connectors sync promise resolved or timed out, checking status:", {
							isUpToDate: handle.isUpToDate,
						});
					} catch (syncErr) {
						console.error("Initial connectors 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;
				}

				syncHandleRef.current = handle;
				setLoading(false);
				setError(null);

				// Fetch connectors after sync is complete (we already waited above)
				await fetchConnectors(electricClient.db);

				// 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
				}
			} 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);
				}
			}
		}

		init();

		return () => {
			mounted = false;
			syncHandleRef.current?.unsubscribe?.();
			liveQueryRef.current?.unsubscribe?.();
			syncHandleRef.current = null;
			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);
		}
	}

	// Manual refresh function (optional, for fallback)
	const refreshConnectors = useCallback(async () => {
		if (!electric) return;
		await fetchConnectors(electric.db);
	}, [electric]);

	return { connectors, loading, error, refreshConnectors };
}