Something went wrong. Try again.
A couple of Bluesky feeds focused around PDSes
Something went wrong. Try again.
pds-feedgen ingest.ts
4.7 kB · 183 lines
TypeScript
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184import { Client, ok, simpleFetchHandler } from "@atcute/client";import { CompositeDidDocumentResolver, PlcDidDocumentResolver, WebDidDocumentResolver,} from "@atcute/identity-resolver";import type { ActorIdentifier, ResourceUri } from "@atcute/lexicons/syntax";import { JetstreamSubscription } from "@atcute/jetstream";
import type {} from "@atcute/atproto";
import { db } from "./common/db.ts";import type { Author, DID } from "./common/types.ts";
const didResolver = new CompositeDidDocumentResolver({ methods: { plc: new PlcDidDocumentResolver(), web: new WebDidDocumentResolver(), },});
const worker = new Worker(new URL("./ingest/worker.ts", import.meta.url).href, { type: "module",});
const getAuthor = db.prepare("SELECT pds FROM authors WHERE did = ?");
async function getPDS(did: DID, ignoreCache = false) { let pds: string | undefined;
if (!ignoreCache) { const author = getAuthor.get<Author>(did); if (author) pds = author.pds; }
if (!pds) { const resolved = await didResolver.resolve(did); for (const service of resolved.service ?? []) { if ( service.type == "AtprotoPersonalDataServer" && typeof service.serviceEndpoint === "string" ) { worker.postMessage({ op: 4, did, pds: service.serviceEndpoint, pds_base: getPDSBase(service.serviceEndpoint), }); pds = service.serviceEndpoint; } } }
return pds;}
function getPDSBase(pds: string) { const url = new URL(pds); const splitDomain = url.hostname.split("."); return `${splitDomain[splitDomain.length - 2]}.${ splitDomain[splitDomain.length - 1] }`;}
const getCursor = db.prepare("SELECT cursor FROM state WHERE id = 1");
const dbCursor = getCursor.get<{ cursor?: string }>();const cursor = dbCursor ? Number(dbCursor.cursor) : 0;const jetstream = new JetstreamSubscription({ wantedCollections: ["app.bsky.feed.post"], cursor: cursor - 10000000, // back up a bit for seamless playback url: Deno.env.get("JETSTREAM")?.split(",") ?? "wss://jetstream1.us-east.bsky.network/subscribe", onConnectionOpen(e) { console.log( `Listening to the jetstream (${(e.target as WebSocket).url})...`, ); }, onConnectionError(event) { console.error(event.error); worker.postMessage({ op: 3, cursor: jetstream.cursor, }); },});
let count = 0;
Deno.serve({ port: Number(Deno.env.get("PORT")) || 4001 }, async (request) => { const url = new URL(request.url); if (url.pathname === "/refresh") { const did = url.searchParams.get("id"); if (!did) { return new Response("No DID/handle provided", { status: 400, }); } const pds = await backfillUser(did as ActorIdentifier); if (!pds) { return new Response(`Failed to refresh ${did}`, { status: 500, }); } return new Response(pds); } return new Response("Pong!");});
for await (const event of jetstream) { if (event.kind === "commit") { count++; if (count >= 1024) { count = 0; worker.postMessage({ op: 3, cursor: event.time_us, }); }
const atUri: ResourceUri = `at://${event.did}/app.bsky.feed.post/${event.commit.rkey}`; let pds; try { pds = await getPDS(event.did as DID); } catch (e) { console.error(`Failed to get PDS for ${event.did}: ${e}`); continue; }
if (!pds) { console.error(`PDS not found for ${event.did}`); continue; }
if (event.commit.operation === "create") { worker.postMessage({ op: 0, atUri, cid: event.commit.cid, did: event.did, pds, }); } else if (event.commit.operation === "delete") { worker.postMessage({ op: 1, atUri, pds, }); } } else if (event.kind === "identity") { await backfillUser(event.did); }}
async function backfillUser(did: ActorIdentifier) { const cached = getAuthor.get<Author>(did); const pds = await getPDS(did as DID, true).catch((e) => { console.error(`Failed to get PDS for ${did}: ${e}`); }); if (!pds || cached?.pds === pds) return pds; const handler = simpleFetchHandler({ service: pds }); const rpc = new Client({ handler }); try { const { records } = await ok( rpc.get("com.atproto.repo.listRecords", { params: { repo: did, collection: "app.bsky.feed.post", }, }), ); worker.postMessage({ op: 2, records, did: did, pds, }); } catch (e) { console.error(`Failed to backfill posts: ${e}`); } return pds;}