nandi.uk
This repository has no description
1import { Result, ok, err } from '../../../../shared/core/Result';
2import { UseCase } from '../../../../shared/core/UseCase';
3import { UseCaseError } from '../../../../shared/core/UseCaseError';
4import { AppError } from '../../../../shared/core/AppError';
5import { ISyncStatusRepository } from '../../domain/repositories/ISyncStatusRepository';
6import { DID } from '../../../user/domain/value-objects/DID';
7import { SyncStatus } from '../../domain/SyncStatus';
8import { ATPROTO_NSID } from '../../../../shared/constants/atproto';
9import { IAtProtoRepoService } from '../../../atproto/application/IAtProtoRepoService';
10import { ProcessMarginBookmarkFirehoseEventUseCase } from '../../../atproto/application/useCases/ProcessMarginBookmarkFirehoseEventUseCase';
11import { ProcessMarginCollectionFirehoseEventUseCase } from '../../../atproto/application/useCases/ProcessMarginCollectionFirehoseEventUseCase';
12import { ProcessMarginCollectionItemFirehoseEventUseCase } from '../../../atproto/application/useCases/ProcessMarginCollectionItemFirehoseEventUseCase';
13import {
14 Environment,
15 EnvironmentConfigService,
16} from 'src/shared/infrastructure/config/EnvironmentConfigService';
17import { IAtUriResolutionService } from '../../../cards/domain/services/IAtUriResolutionService';
18
19export interface SyncAccountDataDTO {
20 curatorId: string;
21 cardId: string;
22}
23
24export class ValidationError extends UseCaseError {
25 constructor(message: string) {
26 super(message);
27 }
28}
29
30export class SyncAccountDataUseCase
31 implements
32 UseCase<
33 SyncAccountDataDTO,
34 Result<void, ValidationError | AppError.UnexpectedError>
35 >
36{
37 constructor(
38 private syncStatusRepo: ISyncStatusRepository,
39 private atProtoRepoService: IAtProtoRepoService,
40 private atUriResolutionService: IAtUriResolutionService,
41 private processMarginBookmarkUseCase: ProcessMarginBookmarkFirehoseEventUseCase,
42 private processMarginCollectionUseCase: ProcessMarginCollectionFirehoseEventUseCase,
43 private processMarginCollectionItemUseCase: ProcessMarginCollectionItemFirehoseEventUseCase,
44 ) {}
45
46 async execute(
47 request: SyncAccountDataDTO,
48 ): Promise<Result<void, ValidationError | AppError.UnexpectedError>> {
49 let syncStatus: SyncStatus | undefined;
50
51 try {
52 const envConfig = new EnvironmentConfigService();
53
54 console.log(
55 `[SYNC] SyncAccountDataUseCase triggered for curator: ${request.curatorId}, card: ${request.cardId}`,
56 );
57
58 // Create DID value object
59 const didResult = DID.create(request.curatorId);
60 if (didResult.isErr()) {
61 return err(new ValidationError(didResult.error.message));
62 }
63 const curatorId = didResult.value;
64
65 // Check if user has already been synced or is currently syncing
66 // Use SELECT FOR UPDATE to prevent concurrent sync attempts
67 const syncStatusResult =
68 await this.syncStatusRepo.findAndLockByCuratorId(curatorId);
69 if (syncStatusResult.isErr()) {
70 console.error(
71 '[SYNC] Error fetching sync status:',
72 syncStatusResult.error,
73 );
74 return err(AppError.UnexpectedError.create(syncStatusResult.error));
75 }
76
77 const existingSyncStatus = syncStatusResult.value;
78
79 if (existingSyncStatus && existingSyncStatus.syncState.isCompleted()) {
80 console.log(
81 `[SYNC] User ${request.curatorId} has already been synced, skipping`,
82 );
83 return ok(undefined);
84 }
85
86 // Defense in depth: check if another process is already syncing
87 if (existingSyncStatus && existingSyncStatus.syncState.isInProgress()) {
88 console.log(
89 `[SYNC] User ${request.curatorId} sync is already in progress, skipping`,
90 );
91 return ok(undefined);
92 }
93
94 // only listen for test account in local env
95 if (
96 envConfig.get().environment === Environment.LOCAL &&
97 curatorId.toString() !== process.env.BSKY_DID
98 ) {
99 return ok(undefined);
100 }
101
102 if (existingSyncStatus) {
103 // Create or update sync status to mark in progress
104 syncStatus = existingSyncStatus;
105 syncStatus.markSyncInProgress();
106 } else {
107 const newSyncStatusResult = SyncStatus.createNew(curatorId);
108 if (newSyncStatusResult.isErr()) {
109 return err(
110 AppError.UnexpectedError.create(newSyncStatusResult.error),
111 );
112 }
113 syncStatus = newSyncStatusResult.value;
114 syncStatus.markSyncInProgress();
115 }
116
117 // Save sync status
118 const saveResult = await this.syncStatusRepo.save(syncStatus);
119 if (saveResult.isErr()) {
120 console.error('[SYNC] Error saving sync status:', saveResult.error);
121 return err(AppError.UnexpectedError.create(saveResult.error));
122 }
123
124 console.log(
125 `[SYNC] Starting sync process for user ${request.curatorId}...`,
126 );
127
128 // Perform the sync
129 const recordsProcessed = await this.performSync(request.curatorId);
130
131 // Mark sync as completed
132 syncStatus.markSyncCompleted(recordsProcessed);
133 await this.syncStatusRepo.save(syncStatus);
134
135 console.log(
136 `[SYNC] Successfully completed sync for ${request.curatorId}: ${recordsProcessed} records processed`,
137 );
138
139 return ok(undefined);
140 } catch (error) {
141 // Mark sync as failed if we have a syncStatus object
142 if (syncStatus) {
143 syncStatus.markSyncFailed(
144 error instanceof Error ? error.message : String(error),
145 );
146 await this.syncStatusRepo.save(syncStatus);
147 }
148
149 console.error('[SYNC] Error in SyncAccountDataUseCase:', error);
150 return err(AppError.UnexpectedError.create(error));
151 }
152 }
153
154 /**
155 * Perform the actual sync process for a user
156 */
157 private async performSync(curatorId: string): Promise<number> {
158 console.log(`[SYNC] Starting sync for curator: ${curatorId}`);
159
160 let totalProcessed = 0;
161
162 // 1. Sync bookmarks first (creates URL cards)
163 console.log(`[SYNC] Syncing bookmarks for ${curatorId}...`);
164 const bookmarksProcessed = await this.syncCollection(
165 curatorId,
166 ATPROTO_NSID.MARGIN.BOOKMARK,
167 this.processMarginBookmarkUseCase,
168 );
169 totalProcessed += bookmarksProcessed;
170 console.log(`[SYNC] Processed ${bookmarksProcessed} bookmarks`);
171
172 // 2. Sync collections (creates collection entities)
173 console.log(`[SYNC] Syncing collections for ${curatorId}...`);
174 const collectionsProcessed = await this.syncCollection(
175 curatorId,
176 ATPROTO_NSID.MARGIN.COLLECTION,
177 this.processMarginCollectionUseCase,
178 );
179 totalProcessed += collectionsProcessed;
180 console.log(`[SYNC] Processed ${collectionsProcessed} collections`);
181
182 // 3. Sync collection items (links cards to collections)
183 // NOTE: Process sequentially to prevent race conditions when multiple items
184 // are added to the same collection concurrently
185 console.log(`[SYNC] Syncing collection items for ${curatorId}...`);
186 const itemsProcessed = await this.syncCollection(
187 curatorId,
188 ATPROTO_NSID.MARGIN.COLLECTION_ITEM,
189 this.processMarginCollectionItemUseCase,
190 true, // sequential = true
191 );
192 totalProcessed += itemsProcessed;
193 console.log(`[SYNC] Processed ${itemsProcessed} collection items`);
194
195 return totalProcessed;
196 }
197
198 /**
199 * Sync a specific collection type with batched concurrent processing
200 * @param sequential - If true, process records sequentially instead of concurrently (prevents race conditions)
201 */
202 private async syncCollection(
203 curatorId: string,
204 collectionType: string,
205 useCase:
206 | ProcessMarginBookmarkFirehoseEventUseCase
207 | ProcessMarginCollectionFirehoseEventUseCase
208 | ProcessMarginCollectionItemFirehoseEventUseCase,
209 sequential: boolean = false,
210 ): Promise<number> {
211 let totalProcessed = 0;
212 let pageCount = 0;
213 const CONCURRENT_BATCH_SIZE = 10; // Process 10 records at a time
214
215 // Use the async generator to fetch all records with automatic pagination
216 for await (const recordsResult of this.atProtoRepoService.listAllRecords(
217 curatorId,
218 collectionType,
219 100, // Fetch 100 records per page
220 )) {
221 // Handle error from service
222 if (recordsResult.isErr()) {
223 console.error(
224 `[SYNC] Error fetching ${collectionType} records: ${recordsResult.error.message}`,
225 );
226 break;
227 }
228
229 const records = recordsResult.value;
230 pageCount++;
231
232 console.log(
233 `[SYNC] Retrieved ${records.length} records from ${collectionType} (page ${pageCount})`,
234 );
235
236 // Process in batches of CONCURRENT_BATCH_SIZE
237 for (let i = 0; i < records.length; i += CONCURRENT_BATCH_SIZE) {
238 const batch = records.slice(i, i + CONCURRENT_BATCH_SIZE);
239 const batchNumber = Math.floor(i / CONCURRENT_BATCH_SIZE) + 1;
240
241 console.log(
242 `[SYNC] Processing batch ${batchNumber} (${batch.length} records) from page ${pageCount}${sequential ? ' [SEQUENTIAL]' : ''}...`,
243 );
244
245 let results: PromiseSettledResult<boolean>[];
246
247 if (sequential) {
248 // Process records one at a time to prevent race conditions
249 results = [];
250 for (const record of batch) {
251 try {
252 const result = await this.processRecord(
253 record,
254 useCase,
255 curatorId,
256 collectionType,
257 );
258 results.push({ status: 'fulfilled', value: result });
259 } catch (error) {
260 results.push({ status: 'rejected', reason: error });
261 }
262 }
263 } else {
264 // Process all records in this batch concurrently
265 results = await Promise.allSettled(
266 batch.map((record) =>
267 this.processRecord(record, useCase, curatorId, collectionType),
268 ),
269 );
270 }
271
272 // Count successful processes
273 const successCount = results.filter(
274 (result) => result.status === 'fulfilled' && result.value === true,
275 ).length;
276
277 totalProcessed += successCount;
278
279 console.log(
280 `[SYNC] Batch ${batchNumber}: processed ${successCount}/${batch.length} records`,
281 );
282 }
283 }
284
285 return totalProcessed;
286 }
287
288 /**
289 * Process a single record from the AT Protocol repository
290 */
291 private async processRecord(
292 record: { uri: string; cid: string; value: any },
293 useCase:
294 | ProcessMarginBookmarkFirehoseEventUseCase
295 | ProcessMarginCollectionFirehoseEventUseCase
296 | ProcessMarginCollectionItemFirehoseEventUseCase,
297 curatorId: string,
298 collectionType: string,
299 ): Promise<boolean> {
300 try {
301 // First, check if this AT URI already exists in our system
302 const resolutionResult = await this.atUriResolutionService.resolveAtUri(
303 record.uri,
304 );
305
306 if (resolutionResult.isErr()) {
307 console.warn(
308 `[SYNC] Error resolving AT URI ${record.uri}: ${resolutionResult.error.message}`,
309 );
310 // Continue processing despite resolution error
311 } else if (resolutionResult.value !== null) {
312 // Resource already exists, skip processing
313 console.log(
314 `[SYNC] Skipping ${collectionType} record ${record.uri} - already exists (type: ${resolutionResult.value.type})`,
315 );
316 return true; // Count as successful (already processed)
317 }
318
319 // Build DTO matching firehose event structure
320 const dto = {
321 atUri: record.uri,
322 cid: record.cid,
323 eventType: 'create' as const,
324 record: record.value,
325 };
326
327 console.log(`[SYNC] Processing ${collectionType} record: ${record.uri}`);
328
329 const result = await useCase.execute(dto);
330
331 if (result.isErr()) {
332 console.warn(
333 `[SYNC] Failed to process record ${record.uri}: ${result.error.message}`,
334 );
335 return false;
336 }
337
338 return true;
339 } catch (error) {
340 console.error(`[SYNC] Error processing record ${record.uri}:`, error);
341 return false;
342 }
343 }
344}