netfelix-audio-fix/src/features/pipeline/PipelinePage.tsx

import { useCallback, useEffect, useRef, useState } from "react";
import { Button } from "~/shared/components/ui/button";
import { api } from "~/shared/lib/api";
import type { PipelineData } from "~/shared/lib/types";
import { DoneColumn } from "./DoneColumn";
import { ProcessingColumn } from "./ProcessingColumn";
import { QueueColumn } from "./QueueColumn";
import { ReviewColumn } from "./ReviewColumn";

interface Progress {
	id: number;
	seconds: number;
	total: number;
}

interface QueueStatus {
	status: string;
	until?: string;
	seconds?: number;
}

export function PipelinePage() {
	const [data, setData] = useState<PipelineData | null>(null);
	const [progress, setProgress] = useState<Progress | null>(null);
	const [queueStatus, setQueueStatus] = useState<QueueStatus | null>(null);
	const [loading, setLoading] = useState(true);

	const load = useCallback(async () => {
		const pipelineRes = await api.get<PipelineData>("/api/review/pipeline");
		setData(pipelineRes);
		setLoading(false);
	}, []);

	const startQueue = useCallback(async () => {
		await api.post("/api/execute/start");
		load();
	}, [load]);

	useEffect(() => {
		load();
	}, [load]);

	// SSE for live updates. job_update fires on every status change and per-line
	// stdout flush of the running job — without coalescing, the pipeline endpoint
	// (a 500-row review query + counts) would re-run several times per second.
	const reloadTimer = useRef<ReturnType<typeof setTimeout> | null>(null);
	useEffect(() => {
		const scheduleReload = () => {
			if (reloadTimer.current) return;
			reloadTimer.current = setTimeout(() => {
				reloadTimer.current = null;
				load();
			}, 1000);
		};
		const es = new EventSource("/api/execute/events");
		es.addEventListener("job_update", (e) => {
			// When a job leaves 'running' (done / error / cancelled), drop any
			// stale progress so the bar doesn't linger on the next job's card.
			try {
				const upd = JSON.parse((e as MessageEvent).data) as { id: number; status: string };
				if (upd.status !== "running") setProgress(null);
			} catch {
				/* ignore malformed events */
			}
			scheduleReload();
		});
		// plan_update lands ~15s after a job finishes — the post-job jellyfin
		// verification writes verified=1 (or flips the plan back to pending).
		// Without refreshing here the Done column would never promote ✓ to ✓✓.
		es.addEventListener("plan_update", () => {
			scheduleReload();
		});
		es.addEventListener("job_progress", (e) => {
			setProgress(JSON.parse((e as MessageEvent).data));
		});
		es.addEventListener("queue_status", (e) => {
			setQueueStatus(JSON.parse((e as MessageEvent).data));
		});
		return () => {
			es.close();
			if (reloadTimer.current) clearTimeout(reloadTimer.current);
		};
	}, [load]);

	if (loading || !data) return <div className="p-6 text-gray-500">Loading pipeline...</div>;

	return (
		<div className="flex flex-col -mx-3 sm:-mx-5 -mt-4 -mb-12 h-[calc(100vh-3rem)] overflow-hidden">
			<div className="flex items-center justify-between px-6 py-3 border-b shrink-0">
				<h1 className="text-lg font-semibold">Pipeline</h1>
				<div className="flex items-center gap-4">
					<span className="text-sm text-gray-500">{data.doneCount} files in desired state</span>
					<Button variant="primary" size="sm" onClick={startQueue}>
						Start queue
					</Button>
				</div>
			</div>
			<div className="flex flex-1 gap-4 p-4 overflow-x-auto overflow-y-hidden min-h-0">
				<ReviewColumn items={data.review} total={data.reviewTotal} jellyfinUrl={data.jellyfinUrl} onMutate={load} />
				<QueueColumn items={data.queued} jellyfinUrl={data.jellyfinUrl} onMutate={load} />
				<ProcessingColumn items={data.processing} progress={progress} queueStatus={queueStatus} onMutate={load} />
				<DoneColumn items={data.done} onMutate={load} />
			</div>
		</div>
	);
}