+2 -5

app/api/inngest/client.ts

··· 1 1 import { Inngest } from "inngest"; 2 2 + 2 3 import { EventSchemas } from "inngest"; 3 3 - import { Json } from "supabase/database.types"; 4 4 5 5 export type Events = { 6 6 "feeds/index-follows": { ··· 51 51 documentUris?: string[]; 52 52 }; 53 53 }; 54 54 - "appview/index-document": { 54 54 + "appview/sync-document-metadata": { 55 55 data: { 56 56 document_uri: string; 57 57 - document_data: Json; 58 57 bsky_post_uri?: string; 59 59 - publication: string | null; 60 60 - did: string; 61 58 }; 62 59 }; 63 60 "user/write-records-to-pds": {

+14 -46

app/api/inngest/functions/index_document.ts app/api/inngest/functions/sync_document_metadata.ts

··· 1 1 import { inngest } from "../client"; 2 2 import { supabaseServerClient } from "supabase/serverClient"; 3 3 - import { AtpAgent } from "@atproto/api"; 3 3 + import { AtpAgent, AtUri } from "@atproto/api"; 4 4 import { idResolver } from "app/(home-pages)/reader/idResolver"; 5 5 6 6 // 1m, 2m, 4m, 8m, 16m, 32m, 1h, 2h, 4h, 8h, 8h, 8h (~37h total) 7 7 const SLEEP_INTERVALS = [ 8 8 - "1m", 9 9 - "2m", 10 10 - "4m", 11 11 - "8m", 12 12 - "16m", 13 13 - "32m", 14 14 - "1h", 15 15 - "2h", 16 16 - "4h", 17 17 - "8h", 18 18 - "8h", 19 19 - "8h", 8 8 + "1m", "2m", "4m", "8m", "16m", "32m", "1h", "2h", "4h", "8h", "8h", "8h", 20 9 ]; 21 10 22 22 - export const index_document = inngest.createFunction( 11 11 + export const sync_document_metadata = inngest.createFunction( 23 12 { 24 24 - id: "index_document_v2", 13 13 + id: "sync_document_metadata_v2", 25 14 debounce: { 26 15 key: "event.data.document_uri", 27 16 period: "60s", ··· 29 18 }, 30 19 concurrency: [{ key: "event.data.document_uri", limit: 1 }], 31 20 }, 32 32 - { event: "appview/index-document" }, 21 21 + { event: "appview/sync-document-metadata" }, 33 22 async ({ event, step }) => { 34 34 - const { document_uri, document_data, bsky_post_uri, publication, did } = 35 35 - event.data; 23 23 + const { document_uri, bsky_post_uri } = event.data; 24 24 + 25 25 + const did = new AtUri(document_uri).host; 36 26 37 27 const handleResult = await step.run("resolve-handle", async () => { 38 28 const doc = await idResolver.did.resolve(did); ··· 49 39 }); 50 40 if (!handleResult) return { error: "No Handle" }; 51 41 52 52 - if (handleResult.isBridgy) { 53 53 - return { handle: handleResult.handle, skipped: true }; 54 54 - } 55 55 - 56 56 - await step.run("write-document", async () => { 57 57 - const docResult = await supabaseServerClient 42 42 + await step.run("set-indexed", async () => { 43 43 + return await supabaseServerClient 58 44 .from("documents") 59 59 - .upsert({ 60 60 - uri: document_uri, 61 61 - data: document_data, 62 62 - indexed: true, 63 63 - }); 64 64 - if (docResult.error) console.log(docResult.error); 65 65 - 66 66 - if (publication) { 67 67 - const docInPubResult = await supabaseServerClient 68 68 - .from("documents_in_publications") 69 69 - .upsert({ 70 70 - publication, 71 71 - document: document_uri, 72 72 - }); 73 73 - await supabaseServerClient 74 74 - .from("documents_in_publications") 75 75 - .delete() 76 76 - .neq("publication", publication) 77 77 - .eq("document", document_uri); 78 78 - if (docInPubResult.error) console.log(docInPubResult.error); 79 79 - } 45 45 + .update({ indexed: !handleResult.isBridgy }) 46 46 + .eq("uri", document_uri) 47 47 + .select(); 80 48 }); 81 49 82 82 - if (!bsky_post_uri) { 50 50 + if (!bsky_post_uri || handleResult.isBridgy) { 83 51 return { handle: handleResult.handle }; 84 52 } 85 53

+2 -2

app/api/inngest/route.tsx

··· 13 13 check_oauth_session, 14 14 } from "./functions/cleanup_expired_oauth_sessions"; 15 15 import { write_records_to_pds } from "./functions/write_records_to_pds"; 16 16 - import { index_document } from "./functions/index_document"; 16 16 + import { sync_document_metadata } from "./functions/sync_document_metadata"; 17 17 18 18 export const { GET, POST, PUT } = serve({ 19 19 client: inngest, ··· 29 29 cleanup_expired_oauth_sessions, 30 30 check_oauth_session, 31 31 write_records_to_pds, 32 32 - index_document, 32 32 + sync_document_metadata, 33 33 ], 34 34 });

+55 -24

appview/index.ts

··· 104 104 console.log(record.error); 105 105 return; 106 106 } 107 107 - let publication: string | null = null; 107 107 + let docResult = await supabase.from("documents").upsert({ 108 108 + uri: evt.uri.toString(), 109 109 + data: record.value as Json, 110 110 + }); 111 111 + if (docResult.error) console.log(docResult.error); 112 112 + await inngest.send({ 113 113 + name: "appview/sync-document-metadata", 114 114 + data: { 115 115 + document_uri: evt.uri.toString(), 116 116 + bsky_post_uri: record.value.postRef?.uri, 117 117 + }, 118 118 + }); 108 119 if (record.value.publication) { 109 120 let publicationURI = new AtUri(record.value.publication); 121 121 + 110 122 if (publicationURI.host !== evt.uri.host) { 111 123 console.log("Unauthorized to create post!"); 112 124 return; 113 125 } 114 114 - publication = record.value.publication; 126 126 + let docInPublicationResult = await supabase 127 127 + .from("documents_in_publications") 128 128 + .upsert({ 129 129 + publication: record.value.publication, 130 130 + document: evt.uri.toString(), 131 131 + }); 132 132 + await supabase 133 133 + .from("documents_in_publications") 134 134 + .delete() 135 135 + .neq("publication", record.value.publication) 136 136 + .eq("document", evt.uri.toString()); 137 137 + 138 138 + if (docInPublicationResult.error) 139 139 + console.log(docInPublicationResult.error); 115 140 } 116 116 - await inngest.send({ 117 117 - name: "appview/index-document", 118 118 - data: { 119 119 - document_uri: evt.uri.toString(), 120 120 - document_data: record.value as Json, 121 121 - bsky_post_uri: record.value.postRef?.uri, 122 122 - publication, 123 123 - did: evt.did, 124 124 - }, 125 125 - }); 126 141 } 127 142 if (evt.event === "delete") { 128 143 await supabase.from("documents").delete().eq("uri", evt.uri.toString()); ··· 256 271 console.log(record.error); 257 272 return; 258 273 } 274 274 + let docResult = await supabase.from("documents").upsert({ 275 275 + uri: evt.uri.toString(), 276 276 + data: record.value as Json, 277 277 + }); 278 278 + if (docResult.error) console.log(docResult.error); 279 279 + await inngest.send({ 280 280 + name: "appview/sync-document-metadata", 281 281 + data: { 282 282 + document_uri: evt.uri.toString(), 283 283 + bsky_post_uri: record.value.bskyPostRef?.uri, 284 284 + }, 285 285 + }); 286 286 + 259 287 // site.standard.document uses "site" field to reference the publication 260 288 // For documents in publications, site is an AT-URI (at://did:plc:xxx/site.standard.publication/rkey) 261 289 // For standalone documents, site is an HTTPS URL (https://leaflet.pub/p/did:plc:xxx) 262 290 // Only link to publications table for AT-URI sites 263 263 - let publication: string | null = null; 264 291 if (record.value.site && record.value.site.startsWith("at://")) { 265 292 let siteURI = new AtUri(record.value.site); 293 293 + 266 294 if (siteURI.host !== evt.uri.host) { 267 295 console.log("Unauthorized to create document in site!"); 268 296 return; 269 297 } 270 270 - publication = record.value.site; 298 298 + let docInPublicationResult = await supabase 299 299 + .from("documents_in_publications") 300 300 + .upsert({ 301 301 + publication: record.value.site, 302 302 + document: evt.uri.toString(), 303 303 + }); 304 304 + await supabase 305 305 + .from("documents_in_publications") 306 306 + .delete() 307 307 + .neq("publication", record.value.site) 308 308 + .eq("document", evt.uri.toString()); 309 309 + 310 310 + if (docInPublicationResult.error) 311 311 + console.log(docInPublicationResult.error); 271 312 } 272 272 - await inngest.send({ 273 273 - name: "appview/index-document", 274 274 - data: { 275 275 - document_uri: evt.uri.toString(), 276 276 - document_data: record.value as Json, 277 277 - bsky_post_uri: record.value.bskyPostRef?.uri, 278 278 - publication, 279 279 - did: evt.did, 280 280 - }, 281 281 - }); 282 313 } 283 314 if (evt.event === "delete") { 284 315 await supabase.from("documents").delete().eq("uri", evt.uri.toString());