import { useCallback, useEffect, useRef, useState } from "react"; import { formatHeaders, parseApiError } from "@/lib/api"; import { generateId } from "@/lib/utils"; import { useFileStore } from "@/stores/file-store"; import type { PipelineStep } from "@/stores/pipeline-store"; interface ProcessResult { jobId: string; downloadUrl: string; previewUrl?: string; originalSize: number; processedSize: number; savedFileId?: string; } export interface PipelineProgress { phase: "idle" | "uploading" | "processing" | "complete"; percent: number; stage?: string; elapsed: number; } const IDLE_PROGRESS: PipelineProgress = { phase: "idle", percent: 0, elapsed: 0, }; const UPLOAD_WEIGHT = 15; export function usePipelineProcessor() { const { processing, error, processedUrl, originalSize, processedSize, setProcessing, setError } = useFileStore(); const [progress, setProgress] = useState(IDLE_PROGRESS); const elapsedRef = useRef | null>(null); const xhrRef = useRef(null); const eventSourceRef = useRef(null); const abortRef = useRef(null); const activeJobIdRef = useRef(null); useEffect(() => { const handleVisibilityChange = () => { if (document.visibilityState !== "visible") return; if (!activeJobIdRef.current) return; if (eventSourceRef.current && eventSourceRef.current.readyState === EventSource.OPEN) { return; } if (eventSourceRef.current) { eventSourceRef.current.close(); eventSourceRef.current = null; } const jobId = activeJobIdRef.current; setTimeout(() => { if (!activeJobIdRef.current || activeJobIdRef.current !== jobId) return; try { const es = new EventSource(`/api/v1/jobs/${jobId}/progress`); eventSourceRef.current = es; es.onmessage = (event) => { try { const data = JSON.parse(event.data); if (data.type === "single" && typeof data.percent === "number") { const scaled = UPLOAD_WEIGHT + (data.percent / 100) * (100 - UPLOAD_WEIGHT); setProgress((prev) => ({ ...prev, phase: "processing", percent: Math.max(prev.percent, scaled), stage: data.stage, })); } if (data.type === "batch") { const pct = data.totalFiles > 0 ? 15 + (data.completedFiles / data.totalFiles) * 85 : 15; setProgress((prev) => ({ ...prev, phase: "processing", percent: pct, stage: data.currentFile ? `Processing ${data.currentFile} (${data.completedFiles}/${data.totalFiles})` : `Processing ${data.completedFiles}/${data.totalFiles}`, })); } } catch { // Ignore malformed SSE } }; es.onerror = () => { es.close(); eventSourceRef.current = null; }; } catch { // EventSource creation failed } }, 500); }; document.addEventListener("visibilitychange", handleVisibilityChange); return () => { document.removeEventListener("visibilitychange", handleVisibilityChange); if (elapsedRef.current) clearInterval(elapsedRef.current); if (eventSourceRef.current) eventSourceRef.current.close(); if (xhrRef.current) xhrRef.current.abort(); if (abortRef.current) abortRef.current.abort(); }; }, []); const processSingle = useCallback( (file: File, steps: PipelineStep[]) => { const capturedIndex = useFileStore.getState().selectedIndex; setError(null); useFileStore.getState().updateEntry(capturedIndex, { processedUrl: null, processedPreviewUrl: null, processedFilename: null, status: "processing", error: null, }); setProcessing(true); setProgress({ phase: "uploading", percent: 0, elapsed: 0 }); const startTime = Date.now(); elapsedRef.current = setInterval(() => { setProgress((prev) => ({ ...prev, elapsed: Math.floor((Date.now() - startTime) / 1000), })); }, 1000); const clientJobId = generateId(); activeJobIdRef.current = clientJobId; // Open SSE for real-time progress from the server try { const es = new EventSource(`/api/v1/jobs/${clientJobId}/progress`); eventSourceRef.current = es; es.onmessage = (event) => { try { const data = JSON.parse(event.data); if (data.type !== "single") return; if (typeof data.percent === "number") { const scaled = UPLOAD_WEIGHT + (data.percent / 100) * (100 - UPLOAD_WEIGHT); setProgress((prev) => ({ ...prev, phase: "processing", percent: Math.max(prev.percent, scaled), stage: data.stage, })); } } catch { // Ignore malformed SSE } }; es.onerror = () => { es.close(); eventSourceRef.current = null; }; } catch { // EventSource creation failed -- proceed without SSE } const pipeline = { steps: steps.map((s) => ({ toolId: s.toolId, settings: s.settings })), }; const formData = new FormData(); formData.append("file", file); formData.append("pipeline", JSON.stringify(pipeline)); formData.append("clientJobId", clientJobId); const xhr = new XMLHttpRequest(); xhrRef.current = xhr; xhr.timeout = 600_000; xhr.upload.onprogress = (event) => { if (event.lengthComputable) { const uploadPercent = (event.loaded / event.total) * UPLOAD_WEIGHT; setProgress((prev) => { if (prev.phase !== "uploading") return prev; return { ...prev, percent: uploadPercent }; }); } }; xhr.upload.onload = () => { setProgress((prev) => ({ ...prev, phase: "processing", percent: UPLOAD_WEIGHT, stage: "Processing...", })); }; xhr.onload = () => { if (elapsedRef.current) clearInterval(elapsedRef.current); if (eventSourceRef.current) { eventSourceRef.current.close(); eventSourceRef.current = null; } if (xhr.status >= 200 && xhr.status < 300) { try { const result: ProcessResult = JSON.parse(xhr.responseText); useFileStore.getState().updateEntry(capturedIndex, { processedUrl: result.downloadUrl, processedPreviewUrl: result.previewUrl ?? null, processedFilename: null, status: "completed", originalSize: result.originalSize, processedSize: result.processedSize, ...(result.savedFileId ? { serverFileId: result.savedFileId } : {}), }); } catch { setError("Invalid response from server"); } } else { try { const body = JSON.parse(xhr.responseText); const parsed = parseApiError(body, xhr.status); if (typeof parsed === "object" && parsed.type === "feature_not_installed") { setError( `The "${parsed.featureName}" feature is not installed. Enable it in Settings → AI Features.`, ); } else { setError(parsed as string); } } catch { setError(`Processing failed: ${xhr.status}`); } } setProcessing(false); setProgress(IDLE_PROGRESS); activeJobIdRef.current = null; }; xhr.onerror = () => { if (elapsedRef.current) clearInterval(elapsedRef.current); if (eventSourceRef.current) { eventSourceRef.current.close(); eventSourceRef.current = null; } setError("Network error - check your connection"); setProcessing(false); setProgress(IDLE_PROGRESS); activeJobIdRef.current = null; }; xhr.ontimeout = () => { if (elapsedRef.current) clearInterval(elapsedRef.current); if (eventSourceRef.current) { eventSourceRef.current.close(); eventSourceRef.current = null; } setError("Request timed out - the server may be overloaded. Try again."); setProcessing(false); setProgress(IDLE_PROGRESS); activeJobIdRef.current = null; }; xhr.open("POST", "/api/v1/pipeline/execute"); formatHeaders().forEach((value, key) => { xhr.setRequestHeader(key, value); }); xhr.send(formData); }, [setProcessing, setError], ); const processAll = useCallback( async (files: File[], steps: PipelineStep[]) => { if (files.length === 0) { setError("No files selected"); return; } if (files.length === 1) { processSingle(files[0], steps); return; } const { updateEntry, setBatchZip } = useFileStore.getState(); setError(null); setProcessing(true); setProgress({ phase: "uploading", percent: 0, elapsed: 0 }); const startTime = Date.now(); elapsedRef.current = setInterval(() => { setProgress((prev) => ({ ...prev, elapsed: Math.floor((Date.now() - startTime) / 1000) })); }, 1000); const clientJobId = generateId(); activeJobIdRef.current = clientJobId; // Open SSE before upload for real-time progress try { const es = new EventSource(`/api/v1/jobs/${clientJobId}/progress`); eventSourceRef.current = es; es.onmessage = (event) => { try { const data = JSON.parse(event.data); if (data.type === "batch") { const pct = data.totalFiles > 0 ? 15 + (data.completedFiles / data.totalFiles) * 85 : 15; setProgress((prev) => ({ ...prev, phase: "processing", percent: pct, stage: data.currentFile ? `Processing ${data.currentFile} (${data.completedFiles}/${data.totalFiles})` : `Processing ${data.completedFiles}/${data.totalFiles}`, })); } } catch { /* ignore malformed SSE */ } }; es.onerror = () => { es.close(); eventSourceRef.current = null; }; } catch { /* SSE failed, proceed without */ } const pipeline = { steps: steps.map((s) => ({ toolId: s.toolId, settings: s.settings })), }; const formData = new FormData(); for (const file of files) formData.append("file", file); formData.append("pipeline", JSON.stringify(pipeline)); formData.append("clientJobId", clientJobId); try { abortRef.current = new AbortController(); const response = await fetch("/api/v1/pipeline/batch", { method: "POST", headers: formatHeaders(), body: formData, signal: abortRef.current.signal, }); if (elapsedRef.current) clearInterval(elapsedRef.current); if (eventSourceRef.current) { eventSourceRef.current.close(); eventSourceRef.current = null; } if (!response.ok) { const text = await response.text(); let errorMsg: string; try { const body = JSON.parse(text); if (body.errors && Array.isArray(body.errors) && body.errors.length > 0) { // Show the first file's step-level error (all files typically fail at the same step) const first = body.errors[0]; errorMsg = first.error; if (body.errors.length > 1) { errorMsg += ` (${body.errors.length} files failed)`; } } else { const parsed = parseApiError(body, response.status); if (typeof parsed === "object" && parsed.type === "feature_not_installed") { errorMsg = `The "${parsed.featureName}" feature is not installed. Enable it in Settings → AI Features.`; } else { errorMsg = parsed as string; } } } catch { errorMsg = `Batch processing failed: ${response.status}`; } setError(errorMsg); setProcessing(false); setProgress(IDLE_PROGRESS); return; } const zipBlob = await response.blob(); setBatchZip(zipBlob, "batch-pipeline.zip"); // Extract files from ZIP using fflate const { unzipSync } = await import("fflate"); const zipBuffer = new Uint8Array((await zipBlob.arrayBuffer()) as ArrayBuffer); const extracted = unzipSync(zipBuffer); const entries = useFileStore.getState().entries; let fileResults: Record = {}; try { fileResults = JSON.parse( decodeURIComponent(response.headers.get("X-File-Results") ?? "%7B%7D"), ); } catch { // Malformed header - fall back to empty mapping, all entries marked failed } for (let i = 0; i < entries.length; i++) { const processedName = fileResults[String(i)]; if (processedName && extracted[processedName]) { const blob = new Blob([extracted[processedName] as BlobPart]); updateEntry(i, { processedUrl: URL.createObjectURL(blob), processedFilename: processedName, processedSize: blob.size, status: "completed", error: null, }); } else { updateEntry(i, { status: "failed", error: "File not found in batch results" }); } } setProcessing(false); setProgress(IDLE_PROGRESS); activeJobIdRef.current = null; } catch (err) { if (elapsedRef.current) clearInterval(elapsedRef.current); if (eventSourceRef.current) { eventSourceRef.current.close(); eventSourceRef.current = null; } setError(err instanceof Error ? err.message : "Batch processing failed"); setProcessing(false); setProgress(IDLE_PROGRESS); activeJobIdRef.current = null; } }, [processSingle, setProcessing, setError], ); return { processSingle, processAll, processing, error, downloadUrl: processedUrl, originalSize, processedSize, progress, }; }