/** * jobs.js — Очередь заданий, SSE-подписка, фолбэк-поллинг и выполнение edit/upscale */ import { apiFetch } from "./api.js"; import { getState, setBusy, setCurrentVersionId } from "./state.js"; import { setJobStatus, resetResult, updateDownloadCurrent, versionFileName } from "./ui.js"; import { refreshHistory } from "./history-tree.js"; import { getReferenceState } from "./reference.js"; let eventSource = null; let currentTrackingJob = null; let fallbackPollingTimer = null; export function initJobsEvents() { if (window.EventSource) { connectSSE(); } else { startFallbackPolling(); } } function connectSSE() { if (eventSource) { try { eventSource.close(); } catch { // ignore } } eventSource = new EventSource("/api/events"); eventSource.addEventListener("snapshot", (event) => { try { const activeJobs = JSON.parse(event.data); updateActiveBadge(activeJobs); } catch { // ignore } }); eventSource.addEventListener("job", (event) => { try { const jobData = JSON.parse(event.data); handleJobEvent(jobData); } catch { // ignore } }); eventSource.onerror = () => { if (eventSource) { eventSource.close(); eventSource = null; } startFallbackPolling(); // Try reconnecting SSE after 15s setTimeout(connectSSE, 15000); }; } function startFallbackPolling() { if (fallbackPollingTimer) return; fallbackPollingTimer = setInterval(async () => { try { const data = await apiFetch("/api/jobs"); if (Array.isArray(data?.jobs)) { updateActiveBadge(data.jobs); } } catch { // ignore } }, 3000); } function updateActiveBadge(jobs) { const active = jobs.filter((j) => j.status === "queued" || j.status === "running"); // Could update badge in topbar if needed } function handleJobEvent(jobData) { if (!currentTrackingJob || currentTrackingJob.id !== jobData.jobId) { return; } const { isEdit, onFinish } = currentTrackingJob; if (jobData.status === "queued") { const pos = jobData.queuePosition ? ` (позиция ${jobData.queuePosition})` : ""; setJobStatus("progress", `Задание ожидает в очереди${pos}…`); } else if (jobData.status === "running") { setJobStatus( "progress", isEdit ? "Codex редактирует фотографию… Это обычно занимает 1–3 минуты." : "ComfyUI увеличивает разрешение… Это займёт несколько секунд." ); } else if (jobData.status === "done") { currentTrackingJob = null; onFinish(null, jobData); } else if (jobData.status === "interrupted") { currentTrackingJob = null; const err = new Error( jobData.error || "Задание прервано перезапуском сервера. Запустите заново." ); err.interrupted = true; onFinish(err); } else if (jobData.status === "error") { currentTrackingJob = null; onFinish(new Error(jobData.error || "Ошибка при выполнении задания.")); } } export async function runJob(kind) { const state = getState(); const isEdit = kind === "edit"; if (!state.currentVersionId || state.isBusy) { return; } const promptInput = document.querySelector("#prompt"); const prompt = String(promptInput?.value || "").trim(); if (isEdit && !prompt) { setJobStatus("error", "Введите текст промпта."); return; } setBusy(true); resetResult(); setJobStatus("progress", "Отправка задания на сервер…"); let lastJobParams = null; try { let res; if (isEdit) { const formData = new FormData(); formData.append("sourceId", state.currentVersionId); formData.append("prompt", prompt); const refState = getReferenceState(); if (refState?.source?.type === "file" && refState.source.file) { formData.append("reference", refState.source.file); } else if (refState?.source?.type === "history" && refState.source.versionId) { formData.append("referenceId", refState.source.versionId); } if (refState?.region) { formData.append("region", JSON.stringify(refState.region)); } lastJobParams = { kind, sourceId: state.currentVersionId, prompt, refState }; res = await apiFetch("/api/edit", { method: "POST", body: formData, }); } else { lastJobParams = { kind, sourceId: state.currentVersionId }; res = await apiFetch("/api/upscale", { method: "POST", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ sourceId: state.currentVersionId }), }); } const jobId = res.jobId; if (res.status === "queued") { const pos = res.queuePosition ? ` (позиция ${res.queuePosition})` : ""; setJobStatus("progress", `Задание добавлено в очередь${pos}…`); } else { setJobStatus( "progress", isEdit ? "Codex редактирует фотографию… Это обычно занимает 1–3 минуты." : "ComfyUI увеличивает разрешение… Это займёт несколько секунд." ); } // Wait for completion via SSE or fallback polling await new Promise((resolve, reject) => { currentTrackingJob = { id: jobId, isEdit, onFinish: (err, resultData) => { if (err) reject(err); else resolve(resultData); }, }; // Fallback polling loop in case SSE is blocked or slow pollJobStatus(jobId, isEdit); }); // Fetch full job status to get version and details const finalJob = await apiFetch(`/api/jobs/${jobId}/status`); renderJobSuccess(finalJob, isEdit); } catch (err) { if (err.interrupted) { setJobStatus( "error", "Задание прервано перезапуском сервера. Нажмите кнопку, чтобы повторить попытку." ); } else { setJobStatus("error", err.message || "Ошибка при выполнении задания."); } } finally { setBusy(false); } } async function pollJobStatus(jobId, isEdit) { let attempts = 0; const poll = async () => { if (!currentTrackingJob || currentTrackingJob.id !== jobId) { return; } try { const data = await apiFetch(`/api/jobs/${jobId}/status`); if (!currentTrackingJob || currentTrackingJob.id !== jobId) return; if (data.status === "queued") { const pos = data.queuePosition ? ` (позиция ${data.queuePosition})` : ""; setJobStatus("progress", `Задание ожидает в очереди${pos}…`); setTimeout(poll, 2000); } else if (data.status === "running") { setJobStatus( "progress", isEdit ? "Codex редактирует фотографию… Это обычно занимает 1–3 минуты." : "ComfyUI увеличивает разрешение… Это займёт несколько секунд." ); setTimeout(poll, 2000); } else if (data.status === "done") { const cb = currentTrackingJob?.onFinish; currentTrackingJob = null; if (cb) cb(null, data); } else if (data.status === "interrupted") { const cb = currentTrackingJob?.onFinish; currentTrackingJob = null; const err = new Error( data.error || "Задание прервано перезапуском сервера. Запустите заново." ); err.interrupted = true; if (cb) cb(err); } else if (data.status === "error") { const cb = currentTrackingJob?.onFinish; currentTrackingJob = null; if (cb) cb(new Error(data.error || "Ошибка при выполнении задания.")); } else { setTimeout(poll, 2000); } } catch { attempts++; if (attempts < 30) { setTimeout(poll, 2500); } } }; setTimeout(poll, 2000); } export function renderJobSuccess(data, isEdit) { const version = data.result?.version; const resultImage = document.querySelector("#result-image"); const resultPlaceholder = document.querySelector("#result-placeholder"); const resultLabel = document.querySelector("#result-label"); const downloadLink = document.querySelector("#download-link"); if (!version) { setJobStatus("error", "Задание завершено, но версия не создана."); return; } setJobStatus("idle", ""); if (resultImage) { resultImage.src = version.url; resultImage.hidden = false; } if (resultPlaceholder) resultPlaceholder.hidden = true; if (resultLabel) { resultLabel.textContent = isEdit ? "Codex" : "4× Upscale"; } if (downloadLink) { downloadLink.href = version.url; downloadLink.download = versionFileName(version); downloadLink.hidden = false; } // Refresh history tree and select newly created version refreshHistory(true); }