leaflet.pub
a tool for shared writing and social publishing
awarm.space
1import { inngest } from "../client";
2import { supabaseServerClient } from "supabase/serverClient";
3import { AtpAgent } from "@atproto/api";
4import { idResolver } from "app/(home-pages)/reader/idResolver";
5
6// 1m, 2m, 4m, 8m, 16m, 32m, 1h, 2h, 4h, 8h, 8h, 8h (~37h total)
7const SLEEP_INTERVALS = [
8 "1m",
9 "2m",
10 "4m",
11 "8m",
12 "16m",
13 "32m",
14 "1h",
15 "2h",
16 "4h",
17 "8h",
18 "8h",
19 "8h",
20];
21
22export const index_document = inngest.createFunction(
23 {
24 id: "index_document_v2",
25 debounce: {
26 key: "event.data.document_uri",
27 period: "60s",
28 timeout: "3m",
29 },
30 concurrency: [{ key: "event.data.document_uri", limit: 1 }],
31 },
32 { event: "appview/index-document" },
33 async ({ event, step }) => {
34 const { document_uri, document_data, bsky_post_uri, publication, did } =
35 event.data;
36
37 const handleResult = await step.run("resolve-handle", async () => {
38 const doc = await idResolver.did.resolve(did);
39 const handle = doc?.alsoKnownAs
40 ?.find((a) => a.startsWith("at://"))
41 ?.replace("at://", "");
42 if (!doc) return null;
43 const isBridgy = !!doc?.service?.find(
44 (s) =>
45 typeof s.serviceEndpoint === "string" &&
46 s.serviceEndpoint.includes("atproto.brid.gy"),
47 );
48 return { handle: handle ?? null, isBridgy, doc };
49 });
50 if (!handleResult) return { error: "No Handle" };
51
52 if (handleResult.isBridgy) {
53 return { handle: handleResult.handle, skipped: true };
54 }
55
56 await step.run("write-document", async () => {
57 const docResult = await supabaseServerClient
58 .from("documents")
59 .upsert({
60 uri: document_uri,
61 data: document_data,
62 indexed: true,
63 });
64 if (docResult.error) console.log(docResult.error);
65
66 if (publication) {
67 const docInPubResult = await supabaseServerClient
68 .from("documents_in_publications")
69 .upsert({
70 publication,
71 document: document_uri,
72 });
73 await supabaseServerClient
74 .from("documents_in_publications")
75 .delete()
76 .neq("publication", publication)
77 .eq("document", document_uri);
78 if (docInPubResult.error) console.log(docInPubResult.error);
79 }
80 });
81
82 if (!bsky_post_uri) {
83 return { handle: handleResult.handle };
84 }
85
86 const agent = new AtpAgent({ service: "https://public.api.bsky.app" });
87
88 const fetchAndUpdate = async () => {
89 const res = await agent.app.bsky.feed.getPosts({
90 uris: [bsky_post_uri],
91 });
92 const post = res.data.posts[0];
93 if (!post) return 0;
94 const likeCount = post.likeCount ?? 0;
95 await supabaseServerClient
96 .from("documents")
97 .update({ bsky_like_count: likeCount })
98 .eq("uri", document_uri);
99 return likeCount;
100 };
101
102 let likeCount = await step.run("sync-0", fetchAndUpdate);
103
104 for (let i = 0; i < SLEEP_INTERVALS.length; i++) {
105 await step.sleep(`wait-${i + 1}`, SLEEP_INTERVALS[i]);
106 likeCount = await step.run(`sync-${i + 1}`, fetchAndUpdate);
107 }
108
109 return { likeCount, handle: handleResult.handle };
110 },
111);