nandi.uk
This repository has no description
20 kB
648 lines
1import {
2 eq,
3 desc,
4 lt,
5 count,
6 sql,
7 and,
8 gte,
9 inArray,
10 notInArray,
11} from 'drizzle-orm';
12import { PostgresJsDatabase } from 'drizzle-orm/postgres-js';
13import {
14 IFeedRepository,
15 FeedQueryOptions,
16 PaginatedFeedResult,
17} from '../../domain/IFeedRepository';
18import { FeedActivity } from '../../domain/FeedActivity';
19import { ActivityId } from '../../domain/value-objects/ActivityId';
20import { feedActivities } from './schema/feedActivity.sql';
21import { followingFeedItems } from './schema/followingFeedItem.sql';
22import {
23 FeedActivityMapper,
24 FeedActivityDTO,
25} from './mappers/FeedActivityMapper';
26import { Result, ok, err } from '../../../../shared/core/Result';
27import { CollectionId } from '../../../cards/domain/value-objects/CollectionId';
28import { CuratorId } from '../../../cards/domain/value-objects/CuratorId';
29import { CardId } from '../../../cards/domain/value-objects/CardId';
30import { ActivityTypeEnum } from '../../domain/value-objects/ActivityType';
31import { ActivitySource } from '@semble/types';
32import { KNOWN_BOT_DIDS } from '../../../../shared/constants/knownBots';
33
34export class DrizzleFeedRepository implements IFeedRepository {
35 constructor(private db: PostgresJsDatabase) {}
36
37 async addActivity(activity: FeedActivity): Promise<Result<void>> {
38 try {
39 const dto = FeedActivityMapper.toPersistence(activity);
40
41 await this.db.insert(feedActivities).values({
42 id: dto.id,
43 actorId: dto.actorId,
44 cardId: dto.cardId,
45 connectionId: dto.connectionId,
46 type: dto.type,
47 metadata: dto.metadata,
48 urlType: dto.urlType,
49 source: dto.source,
50 createdAt: dto.createdAt,
51 });
52
53 return ok(undefined);
54 } catch (error) {
55 return err(error as Error);
56 }
57 }
58
59 async getGlobalFeed(
60 options: FeedQueryOptions,
61 ): Promise<Result<PaginatedFeedResult>> {
62 try {
63 const { page, limit, beforeActivityId } = options;
64 const offset = (page - 1) * limit;
65
66 // Build query conditionally
67 let activitiesResult: Array<{
68 id: string;
69 actorId: string;
70 cardId: string | null;
71 connectionId: string | null;
72 type: string;
73 metadata: any;
74 urlType: string | null;
75 source: string | null;
76 createdAt: Date;
77 }>;
78
79 // Build where conditions
80 const whereConditions = [];
81 if (options.activityTypes && options.activityTypes.length > 0) {
82 if (options.activityTypes.length === 1) {
83 whereConditions.push(
84 eq(feedActivities.type, options.activityTypes[0]!),
85 );
86 } else {
87 whereConditions.push(
88 inArray(feedActivities.type, options.activityTypes),
89 );
90 }
91 }
92 if (options.urlType) {
93 whereConditions.push(eq(feedActivities.urlType, options.urlType));
94 }
95 if (options.source) {
96 if (options.source === ActivitySource.SEMBLE) {
97 // Semble content has source IS NULL
98 whereConditions.push(sql`${feedActivities.source} IS NULL`);
99 } else {
100 // Direct match for other sources (e.g., ActivitySource.MARGIN)
101 whereConditions.push(eq(feedActivities.source, options.source));
102 }
103 }
104 // Filter out known bots by default (unless explicitly included)
105 if (!options.includeKnownBots && KNOWN_BOT_DIDS.length > 0) {
106 whereConditions.push(
107 notInArray(feedActivities.actorId, KNOWN_BOT_DIDS),
108 );
109 }
110
111 if (beforeActivityId) {
112 // Get the timestamp of the beforeActivityId
113 const beforeActivity = await this.db
114 .select({ createdAt: feedActivities.createdAt })
115 .from(feedActivities)
116 .where(eq(feedActivities.id, beforeActivityId.getStringValue()))
117 .limit(1);
118
119 if (beforeActivity.length > 0) {
120 const conditions = [
121 lt(feedActivities.createdAt, beforeActivity[0]!.createdAt),
122 ...whereConditions,
123 ];
124 activitiesResult = await this.db
125 .select()
126 .from(feedActivities)
127 .where(conditions.length > 1 ? and(...conditions) : conditions[0])
128 .orderBy(desc(feedActivities.createdAt), desc(feedActivities.id))
129 .limit(limit);
130 } else {
131 // If beforeActivityId doesn't exist, return empty result
132 activitiesResult = [];
133 }
134 } else {
135 // Regular pagination without cursor
136 const query = this.db.select().from(feedActivities);
137
138 if (whereConditions.length > 0) {
139 query.where(
140 whereConditions.length > 1
141 ? and(...whereConditions)
142 : whereConditions[0],
143 );
144 }
145
146 activitiesResult = await query
147 .orderBy(desc(feedActivities.createdAt), desc(feedActivities.id))
148 .limit(limit)
149 .offset(offset);
150 }
151
152 // Get total count with same filters
153 const countQuery = this.db
154 .select({ count: count() })
155 .from(feedActivities);
156
157 if (whereConditions.length > 0) {
158 countQuery.where(
159 whereConditions.length > 1
160 ? and(...whereConditions)
161 : whereConditions[0],
162 );
163 }
164
165 const totalCountResult = await countQuery;
166
167 const totalCount = totalCountResult[0]?.count || 0;
168
169 // Map to domain objects
170 const activities: FeedActivity[] = [];
171 for (const activityData of activitiesResult) {
172 const dto: FeedActivityDTO = {
173 id: activityData.id,
174 actorId: activityData.actorId,
175 cardId: activityData.cardId || undefined,
176 connectionId: activityData.connectionId || undefined,
177 type: activityData.type,
178 metadata: activityData.metadata as any,
179 urlType: activityData.urlType || undefined,
180 source: activityData.source || undefined,
181 createdAt: activityData.createdAt,
182 };
183
184 const domainResult = FeedActivityMapper.toDomain(dto);
185 if (domainResult.isErr()) {
186 return err(domainResult.error);
187 }
188
189 activities.push(domainResult.value);
190 }
191
192 // Determine if there are more activities
193 const hasMore = offset + activities.length < totalCount;
194
195 // Set next cursor if there are more activities
196 let nextCursor: ActivityId | undefined;
197 if (hasMore && activities.length > 0) {
198 const lastActivity = activities[activities.length - 1]!;
199 nextCursor = lastActivity.activityId;
200 }
201
202 return ok({
203 activities,
204 totalCount,
205 hasMore,
206 nextCursor,
207 });
208 } catch (error) {
209 return err(error as Error);
210 }
211 }
212
213 async getGemsFeed(
214 collectionIds: CollectionId[],
215 options: FeedQueryOptions,
216 ): Promise<Result<PaginatedFeedResult>> {
217 try {
218 const { page, limit, beforeActivityId } = options;
219 const offset = (page - 1) * limit;
220 const collectionIdStrings = collectionIds.map((id) =>
221 id.getStringValue(),
222 );
223
224 // Handle empty collection IDs array
225 if (collectionIdStrings.length === 0) {
226 return ok({
227 activities: [],
228 totalCount: 0,
229 hasMore: false,
230 nextCursor: undefined,
231 });
232 }
233
234 // Build query conditionally
235 let activitiesResult: Array<{
236 id: string;
237 actorId: string;
238 cardId: string | null;
239 connectionId: string | null;
240 type: string;
241 metadata: any;
242 urlType: string | null;
243 source: string | null;
244 createdAt: Date;
245 }>;
246
247 // Build where conditions for gems feed
248 const whereConditions = [];
249 if (options.activityTypes && options.activityTypes.length > 0) {
250 if (options.activityTypes.length === 1) {
251 whereConditions.push(
252 eq(feedActivities.type, options.activityTypes[0]!),
253 );
254 } else {
255 whereConditions.push(
256 inArray(feedActivities.type, options.activityTypes),
257 );
258 }
259 }
260 if (options.urlType) {
261 whereConditions.push(eq(feedActivities.urlType, options.urlType));
262 }
263 if (options.source) {
264 if (options.source === ActivitySource.SEMBLE) {
265 // Semble content has source IS NULL
266 whereConditions.push(sql`${feedActivities.source} IS NULL`);
267 } else {
268 // Direct match for other sources (e.g., ActivitySource.MARGIN)
269 whereConditions.push(eq(feedActivities.source, options.source));
270 }
271 }
272 // Filter out known bots by default (unless explicitly included)
273 if (!options.includeKnownBots && KNOWN_BOT_DIDS.length > 0) {
274 whereConditions.push(
275 notInArray(feedActivities.actorId, KNOWN_BOT_DIDS),
276 );
277 }
278
279 // Create the JSON array condition using jsonb_array_elements_text
280 const arrayLiteral = `{${collectionIdStrings.map((id) => `"${id}"`).join(',')}}`;
281 const jsonArrayCondition = sql`EXISTS (
282 SELECT 1 FROM jsonb_array_elements_text(${feedActivities.metadata}->'collectionIds') AS collection_id
283 WHERE collection_id = ANY(${arrayLiteral}::text[])
284 )`;
285
286 if (beforeActivityId) {
287 // Get the timestamp of the beforeActivityId
288 const beforeActivity = await this.db
289 .select({ createdAt: feedActivities.createdAt })
290 .from(feedActivities)
291 .where(eq(feedActivities.id, beforeActivityId.getStringValue()))
292 .limit(1);
293
294 if (beforeActivity.length > 0) {
295 const conditions = [
296 lt(feedActivities.createdAt, beforeActivity[0]!.createdAt),
297 jsonArrayCondition,
298 ...whereConditions,
299 ];
300 activitiesResult = await this.db
301 .select()
302 .from(feedActivities)
303 .where(and(...conditions))
304 .orderBy(desc(feedActivities.createdAt), desc(feedActivities.id))
305 .limit(limit);
306 } else {
307 // If beforeActivityId doesn't exist, return empty result
308 activitiesResult = [];
309 }
310 } else {
311 // Regular pagination without cursor
312 const conditions = [jsonArrayCondition, ...whereConditions];
313 activitiesResult = await this.db
314 .select()
315 .from(feedActivities)
316 .where(conditions.length > 1 ? and(...conditions) : conditions[0])
317 .orderBy(desc(feedActivities.createdAt), desc(feedActivities.id))
318 .limit(limit)
319 .offset(offset);
320 }
321
322 // Get total count with same filter
323 const conditions = [jsonArrayCondition];
324 if (options.urlType) {
325 conditions.push(eq(feedActivities.urlType, options.urlType));
326 }
327
328 const totalCountResult = await this.db
329 .select({ count: count() })
330 .from(feedActivities)
331 .where(conditions.length > 1 ? and(...conditions) : conditions[0]);
332
333 const totalCount = totalCountResult[0]?.count || 0;
334
335 // Map to domain objects
336 const activities: FeedActivity[] = [];
337 for (const activityData of activitiesResult) {
338 const dto: FeedActivityDTO = {
339 id: activityData.id,
340 actorId: activityData.actorId,
341 cardId: activityData.cardId || undefined,
342 connectionId: activityData.connectionId || undefined,
343 type: activityData.type,
344 metadata: activityData.metadata as any,
345 urlType: activityData.urlType || undefined,
346 source: activityData.source || undefined,
347 createdAt: activityData.createdAt,
348 };
349
350 const domainResult = FeedActivityMapper.toDomain(dto);
351 if (domainResult.isErr()) {
352 return err(domainResult.error);
353 }
354
355 activities.push(domainResult.value);
356 }
357
358 // Determine if there are more activities
359 const hasMore = offset + activities.length < totalCount;
360
361 // Set next cursor if there are more activities
362 let nextCursor: ActivityId | undefined;
363 if (hasMore && activities.length > 0) {
364 const lastActivity = activities[activities.length - 1]!;
365 nextCursor = lastActivity.activityId;
366 }
367
368 return ok({
369 activities,
370 totalCount,
371 hasMore,
372 nextCursor,
373 });
374 } catch (error) {
375 return err(error as Error);
376 }
377 }
378
379 async findById(activityId: ActivityId): Promise<Result<FeedActivity | null>> {
380 try {
381 const activityResult = await this.db
382 .select()
383 .from(feedActivities)
384 .where(eq(feedActivities.id, activityId.getStringValue()))
385 .limit(1);
386
387 if (activityResult.length === 0) {
388 return ok(null);
389 }
390
391 const activityData = activityResult[0]!;
392 const dto: FeedActivityDTO = {
393 id: activityData.id,
394 actorId: activityData.actorId,
395 cardId: activityData.cardId || undefined,
396 type: activityData.type,
397 metadata: activityData.metadata as any,
398 urlType: activityData.urlType || undefined,
399 createdAt: activityData.createdAt,
400 };
401
402 const domainResult = FeedActivityMapper.toDomain(dto);
403 if (domainResult.isErr()) {
404 return err(domainResult.error);
405 }
406
407 return ok(domainResult.value);
408 } catch (error) {
409 return err(error as Error);
410 }
411 }
412
413 async findRecentCardCollectedActivity(
414 actorId: CuratorId,
415 cardId: CardId,
416 withinMinutes: number,
417 ): Promise<Result<FeedActivity | null>> {
418 try {
419 const cutoffTime = new Date(Date.now() - withinMinutes * 60 * 1000);
420
421 const result = await this.db
422 .select()
423 .from(feedActivities)
424 .where(
425 and(
426 eq(feedActivities.actorId, actorId.value),
427 eq(feedActivities.cardId, cardId.getStringValue()),
428 eq(feedActivities.type, ActivityTypeEnum.CARD_COLLECTED),
429 gte(feedActivities.createdAt, cutoffTime),
430 ),
431 )
432 .orderBy(desc(feedActivities.createdAt))
433 .limit(1);
434
435 if (result.length === 0) {
436 return ok(null);
437 }
438
439 const activityData = result[0]!;
440 const dto: FeedActivityDTO = {
441 id: activityData.id,
442 actorId: activityData.actorId,
443 cardId: activityData.cardId || undefined,
444 type: activityData.type,
445 metadata: activityData.metadata as any,
446 urlType: activityData.urlType || undefined,
447 createdAt: activityData.createdAt,
448 };
449
450 const domainResult = FeedActivityMapper.toDomain(dto);
451 if (domainResult.isErr()) {
452 return err(domainResult.error);
453 }
454
455 return ok(domainResult.value);
456 } catch (error) {
457 return err(error as Error);
458 }
459 }
460
461 async updateActivity(activity: FeedActivity): Promise<Result<void>> {
462 try {
463 const dto = FeedActivityMapper.toPersistence(activity);
464
465 await this.db
466 .update(feedActivities)
467 .set({
468 metadata: dto.metadata,
469 urlType: dto.urlType,
470 })
471 .where(eq(feedActivities.id, dto.id));
472
473 return ok(undefined);
474 } catch (error) {
475 return err(error as Error);
476 }
477 }
478
479 async fanOutActivityToFollowers(
480 activityId: ActivityId,
481 followerIds: string[],
482 createdAt: Date,
483 ): Promise<Result<void>> {
484 try {
485 if (followerIds.length === 0) {
486 return ok(undefined);
487 }
488
489 const values = followerIds.map((userId) => ({
490 userId: userId,
491 activityId: activityId.getStringValue(),
492 createdAt: createdAt,
493 }));
494
495 await this.db
496 .insert(followingFeedItems)
497 .values(values)
498 .onConflictDoNothing();
499
500 return ok(undefined);
501 } catch (error) {
502 return err(error as Error);
503 }
504 }
505
506 async getFollowingFeed(
507 userId: string,
508 options: FeedQueryOptions,
509 ): Promise<Result<PaginatedFeedResult>> {
510 try {
511 const { page, limit, beforeActivityId } = options;
512 const offset = (page - 1) * limit;
513
514 // Build where conditions
515 const whereConditions = [eq(followingFeedItems.userId, userId)];
516
517 if (options.activityTypes && options.activityTypes.length > 0) {
518 if (options.activityTypes.length === 1) {
519 whereConditions.push(
520 eq(feedActivities.type, options.activityTypes[0]!),
521 );
522 } else {
523 whereConditions.push(
524 inArray(feedActivities.type, options.activityTypes),
525 );
526 }
527 }
528
529 if (options.urlType) {
530 whereConditions.push(eq(feedActivities.urlType, options.urlType));
531 }
532
533 if (options.source) {
534 if (options.source === ActivitySource.SEMBLE) {
535 whereConditions.push(sql`${feedActivities.source} IS NULL`);
536 } else {
537 whereConditions.push(eq(feedActivities.source, options.source));
538 }
539 }
540
541 // Filter out known bots by default (unless explicitly included)
542 if (!options.includeKnownBots && KNOWN_BOT_DIDS.length > 0) {
543 whereConditions.push(
544 notInArray(feedActivities.actorId, KNOWN_BOT_DIDS),
545 );
546 }
547
548 // Cursor-based pagination
549 if (beforeActivityId) {
550 const beforeActivity = await this.db
551 .select({ createdAt: followingFeedItems.createdAt })
552 .from(followingFeedItems)
553 .where(
554 and(
555 eq(followingFeedItems.userId, userId),
556 eq(
557 followingFeedItems.activityId,
558 beforeActivityId.getStringValue(),
559 ),
560 ),
561 )
562 .limit(1);
563
564 if (beforeActivity.length > 0) {
565 whereConditions.push(
566 lt(followingFeedItems.createdAt, beforeActivity[0]!.createdAt),
567 );
568 }
569 }
570
571 // Main query with JOIN
572 const activitiesResult = await this.db
573 .select({
574 id: feedActivities.id,
575 actorId: feedActivities.actorId,
576 cardId: feedActivities.cardId,
577 connectionId: feedActivities.connectionId,
578 type: feedActivities.type,
579 metadata: feedActivities.metadata,
580 urlType: feedActivities.urlType,
581 source: feedActivities.source,
582 createdAt: followingFeedItems.createdAt, // Use denormalized timestamp
583 })
584 .from(followingFeedItems)
585 .innerJoin(
586 feedActivities,
587 eq(feedActivities.id, followingFeedItems.activityId),
588 )
589 .where(and(...whereConditions))
590 .orderBy(
591 desc(followingFeedItems.createdAt),
592 desc(followingFeedItems.activityId),
593 )
594 .limit(limit)
595 .offset(offset);
596
597 // Count total (with same filters)
598 const totalCountResult = await this.db
599 .select({ count: count() })
600 .from(followingFeedItems)
601 .innerJoin(
602 feedActivities,
603 eq(feedActivities.id, followingFeedItems.activityId),
604 )
605 .where(and(...whereConditions));
606
607 const totalCount = totalCountResult[0]?.count || 0;
608
609 // Map to domain objects
610 const activities: FeedActivity[] = [];
611 for (const activityData of activitiesResult) {
612 const dto: FeedActivityDTO = {
613 id: activityData.id,
614 actorId: activityData.actorId,
615 cardId: activityData.cardId || undefined,
616 connectionId: activityData.connectionId || undefined,
617 type: activityData.type,
618 metadata: activityData.metadata as any,
619 urlType: activityData.urlType || undefined,
620 source: activityData.source || undefined,
621 createdAt: activityData.createdAt,
622 };
623
624 const domainResult = FeedActivityMapper.toDomain(dto);
625 if (domainResult.isErr()) {
626 return err(domainResult.error);
627 }
628
629 activities.push(domainResult.value);
630 }
631
632 const hasMore = offset + activities.length < totalCount;
633 const nextCursor =
634 hasMore && activities.length > 0
635 ? activities[activities.length - 1]!.activityId
636 : undefined;
637
638 return ok({
639 activities,
640 totalCount,
641 hasMore,
642 nextCursor,
643 });
644 } catch (error) {
645 return err(error as Error);
646 }
647 }
648}