Type-safe queues and workers

[yinghaochan]

There's an example of type safe queues and workers here, but it seems like this results in ts errors
#1156 (comment)

export const enum DNQueueName {
  'raw_events_watcher' = '{raw_events_watcher}',
  'raw_events_processor' = '{raw_events_processor}',
};

/** Define the queue jobs and their input/outputs */
interface DNQueue {
  [DNQueueName.raw_events_watcher]: {
    jobs: {
      watchRawEvents: {
        input: {};
        output: void;
      },
      anotherJob: {
        input: { foo: string };
        output: { bar: string };
      },
    }
  };
  [DNQueueName.raw_events_processor]: {
    jobs: {}
  };
}

// @ts-expect-error ignore, works as of when this was written
type QueueData<T extends DNQueueName> = DNQueue[T]['jobs'][keyof DNQueue[T]['jobs']]['input'];

/**

Looks like 'input' or 'output' isn't guaranteed to exist for each job, resulting in the error:
`Type '"input"' cannot be used to index type 'DNQueue[T]["jobs"][keyof DNQueue[T]["jobs"]]'.ts(2536)`


I've tried for a few hours but doesn't seem like a way to fix this with ts@latest.
**/

// @ts-expect-error ignore, works as of when this was written
type QueueReturn<T extends DNQueueName> = DNQueue[T]['jobs'][keyof DNQueue[T]['jobs']]['output'];

type QueueJobNames<T extends DNQueueName> = Extract<keyof DNQueue[T]['jobs'], string>;

type QueueJob<T extends DNQueueName, J = DNQueue[T]['jobs']> = {
  // @ts-expect-error ignore, works as of when this was written
  [K in keyof J]: JobPro<J[K]['input'], J[K]['output'], K>;
}[keyof J];

export class TypedQueue<T extends DNQueueName> extends QueuePro<QueueData<T>, QueueReturn<T>, QueueJobNames<T>> {
  constructor(name: T, opts?: QueueProOptions) {
    super(name, opts)
  }
}

export class TypedWorker<T extends DNQueueName> extends WorkerPro<QueueData<T>, QueueReturn<T>, QueueJobNames<T>> {
  constructor(name: T, fn: <J extends QueueJob<T>>(job: J) => Promise<QueueReturn<T>>, opts?: WorkerProOptions) {
    super(name, fn, opts)
  }
}

How are other users doing type safety when adding to queues?

[yinghaochan]

This is about as good as I could get. It does throw an error on this line: myQueue.add('anotherJob', { raw: 123 }); // NOT OK, invalid job parameters

import {
  Job as JobPro,
  QueueOptions as QueueProOptions,
  WorkerOptions as WorkerProOptions,
  Queue as QueuePro,
  Worker as WorkerPro,
} from 'bullmq';

export const enum DNQueueName {
  'raw_events_watcher' = '{raw_events_watcher}',
  'raw_events_processor' = '{raw_events_processor}',
}

interface DNGeneric {
  [key: string]: {
    jobs: {
      [key: string]: {
        input: any;
        output: any;
      };
    };
  };
}

// Define the queue jobs and their input/outputs
interface DNQueue extends DNGeneric {
  [DNQueueName.raw_events_watcher]: {
    jobs: {
      watchRawEvents: {
        input: { raw: number };
        output: void;
      };
      anotherJob: {
        input: { foo: string };
        output: { bar: string };
      };
    };
  };
  [DNQueueName.raw_events_processor]: {
    jobs: {};
  };
}

// --- THE FIXES ---

/**
 * Helper to get the union of all job definitions for a specific queue.
 * This type is made DISTRIBUTIVE by wrapping it in a conditional.
 * It will now correctly resolve to `never` for 'raw_events_processor'
 * and to a union of jobs for 'raw_events_watcher'.
 */
type QueueJobUnion<T extends DNQueueName> = T extends DNQueueName
  ? DNQueue[T]['jobs'][keyof DNQueue[T]['jobs']]
  : never;

// Now, QueueData and QueueReturn can safely index the result.
type QueueData<T extends DNQueueName> = QueueJobUnion<T>['input'];

type QueueReturn<T extends DNQueueName> = QueueJobUnion<T>['output'];

type QueueJobNames<T extends DNQueueName> = Extract<keyof DNQueue[T]['jobs'], string>;

// This type must also be distributive, as in the original 'Fix 2'.
type QueueJob<T extends DNQueueName> = T extends DNQueueName
  ? {
      [K in keyof DNQueue[T]['jobs']]: JobPro<
        (DNQueue[T]['jobs'][K] & { input: any })['input'],
        (DNQueue[T]['jobs'][K] & { output: any })['output'],
        Extract<K, string>
      >;
    }[keyof DNQueue[T]['jobs']]
  : never;

// --- CLASSES (should now be error-free) ---

export class TypedQueue<T extends DNQueueName> extends QueuePro<
  QueueData<T>,
  QueueReturn<T>,
  QueueJobNames<T>
> {
  constructor(name: T, opts?: QueueProOptions) {
    super(name, opts);
  }
}

export class TypedWorker<T extends DNQueueName> extends WorkerPro<
  QueueData<T>,
  QueueReturn<T>,
  QueueJobNames<T>
> {
  constructor(
    name: T,
    fn: (job: JobPro<QueueData<T>, QueueReturn<T>, QueueJobNames<T>>) => Promise<QueueReturn<T>>,
    opts?: WorkerProOptions
  ) {
    super(name, fn, opts);
  }
}

const myQueue = new TypedQueue(DNQueueName.raw_events_watcher, { connection: '' });

void myQueue.add('watchRawEvents', {}); // OK
void myQueue.add('watchRawEventsXXX', {}); // NOT OK, invalid job name
void myQueue.add('anotherJob', { raw: 123 }); // NOT OK, invalid job parameters

const myWorker = new TypedWorker(
  DNQueueName.raw_events_watcher,
  async (job) => {
    switch (
      job.name // type-checked switch statement
    ) {
      case 'watchRawEvents':
        return;
      case 'anotherJob':
        console.log(job.data.foo); // type-checked job data
        return { bar: '' }; // type-checked response
    }
  },
  { connection: '' }
);