Something went wrong. Try again.
a tool for shared writing and social publishing
Something went wrong. Try again.
TypeScript
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261import { createClient } from "@supabase/supabase-js";import { Database, Json } from "supabase/database.types";import { IdResolver } from "@atproto/identity";const idResolver = new IdResolver();import { Firehose, MemoryRunner, Event } from "@atproto/sync";import { ids } from "lexicons/api/lexicons";import { PubLeafletDocument, PubLeafletGraphSubscription, PubLeafletPublication, PubLeafletComment,} from "lexicons/api";import { AppBskyEmbedExternal, AppBskyEmbedRecordWithMedia, AppBskyFeedPost, AppBskyRichtextFacet,} from "@atproto/api";import { AtUri } from "@atproto/syntax";import { writeFile, readFile } from "fs/promises";import { createIdentity } from "actions/createIdentity";import { drizzle } from "drizzle-orm/node-postgres";import { inngest } from "app/api/inngest/client";import { Client } from "pg";
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 QUOTE_PARAM = "/l-quote/";async function main() { const runner = new MemoryRunner({}); let firehose = new Firehose({ service: "wss://relay1.us-west.bsky.network", subscriptionReconnectDelay: 3000, excludeAccount: true, excludeIdentity: true, runner, idResolver, filterCollections: [ ids.PubLeafletDocument, ids.PubLeafletPublication, ids.PubLeafletGraphSubscription, ids.PubLeafletComment, // ids.AppBskyActorProfile, "app.bsky.feed.post", ], 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 (evt.handle) await supabase .from("bsky_profiles") .update({ handle: evt.handle }) .eq("did", evt.did); } 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; } let docResult = await supabase.from("documents").upsert({ uri: evt.uri.toString(), data: record.value as Json, }); if (docResult.error) console.log(docResult.error); 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(), }); 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; 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 && error.code === "23503") { console.log("creating identity"); let client = new Client({ connectionString: process.env.DB_URL }); let db = drizzle(client); await createIdentity(db, { atp_did: evt.did }); client.end(); 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.PubLeafletGraphSubscription) { if (evt.event === "create" || evt.event === "update") { let record = PubLeafletGraphSubscription.validateRecord(evt.record); if (!record.success) return; let { error } = await supabase.from("publication_subscriptions").upsert({ uri: evt.uri.toString(), identity: evt.did, publication: record.value.publication, record: record.value as Json, }); if (error && error.code === "23503") { console.log("creating identity"); let client = new Client({ connectionString: process.env.DB_URL }); let db = drizzle(client); await createIdentity(db, { atp_did: evt.did }); client.end(); 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) { // //only listen to updates because we should fetch it for the first time when they subscribe! // if (evt.event === "update") { // await supabaseServerClient // .from("bsky_profiles") // .update({ record: evt.record as Json }) // .eq("did", evt.did); // } // } 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; console.log("FOUND EMBED!!!");
// Now validate the record since we know it contains our quote param let record = AppBskyFeedPost.validateRecord(evt.record); if (!record.success) 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 }, }); } }}