feat(translator): carry the job id through the translation lifecycle

apps/dev/src/integration/translator/bootTestPayload.ts

@@ -14,6 +14,7 @@ import {
 } from "@focus-reactive/payload-plugin-translator";
 import type {
   TaskRunnerProvider,
+  TranslationLifecycleCallbacks,
   TranslationProvider,
 } from "@focus-reactive/payload-plugin-translator";
 import { buildConfig } from "payload";
@@ -97,6 +98,8 @@ export async function bootTestPayload(opts?: {
   fake?: FakeTranslationOptions;
   /** Also register the synchronous per-field surface, `POST {basePath}/field`. */
   fieldSurface?: boolean;
+  /** Omitted by default, so a boot without it leaves the runner undecorated. */
+  lifecycle?: TranslationLifecycleCallbacks;
 }): Promise<TestPayload> {
   const dir = mkdtempSync(join(tmpdir(), "translator-int-"));
   const { db, drop } = createTestDatabase(join(dir, "test.db"));
@@ -161,6 +164,7 @@ export async function bootTestPayload(opts?: {
         runner: opts?.runner ?? createSyncRunner(),
         levels: opts?.fieldSurface ? [documentLevel(), fieldLevel()] : [documentLevel()],
         provenance: true,
+        ...(opts?.lifecycle ? { lifecycle: opts.lifecycle } : {}),
         ...(opts?.inlineMarks ? { experimental: { inlineMarks: true } } : {}),
       }),
     ],

apps/dev/src/integration/translator/callEndpoint.ts

@@ -14,7 +14,7 @@ import type { Payload, PayloadRequest } from "payload";
  */
 export async function callEndpoint(
   payload: Payload,
-  method: "get" | "post",
+  method: "get" | "post" | "delete",
   path: string,
   opts?: {
     routeParams?: Record<string, unknown>;

apps/dev/src/integration/translator/cancelled-only-host.int.test.ts

@@ -0,0 +1,59 @@
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+
+import { bootTestPayload } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+type Seen = { collection: string; id: string; targetLng: string; jobId?: string };
+
+const onCancelled = vi.fn<(task: Seen) => void>();
+
+let ctx: TestPayload;
+
+describe("a host that registers only onCancelled", () => {
+  beforeAll(async () => {
+    ctx = await bootTestPayload({
+      runner: createPayloadJobsRunner({ autoRun: false }),
+      lifecycle: { onCancelled },
+    });
+  });
+  afterAll(async () => {
+    await ctx?.cleanup();
+  });
+
+  it("is told what stopped, once per locale, with the document named", async () => {
+    const made = await ctx.payload.create({
+      collection: "docs",
+      locale: "en",
+      data: { _status: "published", title: "To be cancelled", note: "A note" },
+    });
+    const id = String(made.id);
+
+    const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+      body: {
+        source_lng: "en",
+        target_lng: ["de", "fr"],
+        collection_slug: "docs",
+        collection_id: [id],
+        strategy: "overwrite",
+        publish_on_translation: true,
+      },
+    });
+    const jobs = (res.data as { data: { jobs: { job_id: string }[] } }).data.jobs;
+    const jobId = jobs[0]?.job_id;
+
+    onCancelled.mockClear();
+    await callEndpoint(ctx.payload, "delete", "/translate/cancel", { body: { ids: [jobId] } });
+
+    expect(
+      onCancelled.mock.calls.map(([task]) => task.targetLng).sort(),
+      "both locales of the cancelled job were translations the host was tracking"
+    ).toEqual(["de", "fr"]);
+    expect(onCancelled.mock.calls[0]?.[0]).toMatchObject({
+      collection: "docs",
+      id,
+      jobId,
+    });
+  });
+});

apps/dev/src/integration/translator/enqueue-job-ids.int.test.ts

@@ -0,0 +1,80 @@
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+import { bootTestPayload } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+type QueuedJob = {
+  collection_slug: string;
+  collection_id: string;
+  target_lng: string;
+  job_id: string;
+};
+
+let ctx: TestPayload;
+
+const enqueue = (collectionId: string[], targetLng: string | string[]) =>
+  callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+    body: {
+      source_lng: "en",
+      target_lng: targetLng,
+      collection_slug: "docs",
+      collection_id: collectionId,
+      strategy: "overwrite",
+      publish_on_translation: true,
+    },
+  });
+
+const bodyOf = (res: { data: unknown }) =>
+  (res.data as { data: { queued: number; jobs?: QueuedJob[] } }).data;
+
+const jobsOf = (res: { data: unknown }) => bodyOf(res).jobs ?? [];
+
+const create = async (title: string): Promise<string> => {
+  const made = await ctx.payload.create({
+    collection: "docs",
+    locale: "en",
+    data: { _status: "published", title, note: "A note" },
+  });
+  return String(made.id);
+};
+
+describe("the enqueue response names the job that will run each locale", () => {
+  beforeAll(async () => {
+    ctx = await bootTestPayload();
+  });
+  afterAll(async () => {
+    await ctx?.cleanup();
+  });
+
+  it("answers one entry per target locale", async () => {
+    const id = await create("Three locales");
+
+    const jobs = jobsOf(await enqueue([id], ["de", "fr", "es"]));
+
+    expect(jobs.map((job) => job.target_lng).sort()).toEqual(["de", "es", "fr"]);
+    expect(jobs.every((job) => job.job_id.length > 0)).toBe(true);
+  });
+
+  it("names the document each entry belongs to, so a bulk request stays unambiguous", async () => {
+    const first = await create("First");
+    const second = await create("Second");
+
+    const jobs = jobsOf(await enqueue([first, second], "de"));
+
+    expect(
+      jobs.map((job) => job.collection_id).sort(),
+      "a locale-keyed answer could not tell these two apart"
+    ).toEqual([first, second].sort());
+    expect(jobs.every((job) => job.collection_slug === "docs")).toBe(true);
+  });
+
+  it("still reports how many it queued", async () => {
+    const id = await create("Counted");
+
+    const res = await enqueue([id], ["de", "fr"]);
+
+    expect(bodyOf(res).queued, "the count that was already there must survive").toBe(2);
+    expect(jobsOf(res)).toHaveLength(2);
+  });
+});

apps/dev/src/integration/translator/extend-live-job.int.test.ts

@@ -0,0 +1,71 @@
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+
+import { bootTestPayload } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+type Seen = { targetLng: string; jobId?: string; attempt?: number };
+
+const onCompleted = vi.fn<(task: Seen) => void>();
+const onFailed = vi.fn<(task: Seen) => void>();
+
+let ctx: TestPayload;
+
+const create = async (title: string): Promise<string> => {
+  const made = await ctx.payload.create({
+    collection: "docs",
+    locale: "en",
+    data: { _status: "published", title, note: "A note" },
+  });
+  return String(made.id);
+};
+
+const enqueue = (id: string, targetLng: string) =>
+  callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+    body: {
+      source_lng: "en",
+      target_lng: targetLng,
+      collection_slug: "docs",
+      collection_id: [id],
+      strategy: "overwrite",
+      publish_on_translation: true,
+    },
+  });
+
+const jobsOf = (res: { data: unknown }) =>
+  (res.data as { data: { jobs: { target_lng: string; job_id: string }[] } }).data.jobs;
+
+const callbackJobIds = (calls: [Seen][]) =>
+  Object.fromEntries(calls.map(([task]) => [task.targetLng, task.jobId]));
+
+describe("a locale added to a job already in flight", () => {
+  beforeAll(async () => {
+    ctx = await bootTestPayload({
+      runner: createPayloadJobsRunner({ autoRun: false }),
+      lifecycle: { onCompleted, onFailed },
+    });
+  });
+  afterAll(async () => {
+    await ctx?.cleanup();
+  });
+
+  it("reports the live job in the response, and the same id in the callback, with no cross-talk", async () => {
+    onCompleted.mockClear();
+    const id = await create("Extended");
+
+    const first = jobsOf(await enqueue(id, "de"));
+    const second = jobsOf(await enqueue(id, "fr"));
+
+    expect(
+      second[0]?.job_id,
+      "fr joined the job de already had — a new id here would be a lie"
+    ).toBe(first[0]?.job_id);
+
+    await ctx.payload.jobs.run({ queue: "translations" });
+
+    const byLocale = callbackJobIds(onCompleted.mock.calls);
+    expect(byLocale.de, "the first run's locale keeps its own reporting").toBe(first[0]?.job_id);
+    expect(byLocale.fr).toBe(second[0]?.job_id);
+  });
+});

apps/dev/src/integration/translator/lifecycle-job-id.int.test.ts

@@ -0,0 +1,88 @@
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+
+import { bootTestPayload } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+type Seen = { targetLng: string; jobId?: string; attempt?: number };
+
+const onQueued = vi.fn<(task: Seen) => void>();
+const onCompleted = vi.fn<(task: Seen) => void>();
+const onFailed = vi.fn<(task: Seen) => void>();
+
+let ctx: TestPayload;
+
+const create = async (title: string): Promise<string> => {
+  const made = await ctx.payload.create({
+    collection: "docs",
+    locale: "en",
+    data: { _status: "published", title, note: "A note" },
+  });
+  return String(made.id);
+};
+
+const enqueue = (id: string, targetLng: string | string[]) =>
+  callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+    body: {
+      source_lng: "en",
+      target_lng: targetLng,
+      collection_slug: "docs",
+      collection_id: [id],
+      strategy: "overwrite",
+      publish_on_translation: true,
+    },
+  });
+
+const reportedJobs = (res: { data: unknown }) =>
+  (res.data as { data: { jobs: { target_lng: string; job_id: string }[] } }).data.jobs;
+
+describe("a lifecycle callback names the job that ran it", () => {
+  beforeAll(async () => {
+    ctx = await bootTestPayload({
+      runner: createPayloadJobsRunner({ autoRun: false }),
+      lifecycle: { onQueued, onCompleted, onFailed },
+    });
+  });
+  afterAll(async () => {
+    await ctx?.cleanup();
+  });
+
+  it("gives onCompleted the same job id the enqueue response reported for that locale", async () => {
+    onCompleted.mockClear();
+    const id = await create("Matched");
+
+    const reported = reportedJobs(await enqueue(id, ["de", "fr"]));
+    await ctx.payload.jobs.run({ queue: "translations" });
+
+    const byLocale = Object.fromEntries(
+      onCompleted.mock.calls.map(([task]) => [task.targetLng, task.jobId])
+    );
+    for (const job of reported) {
+      expect(
+        byLocale[job.target_lng],
+        `the callback for ${job.target_lng} must name the job the response promised`
+      ).toBe(job.job_id);
+    }
+  });
+
+  it("gives onCompleted the attempt it ran on", async () => {
+    onCompleted.mockClear();
+    const id = await create("Attempted");
+
+    await enqueue(id, "de");
+    await ctx.payload.jobs.run({ queue: "translations" });
+
+    expect(onCompleted.mock.calls[0]?.[0].attempt, "a first run is attempt 1, not 0").toBe(1);
+  });
+
+  it("does not pretend onQueued knows a job id — the job does not exist yet", async () => {
+    onQueued.mockClear();
+    const id = await create("Too early");
+
+    await enqueue(id, "de");
+
+    expect(onQueued).toHaveBeenCalled();
+    expect(onQueued.mock.calls[0]?.[0].jobId).toBeUndefined();
+  });
+});