nandi.uk
This repository has no description
include reference docs
author
Wesley Finck
date
(Jan 29, 2026, 1:18 PM -0800)
commit
910f1791
910f179118a7e169a7d26b001a264c6c4d527aa6
parent
b7d54ff7
b7d54ff72d009b4a349d5c53f9702b5baa6a0edc
+373
+327
.agent/logs/202601229_domain_event_handling_guide.md
+327
.agent/logs/202601229_domain_event_handling_guide.md
···
1
+
# Domain Events Guide
2
+
3
+
This document summarizes how domain events are structured, published, and handled in the codebase, along with a checklist for adding new events.
4
+
5
+
---
6
+
7
+
## Architecture Overview
8
+
9
+
The event system uses **BullMQ** (Redis-backed queues) for asynchronous event processing. Events flow through these layers:
10
+
11
+
```
12
+
Domain Entity → Use Case → Event Publisher → Redis Queue → Worker → Event Handler/Saga
13
+
```
14
+
15
+
---
16
+
17
+
## Key Files & Components
18
+
19
+
| Component | Location | Purpose |
20
+
| -------------------- | ------------------------------------------------------------------ | ------------------------------------------------------ |
21
+
| Event Names Registry | `src/shared/infrastructure/events/EventConfig.ts` | Central registry of all event type names |
22
+
| Event Publisher | `src/shared/infrastructure/events/BullMQEventPublisher.ts` | Routes events to appropriate queues |
23
+
| Event Subscriber | `src/shared/infrastructure/events/BullMQEventSubscriber.ts` | Consumes events from queues and dispatches to handlers |
24
+
| Event Mapper | `src/shared/infrastructure/events/EventMapper.ts` | Serializes/deserializes events for transport |
25
+
| Queue Config | `src/shared/infrastructure/events/QueueConfig.ts` | Queue names and options |
26
+
| Worker Process | `src/shared/infrastructure/processes/NotificationWorkerProcess.ts` | Initializes subscribers and registers handlers |
27
+
28
+
---
29
+
30
+
## Checklist: Adding a New Domain Event
31
+
32
+
### 1. Register the Event Name
33
+
34
+
**File:** `src/shared/infrastructure/events/EventConfig.ts`
35
+
36
+
```typescript
37
+
export const EventNames = {
38
+
// ... existing events
39
+
YOUR_NEW_EVENT: 'YourNewEventEvent',
40
+
} as const;
41
+
```
42
+
43
+
### 2. Create the Domain Event Class
44
+
45
+
**Location:** `src/modules/<module>/domain/events/YourNewEvent.ts`
46
+
47
+
Required structure:
48
+
49
+
```typescript
50
+
import { IDomainEvent } from '../../../../shared/domain/events/IDomainEvent';
51
+
import { UniqueEntityID } from '../../../../shared/domain/UniqueEntityID';
52
+
import { EventNames } from '../../../../shared/infrastructure/events/EventConfig';
53
+
import { Result, ok } from '../../../../shared/core/Result';
54
+
55
+
export class YourNewEvent implements IDomainEvent {
56
+
public readonly eventName = EventNames.YOUR_NEW_EVENT;
57
+
public readonly dateTimeOccurred: Date;
58
+
59
+
private constructor(
60
+
public readonly someId: SomeValueObject,
61
+
dateTimeOccurred?: Date,
62
+
) {
63
+
this.dateTimeOccurred = dateTimeOccurred || new Date();
64
+
}
65
+
66
+
public static create(someId: SomeValueObject): Result<YourNewEvent> {
67
+
return ok(new YourNewEvent(someId));
68
+
}
69
+
70
+
public static reconstruct(
71
+
someId: SomeValueObject,
72
+
dateTimeOccurred: Date,
73
+
): Result<YourNewEvent> {
74
+
return ok(new YourNewEvent(someId, dateTimeOccurred));
75
+
}
76
+
77
+
getAggregateId(): UniqueEntityID {
78
+
return this.someId.getValue();
79
+
}
80
+
}
81
+
```
82
+
83
+
### 3. Raise the Event in the Domain Entity
84
+
85
+
**Location:** Domain entity method (e.g., `src/modules/cards/domain/Card.ts`)
86
+
87
+
```typescript
88
+
import { YourNewEvent } from './events/YourNewEvent';
89
+
90
+
// Inside the domain method that triggers the event:
91
+
public doSomething(): Result<void> {
92
+
// ... business logic ...
93
+
94
+
const domainEvent = YourNewEvent.create(this.someId);
95
+
if (domainEvent.isErr()) {
96
+
return err(new ValidationError(domainEvent.error.message));
97
+
}
98
+
this.addDomainEvent(domainEvent.value);
99
+
100
+
return ok(undefined);
101
+
}
102
+
```
103
+
104
+
### 4. Update the Event Mapper
105
+
106
+
**File:** `src/shared/infrastructure/events/EventMapper.ts`
107
+
108
+
**Step 4a:** Add the import:
109
+
110
+
```typescript
111
+
import { YourNewEvent } from '../../../modules/<module>/domain/events/YourNewEvent';
112
+
```
113
+
114
+
**Step 4b:** Add the serialized interface:
115
+
116
+
```typescript
117
+
export interface SerializedYourNewEvent extends SerializedEvent {
118
+
eventType: typeof EventNames.YOUR_NEW_EVENT;
119
+
someId: string;
120
+
// ... other properties
121
+
}
122
+
```
123
+
124
+
**Step 4c:** Add to the union type:
125
+
126
+
```typescript
127
+
export type SerializedEventUnion =
128
+
| SerializedCardAddedToLibraryEvent
129
+
// ... other events
130
+
| SerializedYourNewEvent;
131
+
```
132
+
133
+
**Step 4d:** Add serialization in `toSerialized()`:
134
+
135
+
```typescript
136
+
if (event instanceof YourNewEvent) {
137
+
return {
138
+
eventType: EventNames.YOUR_NEW_EVENT,
139
+
aggregateId: event.getAggregateId().toString(),
140
+
dateTimeOccurred: event.dateTimeOccurred.toISOString(),
141
+
someId: event.someId.getValue().toString(),
142
+
};
143
+
}
144
+
```
145
+
146
+
**Step 4e:** Add deserialization in `fromSerialized()`:
147
+
148
+
```typescript
149
+
case EventNames.YOUR_NEW_EVENT: {
150
+
const someId = SomeId.createFromString(eventData.someId).unwrap();
151
+
const dateTimeOccurred = new Date(eventData.dateTimeOccurred);
152
+
return YourNewEvent.reconstruct(someId, dateTimeOccurred).unwrap();
153
+
}
154
+
```
155
+
156
+
### 5. Configure Queue Routing
157
+
158
+
**File:** `src/shared/infrastructure/events/BullMQEventPublisher.ts`
159
+
160
+
Update `getTargetQueues()` to route the event to appropriate queues:
161
+
162
+
```typescript
163
+
private getTargetQueues(eventName: EventName): QueueName[] {
164
+
switch (eventName) {
165
+
// ... existing cases
166
+
case EventNames.YOUR_NEW_EVENT:
167
+
return [QueueNames.NOTIFICATIONS]; // or multiple queues
168
+
default:
169
+
return [QueueNames.FEEDS];
170
+
}
171
+
}
172
+
```
173
+
174
+
### 6. Publish Events from the Use Case
175
+
176
+
**File:** The use case that performs the action (e.g., `src/modules/<module>/application/useCases/commands/YourUseCase.ts`)
177
+
178
+
Extend `BaseUseCase` and call `publishEventsForAggregate()`:
179
+
180
+
```typescript
181
+
import { BaseUseCase } from '../../../../../shared/core/UseCase';
182
+
import { IEventPublisher } from '../../../../../shared/application/events/IEventPublisher';
183
+
184
+
export class YourUseCase extends BaseUseCase<RequestDTO, Result<ResponseDTO>> {
185
+
constructor(
186
+
private repository: IRepository,
187
+
eventPublisher: IEventPublisher,
188
+
) {
189
+
super(eventPublisher);
190
+
}
191
+
192
+
async execute(request: RequestDTO): Promise<Result<ResponseDTO>> {
193
+
// ... business logic that raises domain events ...
194
+
195
+
// Publish events for the aggregate
196
+
const publishResult = await this.publishEventsForAggregate(entity);
197
+
if (publishResult.isErr()) {
198
+
console.error('Failed to publish events:', publishResult.error);
199
+
// Don't fail the operation if event publishing fails
200
+
}
201
+
202
+
return ok(response);
203
+
}
204
+
}
205
+
```
206
+
207
+
### 7. Update UseCaseFactory
208
+
209
+
**File:** `src/shared/infrastructure/http/factories/UseCaseFactory.ts`
210
+
211
+
Pass `eventPublisher` to the use case constructor:
212
+
213
+
```typescript
214
+
yourUseCase: new YourUseCase(
215
+
repositories.someRepository,
216
+
services.eventPublisher,
217
+
),
218
+
```
219
+
220
+
### 8. Create the Event Handler
221
+
222
+
**Location:** `src/modules/<module>/application/eventHandlers/YourNewEventHandler.ts`
223
+
224
+
```typescript
225
+
import { YourNewEvent } from '../../../<module>/domain/events/YourNewEvent';
226
+
import { IEventHandler } from '../../../../shared/application/events/IEventSubscriber';
227
+
import { Result, ok } from '../../../../shared/core/Result';
228
+
229
+
export class YourNewEventHandler implements IEventHandler<YourNewEvent> {
230
+
constructor(private someDependency: ISomeDependency) {}
231
+
232
+
async handle(event: YourNewEvent): Promise<Result<void>> {
233
+
// Handle the event
234
+
return ok(undefined);
235
+
}
236
+
}
237
+
```
238
+
239
+
### 9. Register the Handler in the Worker Process
240
+
241
+
**File:** `src/shared/infrastructure/processes/NotificationWorkerProcess.ts` (or appropriate worker)
242
+
243
+
```typescript
244
+
import { YourNewEventHandler } from '../../../modules/<module>/application/eventHandlers/YourNewEventHandler';
245
+
246
+
// In the initialization:
247
+
const yourNewEventHandler = new YourNewEventHandler(dependencies);
248
+
249
+
await subscriber.subscribe(EventNames.YOUR_NEW_EVENT, yourNewEventHandler);
250
+
```
251
+
252
+
---
253
+
254
+
## Event Aggregation with Sagas
255
+
256
+
For events that need to be bundled or coordinated, use a saga pattern:
257
+
258
+
**Example:** `src/modules/notifications/application/sagas/CardNotificationSaga.ts`
259
+
260
+
Sagas can:
261
+
262
+
- Aggregate multiple events within a time window
263
+
- Store pending state in Redis via `ISagaStateStore`
264
+
- Create consolidated notifications
265
+
266
+
Handler delegates to saga:
267
+
268
+
```typescript
269
+
export class YourEventHandler implements IEventHandler<YourEvent> {
270
+
constructor(private saga: YourSaga) {}
271
+
272
+
async handle(event: YourEvent): Promise<Result<void>> {
273
+
return this.saga.handleEvent(event);
274
+
}
275
+
}
276
+
```
277
+
278
+
---
279
+
280
+
## Summary Diagram
281
+
282
+
```
283
+
┌─────────────────────────────────────────────────────────────────────┐
284
+
│ PUBLISHING SIDE │
285
+
├─────────────────────────────────────────────────────────────────────┤
286
+
│ Domain Entity │
287
+
│ └── Raises event via this.addDomainEvent() │
288
+
│ ↓ │
289
+
│ Use Case (extends BaseUseCase) │
290
+
│ └── Calls publishEventsForAggregate() │
291
+
│ ↓ │
292
+
│ BullMQEventPublisher │
293
+
│ ├── EventMapper.toSerialized() - converts to JSON │
294
+
│ └── getTargetQueues() - routes to appropriate queue(s) │
295
+
│ ↓ │
296
+
│ Redis Queues (FEEDS, SEARCH, NOTIFICATIONS) │
297
+
└─────────────────────────────────────────────────────────────────────┘
298
+
↓
299
+
┌─────────────────────────────────────────────────────────────────────┐
300
+
│ CONSUMING SIDE │
301
+
├─────────────────────────────────────────────────────────────────────┤
302
+
│ Worker Process (e.g., NotificationWorkerProcess) │
303
+
│ └── Creates BullMQEventSubscriber │
304
+
│ ↓ │
305
+
│ BullMQEventSubscriber │
306
+
│ ├── Listens to queue │
307
+
│ ├── EventMapper.fromSerialized() - reconstructs event │
308
+
│ └── Dispatches to registered handler │
309
+
│ ↓ │
310
+
│ Event Handler / Saga │
311
+
│ └── Processes event (creates notifications, updates state, etc.) │
312
+
└─────────────────────────────────────────────────────────────────────┘
313
+
```
314
+
315
+
---
316
+
317
+
## Quick Reference: Files to Touch for a New Event
318
+
319
+
1. `src/shared/infrastructure/events/EventConfig.ts` — Add event name
320
+
2. `src/modules/<module>/domain/events/YourNewEvent.ts` — Create event class
321
+
3. `src/modules/<module>/domain/<Entity>.ts` — Raise event in domain method
322
+
4. `src/shared/infrastructure/events/EventMapper.ts` — Add serialization/deserialization
323
+
5. `src/shared/infrastructure/events/BullMQEventPublisher.ts` — Configure queue routing
324
+
6. `src/modules/<module>/application/useCases/...UseCase.ts` — Extend BaseUseCase, publish events
325
+
7. `src/shared/infrastructure/http/factories/UseCaseFactory.ts` — Pass eventPublisher
326
+
8. `src/modules/<module>/application/eventHandlers/...Handler.ts` — Create handler
327
+
9. `src/shared/infrastructure/processes/<Worker>Process.ts` — Register handler
+46
.agent/logs/20260129_repo_sync_notes.md
+46
.agent/logs/20260129_repo_sync_notes.md
···
1
+
```ts
2
+
agent.com.atproto.repo.listRecords
3
+
4
+
listRecords(
5
+
params?: ComAtprotoRepoListRecords.QueryParams,
6
+
opts?: ComAtprotoRepoListRecords.CallOptions,
7
+
): Promise<ComAtprotoRepoListRecords.Response> {
8
+
return this._client.call(
9
+
'com.atproto.repo.listRecords',
10
+
params,
11
+
undefined,
12
+
opts,
13
+
)
14
+
}
15
+
16
+
export type QueryParams = {
17
+
/** The handle or DID of the repo. */
18
+
repo: string
19
+
/** The NSID of the record type. */
20
+
collection: string
21
+
/** The number of records to return. */
22
+
limit?: number
23
+
cursor?: string
24
+
/** Flag to reverse the order of the returned records. */
25
+
reverse?: boolean
26
+
}
27
+
28
+
export interface Response {
29
+
success: boolean
30
+
headers: HeadersMap
31
+
data: OutputSchema
32
+
}
33
+
34
+
export interface OutputSchema {
35
+
cursor?: string
36
+
records: Record[]
37
+
}
38
+
39
+
export interface Record {
40
+
$type?: 'com.atproto.repo.listRecords#record'
41
+
uri: string
42
+
cid: string
43
+
value: { [_ in string]: unknown }
44
+
}
45
+
46
+
```