nandi.uk
This repository has no description
14 kB
452 lines
1import { Hono } from "hono";
2import { cors } from "hono/cors";
3import { createWorker } from "@atmo-dev/contrail/worker";
4import { Contrail } from "@atmo-dev/contrail";
5import { config } from "./contrail.config.js";
6import { lexicons } from "../lexicons/generated/index.js";
7import { LANDING_PAGE } from "./landing-page.js";
8
9// ─── Env ────────────────────────────────────────────────────────────
10
11interface Env {
12 DB: D1Database;
13 VAULT_BUCKET: R2Bucket;
14 CARRY_BUCKET: R2Bucket;
15}
16
17// ─── Contrail setup ─────────────────────────────────────────────────
18
19const contrailWorker = createWorker(config, { lexicons });
20const contrail = new Contrail(config);
21let contrailReady = false;
22
23async function ensureContrailReady(db: D1Database): Promise<void> {
24 if (contrailReady) return;
25 await contrail.init(db);
26 contrailReady = true;
27}
28
29// ─── Island detection (connected components) ────────────────────────
30//
31// Load all connections, build an undirected graph, find connected
32// components via BFS. Each component is an "island" — a cluster of
33// linked URLs that form a constellation in the knowledge graph.
34
35interface Island {
36 id: string;
37 vertices: string[];
38 edges: Array<{
39 uri: string;
40 did: string;
41 source: string;
42 target: string;
43 connectionType: string;
44 note: string;
45 handle?: string | null;
46 }>;
47}
48
49async function detectIslands(db: D1Database): Promise<Island[]> {
50 await ensureContrailReady(db);
51
52 // Load all connections
53 const rows = await db
54 .prepare(
55 `SELECT r.record, r.did, r.rkey, i.handle
56 FROM records_connection r
57 LEFT JOIN identities i ON r.did = i.did
58 ORDER BY r.time_us DESC`,
59 )
60 .all<{ record: string; did: string; rkey: string; handle: string | null }>();
61
62 // Build adjacency list
63 const adj = new Map<string, Set<string>>();
64 const edges: Island["edges"] = [];
65
66 for (const row of rows.results || []) {
67 try {
68 const value = JSON.parse(row.record);
69 const source = value.source as string;
70 const target = value.target as string;
71 if (!source || !target) continue;
72
73 if (!adj.has(source)) adj.set(source, new Set());
74 if (!adj.has(target)) adj.set(target, new Set());
75 adj.get(source)!.add(target);
76 adj.get(target)!.add(source);
77
78 edges.push({
79 uri: `at://${row.did}/network.cosmik.connection/${row.rkey}`,
80 did: row.did,
81 source,
82 target,
83 connectionType: value.connectionType || "relates",
84 note: value.note || "",
85 handle: row.handle,
86 });
87 } catch {}
88 }
89
90 // BFS to find connected components
91 const visited = new Set<string>();
92 const islands: Island[] = [];
93
94 for (const node of adj.keys()) {
95 if (visited.has(node)) continue;
96
97 const component: string[] = [];
98 const queue = [node];
99 while (queue.length > 0) {
100 const current = queue.shift()!;
101 if (visited.has(current)) continue;
102 visited.add(current);
103 component.push(current);
104
105 for (const neighbor of adj.get(current) || []) {
106 if (!visited.has(neighbor)) queue.push(neighbor);
107 }
108 }
109
110 // Only include edges where both endpoints are in this component
111 const componentSet = new Set(component);
112 const componentEdges = edges.filter(
113 e => componentSet.has(e.source) && componentSet.has(e.target),
114 );
115
116 islands.push({
117 id: component.sort().join("|").slice(0, 64), // stable ID from sorted vertices
118 vertices: component,
119 edges: componentEdges,
120 });
121 }
122
123 // Sort by size descending
124 islands.sort((a, b) => b.vertices.length - a.vertices.length);
125 return islands;
126}
127
128// ─── Resolve vertex metadata (batched) ──────────────────────────────
129//
130// Bulk queries instead of N+1 per vertex. Loads all cards and citations
131// matching the given URIs in a few chunked queries.
132
133async function resolveVertexMeta(
134 db: D1Database,
135 uris: string[],
136): Promise<Map<string, { title: string; description: string; type: string }>> {
137 const meta = new Map<string, { title: string; description: string; type: string }>();
138 for (const uri of uris) {
139 meta.set(uri, { title: uri, description: "", type: "unknown" });
140 }
141
142 const httpUris = uris.filter(u => u.startsWith("http"));
143 if (httpUris.length === 0) return meta;
144
145 // Chunk into batches of 50 for IN clauses (D1 limit)
146 const CHUNK = 50;
147
148 // Batch card lookup
149 for (let i = 0; i < httpUris.length; i += CHUNK) {
150 const chunk = httpUris.slice(i, i + CHUNK);
151 const placeholders = chunk.map(() => "?").join(",");
152 const rows = await db
153 .prepare(
154 `SELECT record FROM records_card
155 WHERE json_extract(record, '$.content.url') IN (${placeholders})`,
156 )
157 .bind(...chunk)
158 .all<{ record: string }>();
159
160 for (const row of rows.results || []) {
161 try {
162 const card = JSON.parse(row.record);
163 const url = card?.content?.url;
164 if (!url || !meta.has(url)) continue;
165 const m = card.content.metadata;
166 if (m?.title || m?.description) {
167 meta.set(url, {
168 title: m.title || url,
169 description: m.description || "",
170 type: card.type === "URL" ? "url" : "note",
171 });
172 }
173 } catch {}
174 }
175 }
176
177 // Batch citation lookup (overrides card data if found)
178 for (let i = 0; i < httpUris.length; i += CHUNK) {
179 const chunk = httpUris.slice(i, i + CHUNK);
180 const placeholders = chunk.map(() => "?").join(",");
181 const rows = await db
182 .prepare(
183 `SELECT record FROM records_citation
184 WHERE json_extract(record, '$.url') IN (${placeholders})`,
185 )
186 .bind(...chunk)
187 .all<{ record: string }>();
188
189 for (const row of rows.results || []) {
190 try {
191 const citation = JSON.parse(row.record);
192 const url = citation.url;
193 if (!url || !meta.has(url)) continue;
194 if (citation.title) {
195 const existing = meta.get(url)!;
196 meta.set(url, {
197 title: citation.title,
198 description: citation.takeaway || existing.description,
199 type: "citation",
200 });
201 }
202 } catch {}
203 }
204 }
205
206 return meta;
207}
208
209// ─── Island cache ──────────────────────────────────────────────────
210//
211// The scheduled handler computes islands and writes them to the
212// `island_cache` table. The read path just queries this table
213// instead of recomputing on every request.
214
215async function rebuildIslandCache(db: D1Database): Promise<void> {
216 await ensureContrailReady(db);
217 const islands = await detectIslands(db);
218
219 // Enrich with vertex metadata (batched)
220 const enriched = [];
221 for (const island of islands.slice(0, 50)) {
222 const meta = await resolveVertexMeta(db, island.vertices);
223 enriched.push({
224 ...island,
225 vertexMeta: Object.fromEntries(meta),
226 });
227 }
228
229 const now = Date.now() * 1000;
230
231 // Clear old cache and write new
232 await db.prepare("DELETE FROM island_cache").run();
233
234 const stmts = enriched.map(island =>
235 db
236 .prepare(
237 `INSERT INTO island_cache (id, vertex_count, edge_count, data, updated_at)
238 VALUES (?, ?, ?, ?, ?)`,
239 )
240 .bind(
241 island.id,
242 island.vertices.length,
243 island.edges.length,
244 JSON.stringify(island),
245 now,
246 ),
247 );
248
249 // D1 supports batched statements
250 if (stmts.length > 0) {
251 await db.batch(stmts);
252 }
253}
254
255async function readIslandCache(db: D1Database): Promise<any[]> {
256 const rows = await db
257 .prepare(
258 `SELECT data, summary, strata FROM island_cache ORDER BY vertex_count DESC LIMIT 50`,
259 )
260 .all<{ data: string; summary: string | null; strata: string | null }>();
261
262 return (rows.results || []).map(r => {
263 const island = JSON.parse(r.data);
264 island.summary = r.summary || null;
265 island.strata = r.strata ? JSON.parse(r.strata) : null;
266 return island;
267 });
268}
269
270// ─── Hono app ───────────────────────────────────────────────────────
271
272let app: Hono<{ Bindings: Env }> | null = null;
273
274function buildApp(env: Env): Hono<{ Bindings: Env }> {
275 const app = new Hono<{ Bindings: Env }>();
276 const db = env.DB;
277
278 app.use("*", cors());
279
280 // Landing page — islands feed from cache
281 app.get("/", async (c) => {
282 let islandsJson = "[]";
283 try {
284 const islands = await readIslandCache(db);
285 islandsJson = JSON.stringify(islands);
286 } catch (e) {
287 console.error("Failed to load islands:", e);
288 }
289
290 const page = LANDING_PAGE.replace(
291 "</head>",
292 `<script>window.__ISLANDS__=${islandsJson};</script></head>`,
293 );
294 return c.html(page);
295 });
296
297 // ── Islands API ──────────────────────────────────────────────────
298
299 app.get("/api/strata/islands", async (c) => {
300 const islands = await readIslandCache(db);
301 return c.json({ islands });
302 });
303
304 // Update an island's summary
305 app.put("/api/strata/islands/:id/summary", async (c) => {
306 const id = decodeURIComponent(c.req.param("id"));
307 const body = await c.req.json<{ summary?: string }>().catch(() => ({ summary: undefined }));
308 if (!body.summary) {
309 return c.json({ error: "MissingSummary" }, 400);
310 }
311 await db
312 .prepare("UPDATE island_cache SET summary = ? WHERE id = ?")
313 .bind(body.summary, id)
314 .run();
315 return c.json({ ok: true });
316 });
317
318 // Upload strata analysis for an island
319 app.put("/api/strata/islands/:id/analysis", async (c) => {
320 const id = decodeURIComponent(c.req.param("id"));
321 const body = await c.req.json().catch(() => ({}));
322 if (!body || !body.prose) {
323 return c.json({ error: "MissingRequiredField", message: "prose is required" }, 400);
324 }
325 const strataJson = JSON.stringify(body);
326 await db
327 .prepare("UPDATE island_cache SET strata = ? WHERE id = ?")
328 .bind(strataJson, id)
329 .run();
330 return c.json({ ok: true });
331 });
332
333 // ── R2 sync endpoints ───────────────────────────────────────────
334
335 app.put("/api/sync/vault/:did/*", async (c) => {
336 const did = c.req.param("did");
337 const path = c.req.param("path");
338 if (!did || !path) {
339 return c.json({ error: "Missing path" }, 400);
340 }
341
342 const key = `${did}/${path}`;
343 const body = await c.req.raw.arrayBuffer();
344 await env.VAULT_BUCKET.put(key, body, {
345 httpMetadata: { contentType: c.req.header("content-type") || "text/markdown" },
346 });
347
348 return c.json({ ok: true, key });
349 });
350
351 app.put("/api/sync/carry/:did/*", async (c) => {
352 const did = c.req.param("did");
353 const path = c.req.param("path");
354 if (!did || !path) {
355 return c.json({ error: "Missing path" }, 400);
356 }
357
358 const key = `${did}/${path}`;
359 const body = await c.req.raw.arrayBuffer();
360 await env.CARRY_BUCKET.put(key, body, {
361 httpMetadata: { contentType: c.req.header("content-type") || "application/json" },
362 });
363
364 return c.json({ ok: true, key });
365 });
366
367 // ── Graph neighborhood query ────────────────────────────────────
368
369 app.get("/xrpc/org.latha.strata.connection.getGraph", async (c) => {
370 const uri = c.req.query("uri");
371 if (!uri) {
372 return c.json({ error: "MissingRequiredParameter", message: "uri is required" }, 400);
373 }
374
375 const depth = Math.min(parseInt(c.req.query("depth") || "2"), 3);
376 const limit = Math.min(parseInt(c.req.query("limit") || "50"), 100);
377 const types = c.req.query("types")?.split(",").map((t) => t.trim());
378
379 // Reuse island detection for the subgraph
380 const islands = await detectIslands(db);
381 const island = islands.find(i => i.vertices.includes(uri));
382
383 if (!island) {
384 return c.json({ connections: [], resources: [], depth, uri });
385 }
386
387 // Convert to graph format
388 const resources = island.vertices.map(v => ({
389 uri: v,
390 title: v,
391 type: "unknown",
392 }));
393
394 return c.json({
395 connections: island.edges,
396 resources,
397 depth,
398 uri,
399 });
400 });
401
402 // ── OAuth client metadata ──────────────────────────────────────
403
404 app.get("/oauth-client-metadata.json", (c) => {
405 const host = new URL(c.req.url).host;
406 return c.json({
407 client_id: `https://${host}/oauth-client-metadata.json`,
408 client_name: "Stigmergic",
409 client_uri: `https://${host}`,
410 redirect_uris: [`https://${host}/`],
411 scope: "atproto transition:generic",
412 grant_types: ["authorization_code", "refresh_token"],
413 response_types: ["code"],
414 token_endpoint_auth_method: "none",
415 application_type: "web",
416 dpop_bound_access_tokens: true,
417 });
418 });
419
420 // ── All other routes pass through to contrail ──────────────────
421
422 app.all("*", async (c) => {
423 const response = await contrailWorker.fetch(
424 c.req.raw,
425 c.env as unknown as Record<string, unknown>,
426 );
427 return response;
428 });
429
430 return app;
431}
432
433// ─── Export ─────────────────────────────────────────────────────────
434
435export default {
436 fetch(request: Request, env: Env): Response | Promise<Response> {
437 app ??= buildApp(env);
438 return app.fetch(request, env);
439 },
440 async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
441 // Run Contrail indexing first
442 await contrailWorker.scheduled(
443 event,
444 env as unknown as Record<string, unknown>,
445 ctx,
446 );
447
448 // Rebuild island cache in the background after indexing
449 ctx.waitUntil(rebuildIslandCache(env.DB));
450 },
451};
452