import { ok } from '@atcute/client'; import type { Did, GenericUri, ResourceUri } from '@atcute/lexicons/syntax'; import { CONSTELLATION_SERVICE, STAR_COLLECTION } from './constants'; import { getRpc } from './client'; import { hydrateBacklinks } from './hydration'; import { type BacklinkRecordRef, type HydratedRecord, type PaginatedResult, sortByCreatedAt, timestampFromTid, toNumberMap, } from './records'; import { paginateStatefulRecords, sortStatefulRecordsByUpdatedAt, statefulRecordMatchesQuery, type StatefulHydratedRecord, type StatefulPageOptions, } from './stateful'; export const getBacklinksPage = async ( subject: GenericUri, source: string, limit: number, cursor?: string, ): Promise<{ records: BacklinkRecordRef[]; cursor?: string; total?: number }> => { const constellation = getRpc(CONSTELLATION_SERVICE); const page = await ok( constellation.get('blue.microcosm.links.getBacklinks', { params: { subject, source, limit, ...(cursor ? { cursor } : {}), }, }), ); return page as { records: BacklinkRecordRef[]; cursor?: string; total?: number }; }; const getBacklinkDidsPage = async ( subject: GenericUri, source: string, limit: number, cursor?: string, ): Promise<{ linking_dids: Did[]; cursor?: string | null; total?: number }> => { const constellation = getRpc(CONSTELLATION_SERVICE); const page = await ok( constellation.get('blue.microcosm.links.getBacklinkDids', { params: { subject, source, limit, ...(cursor ? { cursor } : {}), }, }), ); return page as { linking_dids: Did[]; cursor?: string | null; total?: number }; }; export const getBacklinkDids = async (subject: GenericUri, source: string): Promise => { const dids: Did[] = []; let cursor: string | null = null; do { const page = await getBacklinkDidsPage(subject, source, 100, cursor ?? undefined); dids.push(...page.linking_dids); cursor = page.cursor ?? null; } while (cursor); return dids; }; const findBacklinkRefByDidForSource = async ( subject: GenericUri, source: string, did: Did, ): Promise => { let cursor: string | null = null; do { const page = await getBacklinksPage(subject, source, 100, cursor ?? undefined); const ref = page.records.find((record) => record.did === did && record.collection === STAR_COLLECTION); if (ref) { return ref; } cursor = page.cursor ?? null; } while (cursor); }; export const findBacklinkRefByDid = async ( subject: GenericUri, sources: string[], did: Did, ): Promise => { const refs = await Promise.all(sources.map((source) => findBacklinkRefByDidForSource(subject, source, did))); return refs.find((ref) => ref); }; export const getBacklinks = async (subject: GenericUri, source: string): Promise => { const records: BacklinkRecordRef[] = []; let cursor: string | null = null; const constellation = getRpc(CONSTELLATION_SERVICE); do { const page: { records: BacklinkRecordRef[]; cursor?: string | null } = await ok( constellation.get('blue.microcosm.links.getBacklinks', { params: { subject, source, limit: 100, ...(cursor ? { cursor } : {}), }, }), ); records.push(...page.records); cursor = page.cursor ?? null; } while (cursor); return records; }; export const latestBacklinkRefByRkey = (refs: BacklinkRecordRef[]): BacklinkRecordRef | undefined => [...refs].sort((left, right) => left.rkey.localeCompare(right.rkey)).at(-1); export const getUniqueBacklinkRefs = async ( subject: GenericUri, source: string | string[], ): Promise => Array.from( new Map( ( await Promise.all( (Array.isArray(source) ? source : [source]).map((entry) => getBacklinks(subject, entry)), ) ) .flat() .map((ref) => [`${ref.did}/${ref.collection}/${ref.rkey}`, ref] as const), ).values(), ); export const getLatestBacklinkTimestamp = async ( subject: GenericUri, source: string | string[], ): Promise => { try { return timestampFromTid(latestBacklinkRefByRkey(await getUniqueBacklinkRefs(subject, source))?.rkey); } catch { return 0; } }; const BACKLINK_BATCH_LIMIT = 50; export const listBacklinkedRecordsPage = async ( subject: GenericUri, source: string | string[], options: StatefulPageOptions, getState: (uri: ResourceUri) => Promise, getUpdatedAt?: (record: StatefulHydratedRecord) => Promise, hydrateRefs: (refs: BacklinkRecordRef[]) => Promise>> = hydrateBacklinks, ): Promise>> => { if (getUpdatedAt) { const hydrated = await hydrateRefs(await getUniqueBacklinkRefs(subject, source)); const numbers = toNumberMap(hydrated); const states = await Promise.all(hydrated.map((record) => getState(record.uri))); const items = hydrated.map((record, index) => ({ ...record, number: numbers.get(record.uri) ?? index + 1, state: states[index], })); const ordered = await sortStatefulRecordsByUpdatedAt(items, getUpdatedAt); return paginateStatefulRecords(ordered, options); } const requestedEnd = options.offset + options.limit; const items: Array> = []; const sources = Array.isArray(source) ? source : [source]; const batchLimit = Math.min(BACKLINK_BATCH_LIMIT, Math.max(options.limit + 1, 1)); const cursors = new Map(sources.map((entry) => [entry, undefined as string | undefined])); const exhausted = new Set(); const seenRefs = new Set(); let matchingSeen = 0; let scannedSeen = 0; let total: number | undefined; while (items.length <= options.limit) { const pages = await Promise.all( sources .filter((entry) => !exhausted.has(entry)) .map(async (entry) => ({ source: entry, page: await getBacklinksPage(subject, entry, batchLimit, cursors.get(entry)), })), ); if (pages.length === 0) { return { items, totalCount: matchingSeen, hasNext: false }; } const refs = pages .flatMap(({ source: entry, page }) => { total = sources.length === 1 ? page.total ?? total : undefined; cursors.set(entry, page.cursor); if (!page.cursor) { exhausted.add(entry); } return page.records; }) .filter((ref) => { const key = `${ref.did}/${ref.collection}/${ref.rkey}`; if (seenRefs.has(key)) { return false; } seenRefs.add(key); return true; }); if (refs.length === 0) { if (exhausted.size === sources.length) { return { items, totalCount: matchingSeen, hasNext: false }; } continue; } if (pages.every(({ page }) => page.records.length === 0)) { return { items, totalCount: matchingSeen, hasNext: false }; } const hydrated = await hydrateRefs(refs); hydrated.sort(sortByCreatedAt).reverse(); const states = await Promise.all(hydrated.map((record) => getState(record.uri))); for (let index = 0; index < hydrated.length; index += 1) { const record = hydrated[index]; const state = states[index]; const number = total ? Math.max(total - scannedSeen, 1) : scannedSeen + 1; scannedSeen += 1; if (state !== options.state) { continue; } if (!statefulRecordMatchesQuery(record.value, options.query)) { continue; } if (matchingSeen >= options.offset && matchingSeen < requestedEnd + 1) { items.push({ ...record, number, state }); } matchingSeen += 1; if (matchingSeen > requestedEnd) { return { items: items.slice(0, options.limit), hasNext: true }; } } if (exhausted.size === sources.length) { return { items, totalCount: matchingSeen, hasNext: false }; } } return { items: items.slice(0, options.limit), hasNext: true }; };