WRNexusJS/packages/queue/src/defined.ts

import type { AddOptions, Job, JobContext, JobHandler } from "./index.ts";
import {
  createDurableQueue,
  type DurableQueue,
  type DurableQueueOptions,
  type QueueHealth,
} from "./durable.ts";
import { configuredQueueStore } from "./configured.ts";

export type JobStatus = "queued" | "completed" | "failed" | "cancelled" | "missing";

export interface DefinedJobOptions<T> extends Omit<AddOptions, "idempotencyKey"> {
  run: (data: T, context: JobContext & { job: Job<T> }) => void | Promise<void>;
  success?: (data: T, job: Job<T>) => void | Promise<void>;
  failed?: (data: T, error: unknown, job: Job<T>) => void | Promise<void>;
  idempotency?: (data: T) => string | undefined;
  validate?: (data: unknown) => data is T;
}

export type QueueJobDefinitions = Record<string, DefinedJobOptions<any>>;

export interface DefineQueueOptions<TJobs extends QueueJobDefinitions> {
  name: string;
  jobs: TJobs;
  queue?: DurableQueue;
  options?: DurableQueueOptions;
}

type DataOf<T> = T extends DefinedJobOptions<infer I> ? I : never;

export interface DefinedJob<T> {
  readonly name: string;
  add(data: T, options?: AddOptions): Promise<Job<T>>;
  addMany(entries: readonly { data: T; options?: AddOptions }[]): Promise<Job<T>[]>;
  get(id: string): Promise<Job<T> | null>;
  status(id: string): Promise<JobStatus>;
  list(): Promise<Job<T>[]>;
  cancel(id: string): Promise<boolean>;
  retry(id: string): Promise<boolean>;
}

export type DefinedJobs<TJobs extends QueueJobDefinitions> = {
  [K in keyof TJobs]: DefinedJob<DataOf<TJobs[K]>>;
};

export interface DefinedQueue<TJobs extends QueueJobDefinitions> {
  readonly name: string;
  readonly jobs: DefinedJobs<TJobs>;
  readonly raw: DurableQueue;
  start(): void;
  stop(): void;
  shutdown(options?: { force?: boolean }): Promise<void>;
  health(): Promise<QueueHealth>;
  drain(): Promise<number>;
}

function fullName(queue: string, job: string): string {
  return `${queue}:${job}`;
}

export function defineQueue<TJobs extends QueueJobDefinitions>(
  definition: DefineQueueOptions<TJobs>,
): DefinedQueue<TJobs> & DefinedJobs<TJobs> {
  const name = definition.name.trim();
  if (!name) throw new TypeError("queue name cannot be empty");
  const queue =
    definition.queue ??
    createDurableQueue({
      ...definition.options,
      store: definition.options?.store ?? configuredQueueStore(),
      async onDeadLetter(job, error) {
        await definition.options?.onDeadLetter?.(job, error);
        const prefix = `${name}:`;
        if (!job.name.startsWith(prefix)) return;
        const failed = definition.jobs[job.name.slice(prefix.length)]?.failed;
        await failed?.(job.data, error, job);
      },
    });
  const jobs: Record<string, DefinedJob<unknown>> = {};

  for (const [shortName, jobDefinition] of Object.entries(definition.jobs)) {
    const jobName = fullName(name, shortName);
    const handler: JobHandler<unknown> = async (job, context) => {
      if (jobDefinition.validate && !jobDefinition.validate(job.data)) {
        throw new TypeError(`WRN-QUEUE-PAYLOAD: invalid payload for '${jobName}'`);
      }
      await jobDefinition.run(job.data, { ...context, job });
      await jobDefinition.success?.(job.data, job);
    };
    queue.process(jobName, handler);

    jobs[shortName] = {
      name: jobName,
      add(data, options = {}) {
        if (jobDefinition.validate && !jobDefinition.validate(data)) {
          return Promise.reject(
            new TypeError(`WRN-QUEUE-PAYLOAD: invalid payload for '${jobName}'`),
          );
        }
        return queue.add(jobName, data, {
          maxAttempts: jobDefinition.maxAttempts,
          delayMs: jobDefinition.delayMs,
          repeat: jobDefinition.repeat,
          priority: jobDefinition.priority,
          idempotencyKey: jobDefinition.idempotency?.(data),
          ...options,
        });
      },
      addMany(entries) {
        return Promise.all(entries.map((entry) => this.add(entry.data, entry.options)));
      },
      async get(id) {
        const job = await queue.get(id);
        return job?.name === jobName ? (job as Job<unknown>) : null;
      },
      async status(id) {
        return queue.status(id);
      },
      async list() {
        return (await queue.list(jobName)) as Job<unknown>[];
      },
      cancel(id) {
        return queue.cancel(id);
      },
      retry(id) {
        return queue.retry(id);
      },
    };
  }

  const result = {
    name,
    jobs: jobs as DefinedJobs<TJobs>,
    raw: queue,
    start: () => queue.start(),
    stop: () => queue.stop(),
    shutdown: (options?: { force?: boolean }) => queue.shutdown(options),
    health: () => queue.health(),
    drain: () => queue.drain(),
    ...(jobs as DefinedJobs<TJobs>),
  };
  return result;
}