[READ ONLY MIRROR] Spark Social AppView Server
github.com/sprksocial/server
atproto
deno
hono
lexicon
actor and indexing (#6)
authored by
Davi Rodrigues
and committed by
GitHub
23b23fb1
fb6f9eac
+1032
-30
+1
services/appview/package.json
+1
services/appview/package.json
+78
-1
services/appview/pnpm-lock.yaml
+78
-1
services/appview/pnpm-lock.yaml
···
72
72
pino:
73
73
specifier: ^9.6.0
74
74
version: 9.6.0
75
+
pino-pretty:
76
+
specifier: ^13.0.0
77
+
version: 13.0.0
75
78
uint8arrays:
76
79
specifier: ^5.1.0
77
80
version: 5.1.0
···
1241
1244
resolution: {integrity: sha512-dOy+3AuW3a2wNbZHIuMZpTcgjGuLU/uBL/ubcZF9OXbDo8ff4O8yVp5Bf0efS8uEoYo5q4Fx7dY9OgQGXgAsQA==}
1242
1245
dev: true
1243
1246
1247
+
/colorette@2.0.20:
1248
+
resolution: {integrity: sha512-IfEDxwoWIjkeXL1eXcDiow4UbKjhLdq6/EuSVR9GMN7KVH3r9gQ83e73hsz1Nd1T3ijd5xv1wcWRYO+D6kCI2w==}
1249
+
dev: false
1250
+
1244
1251
/commander@9.5.0:
1245
1252
resolution: {integrity: sha512-KRs7WVDKg86PWiuAqhDrAQnTXZKraVcCc6vFdL14qrZ/DcWwuRo7VoiYXalXO7S5GKpqYiVEwCbgFDfxNHKJBQ==}
1246
1253
engines: {node: ^12.20.0 || >=14}
···
1284
1291
shebang-command: 2.0.0
1285
1292
which: 2.0.2
1286
1293
dev: true
1294
+
1295
+
/dateformat@4.6.3:
1296
+
resolution: {integrity: sha512-2P0p0pFGzHS5EMnhdxQi7aJN+iMheud0UhG4dlE1DLAlvL8JHjJJTX/CSm4JXwV0Ka5nGk3zC5mcb5bUQUxxMA==}
1297
+
dev: false
1287
1298
1288
1299
/debug@2.6.9:
1289
1300
resolution: {integrity: sha512-bC7ElrdJaJnPbAP+1EotYvqZsb3ecl5wi6Bfi6BJTUcNowp6cvspg0jXznRTKDjm/E7AdgFBVeAPVMNcKGsHMA==}
···
1353
1364
/encodeurl@2.0.0:
1354
1365
resolution: {integrity: sha512-Q0n9HRi4m6JuGIV1eFlmvJB7ZEVxu93IrMyiMsGC0lrMJMWzRgx6WGquyfQgZVb31vhGgXnfmPNNXmxnOkRBrg==}
1355
1366
engines: {node: '>= 0.8'}
1367
+
dev: false
1368
+
1369
+
/end-of-stream@1.4.4:
1370
+
resolution: {integrity: sha512-+uw1inIHVPQoaVuHzRyXd21icM+cnt4CzD5rW+NC1wjOUSTOs+Te7FOv7AhN7vS9x/oIyhLP5PR1H+phQAHu5Q==}
1371
+
dependencies:
1372
+
once: 1.4.0
1356
1373
dev: false
1357
1374
1358
1375
/envalid@8.0.0:
···
1579
1596
- supports-color
1580
1597
dev: false
1581
1598
1599
+
/fast-copy@3.0.2:
1600
+
resolution: {integrity: sha512-dl0O9Vhju8IrcLndv2eU4ldt1ftXMqqfgN4H1cpmGV7P6jeB9FwpN9a2c8DPGE1Ys88rNUJVYDHq73CGAGOPfQ==}
1601
+
dev: false
1602
+
1582
1603
/fast-deep-equal@3.1.3:
1583
1604
resolution: {integrity: sha512-f3qQ9oQy9j2AhBe/H9VC91wLmKBCCU/gDOnKNAYG5hswO7BLKj09Hc5HYNz9cGI++xlpDCIgDaitVs03ATR84Q==}
1584
1605
dev: true
···
1607
1628
engines: {node: '>=6'}
1608
1629
dev: false
1609
1630
1631
+
/fast-safe-stringify@2.1.1:
1632
+
resolution: {integrity: sha512-W+KJc2dmILlPplD/H4K9l9LcAHAfPtP6BY84uVLXQ6Evcz9Lcg33Y2z1IVblT6xdY54PXYVHEv+0Wpq8Io6zkA==}
1633
+
dev: false
1634
+
1610
1635
/fastq@1.19.1:
1611
1636
resolution: {integrity: sha512-GwLTyxkCXjXbxqIhTsMI2Nui8huMPtnxg7krajPJAjnEG/iiOS7i+zCtWGZR9G0NBKbXKh6X9m9UIsYX/N6vvQ==}
1612
1637
dependencies:
···
1757
1782
function-bind: 1.1.2
1758
1783
dev: false
1759
1784
1785
+
/help-me@5.0.0:
1786
+
resolution: {integrity: sha512-7xgomUX6ADmcYzFik0HzAxh/73YlKR9bmFzf51CZwR+b6YtzU2m0u49hQCqV6SvlqIqsaxovfwdvbnsw3b/zpg==}
1787
+
dev: false
1788
+
1760
1789
/hono@4.7.4:
1761
1790
resolution: {integrity: sha512-Pst8FuGqz3L7tFF+u9Pu70eI0xa5S3LPUmrNd5Jm8nTHze9FxLTK9Kaj5g/k4UcwuJSXTP65SyHOPLrffpcAJg==}
1762
1791
engines: {node: '>=16.9.0'}
···
1856
1885
resolution: {integrity: sha512-s+3Al/p9g32Iq+oqXxkW//7jk2Vig6FF1CFqzVXoTUXt2qz89YWbL+OwS17NFYEvxC35n0FKeGO2LGYSxeM2Gg==}
1857
1886
dev: false
1858
1887
1888
+
/joycon@3.1.1:
1889
+
resolution: {integrity: sha512-34wB/Y7MW7bzjKRjUKTa46I2Z7eV62Rkhva+KkopW7Qvv/OSWBqvkSY7vusOPrNuZcUG3tApvdVgNB8POj3SPw==}
1890
+
engines: {node: '>=10'}
1891
+
dev: false
1892
+
1859
1893
/js-yaml@4.1.0:
1860
1894
resolution: {integrity: sha512-wpxZs9NoxZaJESJGIZTyDEaYpl0FKSA+FB9aJiyemKhMwkxQg63h4T1KJgUGHpTqPDNRcmmYLugrRjJlBtWvRA==}
1861
1895
hasBin: true
···
1987
2021
brace-expansion: 2.0.1
1988
2022
dev: true
1989
2023
2024
+
/minimist@1.2.8:
2025
+
resolution: {integrity: sha512-2yyAR8qBkN3YuheJanUpWC5U3bb5osDywNB8RzDVlDwDHbocAJveqqj1u8+SVD7jkWT4yvsHCpWqqWqAxb0zCA==}
2026
+
dev: false
2027
+
1990
2028
/mkdirp@1.0.4:
1991
2029
resolution: {integrity: sha512-vVqVZQyf3WLx2Shd0qJ9xuvqgAyKPLAiqITEtqW0oIUjzo3PePDd6fW9iFz30ef7Ysp/oiWqbhszeGWW2T6Gzw==}
1992
2030
engines: {node: '>=10'}
···
2120
2158
ee-first: 1.1.1
2121
2159
dev: false
2122
2160
2161
+
/once@1.4.0:
2162
+
resolution: {integrity: sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==}
2163
+
dependencies:
2164
+
wrappy: 1.0.2
2165
+
dev: false
2166
+
2123
2167
/optionator@0.9.4:
2124
2168
resolution: {integrity: sha512-6IpQ7mKUxRcZNLIObR0hz7lxsapSSIYNZJwXPGeF0mTVqGKFIXj1DQcMoT22S3ROcLyY/rz0PWaWZ9ayWmad9g==}
2125
2169
engines: {node: '>= 0.8.0'}
···
2241
2285
split2: 4.2.0
2242
2286
dev: false
2243
2287
2288
+
/pino-pretty@13.0.0:
2289
+
resolution: {integrity: sha512-cQBBIVG3YajgoUjo1FdKVRX6t9XPxwB9lcNJVD5GCnNM4Y6T12YYx8c6zEejxQsU0wrg9TwmDulcE9LR7qcJqA==}
2290
+
hasBin: true
2291
+
dependencies:
2292
+
colorette: 2.0.20
2293
+
dateformat: 4.6.3
2294
+
fast-copy: 3.0.2
2295
+
fast-safe-stringify: 2.1.1
2296
+
help-me: 5.0.0
2297
+
joycon: 3.1.1
2298
+
minimist: 1.2.8
2299
+
on-exit-leak-free: 2.1.2
2300
+
pino-abstract-transport: 2.0.0
2301
+
pump: 3.0.2
2302
+
secure-json-parse: 2.7.0
2303
+
sonic-boom: 4.2.0
2304
+
strip-json-comments: 3.1.1
2305
+
dev: false
2306
+
2244
2307
/pino-std-serializers@6.2.2:
2245
2308
resolution: {integrity: sha512-cHjPPsE+vhj/tnhCy/wiMh3M3z3h/j15zHQX+S9GkTBgqJuTuJzYJ4gUyACLhDaJ7kk9ba9iRDmbH2tJU03OiA==}
2246
2309
dev: false
···
2345
2408
resolution: {integrity: sha512-JZd3gMVBAVQkSs6HdNZo9Sdo0LNcQeMNP3CozBJb3JYC/QUYZTnKxP+f8oWRX4rHP5EurWxqAHTSwUCjlNKa1w==}
2346
2409
dependencies:
2347
2410
punycode: 2.3.1
2411
+
dev: false
2412
+
2413
+
/pump@3.0.2:
2414
+
resolution: {integrity: sha512-tUPXtzlGM8FE3P0ZL6DVs/3P58k9nk8/jZeQCurTJylQA8qFYzHFfhBJkuqyE0FifOsQ0uKWekiZ5g8wtr28cw==}
2415
+
dependencies:
2416
+
end-of-stream: 1.4.4
2417
+
once: 1.4.0
2348
2418
dev: false
2349
2419
2350
2420
/punycode@2.3.1:
···
2434
2504
resolution: {integrity: sha512-YZo3K82SD7Riyi0E1EQPojLz7kpepnSQI9IyPbHHg1XXXevb5dJI7tpyN2ADxGcQbHG7vcyRHk0cbwqcQriUtg==}
2435
2505
dev: false
2436
2506
2507
+
/secure-json-parse@2.7.0:
2508
+
resolution: {integrity: sha512-6aU+Rwsezw7VR8/nyvKTx8QpWH9FrcYiXXlqC4z5d5XQBDRqtbfsRjnwGyqbi3gddNtWHuEk9OANUotL26qKUw==}
2509
+
dev: false
2510
+
2437
2511
/semver@7.7.1:
2438
2512
resolution: {integrity: sha512-hlq8tAfn0m/61p4BVRcPzIGr6LKiMwo4VM6dGi6pt4qcRkmNzTcWq6eCEjEh+qXjkMDvPlOFFSGwQjoEa6gyMA==}
2439
2513
engines: {node: '>=10'}
···
2570
2644
/strip-json-comments@3.1.1:
2571
2645
resolution: {integrity: sha512-6fPc+R4ihwqP6N/aIv2f1gMH8lOVtWQHoqC4yK6oSDVVocumAsfCqjkXnqiYMhmMwS/mEHLp7Vehlt3ql6lEig==}
2572
2646
engines: {node: '>=8'}
2573
-
dev: true
2574
2647
2575
2648
/supports-color@7.2.0:
2576
2649
resolution: {integrity: sha512-qpCAvRl9stuOHveKsn7HncJRvv501qIacKzQlO/+Lwxc9+0q2wLyv4Dfvt80/DPn2pqOBsJdDiogXGR9+OvwRw==}
···
2741
2814
resolution: {integrity: sha512-BN22B5eaMMI9UMtjrGd5g5eCYPpCPDUy0FJXbYsaT5zYxjFOckS53SQDE3pWkVoWpHXVb3BrYcEN4Twa55B5cA==}
2742
2815
engines: {node: '>=0.10.0'}
2743
2816
dev: true
2817
+
2818
+
/wrappy@1.0.2:
2819
+
resolution: {integrity: sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==}
2820
+
dev: false
2744
2821
2745
2822
/ws@8.18.1:
2746
2823
resolution: {integrity: sha512-RKW2aJZMXeMxVpnZ6bck+RswznaxmzdULiBr6KY7XkTnW8uvt0iT9H5DkHUChXrc+uurzwa0rVI16n/Xzjdz1w==}
+50
services/appview/src/db.ts
+50
services/appview/src/db.ts
···
404
404
// Ensure compound index on did + cid for blob takedowns
405
405
blobTakedownSchema.index({ did: 1, cid: 1 }, { unique: true })
406
406
407
+
export interface ActorDocument extends Document {
408
+
uri: string
409
+
did: string
410
+
handle?: string
411
+
profile?: ProfileDocument
412
+
profileCid?: string
413
+
profileTakedownRef?: string
414
+
followersCount: number
415
+
followingCount: number
416
+
postsCount: number
417
+
sortedAt?: Date
418
+
indexedAt: string
419
+
takedownRef?: string
420
+
isLabeler: boolean
421
+
allowIncomingChatsFrom?: string
422
+
upstreamStatus?: string
423
+
createdAt?: Date
424
+
priorityNotifications: boolean
425
+
trustedVerifier?: boolean
426
+
labelsDeclaration?: Record<string, any>
427
+
}
428
+
429
+
export const actorSchema = new Schema<ActorDocument>({
430
+
uri: { type: String, required: true, unique: true, index: true },
431
+
did: { type: String, required: true, index: true },
432
+
handle: { type: String, required: false, index: true },
433
+
profile: { type: Schema.Types.ObjectId, ref: 'Profile', required: false },
434
+
profileCid: { type: String, required: false },
435
+
profileTakedownRef: { type: String, required: false },
436
+
followersCount: { type: Number, required: true, default: 0 },
437
+
followingCount: { type: Number, required: true, default: 0 },
438
+
postsCount: { type: Number, required: true, default: 0 },
439
+
sortedAt: { type: Date, required: false },
440
+
indexedAt: { type: String, required: true },
441
+
takedownRef: { type: String, required: false },
442
+
isLabeler: { type: Boolean, required: true, default: false },
443
+
allowIncomingChatsFrom: { type: String, required: false },
444
+
upstreamStatus: { type: String, required: false },
445
+
createdAt: { type: Date, required: false },
446
+
priorityNotifications: { type: Boolean, required: true, default: false },
447
+
trustedVerifier: { type: Boolean, required: false },
448
+
labelsDeclaration: { type: Object, required: false },
449
+
})
450
+
451
+
// Add compound indexes for Actor
452
+
actorSchema.index({ handle: 'text' })
453
+
actorSchema.index({ did: 1 }, { unique: true })
454
+
407
455
export interface DatabaseModels {
408
456
Like: Model<LikeDocument>
409
457
Post: Model<PostDocument>
···
418
466
Takedown: Model<TakedownDocument>
419
467
RepoTakedown: Model<RepoTakedownDocument>
420
468
BlobTakedown: Model<BlobTakedownDocument>
469
+
Actor: Model<ActorDocument>
421
470
}
422
471
423
472
export class Database {
···
453
502
'BlobTakedown',
454
503
blobTakedownSchema,
455
504
),
505
+
Actor: this.connection.model<ActorDocument>('Actor', actorSchema),
456
506
}
457
507
}
458
508
+16
-1
services/appview/src/index.ts
+16
-1
services/appview/src/index.ts
···
25
25
import { createGetRecordRouter } from './routes/com/atproto/repo/getRecord.js'
26
26
import wellKnownRouter from './well-known.js'
27
27
import { TakedownService } from './services/takedown.js'
28
+
import { IndexingService } from './services/indexing.js'
29
+
30
+
// Extend Hono's context variable map to include our services
31
+
declare module 'hono' {
32
+
interface ContextVariableMap {
33
+
serviceDid: string
34
+
didResolver: DidResolver
35
+
takedownService: TakedownService
36
+
indexingService: IndexingService
37
+
}
38
+
}
28
39
29
40
export type AppContext = {
30
41
db: Database
···
33
44
serviceDid: string
34
45
didResolver: DidResolver
35
46
takedownService: TakedownService
47
+
indexingService: IndexingService
36
48
}
37
49
38
50
export class Server {
···
53
65
// Get service DID from environment
54
66
const serviceDid = env.SERVICE_DID
55
67
56
-
// Create takedown service
68
+
// Create services
57
69
const takedownService = new TakedownService(db)
70
+
const indexingService = new IndexingService(db, resolver)
58
71
59
72
const ctx = {
60
73
db,
···
63
76
serviceDid,
64
77
didResolver: baseIdResolver.did,
65
78
takedownService,
79
+
indexingService,
66
80
}
67
81
68
82
const app = new Hono()
···
73
87
c.set('serviceDid', serviceDid)
74
88
c.set('didResolver', baseIdResolver.did)
75
89
c.set('takedownService', takedownService)
90
+
c.set('indexingService', indexingService)
76
91
await next()
77
92
})
78
93
+29
services/appview/src/routes/com/atproto/admin/getAccountInfos.ts
+29
services/appview/src/routes/com/atproto/admin/getAccountInfos.ts
···
1
+
import { Hono } from 'hono'
2
+
import { zValidator } from '@hono/zod-validator'
3
+
import { z } from 'zod'
4
+
import { HTTPException } from 'hono/http-exception'
5
+
import { TakedownService } from '../../../../services/takedown.js'
6
+
import { authMiddleware } from '../../../../auth/middleware.js'
7
+
import { AppContext } from '../../../../index.js'
8
+
import type * as ComAtprotoAdminUpdateSubjectStatus from '../../../../lexicon/types/com/atproto/admin/updateSubjectStatus.js'
9
+
import type * as ComAtprotoAdminDefs from '../../../../lexicon/types/com/atproto/admin/defs.js'
10
+
import type * as ComAtprotoRepoStrongRef from '../../../../lexicon/types/com/atproto/repo/strongRef.js'
11
+
12
+
export const createGetAccountInfosRouter = (ctx: AppContext) => {
13
+
const router = new Hono()
14
+
15
+
// XRPC endpoint for Ozone integration: com.atproto.admin.getAccountInfos
16
+
router.get(
17
+
'/xrpc/com.atproto.admin.getAccountInfos',
18
+
(c, next) => authMiddleware(c, next, true),
19
+
zValidator(
20
+
'json',
21
+
z.object({
22
+
dids: z.array(z.string()),
23
+
}),
24
+
),
25
+
async (c) => {
26
+
27
+
}
28
+
)
29
+
}
+51
-23
services/appview/src/routes/so/sprk/actor/getProfile.ts
+51
-23
services/appview/src/routes/so/sprk/actor/getProfile.ts
···
15
15
'/xrpc/so.sprk.actor.getProfile',
16
16
optionalAuthMiddleware,
17
17
async (c) => {
18
-
const actor = c.req.query('actor')
18
+
const actorParam = c.req.query('actor')
19
19
const viewerDid = c.get('did') as string | undefined
20
20
21
-
if (!actor) {
21
+
if (!actorParam) {
22
22
return c.json({ error: 'Actor not provided' }, 400)
23
23
}
24
24
25
25
let actorDidDoc
26
-
if (isValidHandle(actor)) {
27
-
actorDidDoc = await ctx.resolver.resolveHandleToDidDoc(actor)
26
+
if (isValidHandle(actorParam)) {
27
+
actorDidDoc = await ctx.resolver.resolveHandleToDidDoc(actorParam)
28
28
} else {
29
29
try {
30
-
ensureValidDid(actor)
31
-
actorDidDoc = await ctx.resolver.resolveDidToDidDoc(actor)
30
+
ensureValidDid(actorParam)
31
+
actorDidDoc = await ctx.resolver.resolveDidToDidDoc(actorParam)
32
32
} catch (err) {
33
33
return c.json({ error: 'Invalid actor' }, 400)
34
34
}
···
36
36
37
37
const actorDid = actorDidDoc.did
38
38
39
-
// Get profile data
40
-
const profile = await ctx.db.models.Profile.findOne({
41
-
authorDid: actorDid,
42
-
}).lean()
39
+
const now = new Date().toISOString()
40
+
41
+
await ctx.indexingService.indexHandle(actorDid, now)
42
+
43
+
// First check if actor exists and has profile
44
+
let actorDoc = await ctx.db.models.Actor.findOne({
45
+
did: actorDid,
46
+
})
47
+
.populate('profile')
48
+
.lean()
43
49
44
-
if (!profile) {
50
+
// If no actor or no profile, try indexing
51
+
if (!actorDoc || !actorDoc.profile) {
52
+
try {
53
+
ctx.logger.info({ did: actorDid }, 'No profile found, attempting to index')
54
+
await ctx.indexingService.indexHandle(actorDid, now, true)
55
+
56
+
// Refetch after indexing
57
+
actorDoc = await ctx.db.models.Actor.findOne({
58
+
did: actorDid,
59
+
})
60
+
.populate('profile')
61
+
.lean()
62
+
63
+
ctx.logger.info({
64
+
did: actorDid,
65
+
hasActor: !!actorDoc,
66
+
hasProfile: !!(actorDoc?.profile)
67
+
}, 'State after indexing')
68
+
} catch (error) {
69
+
ctx.logger.error({ error, did: actorDid }, 'Failed to index handle')
70
+
}
71
+
}
72
+
73
+
if (!actorDoc) {
74
+
return c.json({ error: 'Actor not found' }, 404)
75
+
}
76
+
77
+
if (!actorDoc.profile) {
45
78
return c.json({ error: 'Profile not found' }, 404)
46
79
}
47
80
48
-
const profileHandle = await ctx.resolver.resolveDidToHandle(
49
-
profile.authorDid,
50
-
)
81
+
const profile = actorDoc.profile
51
82
52
83
// Get follower count
53
-
const followersCount = await ctx.db.models.Follow.countDocuments({
54
-
subject: actorDid,
55
-
})
84
+
const followersCount = actorDoc.followersCount || 0
56
85
57
86
// Get follows count
58
-
const followsCount = await ctx.db.models.Follow.countDocuments({
59
-
authorDid: actorDid,
60
-
})
87
+
const followsCount = actorDoc.followingCount || 0
61
88
62
89
// Get posts count
63
-
const postsCount = await ctx.db.models.Post.countDocuments({
64
-
authorDid: actorDid,
65
-
})
90
+
const postsCount = actorDoc.postsCount || 0
91
+
92
+
// Use actor's handle if available, otherwise resolve from DID
93
+
const profileHandle = actorDoc.handle || await ctx.resolver.resolveDidToHandle(actorDid)
66
94
67
95
// Build viewer state if a user is authenticated
68
96
const viewer: SoSprkActorDefs.ViewerState = {}
+1
-1
services/appview/src/routes/so/sprk/actor/searchActor.ts
+1
-1
services/appview/src/routes/so/sprk/actor/searchActor.ts
+222
services/appview/src/services/indexing.ts
+222
services/appview/src/services/indexing.ts
···
1
+
import { AtUri } from '@atproto/syntax'
2
+
import { CID } from 'multiformats/cid'
3
+
import { Document } from 'mongoose'
4
+
import { BidirectionalResolver } from '../id-resolver.js'
5
+
import { Database } from '../db.js'
6
+
import { pino } from 'pino'
7
+
import * as Post from './plugins/post.js'
8
+
9
+
const logger = pino({ name: 'indexing-service' })
10
+
11
+
// Generic type for model processors
12
+
type RecordProcessor<T extends Document> = {
13
+
collection: string
14
+
insertRecord: (
15
+
uri: AtUri,
16
+
cid: CID,
17
+
record: unknown,
18
+
timestamp: string,
19
+
opts?: { disableNotifs?: boolean },
20
+
) => Promise<T | null>
21
+
updateRecord: (
22
+
uri: AtUri,
23
+
cid: CID,
24
+
record: unknown,
25
+
timestamp: string,
26
+
) => Promise<T | null>
27
+
deleteRecord: (uri: AtUri, cascading?: boolean) => Promise<void>
28
+
}
29
+
30
+
/**
31
+
* Service to handle indexing of records from the Atproto network
32
+
*/
33
+
export class IndexingService {
34
+
private records: Record<string, RecordProcessor<any>> = {}
35
+
private logger = pino({ name: 'indexing-service' })
36
+
37
+
constructor(
38
+
private db: Database,
39
+
private resolver: BidirectionalResolver,
40
+
) {
41
+
// Register record processors
42
+
this.records.post = Post.makePlugin(db)
43
+
44
+
// Additional plugins would be registered here
45
+
// Example:
46
+
// this.records.like = Like.makePlugin(db)
47
+
// this.records.follow = Follow.makePlugin(db)
48
+
// etc.
49
+
}
50
+
51
+
/**
52
+
* Index a record in the database
53
+
*
54
+
* @param uri The URI of the record
55
+
* @param cid The CID of the record
56
+
* @param obj The record data
57
+
* @param action The action type (create/update)
58
+
* @param timestamp The timestamp of the operation
59
+
* @param opts Optional parameters
60
+
*/
61
+
async indexRecord(
62
+
uri: AtUri,
63
+
cid: CID,
64
+
obj: unknown,
65
+
action: 'create' | 'update',
66
+
timestamp: string,
67
+
opts?: { disableNotifs?: boolean },
68
+
): Promise<void> {
69
+
try {
70
+
const indexer = this.findIndexerForCollection(uri.collection)
71
+
if (!indexer) {
72
+
this.logger.debug({ collection: uri.collection }, 'No indexer found for collection')
73
+
return
74
+
}
75
+
76
+
if (action === 'create') {
77
+
await indexer.insertRecord(uri, cid, obj, timestamp, opts)
78
+
} else {
79
+
await indexer.updateRecord(uri, cid, obj, timestamp)
80
+
}
81
+
} catch (error) {
82
+
this.logger.error(
83
+
{ error, uri: uri.toString(), cid: cid.toString(), action },
84
+
'Error indexing record',
85
+
)
86
+
}
87
+
}
88
+
89
+
/**
90
+
* Delete a record from the database
91
+
*
92
+
* @param uri The URI of the record to delete
93
+
* @param cascading Whether to cascade the deletion to related records
94
+
*/
95
+
async deleteRecord(uri: AtUri, cascading = false): Promise<void> {
96
+
try {
97
+
const indexer = this.findIndexerForCollection(uri.collection)
98
+
if (!indexer) {
99
+
this.logger.debug({ collection: uri.collection }, 'No indexer found for collection')
100
+
return
101
+
}
102
+
103
+
await indexer.deleteRecord(uri, cascading)
104
+
} catch (error) {
105
+
this.logger.error(
106
+
{ error, uri: uri.toString() },
107
+
'Error deleting record',
108
+
)
109
+
}
110
+
}
111
+
112
+
/**
113
+
* Index or update actor handle information
114
+
*
115
+
* @param did The DID of the actor
116
+
* @param timestamp The timestamp of the operation
117
+
* @param force Force reindexing even if recently indexed
118
+
*/
119
+
async indexHandle(did: string, timestamp: string, force = false): Promise<void> {
120
+
try {
121
+
// Find existing actor
122
+
const actor = await this.db.models.Actor.findOne({ did })
123
+
124
+
// Skip if recently indexed and not forced
125
+
if (!force && actor && this.isHandleRecentlyIndexed(actor, timestamp)) {
126
+
return
127
+
}
128
+
129
+
// Resolve DID to handle
130
+
const didDoc = await this.resolver.resolveDidToDidDoc(did)
131
+
132
+
// Verify handle ownership
133
+
let handle: string | undefined = undefined
134
+
if (didDoc.handle) {
135
+
const handleDidDoc = await this.resolver.resolveHandleToDidDoc(didDoc.handle)
136
+
handle = did === handleDidDoc.did ? didDoc.handle.toLowerCase() : undefined
137
+
}
138
+
139
+
// Handle conflict resolution - if another actor has this handle
140
+
if (handle) {
141
+
const actorWithHandle = await this.db.models.Actor.findOne({ handle })
142
+
if (actorWithHandle && actorWithHandle.did !== did) {
143
+
// Clear handle from the other actor
144
+
await this.db.models.Actor.updateOne(
145
+
{ did: actorWithHandle.did },
146
+
{ $set: { handle: null } }
147
+
)
148
+
}
149
+
}
150
+
151
+
const existingProfile = await this.db.models.Profile.findOne({ authorDid: did })
152
+
if (existingProfile) {
153
+
console.log('existingProfile: ', existingProfile)
154
+
}
155
+
156
+
// Update or create actor
157
+
await this.db.models.Actor.updateOne(
158
+
{ did },
159
+
{
160
+
$set: {
161
+
handle,
162
+
indexedAt: timestamp,
163
+
...(existingProfile && existingProfile._id ? {
164
+
profile: existingProfile._id,
165
+
profileCid: existingProfile.cid
166
+
} : {})
167
+
},
168
+
$setOnInsert: {
169
+
uri: `at://${did}/so.sprk.actor.profile`,
170
+
followersCount: 0,
171
+
followingCount: 0,
172
+
postsCount: 0,
173
+
isLabeler: false,
174
+
priorityNotifications: false,
175
+
}
176
+
},
177
+
{ upsert: true }
178
+
)
179
+
180
+
if (existingProfile) {
181
+
this.logger.info({ did, profileId: existingProfile._id }, 'Linked existing profile to actor during indexing')
182
+
}
183
+
} catch (error) {
184
+
this.logger.error({ error, did }, 'Error indexing handle')
185
+
}
186
+
}
187
+
188
+
/**
189
+
* Find the indexer responsible for a collection
190
+
*
191
+
* @param collection The collection to find an indexer for
192
+
* @returns The indexer or undefined if not found
193
+
*/
194
+
findIndexerForCollection(collection: string): RecordProcessor<any> | undefined {
195
+
return Object.values(this.records).find(
196
+
(indexer) => indexer.collection === collection
197
+
)
198
+
}
199
+
200
+
/**
201
+
* Check if an actor's handle was recently indexed
202
+
*
203
+
* @param actor The actor document
204
+
* @param timestamp Current timestamp
205
+
* @returns Whether the actor was recently indexed
206
+
*/
207
+
private isHandleRecentlyIndexed(actor: any, timestamp: string): boolean {
208
+
if (!actor.indexedAt) return false
209
+
210
+
const timeDiff = new Date(timestamp).getTime() - new Date(actor.indexedAt).getTime()
211
+
const ONE_DAY = 24 * 60 * 60 * 1000
212
+
const ONE_HOUR = 60 * 60 * 1000
213
+
214
+
// Reindex daily for all actors
215
+
if (timeDiff > ONE_DAY) return false
216
+
217
+
// Reindex more frequently for actors without handles
218
+
if (actor.handle === null && timeDiff > ONE_HOUR) return false
219
+
220
+
return true
221
+
}
222
+
}
+178
services/appview/src/services/plugins/post.ts
+178
services/appview/src/services/plugins/post.ts
···
1
+
import { AtUri } from '@atproto/syntax'
2
+
import { CID } from 'multiformats/cid'
3
+
import { pino } from 'pino'
4
+
import { Database, PostDocument } from '../../db.js'
5
+
6
+
const logger = pino({ name: 'post-processor' })
7
+
8
+
export type PostRecord = {
9
+
text: string
10
+
createdAt: string
11
+
facets?: Array<unknown>
12
+
reply?: {
13
+
root: { uri: string; cid: string }
14
+
parent: { uri: string; cid: string }
15
+
}
16
+
embed?: unknown
17
+
langs?: string[]
18
+
labels?: unknown
19
+
tags?: string[]
20
+
}
21
+
22
+
export type PostPluginType = {
23
+
collection: string
24
+
insertRecord: (
25
+
uri: AtUri,
26
+
cid: CID,
27
+
record: unknown,
28
+
timestamp: string,
29
+
opts?: { disableNotifs?: boolean },
30
+
) => Promise<PostDocument | null>
31
+
updateRecord: (
32
+
uri: AtUri,
33
+
cid: CID,
34
+
record: unknown,
35
+
timestamp: string,
36
+
) => Promise<PostDocument | null>
37
+
deleteRecord: (uri: AtUri, cascading?: boolean) => Promise<void>
38
+
}
39
+
40
+
/**
41
+
* Create a post processor plugin for the indexing service
42
+
*/
43
+
export function makePlugin(db: Database): PostPluginType {
44
+
return {
45
+
collection: 'so.sprk.feed.post',
46
+
47
+
async insertRecord(
48
+
uri: AtUri,
49
+
cid: CID,
50
+
recordObj: unknown,
51
+
timestamp: string,
52
+
opts = {},
53
+
): Promise<PostDocument | null> {
54
+
const record = recordObj as PostRecord
55
+
if (!record) return null
56
+
57
+
try {
58
+
// Find author information
59
+
const actor = await db.models.Actor.findOne({ did: uri.hostname })
60
+
if (!actor) {
61
+
logger.warn({ did: uri.hostname }, 'Actor not found when indexing post')
62
+
return null
63
+
}
64
+
65
+
// Extract post data
66
+
const postData = {
67
+
uri: uri.toString(),
68
+
cid: cid.toString(),
69
+
text: record.text,
70
+
facets: record.facets || [],
71
+
reply: record.reply || null,
72
+
embed: record.embed || null,
73
+
langs: record.langs || [],
74
+
labels: record.labels || null,
75
+
tags: record.tags || [],
76
+
authorDid: uri.hostname,
77
+
authorHandle: actor.handle || uri.hostname,
78
+
createdAt: record.createdAt,
79
+
indexedAt: timestamp,
80
+
}
81
+
82
+
// Create the post
83
+
const post = await db.models.Post.findOneAndUpdate(
84
+
{ uri: uri.toString() },
85
+
{ $set: postData },
86
+
{ upsert: true, new: true }
87
+
)
88
+
89
+
// Increment post count for actor
90
+
await db.models.Actor.updateOne(
91
+
{ did: uri.hostname },
92
+
{ $inc: { postsCount: 1 } }
93
+
)
94
+
95
+
return post
96
+
} catch (error) {
97
+
logger.error(
98
+
{ error, uri: uri.toString(), cid: cid.toString() },
99
+
'Error inserting post record'
100
+
)
101
+
return null
102
+
}
103
+
},
104
+
105
+
async updateRecord(
106
+
uri: AtUri,
107
+
cid: CID,
108
+
recordObj: unknown,
109
+
timestamp: string,
110
+
): Promise<PostDocument | null> {
111
+
const record = recordObj as PostRecord
112
+
if (!record) return null
113
+
114
+
try {
115
+
// Find author information
116
+
const actor = await db.models.Actor.findOne({ did: uri.hostname })
117
+
if (!actor) {
118
+
logger.warn({ did: uri.hostname }, 'Actor not found when updating post')
119
+
return null
120
+
}
121
+
122
+
// Update post data
123
+
const postData = {
124
+
cid: cid.toString(),
125
+
text: record.text,
126
+
facets: record.facets || [],
127
+
reply: record.reply || null,
128
+
embed: record.embed || null,
129
+
langs: record.langs || [],
130
+
labels: record.labels || null,
131
+
tags: record.tags || [],
132
+
authorHandle: actor.handle || uri.hostname,
133
+
indexedAt: timestamp,
134
+
}
135
+
136
+
// Update the post
137
+
const post = await db.models.Post.findOneAndUpdate(
138
+
{ uri: uri.toString() },
139
+
{ $set: postData },
140
+
{ new: true }
141
+
)
142
+
143
+
return post
144
+
} catch (error) {
145
+
logger.error(
146
+
{ error, uri: uri.toString(), cid: cid.toString() },
147
+
'Error updating post record'
148
+
)
149
+
return null
150
+
}
151
+
},
152
+
153
+
async deleteRecord(uri: AtUri): Promise<void> {
154
+
try {
155
+
const post = await db.models.Post.findOne({ uri: uri.toString() })
156
+
157
+
if (post) {
158
+
// Delete the post
159
+
await db.models.Post.deleteOne({ uri: uri.toString() })
160
+
161
+
// Decrement post count for actor
162
+
await db.models.Actor.updateOne(
163
+
{ did: uri.hostname },
164
+
{ $inc: { postsCount: -1 } }
165
+
)
166
+
167
+
// Delete any associated likes, reposts, etc. as needed
168
+
// This would be the cascading deletion logic
169
+
}
170
+
} catch (error) {
171
+
logger.error(
172
+
{ error, uri: uri.toString() },
173
+
'Error deleting post record'
174
+
)
175
+
}
176
+
},
177
+
}
178
+
}
+2
services/ingester/src/db/connection.ts
+2
services/ingester/src/db/connection.ts
···
11
11
musicSchema,
12
12
lookSchema,
13
13
generatorSchema,
14
+
actorSchema,
14
15
} from './models.js'
15
16
import { env } from '../utils/env.js'
16
17
import { pino } from 'pino'
···
33
34
Music: this.connection.model('Music', musicSchema),
34
35
Look: this.connection.model('Look', lookSchema),
35
36
Generator: this.connection.model('Generator', generatorSchema),
37
+
Actor: this.connection.model('Actor', actorSchema),
36
38
}
37
39
}
38
40
+49
services/ingester/src/db/models.ts
+49
services/ingester/src/db/models.ts
···
342
342
generatorSchema.index({ authorDid: 1, createdAt: -1 })
343
343
generatorSchema.index({ did: 1, createdAt: -1 })
344
344
345
+
export interface ActorDocument extends Document {
346
+
uri: string
347
+
did: string
348
+
handle?: string
349
+
profile?: ProfileDocument
350
+
profileCid?: string
351
+
profileTakedownRef?: string
352
+
followersCount: number
353
+
followingCount: number
354
+
postsCount: number
355
+
sortedAt?: Date
356
+
indexedAt: string
357
+
takedownRef?: string
358
+
isLabeler: boolean
359
+
allowIncomingChatsFrom?: string
360
+
upstreamStatus?: string
361
+
createdAt?: Date
362
+
priorityNotifications: boolean
363
+
trustedVerifier?: boolean
364
+
labelsDeclaration?: Record<string, any>
365
+
}
366
+
367
+
export const actorSchema = new Schema<ActorDocument>({
368
+
uri: { type: String, required: true, unique: true, index: true },
369
+
did: { type: String, required: true, index: true },
370
+
handle: { type: String, required: false, index: true },
371
+
profile: { type: Schema.Types.ObjectId, ref: 'Profile', required: false },
372
+
profileCid: { type: String, required: false },
373
+
profileTakedownRef: { type: String, required: false },
374
+
followersCount: { type: Number, required: true, default: 0 },
375
+
followingCount: { type: Number, required: true, default: 0 },
376
+
postsCount: { type: Number, required: true, default: 0 },
377
+
sortedAt: { type: Date, required: false },
378
+
indexedAt: { type: String, required: true },
379
+
takedownRef: { type: String, required: false },
380
+
isLabeler: { type: Boolean, required: true, default: false },
381
+
allowIncomingChatsFrom: { type: String, required: false },
382
+
upstreamStatus: { type: String, required: false },
383
+
createdAt: { type: Date, required: false },
384
+
priorityNotifications: { type: Boolean, required: true, default: false },
385
+
trustedVerifier: { type: Boolean, required: false },
386
+
labelsDeclaration: { type: Object, required: false },
387
+
})
388
+
389
+
// Add compound indexes for Actor
390
+
actorSchema.index({ handle: 'text' })
391
+
actorSchema.index({ did: 1 }, { unique: true })
392
+
345
393
export interface DatabaseModels {
346
394
Like: Model<LikeDocument>
347
395
Post: Model<PostDocument>
···
353
401
Music: Model<MusicDocument>
354
402
Look: Model<LookDocument>
355
403
Generator: Model<GeneratorDocument>
404
+
Actor: Model<ActorDocument>
356
405
}
+75
services/ingester/src/handlers/actor-handler.ts
+75
services/ingester/src/handlers/actor-handler.ts
···
1
+
import { pino } from 'pino'
2
+
import { Database } from '../db/connection.js'
3
+
import type { NormalizedEvent } from '../types/events.js'
4
+
import { IndexingService } from '../services/indexing.js'
5
+
import { BidirectionalResolver } from '../id-resolver.js'
6
+
7
+
const logger = pino({ name: 'actor-handler' })
8
+
9
+
/**
10
+
* This handler is called by all other handlers to ensure that
11
+
* any DID referenced in an event has a corresponding actor entry.
12
+
*
13
+
* @param evt The normalized event to process
14
+
* @param db Database connection
15
+
*/
16
+
export async function handleActorReferences(evt: NormalizedEvent, db: Database): Promise<void> {
17
+
try {
18
+
const now = new Date().toISOString()
19
+
const resolver = new BidirectionalResolver()
20
+
const indexingService = new IndexingService(db, resolver)
21
+
22
+
// Always ensure the author DID has an actor
23
+
if (evt.did) {
24
+
await indexingService.indexHandle(evt.did, now)
25
+
}
26
+
27
+
// Handle subject DIDs for follow, block, like events
28
+
if (['follow', 'block', 'like'].includes(evt.event) && evt.record?.subject) {
29
+
// Subject is usually a DID in format did:plc:12345
30
+
const subjectDid = evt.record.subject as string
31
+
if (subjectDid && subjectDid.startsWith('did:')) {
32
+
await indexingService.indexHandle(subjectDid, now)
33
+
}
34
+
}
35
+
36
+
// Handle reply references for posts
37
+
if (evt.collection === 'so.sprk.feed.post' && evt.record?.reply) {
38
+
const reply = evt.record.reply as { root?: { uri?: string }, parent?: { uri?: string } }
39
+
40
+
// Extract DIDs from reply URIs (format: at://did:plc:12345/...)
41
+
if (reply.root?.uri) {
42
+
const rootDid = extractDidFromUri(reply.root.uri)
43
+
if (rootDid) {
44
+
await indexingService.indexHandle(rootDid, now)
45
+
}
46
+
}
47
+
48
+
if (reply.parent?.uri) {
49
+
const parentDid = extractDidFromUri(reply.parent.uri)
50
+
if (parentDid && parentDid !== extractDidFromUri(reply.root?.uri || '')) {
51
+
await indexingService.indexHandle(parentDid, now)
52
+
}
53
+
}
54
+
}
55
+
56
+
// Handle repost subjects
57
+
if (evt.collection === 'so.sprk.feed.repost' && evt.record?.subject?.uri) {
58
+
const subjectUri = evt.record.subject.uri as string
59
+
const subjectDid = extractDidFromUri(subjectUri)
60
+
if (subjectDid) {
61
+
await indexingService.indexHandle(subjectDid, now)
62
+
}
63
+
}
64
+
} catch (error) {
65
+
logger.error({ error, uri: evt.uri }, 'Error while handling actor references')
66
+
}
67
+
}
68
+
69
+
/**
70
+
* Helper function to extract DID from a URI
71
+
*/
72
+
function extractDidFromUri(uri: string): string | null {
73
+
const match = uri.match(/at:\/\/(did:[^/]+)/)
74
+
return match ? match[1] : null
75
+
}
+5
-1
services/ingester/src/handlers/index.ts
+5
-1
services/ingester/src/handlers/index.ts
···
11
11
import { handleMusicEvent } from './music-handler.js'
12
12
import { handleLookEvent } from './look-handler.js'
13
13
import { handleGeneratorEvent } from './generator-handler.js'
14
+
import { handleActorReferences } from './actor-handler.js'
14
15
15
16
const logger = pino({ name: 'event-handler' })
16
17
17
18
export async function handleEvent(evt: NormalizedEvent, db: Database): Promise<void> {
18
19
try {
19
-
// Handle different events based on collection
20
+
// First, ensure all actor references are handled properly
21
+
await handleActorReferences(evt, db)
22
+
23
+
// Then handle different events based on collection
20
24
if (evt.collection === 'so.sprk.feed.like') {
21
25
await handleLikeEvent(evt, db)
22
26
return
+14
-3
services/ingester/src/handlers/profile-handler.ts
+14
-3
services/ingester/src/handlers/profile-handler.ts
···
1
1
import { pino } from 'pino'
2
2
import { Database } from '../db/connection.js'
3
3
import type { NormalizedEvent } from '../types/events.js'
4
+
import { ensureActor, linkProfileToActor } from '../utils/actor-utils.js'
5
+
import type { ProfileDocument } from '../db/models.js'
4
6
5
7
const logger = pino({ name: 'profile-handler' })
6
8
···
37
39
}, 'Processing profile event')
38
40
39
41
try {
42
+
// First, ensure we have an actor for this DID
43
+
await ensureActor(evt.did, evt.handle || undefined, db)
44
+
40
45
const profileData = {
41
46
uri: evt.uri,
42
47
displayName: record.displayName,
···
53
58
cid: evt.commit.cid
54
59
}
55
60
56
-
await db.models.Profile.findOneAndUpdate(
61
+
// Save the profile
62
+
const profile = await db.models.Profile.findOneAndUpdate(
57
63
{ uri: evt.uri },
58
64
profileData,
59
65
{ upsert: true, new: true }
60
-
)
66
+
) as ProfileDocument
67
+
68
+
if (profile && profile._id) {
69
+
// Link the profile to the actor
70
+
await linkProfileToActor(evt.did, profile._id.toString(), evt.commit.cid, db)
71
+
}
61
72
62
73
logger.info(
63
74
{ uri: evt.uri },
64
-
'Successfully saved profile to database'
75
+
'Successfully saved profile to database and linked to actor'
65
76
)
66
77
} catch (error) {
67
78
logger.error(
+61
services/ingester/src/id-resolver.ts
+61
services/ingester/src/id-resolver.ts
···
1
+
import { pino } from 'pino'
2
+
3
+
const logger = pino({ name: 'id-resolver' })
4
+
5
+
export interface DidDocument {
6
+
did: string
7
+
handle?: string
8
+
}
9
+
10
+
/**
11
+
* Service to handle resolving DIDs to handles and vice versa
12
+
*/
13
+
export class BidirectionalResolver {
14
+
private logger = pino({ name: 'id-resolver' })
15
+
16
+
/**
17
+
* Resolve a DID to its DID document
18
+
*/
19
+
async resolveDidToDidDoc(did: string): Promise<DidDocument> {
20
+
try {
21
+
// TODO: Implement actual DID resolution
22
+
// For now, return basic document
23
+
return {
24
+
did,
25
+
}
26
+
} catch (error) {
27
+
this.logger.error({ error, did }, 'Failed to resolve DID to DID document')
28
+
throw error
29
+
}
30
+
}
31
+
32
+
/**
33
+
* Resolve a handle to its DID document
34
+
*/
35
+
async resolveHandleToDidDoc(handle: string): Promise<DidDocument> {
36
+
try {
37
+
// TODO: Implement actual handle resolution
38
+
// For now, return basic document
39
+
return {
40
+
did: `did:plc:${handle.toLowerCase()}`,
41
+
handle,
42
+
}
43
+
} catch (error) {
44
+
this.logger.error({ error, handle }, 'Failed to resolve handle to DID document')
45
+
throw error
46
+
}
47
+
}
48
+
49
+
/**
50
+
* Resolve a DID to its handle
51
+
*/
52
+
async resolveDidToHandle(did: string): Promise<string> {
53
+
try {
54
+
const doc = await this.resolveDidToDidDoc(did)
55
+
return doc.handle || did
56
+
} catch (error) {
57
+
this.logger.error({ error, did }, 'Failed to resolve DID to handle')
58
+
throw error
59
+
}
60
+
}
61
+
}
+109
services/ingester/src/services/indexing.ts
+109
services/ingester/src/services/indexing.ts
···
1
+
import { pino } from 'pino'
2
+
import { Database } from '../db/connection.js'
3
+
import { BidirectionalResolver } from '../id-resolver.js'
4
+
5
+
const logger = pino({ name: 'indexing-service' })
6
+
7
+
/**
8
+
* Service to handle indexing of actors and their handles
9
+
*/
10
+
export class IndexingService {
11
+
private logger = pino({ name: 'indexing-service' })
12
+
13
+
constructor(
14
+
private db: Database,
15
+
private resolver: BidirectionalResolver,
16
+
) {}
17
+
18
+
/**
19
+
* Index or update actor handle information
20
+
*
21
+
* @param did The DID of the actor
22
+
* @param timestamp The timestamp of the operation
23
+
* @param force Force reindexing even if recently indexed
24
+
*/
25
+
async indexHandle(did: string, timestamp: string, force = false): Promise<void> {
26
+
try {
27
+
// Find existing actor
28
+
const actor = await this.db.models.Actor.findOne({ did })
29
+
30
+
// Skip if recently indexed and not forced
31
+
if (!force && actor && this.isHandleRecentlyIndexed(actor, timestamp)) {
32
+
return
33
+
}
34
+
35
+
// Resolve DID to handle
36
+
const didDoc = await this.resolver.resolveDidToDidDoc(did)
37
+
38
+
// Verify handle ownership
39
+
let handle: string | undefined = undefined
40
+
if (didDoc.handle) {
41
+
const handleDidDoc = await this.resolver.resolveHandleToDidDoc(didDoc.handle)
42
+
handle = did === handleDidDoc.did ? didDoc.handle.toLowerCase() : undefined
43
+
}
44
+
45
+
// Handle conflict resolution - if another actor has this handle
46
+
if (handle) {
47
+
const actorWithHandle = await this.db.models.Actor.findOne({ handle })
48
+
if (actorWithHandle && actorWithHandle.did !== did) {
49
+
// Clear handle from the other actor
50
+
await this.db.models.Actor.updateOne(
51
+
{ did: actorWithHandle.did },
52
+
{ $set: { handle: null } }
53
+
)
54
+
}
55
+
}
56
+
57
+
const existingProfile = await this.db.models.Profile.findOne({ authorDid: did })
58
+
59
+
// Update or create actor
60
+
await this.db.models.Actor.updateOne(
61
+
{ did },
62
+
{
63
+
$set: {
64
+
handle,
65
+
indexedAt: timestamp,
66
+
...(existingProfile && existingProfile._id ? {
67
+
profile: existingProfile._id,
68
+
profileCid: existingProfile.cid
69
+
} : {})
70
+
},
71
+
$setOnInsert: {
72
+
uri: `at://${did}/so.sprk.actor.profile`,
73
+
followersCount: 0,
74
+
followingCount: 0,
75
+
postsCount: 0,
76
+
isLabeler: false,
77
+
priorityNotifications: false,
78
+
}
79
+
},
80
+
{ upsert: true }
81
+
)
82
+
83
+
if (existingProfile) {
84
+
this.logger.info({ did, profileId: existingProfile._id }, 'Linked existing profile to actor during indexing')
85
+
}
86
+
} catch (error) {
87
+
this.logger.error({ error, did }, 'Error indexing handle')
88
+
}
89
+
}
90
+
91
+
/**
92
+
* Check if an actor's handle was recently indexed
93
+
*/
94
+
private isHandleRecentlyIndexed(actor: any, timestamp: string): boolean {
95
+
if (!actor.indexedAt) return false
96
+
97
+
const timeDiff = new Date(timestamp).getTime() - new Date(actor.indexedAt).getTime()
98
+
const ONE_DAY = 24 * 60 * 60 * 1000
99
+
const ONE_HOUR = 60 * 60 * 1000
100
+
101
+
// Reindex daily for all actors
102
+
if (timeDiff > ONE_DAY) return false
103
+
104
+
// Reindex more frequently for actors without handles
105
+
if (actor.handle === null && timeDiff > ONE_HOUR) return false
106
+
107
+
return true
108
+
}
109
+
}
+91
services/ingester/src/utils/actor-utils.ts
+91
services/ingester/src/utils/actor-utils.ts
···
1
+
import { pino } from 'pino'
2
+
import { Database } from '../db/connection.js'
3
+
4
+
const logger = pino({ name: 'actor-utils' })
5
+
6
+
/**
7
+
* Ensures that an actor exists for the given DID.
8
+
* If the actor doesn't exist, it creates a new one.
9
+
*
10
+
* @param did The DID to ensure has an actor
11
+
* @param handle Optional handle associated with the DID
12
+
* @param db Database connection
13
+
* @returns The actor document, either existing or newly created
14
+
*/
15
+
export async function ensureActor(
16
+
did: string,
17
+
handle?: string,
18
+
db?: Database
19
+
): Promise<any> {
20
+
if (!db) {
21
+
logger.warn({ did }, 'No database connection provided to ensureActor')
22
+
return null
23
+
}
24
+
25
+
try {
26
+
// Try to find existing actor
27
+
const existingActor = await db.models.Actor.findOne({ did })
28
+
29
+
if (existingActor) {
30
+
// If handle is provided and different from existing, update it
31
+
if (handle && existingActor.handle !== handle) {
32
+
existingActor.handle = handle
33
+
await existingActor.save()
34
+
logger.info({ did, handle }, 'Updated actor handle')
35
+
}
36
+
return existingActor
37
+
}
38
+
39
+
// Create new actor if none exists
40
+
const now = new Date()
41
+
const uri = `at://${did}/so.sprk.actor.profile`
42
+
43
+
const newActor = await db.models.Actor.create({
44
+
uri,
45
+
did,
46
+
handle: handle || undefined,
47
+
followersCount: 0,
48
+
followingCount: 0,
49
+
postsCount: 0,
50
+
indexedAt: now.toISOString(),
51
+
isLabeler: false,
52
+
priorityNotifications: false
53
+
})
54
+
55
+
logger.info({ did, handle }, 'Created new actor')
56
+
return newActor
57
+
} catch (error) {
58
+
logger.error({ error, did, handle }, 'Failed to ensure actor exists')
59
+
return null
60
+
}
61
+
}
62
+
63
+
/**
64
+
* Links a profile to an actor
65
+
*
66
+
* @param did The DID of the actor
67
+
* @param profileId The MongoDB ID of the profile
68
+
* @param profileCid The CID of the profile
69
+
* @param db Database connection
70
+
*/
71
+
export async function linkProfileToActor(
72
+
did: string,
73
+
profileId: string,
74
+
profileCid: string,
75
+
db: Database
76
+
): Promise<void> {
77
+
try {
78
+
await db.models.Actor.findOneAndUpdate(
79
+
{ did },
80
+
{
81
+
profile: profileId,
82
+
profileCid
83
+
},
84
+
{ new: true }
85
+
)
86
+
87
+
logger.info({ did, profileId }, 'Linked profile to actor')
88
+
} catch (error) {
89
+
logger.error({ error, did, profileId }, 'Failed to link profile to actor')
90
+
}
91
+
}