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, enrichWithLetta } 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 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 authors: value?.authors || [],
68 summary: value?.summary || "",
69 domains: value?.domains || [],
70 posterHandle: value?.posterHandle || "",
71 indexedAt: value?.indexedAt || "",
72 };
73 } catch {
74 return null;
75 }
76 }).filter(Boolean);
77 } catch {
78 // empty list is fine
79 }
80
81 const page = LANDING_PAGE.replace(
82 "</head>",
83 `<script>window.__SUMMARIES__=${JSON.stringify(summaries)};</script></head>`
84 );
85 return c.html(page);
86 });
87
88 // Summary detail page
89 app.get("/paper/:rkey", async (c) => {
90 const rkey = c.req.param("rkey");
91 await ensureContrailReady(db);
92
93 let summary: any = null;
94 try {
95 const row = await db
96 .prepare("SELECT uri, record, did, rkey FROM records_summary WHERE rkey = ? LIMIT 1")
97 .bind(rkey)
98 .first<{ uri: string; record: string; did: string; rkey: string }>();
99 if (row) {
100 const value = JSON.parse(row.record);
101 summary = {
102 uri: row.uri,
103 rkey: row.rkey,
104 ...value,
105 };
106 }
107 } catch {
108 // not found
109 }
110
111 if (!summary) {
112 return c.html("<h1>Not Found</h1>", 404);
113 }
114
115 const page = SUMMARY_PAGE.replace(
116 "</head>",
117 `<script>window.__SUMMARY__=${JSON.stringify(summary)};</script></head>`
118 );
119 return c.html(page);
120 });
121
122 // API: list summaries as JSON
123 app.get("/api/summaries", async (c) => {
124 await ensureContrailReady(db);
125 const limit = Math.min(Number(c.req.query("limit") || 50), 200);
126 const cursor = c.req.query("cursor");
127
128 let rows: D1Result<any>;
129 if (cursor) {
130 rows = await db
131 .prepare("SELECT uri, record, did, rkey, time_us FROM records_summary WHERE time_us < ? ORDER BY time_us DESC LIMIT ?")
132 .bind(cursor, limit)
133 .all();
134 } else {
135 rows = await db
136 .prepare("SELECT uri, record, did, rkey, time_us FROM records_summary ORDER BY time_us DESC LIMIT ?")
137 .bind(limit)
138 .all();
139 }
140
141 const summaries = (rows.results || []).map((r: any) => {
142 try {
143 const value = JSON.parse(r.record);
144 return { uri: r.uri, rkey: r.rkey, ...value };
145 } catch {
146 return null;
147 }
148 }).filter(Boolean);
149
150 const lastRow = rows.results?.[rows.results.length - 1];
151 const nextCursor = lastRow?.time_us || undefined;
152
153 return c.json({ summaries, cursor: nextCursor });
154 });
155
156 // API: trigger feed poll + summary generation
157 app.post("/api/poll", async (c) => {
158 await ensureContrailReady(db);
159
160 // Get last cursor from D1
161 let lastCursor: string | undefined;
162 try {
163 const row = await db
164 .prepare("SELECT value FROM instance_settings WHERE key = 'feed_cursor' LIMIT 1")
165 .first<{ value: string }>();
166 lastCursor = row?.value || undefined;
167 } catch {
168 // first run
169 }
170
171 const { posts, newCursor } = await pollNewPapers(lastCursor);
172
173 // Deduplicate against existing records
174 const existingUrls = new Set<string>();
175 if (posts.length > 0) {
176 const placeholders = posts.map(() => "?").join(",");
177 const urls = posts.map(p => p.paperUrl);
178 const rows = await db
179 .prepare(
180 `SELECT DISTINCT json_extract(record, '$.paperUrl') as paperUrl FROM records_summary WHERE paperUrl IN (${placeholders})`
181 )
182 .bind(...urls)
183 .all<{ paperUrl: string }>();
184 for (const r of rows.results || []) {
185 if (r.paperUrl) existingUrls.add(r.paperUrl);
186 }
187 }
188
189 const newPosts = posts.filter(p => !existingUrls.has(p.paperUrl));
190 let written = 0;
191 let errors = 0;
192
193 for (const post of newPosts) {
194 try {
195 const summary = await buildSummary(post);
196 const result = await writeSummaryToPds(post, summary);
197
198 // Notify contrail to index the new record
199 await contrail.notify(result.uri);
200 written++;
201 } catch (err: any) {
202 console.error(`Failed to write summary for ${post.paperUrl}: ${err?.message ?? err}`);
203 errors++;
204 }
205 }
206
207 // Save cursor
208 if (newCursor) {
209 await db
210 .prepare("INSERT OR REPLACE INTO instance_settings (key, value) VALUES ('feed_cursor', ?)")
211 .bind(newCursor)
212 .run();
213 }
214
215 return c.json({
216 polled: posts.length,
217 new: newPosts.length,
218 written,
219 errors,
220 });
221 });
222
223 // Health check
224 app.get("/api/health", async (c) => {
225 await ensureContrailReady(db);
226 let count = 0;
227 try {
228 const row = await db
229 .prepare("SELECT COUNT(*) as cnt FROM records_summary")
230 .first<{ cnt: number }>();
231 count = row?.cnt || 0;
232 } catch {
233 // table might not exist yet
234 }
235 return c.json({ ok: true, summaries: count });
236 });
237
238 // Enrich existing summaries using Letta API
239 app.post("/api/enrich", async (c) => {
240 const apiKey = process.env.LETTA_API_KEY;
241 const agentId = process.env.LETTA_AGENT_ID;
242 if (!apiKey || !agentId) {
243 return c.json({ error: "Missing LETTA_API_KEY or LETTA_AGENT_ID" }, 500);
244 }
245
246 await ensureContrailReady(db);
247
248 // Get records that need enrichment (empty/placeholder summaries, no takeaway, or Untitled)
249 const rows = await db
250 .prepare(
251 `SELECT rkey, record FROM records_summary
252 WHERE json_extract(record, '$.summary') = ''
253 OR json_extract(record, '$.summary') LIKE 'Published in%'
254 OR json_extract(record, '$.title') = 'Untitled'
255 OR json_extract(record, '$.takeaway') IS NULL
256 OR json_extract(record, '$.takeaway') = ''
257 ORDER BY time_us DESC LIMIT 20`
258 )
259 .all<{ rkey: string; record: string }>();
260
261 if (!rows.results?.length) {
262 return c.json({ enriched: 0, message: "No records need enrichment" });
263 }
264
265 let enriched = 0;
266 let errors = 0;
267
268 for (const row of rows.results) {
269 try {
270 const record = JSON.parse(row.record) as {
271 title: string;
272 paperUrl: string;
273 abstract: string;
274 domains: string[];
275 venue: string;
276 summary: string;
277 takeaway?: string;
278 };
279
280 const result = await enrichWithLetta(
281 {
282 title: record.title,
283 paperUrl: record.paperUrl,
284 abstract: record.abstract,
285 domains: record.domains || [],
286 venue: record.venue || "",
287 },
288 apiKey,
289 agentId,
290 );
291
292 // Update the record in D1
293 record.summary = result.summary;
294 record.domains = result.domains;
295 if (result.takeaway) record.takeaway = result.takeaway;
296 if (result.title) record.title = result.title;
297
298 await db
299 .prepare("UPDATE records_summary SET record = ? WHERE rkey = ?")
300 .bind(JSON.stringify(record), row.rkey)
301 .run();
302
303 enriched++;
304 console.log(`Enriched: ${record.title?.substring(0, 50)}...`);
305 } catch (err: any) {
306 console.error(`Enrichment failed for ${row.rkey}: ${err?.message ?? err}`);
307 errors++;
308 }
309 }
310
311 return c.json({ enriched, errors, total: rows.results.length });
312 });
313
314 // Crawl endpoint: register a DID and trigger immediate ingest
315 app.post("/xrpc/com.atproto.sync.requestCrawl", async (c) => {
316 let body: { did?: string; hostname?: string };
317 try {
318 body = await c.req.json();
319 } catch {
320 body = {};
321 }
322 const did = body.did;
323 if (!did) {
324 return c.json({ error: "BadRequest", message: "Provide did" }, 400);
325 }
326
327 await ensureContrailReady(db);
328 try {
329 // Ensure identities table has this DID
330 await db.exec(
331 "CREATE TABLE IF NOT EXISTS identities (did TEXT PRIMARY KEY, handle TEXT, time_us INTEGER)"
332 );
333 await db
334 .prepare("INSERT OR IGNORE INTO identities (did, handle, time_us) VALUES (?, ?, ?)")
335 .bind(did, "researcher.pds.latha.org", Date.now() * 1000)
336 .run();
337
338 // Ensure records_summary table exists
339 await db.exec(
340 "CREATE TABLE IF NOT EXISTS records_summary (uri TEXT PRIMARY KEY, cid TEXT, did TEXT, rkey TEXT, record TEXT, time_us INTEGER)"
341 );
342
343 // Fetch records from PDS and insert directly
344 const pdsUrl = `https://pds.latha.org/xrpc/com.atproto.repo.listRecords?repo=${did}&collection=org.latha.papers.summary&limit=100`;
345 const res = await fetch(pdsUrl);
346 if (res.ok) {
347 const data = await res.json() as {
348 records: Array<{ uri: string; cid: string; value: any }>;
349 };
350 for (const rec of data.records) {
351 const rkey = rec.uri.split("/").pop() || "";
352 await db
353 .prepare(
354 "INSERT OR REPLACE INTO records_summary (uri, cid, did, rkey, record, time_us) VALUES (?, ?, ?, ?, ?, ?)"
355 )
356 .bind(rec.uri, rec.cid, did, rkey, JSON.stringify(rec.value), Date.now() * 1000)
357 .run();
358 }
359 }
360 } catch (err: any) {
361 console.error(`Crawl error: ${err?.message ?? err}`);
362 }
363
364 return c.json({ ok: true, did });
365 });
366
367 // All other routes pass through to contrail
368 app.all("*", async (c) => {
369 const response = await contrailWorker.fetch(
370 c.req.raw,
371 c.env as unknown as Record<string, unknown>
372 );
373 return response;
374 });
375
376 return app;
377}
378
379export default {
380 fetch(request: Request, env: Env): Response | Promise<Response> {
381 app ??= buildApp(env);
382 return app.fetch(request, env);
383 },
384 async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
385 // Cron: poll feed and generate summaries
386 const db = env.DB;
387 await ensureContrailReady(db);
388
389 let lastCursor: string | undefined;
390 try {
391 const row = await db
392 .prepare("SELECT value FROM instance_settings WHERE key = 'feed_cursor' LIMIT 1")
393 .first<{ value: string }>();
394 lastCursor = row?.value || undefined;
395 } catch {
396 // first run
397 }
398
399 const { posts, newCursor } = await pollNewPapers(lastCursor);
400
401 // Deduplicate
402 const existingUrls = new Set<string>();
403 if (posts.length > 0) {
404 const placeholders = posts.map(() => "?").join(",");
405 const urls = posts.map(p => p.paperUrl);
406 const rows = await db
407 .prepare(
408 `SELECT DISTINCT json_extract(record, '$.paperUrl') as paperUrl FROM records_summary WHERE paperUrl IN (${placeholders})`
409 )
410 .bind(...urls)
411 .all<{ paperUrl: string }>();
412 for (const r of rows.results || []) {
413 if (r.paperUrl) existingUrls.add(r.paperUrl);
414 }
415 }
416
417 const newPosts = posts.filter(p => !existingUrls.has(p.paperUrl));
418
419 for (const post of newPosts) {
420 try {
421 const summary = await buildSummary(post);
422 const result = await writeSummaryToPds(post, summary);
423 await contrail.notify(result.uri);
424 } catch (err: any) {
425 console.error(`Cron: failed to write summary for ${post.paperUrl}: ${err?.message ?? err}`);
426 }
427 }
428
429 if (newCursor) {
430 await db
431 .prepare("INSERT OR REPLACE INTO instance_settings (key, value) VALUES ('feed_cursor', ?)")
432 .bind(newCursor)
433 .run();
434 }
435 },
436};