nandi.uk
This repository has no description
2.3 kB
76 lines
1import { Queue } from 'bullmq';
2import Redis from 'ioredis';
3import { IEventPublisher } from '../../application/events/IEventPublisher';
4import { IDomainEvent } from '../../domain/events/IDomainEvent';
5import { Result, ok, err } from '../../core/Result';
6import { QueueNames, QueueOptions, QueueName } from './QueueConfig';
7import { EventMapper } from './EventMapper';
8import { EventName, EventNames } from './EventConfig';
9
10export class BullMQEventPublisher implements IEventPublisher {
11 private queues: Map<string, Queue> = new Map();
12
13 constructor(private redisConnection: Redis) {}
14
15 async publishEvents(events: IDomainEvent[]): Promise<Result<void>> {
16 try {
17 for (const event of events) {
18 await this.publishSingleEvent(event);
19 }
20 return ok(undefined);
21 } catch (error) {
22 return err(error as Error);
23 }
24 }
25
26 private async publishSingleEvent(event: IDomainEvent): Promise<void> {
27 const targetQueues = this.getTargetQueues(event.eventName);
28
29 for (const queueName of targetQueues) {
30 await this.publishToQueue(queueName, event);
31 }
32 }
33
34 private async publishToQueue(
35 queueName: QueueName,
36 event: IDomainEvent,
37 ): Promise<void> {
38 if (!this.queues.has(queueName)) {
39 this.queues.set(
40 queueName,
41 new Queue(queueName, {
42 connection: this.redisConnection,
43 defaultJobOptions: QueueOptions[queueName],
44 }),
45 );
46 }
47
48 const queue = this.queues.get(queueName)!;
49 const serializedEvent = EventMapper.toSerialized(event);
50 await queue.add(serializedEvent.eventType, serializedEvent);
51 }
52
53 private getTargetQueues(eventName: EventName): QueueName[] {
54 switch (eventName) {
55 case EventNames.CARD_ADDED_TO_LIBRARY:
56 return [
57 QueueNames.FEEDS,
58 QueueNames.SEARCH,
59 QueueNames.NOTIFICATIONS,
60 QueueNames.SYNC,
61 ];
62 case EventNames.CARD_ADDED_TO_COLLECTION:
63 return [QueueNames.FEEDS, QueueNames.NOTIFICATIONS];
64 case EventNames.CARD_REMOVED_FROM_LIBRARY:
65 return [QueueNames.NOTIFICATIONS];
66 default:
67 return [QueueNames.FEEDS];
68 }
69 }
70
71 async close(): Promise<void> {
72 await Promise.all(
73 Array.from(this.queues.values()).map((queue) => queue.close()),
74 );
75 }
76}