feat: Adds support for a context exposed in job processing

This adds suport for a context object, created at the start of an
individual job's processing lifecycle and then exposed in emitter events
such as ack, fail, ping, and dead.

Resolves #13

Author: thecodedrift

README.md

@@ -93,32 +93,53 @@ const queue = new Queue<SimpleJob>(new MemoryDriver("default"), "docmq");
 
 #### `new Queue()` options
 
-`new Queue<T>(driver: Driver, name: string, options?: QueueOptions)`
+```ts
+new Queue<
+  TData,
+  TAck = unknown,
+  TFail extends Error = Error,
+  TContext = DefaultContext,
+>(driver: Driver, name: string, options?: QueueOptions)
+```
 
 - `driver` a Driver implementation to use such as the `MemoryDriver`
 - `name` a string for the queue's name
 - `options?` additional options
   - `retention.jobs?` number of seconds to retain jobs with no further work. Default `3600` (1 hour)
   - `statInterval?` number of seconds between emitting a `stat` event with queue statistics, defaults to `5`
 
+### A Note on TypeScript
+
+This library uses TypeScript to provide a better developer experience regarding the objects passed into your queue and the responses your job processor provides back to DocMQ. There are four main types used throughout this documentation, and all are set during the creation of the `Queue` class.
+
+`TData` refers specifically to the typing of your **job payload**. It's the payload you're expecting to pass when calling `enqueue()`, and it's the payload you're expecting to receive inside of your `process()` callback.
+
+`TAck = unknown` refers to the typing of your **ack response** when calling `api.ack()` inside of your job processor and is by default an unknown type. Setting `TAck` also sets the typings for the `ack` event.
+
+`TFail extends Error = Error` refers to the typing of your **error object** created and passed to `api.fail()` inside of your job processor and defaults to the base `Error` class. Setting `TFail` also sets the typings for your `fail` event.
+
+`TContext = Record<string, unknown>` refers to the **context object** available during processing, and is by default an empty object. The context is available inside of `process()` as well as inside of event callbacks after the processing context is available (`ack`, `fail`, `ping`, `dead`, etc). A `DefaultContext` is made available as a convienence for the Record definition.
+
 ### Adding a Job to the Queue
 
 ```ts
 await queue.enqueue({
   ref: "sample-id",
-  /* SimpleJob */ payload: {
+  /* TData */ payload: {
     success: true,
   },
 });
 ```
 
-#### `enqueue()` Options
+#### `enqueue()` API
 
-`queue.enqueue(job: JobDefinition<T> | JobDefinition<T>[])`
+```ts
+queue.enqueue(job: JobDefinition<TData> | JobDefinition<TData>[])
+```
 
-- `job` the JSON Job object, consisting of
+- `job` the JSON Job object (or an array of job objects), consisting of
   - `ref?: string` an identifier for the job, allowing future `enqueue()` calls to replace the job with new data. Defaults to a v4 UUID
-  - `payload: T` the job's payload which will be saved and sent to the handler
+  - `payload: TData` the job's payload which will be saved and sent to the handler
   - `runAt?: Date` a date object describing when the job should run. Defaults to `now()`
   - `runEvery?: string | null` Either a cron interval or an ISO-8601 duration, or `null` to remove recurrence
   - `timezone?: string | null` When using `runEvery`, you can specify a timezone to make DocMQ aware of durations that cross date-modifying thresholds such as Daylight Savings Time; recommended when using cron and duration values outside of UTC.
@@ -155,7 +176,7 @@ export interface LinearRetryStrategy {
 
 ```ts
 queue.process(
-  async (job /* SampleJob */, api) => {
+  async (job: TData, api: HandlerAPI<TAck, TFail, TContext>) => {
     await api.ack();
   },
   {
@@ -166,27 +187,29 @@ queue.process(
 
 #### `process()` Options
 
-`queue.process(handler: JobHandler<T, A, F>, config?: ProcessorConfig)`
+`queue.process(handler: JobHandler<T, A, F>, config?: ProcessorConfig<C>)`
 
 - `handler` the job handler function, taking the job `T` and the api as arguments, returns a promise
 - `config?: ProcessorConfig` an optional configuration for the processor including
   - `pause?: boolean` should the processor wait to be started, default `false`
   - `concurrency?: number` the number of concurrent processor loops to run, default `1`
   - `visibility?: number` specify the visibility window (how long a job is held for by default) in seconds, default `30`
   - `pollInterval?: number` as a fallback, define how often to check for new jobs in the event that driver does not support evented notifications. Defaults to `5`
+  - `createContext?: () => Promise<TContext> | TContext` generates a unique context of type `TContext` for this run. It will be available in the handler API.
 
 #### `api` Methods and Members
 
-- `api.ref` (string) the ref value of the job
-- `api.attempt` (number) the attempt number for this job
-- `api.visible` (number) the number of seconds this job was originally reserved for
-- `api.ack(result: A)` acknowlegde the job, marking it complete, and scheduling future work
-- `api.fail(reason: string | F)` fail the job and emit the `reason`, scheduling a retry if required
+- `api.ref` (`string`) the ref value of the job
+- `api.attempt` (`number`) the attempt number for this job
+- `api.visible` (`number`) the number of seconds this job was originally reserved for
+- `api.context` (`TContext`) the context object, generated for this run
+- `api.ack(result: TAck)` acknowlegde the job, marking it complete, and scheduling future work
+- `api.fail(reason: string | TFail)` fail the job and emit the `reason`, scheduling a retry if required
 - `api.ping(extendBy: number)` on a long running job, extend the runtime by `extendBy` seconds
 
 ### Events
 
-The `Queue` object has a large number of emitted events available through `queue.events`. It extends `EventEmitter`, and the most common events are below:
+The `Queue` object has a large number of emitted events available through `queue.events`. It extends `EventEmitter`, and the most common events are below. Events related to the processing of a job (`ack`, `fail`, `dead`, and `ping`) will all receive `context: TContext` as a second argument to the event callback
 
 - `ack` when a job was acked successfully
 - `fail` when a job was failed

src/queue.ts

@@ -15,7 +15,8 @@ import {
   type JobDefinition,
   type Driver,
   type EmitterJob,
-  ProcessAPI,
+  type ProcessAPI,
+  type DefaultContext,
 } from "./types.js";
 import { Worker } from "./worker.js";
 import {
@@ -70,17 +71,22 @@ const resetStats = (): QueueStats => ({
  * })
  * ```
  */
-export class Queue<T, A = unknown, F extends Error = Error> {
+export class Queue<
+  TData,
+  TAck = unknown,
+  TFail extends Error = Error,
+  TContext = DefaultContext
+> {
   /**
    * An emitter associated with all interesting events that a queue can create
    * See: {@link Emitter}
    */
-  events: Emitter<T, A, F>;
+  events: Emitter<TData, TAck, TFail, TContext>;
 
   protected name: string;
   protected driver: Driver;
   protected opts: Required<QueueOptions>;
-  protected workers: Worker<T, A, F>[];
+  protected workers: Worker<TData, TAck, TFail, TContext>[];
   protected destroyed: boolean;
   protected statInterval?: ReturnType<typeof setInterval>;
   protected stats: QueueStats;
@@ -104,7 +110,7 @@ export class Queue<T, A = unknown, F extends Error = Error> {
     this.destroyed = false;
     this.workers = [];
     this.driver = driver;
-    this.events = new EventEmitter() as Emitter<T, A, F>;
+    this.events = new EventEmitter() as Emitter<TData, TAck, TFail, TContext>;
     this.opts = {
       retention: {
         jobs: options?.retention?.jobs ?? 3600,
@@ -162,7 +168,7 @@ export class Queue<T, A = unknown, F extends Error = Error> {
    * Add a job to DocMQ
    * @param job A job, specified by {@link JobDefinition}
    */
-  async enqueue(job: JobDefinition<T> | JobDefinition<T>[]) {
+  async enqueue(job: JobDefinition<TData> | JobDefinition<TData>[]) {
     const bulkJobs = Array.isArray(job) ? job : [job];
 
     if (this.destroyed) {
@@ -252,14 +258,12 @@ export class Queue<T, A = unknown, F extends Error = Error> {
     );
 
     // split into success/failure
-    const success: JobDefinition<T>[] = [];
-    const failure: JobDefinition<T>[] = [];
-    const errors: unknown[] = [];
+    const success: JobDefinition<TData>[] = [];
+    const failure: JobDefinition<TData>[] = [];
     results.forEach((r, idx) => {
       const j = bulkJobs[idx];
       if (r.status === "rejected") {
         failure.push(j);
-        errors.push(r.reason);
       } else {
         success.push(j);
       }
@@ -274,7 +278,6 @@ export class Queue<T, A = unknown, F extends Error = Error> {
         "Unable to add the included jobs to the queue"
       );
       err.jobs = failure;
-      err.errors = errors;
       this.events.emit("error", err);
     }
   }
@@ -340,7 +343,10 @@ export class Queue<T, A = unknown, F extends Error = Error> {
    * visibility window for processing, and change how often DocMQ polls for new
    * events when in an idle state.
    */
-  process(handler: JobHandler<T, A, F>, config?: ProcessorConfig): ProcessAPI {
+  process(
+    handler: JobHandler<TData, TAck, TFail, TContext>,
+    config?: ProcessorConfig<TContext>
+  ): ProcessAPI {
     if (this.destroyed) {
       throw new Error("Cannot process a destroyed queue");
     }
@@ -402,14 +408,15 @@ export class Queue<T, A = unknown, F extends Error = Error> {
       // map into a collection of async functions and then run them
       next
         .map((doc) => async () => {
-          const w = new Worker<T, A, F>({
+          const w = new Worker<TData, TAck, TFail, TContext>({
             driver: this.driver,
             name: this.name,
             doc,
-            payload: Queue.decodePayload<T>(doc.payload),
+            payload: Queue.decodePayload<TData>(doc.payload),
             handler,
             emitter: this.events,
             visibility,
+            createContext: config?.createContext,
           });
           this.workers.push(w);
           this.events.emit("process", {
@@ -567,7 +574,7 @@ export class Queue<T, A = unknown, F extends Error = Error> {
 
   /** Add the stat listeners, using our own event system to capture outcomes */
   protected enableStats() {
-    const onFail = (info: EmitterJob<T, A, F>) => {
+    const onFail = (info: EmitterJob<TData, TAck, TFail>) => {
       this.stats.outcomes.failure += 1;
 
       let errorType = "Error";

src/types.ts

@@ -8,6 +8,9 @@ type Returnable = void | null | undefined;
 /** Makes all keys in T required */
 type RequireKeyed<T, K extends keyof T> = T & { [P in K]-?: T[P] };
 
+/** The default context for jobs when processed */
+export type DefaultContext = Record<string, unknown>;
+
 export interface QueueOptions {
   /** Specify alternate retentions for message types */
   retention?: {
@@ -21,7 +24,7 @@ export interface QueueOptions {
   statInterval?: number;
 }
 
-export interface ProcessorConfig {
+export interface ProcessorConfig<C> {
   /** Should the processor be paused on creation? If so, no events will be called until you emit a "start" event. */
   pause?: boolean;
   /** The number of concurrent handlers to run, defaults to `5`. Jobs tend to be IO bound, increasing this number allows for more jobs to run in parallel, but at a higher RPU load in serverless environments such as Mongo Atlas */
@@ -35,6 +38,7 @@ export interface ProcessorConfig {
    * seconds. Defaults to `5`
    */
   pollInterval?: number;
+  createContext?: () => MaybePromise<C>;
 }
 
 export interface FixedRetryStrategy {
@@ -153,13 +157,18 @@ export interface EmitterJob<T = unknown, A = unknown, F = unknown> {
   next?: Date;
 }
 
-export type EmitterJobWithPayload<T, A, F> = RequireKeyed<
-  EmitterJob<T, A, F>,
+export type EmitterJobWithPayload<TData, TAck, TFail> = RequireKeyed<
+  EmitterJob<TData, TAck, TFail>,
   "payload"
 >;
 
 /** DocMQ's EventEmitter makes it easy to attach logging or additional behavior to your workflow */
-export type Emitter<T, A, F extends Error = Error> = EventEmitter<{
+export type Emitter<
+  TData,
+  TAck,
+  TFail extends Error = Error,
+  TContext = DefaultContext
+> = EventEmitter<{
   /** Triggered when the Processor loop goes idle, meaning 0 jobs are currently in-process */
   idle: () => void;
   /** A debug message with additional logging details */
@@ -177,17 +186,24 @@ export type Emitter<T, A, F extends Error = Error> = EventEmitter<{
   /** The processor is stopping */
   stop: () => void;
   /** A set of jobs was added to the queue */
-  add: (jobs: JobDefinition<T>[]) => void;
+  add: (jobs: JobDefinition<TData>[]) => void;
   /** A job was pulled for processing */
-  process: (info: EmitterJob<T, A, F>) => void;
+  process: (info: EmitterJob<TData, TAck, TFail>) => void;
   /** A job was completed successfully */
-  ack: (info: EmitterJobWithPayload<T, A, F>) => void;
+  ack: (
+    info: EmitterJobWithPayload<TData, TAck, TFail>,
+    context: TContext
+  ) => void;
   /** A job has failed one of its execution attempts */
-  fail: (info: EmitterJob<T, A, F>) => void;
+  fail: (info: EmitterJob<TData, TAck, TFail>, context: TContext) => void;
   /** A job has failed all of its execution attempts */
-  dead: (info: EmitterJob<T, A, F>) => void;
+  dead: (info: EmitterJob<TData, TAck, TFail>, context: TContext) => void;
   /** A job asked to extend its visibility window */
-  ping: (info: EmitterJob<T, A, F>, extendBy: number) => void;
+  ping: (
+    info: EmitterJob<TData, TAck, TFail>,
+    extendBy: number,
+    context: TContext
+  ) => void;
   /** A report of statistics for this queue */
   stats: (stats: QueueStats & { queue: string }) => void;
 }>;
@@ -208,27 +224,38 @@ export interface FailureRetryOptions {
   attempt?: number;
 }
 
-export interface HandlerApi<A = unknown, F extends Error = Error> {
+export interface HandlerApi<
+  TAck = unknown,
+  TFail extends Error = Error,
+  TContext = DefaultContext
+> {
   /** The reference value for the job */
   ref: string;
   /** The number of attempts made for this job */
   attempt: number;
   /** How long (seconds) the Job was initially reserved for */
   visible: number;
+  /** The current context for this execution */
+  context: TContext;
   /** Acknowledge "ack" the job, marking it as successfully handled */
-  ack: (result?: A) => Promise<void>;
+  ack: (result?: TAck) => Promise<void>;
   /** Fail the job, triggering any requeue/rescheduling logic */
   fail: (
-    error: DocMQError | F | string,
+    error: DocMQError | TFail | string,
     retryOptions?: FailureRetryOptions
   ) => Promise<void>;
   /** Request to extend the running time for the current job */
   ping: (extendBy: number) => Promise<void>;
 }
 
-export type JobHandler<T = unknown, A = unknown, F extends Error = Error> = (
-  payload: T,
-  api: HandlerApi<A, F>
+export type JobHandler<
+  TData,
+  TAck = unknown,
+  TFail extends Error = Error,
+  TContext = DefaultContext
+> = (
+  payload: TData,
+  api: HandlerApi<TAck, TFail, TContext>
 ) => Promise<unknown>;
 
 /** The DriverEmitter controls events related to the handling of the DB driver */
@@ -306,14 +333,20 @@ export interface Driver<Schema = unknown, Table = unknown, TxInfo = unknown> {
   destroy(): Returnable;
 }
 
-export interface WorkerOptions<T, A, F extends Error = Error> {
+export interface WorkerOptions<
+  TData,
+  TAck,
+  TFail extends Error = Error,
+  TContext = DefaultContext
+> {
   driver: Driver;
   name: string;
   doc: QueueDoc;
-  payload: T;
-  handler: JobHandler<T, A, F>;
-  emitter: Emitter<T, A, F>;
+  payload: TData;
+  handler: JobHandler<TData, TAck, TFail, TContext>;
+  emitter: Emitter<TData, TAck, TFail, TContext>;
   visibility: number;
+  createContext?: () => MaybePromise<TContext>;
 }
 
 export interface QueueStats {