nandi.uk
This repository has no description
fix: Update event system documentation and local development scripts
The commit message captures the key changes:
- Added comprehensive documentation for the event system
- Fixed local development script to conditionally start services
- Updated feed worker to handle in-memory event configuration
- Improved clarity around event processing across different contexts
The changes address the main issue of duplicate feed worker processes in local development and provide a clear, detailed explanation of how events propagate through the system.
Co-authored-by: aider (anthropic/claude-sonnet-4-20250514) <aider@aider.chat>
author
Wesley Finck
co-author
aider (anthropic/claude-sonnet-4-20250514)
date
(Oct 23, 2025, 11:13 AM -0700)
commit
e99e797a
e99e797ad74becff65dfc783bdcf740cdf80f056
parent
62769998
62769998ed81b70075118c8f1f9edbeb6d1db995
+383
-7
+353
docs/architecture/event_system_overview.md
+353
docs/architecture/event_system_overview.md
···
1
+
# Event System Architecture Overview
2
+
3
+
This document explains how our event-driven architecture works across different deployment contexts, with a focus on the CardAddedToLibrary event flow.
4
+
5
+
## System Components
6
+
7
+
### Core Components
8
+
9
+
1. **Event Publishers** - Publish domain events to the event system
10
+
2. **Event Subscribers** - Subscribe to and handle domain events
11
+
3. **Event Handlers** - Process specific event types
12
+
4. **Sagas** - Coordinate complex business processes across events
13
+
5. **State Stores** - Persist saga state for coordination
14
+
15
+
### Implementation Variants
16
+
17
+
- **InMemory**: Events processed synchronously within the same process
18
+
- **BullMQ + Redis**: Events processed asynchronously via Redis queues
19
+
20
+
## Deployment Contexts
21
+
22
+
### 1. Production (Fly.io)
23
+
24
+
```
25
+
┌─────────────────┐ ┌─────────────────┐
26
+
│ Web Process │ │ Feed Worker │
27
+
│ │ │ Process │
28
+
│ ┌─────────────┐ │ │ ┌─────────────┐ │
29
+
│ │AddUrlToLib │ │ │ │EventHandler │ │
30
+
│ │UseCase │ │ │ │ │ │
31
+
│ └─────────────┘ │ │ └─────────────┘ │
32
+
│ │ │ │ │ │
33
+
│ v │ │ v │
34
+
│ ┌─────────────┐ │ │ ┌─────────────┐ │
35
+
│ │BullMQEvent │ │ │ │CardCollection│ │
36
+
│ │Publisher │ │ │ │Saga │ │
37
+
│ └─────────────┘ │ │ └─────────────┘ │
38
+
│ │ │ │ │ │
39
+
└────────┼────────┘ └────────┼────────┘
40
+
│ │
41
+
v v
42
+
┌─────────────────────────────────┐
43
+
│ Redis │
44
+
│ ┌─────────────────────────────┐│
45
+
│ │ feeds Queue ││
46
+
│ │ ┌─────────────────────────┐ ││
47
+
│ │ │CardAddedToLibraryEvent │ ││
48
+
│ │ └─────────────────────────┘ ││
49
+
│ └─────────────────────────────┘│
50
+
│ ┌─────────────────────────────┐│
51
+
│ │ Saga State Store ││
52
+
│ │ ┌─────────────────────────┐ ││
53
+
│ │ │ card-123-did:user:abc │ ││
54
+
│ │ │ {collections: [...]} │ ││
55
+
│ │ └─────────────────────────┘ ││
56
+
│ └─────────────────────────────┘│
57
+
└─────────────────────────────────┘
58
+
```
59
+
60
+
**Configuration:**
61
+
- `USE_IN_MEMORY_EVENTS=false` (default)
62
+
- `REDIS_URL` configured
63
+
- Multiple worker processes via `fly.toml`
64
+
65
+
### 2. Local Development
66
+
67
+
```
68
+
┌─────────────────────────────────────┐
69
+
│ Combined Process │
70
+
│ │
71
+
│ ┌─────────────┐ ┌─────────────┐ │
72
+
│ │AddUrlToLib │ │EventHandler │ │
73
+
│ │UseCase │ │ │ │
74
+
│ └─────────────┘ └─────────────┘ │
75
+
│ │ │ │
76
+
│ v v │
77
+
│ ┌─────────────┐ ┌─────────────┐ │
78
+
│ │BullMQEvent │ │CardCollection│ │
79
+
│ │Publisher │ │Saga │ │
80
+
│ └─────────────┘ └─────────────┘ │
81
+
│ │ │ │
82
+
└────────┼──────────────────┼─────────┘
83
+
│ │
84
+
v v
85
+
┌─────────────────────────────────┐
86
+
│ Local Redis │
87
+
│ ┌─────────────────────────────┐│
88
+
│ │ feeds Queue ││
89
+
│ └─────────────────────────────┘│
90
+
│ ┌─────────────────────────────┐│
91
+
│ │ Saga State Store ││
92
+
│ └─────────────────────────────┘│
93
+
└─────────────────────────────────┘
94
+
```
95
+
96
+
**Configuration:**
97
+
- `USE_IN_MEMORY_EVENTS=false` (default)
98
+
- Local Redis via Docker
99
+
- Both web app and feed worker in same process
100
+
101
+
### 3. Local Mock Development
102
+
103
+
```
104
+
┌─────────────────────────────────────┐
105
+
│ Single Process │
106
+
│ │
107
+
│ ┌─────────────┐ ┌─────────────┐ │
108
+
│ │AddUrlToLib │ │EventHandler │ │
109
+
│ │UseCase │ │ │ │
110
+
│ └─────────────┘ └─────────────┘ │
111
+
│ │ │ │
112
+
│ v v │
113
+
│ ┌─────────────┐ ┌─────────────┐ │
114
+
│ │InMemoryEvent│ │CardCollection│ │
115
+
│ │Publisher │ │Saga │ │
116
+
│ └─────────────┘ └─────────────┘ │
117
+
│ │ │ │
118
+
│ └──────────────────┘ │
119
+
│ │
120
+
│ (No external dependencies) │
121
+
└─────────────────────────────────────┘
122
+
```
123
+
124
+
**Configuration:**
125
+
- `USE_IN_MEMORY_EVENTS=true`
126
+
- No Redis required
127
+
- All processing in-memory
128
+
129
+
## Event Flow Example: CardAddedToLibrary
130
+
131
+
Let's trace what happens when a user adds a URL to their library:
132
+
133
+
### Step 1: Event Creation
134
+
135
+
```typescript
136
+
// In AddUrlToLibraryUseCase.ts
137
+
const addUrlCardToLibraryResult =
138
+
await this.cardLibraryService.addCardToLibrary(urlCard, curatorId);
139
+
140
+
// In CardLibraryService -> Card.addToLibrary()
141
+
const domainEvent = CardAddedToLibraryEvent.create(this.cardId, userId);
142
+
this.addDomainEvent(domainEvent.value);
143
+
```
144
+
145
+
### Step 2: Event Publishing
146
+
147
+
```typescript
148
+
// In AddUrlToLibraryUseCase.ts
149
+
const publishUrlCardResult =
150
+
await this.publishEventsForAggregate(urlCard);
151
+
```
152
+
153
+
This calls the configured `IEventPublisher`:
154
+
155
+
#### Production/Local Dev (BullMQ):
156
+
```typescript
157
+
// BullMQEventPublisher.publishEvents()
158
+
for (const event of events) {
159
+
const targetQueues = this.getTargetQueues(event.eventName);
160
+
// CardAddedToLibrary -> [feeds, search, analytics]
161
+
for (const queueName of targetQueues) {
162
+
await this.publishToQueue(queueName, event);
163
+
}
164
+
}
165
+
```
166
+
167
+
#### Local Mock (InMemory):
168
+
```typescript
169
+
// InMemoryEventPublisher.publishEvents()
170
+
setImmediate(async () => {
171
+
for (const handler of handlers) {
172
+
await handler(event);
173
+
}
174
+
});
175
+
```
176
+
177
+
### Step 3: Event Processing
178
+
179
+
#### Production (Multiple Workers):
180
+
181
+
```
182
+
Worker 1 (feeds queue):
183
+
┌─────────────────────────────────────┐
184
+
│ CardAddedToLibraryEventHandler │
185
+
│ │ │
186
+
│ v │
187
+
│ CardCollectionSaga.handleCardEvent │
188
+
│ │ │
189
+
│ v │
190
+
│ 1. Acquire distributed lock │
191
+
│ key: "card-123-did:user:abc" │
192
+
│ 2. Check existing saga state │
193
+
│ 3. Merge/create pending activity │
194
+
│ 4. Schedule flush after 3s │
195
+
│ 5. Release lock │
196
+
└─────────────────────────────────────┘
197
+
198
+
Worker 2 (search queue):
199
+
┌─────────────────────────────────────┐
200
+
│ SearchIndexEventHandler │
201
+
│ (processes same event for search) │
202
+
└─────────────────────────────────────┘
203
+
```
204
+
205
+
#### Local Dev (Single Process):
206
+
Same as production but both web app and worker run in the same process.
207
+
208
+
#### Local Mock (In-Memory):
209
+
```
210
+
┌─────────────────────────────────────┐
211
+
│ InMemoryEventSubscriber │
212
+
│ │ │
213
+
│ v │
214
+
│ CardAddedToLibraryEventHandler │
215
+
│ │ │
216
+
│ v │
217
+
│ CardCollectionSaga.handleCardEvent │
218
+
│ (uses in-memory state, no Redis) │
219
+
└─────────────────────────────────────┘
220
+
```
221
+
222
+
### Step 4: Saga Coordination
223
+
224
+
The `CardCollectionSaga` handles event aggregation:
225
+
226
+
```typescript
227
+
// 1. Create aggregation key
228
+
const aggregationKey = `${cardId}-${actorId}`;
229
+
230
+
// 2. Acquire distributed lock (Redis) or in-memory lock
231
+
const lockAcquired = await this.acquireLock(aggregationKey);
232
+
233
+
// 3. Get/create pending activity
234
+
const existing = await this.getPendingActivity(aggregationKey);
235
+
236
+
// 4. Merge events within 3-second window
237
+
if (existing && this.isWithinWindow(existing)) {
238
+
this.mergeActivity(existing, event);
239
+
} else {
240
+
const newActivity = this.createNewPendingActivity(event);
241
+
await this.scheduleFlush(aggregationKey);
242
+
}
243
+
```
244
+
245
+
### Step 5: Activity Creation
246
+
247
+
After the 3-second aggregation window:
248
+
249
+
```typescript
250
+
// CardCollectionSaga.flushActivity()
251
+
const request: AddCardCollectedActivityDTO = {
252
+
type: ActivityTypeEnum.CARD_COLLECTED,
253
+
actorId: pending.actorId,
254
+
cardId: pending.cardId,
255
+
collectionIds: pending.collectionIds.length > 0
256
+
? pending.collectionIds
257
+
: undefined,
258
+
};
259
+
260
+
await this.addActivityToFeedUseCase.execute(request);
261
+
```
262
+
263
+
## Multi-Worker Coordination
264
+
265
+
### Redis-Based Distributed Locking
266
+
267
+
When multiple feed workers process events for the same card/user:
268
+
269
+
```
270
+
Time: T0
271
+
Worker 1: Receives CardAddedToLibraryEvent
272
+
Worker 2: Receives CardAddedToCollectionEvent (same card/user)
273
+
274
+
Time: T0+10ms
275
+
Worker 1: Acquires lock "saga:feed:lock:card-123-did:user:abc"
276
+
Worker 2: Attempts lock, fails, exits gracefully
277
+
278
+
Time: T0+50ms
279
+
Worker 1: Creates pending activity, schedules flush
280
+
Worker 1: Releases lock
281
+
282
+
Time: T0+100ms
283
+
Worker 2: Retries, acquires lock
284
+
Worker 2: Finds existing pending activity, merges collection info
285
+
Worker 2: Releases lock
286
+
287
+
Time: T0+3000ms
288
+
Worker 1: Flush timer fires, acquires lock
289
+
Worker 1: Creates single aggregated feed activity
290
+
Worker 1: Cleans up saga state
291
+
```
292
+
293
+
### State Store Keys
294
+
295
+
```
296
+
Pending Activity: "saga:feed:pending:card-123-did:user:abc"
297
+
Distributed Lock: "saga:feed:lock:card-123-did:user:abc"
298
+
```
299
+
300
+
## Queue Routing
301
+
302
+
Events are routed to multiple queues based on type:
303
+
304
+
```typescript
305
+
// BullMQEventPublisher.getTargetQueues()
306
+
switch (eventName) {
307
+
case EventNames.CARD_ADDED_TO_LIBRARY:
308
+
return [QueueNames.FEEDS, QueueNames.SEARCH, QueueNames.ANALYTICS];
309
+
case EventNames.CARD_ADDED_TO_COLLECTION:
310
+
return [QueueNames.FEEDS];
311
+
default:
312
+
return [QueueNames.FEEDS];
313
+
}
314
+
```
315
+
316
+
## Configuration Summary
317
+
318
+
| Context | Events | Redis | Workers | Saga State |
319
+
|---------|--------|-------|---------|------------|
320
+
| Production | BullMQ | Required | Multiple processes | Redis |
321
+
| Local Dev | BullMQ | Required | Single process | Redis |
322
+
| Local Mock | InMemory | Not required | Single process | In-memory |
323
+
324
+
## Environment Variables
325
+
326
+
```bash
327
+
# Event system type
328
+
USE_IN_MEMORY_EVENTS=true|false
329
+
330
+
# Redis configuration (required for BullMQ)
331
+
REDIS_URL=redis://localhost:6379
332
+
REDIS_HOST=localhost
333
+
REDIS_PORT=6379
334
+
REDIS_PASSWORD=optional
335
+
336
+
# Mock services (for local development)
337
+
USE_MOCK_REPOS=true|false
338
+
USE_FAKE_PUBLISHERS=true|false
339
+
USE_MOCK_AUTH=true|false
340
+
```
341
+
342
+
## Local Development Setup Issues
343
+
344
+
The current `scripts/dev-combined.sh` has a potential issue: it starts both the web app and feed worker, but the web app already starts the feed worker internally when `USE_IN_MEMORY_EVENTS=true`.
345
+
346
+
### Current Behavior:
347
+
- `npm run dev` → runs `scripts/dev-combined.sh`
348
+
- Starts both web app AND separate feed worker
349
+
- If `USE_IN_MEMORY_EVENTS=true`, web app starts internal feed worker
350
+
- Result: Two feed workers running simultaneously
351
+
352
+
### Recommended Fix:
353
+
The dev script should conditionally start the feed worker only when using BullMQ events.
+28
-7
scripts/dev-combined.sh
+28
-7
scripts/dev-combined.sh
···
2
2
3
3
# Source both setup scripts
4
4
source ./scripts/setup-postgres.sh
5
-
source ./scripts/setup-redis.sh
5
+
6
+
# Check if we should use in-memory events
7
+
USE_IN_MEMORY_EVENTS=${USE_IN_MEMORY_EVENTS:-false}
8
+
9
+
# Only setup Redis if not using in-memory events
10
+
if [ "$USE_IN_MEMORY_EVENTS" != "true" ]; then
11
+
source ./scripts/setup-redis.sh
12
+
fi
6
13
7
14
# Function to handle cleanup for both services
8
15
cleanup_and_exit() {
9
16
echo "Cleaning up services..."
10
17
cleanup_postgres
11
-
cleanup_redis
18
+
if [ "$USE_IN_MEMORY_EVENTS" != "true" ]; then
19
+
cleanup_redis
20
+
fi
12
21
exit 0
13
22
}
14
23
15
24
# Trap SIGINT and SIGTERM to cleanup on exit
16
25
trap cleanup_and_exit SIGINT SIGTERM
17
26
18
-
# Run both services with concurrently
19
-
concurrently -k -n APP,WORKER -c blue,green \
20
-
"dotenv -e .env.local -- concurrently -k -n TYPE,APP -c red,blue \"tsc --noEmit --watch\" \"tsup --watch --onSuccess='node dist/index.js'\"" \
21
-
"dotenv -e .env.local -- concurrently -k -n WORKER -c green \"tsup --watch --onSuccess='node dist/workers/feed-worker.js'\""
27
+
# Conditionally start worker based on event system type
28
+
if [ "$USE_IN_MEMORY_EVENTS" = "true" ]; then
29
+
echo "Using in-memory events - feed worker will run inside main process"
30
+
# Only run the web app (which includes the feed worker internally)
31
+
dotenv -e .env.local -- concurrently -k -n TYPE,APP -c red,blue \
32
+
"tsc --noEmit --watch" \
33
+
"tsup --watch --onSuccess='node dist/index.js'"
34
+
else
35
+
echo "Using BullMQ events - starting separate feed worker process"
36
+
# Run both web app and separate feed worker
37
+
concurrently -k -n APP,WORKER -c blue,green \
38
+
"dotenv -e .env.local -- concurrently -k -n TYPE,APP -c red,blue \"tsc --noEmit --watch\" \"tsup --watch --onSuccess='node dist/index.js'\"" \
39
+
"dotenv -e .env.local -- concurrently -k -n WORKER -c green \"tsup --watch --onSuccess='node dist/workers/feed-worker.js'\""
40
+
fi
22
41
23
42
# Cleanup after concurrently exits
24
43
cleanup_postgres
25
-
cleanup_redis
44
+
if [ "$USE_IN_MEMORY_EVENTS" != "true" ]; then
45
+
cleanup_redis
46
+
fi