Skip to content

Commit d35efbc

Browse files
danielroedanielroe
committed
refactor: use @atproto/lex client and paginate listRepos in PDS endpoints
1 parent 9047e49 commit d35efbc

3 files changed

Lines changed: 117 additions & 65 deletions

File tree

‎server/api/atproto/pds-graphs.get.ts‎

Lines changed: 68 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,9 @@
1+
import { Client } from '@atproto/lex'
2+
import * as app from '#shared/types/lexicons/app'
13
import type { AtprotoProfile } from '#shared/types/atproto'
24

35
import {
4-
ONE_THOUSAND_NPMX_USER_ACCOUNTS_XRPC,
5-
BSKY_APP_VIEW_USER_PROFILES_XRPC,
6+
BLUESKY_API,
67
ERROR_PDS_FETCH_FAILED,
78
} from '#shared/utils/constants'
89

@@ -11,11 +12,25 @@ interface GraphLink {
1112
target: string
1213
}
1314

15+
const NPMX_PDS_HOST = 'https://npmx.social'
16+
const LIST_REPOS_LIMIT = 1000
1417
const USER_BATCH_AMOUNT = 25
1518

16-
export default defineCachedEventHandler(
17-
async (): Promise<{ nodes: AtprotoProfile[]; links: GraphLink[] }> => {
18-
const response = await fetch(ONE_THOUSAND_NPMX_USER_ACCOUNTS_XRPC)
19+
const blueskyClient = new Client({ service: BLUESKY_API })
20+
21+
/**
22+
* Paginate through all repos on the npmx PDS via com.atproto.sync.listRepos.
23+
*/
24+
async function fetchAllDids(): Promise<string[]> {
25+
const dids: string[] = []
26+
let cursor: string | undefined
27+
28+
do {
29+
const url = new URL(`${NPMX_PDS_HOST}/xrpc/com.atproto.sync.listRepos`)
30+
url.searchParams.set('limit', String(LIST_REPOS_LIMIT))
31+
if (cursor) url.searchParams.set('cursor', cursor)
32+
33+
const response = await fetch(url.toString())
1934

2035
if (!response.ok) {
2136
throw createError({
@@ -24,61 +39,75 @@ export default defineCachedEventHandler(
2439
})
2540
}
2641

27-
const listRepos = (await response.json()) as { repos: { did: string }[] }
28-
const dids = listRepos.repos.map(repo => repo.did)
42+
const data = (await response.json()) as {
43+
repos: { did: string }[]
44+
cursor?: string
45+
}
46+
47+
dids.push(...data.repos.map(repo => repo.did))
48+
cursor = data.cursor
49+
} while (cursor)
50+
51+
return dids
52+
}
53+
54+
export default defineCachedEventHandler(
55+
async (): Promise<{ nodes: AtprotoProfile[]; links: GraphLink[] }> => {
56+
const dids = await fetchAllDids()
2957
const localDids = new Set(dids)
3058

3159
const nodes: AtprotoProfile[] = []
3260
const links: GraphLink[] = []
3361

62+
// Fetch profiles in batches using the @atproto/lex client
3463
for (let i = 0; i < dids.length; i += USER_BATCH_AMOUNT) {
3564
const batch = dids.slice(i, i + USER_BATCH_AMOUNT)
3665

37-
const url = new URL(BSKY_APP_VIEW_USER_PROFILES_XRPC)
38-
for (const did of batch) {
39-
url.searchParams.append('actors', did)
40-
}
41-
4266
try {
43-
const profilesResponse = await fetch(url.toString())
44-
45-
if (!profilesResponse.ok) {
46-
console.warn(`Failed to fetch atproto profiles: ${profilesResponse.status}`)
47-
continue
48-
}
49-
50-
const profilesData = (await profilesResponse.json()) as { profiles: AtprotoProfile[] }
51-
52-
if (profilesData.profiles) {
53-
nodes.push(...profilesData.profiles)
54-
}
67+
const data = await blueskyClient.call(app.bsky.actor.getProfiles, {
68+
actors: batch,
69+
})
70+
71+
nodes.push(
72+
...data.profiles.map(profile => ({
73+
did: profile.did,
74+
handle: profile.handle,
75+
displayName: profile.displayName,
76+
avatar: profile.avatar,
77+
})),
78+
)
5579
} catch (error) {
5680
console.warn('Failed to fetch atproto profiles:', error)
5781
}
5882
}
5983

84+
// Fetch follow graphs (no lexicon type for getFollows, using raw fetch)
6085
for (const did of dids) {
61-
const followResponse = await fetch(
62-
`https://public.api.bsky.app/xrpc/app.bsky.graph.getFollows?actor=${did}`,
63-
)
86+
try {
87+
const followResponse = await fetch(
88+
`https://public.api.bsky.app/xrpc/app.bsky.graph.getFollows?actor=${did}`,
89+
)
6490

65-
if (!followResponse.ok) {
66-
console.warn(`Failed to fetch follows: ${followResponse.status}`)
67-
continue
68-
}
91+
if (!followResponse.ok) {
92+
console.warn(`Failed to fetch follows: ${followResponse.status}`)
93+
continue
94+
}
6995

70-
const followData = await followResponse.json()
96+
const followData = (await followResponse.json()) as {
97+
follows: { did: string }[]
98+
}
7199

72-
for (const followedUser of followData.follows) {
73-
if (localDids.has(followedUser.did)) {
74-
links.push({ source: did, target: followedUser.did })
100+
for (const followedUser of followData.follows) {
101+
if (localDids.has(followedUser.did)) {
102+
links.push({ source: did, target: followedUser.did })
103+
}
75104
}
105+
} catch (error) {
106+
console.warn('Failed to fetch follows:', error)
76107
}
77108
}
78-
return {
79-
nodes,
80-
links,
81-
}
109+
110+
return { nodes, links }
82111
},
83112
{
84113
maxAge: 3600,

‎server/api/atproto/pds-users.get.ts‎

Lines changed: 49 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -1,16 +1,32 @@
1+
import { Client } from '@atproto/lex'
2+
import * as app from '#shared/types/lexicons/app'
13
import {
2-
ONE_THOUSAND_NPMX_USER_ACCOUNTS_XRPC,
3-
BSKY_APP_VIEW_USER_PROFILES_XRPC,
4+
BLUESKY_API,
45
ERROR_PDS_FETCH_FAILED,
56
} from '#shared/utils/constants'
67
import type { AtprotoProfile } from '#shared/types/atproto'
78

9+
const NPMX_PDS_HOST = 'https://npmx.social'
10+
const LIST_REPOS_LIMIT = 1000
811
const USER_BATCH_AMOUNT = 25
912

10-
export default defineCachedEventHandler(
11-
async (): Promise<AtprotoProfile[]> => {
12-
// INFO: Request npmx.social PDS for every hosted user account
13-
const response = await fetch(ONE_THOUSAND_NPMX_USER_ACCOUNTS_XRPC)
13+
const blueskyClient = new Client({ service: BLUESKY_API })
14+
15+
/**
16+
* Paginate through all repos on the npmx PDS via com.atproto.sync.listRepos.
17+
* The lexicon type isn't generated in this codebase, so we use raw fetch
18+
* with cursor-based pagination to handle >1k accounts.
19+
*/
20+
async function fetchAllDids(): Promise<string[]> {
21+
const dids: string[] = []
22+
let cursor: string | undefined
23+
24+
do {
25+
const url = new URL(`${NPMX_PDS_HOST}/xrpc/com.atproto.sync.listRepos`)
26+
url.searchParams.set('limit', String(LIST_REPOS_LIMIT))
27+
if (cursor) url.searchParams.set('cursor', cursor)
28+
29+
const response = await fetch(url.toString())
1430

1531
if (!response.ok) {
1632
throw createError({
@@ -19,34 +35,46 @@ export default defineCachedEventHandler(
1935
})
2036
}
2137

22-
const listRepos = (await response.json()) as { repos: { did: string }[] }
23-
const dids = listRepos.repos.map(repo => repo.did)
38+
const data = (await response.json()) as {
39+
repos: { did: string }[]
40+
cursor?: string
41+
}
42+
43+
dids.push(...data.repos.map(repo => repo.did))
44+
cursor = data.cursor
45+
} while (cursor)
46+
47+
return dids
48+
}
49+
50+
export default defineCachedEventHandler(
51+
async (): Promise<AtprotoProfile[]> => {
52+
const dids = await fetchAllDids()
2453

25-
// INFO: Request the list of user profiles from the Bluesky AppView
54+
// Fetch profiles in batches of 25 (API limit) using the @atproto/lex client
2655
const batchPromises: Promise<AtprotoProfile[]>[] = []
2756

2857
for (let i = 0; i < dids.length; i += USER_BATCH_AMOUNT) {
2958
const batch = dids.slice(i, i + USER_BATCH_AMOUNT)
30-
const url = new URL(BSKY_APP_VIEW_USER_PROFILES_XRPC)
31-
32-
for (const did of batch) url.searchParams.append('actors', did)
3359

3460
batchPromises.push(
35-
fetch(url.toString())
36-
.then(res => {
37-
if (!res.ok) throw new Error(`Status ${res.status}`)
38-
return res.json() as Promise<{ profiles: AtprotoProfile[] }>
39-
})
40-
.then(data => data.profiles || [])
41-
.catch(err => {
61+
blueskyClient
62+
.call(app.bsky.actor.getProfiles, { actors: batch })
63+
.then(data =>
64+
data.profiles.map(profile => ({
65+
did: profile.did,
66+
handle: profile.handle,
67+
displayName: profile.displayName,
68+
avatar: profile.avatar,
69+
})),
70+
)
71+
.catch((err) => {
4272
console.warn('Failed to fetch batch:', err)
43-
// Return empty array on failure so Promise.all doesn't crash
4473
return []
4574
}),
4675
)
4776
}
4877

49-
// INFO: Await all batches in parallel and flatten the results
5078
return (await Promise.all(batchPromises)).flat()
5179
},
5280
{

‎shared/utils/constants.ts‎

Lines changed: 0 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -15,11 +15,6 @@ export const BLUESKY_API = 'https://public.api.bsky.app'
1515
export const BLUESKY_COMMENTS_REQUEST = '/api/atproto/bluesky-comments'
1616
export const NPM_REGISTRY = 'https://registry.npmjs.org'
1717
export const NPM_API = 'https://api.npmjs.org'
18-
export const ONE_THOUSAND_NPMX_USER_ACCOUNTS_XRPC =
19-
'https://npmx.social/xrpc/com.atproto.sync.listRepos?limit=1000'
20-
export const BSKY_APP_VIEW_USER_PROFILES_XRPC =
21-
'https://public.api.bsky.app/xrpc/app.bsky.actor.getProfiles'
22-
2318
// Error Messages
2419
export const ERROR_PACKAGE_ANALYSIS_FAILED = 'Failed to analyze package.'
2520
export const ERROR_PACKAGE_VERSION_AND_FILE_FAILED = 'Version and file path are required.'

0 commit comments

Comments
 (0)