nandi.uk
This repository has no description
18 kB
577 lines
1import { eq, desc, lt, count, sql, and, gte } from 'drizzle-orm';
2import { PostgresJsDatabase } from 'drizzle-orm/postgres-js';
3import {
4 IFeedRepository,
5 FeedQueryOptions,
6 PaginatedFeedResult,
7} from '../../domain/IFeedRepository';
8import { FeedActivity } from '../../domain/FeedActivity';
9import { ActivityId } from '../../domain/value-objects/ActivityId';
10import { feedActivities } from './schema/feedActivity.sql';
11import { followingFeedItems } from './schema/followingFeedItem.sql';
12import {
13 FeedActivityMapper,
14 FeedActivityDTO,
15} from './mappers/FeedActivityMapper';
16import { Result, ok, err } from '../../../../shared/core/Result';
17import { CollectionId } from '../../../cards/domain/value-objects/CollectionId';
18import { CuratorId } from '../../../cards/domain/value-objects/CuratorId';
19import { CardId } from '../../../cards/domain/value-objects/CardId';
20import { ActivityTypeEnum } from '../../domain/value-objects/ActivityType';
21import { ActivitySource } from '@semble/types';
22
23export class DrizzleFeedRepository implements IFeedRepository {
24 constructor(private db: PostgresJsDatabase) {}
25
26 async addActivity(activity: FeedActivity): Promise<Result<void>> {
27 try {
28 const dto = FeedActivityMapper.toPersistence(activity);
29
30 await this.db.insert(feedActivities).values({
31 id: dto.id,
32 actorId: dto.actorId,
33 cardId: dto.cardId,
34 type: dto.type,
35 metadata: dto.metadata,
36 urlType: dto.urlType,
37 source: dto.source,
38 createdAt: dto.createdAt,
39 });
40
41 return ok(undefined);
42 } catch (error) {
43 return err(error as Error);
44 }
45 }
46
47 async getGlobalFeed(
48 options: FeedQueryOptions,
49 ): Promise<Result<PaginatedFeedResult>> {
50 try {
51 const { page, limit, beforeActivityId } = options;
52 const offset = (page - 1) * limit;
53
54 // Build query conditionally
55 let activitiesResult: Array<{
56 id: string;
57 actorId: string;
58 cardId: string | null;
59 type: string;
60 metadata: any;
61 urlType: string | null;
62 source: string | null;
63 createdAt: Date;
64 }>;
65
66 // Build where conditions
67 const whereConditions = [];
68 if (options.urlType) {
69 whereConditions.push(eq(feedActivities.urlType, options.urlType));
70 }
71 if (options.source) {
72 if (options.source === ActivitySource.SEMBLE) {
73 // Semble content has source IS NULL
74 whereConditions.push(sql`${feedActivities.source} IS NULL`);
75 } else {
76 // Direct match for other sources (e.g., ActivitySource.MARGIN)
77 whereConditions.push(eq(feedActivities.source, options.source));
78 }
79 }
80
81 if (beforeActivityId) {
82 // Get the timestamp of the beforeActivityId
83 const beforeActivity = await this.db
84 .select({ createdAt: feedActivities.createdAt })
85 .from(feedActivities)
86 .where(eq(feedActivities.id, beforeActivityId.getStringValue()))
87 .limit(1);
88
89 if (beforeActivity.length > 0) {
90 const conditions = [
91 lt(feedActivities.createdAt, beforeActivity[0]!.createdAt),
92 ...whereConditions,
93 ];
94 activitiesResult = await this.db
95 .select()
96 .from(feedActivities)
97 .where(conditions.length > 1 ? and(...conditions) : conditions[0])
98 .orderBy(desc(feedActivities.createdAt), desc(feedActivities.id))
99 .limit(limit);
100 } else {
101 // If beforeActivityId doesn't exist, return empty result
102 activitiesResult = [];
103 }
104 } else {
105 // Regular pagination without cursor
106 const query = this.db.select().from(feedActivities);
107
108 if (whereConditions.length > 0) {
109 query.where(
110 whereConditions.length > 1
111 ? and(...whereConditions)
112 : whereConditions[0],
113 );
114 }
115
116 activitiesResult = await query
117 .orderBy(desc(feedActivities.createdAt), desc(feedActivities.id))
118 .limit(limit)
119 .offset(offset);
120 }
121
122 // Get total count with same filters
123 const countQuery = this.db
124 .select({ count: count() })
125 .from(feedActivities);
126
127 if (whereConditions.length > 0) {
128 countQuery.where(
129 whereConditions.length > 1
130 ? and(...whereConditions)
131 : whereConditions[0],
132 );
133 }
134
135 const totalCountResult = await countQuery;
136
137 const totalCount = totalCountResult[0]?.count || 0;
138
139 // Map to domain objects
140 const activities: FeedActivity[] = [];
141 for (const activityData of activitiesResult) {
142 const dto: FeedActivityDTO = {
143 id: activityData.id,
144 actorId: activityData.actorId,
145 cardId: activityData.cardId || undefined,
146 type: activityData.type,
147 metadata: activityData.metadata as any,
148 urlType: activityData.urlType || undefined,
149 source: activityData.source || undefined,
150 createdAt: activityData.createdAt,
151 };
152
153 const domainResult = FeedActivityMapper.toDomain(dto);
154 if (domainResult.isErr()) {
155 return err(domainResult.error);
156 }
157
158 activities.push(domainResult.value);
159 }
160
161 // Determine if there are more activities
162 const hasMore = offset + activities.length < totalCount;
163
164 // Set next cursor if there are more activities
165 let nextCursor: ActivityId | undefined;
166 if (hasMore && activities.length > 0) {
167 const lastActivity = activities[activities.length - 1]!;
168 nextCursor = lastActivity.activityId;
169 }
170
171 return ok({
172 activities,
173 totalCount,
174 hasMore,
175 nextCursor,
176 });
177 } catch (error) {
178 return err(error as Error);
179 }
180 }
181
182 async getGemsFeed(
183 collectionIds: CollectionId[],
184 options: FeedQueryOptions,
185 ): Promise<Result<PaginatedFeedResult>> {
186 try {
187 const { page, limit, beforeActivityId } = options;
188 const offset = (page - 1) * limit;
189 const collectionIdStrings = collectionIds.map((id) =>
190 id.getStringValue(),
191 );
192
193 // Handle empty collection IDs array
194 if (collectionIdStrings.length === 0) {
195 return ok({
196 activities: [],
197 totalCount: 0,
198 hasMore: false,
199 nextCursor: undefined,
200 });
201 }
202
203 // Build query conditionally
204 let activitiesResult: Array<{
205 id: string;
206 actorId: string;
207 cardId: string | null;
208 type: string;
209 metadata: any;
210 urlType: string | null;
211 source: string | null;
212 createdAt: Date;
213 }>;
214
215 // Build where conditions for gems feed
216 const whereConditions = [];
217 if (options.urlType) {
218 whereConditions.push(eq(feedActivities.urlType, options.urlType));
219 }
220 if (options.source) {
221 if (options.source === ActivitySource.SEMBLE) {
222 // Semble content has source IS NULL
223 whereConditions.push(sql`${feedActivities.source} IS NULL`);
224 } else {
225 // Direct match for other sources (e.g., ActivitySource.MARGIN)
226 whereConditions.push(eq(feedActivities.source, options.source));
227 }
228 }
229
230 // Create the JSON array condition using jsonb_array_elements_text
231 const arrayLiteral = `{${collectionIdStrings.map((id) => `"${id}"`).join(',')}}`;
232 const jsonArrayCondition = sql`EXISTS (
233 SELECT 1 FROM jsonb_array_elements_text(${feedActivities.metadata}->'collectionIds') AS collection_id
234 WHERE collection_id = ANY(${arrayLiteral}::text[])
235 )`;
236
237 if (beforeActivityId) {
238 // Get the timestamp of the beforeActivityId
239 const beforeActivity = await this.db
240 .select({ createdAt: feedActivities.createdAt })
241 .from(feedActivities)
242 .where(eq(feedActivities.id, beforeActivityId.getStringValue()))
243 .limit(1);
244
245 if (beforeActivity.length > 0) {
246 const conditions = [
247 lt(feedActivities.createdAt, beforeActivity[0]!.createdAt),
248 jsonArrayCondition,
249 ...whereConditions,
250 ];
251 activitiesResult = await this.db
252 .select()
253 .from(feedActivities)
254 .where(and(...conditions))
255 .orderBy(desc(feedActivities.createdAt), desc(feedActivities.id))
256 .limit(limit);
257 } else {
258 // If beforeActivityId doesn't exist, return empty result
259 activitiesResult = [];
260 }
261 } else {
262 // Regular pagination without cursor
263 const conditions = [jsonArrayCondition, ...whereConditions];
264 activitiesResult = await this.db
265 .select()
266 .from(feedActivities)
267 .where(conditions.length > 1 ? and(...conditions) : conditions[0])
268 .orderBy(desc(feedActivities.createdAt), desc(feedActivities.id))
269 .limit(limit)
270 .offset(offset);
271 }
272
273 // Get total count with same filter
274 const conditions = [jsonArrayCondition];
275 if (options.urlType) {
276 conditions.push(eq(feedActivities.urlType, options.urlType));
277 }
278
279 const totalCountResult = await this.db
280 .select({ count: count() })
281 .from(feedActivities)
282 .where(conditions.length > 1 ? and(...conditions) : conditions[0]);
283
284 const totalCount = totalCountResult[0]?.count || 0;
285
286 // Map to domain objects
287 const activities: FeedActivity[] = [];
288 for (const activityData of activitiesResult) {
289 const dto: FeedActivityDTO = {
290 id: activityData.id,
291 actorId: activityData.actorId,
292 cardId: activityData.cardId || undefined,
293 type: activityData.type,
294 metadata: activityData.metadata as any,
295 urlType: activityData.urlType || undefined,
296 source: activityData.source || undefined,
297 createdAt: activityData.createdAt,
298 };
299
300 const domainResult = FeedActivityMapper.toDomain(dto);
301 if (domainResult.isErr()) {
302 return err(domainResult.error);
303 }
304
305 activities.push(domainResult.value);
306 }
307
308 // Determine if there are more activities
309 const hasMore = offset + activities.length < totalCount;
310
311 // Set next cursor if there are more activities
312 let nextCursor: ActivityId | undefined;
313 if (hasMore && activities.length > 0) {
314 const lastActivity = activities[activities.length - 1]!;
315 nextCursor = lastActivity.activityId;
316 }
317
318 return ok({
319 activities,
320 totalCount,
321 hasMore,
322 nextCursor,
323 });
324 } catch (error) {
325 return err(error as Error);
326 }
327 }
328
329 async findById(activityId: ActivityId): Promise<Result<FeedActivity | null>> {
330 try {
331 const activityResult = await this.db
332 .select()
333 .from(feedActivities)
334 .where(eq(feedActivities.id, activityId.getStringValue()))
335 .limit(1);
336
337 if (activityResult.length === 0) {
338 return ok(null);
339 }
340
341 const activityData = activityResult[0]!;
342 const dto: FeedActivityDTO = {
343 id: activityData.id,
344 actorId: activityData.actorId,
345 cardId: activityData.cardId || undefined,
346 type: activityData.type,
347 metadata: activityData.metadata as any,
348 urlType: activityData.urlType || undefined,
349 createdAt: activityData.createdAt,
350 };
351
352 const domainResult = FeedActivityMapper.toDomain(dto);
353 if (domainResult.isErr()) {
354 return err(domainResult.error);
355 }
356
357 return ok(domainResult.value);
358 } catch (error) {
359 return err(error as Error);
360 }
361 }
362
363 async findRecentCardCollectedActivity(
364 actorId: CuratorId,
365 cardId: CardId,
366 withinMinutes: number,
367 ): Promise<Result<FeedActivity | null>> {
368 try {
369 const cutoffTime = new Date(Date.now() - withinMinutes * 60 * 1000);
370
371 const result = await this.db
372 .select()
373 .from(feedActivities)
374 .where(
375 and(
376 eq(feedActivities.actorId, actorId.value),
377 eq(feedActivities.cardId, cardId.getStringValue()),
378 eq(feedActivities.type, ActivityTypeEnum.CARD_COLLECTED),
379 gte(feedActivities.createdAt, cutoffTime),
380 ),
381 )
382 .orderBy(desc(feedActivities.createdAt))
383 .limit(1);
384
385 if (result.length === 0) {
386 return ok(null);
387 }
388
389 const activityData = result[0]!;
390 const dto: FeedActivityDTO = {
391 id: activityData.id,
392 actorId: activityData.actorId,
393 cardId: activityData.cardId || undefined,
394 type: activityData.type,
395 metadata: activityData.metadata as any,
396 urlType: activityData.urlType || undefined,
397 createdAt: activityData.createdAt,
398 };
399
400 const domainResult = FeedActivityMapper.toDomain(dto);
401 if (domainResult.isErr()) {
402 return err(domainResult.error);
403 }
404
405 return ok(domainResult.value);
406 } catch (error) {
407 return err(error as Error);
408 }
409 }
410
411 async updateActivity(activity: FeedActivity): Promise<Result<void>> {
412 try {
413 const dto = FeedActivityMapper.toPersistence(activity);
414
415 await this.db
416 .update(feedActivities)
417 .set({
418 metadata: dto.metadata,
419 urlType: dto.urlType,
420 })
421 .where(eq(feedActivities.id, dto.id));
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 });
573 } catch (error) {
574 return err(error as Error);
575 }
576 }
577}