nandi.uk
This repository has no description
implement fan-out feed activity to user and collection followers
author
Wesley Finck
date
(Feb 6, 2026, 5:35 PM -0800)
commit
05bfe009
05bfe009ec205b8cd1fafc3f75f14ec425837ecd
parent
21e91504
21e9150438941bdf0b8b24f795a955962c55a6cd
+1999
-1
+13
src/modules/cards/tests/test-utils/createTestSchema.ts
+13
src/modules/cards/tests/test-utils/createTestSchema.ts
···
118
118
created_at TIMESTAMP WITH TIME ZONE NOT NULL DEFAULT NOW(),
119
119
PRIMARY KEY (follower_id, target_id, target_type)
120
120
)`,
121
+
122
+
// Following feed items table (references feed_activities)
123
+
sql`CREATE TABLE IF NOT EXISTS following_feed_items (
124
+
user_id TEXT NOT NULL,
125
+
activity_id UUID NOT NULL REFERENCES feed_activities(id) ON DELETE CASCADE,
126
+
created_at TIMESTAMP NOT NULL,
127
+
PRIMARY KEY (user_id, activity_id)
128
+
)`,
121
129
];
122
130
123
131
// Execute table creation queries in order
···
255
263
`);
256
264
await db.execute(sql`
257
265
CREATE INDEX IF NOT EXISTS idx_follows_target ON follows(target_id, target_type);
266
+
`);
267
+
268
+
// Following feed items indexes
269
+
await db.execute(sql`
270
+
CREATE INDEX IF NOT EXISTS idx_following_feed_user_time ON following_feed_items(user_id, created_at DESC);
258
271
`);
259
272
}
+84
-1
src/modules/feeds/application/useCases/commands/AddActivityToFeedUseCase.ts
+84
-1
src/modules/feeds/application/useCases/commands/AddActivityToFeedUseCase.ts
···
10
10
import { ICardRepository } from '../../../../cards/domain/ICardRepository';
11
11
import { SourceTypeEnum } from '../../../domain/value-objects/SourceType';
12
12
import { ATPROTO_NSID } from '../../../../../shared/constants/atproto';
13
+
import { IFollowsRepository } from '../../../../user/domain/repositories/IFollowsRepository';
14
+
import { IFeedRepository } from '../../../domain/IFeedRepository';
15
+
import {
16
+
FollowTargetType,
17
+
FollowTargetTypeEnum,
18
+
} from '../../../../user/domain/value-objects/FollowTargetType';
13
19
14
20
export interface AddCardCollectedActivityDTO {
15
21
type: ActivityTypeEnum.CARD_COLLECTED;
···
44
50
constructor(
45
51
private feedService: FeedService,
46
52
private cardRepository: ICardRepository,
53
+
private followsRepository: IFollowsRepository,
54
+
private feedRepository: IFeedRepository,
47
55
) {}
48
56
49
57
async execute(
···
153
161
return err(new ValidationError(activityResult.error.message));
154
162
}
155
163
164
+
const activity = activityResult.value;
165
+
166
+
// ========================================
167
+
// PHASE 4: GET FOLLOWERS
168
+
// ========================================
169
+
170
+
// 4a. Get followers of the actor (user who created activity)
171
+
const targetTypeResult = FollowTargetType.create(
172
+
FollowTargetTypeEnum.USER,
173
+
);
174
+
if (targetTypeResult.isErr()) {
175
+
console.error(
176
+
'Failed to create FollowTargetType:',
177
+
targetTypeResult.error,
178
+
);
179
+
return ok({
180
+
activityId: activity.activityId.getStringValue(),
181
+
});
182
+
}
183
+
184
+
const userFollowersResult = await this.followsRepository.getFollowers(
185
+
actorId.value,
186
+
targetTypeResult.value,
187
+
);
188
+
189
+
const userFollowers = userFollowersResult.isOk()
190
+
? userFollowersResult.value.map((f) => f.followerId.value)
191
+
: [];
192
+
193
+
// 4b. Get followers of collections (if any)
194
+
let collectionFollowers: string[] = [];
195
+
if (collectionIds && collectionIds.length > 0) {
196
+
const collectionIdStrings = collectionIds.map((id) =>
197
+
id.getStringValue(),
198
+
);
199
+
const collectionFollowersResult =
200
+
await this.followsRepository.getFollowersOfCollections(
201
+
collectionIdStrings,
202
+
);
203
+
204
+
collectionFollowers = collectionFollowersResult.isOk()
205
+
? collectionFollowersResult.value.map((f) => f.followerId.value)
206
+
: [];
207
+
}
208
+
209
+
// 4c. Combine and deduplicate follower IDs
210
+
const allFollowerIds = new Set([
211
+
...userFollowers,
212
+
...collectionFollowers,
213
+
]);
214
+
215
+
// ========================================
216
+
// PHASE 5: FAN-OUT
217
+
// ========================================
218
+
219
+
if (allFollowerIds.size > 0) {
220
+
const fanOutResult =
221
+
await this.feedRepository.fanOutActivityToFollowers(
222
+
activity.activityId,
223
+
Array.from(allFollowerIds),
224
+
activity.createdAt,
225
+
);
226
+
227
+
// Error handling: Log but don't fail the use case
228
+
// Activity already exists in global feed
229
+
// Event retries will eventually distribute it
230
+
if (fanOutResult.isErr()) {
231
+
console.error(
232
+
'Fan-out failed (will retry on event retry):',
233
+
fanOutResult.error,
234
+
);
235
+
// Note: We do NOT return err here - activity was created successfully
236
+
}
237
+
}
238
+
156
239
return ok({
157
-
activityId: activityResult.value.activityId.getStringValue(),
240
+
activityId: activity.activityId.getStringValue(),
158
241
});
159
242
} catch (error) {
160
243
return err(AppError.UnexpectedError.create(error));
+41
src/modules/feeds/domain/IFeedRepository.ts
+41
src/modules/feeds/domain/IFeedRepository.ts
···
36
36
withinMinutes: number,
37
37
): Promise<Result<FeedActivity | null>>;
38
38
updateActivity(activity: FeedActivity): Promise<Result<void>>;
39
+
40
+
/**
41
+
* Fan-out an activity to multiple followers' following feeds.
42
+
*
43
+
* @param activityId - Activity to distribute
44
+
* @param followerIds - User DIDs to receive this activity (deduplicated by caller)
45
+
* @param createdAt - Activity timestamp (denormalized for sorting)
46
+
* @returns Success or error
47
+
*
48
+
* Idempotency guarantee:
49
+
* - Uses ON CONFLICT DO NOTHING on primary key (user_id, activity_id)
50
+
* - Safe to call multiple times with same inputs
51
+
* - Retries are silent (no error on duplicate)
52
+
*
53
+
* Performance:
54
+
* - Bulk insert operation (single query)
55
+
* - Returns immediately if followerIds is empty (no-op)
56
+
*/
57
+
fanOutActivityToFollowers(
58
+
activityId: ActivityId,
59
+
followerIds: string[],
60
+
createdAt: Date,
61
+
): Promise<Result<void>>;
62
+
63
+
/**
64
+
* Get a user's following feed (paginated).
65
+
*
66
+
* @param userId - User DID whose feed to fetch
67
+
* @param options - Pagination, filters (urlType, source, beforeActivityId)
68
+
* @returns Paginated feed activities
69
+
*
70
+
* Query pattern:
71
+
* - Filters by user_id on following_feed_items
72
+
* - JOINs to feed_activities for full activity data
73
+
* - Supports same filters as global feed (urlType, source)
74
+
* - Cursor-based pagination via beforeActivityId
75
+
*/
76
+
getFollowingFeed(
77
+
userId: string,
78
+
options: FeedQueryOptions,
79
+
): Promise<Result<PaginatedFeedResult>>;
39
80
}
+150
src/modules/feeds/infrastructure/repositories/DrizzleFeedRepository.ts
+150
src/modules/feeds/infrastructure/repositories/DrizzleFeedRepository.ts
···
8
8
import { FeedActivity } from '../../domain/FeedActivity';
9
9
import { ActivityId } from '../../domain/value-objects/ActivityId';
10
10
import { feedActivities } from './schema/feedActivity.sql';
11
+
import { followingFeedItems } from './schema/followingFeedItem.sql';
11
12
import {
12
13
FeedActivityMapper,
13
14
FeedActivityDTO,
···
420
421
.where(eq(feedActivities.id, dto.id));
421
422
422
423
return ok(undefined);
424
+
} catch (error) {
425
+
return err(error as Error);
426
+
}
427
+
}
428
+
429
+
async fanOutActivityToFollowers(
430
+
activityId: ActivityId,
431
+
followerIds: string[],
432
+
createdAt: Date,
433
+
): Promise<Result<void>> {
434
+
try {
435
+
if (followerIds.length === 0) {
436
+
return ok(undefined);
437
+
}
438
+
439
+
const values = followerIds.map((userId) => ({
440
+
userId: userId,
441
+
activityId: activityId.getStringValue(),
442
+
createdAt: createdAt,
443
+
}));
444
+
445
+
await this.db
446
+
.insert(followingFeedItems)
447
+
.values(values)
448
+
.onConflictDoNothing();
449
+
450
+
return ok(undefined);
451
+
} catch (error) {
452
+
return err(error as Error);
453
+
}
454
+
}
455
+
456
+
async getFollowingFeed(
457
+
userId: string,
458
+
options: FeedQueryOptions,
459
+
): Promise<Result<PaginatedFeedResult>> {
460
+
try {
461
+
const { page, limit, beforeActivityId } = options;
462
+
const offset = (page - 1) * limit;
463
+
464
+
// Build where conditions
465
+
const whereConditions = [eq(followingFeedItems.userId, userId)];
466
+
467
+
if (options.urlType) {
468
+
whereConditions.push(eq(feedActivities.urlType, options.urlType));
469
+
}
470
+
471
+
if (options.source) {
472
+
if (options.source === ActivitySource.SEMBLE) {
473
+
whereConditions.push(sql`${feedActivities.source} IS NULL`);
474
+
} else {
475
+
whereConditions.push(eq(feedActivities.source, options.source));
476
+
}
477
+
}
478
+
479
+
// Cursor-based pagination
480
+
if (beforeActivityId) {
481
+
const beforeActivity = await this.db
482
+
.select({ createdAt: followingFeedItems.createdAt })
483
+
.from(followingFeedItems)
484
+
.where(
485
+
and(
486
+
eq(followingFeedItems.userId, userId),
487
+
eq(
488
+
followingFeedItems.activityId,
489
+
beforeActivityId.getStringValue(),
490
+
),
491
+
),
492
+
)
493
+
.limit(1);
494
+
495
+
if (beforeActivity.length > 0) {
496
+
whereConditions.push(
497
+
lt(followingFeedItems.createdAt, beforeActivity[0]!.createdAt),
498
+
);
499
+
}
500
+
}
501
+
502
+
// Main query with JOIN
503
+
const activitiesResult = await this.db
504
+
.select({
505
+
id: feedActivities.id,
506
+
actorId: feedActivities.actorId,
507
+
cardId: feedActivities.cardId,
508
+
type: feedActivities.type,
509
+
metadata: feedActivities.metadata,
510
+
urlType: feedActivities.urlType,
511
+
source: feedActivities.source,
512
+
createdAt: followingFeedItems.createdAt, // Use denormalized timestamp
513
+
})
514
+
.from(followingFeedItems)
515
+
.innerJoin(
516
+
feedActivities,
517
+
eq(feedActivities.id, followingFeedItems.activityId),
518
+
)
519
+
.where(and(...whereConditions))
520
+
.orderBy(
521
+
desc(followingFeedItems.createdAt),
522
+
desc(followingFeedItems.activityId),
523
+
)
524
+
.limit(limit)
525
+
.offset(offset);
526
+
527
+
// Count total (with same filters)
528
+
const totalCountResult = await this.db
529
+
.select({ count: count() })
530
+
.from(followingFeedItems)
531
+
.innerJoin(
532
+
feedActivities,
533
+
eq(feedActivities.id, followingFeedItems.activityId),
534
+
)
535
+
.where(and(...whereConditions));
536
+
537
+
const totalCount = totalCountResult[0]?.count || 0;
538
+
539
+
// Map to domain objects
540
+
const activities: FeedActivity[] = [];
541
+
for (const activityData of activitiesResult) {
542
+
const dto: FeedActivityDTO = {
543
+
id: activityData.id,
544
+
actorId: activityData.actorId,
545
+
cardId: activityData.cardId || undefined,
546
+
type: activityData.type,
547
+
metadata: activityData.metadata as any,
548
+
urlType: activityData.urlType || undefined,
549
+
source: activityData.source || undefined,
550
+
createdAt: activityData.createdAt,
551
+
};
552
+
553
+
const domainResult = FeedActivityMapper.toDomain(dto);
554
+
if (domainResult.isErr()) {
555
+
return err(domainResult.error);
556
+
}
557
+
558
+
activities.push(domainResult.value);
559
+
}
560
+
561
+
const hasMore = offset + activities.length < totalCount;
562
+
const nextCursor =
563
+
hasMore && activities.length > 0
564
+
? activities[activities.length - 1]!.activityId
565
+
: undefined;
566
+
567
+
return ok({
568
+
activities,
569
+
totalCount,
570
+
hasMore,
571
+
nextCursor,
572
+
});
423
573
} catch (error) {
424
574
return err(error as Error);
425
575
}
+29
src/modules/feeds/infrastructure/repositories/schema/followingFeedItem.sql.ts
+29
src/modules/feeds/infrastructure/repositories/schema/followingFeedItem.sql.ts
···
1
+
import {
2
+
pgTable,
3
+
text,
4
+
timestamp,
5
+
uuid,
6
+
index,
7
+
primaryKey,
8
+
} from 'drizzle-orm/pg-core';
9
+
import { feedActivities } from './feedActivity.sql';
10
+
11
+
export const followingFeedItems = pgTable(
12
+
'following_feed_items',
13
+
{
14
+
userId: text('user_id').notNull(), // DID of feed owner
15
+
activityId: uuid('activity_id')
16
+
.notNull()
17
+
.references(() => feedActivities.id, { onDelete: 'cascade' }),
18
+
createdAt: timestamp('created_at').notNull(), // Denormalized from activity for sorting
19
+
},
20
+
(table) => ({
21
+
// Composite primary key
22
+
pk: primaryKey({ columns: [table.userId, table.activityId] }),
23
+
// Index for efficient user feed queries sorted by time
24
+
userTimeIdx: index('idx_following_feed_user_time').on(
25
+
table.userId,
26
+
table.createdAt.desc(),
27
+
),
28
+
}),
29
+
);
+87
src/modules/feeds/tests/infrastructure/InMemoryFeedRepository.ts
+87
src/modules/feeds/tests/infrastructure/InMemoryFeedRepository.ts
···
11
11
export class InMemoryFeedRepository implements IFeedRepository {
12
12
private static instance: InMemoryFeedRepository | null = null;
13
13
private activities: FeedActivity[] = [];
14
+
// Store following feed items as Map<userId, Set<activityId>>
15
+
private followingFeedItems: Map<string, Set<string>> = new Map();
14
16
15
17
private constructor() {}
16
18
···
212
214
}
213
215
}
214
216
217
+
async fanOutActivityToFollowers(
218
+
activityId: ActivityId,
219
+
followerIds: string[],
220
+
createdAt: Date,
221
+
): Promise<Result<void>> {
222
+
try {
223
+
if (followerIds.length === 0) {
224
+
return ok(undefined);
225
+
}
226
+
227
+
const activityIdString = activityId.getStringValue();
228
+
229
+
for (const userId of followerIds) {
230
+
if (!this.followingFeedItems.has(userId)) {
231
+
this.followingFeedItems.set(userId, new Set());
232
+
}
233
+
this.followingFeedItems.get(userId)!.add(activityIdString);
234
+
}
235
+
236
+
return ok(undefined);
237
+
} catch (error) {
238
+
return err(error as Error);
239
+
}
240
+
}
241
+
242
+
async getFollowingFeed(
243
+
userId: string,
244
+
options: FeedQueryOptions,
245
+
): Promise<Result<PaginatedFeedResult>> {
246
+
try {
247
+
const { page, limit, beforeActivityId, urlType } = options;
248
+
249
+
// Get activity IDs for this user's following feed
250
+
const userActivityIds = this.followingFeedItems.get(userId) || new Set();
251
+
252
+
// Filter activities that are in this user's following feed
253
+
let filteredActivities = this.activities.filter((activity) =>
254
+
userActivityIds.has(activity.activityId.getStringValue()),
255
+
);
256
+
257
+
// Filter by URL type if provided
258
+
if (urlType) {
259
+
filteredActivities = filteredActivities.filter(
260
+
(activity) => activity.urlType === urlType,
261
+
);
262
+
}
263
+
264
+
// Filter by cursor if provided
265
+
if (beforeActivityId) {
266
+
const beforeIndex = filteredActivities.findIndex((activity) =>
267
+
activity.activityId.equals(beforeActivityId),
268
+
);
269
+
if (beforeIndex >= 0) {
270
+
filteredActivities = filteredActivities.slice(beforeIndex + 1);
271
+
}
272
+
}
273
+
274
+
// Paginate
275
+
const offset = (page - 1) * limit;
276
+
const paginatedActivities = filteredActivities.slice(
277
+
offset,
278
+
offset + limit,
279
+
);
280
+
281
+
const totalCount = filteredActivities.length;
282
+
const hasMore = offset + paginatedActivities.length < totalCount;
283
+
284
+
let nextCursor: ActivityId | undefined;
285
+
if (hasMore && paginatedActivities.length > 0) {
286
+
nextCursor =
287
+
paginatedActivities[paginatedActivities.length - 1]!.activityId;
288
+
}
289
+
290
+
return ok({
291
+
activities: paginatedActivities,
292
+
totalCount,
293
+
hasMore,
294
+
nextCursor,
295
+
});
296
+
} catch (error) {
297
+
return err(error as Error);
298
+
}
299
+
}
300
+
215
301
// Test helper methods
216
302
clear(): void {
217
303
this.activities = [];
304
+
this.followingFeedItems.clear();
218
305
}
219
306
220
307
getAll(): FeedActivity[] {