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