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, pollFollowedAccounts, type PaperPost } from "./feed-watcher.js";
15import { buildSummary, enrichWithLetta, type PaperSummary } 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 RESEARCHER_KEY_PEM?: string;
24 LETTA_API_KEY?: string;
25 LETTA_AGENT_ID?: string;
26}
27
28const contrailWorker = createWorker(config, { lexicons });
29const contrail = new Contrail(config);
30let contrailReady = false;
31
32async function ensureContrailReady(db: D1Database): Promise<void> {
33 if (contrailReady) return;
34 await contrail.init(db);
35 contrailReady = true;
36}
37
38let app: Hono<{ Bindings: Env }> | null = null;
39
40function buildApp(env: Env): Hono<{ Bindings: Env }> {
41 const app = new Hono<{ Bindings: Env }>();
42 const db = env.DB;
43
44 app.use("*", cors());
45
46 // Landing page — browse all indexed summaries
47 app.get("/", async (c) => {
48 await ensureContrailReady(db);
49
50 let summaries: any[] = [];
51 try {
52 const rows = await db
53 .prepare(
54 "SELECT uri, record, did, rkey, time_us FROM records_summary ORDER BY json_extract(record, '$.publishedDate') DESC NULLS LAST, time_us DESC LIMIT 100"
55 )
56 .all();
57 summaries = (rows.results || []).map((r: any) => {
58 try {
59 const value = JSON.parse(r.record);
60 return {
61 uri: r.uri,
62 rkey: r.rkey,
63 title: value?.title || "Untitled",
64 paperUrl: value?.paperUrl || "",
65 venue: value?.venue || "",
66 year: value?.year || null,
67 publishedDate: value?.publishedDate || null,
68 authors: value?.authors || [],
69 summary: value?.summary || "",
70 domains: value?.domains || [],
71 posterHandle: value?.posterHandle || "",
72 indexedAt: value?.indexedAt || "",
73 };
74 } catch {
75 return null;
76 }
77 }).filter(Boolean);
78 } catch {
79 // empty list is fine
80 }
81
82 const page = LANDING_PAGE.replace(
83 "</head>",
84 `<script>window.__SUMMARIES__=${JSON.stringify(summaries).replace(/<\//g, "<\\/")};</script></head>`
85 );
86 return c.html(page);
87 });
88
89 // Summary detail page
90 app.get("/paper/:rkey", async (c) => {
91 const rkey = c.req.param("rkey");
92 await ensureContrailReady(db);
93
94 let summary: any = null;
95 try {
96 const row = await db
97 .prepare("SELECT uri, record, did, rkey FROM records_summary WHERE rkey = ? LIMIT 1")
98 .bind(rkey)
99 .first<{ uri: string; record: string; did: string; rkey: string }>();
100 if (row) {
101 const value = JSON.parse(row.record);
102 summary = {
103 uri: row.uri,
104 rkey: row.rkey,
105 ...value,
106 };
107 }
108 } catch {
109 // not found
110 }
111
112 if (!summary) {
113 return c.html("<h1>Not Found</h1>", 404);
114 }
115
116 const page = SUMMARY_PAGE.replace(
117 "</head>",
118 `<script>window.__SUMMARY__=${JSON.stringify(summary)};</script></head>`
119 );
120 return c.html(page);
121 });
122
123 // API: list summaries as JSON
124 app.get("/api/summaries", async (c) => {
125 await ensureContrailReady(db);
126 const limit = Math.min(Number(c.req.query("limit") || 50), 200);
127 const cursor = c.req.query("cursor");
128
129 let rows: D1Result<any>;
130 if (cursor) {
131 rows = await db
132 .prepare("SELECT uri, record, did, rkey, time_us FROM records_summary WHERE time_us < ? ORDER BY time_us DESC LIMIT ?")
133 .bind(cursor, limit)
134 .all();
135 } else {
136 rows = await db
137 .prepare("SELECT uri, record, did, rkey, time_us FROM records_summary ORDER BY time_us DESC LIMIT ?")
138 .bind(limit)
139 .all();
140 }
141
142 const summaries = (rows.results || []).map((r: any) => {
143 try {
144 const value = JSON.parse(r.record);
145 return { uri: r.uri, rkey: r.rkey, ...value };
146 } catch {
147 return null;
148 }
149 }).filter(Boolean);
150
151 const lastRow = rows.results?.[rows.results.length - 1];
152 const nextCursor = lastRow?.time_us || undefined;
153
154 return c.json({ summaries, cursor: nextCursor });
155 });
156
157 // API: trigger feed poll + summary generation
158 app.post("/api/poll", async (c) => {
159 await ensureContrailReady(db);
160
161 // Get last cursor from D1
162 let lastCursor: string | undefined;
163 try {
164 const row = await db
165 .prepare("SELECT value FROM instance_settings WHERE key = 'feed_cursor' LIMIT 1")
166 .first<{ value: string }>();
167 lastCursor = row?.value || undefined;
168 } catch {
169 // first run
170 }
171
172 const { posts: feedPosts, newCursor } = await pollNewPapers(lastCursor);
173
174 // Also poll followed accounts for paper links
175 let followedPosts: PaperPost[] = [];
176 try {
177 followedPosts = await pollFollowedAccounts(5);
178 } catch (err: any) {
179 console.error(`Followed accounts poll failed: ${err?.message ?? err}`);
180 }
181
182 // Merge and deduplicate
183 const seenPaperUrls = new Set<string>();
184 const posts: PaperPost[] = [];
185 for (const p of [...feedPosts, ...followedPosts]) {
186 if (!seenPaperUrls.has(p.paperUrl)) {
187 seenPaperUrls.add(p.paperUrl);
188 posts.push(p);
189 }
190 }
191
192 // Deduplicate against existing records
193 const existingUrls = new Set<string>();
194 if (posts.length > 0) {
195 const placeholders = posts.map(() => "?").join(",");
196 const urls = posts.map(p => p.paperUrl);
197 const rows = await db
198 .prepare(
199 `SELECT DISTINCT json_extract(record, '$.paperUrl') as paperUrl FROM records_summary WHERE paperUrl IN (${placeholders})`
200 )
201 .bind(...urls)
202 .all<{ paperUrl: string }>();
203 for (const r of rows.results || []) {
204 if (r.paperUrl) existingUrls.add(r.paperUrl);
205 }
206 }
207
208 const newPosts = posts.filter(p => !existingUrls.has(p.paperUrl));
209 let written = 0;
210 let errors = 0;
211 const errorDetails: string[] = [];
212
213 const apiKey = process.env.LETTA_API_KEY;
214 const agentId = process.env.LETTA_AGENT_ID;
215
216 // Build heuristic summaries in parallel
217 const heuristicResults = await Promise.all(newPosts.map(async (post) => {
218 try {
219 return { post, summary: await buildSummary(post) };
220 } catch (err: any) {
221 return { post, error: err?.message ?? err };
222 }
223 }));
224
225 // Enrich in parallel (batch of 5 to avoid rate limits)
226 const enrichedResults: Array<{ post: PaperPost; summary: PaperSummary; error?: string }> = [];
227 for (let i = 0; i < heuristicResults.length; i += 5) {
228 const batch = heuristicResults.slice(i, i + 5);
229 const batchResults = await Promise.all(batch.map(async ({ post, summary, error }) => {
230 if (error || !summary) return { post, summary: summary!, error };
231 if (!apiKey || !agentId) return { post, summary };
232 try {
233 const enrichment = await enrichWithLetta(
234 { title: summary.title, paperUrl: summary.paperUrl, abstract: summary.abstract, domains: summary.domains, venue: summary.venue },
235 apiKey,
236 agentId,
237 );
238 if (enrichment.title) summary.title = enrichment.title;
239 summary.summary = enrichment.summary;
240 summary.domains = enrichment.domains;
241 if (enrichment.takeaway) (summary as any).takeaway = enrichment.takeaway;
242 if (enrichment.year) summary.year = enrichment.year;
243 if (enrichment.publishedDate) (summary as any).publishedDate = enrichment.publishedDate;
244 return { post, summary };
245 } catch (err: any) {
246 console.error(`Enrichment failed for ${post.paperUrl}: ${err?.message ?? err}, using heuristic`);
247 return { post, summary };
248 }
249 }));
250 enrichedResults.push(...batchResults);
251 }
252
253 // Write all to PDS and D1
254 for (const { post, summary, error } of enrichedResults) {
255 if (error || !summary) {
256 const msg = `${post.paperUrl}: ${error}`;
257 console.error(`Failed to build summary for ${msg}`);
258 if (errors < 3) errorDetails.push(msg.substring(0, 200));
259 errors++;
260 continue;
261 }
262 try {
263 const result = await writeSummaryToPds(post, summary);
264
265 const rkey = result.uri.split("/").pop() || "";
266 await db
267 .prepare("INSERT OR REPLACE INTO records_summary (uri, cid, did, rkey, record, time_us, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?)")
268 .bind(result.uri, result.cid, "did:plc:3kkhul7jznlb6ba7rprzawnj", rkey, JSON.stringify({
269 $type: "org.latha.papers.summary",
270 ...summary,
271 sourceUri: post.postUri,
272 posterDid: post.posterDid,
273 posterHandle: post.posterHandle,
274 postText: post.postText,
275 indexedAt: new Date().toISOString(),
276 }), Date.now() * 1000, Date.now() * 1000)
277 .run();
278 written++;
279 } catch (err: any) {
280 const msg = `${post.paperUrl}: ${err?.message ?? err}`;
281 console.error(`Failed to write summary for ${msg}`);
282 if (errors < 3) errorDetails.push(msg.substring(0, 200));
283 errors++;
284 }
285 }
286
287 // Save cursor
288 if (newCursor) {
289 await db
290 .prepare("INSERT OR REPLACE INTO instance_settings (key, value) VALUES ('feed_cursor', ?)")
291 .bind(newCursor)
292 .run();
293 }
294
295 return c.json({
296 polled: posts.length,
297 fromFeed: feedPosts.length,
298 fromFollows: followedPosts.length,
299 new: newPosts.length,
300 written,
301 errors,
302 lastError: errors > 0 ? errorDetails[0] : undefined,
303 });
304 });
305
306 // Health check
307 app.get("/api/health", async (c) => {
308 await ensureContrailReady(db);
309 let count = 0;
310 try {
311 const row = await db
312 .prepare("SELECT COUNT(*) as cnt FROM records_summary")
313 .first<{ cnt: number }>();
314 count = row?.cnt || 0;
315 } catch {
316 // table might not exist yet
317 }
318 const hasKey = !!process.env.RESEARCHER_KEY_PEM;
319 return c.json({ ok: true, summaries: count, hasPdsKey: hasKey });
320 });
321
322 // Enrich existing summaries using Letta API
323 app.post("/api/enrich", async (c) => {
324 const apiKey = process.env.LETTA_API_KEY;
325 const agentId = process.env.LETTA_AGENT_ID;
326 if (!apiKey || !agentId) {
327 return c.json({ error: "Missing LETTA_API_KEY or LETTA_AGENT_ID" }, 500);
328 }
329
330 await ensureContrailReady(db);
331
332 // Get records that need enrichment
333 const rows = await db
334 .prepare(
335 `SELECT rkey, record FROM records_summary
336 WHERE json_extract(record, '$.summary') = ''
337 OR json_extract(record, '$.summary') LIKE 'Published in%'
338 OR json_extract(record, '$.title') = 'Untitled'
339 OR json_extract(record, '$.takeaway') IS NULL
340 OR json_extract(record, '$.takeaway') = ''
341 OR json_extract(record, '$.year') IS NULL
342 OR json_extract(record, '$.year') = 0
343 OR json_extract(record, '$.publishedDate') IS NULL
344 ORDER BY time_us DESC LIMIT 20`
345 )
346 .all<{ rkey: string; record: string }>();
347
348 if (!rows.results?.length) {
349 return c.json({ enriched: 0, message: "No records need enrichment" });
350 }
351
352 let enriched = 0;
353 let errors = 0;
354
355 // Enrich in parallel batches of 5
356 for (let i = 0; i < rows.results.length; i += 5) {
357 const batch = rows.results.slice(i, i + 5);
358 const results = await Promise.allSettled(batch.map(async (row) => {
359 const record = JSON.parse(row.record) as {
360 title: string;
361 paperUrl: string;
362 abstract: string;
363 domains: string[];
364 venue: string;
365 summary: string;
366 takeaway?: string;
367 year?: number;
368 };
369
370 const result = await enrichWithLetta(
371 {
372 title: record.title,
373 paperUrl: record.paperUrl,
374 abstract: record.abstract,
375 domains: record.domains || [],
376 venue: record.venue || "",
377 },
378 apiKey,
379 agentId,
380 );
381
382 record.summary = result.summary;
383 record.domains = result.domains;
384 if (result.takeaway) record.takeaway = result.takeaway;
385 if (result.title) record.title = result.title;
386 if (result.year) record.year = result.year;
387 if (result.publishedDate) record.publishedDate = result.publishedDate;
388
389 await db
390 .prepare("UPDATE records_summary SET record = ? WHERE rkey = ?")
391 .bind(JSON.stringify(record), row.rkey)
392 .run();
393
394 return record.title;
395 }));
396
397 for (const r of results) {
398 if (r.status === "fulfilled") {
399 enriched++;
400 console.log(`Enriched: ${r.value?.substring(0, 50)}...`);
401 } else {
402 console.error(`Enrichment failed: ${r.reason}`);
403 errors++;
404 }
405 }
406 }
407
408 return c.json({ enriched, errors, total: rows.results.length });
409 });
410
411 // Crawl endpoint: register a DID and trigger immediate ingest
412 app.post("/xrpc/com.atproto.sync.requestCrawl", async (c) => {
413 let body: { did?: string; hostname?: string };
414 try {
415 body = await c.req.json();
416 } catch {
417 body = {};
418 }
419 const did = body.did;
420 if (!did) {
421 return c.json({ error: "BadRequest", message: "Provide did" }, 400);
422 }
423
424 await ensureContrailReady(db);
425 try {
426 // Ensure identities table has this DID
427 await db.exec(
428 "CREATE TABLE IF NOT EXISTS identities (did TEXT PRIMARY KEY, handle TEXT, time_us INTEGER)"
429 );
430 await db
431 .prepare("INSERT OR IGNORE INTO identities (did, handle, time_us) VALUES (?, ?, ?)")
432 .bind(did, "researcher.pds.latha.org", Date.now() * 1000)
433 .run();
434
435 // Ensure records_summary table exists
436 await db.exec(
437 "CREATE TABLE IF NOT EXISTS records_summary (uri TEXT PRIMARY KEY, cid TEXT, did TEXT, rkey TEXT, record TEXT, time_us INTEGER)"
438 );
439
440 // Fetch records from PDS and insert directly
441 const pdsUrl = `https://pds.latha.org/xrpc/com.atproto.repo.listRecords?repo=${did}&collection=org.latha.papers.summary&limit=100`;
442 const res = await fetch(pdsUrl);
443 if (res.ok) {
444 const data = await res.json() as {
445 records: Array<{ uri: string; cid: string; value: any }>;
446 };
447 for (const rec of data.records) {
448 const rkey = rec.uri.split("/").pop() || "";
449 await db
450 .prepare(
451 "INSERT OR REPLACE INTO records_summary (uri, cid, did, rkey, record, time_us) VALUES (?, ?, ?, ?, ?, ?)"
452 )
453 .bind(rec.uri, rec.cid, did, rkey, JSON.stringify(rec.value), Date.now() * 1000)
454 .run();
455 }
456 }
457 } catch (err: any) {
458 console.error(`Crawl error: ${err?.message ?? err}`);
459 }
460
461 return c.json({ ok: true, did });
462 });
463
464 // All other routes pass through to contrail
465 app.all("*", async (c) => {
466 const response = await contrailWorker.fetch(
467 c.req.raw,
468 c.env as unknown as Record<string, unknown>
469 );
470 return response;
471 });
472
473 return app;
474}
475
476export default {
477 fetch(request: Request, env: Env): Response | Promise<Response> {
478 app ??= buildApp(env);
479 return app.fetch(request, env);
480 },
481 async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
482 // Cron: poll feed and generate summaries
483 const db = env.DB;
484 await ensureContrailReady(db);
485
486 let lastCursor: string | undefined;
487 try {
488 const row = await db
489 .prepare("SELECT value FROM instance_settings WHERE key = 'feed_cursor' LIMIT 1")
490 .first<{ value: string }>();
491 lastCursor = row?.value || undefined;
492 } catch {
493 // first run
494 }
495
496 const { posts: feedPosts, newCursor } = await pollNewPapers(lastCursor);
497
498 // Also poll followed accounts
499 let followedPosts: PaperPost[] = [];
500 try {
501 followedPosts = await pollFollowedAccounts(5);
502 } catch {}
503
504 // Merge and deduplicate
505 const seenPaperUrls = new Set<string>();
506 const posts: PaperPost[] = [];
507 for (const p of [...feedPosts, ...followedPosts]) {
508 if (!seenPaperUrls.has(p.paperUrl)) {
509 seenPaperUrls.add(p.paperUrl);
510 posts.push(p);
511 }
512 }
513
514 // Deduplicate against existing records
515 const existingUrls = new Set<string>();
516 if (posts.length > 0) {
517 const placeholders = posts.map(() => "?").join(",");
518 const urls = posts.map(p => p.paperUrl);
519 const rows = await db
520 .prepare(
521 `SELECT DISTINCT json_extract(record, '$.paperUrl') as paperUrl FROM records_summary WHERE paperUrl IN (${placeholders})`
522 )
523 .bind(...urls)
524 .all<{ paperUrl: string }>();
525 for (const r of rows.results || []) {
526 if (r.paperUrl) existingUrls.add(r.paperUrl);
527 }
528 }
529
530 const newPosts = posts.filter(p => !existingUrls.has(p.paperUrl));
531 const apiKey = process.env.LETTA_API_KEY;
532 const agentId = process.env.LETTA_AGENT_ID;
533
534 // Build heuristic summaries in parallel
535 const heuristicResults = await Promise.all(newPosts.map(async (post) => {
536 try {
537 return { post, summary: await buildSummary(post) };
538 } catch (err: any) {
539 return { post, error: err?.message ?? err };
540 }
541 }));
542
543 // Enrich in parallel (batch of 5)
544 const enrichedResults: Array<{ post: PaperPost; summary: PaperSummary; error?: string }> = [];
545 for (let i = 0; i < heuristicResults.length; i += 5) {
546 const batch = heuristicResults.slice(i, i + 5);
547 const batchResults = await Promise.all(batch.map(async ({ post, summary, error }) => {
548 if (error || !summary) return { post, summary: summary!, error };
549 if (!apiKey || !agentId) return { post, summary };
550 try {
551 const enrichment = await enrichWithLetta(
552 { title: summary.title, paperUrl: summary.paperUrl, abstract: summary.abstract, domains: summary.domains, venue: summary.venue },
553 apiKey,
554 agentId,
555 );
556 if (enrichment.title) summary.title = enrichment.title;
557 summary.summary = enrichment.summary;
558 summary.domains = enrichment.domains;
559 if (enrichment.takeaway) (summary as any).takeaway = enrichment.takeaway;
560 if (enrichment.year) summary.year = enrichment.year;
561 if (enrichment.publishedDate) (summary as any).publishedDate = enrichment.publishedDate;
562 return { post, summary };
563 } catch {
564 return { post, summary };
565 }
566 }));
567 enrichedResults.push(...batchResults);
568 }
569
570 // Write all to PDS and D1
571 for (const { post, summary, error } of enrichedResults) {
572 if (error || !summary) {
573 console.error(`Cron: failed to build summary for ${post.paperUrl}: ${error}`);
574 continue;
575 }
576 try {
577 const result = await writeSummaryToPds(post, summary);
578
579 const rkey = result.uri.split("/").pop() || "";
580 await db
581 .prepare("INSERT OR REPLACE INTO records_summary (uri, cid, did, rkey, record, time_us, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?)")
582 .bind(result.uri, result.cid, "did:plc:3kkhul7jznlb6ba7rprzawnj", rkey, JSON.stringify({
583 $type: "org.latha.papers.summary",
584 ...summary,
585 sourceUri: post.postUri,
586 posterDid: post.posterDid,
587 posterHandle: post.posterHandle,
588 postText: post.postText,
589 indexedAt: new Date().toISOString(),
590 }), Date.now() * 1000, Date.now() * 1000)
591 .run();
592 } catch (err: any) {
593 console.error(`Cron: failed to write summary for ${post.paperUrl}: ${err?.message ?? err}`);
594 }
595 }
596
597 if (newCursor) {
598 await db
599 .prepare("INSERT OR REPLACE INTO instance_settings (key, value) VALUES ('feed_cursor', ?)")
600 .bind(newCursor)
601 .run();
602 }
603 },
604};