nandi.uk
This repository has no description
49 kB
1389 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";
8import { StrataContainer } from "./container.js";
9
10// ─── Env ────────────────────────────────────────────────────────────
11
12interface Env {
13 DB: D1Database;
14 VAULT_BUCKET: R2Bucket;
15 CARRY_BUCKET: R2Bucket;
16 STRATA_CONTAINER: DurableObjectNamespace;
17 PDS_APP_PASSWORD: string;
18 CARRY_GITHUB_TOKEN: string;
19 OBSIDIAN_AUTH_TOKEN: string;
20 OBSIDIAN_ENCRYPTION_KEY: string;
21 OBSIDIAN_ENCRYPTION_SALT: string;
22}
23
24// ─── Contrail setup ─────────────────────────────────────────────────
25
26const contrailWorker = createWorker(config, { lexicons });
27const contrail = new Contrail(config);
28let contrailReady = false;
29
30async function ensureContrailReady(db: D1Database): Promise<void> {
31 if (contrailReady) return;
32 await contrail.init(db);
33 contrailReady = true;
34}
35
36// ─── Island detection (connected components) ────────────────────────
37//
38// Load all connections, build an undirected graph, find connected
39// components via BFS. Each component is an "island" — a cluster of
40// linked URLs that form a constellation in the knowledge graph.
41
42interface Island {
43 id: string;
44 centroid: string;
45 vertices: string[];
46 edges: Array<{
47 uri: string;
48 did: string;
49 source: string;
50 target: string;
51 connectionType: string;
52 note: string;
53 handle?: string | null;
54 }>;
55}
56
57/** Compute centroid: vertex with highest closeness centrality (smallest avg BFS distance). */
58function computeCentroid(adj: Map<string, Set<string>>, nodes: Set<string>): string {
59 if (nodes.size <= 1) return [...nodes][0] ?? '';
60
61 let bestNode = '';
62 let bestAvg = Infinity;
63
64 for (const start of nodes) {
65 const dist = new Map<string, number>();
66 dist.set(start, 0);
67 const queue = [start];
68 while (queue.length > 0) {
69 const current = queue.shift()!;
70 for (const neighbor of adj.get(current) || []) {
71 if (!dist.has(neighbor) && nodes.has(neighbor)) {
72 dist.set(neighbor, dist.get(current)! + 1);
73 queue.push(neighbor);
74 }
75 }
76 }
77
78 let totalDist = 0;
79 let reachable = 0;
80 for (const [node, d] of dist) {
81 if (nodes.has(node)) {
82 totalDist += d;
83 reachable++;
84 }
85 }
86
87 const avg = reachable > 0 ? totalDist / reachable : Infinity;
88 if (avg < bestAvg) {
89 bestAvg = avg;
90 bestNode = start;
91 }
92 }
93
94 return bestNode;
95}
96
97/** Hash a URL to a 12-char hex island ID (matches CLI's islandId()). */
98async function islandHash(url: string): Promise<string> {
99 const hash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(url));
100 return Array.from(new Uint8Array(hash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 12);
101}
102
103async function detectIslands(db: D1Database): Promise<Island[]> {
104 await ensureContrailReady(db);
105
106 // Load all connections
107 const rows = await db
108 .prepare(
109 `SELECT r.record, r.did, r.rkey, i.handle
110 FROM records_connection r
111 LEFT JOIN identities i ON r.did = i.did
112 ORDER BY r.time_us DESC`,
113 )
114 .all<{ record: string; did: string; rkey: string; handle: string | null }>();
115
116 // Build adjacency list
117 const adj = new Map<string, Set<string>>();
118 const edges: Island["edges"] = [];
119
120 for (const row of rows.results || []) {
121 try {
122 const value = JSON.parse(row.record);
123 const source = value.source as string;
124 const target = value.target as string;
125 if (!source || !target) continue;
126
127 if (!adj.has(source)) adj.set(source, new Set());
128 if (!adj.has(target)) adj.set(target, new Set());
129 adj.get(source)!.add(target);
130 adj.get(target)!.add(source);
131
132 edges.push({
133 uri: `at://${row.did}/network.cosmik.connection/${row.rkey}`,
134 did: row.did,
135 source,
136 target,
137 connectionType: value.connectionType || "relates",
138 note: value.note || "",
139 handle: row.handle,
140 });
141 } catch {}
142 }
143
144 // BFS to find connected components
145 const visited = new Set<string>();
146 const islands: Island[] = [];
147
148 for (const node of adj.keys()) {
149 if (visited.has(node)) continue;
150
151 const component: string[] = [];
152 const queue = [node];
153 while (queue.length > 0) {
154 const current = queue.shift()!;
155 if (visited.has(current)) continue;
156 visited.add(current);
157 component.push(current);
158
159 for (const neighbor of adj.get(current) || []) {
160 if (!visited.has(neighbor)) queue.push(neighbor);
161 }
162 }
163
164 // Only include edges where both endpoints are in this component
165 const componentSet = new Set(component);
166 const componentEdges = edges.filter(
167 e => componentSet.has(e.source) && componentSet.has(e.target),
168 );
169
170 // Stable ID: truncated SHA-256 of the centroid vertex (highest closeness centrality)
171 const centroid = computeCentroid(adj, componentSet);
172 const id = await islandHash(centroid);
173
174 islands.push({
175 id,
176 centroid,
177 vertices: component,
178 edges: componentEdges,
179 });
180 }
181
182 // Sort by size descending
183 islands.sort((a, b) => b.vertices.length - a.vertices.length);
184 return islands;
185}
186
187// ─── Resolve vertex metadata (batched) ──────────────────────────────
188//
189// Bulk queries instead of N+1 per vertex. Loads all cards and citations
190// matching the given URIs in a few chunked queries.
191
192async function resolveVertexMeta(
193 db: D1Database,
194 uris: string[],
195): Promise<Map<string, { title: string; description: string; type: string }>> {
196 const meta = new Map<string, { title: string; description: string; type: string }>();
197 for (const uri of uris) {
198 meta.set(uri, { title: uri, description: "", type: "unknown" });
199 }
200
201 const httpUris = uris.filter(u => u.startsWith("http"));
202 const atUris = uris.filter(u => u.startsWith("at://"));
203 const CHUNK = 50;
204
205 // Build all queries upfront, then batch-execute in one D1 round trip
206 const stmts: D1PreparedStatement[] = [];
207
208 // HTTP URI queries — cards
209 const httpChunks: string[][] = [];
210 for (let i = 0; i < httpUris.length; i += CHUNK) {
211 const chunk = httpUris.slice(i, i + CHUNK);
212 httpChunks.push(chunk);
213 const placeholders = chunk.map(() => "?").join(",");
214 stmts.push(
215 db.prepare(
216 `SELECT record FROM records_card WHERE json_extract(record, '$.content.url') IN (${placeholders})`
217 ).bind(...chunk)
218 );
219 }
220
221 // HTTP URI queries — citations
222 for (let i = 0; i < httpUris.length; i += CHUNK) {
223 const chunk = httpUris.slice(i, i + CHUNK);
224 const placeholders = chunk.map(() => "?").join(",");
225 stmts.push(
226 db.prepare(
227 `SELECT record FROM records_citation WHERE json_extract(record, '$.url') IN (${placeholders})`
228 ).bind(...chunk)
229 );
230 }
231
232 // AT URI queries — cards
233 const atChunks: string[][] = [];
234 for (let i = 0; i < atUris.length; i += CHUNK) {
235 const chunk = atUris.slice(i, i + CHUNK);
236 atChunks.push(chunk);
237 const placeholders = chunk.map(() => "?").join(",");
238 stmts.push(
239 db.prepare(`SELECT uri, record FROM records_card WHERE uri IN (${placeholders})`).bind(...chunk)
240 );
241 }
242
243 // AT URI queries — collections
244 for (let i = 0; i < atUris.length; i += CHUNK) {
245 const chunk = atUris.slice(i, i + CHUNK);
246 const placeholders = chunk.map(() => "?").join(",");
247 stmts.push(
248 db.prepare(`SELECT uri, record FROM records_collection WHERE uri IN (${placeholders})`).bind(...chunk)
249 );
250 }
251
252 // Execute all queries in one batch
253 const results = stmts.length > 0 ? await db.batch(stmts) : [];
254 let resultIdx = 0;
255
256 // Process card results (HTTP)
257 for (const chunk of httpChunks) {
258 const rows = (results[resultIdx++] as { results: { record: string }[] } | null)?.results || [];
259 for (const row of rows) {
260 try {
261 const card = JSON.parse(row.record);
262 const url = card?.content?.url;
263 if (!url || !meta.has(url)) continue;
264 const m = card.content.metadata;
265 if (m?.title || m?.description) {
266 meta.set(url, { title: m.title || url, description: m.description || "", type: card.type === "URL" ? "url" : "note" });
267 }
268 } catch {}
269 }
270 }
271
272 // Process citation results (HTTP) — overrides card data
273 for (let i = 0; i < httpChunks.length; i++) {
274 const rows = (results[resultIdx++] as { results: { record: string }[] } | null)?.results || [];
275 for (const row of rows) {
276 try {
277 const citation = JSON.parse(row.record);
278 const url = citation.url;
279 if (!url || !meta.has(url)) continue;
280 if (citation.title) {
281 const existing = meta.get(url)!;
282 meta.set(url, { title: citation.title, description: citation.takeaway || existing.description, type: "citation" });
283 }
284 } catch {}
285 }
286 }
287
288 // Process card results (AT)
289 for (const chunk of atChunks) {
290 const rows = (results[resultIdx++] as { results: { uri: string; record: string }[] } | null)?.results || [];
291 for (const row of rows) {
292 try {
293 const card = JSON.parse(row.record);
294 const m = card?.content?.metadata;
295 const url = card?.content?.url;
296 if (meta.has(row.uri)) {
297 meta.set(row.uri, { title: m?.title || url || row.uri, description: m?.description || "", type: "card" });
298 }
299 } catch {}
300 }
301 }
302
303 // Process collection results (AT)
304 for (let i = 0; i < atChunks.length; i++) {
305 const rows = (results[resultIdx++] as { results: { uri: string; record: string }[] } | null)?.results || [];
306 for (const row of rows) {
307 try {
308 const coll = JSON.parse(row.record);
309 if (meta.has(row.uri)) {
310 meta.set(row.uri, { title: coll.name || row.uri, description: coll.description || "", type: "collection" });
311 }
312 } catch {}
313 }
314 }
315
316 return meta;
317}
318
319// ─── Derive island from centroid ────────────────────────────────────
320//
321// Given a centroid vertex (canonical component reference), derive the
322// connected component by traversing network.cosmik.connection records.
323
324async function deriveIsland(
325 db: D1Database,
326 seed: string,
327): Promise<Island | null> {
328 await ensureContrailReady(db);
329
330 // BFS in application code — recursive CTE OOMs in D1 due to cross join
331 const visited = new Set<string>();
332 const queue = [seed];
333 const allEdges: Array<{
334 uri: string; did: string; rkey: string;
335 source: string; target: string; connectionType: string;
336 note: string; handle: string | null;
337 }> = [];
338
339 while (queue.length > 0) {
340 // Drain the current queue into a batch
341 const batch = queue.splice(0, 50);
342 const toQuery = batch.filter(v => !visited.has(v));
343 if (toQuery.length === 0) continue;
344
345 // Mark visited
346 for (const v of toQuery) visited.add(v);
347
348 // Query connections where any of these vertices are source or target
349 const placeholders = toQuery.map(() => "?").join(",");
350 const rows = await db
351 .prepare(
352 `SELECT r.record, r.did, r.rkey, i.handle
353 FROM records_connection r
354 LEFT JOIN identities i ON r.did = i.did
355 WHERE json_extract(r.record, '$.source') IN (${placeholders})
356 OR json_extract(r.record, '$.target') IN (${placeholders})`,
357 )
358 .bind(...toQuery, ...toQuery)
359 .all<{ record: string; did: string; rkey: string; handle: string | null }>();
360
361 for (const row of rows.results || []) {
362 try {
363 const value = JSON.parse(row.record);
364 const source = value.source as string;
365 const target = value.target as string;
366 if (!source || !target) continue;
367
368 // Collect edge
369 const edgeKey = `${source}|${target}|${row.rkey}`;
370 allEdges.push({
371 uri: `at://${row.did}/network.cosmik.connection/${row.rkey}`,
372 did: row.did,
373 rkey: row.rkey,
374 source,
375 target,
376 connectionType: value.connectionType || "relates",
377 note: value.note || "",
378 handle: row.handle,
379 });
380
381 // Enqueue unvisited neighbors
382 if (!visited.has(source)) queue.push(source);
383 if (!visited.has(target)) queue.push(target);
384 } catch {}
385 }
386 }
387
388 if (visited.size === 0) return null;
389
390 const vertices = [...visited];
391
392 // Filter edges to only those with both endpoints in the component
393 const vertexSet = new Set(vertices);
394 const edges: Island["edges"] = [];
395 const seenEdges = new Set<string>();
396 for (const e of allEdges) {
397 if (!vertexSet.has(e.source) || !vertexSet.has(e.target)) continue;
398 const edgeKey = `${e.source}|${e.target}|${e.rkey}`;
399 if (seenEdges.has(edgeKey)) continue;
400 seenEdges.add(edgeKey);
401 edges.push({
402 uri: e.uri,
403 did: e.did,
404 source: e.source,
405 target: e.target,
406 connectionType: e.connectionType,
407 note: e.note,
408 handle: e.handle,
409 });
410 }
411
412 // Compute stable ID from centroid (highest closeness centrality)
413 const adj = new Map<string, Set<string>>();
414 for (const e of edges) {
415 if (!adj.has(e.source)) adj.set(e.source, new Set());
416 if (!adj.has(e.target)) adj.set(e.target, new Set());
417 adj.get(e.source)!.add(e.target);
418 adj.get(e.target)!.add(e.source);
419 }
420 const centroid = computeCentroid(adj, vertexSet);
421 const id = await islandHash(centroid);
422
423 return { id, centroid, vertices, edges };
424}
425
426// ─── List strata islands from records_strata ──────────────────────
427
428// ─── Resolve edge URIs to full edge objects ──────────────────────
429
430/** Given a list of connections (either full objects or just {uri}), resolve URIs to full edges from D1. */
431async function resolveEdgeUris(db: D1Database, connections: any[]): Promise<any[]> {
432 const resolved: any[] = [];
433 const uriIndices: Map<string, number[]> = new Map(); // uri → indices in connections that need it
434
435 for (let i = 0; i < connections.length; i++) {
436 const conn = connections[i];
437 // Already has source/target — use as-is
438 if (typeof conn === "object" && conn.source && conn.target) {
439 resolved.push(conn);
440 continue;
441 }
442 // Just a URI reference — needs lookup
443 const uri = conn.uri;
444 if (!uri) continue;
445 resolved.push(null); // placeholder
446 if (!uriIndices.has(uri)) uriIndices.set(uri, []);
447 uriIndices.get(uri)!.push(i);
448 }
449
450 if (uriIndices.size === 0) return resolved.filter(Boolean);
451
452 // Batch-fetch connection records from D1
453 const uris = [...uriIndices.keys()];
454 const placeholders = uris.map(() => "?").join(",");
455 const rows = await db
456 .prepare(
457 `SELECT r.uri, r.record, r.did, r.rkey, i.handle
458 FROM records_connection r
459 LEFT JOIN identities i ON r.did = i.did
460 WHERE r.uri IN (${placeholders})`,
461 )
462 .bind(...uris)
463 .all<{ uri: string; record: string; did: string; rkey: string; handle: string | null }>();
464
465 for (const row of rows.results || []) {
466 try {
467 const value = JSON.parse(row.record);
468 const edge = {
469 uri: row.uri,
470 did: row.did,
471 source: value.source,
472 target: value.target,
473 connectionType: value.connectionType || "relates",
474 note: value.note || "",
475 handle: row.handle,
476 };
477 for (const idx of uriIndices.get(row.uri) || []) {
478 resolved[idx] = edge;
479 }
480 } catch {}
481 }
482
483 return resolved.filter(Boolean);
484}
485
486// ─── Hono app ───────────────────────────────────────────────────────
487
488let app: Hono<{ Bindings: Env }> | null = null;
489
490function buildApp(env: Env): Hono<{ Bindings: Env }> {
491 const app = new Hono<{ Bindings: Env }>();
492 const db = env.DB;
493
494 app.use("*", cors());
495
496 // Landing page — strata records feed
497 app.get("/", async (c) => {
498 // Load only the fields we need — records are 100KB+ with embedded connections
499 // No ensureContrailReady needed — just D1
500 let islandsJson = "[]";
501 try {
502 const rows = await db
503 .prepare("SELECT uri, json_extract(record, '$.source.uri') as centroid, json_extract(record, '$.analysis.title') as title FROM records_strata ORDER BY time_us DESC LIMIT 10")
504 .all<{ uri: string; centroid: string | null; title: string | null }>();
505
506 const islands: any[] = [];
507 for (const row of rows.results || []) {
508 if (!row.centroid) continue;
509 const id = row.uri.split("/").pop() || row.uri;
510 islands.push({ id, centroid: row.centroid, title: row.title || null, recordUri: row.uri });
511 }
512 islandsJson = JSON.stringify(islands);
513 } catch (e) {
514 console.error("Failed to load islands:", e);
515 }
516
517 const page = LANDING_PAGE.replace(
518 "</head>",
519 `<script>window.__ISLANDS__=${islandsJson};</script></head>`,
520 );
521 return c.html(page);
522 });
523
524 // ── Islands API ──────────────────────────────────────────────────
525
526 app.get("/xrpc/org.latha.island.getIslands", async (c) => {
527 const limit = Math.min(parseInt(c.req.query("limit") || "50"), 100);
528 const summary = c.req.query("summary") === "true";
529 // No ensureContrailReady — just D1 queries
530
531 const rows = await db
532 .prepare("SELECT uri, record FROM records_strata ORDER BY time_us DESC LIMIT ?")
533 .bind(limit)
534 .all<{ uri: string; record: string }>();
535
536 const islands: any[] = [];
537 for (const row of rows.results || []) {
538 try {
539 const rec = JSON.parse(row.record);
540 const centroid = rec.source?.uri;
541 if (!centroid) continue;
542
543 // ID is the record rkey — unique per strata record even if same graph
544 const id = row.uri.split("/").pop() || row.uri;
545
546 // Build graph from connections — resolve URIs if edges are just {uri} references
547 const connections: any[] = rec.connections || [];
548 const resolvedEdges = await resolveEdgeUris(db, connections);
549 const vertexSet = new Set<string>();
550 const edges: any[] = [];
551
552 for (const conn of resolvedEdges) {
553 if (conn.source && conn.target) {
554 vertexSet.add(conn.source);
555 vertexSet.add(conn.target);
556 edges.push(conn);
557 }
558 }
559
560 const vertices = [...vertexSet].sort();
561 const analysis = rec.analysis || {};
562
563 // Summary mode: minimal payload for explore page canvas
564 // Skip vertexMeta (122KB) and trim edges to just source/target
565 const vertexMeta = summary ? {} : Object.fromEntries(await resolveVertexMeta(db, vertices));
566 const trimmedEdges = summary
567 ? edges.map(e => ({ source: e.source, target: e.target, connectionType: e.connectionType }))
568 : edges;
569
570 islands.push({
571 id, centroid, vertices, edges: trimmedEdges, vertexMeta,
572 title: analysis.title || null,
573 strata: analysis.synthesis ? {
574 prose: analysis.synthesis || "", title: analysis.title || "",
575 themes: analysis.themes || [],
576 relationships: [],
577 highlights: analysis.highlights || [],
578 tensions: analysis.tensions || [],
579 open_questions: analysis.openQuestions || [],
580 synthesis: analysis.synthesis || "",
581 } : null,
582 recordUri: row.uri,
583 });
584 } catch {}
585 }
586
587 return c.json({ islands });
588 });
589
590 // ── Get single island by ID or AT URI ────────────────────────────
591
592 app.get("/xrpc/org.latha.island.getIsland", async (c) => {
593 const id = c.req.query("id");
594 if (!id) {
595 return c.json({ error: "MissingRequiredParameter", message: "id is required" }, 400);
596 }
597
598 await ensureContrailReady(db);
599
600 // If it's an AT URI, look up the strata record and derive from its centroid
601 if (id.startsWith("at://")) {
602 const row = await db
603 .prepare("SELECT uri, record FROM records_strata WHERE uri = ?")
604 .bind(id)
605 .first<{ uri: string; record: string }>();
606
607 if (!row) {
608 return c.json({ error: "RecordNotFound" }, 404);
609 }
610
611 const rec = JSON.parse(row.record);
612 const centroid = rec.source?.uri;
613 if (!centroid) {
614 return c.json({ error: "InvalidRecord" }, 400);
615 }
616
617 const island = await deriveIsland(db, centroid);
618 if (!island) {
619 return c.json({ error: "IslandNotFound" }, 404);
620 }
621
622 const meta = await resolveVertexMeta(db, island.vertices);
623 const analysis = rec.analysis || {};
624
625 return c.json({
626 island: {
627 ...island,
628 centroid: island.centroid,
629 vertexMeta: Object.fromEntries(meta),
630 summary: analysis.title || null,
631 strata: {
632 prose: analysis.synthesis || "",
633 title: analysis.title || "",
634 themes: analysis.themes || [],
635 relationships: [],
636 highlights: analysis.highlights || [],
637 tensions: analysis.tensions || [],
638 open_questions: analysis.openQuestions || [],
639 synthesis: analysis.synthesis || "",
640 },
641 },
642 recordUri: row.uri,
643 });
644 }
645
646 // ID is the record rkey — look up directly by URI
647 const recordUri = `at://did:plc:ngokl2gnmpbvuvrfckja3g7p/org.latha.island/${id}`;
648 const row = await db
649 .prepare("SELECT uri, record FROM records_strata WHERE uri = ?")
650 .bind(recordUri)
651 .first<{ uri: string; record: string }>();
652
653 if (row) {
654 try {
655 const rec = JSON.parse(row.record);
656 const centroid = rec.source?.uri;
657
658 // Build graph from connections — resolve URIs if edges are just {uri} references
659 const connections: any[] = rec.connections || [];
660 const resolvedEdges = await resolveEdgeUris(db, connections);
661 const vertexSet = new Set<string>();
662 const edges: any[] = [];
663 for (const conn of resolvedEdges) {
664 if (conn.source && conn.target) {
665 vertexSet.add(conn.source);
666 vertexSet.add(conn.target);
667 edges.push(conn);
668 }
669 }
670 const vertices = [...vertexSet].sort();
671 const meta = await resolveVertexMeta(db, vertices);
672 const analysis = rec.analysis || {};
673
674 return c.json({
675 island: {
676 id,
677 vertices,
678 edges,
679 centroid: centroid || vertices[0],
680 vertexMeta: Object.fromEntries(meta),
681 summary: analysis.title || null,
682 strata: {
683 prose: analysis.synthesis || "",
684 title: analysis.title || "",
685 themes: analysis.themes || [],
686 relationships: [],
687 highlights: analysis.highlights || [],
688 tensions: analysis.tensions || [],
689 open_questions: analysis.openQuestions || [],
690 synthesis: analysis.synthesis || "",
691 },
692 recordUri: row.uri,
693 },
694 });
695 } catch {}
696 }
697
698 // No strata record found — try deriving island by treating the ID as a centroid seed
699 // If the ID is a 12-char hex string (centroid hash), scan strata records for a match
700 if (/^[0-9a-f]{12}$/.test(id)) {
701 const rows = await db
702 .prepare("SELECT uri, record FROM records_strata ORDER BY time_us DESC LIMIT 500")
703 .all<{ uri: string; record: string }>();
704 for (const row of rows.results || []) {
705 try {
706 const rec = JSON.parse(row.record);
707 const centroid = rec.source?.uri;
708 if (!centroid) continue;
709 const hash = await islandHash(centroid);
710 if (hash === id) {
711 // Found the record — redirect to the AT URI path
712 const connections: any[] = rec.connections || [];
713 const resolvedEdges = await resolveEdgeUris(db, connections);
714 const vertexSet = new Set<string>();
715 const edges: any[] = [];
716 for (const conn of resolvedEdges) {
717 if (conn.source && conn.target) {
718 vertexSet.add(conn.source);
719 vertexSet.add(conn.target);
720 edges.push(conn);
721 }
722 }
723 const vertices = [...vertexSet].sort();
724 const meta = await resolveVertexMeta(db, vertices);
725 const analysis = rec.analysis || {};
726 return c.json({
727 island: {
728 id,
729 vertices,
730 edges,
731 centroid,
732 vertexMeta: Object.fromEntries(meta),
733 summary: analysis.title || null,
734 strata: {
735 prose: analysis.synthesis || "",
736 title: analysis.title || "",
737 themes: analysis.themes || [],
738 relationships: [],
739 highlights: analysis.highlights || [],
740 tensions: analysis.tensions || [],
741 open_questions: analysis.openQuestions || [],
742 synthesis: analysis.synthesis || "",
743 },
744 },
745 recordUri: row.uri,
746 });
747 }
748 } catch {}
749 }
750 }
751
752 // Fall back: treat ID as a vertex URL for BFS derivation
753 const island = await deriveIsland(db, id);
754 if (!island) {
755 return c.json({ error: "IslandNotFound" }, 404);
756 }
757
758 const meta = await resolveVertexMeta(db, island.vertices);
759 return c.json({
760 island: {
761 ...island,
762 centroid: island.centroid,
763 vertexMeta: Object.fromEntries(meta),
764 summary: null,
765 strata: null,
766 },
767 recordUri: null,
768 });
769 });
770
771 // ── List strata records (AT URI-based) ──────────────────────────
772
773 app.get("/xrpc/org.latha.island.listRecords", async (c) => {
774 const rows = await db
775 .prepare("SELECT uri, record FROM records_strata ORDER BY time_us DESC LIMIT 50")
776 .all<{ uri: string; record: string }>();
777
778 const records = (rows.results || []).map(r => {
779 try {
780 const rec = JSON.parse(r.record);
781 return {
782 uri: r.uri,
783 source: rec.source?.uri || "",
784 title: rec.analysis?.title || "",
785 themes: rec.analysis?.themes || [],
786 createdAt: rec.createdAt || "",
787 };
788 } catch {
789 return null;
790 }
791 }).filter(Boolean);
792
793 return c.json({ records });
794 });
795
796 // ── Strata record endpoint (AT URI-based) ──────────────────────
797
798 app.get("/xrpc/org.latha.island.getRecord", async (c) => {
799 const uri = c.req.query("uri");
800 if (!uri) {
801 return c.json({ error: "MissingRequiredParameter", message: "uri is required" }, 400);
802 }
803
804 // Fetch the strata record from D1 (indexed by Contrail)
805 const row = await db
806 .prepare("SELECT record FROM records_strata WHERE uri = ?")
807 .bind(uri)
808 .first<{ record: string }>();
809
810 if (!row) {
811 return c.json({ error: "NotFound", message: "Strata record not found" }, 404);
812 }
813
814 let strataRecord: any;
815 try {
816 strataRecord = JSON.parse(row.record);
817 } catch {
818 return c.json({ error: "InvalidRecord" }, 500);
819 }
820
821 // Derive island from centroid (source.uri)
822 const centroid = strataRecord.source?.uri;
823 if (!centroid) {
824 return c.json({ error: "MissingLexmin", message: "Strata record has no source.uri" }, 400);
825 }
826
827 const island = await deriveIsland(db, centroid);
828 if (!island) {
829 return c.json({ error: "IslandNotFound", message: "Could not derive island from centroid" }, 404);
830 }
831
832 // Resolve vertex metadata
833 const meta = await resolveVertexMeta(db, island.vertices);
834
835 // Build the response in the same format as the island cache
836 const analysis = strataRecord.analysis || {};
837 const strataData = {
838 prose: analysis.synthesis || "",
839 title: analysis.title || "",
840 themes: analysis.themes || [],
841 relationships: [],
842 highlights: analysis.highlights || [],
843 tensions: analysis.tensions || [],
844 open_questions: analysis.openQuestions || [],
845 synthesis: analysis.synthesis || "",
846 };
847
848 return c.json({
849 island: {
850 ...island,
851 vertexMeta: Object.fromEntries(meta),
852 summary: analysis.title || null,
853 strata: strataData,
854 },
855 record: strataRecord,
856 recordUri: uri,
857 });
858 });
859
860 // ── Container-backed xrpc methods ────────────────────────────────
861
862 // Helper: read carry data from D1 for analysis
863 async function getCarryData(db: D1Database, did?: string): Promise<{
864 entities: Array<{ atUri: string; name: string; entityType: string; description?: string; url?: string }>;
865 citations: Array<{ atUri: string; url: string; title: string; takeaway: string; confidence?: string }>;
866 reasonings: Array<{ atUri: string; claim: string; evidence: string; reasoningType?: string; url?: string }>;
867 connections: Array<{ atUri: string; source: string; target: string; relation: string; sourceUrl: string; targetUrl: string; context: string }>;
868 }> {
869 const entities: Array<{ atUri: string; name: string; entityType: string; description?: string; url?: string }> = [];
870 const citations: Array<{ atUri: string; url: string; title: string; takeaway: string; confidence?: string }> = [];
871 const reasonings: Array<{ atUri: string; claim: string; evidence: string; reasoningType?: string; url?: string }> = [];
872 const connections: Array<{ atUri: string; source: string; target: string; relation: string; sourceUrl: string; targetUrl: string; context: string }> = [];
873
874 try {
875 // Read entities
876 const entityRows = await db
877 .prepare("SELECT uri, record FROM records_entity ORDER BY time_us DESC LIMIT 100")
878 .all<{ uri: string; record: string }>();
879 for (const row of entityRows.results || []) {
880 try {
881 const r = JSON.parse(row.record);
882 entities.push({
883 atUri: row.uri,
884 name: r.name || "",
885 entityType: r.entityType || "",
886 description: r.description || "",
887 url: r.url || "",
888 });
889 } catch {}
890 }
891
892 // Read citations
893 const citeRows = await db
894 .prepare("SELECT uri, record FROM records_citation ORDER BY time_us DESC LIMIT 100")
895 .all<{ uri: string; record: string }>();
896 for (const row of citeRows.results || []) {
897 try {
898 const r = JSON.parse(row.record);
899 citations.push({
900 atUri: row.uri,
901 url: r.url || "",
902 title: r.title || "",
903 takeaway: r.takeaway || "",
904 confidence: r.confidence || "",
905 });
906 } catch {}
907 }
908
909 // Read reasonings
910 const reasonRows = await db
911 .prepare("SELECT uri, record FROM records_reasoning ORDER BY time_us DESC LIMIT 100")
912 .all<{ uri: string; record: string }>();
913 for (const row of reasonRows.results || []) {
914 try {
915 const r = JSON.parse(row.record);
916 reasonings.push({
917 atUri: row.uri,
918 claim: r.claim || "",
919 evidence: r.evidence || "",
920 reasoningType: r.reasoningType || "",
921 url: r.url || "",
922 });
923 } catch {}
924 }
925 } catch (e: any) {
926 console.error("getCarryData error:", e.message);
927 }
928
929 // Read connections from D1
930 try {
931 const connRows = await db
932 .prepare("SELECT uri, record FROM records_connection ORDER BY time_us DESC LIMIT 100")
933 .all<{ uri: string; record: string }>();
934 for (const row of connRows.results || []) {
935 try {
936 const r = JSON.parse(row.record);
937 connections.push({
938 atUri: row.uri,
939 source: r.source || "",
940 target: r.target || "",
941 relation: r.relation || "",
942 sourceUrl: r.sourceUrl || "",
943 targetUrl: r.targetUrl || "",
944 context: r.context || "",
945 });
946 } catch {}
947 }
948 } catch (e: any) {
949 console.error("getCarryData connections error:", e.message);
950 }
951
952 return { entities, citations, reasonings, connections };
953 }
954
955 // Helper: call the strata container
956 async function callContainer(path: string, method: string, body?: any): Promise<any> {
957 const id = env.STRATA_CONTAINER.idFromName("strata");
958 const stub = env.STRATA_CONTAINER.get(id);
959 const url = new URL(path, "http://container");
960 const init: RequestInit = { method };
961 if (body) {
962 init.body = JSON.stringify(body);
963 init.headers = { "Content-Type": "application/json" };
964 }
965 const resp = await stub.fetch(new Request(url.toString(), init));
966 const text = await resp.text();
967 try {
968 return JSON.parse(text);
969 } catch {
970 throw new Error(`Container returned non-JSON: ${text.slice(0, 500)}`);
971 }
972 }
973
974 // Sync container data (vault + carry)
975 app.post("/xrpc/org.latha.island.syncContainer", async (c) => {
976 try {
977 const result = await callContainer("/sync", "POST");
978 return c.json(result);
979 } catch (e: any) {
980 return c.json({ error: "ContainerError", message: e.message }, 500);
981 }
982 });
983
984 // Notify appview of a newly published record — crawl the DID's PDS and upsert into D1
985 app.post("/xrpc/org.latha.island.notifyOfUpdate", async (c) => {
986 const body = await c.req.json().catch(() => ({})) as { did?: string; uri?: string };
987 const did = body.did;
988 if (!did) return c.json({ error: "InvalidRequest", message: "did is required" }, 400);
989
990 await ensureContrailReady(db);
991
992 // Look up the DID's PDS
993 const identity = await db
994 .prepare("SELECT pds FROM identities WHERE did = ?")
995 .bind(did)
996 .first<{ pds: string }>();
997
998 const pds = identity?.pds;
999 if (!pds) return c.json({ error: "IdentityNotFound", message: `DID ${did} not in identities table` }, 404);
1000
1001 // Fetch org.latha.island records from the DID's PDS
1002 const url = `${pds}/xrpc/com.atproto.repo.listRecords?repo=${did}&collection=org.latha.island&limit=50`;
1003 const resp = await fetch(url);
1004 if (!resp.ok) return c.json({ error: "PdsError", message: `PDS returned ${resp.status}` }, 502);
1005
1006 const data = await resp.json() as { records: Array<{ uri: string; cid: string; value: Record<string, unknown> }> };
1007 const records = data.records || [];
1008 if (records.length === 0) return c.json({ ingested: 0 });
1009
1010 // Upsert each record into records_strata
1011 const now = Date.now() * 1000; // time_us
1012 const indexedAt = Date.now();
1013 const stmt = db.prepare(
1014 `INSERT INTO records_strata (uri, did, rkey, cid, record, time_us, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(uri) DO UPDATE SET cid = excluded.cid, record = excluded.record, time_us = excluded.time_us, indexed_at = excluded.indexed_at`
1015 );
1016
1017 let ingested = 0;
1018 const batch: D1PreparedStatement[] = [];
1019 for (const r of records) {
1020 const rkey = r.uri.split("/").pop() || "";
1021 batch.push(stmt.bind(r.uri, did, rkey, r.cid || "", JSON.stringify(r.value), now + ingested, indexedAt));
1022 ingested++;
1023 }
1024
1025 await db.batch(batch);
1026 return c.json({ ingested });
1027 });
1028
1029 // Container health + data status
1030 app.get("/xrpc/org.latha.island.containerHealth", async (c) => {
1031 try {
1032 const result = await callContainer("/health", "GET");
1033 return c.json(result);
1034 } catch (e: any) {
1035 return c.json({ error: "ContainerError", message: e.message }, 500);
1036 }
1037 });
1038
1039 // Derive islands — POST, delegates to container
1040 // With seed: single island. Without: all islands.
1041 app.post("/xrpc/org.latha.island.deriveIsland", async (c) => {
1042 const body = await c.req.json().catch(() => ({})) as { seed?: string };
1043 const seed = body.seed;
1044
1045 // Load all connections from D1 for the container to traverse
1046 await ensureContrailReady(db);
1047 const rows = await db
1048 .prepare(
1049 `SELECT r.record, r.did, r.rkey, i.handle
1050 FROM records_connection r
1051 LEFT JOIN identities i ON r.did = i.did`,
1052 )
1053 .all<{ record: string; did: string; rkey: string; handle: string | null }>();
1054
1055 const connections = (rows.results || []).map(r => {
1056 try {
1057 const value = JSON.parse(r.record);
1058 return {
1059 source: value.source,
1060 target: value.target,
1061 connectionType: value.connectionType || "relates",
1062 note: value.note || "",
1063 did: r.did,
1064 rkey: r.rkey,
1065 };
1066 } catch {
1067 return null;
1068 }
1069 }).filter(Boolean);
1070
1071 try {
1072 const result = await callContainer("/deriveIsland", "POST", { connections, seed: seed || undefined });
1073 return c.json(result);
1074 } catch (e: any) {
1075 return c.json({ error: "ContainerError", message: e.message }, 500);
1076 }
1077 });
1078
1079 // Run strata analysis — delegates to container
1080 app.post("/xrpc/org.latha.island.analyze", async (c) => {
1081 const body = await c.req.json().catch(() => ({}));
1082 if (!body.island) {
1083 return c.json({ error: "MissingRequiredField", message: "island is required" }, 400);
1084 }
1085
1086 try {
1087 // Read carry data from D1 for the user's DID
1088 const carryData = await getCarryData(db, body.did);
1089
1090 // Pass carry data to container alongside island
1091 const payload = { ...body, carryData };
1092 const result = await callContainer("/analyze", "POST", payload);
1093
1094 // Create new connection records on PDS
1095 const newConnections: Array<{ source: string; target: string; connectionType: string; note: string }> = result.newConnections || [];
1096 const createdUris: string[] = [];
1097
1098 if (newConnections.length > 0) {
1099 const appPassword = env.PDS_APP_PASSWORD;
1100 if (appPassword && body.did) {
1101 const pdsHost = "amanita.us-east.host.bsky.network";
1102 const authResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.server.createSession`, {
1103 method: "POST",
1104 headers: { "Content-Type": "application/json" },
1105 body: JSON.stringify({ identifier: body.did, password: appPassword }),
1106 });
1107 if (authResp.ok) {
1108 const { accessJwt } = await authResp.json() as { accessJwt: string };
1109
1110 for (const conn of newConnections) {
1111 const rkey = crypto.randomUUID().replace(/-/g, "").slice(0, 13); // TID format
1112 const record = {
1113 "$type": "network.cosmik.connection",
1114 source: conn.source,
1115 target: conn.target,
1116 connectionType: conn.connectionType,
1117 note: conn.note,
1118 createdAt: new Date().toISOString(),
1119 };
1120
1121 const createResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.repo.createRecord`, {
1122 method: "POST",
1123 headers: {
1124 "Content-Type": "application/json",
1125 "Authorization": `Bearer ${accessJwt}`,
1126 },
1127 body: JSON.stringify({
1128 repo: body.did,
1129 collection: "network.cosmik.connection",
1130 rkey,
1131 record,
1132 }),
1133 });
1134
1135 if (createResp.ok) {
1136 const createResult = await createResp.json() as { uri: string };
1137 createdUris.push(createResult.uri);
1138 }
1139 }
1140 }
1141 }
1142 }
1143
1144 // Add new connection URIs to highlights
1145 result.highlights = [...(result.highlights || []), ...createdUris];
1146 result.createdConnections = createdUris;
1147
1148 return c.json(result);
1149 } catch (e: any) {
1150 return c.json({ error: "ContainerError", message: e.message }, 500);
1151 }
1152 });
1153
1154 // ── Update strata record (write back to PDS) ────────────────────
1155 app.put("/xrpc/org.latha.island.updateRecord", async (c) => {
1156 const body = await c.req.json().catch(() => ({})) as {
1157 uri?: string;
1158 analysis?: Record<string, any>;
1159 };
1160 if (!body.uri || !body.analysis) {
1161 return c.json({ error: "MissingRequiredField", message: "uri and analysis are required" }, 400);
1162 }
1163
1164 const appPassword = env.PDS_APP_PASSWORD;
1165 if (!appPassword) {
1166 return c.json({ error: "NotConfigured", message: "PDS_APP_PASSWORD not set" }, 500);
1167 }
1168
1169 // Parse the AT URI to extract repo DID, collection, rkey
1170 const parts = body.uri.replace("at://", "").split("/");
1171 const repoDid = parts[0];
1172 const collection = parts[1];
1173 const rkey = parts[2];
1174 if (!repoDid || !collection || !rkey) {
1175 return c.json({ error: "InvalidUri", message: "Expected at://did:.../org.latha.island/rkey" }, 400);
1176 }
1177
1178 // Read existing record from D1
1179 const row = await db
1180 .prepare("SELECT record FROM records_strata WHERE uri = ?")
1181 .bind(body.uri)
1182 .first<{ record: string }>();
1183 if (!row) {
1184 return c.json({ error: "RecordNotFound" }, 404);
1185 }
1186
1187 // Merge updated analysis into existing record
1188 const record = JSON.parse(row.record);
1189 record.analysis = { ...record.analysis, ...body.analysis };
1190 // Remove stale fields that are no longer in the lexicon
1191 delete record.analysis.connections;
1192 delete record.connections;
1193 record.updatedAt = new Date().toISOString();
1194
1195 // Write back to PDS via com.atproto.repo.putRecord
1196 const pdsHost = "amanita.us-east.host.bsky.network";
1197 const authResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.server.createSession`, {
1198 method: "POST",
1199 headers: { "Content-Type": "application/json" },
1200 body: JSON.stringify({ identifier: repoDid, password: appPassword }),
1201 });
1202 if (!authResp.ok) {
1203 const err = await authResp.text();
1204 return c.json({ error: "AuthFailed", message: err }, 500);
1205 }
1206 const { accessJwt } = await authResp.json() as { accessJwt: string };
1207
1208 const putResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.repo.putRecord`, {
1209 method: "POST",
1210 headers: {
1211 "Content-Type": "application/json",
1212 "Authorization": `Bearer ${accessJwt}`,
1213 },
1214 body: JSON.stringify({
1215 repo: repoDid,
1216 collection,
1217 rkey,
1218 record,
1219 }),
1220 });
1221 if (!putResp.ok) {
1222 const err = await putResp.text();
1223 return c.json({ error: "PutRecordFailed", message: err }, 500);
1224 }
1225
1226 // Update D1 cache
1227 await db
1228 .prepare("UPDATE records_strata SET record = ? WHERE uri = ?")
1229 .bind(JSON.stringify(record), body.uri)
1230 .run();
1231
1232 return c.json({ ok: true, uri: body.uri });
1233 });
1234
1235 // ── R2 sync endpoints ───────────────────────────────────────────
1236
1237 app.put("/api/sync/vault/:did/*", async (c) => {
1238 const did = c.req.param("did");
1239 const path = c.req.param("path");
1240 if (!did || !path) {
1241 return c.json({ error: "Missing path" }, 400);
1242 }
1243
1244 const key = `${did}/${path}`;
1245 const body = await c.req.raw.arrayBuffer();
1246 await env.VAULT_BUCKET.put(key, body, {
1247 httpMetadata: { contentType: c.req.header("content-type") || "text/markdown" },
1248 });
1249
1250 return c.json({ ok: true, key });
1251 });
1252
1253 app.put("/api/sync/carry/:did/*", async (c) => {
1254 const did = c.req.param("did");
1255 const path = c.req.param("path");
1256 if (!did || !path) {
1257 return c.json({ error: "Missing path" }, 400);
1258 }
1259
1260 const key = `${did}/${path}`;
1261 const body = await c.req.raw.arrayBuffer();
1262 await env.CARRY_BUCKET.put(key, body, {
1263 httpMetadata: { contentType: c.req.header("content-type") || "application/json" },
1264 });
1265
1266 return c.json({ ok: true, key });
1267 });
1268
1269 // ── Graph neighborhood query ────────────────────────────────────
1270
1271 app.get("/xrpc/org.latha.island.connection.getGraph", async (c) => {
1272 const uri = c.req.query("uri");
1273 if (!uri) {
1274 return c.json({ error: "MissingRequiredParameter", message: "uri is required" }, 400);
1275 }
1276
1277 const depth = Math.min(parseInt(c.req.query("depth") || "2"), 3);
1278 const limit = Math.min(parseInt(c.req.query("limit") || "50"), 100);
1279 const types = c.req.query("types")?.split(",").map((t) => t.trim());
1280
1281 // Reuse island detection for the subgraph
1282 const islands = await detectIslands(db);
1283 const island = islands.find(i => i.vertices.includes(uri));
1284
1285 if (!island) {
1286 return c.json({ connections: [], resources: [], depth, uri });
1287 }
1288
1289 // Convert to graph format
1290 const resources = island.vertices.map(v => ({
1291 uri: v,
1292 title: v,
1293 type: "unknown",
1294 }));
1295
1296 return c.json({
1297 connections: island.edges,
1298 resources,
1299 depth,
1300 uri,
1301 });
1302 });
1303
1304 // ── OAuth client metadata ──────────────────────────────────────
1305
1306 app.get("/oauth-client-metadata.json", (c) => {
1307 const host = new URL(c.req.url).host;
1308 return c.json({
1309 client_id: `https://${host}/oauth-client-metadata.json`,
1310 client_name: "Stigmergic",
1311 client_uri: `https://${host}`,
1312 redirect_uris: [`https://${host}/`],
1313 scope: "atproto transition:generic",
1314 grant_types: ["authorization_code", "refresh_token"],
1315 response_types: ["code"],
1316 token_endpoint_auth_method: "none",
1317 application_type: "web",
1318 dpop_bound_access_tokens: true,
1319 });
1320 });
1321
1322 // ── SPA fallback: serve landing page for client-side routes ──────
1323 for (const path of ["/island", "/island/*", "/strata", "/connect"]) {
1324 app.get(path, async (c) => {
1325 let islandsJson = "[]";
1326 try {
1327 await ensureContrailReady(db);
1328 const rows = await db
1329 .prepare("SELECT uri, record FROM records_strata ORDER BY time_us DESC LIMIT 50")
1330 .all<{ uri: string; record: string }>();
1331 const islands: any[] = [];
1332 for (const row of rows.results || []) {
1333 try {
1334 const rec = JSON.parse(row.record);
1335 const centroid = rec.source?.uri;
1336 if (!centroid) continue;
1337 const id = row.uri.split("/").pop() || row.uri;
1338 // Compute island hash from centroid (same as island-shared.ts)
1339 const encoder = new TextEncoder();
1340 const data = encoder.encode(centroid);
1341 const hashBuffer = await crypto.subtle.digest("SHA-256", data);
1342 const hashArray = Array.from(new Uint8Array(hashBuffer));
1343 const islandHash = hashArray.map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 12);
1344 islands.push({ id, islandHash, centroid, title: rec.analysis?.title || null, recordUri: row.uri });
1345 } catch {}
1346 }
1347 islandsJson = JSON.stringify(islands);
1348 } catch {}
1349
1350 const page = LANDING_PAGE.replace(
1351 "</head>",
1352 `<script>window.__ISLANDS__=${islandsJson};</script></head>`,
1353 );
1354 return c.html(page);
1355 });
1356 }
1357
1358 // ── All other routes pass through to contrail ──────────────────
1359
1360 app.all("*", async (c) => {
1361 const response = await contrailWorker.fetch(
1362 c.req.raw,
1363 c.env as unknown as Record<string, unknown>,
1364 );
1365 return response;
1366 });
1367
1368 return app;
1369}
1370
1371// ─── Export ─────────────────────────────────────────────────────────
1372
1373export { StrataContainer };
1374
1375export default {
1376 fetch(request: Request, env: Env): Response | Promise<Response> {
1377 app ??= buildApp(env);
1378 return app.fetch(request, env);
1379 },
1380 async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
1381 // Run Contrail indexing
1382 await contrailWorker.scheduled(
1383 event,
1384 env as unknown as Record<string, unknown>,
1385 ctx,
1386 );
1387 },
1388};
1389