cosmik.network
A social knowledge tool for researchers built on ATProto
updates to event structure
+18
-21
4 changed files
src
shared
domain
events
infrastructure
+2
-4
src/shared/domain/events/IDomainEvent.ts
+2
-4
src/shared/domain/events/IDomainEvent.ts
···
1
1
+
import { EventName } from 'src/shared/infrastructure/events/EventConfig';
1
2
import { UniqueEntityID } from '../UniqueEntityID';
2
3
3
4
export interface IDomainEvent {
5
5
+
eventName: EventName;
4
6
dateTimeOccurred: Date;
5
7
getAggregateId(): UniqueEntityID;
6
8
}
7
7
-
8
8
-
export interface IDomainEventClass {
9
9
-
eventName: string;
10
10
-
}
+3
-8
src/shared/infrastructure/events/BullMQEventPublisher.ts
+3
-8
src/shared/infrastructure/events/BullMQEventPublisher.ts
···
4
4
import { IDomainEvent } from '../../domain/events/IDomainEvent';
5
5
import { Result, ok, err } from '../../core/Result';
6
6
import { QueueNames, QueueOptions } from './QueueConfig';
7
7
-
import { EventNames } from './EventConfig';
7
7
+
import { EventName } from './EventConfig';
8
8
9
9
export class BullMQEventPublisher implements IEventPublisher {
10
10
private queues: Map<string, Queue> = new Map();
···
45
45
});
46
46
}
47
47
48
48
-
private getEventType(event: IDomainEvent): string {
49
49
-
// Use the static eventName property from the event class
50
50
-
const eventClass = event.constructor as any;
51
51
-
if (eventClass.eventName) {
52
52
-
return eventClass.eventName;
53
53
-
}
54
54
-
throw new Error(`Event class ${event.constructor.name} does not have a static eventName property`);
48
48
+
private getEventType(event: IDomainEvent): EventName {
49
49
+
return event.eventName;
55
50
}
56
51
57
52
async close(): Promise<void> {
+12
-8
src/shared/infrastructure/events/BullMQEventSubscriber.ts
+12
-8
src/shared/infrastructure/events/BullMQEventSubscriber.ts
···
1
1
import { Worker, Job } from 'bullmq';
2
2
import Redis from 'ioredis';
3
3
-
import { IEventSubscriber, IEventHandler } from '../../application/events/IEventSubscriber';
3
3
+
import {
4
4
+
IEventSubscriber,
5
5
+
IEventHandler,
6
6
+
} from '../../application/events/IEventSubscriber';
4
7
import { IDomainEvent } from '../../domain/events/IDomainEvent';
5
8
import { CardAddedToLibraryEvent } from '../../../modules/cards/domain/events/CardAddedToLibraryEvent';
6
9
import { CardId } from '../../../modules/cards/domain/value-objects/CardId';
···
16
19
17
20
async subscribe<T extends IDomainEvent>(
18
21
eventType: string,
19
19
-
handler: IEventHandler<T>
22
22
+
handler: IEventHandler<T>,
20
23
): Promise<void> {
21
24
this.handlers.set(eventType, handler);
22
25
}
···
30
33
{
31
34
connection: this.redisConnection,
32
35
concurrency: 10,
33
33
-
}
36
36
+
},
34
37
);
35
38
36
39
worker.on('completed', (job) => {
···
49
52
}
50
53
51
54
async stop(): Promise<void> {
52
52
-
await Promise.all(this.workers.map(worker => worker.close()));
55
55
+
await Promise.all(this.workers.map((worker) => worker.close()));
53
56
this.workers = [];
54
57
}
55
58
56
59
private async processJob(job: Job): Promise<void> {
57
60
const eventData = job.data;
58
61
const eventType = eventData.eventType;
59
59
-
62
62
+
60
63
const handler = this.handlers.get(eventType);
61
64
if (!handler) {
62
65
console.warn(`No handler registered for event type: ${eventType}`);
···
65
68
66
69
const event = this.reconstructEvent(eventData);
67
70
const result = await handler.handle(event);
68
68
-
71
71
+
69
72
if (result.isErr()) {
70
73
throw result.error;
71
74
}
···
76
79
case EventNames.CARD_ADDED_TO_LIBRARY: {
77
80
const cardId = CardId.create(eventData.cardId).unwrap();
78
81
const curatorId = CuratorId.create(eventData.curatorId).unwrap();
79
79
-
82
82
+
80
83
const event = new CardAddedToLibraryEvent(cardId, curatorId);
81
84
(event as any).dateTimeOccurred = new Date(eventData.dateTimeOccurred);
82
82
-
85
85
+
(event as any).eventName = EventNames.CARD_ADDED_TO_LIBRARY;
86
86
+
83
87
return event;
84
88
}
85
89
default: