feat(node): enqueue interceptors

sdks/node/src/errors.ts

@@ -150,3 +150,11 @@ export class PredicateRejectedError extends TaskitoError {
     this.name = "PredicateRejectedError";
   }
 }
+
+/** Thrown when an enqueue interceptor rejects, misbehaves, or redirects illegally. */
+export class InterceptionError extends TaskitoError {
+  constructor(message: string) {
+    super(message);
+    this.name = "InterceptionError";
+  }
+}

sdks/node/src/index.ts

@@ -2,6 +2,7 @@ export { currentJob, type JobContext } from "./context";
 export { type DashboardAuth, type DashboardOptions, serveDashboard } from "./dashboard";
 export {
   CryptoError,
+  InterceptionError,
   JobCancelledError,
   JobFailedError,
   LockLostError,
@@ -19,6 +20,7 @@ export {
   WorkflowError,
 } from "./errors";
 export type { EventHandler, EventName, OutcomeEvent } from "./events";
+export { Interception, type Interceptor } from "./interception";
 export { Lock, type LockInfo, type LockOptions } from "./locks";
 export type { EnqueueContext, Middleware, TaskContext } from "./middleware";
 export {

sdks/node/src/interception.ts

@@ -0,0 +1,43 @@
+/**
+ * Producer-side enqueue interception. Interceptors registered via
+ * `Queue.intercept` run at the very start of every enqueue — before per-task
+ * defaults, middleware, gates, and serialization — and decide what happens
+ * to the call:
+ *
+ * - `pass` — enqueue unchanged.
+ * - `convert` — replace the args (task name unchanged), e.g. redact a field.
+ * - `redirect` — enqueue a different task with new args instead.
+ * - `reject` — block the enqueue; `Queue.enqueue` throws `InterceptionError`.
+ *
+ * Interceptors chain in registration order, each seeing the previous one's
+ * output. Distinct from `Middleware.onEnqueue` (mutate-in-place hooks) and
+ * from the Python SDK's argument-type interception.
+ */
+export type Interception =
+  | { readonly type: "pass" }
+  | { readonly type: "convert"; readonly args: unknown[] }
+  | { readonly type: "redirect"; readonly taskName: string; readonly args: unknown[] }
+  | { readonly type: "reject"; readonly reason: string };
+
+/** Decides the outcome of an enqueue. Runs synchronously before serialization. */
+export type Interceptor = (taskName: string, args: unknown[]) => Interception;
+
+/** Factories for the four {@link Interception} outcomes. */
+export const Interception = {
+  /** Enqueue the original args unchanged. */
+  pass(): Interception {
+    return { type: "pass" };
+  },
+  /** Replace the args; the task name stays the same. */
+  convert(args: unknown[]): Interception {
+    return { type: "convert", args };
+  },
+  /** Enqueue a different task with new args instead of the original. */
+  redirect(taskName: string, args: unknown[]): Interception {
+    return { type: "redirect", taskName, args };
+  },
+  /** Block the enqueue; the reason is surfaced on the thrown error. */
+  reject(reason: string): Interception {
+    return { type: "reject", reason };
+  },
+};

sdks/node/src/queue.ts

@@ -1,6 +1,7 @@
 import { mkdirSync } from "node:fs";
 import { dirname } from "node:path";
 import {
+  InterceptionError,
   JobCancelledError,
   JobFailedError,
   LockLostError,
@@ -12,6 +13,7 @@ import {
   TaskitoError,
 } from "./errors";
 import { Emitter, type EventHandler, type EventName } from "./events";
+import type { Interceptor } from "./interception";
 import { Lock, type LockOptions } from "./locks";
 import type { EnqueueContext, Middleware } from "./middleware";
 import {
@@ -100,6 +102,7 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
   private readonly tasks = new Map<string, RegisteredTask>();
   private readonly queueLimits = new Map<string, QueueLimits>();
   private readonly middleware: Middleware[] = [];
+  private readonly interceptors: Interceptor[] = [];
   private readonly gates = new Map<string, Predicate[]>();
   private readonly emitter = new Emitter();
   private readonly resources = new ResourceRuntime();
@@ -281,6 +284,19 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
    * `onEnqueue`). If it returns `false`, the enqueue throws
    * {@link PredicateRejectedError}. Multiple gates on a task all must pass.
    */
+  /**
+   * Register an enqueue interceptor. Interceptors run at the start of every
+   * enqueue — before defaults, middleware, and gates — chained in
+   * registration order, each seeing the previous one's task name and args.
+   * Returning `Interception.reject(...)` (or a null-ish value) makes the
+   * enqueue throw {@link InterceptionError}; `redirect` is not supported for
+   * batch enqueue or for tasks with per-task codecs.
+   */
+  intercept(interceptor: Interceptor): this {
+    this.interceptors.push(interceptor);
+    return this;
+  }
+
   gate<Name extends keyof TTasks & string>(
     name: Name,
     predicate: (ctx: { taskName: Name; args: Parameters<TTasks[Name]> }) => boolean,
@@ -309,8 +325,8 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
     args?: Parameters<TTasks[Name]>,
     options?: EnqueueOptions,
   ): string {
-    const { payload, options: nativeOpts } = this.prepareEnqueue(name, args, options);
-    return this.native.enqueue(name, payload, nativeOpts);
+    const { taskName, payload, options: nativeOpts } = this.prepareEnqueue(name, args, options);
+    return this.native.enqueue(taskName, payload, nativeOpts);
   }
 
   /**
@@ -322,43 +338,100 @@ export class Queue<TTasks extends TaskMap = TaskMap> {
     name: Name,
     jobs: ReadonlyArray<{ args?: Parameters<TTasks[Name]>; options?: EnqueueOptions }>,
   ): string[] {
-    const prepared = jobs.map((job) => this.prepareEnqueue(name, job.args, job.options));
+    const prepared = jobs.map((job) => {
+      const { payload, options } = this.prepareEnqueue(name, job.args, job.options, {
+        batch: true,
+      });
+      return { payload, options };
+    });
     return this.native.enqueueMany(name, prepared);
   }
 
   /**
-   * Merge per-task defaults, run the `onEnqueue` interception hooks, then
-   * serialize the args and encode the options — the shared path for
-   * {@link Queue.enqueue} and {@link Queue.enqueueMany}.
+   * Run enqueue interceptors, merge per-task defaults, run the middleware
+   * `onEnqueue` hooks, then serialize the args and encode the options — the
+   * shared path for {@link Queue.enqueue} and {@link Queue.enqueueMany}.
+   * Everything downstream of the interceptors keys off the (possibly
+   * redirected) final task name.
    */
   private prepareEnqueue<Name extends keyof TTasks & string>(
     name: Name,
     args: Parameters<TTasks[Name]> | undefined,
     options: EnqueueOptions | undefined,
-  ): { payload: Buffer; options: NativeEnqueueOptions } {
-    const defaults = this.tasks.get(name)?.options;
+    mode?: { batch: boolean },
+  ): { taskName: string; payload: Buffer; options: NativeEnqueueOptions } {
+    const { taskName, args: finalArgs } = this.runInterceptors(name, [...(args ?? [])], mode);
+    const defaults = this.tasks.get(taskName)?.options;
     const merged: EnqueueOptions = {
       ...options,
       maxRetries: options?.maxRetries ?? defaults?.maxRetries,
       timeoutMs: options?.timeoutMs ?? defaults?.timeoutMs,
     };
-    // Interception seam: let middleware validate/redact/rewrite before serializing.
-    const ctx: EnqueueContext = { taskName: name, args: [...(args ?? [])], options: merged };
+    // Middleware seam: let onEnqueue hooks validate/redact/rewrite before serializing.
+    const ctx: EnqueueContext = { taskName, args: finalArgs, options: merged };
     for (const mw of this.middleware) {
       mw.onEnqueue?.(ctx);
     }
     // Gate: predicates see the (possibly rewritten) args and may reject the enqueue.
-    for (const gate of this.gates.get(name) ?? []) {
-      if (!gate({ taskName: name, args: ctx.args })) {
-        throw new PredicateRejectedError(name);
+    for (const gate of this.gates.get(taskName) ?? []) {
+      if (!gate({ taskName, args: ctx.args })) {
+        throw new PredicateRejectedError(taskName);
       }
     }
     return {
-      payload: this.encodeTaskPayload(name, ctx.args),
+      taskName,
+      payload: this.encodeTaskPayload(taskName, ctx.args),
       options: toNativeEnqueueOptions(ctx.options),
     };
   }
 
+  /**
+   * Chain the registered interceptors over `(taskName, args)`: a null-ish
+   * outcome or `reject` throws; `redirect` swaps both
+   * the task name and args, and is rejected for batch enqueue (the batch is
+   * stored under one task name) and for tasks with per-task codecs (the
+   * redirect target's codec chain cannot be resolved from a bare name —
+   * a cross-SDK behavioral contract).
+   */
+  private runInterceptors(
+    name: string,
+    args: unknown[],
+    mode?: { batch: boolean },
+  ): { taskName: string; args: unknown[] } {
+    let taskName = name;
+    let currentArgs = args;
+    for (const interceptor of this.interceptors) {
+      const outcome = interceptor(taskName, currentArgs);
+      if (!outcome) {
+        throw new InterceptionError(`interceptor returned null for task "${taskName}"`);
+      }
+      switch (outcome.type) {
+        case "pass":
+          break;
+        case "convert":
+          currentArgs = outcome.args;
+          break;
+        case "redirect":
+          if (mode?.batch) {
+            throw new InterceptionError(
+              `interceptor Redirect is not supported for batch enqueue of task "${taskName}"`,
+            );
+          }
+          taskName = outcome.taskName;
+          currentArgs = outcome.args;
+          break;
+        case "reject":
+          throw new InterceptionError(`enqueue of "${taskName}" rejected: ${outcome.reason}`);
+      }
+    }
+    if (taskName !== name && (this.tasks.get(name)?.options?.codecs?.length ?? 0) > 0) {
+      throw new InterceptionError(
+        `interceptor Redirect is not supported for a task with payload codecs ("${name}")`,
+      );
+    }
+    return { taskName, args: currentArgs };
+  }
+
   /**
    * Serialize a task payload and apply the task's named codecs in order.
    * Payload only — results stay on the queue serializer.

sdks/node/test/core/interceptors.test.ts

@@ -0,0 +1,147 @@
+import { mkdtempSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, describe, expect, it } from "vitest";
+import { GzipCodec, Interception, InterceptionError, Queue, type Worker } from "../../src/index";
+
+let worker: Worker | undefined;
+
+afterEach(() => {
+  worker?.stop();
+  worker = undefined;
+});
+
+function tempDb(): string {
+  return join(mkdtempSync(join(tmpdir(), "taskito-intercept-")), "queue.db");
+}
+
+async function waitFor<T>(read: () => T | undefined, timeoutMs = 10_000): Promise<T> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    const value = read();
+    if (value !== undefined) {
+      return value;
+    }
+    await new Promise((resolve) => setTimeout(resolve, 25));
+  }
+  throw new Error("timed out waiting for job result");
+}
+
+describe("enqueue interceptors", () => {
+  it("convert rewrites the args the worker sees", async () => {
+    const queue = new Queue({ dbPath: tempDb() })
+      .task("ic.echo", (x: number) => x)
+      .intercept((_taskName, args) => Interception.convert([(args[0] as number) * 10]));
+
+    const id = queue.enqueue("ic.echo", [5]);
+    worker = queue.runWorker();
+    await expect(waitFor(() => queue.getResult(id))).resolves.toBe(50);
+  });
+
+  it("redirect enqueues the other task; the original handler never runs", async () => {
+    const ranA: unknown[] = [];
+    const queue = new Queue({ dbPath: tempDb() })
+      .task("ic.a", (x: number) => {
+        ranA.push(x);
+        return x;
+      })
+      .task("ic.b", (x: number) => x + 100)
+      .intercept((taskName, args) =>
+        taskName === "ic.a" ? Interception.redirect("ic.b", args) : Interception.pass(),
+      );
+
+    const id = queue.enqueue("ic.a", [1]);
+    expect(queue.getJob(id)?.taskName).toBe("ic.b");
+    worker = queue.runWorker();
+    await expect(waitFor(() => queue.getResult(id))).resolves.toBe(101);
+    expect(ranA).toEqual([]);
+  });
+
+  it("reject throws InterceptionError and enqueues nothing", async () => {
+    const queue = new Queue({ dbPath: tempDb() })
+      .task("ic.noop", () => null)
+      .intercept(() => Interception.reject("blocked by policy"));
+
+    expect(() => queue.enqueue("ic.noop", [])).toThrow(InterceptionError);
+    expect(() => queue.enqueue("ic.noop", [])).toThrow(/blocked by policy/);
+    expect((await queue.stats()).pending).toBe(0);
+  });
+
+  it("a null-ish interceptor result throws InterceptionError", () => {
+    const queue = new Queue({ dbPath: tempDb() })
+      .task("ic.noop", () => null)
+      // biome-ignore lint/suspicious/noExplicitAny: deliberately misbehaving interceptor
+      .intercept((() => undefined) as any);
+
+    expect(() => queue.enqueue("ic.noop", [])).toThrow(/returned null/);
+  });
+
+  it("interceptors chain in registration order", async () => {
+    const queue = new Queue({ dbPath: tempDb() })
+      .task("ic.echo", (x: number) => x)
+      .intercept((_name, args) => Interception.convert([(args[0] as number) + 1]))
+      .intercept((_name, args) => Interception.convert([(args[0] as number) * 2]));
+
+    const id = queue.enqueue("ic.echo", [3]); // (3 + 1) * 2
+    worker = queue.runWorker();
+    await expect(waitFor(() => queue.getResult(id))).resolves.toBe(8);
+  });
+
+  it("batch enqueue converts per item", async () => {
+    const queue = new Queue({ dbPath: tempDb() })
+      .task("ic.echo", (x: number) => x)
+      .intercept((_name, args) => Interception.convert([(args[0] as number) * 10]));
+
+    const ids = queue.enqueueMany("ic.echo", [{ args: [1] }, { args: [2] }]);
+    worker = queue.runWorker();
+    await expect(waitFor(() => queue.getResult(ids[0] as string))).resolves.toBe(10);
+    await expect(waitFor(() => queue.getResult(ids[1] as string))).resolves.toBe(20);
+  });
+
+  it("batch reject fails the whole batch — zero jobs stored", async () => {
+    const queue = new Queue({ dbPath: tempDb() })
+      .task("ic.echo", (x: number) => x)
+      .intercept((_name, args) =>
+        (args[0] as number) < 0 ? Interception.reject("negative") : Interception.pass(),
+      );
+
+    expect(() =>
+      queue.enqueueMany("ic.echo", [{ args: [1] }, { args: [-2] }, { args: [3] }]),
+    ).toThrow(InterceptionError);
+    expect((await queue.stats()).pending).toBe(0);
+  });
+
+  it("batch redirect is unsupported", () => {
+    const queue = new Queue({ dbPath: tempDb() })
+      .task("ic.a", (x: number) => x)
+      .task("ic.b", (x: number) => x)
+      .intercept((_name, args) => Interception.redirect("ic.b", args));
+
+    expect(() => queue.enqueueMany("ic.a", [{ args: [1] }])).toThrow(
+      /Redirect is not supported for batch/,
+    );
+  });
+
+  it("redirect of a task with per-task codecs is rejected", () => {
+    const queue = new Queue({ dbPath: tempDb(), codecs: { gzip: new GzipCodec() } })
+      .task("ic.codec", (x: number) => x, { codecs: ["gzip"] })
+      .task("ic.b", (x: number) => x)
+      .intercept((_name, args) => Interception.redirect("ic.b", args));
+
+    expect(() => queue.enqueue("ic.codec", [1])).toThrow(
+      /Redirect is not supported for a task with payload codecs/,
+    );
+  });
+
+  it("interceptors run before gates: a gate sees the converted args", async () => {
+    const queue = new Queue({ dbPath: tempDb() })
+      .task("ic.echo", (x: number) => x)
+      .gate("ic.echo", ({ args }) => (args[0] as number) > 0)
+      .intercept((_name, args) => Interception.convert([Math.abs(args[0] as number)]));
+
+    // Raw -5 would be gated; the interceptor converts to 5 first.
+    const id = queue.enqueue("ic.echo", [-5]);
+    worker = queue.runWorker();
+    await expect(waitFor(() => queue.getResult(id))).resolves.toBe(5);
+  });
+});