photo-editor/test/job-queue.test.js

"use strict";

const { test, describe, beforeEach, afterEach } = require("node:test");
const assert = require("node:assert/strict");
const fs = require("fs");
const path = require("path");
const os = require("os");
const { EventEmitter } = require("events");

const { JobStore } = require("../src/core/jobs/job-store");
const { Queue } = require("../src/core/jobs/queue");
const { EngineRegistry } = require("../src/core/engines/engine");
const { TooManyRequestsError } = require("../src/errors");

describe("job-queue", () => {
  let tempDir;
  let jobFile;

  beforeEach(() => {
    tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "kadr-job-test-"));
    jobFile = path.join(tempDir, "jobs.jsonl");
  });

  afterEach(() => {
    try {
      fs.rmSync(tempDir, { recursive: true, force: true });
    } catch {
      // ignore
    }
  });

  test("processes jobs in FIFO order respecting group concurrency", async () => {
    const jobStore = new JobStore({ file: jobFile });
    const engineRegistry = new EngineRegistry();
    const events = new EventEmitter();

    const executed = [];

    // Fake edit engine with delay
    engineRegistry.register("edit", () => ({
      async run(job) {
        executed.push(`start:${job.id}`);
        await new Promise((r) => setTimeout(r, 50));
        executed.push(`done:${job.id}`);
        return { ok: true };
      },
    }));

    const queue = new Queue({
      concurrencyByGroup: { codex: 1, "comfy-upscale": 2 },
      maxQueuedJobs: 5,
      jobStore,
      engineRegistry,
      engineContext: {},
      events,
    });

    const job1 = { id: "job-1", type: "edit", status: "queued", payload: {} };
    const job2 = { id: "job-2", type: "edit", status: "queued", payload: {} };

    jobStore.saveJob(job1);
    jobStore.saveJob(job2);

    await queue.enqueue(job1);
    const { queuePosition } = await queue.enqueue(job2);

    assert.equal(queuePosition, 2);

    queue.start();

    // Wait until both jobs finish
    while (jobStore.getJob("job-2")?.status !== "done") {
      await new Promise((r) => setTimeout(r, 20));
    }

    // Because codex concurrency = 1, job-1 should complete before job-2 starts
    assert.deepEqual(executed, [
      "start:job-1",
      "done:job-1",
      "start:job-2",
      "done:job-2",
    ]);

    await queue.stop();
  });

  test("rejects with 429 when MAX_QUEUED_JOBS limit is exceeded", async () => {
    const jobStore = new JobStore({ file: jobFile });
    const engineRegistry = new EngineRegistry();
    engineRegistry.register("edit", () => ({
      run: async () => new Promise((r) => setTimeout(r, 500)),
    }));

    const queue = new Queue({
      concurrencyByGroup: { codex: 1 },
      maxQueuedJobs: 2,
      jobStore,
      engineRegistry,
      engineContext: {},
    });

    const j1 = { id: "j1", type: "edit", status: "queued", payload: {} };
    const j2 = { id: "j2", type: "edit", status: "queued", payload: {} };
    const j3 = { id: "j3", type: "edit", status: "queued", payload: {} };

    jobStore.saveJob(j1);
    jobStore.saveJob(j2);
    jobStore.saveJob(j3);

    await queue.enqueue(j1);
    await queue.enqueue(j2);

    // 3rd job exceeds limit of 2
    await assert.rejects(
      async () => queue.enqueue(j3),
      (err) => err instanceof TooManyRequestsError
    );
  });

  test("recovers jobs from journal after server restart", () => {
    // Write pre-existing jobs into jobs.jsonl
    const lines = [
      JSON.stringify({
        seq: 1,
        jobId: "j-codex-running",
        at: new Date().toISOString(),
        event: "created",
        job: { id: "j-codex-running", type: "edit", status: "running", engine: "codex", attempts: 0 },
      }),
      JSON.stringify({
        seq: 2,
        jobId: "j-upscale-running",
        at: new Date().toISOString(),
        event: "created",
        job: { id: "j-upscale-running", type: "upscale", status: "running", engine: "comfy-upscale", attempts: 0 },
      }),
      JSON.stringify({
        seq: 3,
        jobId: "j-queued",
        at: new Date().toISOString(),
        event: "created",
        job: { id: "j-queued", type: "edit", status: "queued", engine: "codex", attempts: 0 },
      }),
    ];
    fs.writeFileSync(jobFile, lines.join("\n") + "\n", "utf8");

    // Load JobStore -> recovery happens in constructor
    const jobStore = new JobStore({ file: jobFile });

    const codexJob = jobStore.getJob("j-codex-running");
    assert.equal(codexJob.status, "interrupted");
    assert.equal(codexJob.attempts, 1);

    const upscaleJob = jobStore.getJob("j-upscale-running");
    assert.equal(upscaleJob.status, "queued");
    assert.equal(upscaleJob.attempts, 1);

    const queuedJob = jobStore.getJob("j-queued");
    assert.equal(queuedJob.status, "queued");
  });
});