feat(translator): carry a run's handle through the lifecycle, and say it once

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";
@@ -86,6 +87,7 @@ export async function bootTestPayload(opts?: {
   failFor?: string[];
   onTranslate?: (targetLng: string) => Promise<void> | void;
   runner?: TaskRunnerProvider;
+  lifecycle?: TranslationLifecycleCallbacks;
   /** Turn on container-granular rich-text translation, and declare the provider able to keep marks. */
   inlineMarks?: boolean;
   /**
@@ -161,6 +163,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/cancel-after-giving-up.int.test.ts

@@ -0,0 +1,104 @@
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+
+import { bootTestPayload } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+type Seen = { targetLng: string };
+type Assignment = { target_lng: string; handle: string };
+
+const ONE_ATTEMPT_SO_THE_RUN_GIVES_UP_AT_ONCE = 1;
+/** Enough passes that the run is finished one way or another before anything is read. */
+const DRAIN = 8;
+
+const onFailed = vi.fn<(task: Seen, error: unknown) => void>();
+const onCancelled = vi.fn<(task: Seen) => void>();
+
+let ctx: TestPayload;
+
+describe("cancelling a run that has already given up", () => {
+  beforeAll(async () => {
+    ctx = await bootTestPayload({
+      runner: createPayloadJobsRunner({
+        autoRun: false,
+        retries: {
+          attempts: ONE_ATTEMPT_SO_THE_RUN_GIVES_UP_AT_ONCE,
+          backoff: { type: "fixed", delay: 0 },
+        },
+      }),
+      lifecycle: { onFailed, onCancelled },
+      onTranslate: () => {
+        throw new Error("provider down");
+      },
+    });
+  });
+  afterAll(async () => {
+    await ctx?.cleanup();
+  });
+
+  it("announces nothing: every locale was settled when the run gave up", async () => {
+    const made = await ctx.payload.create({
+      collection: "docs",
+      locale: "en",
+      data: { _status: "published", title: "Spent", note: "A note" },
+    });
+    const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+      body: {
+        source_lng: "en",
+        target_lng: ["de", "fr"],
+        collection_slug: "docs",
+        collection_id: [String(made.id)],
+        strategy: "overwrite",
+        publish_on_translation: false,
+      },
+    });
+    const handle = ((res.data as { data: { assignments?: Assignment[] } }).data.assignments ??
+      [])[0]?.handle;
+    for (let pass = 0; pass < DRAIN; pass++) {
+      await ctx.payload.jobs.run({ queue: "translations" });
+    }
+    expect(
+      onFailed.mock.calls.map(([task]) => task.targetLng).sort(),
+      "the run must be spent before the cancellation this check is about"
+    ).toEqual(["de", "fr"]);
+
+    await callEndpoint(ctx.payload, "delete", "/translate/cancel", { body: { ids: [handle] } });
+
+    expect(
+      onCancelled.mock.calls.map(([task]) => task.targetLng),
+      "a locale already told it failed is settled; telling it again is a second ending"
+    ).toEqual([]);
+  });
+
+  it("announces nothing when the whole collection is cancelled either", async () => {
+    const made = await ctx.payload.create({
+      collection: "docs",
+      locale: "en",
+      data: { _status: "published", title: "Spent too", note: "A note" },
+    });
+    await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+      body: {
+        source_lng: "en",
+        target_lng: ["de", "fr"],
+        collection_slug: "docs",
+        collection_id: [String(made.id)],
+        strategy: "overwrite",
+        publish_on_translation: false,
+      },
+    });
+    for (let pass = 0; pass < DRAIN; pass++) {
+      await ctx.payload.jobs.run({ queue: "translations" });
+    }
+    onCancelled.mockClear();
+
+    await callEndpoint(ctx.payload, "delete", "/translate/cancel-by-collection/:collection_slug", {
+      routeParams: { collection_slug: "docs" },
+    });
+
+    expect(
+      onCancelled.mock.calls.map(([task]) => task.targetLng),
+      "a spent run is neither completed nor running, so this sweep reaches it"
+    ).toEqual([]);
+  });
+});

apps/dev/src/integration/translator/cancel-announcement.int.test.ts

@@ -0,0 +1,80 @@
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+
+import { bootTestPayload } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+type Seen = { targetLng: string; handle?: string };
+type Assignment = { target_lng: string; handle: string };
+
+const ATTEMPTS_SO_THE_RUN_STILL_HAS_A_TRY_LEFT = 3;
+
+let ctx: TestPayload;
+
+const runStillOnRecordWhenAnnounced: boolean[] = [];
+
+const onCancelled = vi.fn(async (task: Seen) => {
+  const row = await ctx.payload.find({
+    collection: "payload-jobs" as "docs",
+    where: { id: { equals: task.handle } } as never,
+  });
+  runStillOnRecordWhenAnnounced.push(row.docs.length === 1);
+});
+
+describe("cancelling announces what it stopped, to a host that registered nothing else", () => {
+  beforeAll(async () => {
+    ctx = await bootTestPayload({
+      runner: createPayloadJobsRunner({
+        autoRun: false,
+        retries: {
+          attempts: ATTEMPTS_SO_THE_RUN_STILL_HAS_A_TRY_LEFT,
+          backoff: { type: "fixed", delay: 0 },
+        },
+      }),
+      lifecycle: { onCancelled },
+      onTranslate: (targetLng) => {
+        if (targetLng === "fr") throw new Error("provider down");
+      },
+    });
+  });
+  afterAll(async () => {
+    await ctx?.cleanup();
+  });
+
+  it("names every locale the run still owed, and no locale it had delivered", async () => {
+    const made = await ctx.payload.create({
+      collection: "docs",
+      locale: "en",
+      data: { _status: "published", title: "Stopped midway", note: "A note" },
+    });
+    const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+      body: {
+        source_lng: "en",
+        target_lng: ["de", "fr", "es"],
+        collection_slug: "docs",
+        collection_id: [String(made.id)],
+        strategy: "overwrite",
+        publish_on_translation: false,
+      },
+    });
+    const handle = ((res.data as { data: { assignments?: Assignment[] } }).data.assignments ??
+      [])[0]?.handle;
+    await ctx.payload.jobs.run({ queue: "translations" });
+
+    await callEndpoint(ctx.payload, "delete", "/translate/cancel", { body: { ids: [handle] } });
+
+    expect(onCancelled.mock.calls.map(([task]) => task.targetLng).sort()).toEqual(["es", "fr"]);
+    expect(
+      onCancelled.mock.calls.every(([task]) => task.handle === handle),
+      "each locale is announced against the run that was cancelled"
+    ).toBe(true);
+  });
+
+  it("announces while the run is still on record", async () => {
+    expect(
+      runStillOnRecordWhenAnnounced,
+      "a host told after the deletion could no longer look the run up"
+    ).toEqual([true, true]);
+  });
+});

apps/dev/src/integration/translator/cancel-while-running.int.test.ts

@@ -0,0 +1,73 @@
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+import { afterAll, beforeAll, describe, expect, it } from "vitest";
+
+import { bootTestPayload } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+type Assignment = { target_lng: string; handle: string };
+
+// SQLite only. Cancelling from inside the running job deletes a row that job's own transaction
+// holds, which on a real transactional adapter blocks until the suite's teardown times out. What is
+// under test is which callbacks the plugin fires, and that is the same on every adapter.
+const SQLITE = (process.env.DB_ADAPTER ?? "sqlite") === "sqlite";
+
+let ctx: TestPayload;
+let handle: string | undefined;
+const announced: string[] = [];
+
+const say = (what: string) => (task: { targetLng: string }) => {
+  announced.push(`${what}:${task.targetLng}`);
+};
+
+describe.skipIf(!SQLITE)("cancelling a run that is already translating", () => {
+  beforeAll(async () => {
+    ctx = await bootTestPayload({
+      runner: createPayloadJobsRunner({ autoRun: false, retries: { attempts: 0 } }),
+      lifecycle: {
+        onCompleted: say("completed"),
+        onFailed: say("failed"),
+        onCancelled: say("cancelled"),
+      },
+      onTranslate: async (targetLng) => {
+        if (targetLng === "de" && handle) {
+          await callEndpoint(ctx.payload, "delete", "/translate/cancel", {
+            body: { ids: [handle] },
+          });
+        }
+      },
+    });
+  });
+  afterAll(async () => {
+    await ctx?.cleanup();
+  });
+
+  it("does not also report the locales as failed", async () => {
+    const made = await ctx.payload.create({
+      collection: "docs",
+      locale: "en",
+      data: { _status: "published", title: "Stopped mid-flight", note: "A note" },
+    });
+    const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+      body: {
+        source_lng: "en",
+        target_lng: ["de", "fr"],
+        collection_slug: "docs",
+        collection_id: [String(made.id)],
+        strategy: "overwrite",
+        publish_on_translation: false,
+      },
+    });
+    handle = ((res.data as { data: { assignments?: Assignment[] } }).data.assignments ?? [])[0]
+      ?.handle;
+
+    // Payload's own update throws when the row it is writing was deleted under it.
+    await ctx.payload.jobs.run({ queue: "translations" }).catch(() => undefined);
+
+    expect(announced, "the run was cancelled, so the host is told that").toContain("cancelled:de");
+    expect(
+      announced.filter((a) => a.startsWith("failed:")),
+      `nothing translated failed — a machinery error is not a locale giving up (saw: ${announced.join(", ")})`
+    ).toEqual([]);
+  });
+});

apps/dev/src/integration/translator/completed-handle.int.test.ts

@@ -0,0 +1,55 @@
+import { createPayloadJobsRunner } from "@focus-reactive/payload-plugin-translator";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+
+import { bootTestPayload } from "./bootTestPayload";
+import type { TestPayload } from "./bootTestPayload";
+import { callEndpoint } from "./callEndpoint";
+
+type Seen = { targetLng: string; handle?: string };
+type Assignment = { target_lng: string; handle: string };
+
+const onCompleted = vi.fn<(task: Seen) => void>();
+
+let ctx: TestPayload;
+
+describe("the handle the enqueue answer named is the handle onCompleted carries", () => {
+  beforeAll(async () => {
+    ctx = await bootTestPayload({
+      runner: createPayloadJobsRunner({ autoRun: false }),
+      lifecycle: { onCompleted },
+    });
+  });
+  afterAll(async () => {
+    await ctx?.cleanup();
+  });
+
+  it("matches for every locale of the request", async () => {
+    const made = await ctx.payload.create({
+      collection: "docs",
+      locale: "en",
+      data: { _status: "published", title: "Two locales", note: "A note" },
+    });
+    const res = await callEndpoint(ctx.payload, "post", "/translate/enqueue", {
+      body: {
+        source_lng: "en",
+        target_lng: ["de", "fr"],
+        collection_slug: "docs",
+        collection_id: [String(made.id)],
+        strategy: "overwrite",
+        publish_on_translation: false,
+      },
+    });
+    const promised = (res.data as { data: { assignments?: Assignment[] } }).data.assignments ?? [];
+
+    await ctx.payload.jobs.run({ queue: "translations" });
+
+    const delivered = onCompleted.mock.calls.map(([task]) => task);
+    expect(delivered.map((t) => t.targetLng).sort()).toEqual(["de", "fr"]);
+    for (const task of delivered) {
+      expect(
+        task.handle,
+        `the caller was told ${task.targetLng} would run as a named job; it has to be that job`
+      ).toBe(promised.find((j) => j.target_lng === task.targetLng)?.handle);
+    }
+  });
+});