From 281099be2e88eeb0ebe1097ae69049a914482f2c Mon Sep 17 00:00:00 2001 From: Jared Pereira Date: Mon, 22 Sep 2025 00:14:53 +0900 Subject: [PATCH] don't persist cursor for now --- appview/index.ts | 381 ++++++++++++++++++++++------------------------- 1 file changed, 182 insertions(+), 199 deletions(-) diff --git a/appview/index.ts b/appview/index.ts index f1397c57..adb8fde2 100644 --- a/appview/index.ts +++ b/appview/index.ts @@ -31,205 +31,7 @@ let supabase = createClient( ); const QUOTE_PARAM = "/l-quote/"; async function main() { - let startCursor; - try { - let file = (await readFile(cursorFile)).toString(); - console.log("START CURSOR: " + file); - startCursor = parseInt(file); - console.log(startCursor); - if (Number.isNaN(startCursor)) startCursor = undefined; - } catch (e) {} - - 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.seq}: ${evt.uri}`); - if (evt.collection === ids.PubLeafletDocument) { - if (evt.event === "create" || evt.event === "update") { - let record = PubLeafletDocument.validateRecord(evt.record); - if (!record.success) { - return; - } - await supabase.from("documents").upsert({ - uri: evt.uri.toString(), - data: record.value as Json, - }); - let publicationURI = new AtUri(record.value.publication); - - if (publicationURI.host !== evt.uri.host) { - console.log("Unauthorized to create post!"); - return; - } - await supabase.from("documents_in_publications").insert({ - publication: record.value.publication, - document: evt.uri.toString(), - }); - } - 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 }, - }); - } - } - } - - const runner = new MemoryRunner({ - startCursor, - setCursor: async (cursor) => { - // persist cursor - await writeFile(cursorFile, cursor.toString(), { flush: true }); - }, - }); + const runner = new MemoryRunner({}); let firehose = new Firehose({ service: "wss://relay1.us-west.bsky.network", subscriptionReconnectDelay: 3000, @@ -267,3 +69,184 @@ async function main() { } 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.seq}: ${evt.uri}`); + if (evt.collection === ids.PubLeafletDocument) { + if (evt.event === "create" || evt.event === "update") { + let record = PubLeafletDocument.validateRecord(evt.record); + if (!record.success) { + return; + } + await supabase.from("documents").upsert({ + uri: evt.uri.toString(), + data: record.value as Json, + }); + let publicationURI = new AtUri(record.value.publication); + + if (publicationURI.host !== evt.uri.host) { + console.log("Unauthorized to create post!"); + return; + } + await supabase.from("documents_in_publications").insert({ + publication: record.value.publication, + document: evt.uri.toString(), + }); + } + 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 }, + }); + } + } +} -- 2.54.0