nandi.uk
This repository has no description
1/**
2 * Papers Appview — watches the Paper Skygest feed, detects linked papers,
3 * generates summaries, and writes org.latha.papers.summary records to PDS.
4 *
5 * Hosted at papers.latha.org
6 */
7
8import { Hono } from "hono";
9import { cors } from "hono/cors";
10import { createWorker } from "@atmo-dev/contrail/worker";
11import { Contrail } from "@atmo-dev/contrail";
12import { lexicons } from "../lexicons/generated/index.js";
13import { config } from "./contrail.config.js";
14import { pollNewPapers } from "./feed-watcher.js";
15import { buildSummary } from "./paper-summarizer.js";
16import { writeSummaryToPds } from "./pds-writer.js";
17import { LANDING_PAGE } from "./landing-page.js";
18import { SUMMARY_PAGE } from "./summary-page.js";
19
20interface Env {
21 DB: D1Database;
22 FEED_CURSOR?: string;
23}
24
25const contrailWorker = createWorker(config, { lexicons });
26const contrail = new Contrail(config);
27let contrailReady = false;
28
29async function ensureContrailReady(db: D1Database): Promise<void> {
30 if (contrailReady) return;
31 await contrail.init(db);
32 contrailReady = true;
33}
34
35let app: Hono<{ Bindings: Env }> | null = null;
36
37function buildApp(env: Env): Hono<{ Bindings: Env }> {
38 const app = new Hono<{ Bindings: Env }>();
39 const db = env.DB;
40
41 app.use("*", cors());
42
43 // Landing page — browse all indexed summaries
44 app.get("/", async (c) => {
45 await ensureContrailReady(db);
46
47 let summaries: any[] = [];
48 try {
49 const rows = await db
50 .prepare(
51 "SELECT uri, record, did, rkey, time_us FROM records_summary ORDER BY time_us DESC LIMIT 100"
52 )
53 .all();
54 summaries = (rows.results || []).map((r: any) => {
55 try {
56 const value = JSON.parse(r.record);
57 return {
58 uri: r.uri,
59 rkey: r.rkey,
60 title: value?.title || "Untitled",
61 paperUrl: value?.paperUrl || "",
62 venue: value?.venue || "",
63 year: value?.year || null,
64 authors: value?.authors || [],
65 summary: value?.summary || "",
66 domains: value?.domains || [],
67 posterHandle: value?.posterHandle || "",
68 indexedAt: value?.indexedAt || "",
69 };
70 } catch {
71 return null;
72 }
73 }).filter(Boolean);
74 } catch {
75 // empty list is fine
76 }
77
78 const page = LANDING_PAGE.replace(
79 "</head>",
80 `<script>window.__SUMMARIES__=${JSON.stringify(summaries)};</script></head>`
81 );
82 return c.html(page);
83 });
84
85 // Summary detail page
86 app.get("/paper/:rkey", async (c) => {
87 const rkey = c.req.param("rkey");
88 await ensureContrailReady(db);
89
90 let summary: any = null;
91 try {
92 const row = await db
93 .prepare("SELECT uri, record, did, rkey FROM records_summary WHERE rkey = ? LIMIT 1")
94 .bind(rkey)
95 .first<{ uri: string; record: string; did: string; rkey: string }>();
96 if (row) {
97 const value = JSON.parse(row.record);
98 summary = {
99 uri: row.uri,
100 rkey: row.rkey,
101 ...value,
102 };
103 }
104 } catch {
105 // not found
106 }
107
108 if (!summary) {
109 return c.html("<h1>Not Found</h1>", 404);
110 }
111
112 const page = SUMMARY_PAGE.replace(
113 "</head>",
114 `<script>window.__SUMMARY__=${JSON.stringify(summary)};</script></head>`
115 );
116 return c.html(page);
117 });
118
119 // API: list summaries as JSON
120 app.get("/api/summaries", async (c) => {
121 await ensureContrailReady(db);
122 const limit = Math.min(Number(c.req.query("limit") || 50), 200);
123 const cursor = c.req.query("cursor");
124
125 let rows: D1Result<any>;
126 if (cursor) {
127 rows = await db
128 .prepare("SELECT uri, record, did, rkey, time_us FROM records_summary WHERE time_us < ? ORDER BY time_us DESC LIMIT ?")
129 .bind(cursor, limit)
130 .all();
131 } else {
132 rows = await db
133 .prepare("SELECT uri, record, did, rkey, time_us FROM records_summary ORDER BY time_us DESC LIMIT ?")
134 .bind(limit)
135 .all();
136 }
137
138 const summaries = (rows.results || []).map((r: any) => {
139 try {
140 const value = JSON.parse(r.record);
141 return { uri: r.uri, rkey: r.rkey, ...value };
142 } catch {
143 return null;
144 }
145 }).filter(Boolean);
146
147 const lastRow = rows.results?.[rows.results.length - 1];
148 const nextCursor = lastRow?.time_us || undefined;
149
150 return c.json({ summaries, cursor: nextCursor });
151 });
152
153 // API: trigger feed poll + summary generation
154 app.post("/api/poll", async (c) => {
155 await ensureContrailReady(db);
156
157 // Get last cursor from D1
158 let lastCursor: string | undefined;
159 try {
160 const row = await db
161 .prepare("SELECT value FROM instance_settings WHERE key = 'feed_cursor' LIMIT 1")
162 .first<{ value: string }>();
163 lastCursor = row?.value || undefined;
164 } catch {
165 // first run
166 }
167
168 const { posts, newCursor } = await pollNewPapers(lastCursor);
169
170 // Deduplicate against existing records
171 const existingUrls = new Set<string>();
172 if (posts.length > 0) {
173 const placeholders = posts.map(() => "?").join(",");
174 const urls = posts.map(p => p.paperUrl);
175 const rows = await db
176 .prepare(
177 `SELECT DISTINCT json_extract(record, '$.paperUrl') as paperUrl FROM records_summary WHERE paperUrl IN (${placeholders})`
178 )
179 .bind(...urls)
180 .all<{ paperUrl: string }>();
181 for (const r of rows.results || []) {
182 if (r.paperUrl) existingUrls.add(r.paperUrl);
183 }
184 }
185
186 const newPosts = posts.filter(p => !existingUrls.has(p.paperUrl));
187 let written = 0;
188 let errors = 0;
189
190 for (const post of newPosts) {
191 try {
192 const summary = await buildSummary(post);
193 const result = await writeSummaryToPds(post, summary);
194
195 // Notify contrail to index the new record
196 await contrail.notify(result.uri);
197 written++;
198 } catch (err: any) {
199 console.error(`Failed to write summary for ${post.paperUrl}: ${err?.message ?? err}`);
200 errors++;
201 }
202 }
203
204 // Save cursor
205 if (newCursor) {
206 await db
207 .prepare("INSERT OR REPLACE INTO instance_settings (key, value) VALUES ('feed_cursor', ?)")
208 .bind(newCursor)
209 .run();
210 }
211
212 return c.json({
213 polled: posts.length,
214 new: newPosts.length,
215 written,
216 errors,
217 });
218 });
219
220 // Health check
221 app.get("/api/health", async (c) => {
222 await ensureContrailReady(db);
223 let count = 0;
224 try {
225 const row = await db
226 .prepare("SELECT COUNT(*) as cnt FROM records_summary")
227 .first<{ cnt: number }>();
228 count = row?.cnt || 0;
229 } catch {
230 // table might not exist yet
231 }
232 return c.json({ ok: true, summaries: count });
233 });
234
235 // Crawl endpoint: register a DID and trigger immediate ingest
236 app.post("/xrpc/com.atproto.sync.requestCrawl", async (c) => {
237 let body: { did?: string; hostname?: string };
238 try {
239 body = await c.req.json();
240 } catch {
241 body = {};
242 }
243 const did = body.did;
244 if (!did) {
245 return c.json({ error: "BadRequest", message: "Provide did" }, 400);
246 }
247
248 await ensureContrailReady(db);
249 try {
250 // Ensure identities table has this DID
251 await db.exec(
252 "CREATE TABLE IF NOT EXISTS identities (did TEXT PRIMARY KEY, handle TEXT, time_us INTEGER)"
253 );
254 await db
255 .prepare("INSERT OR IGNORE INTO identities (did, handle, time_us) VALUES (?, ?, ?)")
256 .bind(did, "researcher.pds.latha.org", Date.now() * 1000)
257 .run();
258
259 // Ensure records_summary table exists
260 await db.exec(
261 "CREATE TABLE IF NOT EXISTS records_summary (uri TEXT PRIMARY KEY, cid TEXT, did TEXT, rkey TEXT, record TEXT, time_us INTEGER)"
262 );
263
264 // Fetch records from PDS and insert directly
265 const pdsUrl = `https://pds.latha.org/xrpc/com.atproto.repo.listRecords?repo=${did}&collection=org.latha.papers.summary&limit=100`;
266 const res = await fetch(pdsUrl);
267 if (res.ok) {
268 const data = await res.json() as {
269 records: Array<{ uri: string; cid: string; value: any }>;
270 };
271 for (const rec of data.records) {
272 const rkey = rec.uri.split("/").pop() || "";
273 await db
274 .prepare(
275 "INSERT OR REPLACE INTO records_summary (uri, cid, did, rkey, record, time_us) VALUES (?, ?, ?, ?, ?, ?)"
276 )
277 .bind(rec.uri, rec.cid, did, rkey, JSON.stringify(rec.value), Date.now() * 1000)
278 .run();
279 }
280 }
281 } catch (err: any) {
282 console.error(`Crawl error: ${err?.message ?? err}`);
283 }
284
285 return c.json({ ok: true, did });
286 });
287
288 // All other routes pass through to contrail
289 app.all("*", async (c) => {
290 const response = await contrailWorker.fetch(
291 c.req.raw,
292 c.env as unknown as Record<string, unknown>
293 );
294 return response;
295 });
296
297 return app;
298}
299
300export default {
301 fetch(request: Request, env: Env): Response | Promise<Response> {
302 app ??= buildApp(env);
303 return app.fetch(request, env);
304 },
305 async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
306 // Cron: poll feed and generate summaries
307 const db = env.DB;
308 await ensureContrailReady(db);
309
310 let lastCursor: string | undefined;
311 try {
312 const row = await db
313 .prepare("SELECT value FROM instance_settings WHERE key = 'feed_cursor' LIMIT 1")
314 .first<{ value: string }>();
315 lastCursor = row?.value || undefined;
316 } catch {
317 // first run
318 }
319
320 const { posts, newCursor } = await pollNewPapers(lastCursor);
321
322 // Deduplicate
323 const existingUrls = new Set<string>();
324 if (posts.length > 0) {
325 const placeholders = posts.map(() => "?").join(",");
326 const urls = posts.map(p => p.paperUrl);
327 const rows = await db
328 .prepare(
329 `SELECT DISTINCT json_extract(record, '$.paperUrl') as paperUrl FROM records_summary WHERE paperUrl IN (${placeholders})`
330 )
331 .bind(...urls)
332 .all<{ paperUrl: string }>();
333 for (const r of rows.results || []) {
334 if (r.paperUrl) existingUrls.add(r.paperUrl);
335 }
336 }
337
338 const newPosts = posts.filter(p => !existingUrls.has(p.paperUrl));
339
340 for (const post of newPosts) {
341 try {
342 const summary = await buildSummary(post);
343 const result = await writeSummaryToPds(post, summary);
344 await contrail.notify(result.uri);
345 } catch (err: any) {
346 console.error(`Cron: failed to write summary for ${post.paperUrl}: ${err?.message ?? err}`);
347 }
348 }
349
350 if (newCursor) {
351 await db
352 .prepare("INSERT OR REPLACE INTO instance_settings (key, value) VALUES ('feed_cursor', ?)")
353 .bind(newCursor)
354 .run();
355 }
356 },
357};