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<Did[]> => {
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<BacklinkRecordRef | undefined> => {
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<BacklinkRecordRef | undefined> => {
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<BacklinkRecordRef[]> => {
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<BacklinkRecordRef[]> =>
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<number> => {
try {
return timestampFromTid(latestBacklinkRefByRkey(await getUniqueBacklinkRefs(subject, source))?.rkey);
} catch {
return 0;
}
};
const BACKLINK_BATCH_LIMIT = 50;
export const listBacklinkedRecordsPage = async <T extends { createdAt?: string }, State extends string>(
subject: GenericUri,
source: string | string[],
options: StatefulPageOptions<State>,
getState: (uri: ResourceUri) => Promise<State>,
getUpdatedAt?: (record: StatefulHydratedRecord<T, State>) => Promise<number>,
hydrateRefs: (refs: BacklinkRecordRef[]) => Promise<Array<HydratedRecord<T>>> = hydrateBacklinks<T>,
): Promise<PaginatedResult<StatefulHydratedRecord<T, State>>> => {
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<StatefulHydratedRecord<T, State>> = [];
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<string>();
const seenRefs = new Set<string>();
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 };
};