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