nandi.uk
This repository has no description
1import { eq, and } from 'drizzle-orm';
2import { PostgresJsDatabase } from 'drizzle-orm/postgres-js';
3import {
4 IFirehoseEventDuplicationService,
5 FirehoseEventType,
6} from '../../domain/services/IFirehoseEventDuplicationService';
7import { IAtUriResolutionService } from '../../../cards/domain/services/IAtUriResolutionService';
8import { ATUri } from '../../domain/ATUri';
9import { publishedRecords } from '../../../cards/infrastructure/repositories/schema/publishedRecord.sql';
10import { Result, ok, err } from 'src/shared/core/Result';
11import { EnvironmentConfigService } from 'src/shared/infrastructure/config/EnvironmentConfigService';
12
13export class DrizzleFirehoseEventDuplicationService
14 implements IFirehoseEventDuplicationService
15{
16 constructor(
17 private db: PostgresJsDatabase,
18 private atUriResolver: IAtUriResolutionService,
19 private configService: EnvironmentConfigService,
20 ) {}
21
22 async hasEventBeenProcessed(
23 atUri: string,
24 cid: string | null,
25 operation: FirehoseEventType,
26 ): Promise<Result<boolean>> {
27 try {
28 // For CREATE/UPDATE: check if (uri, cid) exists
29 if (operation === 'create' || operation === 'update') {
30 if (!cid) return ok(false);
31
32 const result = await this.db
33 .select({ id: publishedRecords.id })
34 .from(publishedRecords)
35 .where(
36 and(eq(publishedRecords.uri, atUri), eq(publishedRecords.cid, cid)),
37 )
38 .limit(1);
39
40 return ok(result.length > 0);
41 }
42
43 // For DELETE: use more complex logic
44 return this.hasBeenDeleted(atUri);
45 } catch (error) {
46 return err(error as Error);
47 }
48 }
49
50 async hasBeenDeleted(atUri: string): Promise<Result<boolean>> {
51 try {
52 // 1. Find all publishedRecords with matching URI
53 const records = await this.db
54 .select()
55 .from(publishedRecords)
56 .where(eq(publishedRecords.uri, atUri));
57
58 if (records.length === 0) {
59 return ok(true); // No records = was deleted
60 }
61
62 // 2. Parse AT URI to get collection
63 const atUriResult = ATUri.create(atUri);
64 if (atUriResult.isErr()) {
65 return err(atUriResult.error);
66 }
67
68 const collection = atUriResult.value.collection;
69 const collections = this.configService.getAtProtoCollections();
70
71 // 3. Check if entity still exists based on collection type
72 switch (collection) {
73 case collections.collection:
74 case collections.marginCollection: {
75 const collectionIdResult =
76 await this.atUriResolver.resolveCollectionId(atUri);
77 if (collectionIdResult.isErr()) {
78 return err(collectionIdResult.error);
79 }
80 return ok(collectionIdResult.value === null);
81 }
82 case collections.card:
83 case collections.marginBookmark: {
84 const cardIdResult = await this.atUriResolver.resolveCardId(atUri);
85 if (cardIdResult.isErr()) {
86 return err(cardIdResult.error);
87 }
88 return ok(cardIdResult.value === null);
89 }
90 case collections.collectionLink:
91 case collections.marginCollectionItem: {
92 const linkInfoResult =
93 await this.atUriResolver.resolveCollectionLinkId(atUri);
94 if (linkInfoResult.isErr()) {
95 return err(linkInfoResult.error);
96 }
97 return ok(linkInfoResult.value === null);
98 }
99 case collections.collectionLinkRemoval: {
100 // For removal records, we track them in published_records table
101 // If a record exists, it hasn't been deleted
102 return ok(records.length === 0);
103 }
104 case collections.follow: {
105 const followInfoResult =
106 await this.atUriResolver.resolveFollowId(atUri);
107 if (followInfoResult.isErr()) {
108 return err(followInfoResult.error);
109 }
110 return ok(followInfoResult.value === null);
111 }
112 default:
113 return err(new Error(`Unknown collection type: ${collection}`));
114 }
115 } catch (error) {
116 return err(error as Error);
117 }
118 }
119}