+5 -2

app/api/inngest/client.ts

··· 1 1 import { Inngest } from "inngest"; 2 2 - 3 2 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/sync-document-metadata": { 54 54 + "appview/index-document": { 55 55 data: { 56 56 document_uri: string; 57 57 + document_data: Json; 57 58 bsky_post_uri?: string; 59 59 + publication: string | null; 60 60 + did: string; 58 61 }; 59 62 }; 60 63 "user/write-records-to-pds": {

+46 -14

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

··· 1 1 import { inngest } from "../client"; 2 2 import { supabaseServerClient } from "supabase/serverClient"; 3 3 - import { AtpAgent, AtUri } from "@atproto/api"; 3 3 + import { AtpAgent } 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", "2m", "4m", "8m", "16m", "32m", "1h", "2h", "4h", "8h", "8h", "8h", 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", 9 20 ]; 10 21 11 11 - export const sync_document_metadata = inngest.createFunction( 22 22 + export const index_document = inngest.createFunction( 12 23 { 13 13 - id: "sync_document_metadata_v2", 24 24 + id: "index_document_v2", 14 25 debounce: { 15 26 key: "event.data.document_uri", 16 27 period: "60s", ··· 18 29 }, 19 30 concurrency: [{ key: "event.data.document_uri", limit: 1 }], 20 31 }, 21 21 - { event: "appview/sync-document-metadata" }, 32 32 + { event: "appview/index-document" }, 22 33 async ({ event, step }) => { 23 23 - const { document_uri, bsky_post_uri } = event.data; 24 24 - 25 25 - const did = new AtUri(document_uri).host; 34 34 + const { document_uri, document_data, bsky_post_uri, publication, did } = 35 35 + event.data; 26 36 27 37 const handleResult = await step.run("resolve-handle", async () => { 28 38 const doc = await idResolver.did.resolve(did); ··· 39 49 }); 40 50 if (!handleResult) return { error: "No Handle" }; 41 51 42 42 - await step.run("set-indexed", async () => { 43 43 - return await supabaseServerClient 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 44 58 .from("documents") 45 45 - .update({ indexed: !handleResult.isBridgy }) 46 46 - .eq("uri", document_uri) 47 47 - .select(); 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 + } 48 80 }); 49 81 50 50 - if (!bsky_post_uri || handleResult.isBridgy) { 82 82 + if (!bsky_post_uri) { 51 83 return { handle: handleResult.handle }; 52 84 } 53 85

+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 { sync_document_metadata } from "./functions/sync_document_metadata"; 16 16 + import { index_document } from "./functions/index_document"; 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 - sync_document_metadata, 32 32 + index_document, 33 33 ], 34 34 });

+24 -55

appview/index.ts

··· 104 104 console.log(record.error); 105 105 return; 106 106 } 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 - }); 107 107 + let publication: string | null = null; 119 108 if (record.value.publication) { 120 109 let publicationURI = new AtUri(record.value.publication); 121 121 - 122 110 if (publicationURI.host !== evt.uri.host) { 123 111 console.log("Unauthorized to create post!"); 124 112 return; 125 113 } 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); 114 114 + publication = record.value.publication; 140 115 } 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 + }); 141 126 } 142 127 if (evt.event === "delete") { 143 128 await supabase.from("documents").delete().eq("uri", evt.uri.toString()); ··· 271 256 console.log(record.error); 272 257 return; 273 258 } 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 - 287 259 // site.standard.document uses "site" field to reference the publication 288 260 // For documents in publications, site is an AT-URI (at://did:plc:xxx/site.standard.publication/rkey) 289 261 // For standalone documents, site is an HTTPS URL (https://leaflet.pub/p/did:plc:xxx) 290 262 // Only link to publications table for AT-URI sites 263 263 + let publication: string | null = null; 291 264 if (record.value.site && record.value.site.startsWith("at://")) { 292 265 let siteURI = new AtUri(record.value.site); 293 293 - 294 266 if (siteURI.host !== evt.uri.host) { 295 267 console.log("Unauthorized to create document in site!"); 296 268 return; 297 269 } 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); 270 270 + publication = record.value.site; 312 271 } 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 + }); 313 282 } 314 283 if (evt.event === "delete") { 315 284 await supabase.from("documents").delete().eq("uri", evt.uri.toString());