import {JobHandler, JobHandlerOptions} from 'job-handler';
// ...
export class MyApplication extends BootMixin(
  ServiceMixin(RepositoryMixin(RestApplication)),
) {
  constructor(options: ApplicationConfig = {}) {
    const opts: JobHandlerOptions = {
      // your configuration
    };
    this.configure(JobHandlerBindings.COMPONENT).to(opts);
    this.component(JobHandler);
    // ...
  }
  // ...
}

import {ContextTags, injectable} from '@loopback/core';
import {Consumer, Job} from 'job-handler';
import {QueueName} from '../config/job-handler.config.ts';

/**
 * type your job as you want
 * be careful, this type cannot be verified and is serialized
 * by bullmq, you can refer to their documentation
 */
type GenerateReportJob = Job<{ids: string[]}, {link: string}>;

@injectable({
  tags: {
    [ContextTags.KEY]: JobHandlerBindings.generateConsumerPathFromName(
      QueueName.GenerateReport,
      'Consumer',
    ),
  },
})
export class GenerateReportConsumer implements Consumer {
  // you can inject anything as for any service in Loopback4
  constructor() {}

  async process(job: GenerateReportJob): Promise<{link: string}> {
    // handle your job
  }
}

import {inject} from '@loopback/context';
import {Job, JobHandler, JobHandlerBindings} from 'job-handler';

import {JobName, QueueName} from '../config/job-handler.config';

export enum JobErrorCode {
  UNKNOWN = 'UNKNOWN',
  CANNOT_GET_QUEUE = 'CANNOT_GET_QUEUE',
  CANNOT_GET_JOB = 'CANNOT_GET_JOB',
  CANNOT_GET_FINISHED_JOB = 'CANNOT_GET_FINISHED_JOB',
  ERROR_ON_JOB = 'ERROR_ON_JOB',
}

export class JobHandlerService {
  constructor(
    @inject(JobHandlerBindings.COMPONENT)
    private readonly jobHandler: JobHandler<QueueName>,
  ) {}

  public async createJobAndWaitUntilFinished<
    R,
    Params extends Record<string, unknown>,
  >(
    queueName: QueueName,
    jobName: JobName,
    params: Params,
    ttl = 10000,
  ): Promise<R> {
    const queue = this.jobHandler.getQueue(queueName);
    if (!queue) {
      throw new Error(JobErrorCode.CANNOT_GET_QUEUE);
    }
    const job = await queue.queue.add(jobName, params);
    if (!job.id) {
      throw new Error(JobErrorCode.CANNOT_GET_JOB);
    }

    /** throws when job fails */
    try {
      // wait TTL sec maximum
      await job.waitUntilFinished(queue.events, ttl);
      const finishedJob = (await Job.fromId(queue.queue, job.id)) as Job<
        Params,
        R
      >;
      if (!finishedJob) {
        throw new Error(JobErrorCode.CANNOT_GET_FINISHED_JOB);
      }
      return finishedJob.returnvalue;
    } catch (error) {
      throw new Error(JobErrorCode.ERROR_ON_JOB);
    }
  }
}

export default JobHandlerService;

import {JobHandlerOptions} from 'job-handler';

export enum QueueName {
  GenerateReports = 'GenerateReports',
  SendReports = 'SendReports',
}

export const JOB_HANDLER_CONFIG: JobHandlerOptions<QueueName> = {
  redisConfig: {
    host: process.env.JOB_HANDLER_REDIS_HOST ?? '127.0.0.1',
    port: +(process.env.JOB_HANDLER_REDIS_PORT ?? 6379),
    password: process.env.JOB_HANDLER_REDIS_PASSWORD ?? '',
  },
  // be careful, this assumes that "JOB_HANDLER_CAN_CONSUME=false" is true
  canConsume: !!(process.env.canConsume ?? false),
  canSchedule: !!(process.env.canSchedule ?? false),
  queues: [
    {
      name: QueueName.GenerateReports,
      workerOptions: {
        concurrency: 1,
      },
    },
    {
      name: QueueName.SendReports,
      workerOptions: {
        concurrency: 2,
      },
    },
  ],
  enabledQueueNames: (process.env.JOB_HANDLER_ENABLED_QUEUE_NAMES ?? '').split(
    ';',
  ) as QueueName[],
  suffixForConsumers: 'Consumer',
};