Something went wrong. Try again.
leaflet fork for copy edit / bug fixes forked from leaflet.pub/leaflet
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596import { createClient } from "@supabase/supabase-js";import { Database, Json } from "supabase/database.types";import { IdResolver } from "@atproto/identity";import Client from "ioredis";const idResolver = new IdResolver();import { Firehose, MemoryRunner, Event } from "@atproto/sync";import { ids } from "lexicons/api/lexicons";import { PubLeafletContent, PubLeafletDocument, PubLeafletGraphSubscription, PubLeafletPublication, PubLeafletComment, PubLeafletPollVote, PubLeafletPollDefinition, PubLeafletInteractionsRecommend, SiteStandardDocument, SiteStandardPublication, SiteStandardGraphSubscription, SiteStandardGraphRecommend,} from "lexicons/api";import { AppBskyEmbedExternal, AppBskyEmbedRecordWithMedia, AppBskyFeedPost, AppBskyRichtextFacet,} from "@atproto/api";import { AtUri } from "@atproto/syntax";import { writeFile, readFile } from "fs/promises";import { inngest } from "app/api/inngest/client";import { stripThemeWithoutType } from "src/utils/stripThemeWithoutType";import { pageHasMembersDelimiter } from "src/membership";
const cursorFile = process.env.CURSOR_FILE || "/cursor/cursor";
let supabase = createClient<Database>( process.env.NEXT_PUBLIC_SUPABASE_API_URL as string, process.env.SUPABASE_SERVICE_ROLE_KEY as string,);
const redisClient: Client | null = process.env.REDIS_URL ? new Client(process.env.REDIS_URL) : null;
class RedisProfileCache { constructor(private client: Client) {} async clearEntry(did: string): Promise<void> { await this.client.del(`bsky-profile:${did}`); }}
const profileCache: RedisProfileCache | null = redisClient ? new RedisProfileCache(redisClient) : null;
const QUOTE_PARAM = "/l-quote/";
// The PDS copy of a members-only post is truncated at the delimiter (see// truncateDocumentRecordForPDS), so for posts published through Leaflet the// firehose record must not clobber the full-content copy publishToPublication// wrote to `documents`. Third-party gated posts (no leaflet draft) still// index from the firehose — their PDS record is all the content there is.// Besides the leaflet-draft link we also keep any stored copy that has blocks// below the delimiter: on a first publish the firehose event can beat the// leaflets_in_publications insert, and the full copy must win that race.async function isGatedLeafletManagedDoc( uri: string, isPublicationDoc: boolean, firstPage: unknown,): Promise<boolean> { if (!isPublicationDoc || !pageHasMembersDelimiter(firstPage)) return false; const [{ data: managed }, { data: existing }] = await Promise.all([ supabase .from("leaflets_in_publications") .select("leaflet") .eq("doc", uri) .limit(1), supabase.from("documents").select("data").eq("uri", uri).maybeSingle(), ]); if (managed?.length) return true; return recordHasContentBelowDelimiter(existing?.data);}
function recordHasContentBelowDelimiter(data: unknown): boolean { const record = data as { $type?: string; pages?: unknown[]; content?: { pages?: unknown[] }; } | null; const pages = record?.$type === "site.standard.document" ? record.content?.pages : record?.pages; const blocks = (pages?.[0] as { blocks?: { block?: { $type?: string } }[] }) ?.blocks; if (!Array.isArray(blocks)) return false; const idx = blocks.findIndex( (b) => b?.block?.$type === ids.PubLeafletBlocksMembersOnlyDelimiter, ); return idx !== -1 && idx < blocks.length - 1;}
async function main() { const runner = new MemoryRunner({}); let firehose = new Firehose({ service: "wss://relay1.us-west.bsky.network", subscriptionReconnectDelay: 3000, excludeAccount: true, excludeIdentity: false, runner, idResolver, filterCollections: [ ids.PubLeafletDocument, ids.PubLeafletPublication, ids.PubLeafletGraphSubscription, ids.PubLeafletComment, ids.PubLeafletPollVote, ids.PubLeafletPollDefinition, ids.PubLeafletInteractionsRecommend, ids.AppBskyActorProfile, "app.bsky.feed.post", ids.SiteStandardDocument, ids.SiteStandardPublication, ids.SiteStandardGraphSubscription, ids.SiteStandardGraphRecommend, "parts.page.mention.service", "parts.page.mention.config", ], handleEvent, onError: (err) => { console.error(err); }, }); console.log("starting firehose consumer"); firehose.start(); let cleaningUp = false; const cleanup = async () => { if (cleaningUp) return; cleaningUp = true; console.log("shutting down firehose..."); await firehose.destroy(); await runner.destroy(); process.exit(); };
process.on("SIGINT", cleanup); process.on("SIGTERM", cleanup);}
main();
async function handleEvent(evt: Event) { if (evt.event === "identity") { if (profileCache) { try { await profileCache.clearEntry(evt.did); } catch (err) { console.error("Failed to clear profile cache for", evt.did, err); } } } if ( evt.event == "account" || evt.event === "identity" || evt.event === "sync" ) return; if (evt.collection !== "app.bsky.feed.post") console.log( `${evt.event} in ${evt.collection} ${evt.uri}: ${evt.seq} ${evt.time}`, ); if (evt.collection === ids.PubLeafletDocument) { if (evt.event === "create" || evt.event === "update") { let record = PubLeafletDocument.validateRecord(evt.record); if (!record.success) { console.log(record.error); return; } if ( !(await isGatedLeafletManagedDoc( evt.uri.toString(), !!record.value.publication, record.value.pages?.[0], )) ) { let docResult = await supabase.from("documents").upsert({ uri: evt.uri.toString(), data: record.value as Json, }); if (docResult.error) console.log(docResult.error); } await inngest.send({ name: "appview/sync-document-metadata", data: { document_uri: evt.uri.toString(), bsky_post_uri: record.value.postRef?.uri, event_type: evt.event, }, }); if (record.value.publication) { let publicationURI = new AtUri(record.value.publication);
if (publicationURI.host !== evt.uri.host) { console.log("Unauthorized to create post!"); return; } let docInPublicationResult = await supabase .from("documents_in_publications") .upsert({ publication: record.value.publication, document: evt.uri.toString(), members_only: pageHasMembersDelimiter(record.value.pages?.[0]), }); await supabase .from("documents_in_publications") .delete() .neq("publication", record.value.publication) .eq("document", evt.uri.toString());
if (docInPublicationResult.error) console.log(docInPublicationResult.error); } } if (evt.event === "delete") { await supabase.from("documents").delete().eq("uri", evt.uri.toString()); } } if (evt.collection === ids.PubLeafletPublication) { if (evt.event === "create" || evt.event === "update") { let record = PubLeafletPublication.validateRecord(evt.record); if (!record.success) return; await supabase .from("identities") .upsert({ atp_did: evt.did }, { onConflict: "atp_did" }); await supabase.from("publications").upsert({ uri: evt.uri.toString(), identity_did: evt.did, name: record.value.name, record: record.value as Json, }); } if (evt.event === "delete") { await supabase .from("publications") .delete() .eq("uri", evt.uri.toString()); } } if (evt.collection === ids.PubLeafletComment) { if (evt.event === "create" || evt.event === "update") { let record = PubLeafletComment.validateRecord(evt.record); if (!record.success) return; let { error } = await supabase.from("comments_on_documents").upsert({ uri: evt.uri.toString(), profile: evt.did, document: record.value.subject, record: record.value as Json, }); } if (evt.event === "delete") { await supabase .from("comments_on_documents") .delete() .eq("uri", evt.uri.toString()); } } if (evt.collection === ids.PubLeafletPollVote) { if (evt.event === "create" || evt.event === "update") { let record = PubLeafletPollVote.validateRecord(evt.record); if (!record.success) return; let { error } = await supabase.from("atp_poll_votes").upsert({ uri: evt.uri.toString(), voter_did: evt.did, poll_uri: record.value.poll.uri, poll_cid: record.value.poll.cid, record: record.value as Json, }); } if (evt.event === "delete") { await supabase .from("atp_poll_votes") .delete() .eq("uri", evt.uri.toString()); } } if (evt.collection === ids.PubLeafletPollDefinition) { if (evt.event === "create" || evt.event === "update") { let record = PubLeafletPollDefinition.validateRecord(evt.record); if (!record.success) return; let { error } = await supabase.from("atp_poll_records").upsert({ uri: evt.uri.toString(), cid: evt.cid.toString(), record: record.value as Json, }); if (error) console.log("Error upserting poll definition:", error); } if (evt.event === "delete") { await supabase .from("atp_poll_records") .delete() .eq("uri", evt.uri.toString()); } } if (evt.collection === ids.PubLeafletInteractionsRecommend) { if (evt.event === "create" || evt.event === "update") { let record = PubLeafletInteractionsRecommend.validateRecord(evt.record); if (!record.success) return; await supabase .from("identities") .upsert({ atp_did: evt.did }, { onConflict: "atp_did" }); let { error } = await supabase.from("recommends_on_documents").upsert({ uri: evt.uri.toString(), recommender_did: evt.did, document: record.value.subject, record: record.value as Json, }); if (error) console.log("Error upserting recommend:", error); } if (evt.event === "delete") { await supabase .from("recommends_on_documents") .delete() .eq("uri", evt.uri.toString()); } } if (evt.collection === ids.PubLeafletGraphSubscription) { if (evt.event === "create" || evt.event === "update") { let record = PubLeafletGraphSubscription.validateRecord(evt.record); if (!record.success) return; await supabase .from("identities") .upsert({ atp_did: evt.did }, { onConflict: "atp_did" }); await supabase.from("publication_subscriptions").upsert({ uri: evt.uri.toString(), identity: evt.did, publication: record.value.publication, record: record.value as Json, }); } if (evt.event === "delete") { await supabase .from("publication_subscriptions") .delete() .eq("uri", evt.uri.toString()); } } // site.standard.document records go into the main "documents" table // The normalization layer handles reading both pub.leaflet and site.standard formats if (evt.collection === ids.SiteStandardDocument) { if (evt.event === "create" || evt.event === "update") { let record = SiteStandardDocument.validateRecord(evt.record); if (!record.success) { console.log(record.error); return; } // When the record offloads pages to a blob, skip the documents upsert — // the firehose record has `pages: []` and would clobber the fully // inflated copy that publishToPublication wrote optimistically. The // inngest sync_document_metadata function is the writer in that case. const hasBlobPages = PubLeafletContent.isMain(record.value.content) && !!record.value.content.blobPages; if ( !hasBlobPages && !(await isGatedLeafletManagedDoc( evt.uri.toString(), !!record.value.site?.startsWith("at://"), PubLeafletContent.isMain(record.value.content) ? record.value.content.pages?.[0] : undefined, )) ) { let docResult = await supabase.from("documents").upsert({ uri: evt.uri.toString(), data: record.value as Json, }); if (docResult.error) console.log(docResult.error); } await inngest.send({ name: "appview/sync-document-metadata", data: { document_uri: evt.uri.toString(), bsky_post_uri: record.value.bskyPostRef?.uri, event_type: evt.event, }, });
// site.standard.document uses "site" field to reference the publication // For documents in publications, site is an AT-URI (at://did:plc:xxx/site.standard.publication/rkey) // For standalone documents, site is an HTTPS URL (https://leaflet.pub/p/did:plc:xxx) // Only link to publications table for AT-URI sites if (record.value.site && record.value.site.startsWith("at://")) { let siteURI = new AtUri(record.value.site);
if (siteURI.host !== evt.uri.host) { console.log("Unauthorized to create document in site!"); return; } let docInPublicationResult = await supabase .from("documents_in_publications") .upsert({ publication: record.value.site, document: evt.uri.toString(), // With offloaded blob pages the firehose record has `pages: []`, // so we can't tell whether a delimiter exists — leave the flag // publishToPublication wrote untouched. ...(!hasBlobPages && PubLeafletContent.isMain(record.value.content) ? { members_only: pageHasMembersDelimiter( record.value.content.pages?.[0], ), } : {}), }); await supabase .from("documents_in_publications") .delete() .neq("publication", record.value.site) .eq("document", evt.uri.toString());
if (docInPublicationResult.error) console.log(docInPublicationResult.error); } } if (evt.event === "delete") { await supabase.from("documents").delete().eq("uri", evt.uri.toString()); } }
// site.standard.publication records go into the main "publications" table if (evt.collection === ids.SiteStandardPublication) { if (evt.event === "create" || evt.event === "update") { let record = SiteStandardPublication.validateRecord( stripThemeWithoutType(evt.record), ); if (!record.success) return; await supabase .from("identities") .upsert({ atp_did: evt.did }, { onConflict: "atp_did" }); let { error } = await supabase.from("publications").upsert({ uri: evt.uri.toString(), identity_did: evt.did, name: record.value.name, record: record.value as Json, }); if (error) console.log(error); } if (evt.event === "delete") { await supabase .from("publications") .delete() .eq("uri", evt.uri.toString()); } }
// site.standard.graph.recommend records go into the main "recommends_on_documents" table if (evt.collection === ids.SiteStandardGraphRecommend) { if (evt.event === "create" || evt.event === "update") { let record = SiteStandardGraphRecommend.validateRecord(evt.record); if (!record.success) return; await supabase .from("identities") .upsert({ atp_did: evt.did }, { onConflict: "atp_did" }); let { error } = await supabase.from("recommends_on_documents").upsert({ uri: evt.uri.toString(), recommender_did: evt.did, document: record.value.document, record: record.value as Json, }); if (error) console.log("Error upserting recommend:", error); } if (evt.event === "delete") { await supabase .from("recommends_on_documents") .delete() .eq("uri", evt.uri.toString()); } }
// site.standard.graph.subscription records go into the main "publication_subscriptions" table if (evt.collection === ids.SiteStandardGraphSubscription) { if (evt.event === "create" || evt.event === "update") { let record = SiteStandardGraphSubscription.validateRecord(evt.record); if (!record.success) return; await supabase .from("identities") .upsert({ atp_did: evt.did }, { onConflict: "atp_did" }); await supabase.from("publication_subscriptions").upsert({ uri: evt.uri.toString(), identity: evt.did, publication: record.value.publication, record: record.value as Json, }); } if (evt.event === "delete") { await supabase .from("publication_subscriptions") .delete() .eq("uri", evt.uri.toString()); } } if (evt.collection === ids.AppBskyActorProfile) { if (profileCache) { try { await profileCache.clearEntry(evt.did); } catch (err) { console.error("Failed to clear profile cache for", evt.did, err); } } } if (evt.collection === "parts.page.mention.service") { if (evt.event === "create" || evt.event === "update") { let { error } = await supabase.from("mention_services").upsert({ uri: evt.uri.toString(), identity_did: evt.did, record: evt.record as Json, }); if (error) console.log("Error upserting mention service:", error); } if (evt.event === "delete") { await supabase .from("mention_services") .delete() .eq("uri", evt.uri.toString()); } } if (evt.collection === "parts.page.mention.config") { if (evt.event === "create" || evt.event === "update") { let record = evt.record as Record<string, unknown> | undefined; if (!Array.isArray(record?.services)) return; let { error } = await supabase.from("mention_service_configs").upsert({ uri: evt.uri.toString(), identity_did: evt.did, record: evt.record as Json, }); if (error) console.log("Error upserting mention config:", error); } if (evt.event === "delete") { await supabase .from("mention_service_configs") .delete() .eq("uri", evt.uri.toString()); } } if (evt.collection === "app.bsky.feed.post") { if (evt.event !== "create") return;
// Early exit if no embed if ( !evt.record || typeof evt.record !== "object" || !("embed" in evt.record) ) return;
// Check if embed contains our quote param using optional chaining const embedRecord = evt.record as any; const hasQuoteParam = embedRecord.embed?.external?.uri?.includes(QUOTE_PARAM) || embedRecord.embed?.media?.external?.uri?.includes(QUOTE_PARAM);
if (!hasQuoteParam) return;
// Now validate the record since we know it contains our quote param let record = AppBskyFeedPost.validateRecord(evt.record); if (!record.success) { console.log(record.error); return; }
let embed: string | null = null; if ( AppBskyEmbedExternal.isMain(record.value.embed) && record.value.embed.external.uri.includes(QUOTE_PARAM) ) { embed = record.value.embed.external.uri; } if ( AppBskyEmbedRecordWithMedia.isMain(record.value.embed) && AppBskyEmbedExternal.isMain(record.value.embed.media) && record.value.embed.media?.external?.uri.includes(QUOTE_PARAM) ) { embed = record.value.embed.media.external.uri; } if (embed) { console.log( "processing post mention: " + embed + " in " + evt.uri.toString(), ); await inngest.send({ name: "appview/index-bsky-post-mention", data: { post_uri: evt.uri.toString(), document_link: embed }, }); } }}