JobBuffer

// This is a buffer which will collect all the// 'apply-collection-filters' jobs and buffer them.export class CollectionJobBuffer implements JobBuffer<ApplyCollectionFiltersJobData> {  readonly id = 'apply-collection-filters-buffer';  collect(job: Job): boolean {    return job.queueName === 'apply-collection-filters';  }  // When the buffer gets flushed, this function will be passed all the collected jobs  // and will reduce them down to a single job that has aggregated all of the collectionIds.  reduce(collectedJobs: Array<Job<ApplyCollectionFiltersJobData>>): Array<Job<any>> {    // Concatenate all the collectionIds from all the events that were buffered    const collectionIdsToUpdate = collectedJobs.reduce((result, job) => {      return [...result, ...job.data.collectionIds];    }, [] as ID[]);    const referenceJob = collectedJobs[0];    // Create a new Job containing all the concatenated collectionIds,    // de-duplicated to include each collectionId only once.    const batchedCollectionJob = new Job<ApplyCollectionFiltersJobData>({      ...referenceJob,      id: undefined,      data: {        collectionIds: unique(collectionIdsToUpdate),        ctx: referenceJob.data.ctx,        applyToChangedVariantsOnly: referenceJob.data.applyToChangedVariantsOnly,      },    });    // Only this single job will get added to the job queue    return [batchedCollectionJob];  }}

const collectionBuffer = new CollectionJobBuffer();await this.jobQueueService.addBuffer(collectionBuffer);// Here you can perform some work which would ordinarily// trigger the 'apply-collection-filters' job, such as updating// collection filters or changing ProductVariant prices.await this.jobQueueService.flush(collectionBuffer);await this.jobQueueService.removeBuffer(collectionBuffer);

interface JobBuffer<Data extends JobData<Data> = object> {    readonly id: string;    collect(job: Job<Data>): boolean | Promise<boolean>;    reduce(collectedJobs: Array<Job<Data>>): Array<Job<Data>> | Promise<Array<Job<Data>>>;}