Changing all queue-related services and modules to use async provider. Changed email jobs to not be named which was preventing processing

src/email/client.ts

@@ -78,7 +78,7 @@ export class EmailClient {
 
 	private async handleNewMail(
 		numMessages: number,
-		callback: (emails: Email[]) => void,
+		callback: (emails: Email[]) => Promise<void>,
 	) {
 		this.loggerService.log(`Received ${numMessages} emails`)
 		const mailbox = await new Promise<Imap.Box>((res) => this.getMailbox(res))
@@ -96,11 +96,11 @@ export class EmailClient {
 				totalMessages - (numMessages - 1),
 			)
 			this.loggerService.debug(`Fetched ${emails.length} emails`)
-			callback(emails)
+			await callback(emails)
 		}
 	}
 
-	private _listen(callback: (emails: Email[]) => void) {
+	private _listen(callback: (emails: Email[]) => Promise<void>) {
 		// Don't start listening until we are ready
 		if (this.status === EmailClientStatus.NotReady) {
 			throw new Error('Not ready to listen')
@@ -116,10 +116,13 @@ export class EmailClient {
 			this.mailbox = mailbox
 		})
 
-		this.imapClient.on('mail', (n: number) => this.handleNewMail(n, callback))
+		this.imapClient.on(
+			'mail',
+			(n: number) => this.handleNewMail(n, callback),
+		)
 	}
 
-	public listen(callback: (emails: Email[]) => void) {
+	public listen(callback: (emails: Email[]) => Promise<void>) {
 		this.attempt('listen', 0, 5, 1000, () => {
 			this._listen(callback)
 		})
@@ -198,7 +201,7 @@ export class EmailClient {
 	}
 
 	public async markMailsSeen(messageIds: string[]) {
-		this.loggerService.debug('Marking mails as seen', { messageIds })
+		this.loggerService.debug(`Marking mails as seen (${messageIds.join(',')})`)
 		return new Promise<void>((res, rej) => {
 			this.imapClient.addFlags(messageIds.join(','), ['\\Seen'], (error) => {
 				if (error != null) {

src/email/module.ts

@@ -9,7 +9,7 @@ import { EmailProvider } from './provider'
 @Module({
 	imports: [
 		LoggerModule,
-		BullModule.registerQueue({ name: EMAILS_QUEUE_NAME }),
+		BullModule.registerQueueAsync({ name: EMAILS_QUEUE_NAME }),
 	],
 	providers: [EmailClient, EmailProvider],
 	exports: [EmailClient, EmailProvider],

src/email/provider.ts

@@ -20,7 +20,7 @@ export class EmailProvider {
 
 	private async handleReceivedEmails(emails: Email[]) {
 		await this.emailsQueue.addBulk(
-			emails.map((email) => ({ name: email.id, data: email })),
+			emails.map((email) => ({ data: email })),
 		)
 	}
 

src/ntfy/module.ts

@@ -8,7 +8,7 @@ import { NTFY_PUBLISH_QUEUE_NAME } from './types'
 
 @Module({
 	imports: [
-		BullModule.registerQueue({ name: NTFY_PUBLISH_QUEUE_NAME }),
+		BullModule.registerQueueAsync({ name: NTFY_PUBLISH_QUEUE_NAME }),
 		LoggerModule,
 	],
 	providers: [NtfyProvider, NtfyClient],

src/ntfy/provider.ts

@@ -5,12 +5,7 @@ import { Job, JobOptions, Queue } from 'bull'
 import { Dayjs } from 'dayjs'
 
 import { NtfyClient } from './client'
-import {
-	MessageConfig,
-	MessagePriority,
-	MessageTags,
-	NTFY_PUBLISH_QUEUE_NAME,
-} from './types'
+import { MessageConfig, MessageTags, NTFY_PUBLISH_QUEUE_NAME } from './types'
 
 @Processor(NTFY_PUBLISH_QUEUE_NAME)
 @Injectable()

src/reservations/module.ts

@@ -16,7 +16,7 @@ import { ReservationsWorker } from './worker'
 	imports: [
 		LoggerModule,
 		TypeOrmModule.forFeature([Reservation]),
-		BullModule.registerQueue({ name: RESERVATIONS_QUEUE_NAME }),
+		BullModule.registerQueueAsync({ name: RESERVATIONS_QUEUE_NAME }),
 		RunnerModule,
 		NtfyModule,
 	],

src/runner/module.ts

@@ -8,7 +8,10 @@ import { RunnerService } from './service'
 
 @Module({
 	providers: [RunnerService, BaanReserverenService, EmptyPageFactory],
-	imports: [LoggerModule, BullModule.registerQueue({ name: 'reservations' })],
+	imports: [
+		LoggerModule,
+		BullModule.registerQueueAsync({ name: 'reservations' }),
+	],
 	exports: [EmptyPageFactory, BaanReserverenService],
 })
 export class RunnerModule {}

src/waitingList/module.ts

@@ -13,8 +13,8 @@ import { WaitingListService } from './service'
 	imports: [
 		LoggerModule,
 		ReservationsModule,
-		BullModule.registerQueue({ name: EMAILS_QUEUE_NAME }),
-		BullModule.registerQueue({ name: RESERVATIONS_QUEUE_NAME }),
+		BullModule.registerQueueAsync({ name: EMAILS_QUEUE_NAME }),
+		BullModule.registerQueueAsync({ name: RESERVATIONS_QUEUE_NAME }),
 		EmailModule,
 		NtfyModule,
 	],

src/waitingList/service.ts

@@ -1,5 +1,5 @@
 import { InjectQueue, Process, Processor } from '@nestjs/bull'
-import { Inject } from '@nestjs/common'
+import { Inject, Injectable } from '@nestjs/common'
 import { Job, Queue } from 'bull'
 import { NtfyProvider } from 'src/ntfy/provider'
 
@@ -25,6 +25,7 @@ const EMAIL_START_TIME_REGEX = new RegExp(
 const EMAIL_END_TIME_REGEX = new RegExp(/^Eindtijd: ([0-9]{1,2}:[0-9]{1,2})$/im)
 
 @Processor(EMAILS_QUEUE_NAME)
+@Injectable()
 export class WaitingListService {
 	constructor(
 		@InjectQueue(RESERVATIONS_QUEUE_NAME)