photo-editor/public/js/jobs.js

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