nandi.uk
This repository has no description
40 kB
1145 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.strata.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.strata.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.strata/${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.strata.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.strata.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 }>;
699 citations: Array<{ atUri: string; url: string; title: string; takeaway: string; confidence?: string }>;
700 reasonings: Array<{ atUri: string; claim: string; evidence: string; reasoningType?: string }>;
701 }> {
702 const entities: Array<{ atUri: string; name: string; entityType: string; description?: string }> = [];
703 const citations: Array<{ atUri: string; url: string; title: string; takeaway: string; confidence?: string }> = [];
704 const reasonings: Array<{ atUri: string; claim: string; evidence: string; reasoningType?: string }> = [];
705
706 try {
707 // Read entities
708 const entityRows = await db
709 .prepare("SELECT uri, record FROM records_entity ORDER BY time_us DESC LIMIT 100")
710 .all<{ uri: string; record: string }>();
711 for (const row of entityRows.results || []) {
712 try {
713 const r = JSON.parse(row.record);
714 entities.push({
715 atUri: row.uri,
716 name: r.name || "",
717 entityType: r.entityType || "",
718 description: r.description || "",
719 });
720 } catch {}
721 }
722
723 // Read citations
724 const citeRows = await db
725 .prepare("SELECT uri, record FROM records_citation ORDER BY time_us DESC LIMIT 100")
726 .all<{ uri: string; record: string }>();
727 for (const row of citeRows.results || []) {
728 try {
729 const r = JSON.parse(row.record);
730 citations.push({
731 atUri: row.uri,
732 url: r.url || "",
733 title: r.title || "",
734 takeaway: r.takeaway || "",
735 confidence: r.confidence || "",
736 });
737 } catch {}
738 }
739
740 // Read reasonings
741 const reasonRows = await db
742 .prepare("SELECT uri, record FROM records_reasoning ORDER BY time_us DESC LIMIT 100")
743 .all<{ uri: string; record: string }>();
744 for (const row of reasonRows.results || []) {
745 try {
746 const r = JSON.parse(row.record);
747 reasonings.push({
748 atUri: row.uri,
749 claim: r.claim || "",
750 evidence: r.evidence || "",
751 reasoningType: r.reasoningType || "",
752 });
753 } catch {}
754 }
755 } catch (e: any) {
756 console.error("getCarryData error:", e.message);
757 }
758
759 return { entities, citations, reasonings };
760 }
761
762 // Helper: call the strata container
763 async function callContainer(path: string, method: string, body?: any): Promise<any> {
764 const id = env.STRATA_CONTAINER.idFromName("strata");
765 const stub = env.STRATA_CONTAINER.get(id);
766 const url = new URL(path, "http://container");
767 const init: RequestInit = { method };
768 if (body) {
769 init.body = JSON.stringify(body);
770 init.headers = { "Content-Type": "application/json" };
771 }
772 const resp = await stub.fetch(new Request(url.toString(), init));
773 const text = await resp.text();
774 try {
775 return JSON.parse(text);
776 } catch {
777 throw new Error(`Container returned non-JSON: ${text.slice(0, 500)}`);
778 }
779 }
780
781 // Sync container data (vault + carry)
782 app.post("/xrpc/org.latha.strata.syncContainer", async (c) => {
783 try {
784 const result = await callContainer("/sync", "POST");
785 return c.json(result);
786 } catch (e: any) {
787 return c.json({ error: "ContainerError", message: e.message }, 500);
788 }
789 });
790
791 // Container health + data status
792 app.get("/xrpc/org.latha.strata.containerHealth", async (c) => {
793 try {
794 const result = await callContainer("/health", "GET");
795 return c.json(result);
796 } catch (e: any) {
797 return c.json({ error: "ContainerError", message: e.message }, 500);
798 }
799 });
800
801 // Derive islands — POST, delegates to container
802 // With seed: single island. Without: all islands.
803 app.post("/xrpc/org.latha.strata.deriveIsland", async (c) => {
804 const body = await c.req.json().catch(() => ({})) as { seed?: string };
805 const seed = body.seed;
806
807 // Load all connections from D1 for the container to traverse
808 await ensureContrailReady(db);
809 const rows = await db
810 .prepare(
811 `SELECT r.record, r.did, r.rkey, i.handle
812 FROM records_connection r
813 LEFT JOIN identities i ON r.did = i.did`,
814 )
815 .all<{ record: string; did: string; rkey: string; handle: string | null }>();
816
817 const connections = (rows.results || []).map(r => {
818 try {
819 const value = JSON.parse(r.record);
820 return {
821 source: value.source,
822 target: value.target,
823 connectionType: value.connectionType || "relates",
824 note: value.note || "",
825 did: r.did,
826 rkey: r.rkey,
827 };
828 } catch {
829 return null;
830 }
831 }).filter(Boolean);
832
833 try {
834 const result = await callContainer("/deriveIsland", "POST", { connections, seed: seed || undefined });
835 return c.json(result);
836 } catch (e: any) {
837 return c.json({ error: "ContainerError", message: e.message }, 500);
838 }
839 });
840
841 // Run strata analysis — delegates to container
842 app.post("/xrpc/org.latha.strata.analyze", async (c) => {
843 const body = await c.req.json().catch(() => ({}));
844 if (!body.island) {
845 return c.json({ error: "MissingRequiredField", message: "island is required" }, 400);
846 }
847
848 try {
849 // Read carry data from D1 for the user's DID
850 const carryData = await getCarryData(db, body.did);
851
852 // Pass carry data to container alongside island
853 const payload = { ...body, carryData };
854 const result = await callContainer("/analyze", "POST", payload);
855
856 // Create new connection records on PDS
857 const newConnections: Array<{ source: string; target: string; connectionType: string; note: string }> = result.newConnections || [];
858 const createdUris: string[] = [];
859
860 if (newConnections.length > 0) {
861 const appPassword = env.PDS_APP_PASSWORD;
862 if (appPassword && body.did) {
863 const pdsHost = "amanita.us-east.host.bsky.network";
864 const authResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.server.createSession`, {
865 method: "POST",
866 headers: { "Content-Type": "application/json" },
867 body: JSON.stringify({ identifier: body.did, password: appPassword }),
868 });
869 if (authResp.ok) {
870 const { accessJwt } = await authResp.json() as { accessJwt: string };
871
872 for (const conn of newConnections) {
873 const rkey = crypto.randomUUID().replace(/-/g, "").slice(0, 13); // TID format
874 const record = {
875 "$type": "network.cosmik.connection",
876 source: conn.source,
877 target: conn.target,
878 connectionType: conn.connectionType,
879 note: conn.note,
880 createdAt: new Date().toISOString(),
881 };
882
883 const createResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.repo.createRecord`, {
884 method: "POST",
885 headers: {
886 "Content-Type": "application/json",
887 "Authorization": `Bearer ${accessJwt}`,
888 },
889 body: JSON.stringify({
890 repo: body.did,
891 collection: "network.cosmik.connection",
892 rkey,
893 record,
894 }),
895 });
896
897 if (createResp.ok) {
898 const createResult = await createResp.json() as { uri: string };
899 createdUris.push(createResult.uri);
900 }
901 }
902 }
903 }
904 }
905
906 // Add new connection URIs to highlights
907 result.highlights = [...(result.highlights || []), ...createdUris];
908 result.createdConnections = createdUris;
909
910 return c.json(result);
911 } catch (e: any) {
912 return c.json({ error: "ContainerError", message: e.message }, 500);
913 }
914 });
915
916 // ── Update strata record (write back to PDS) ────────────────────
917 app.put("/xrpc/org.latha.strata.updateRecord", async (c) => {
918 const body = await c.req.json().catch(() => ({})) as {
919 uri?: string;
920 analysis?: Record<string, any>;
921 };
922 if (!body.uri || !body.analysis) {
923 return c.json({ error: "MissingRequiredField", message: "uri and analysis are required" }, 400);
924 }
925
926 const appPassword = env.PDS_APP_PASSWORD;
927 if (!appPassword) {
928 return c.json({ error: "NotConfigured", message: "PDS_APP_PASSWORD not set" }, 500);
929 }
930
931 // Parse the AT URI to extract repo DID, collection, rkey
932 const parts = body.uri.replace("at://", "").split("/");
933 const repoDid = parts[0];
934 const collection = parts[1];
935 const rkey = parts[2];
936 if (!repoDid || !collection || !rkey) {
937 return c.json({ error: "InvalidUri", message: "Expected at://did:.../org.latha.strata/rkey" }, 400);
938 }
939
940 // Read existing record from D1
941 const row = await db
942 .prepare("SELECT record FROM records_strata WHERE uri = ?")
943 .bind(body.uri)
944 .first<{ record: string }>();
945 if (!row) {
946 return c.json({ error: "RecordNotFound" }, 404);
947 }
948
949 // Merge updated analysis into existing record
950 const record = JSON.parse(row.record);
951 record.analysis = { ...record.analysis, ...body.analysis };
952 // Remove stale fields that are no longer in the lexicon
953 delete record.analysis.connections;
954 delete record.connections;
955 record.updatedAt = new Date().toISOString();
956
957 // Write back to PDS via com.atproto.repo.putRecord
958 const pdsHost = "amanita.us-east.host.bsky.network";
959 const authResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.server.createSession`, {
960 method: "POST",
961 headers: { "Content-Type": "application/json" },
962 body: JSON.stringify({ identifier: repoDid, password: appPassword }),
963 });
964 if (!authResp.ok) {
965 const err = await authResp.text();
966 return c.json({ error: "AuthFailed", message: err }, 500);
967 }
968 const { accessJwt } = await authResp.json() as { accessJwt: string };
969
970 const putResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.repo.putRecord`, {
971 method: "POST",
972 headers: {
973 "Content-Type": "application/json",
974 "Authorization": `Bearer ${accessJwt}`,
975 },
976 body: JSON.stringify({
977 repo: repoDid,
978 collection,
979 rkey,
980 record,
981 }),
982 });
983 if (!putResp.ok) {
984 const err = await putResp.text();
985 return c.json({ error: "PutRecordFailed", message: err }, 500);
986 }
987
988 // Update D1 cache
989 await db
990 .prepare("UPDATE records_strata SET record = ? WHERE uri = ?")
991 .bind(JSON.stringify(record), body.uri)
992 .run();
993
994 return c.json({ ok: true, uri: body.uri });
995 });
996
997 // ── R2 sync endpoints ───────────────────────────────────────────
998
999 app.put("/api/sync/vault/:did/*", async (c) => {
1000 const did = c.req.param("did");
1001 const path = c.req.param("path");
1002 if (!did || !path) {
1003 return c.json({ error: "Missing path" }, 400);
1004 }
1005
1006 const key = `${did}/${path}`;
1007 const body = await c.req.raw.arrayBuffer();
1008 await env.VAULT_BUCKET.put(key, body, {
1009 httpMetadata: { contentType: c.req.header("content-type") || "text/markdown" },
1010 });
1011
1012 return c.json({ ok: true, key });
1013 });
1014
1015 app.put("/api/sync/carry/:did/*", async (c) => {
1016 const did = c.req.param("did");
1017 const path = c.req.param("path");
1018 if (!did || !path) {
1019 return c.json({ error: "Missing path" }, 400);
1020 }
1021
1022 const key = `${did}/${path}`;
1023 const body = await c.req.raw.arrayBuffer();
1024 await env.CARRY_BUCKET.put(key, body, {
1025 httpMetadata: { contentType: c.req.header("content-type") || "application/json" },
1026 });
1027
1028 return c.json({ ok: true, key });
1029 });
1030
1031 // ── Graph neighborhood query ────────────────────────────────────
1032
1033 app.get("/xrpc/org.latha.strata.connection.getGraph", async (c) => {
1034 const uri = c.req.query("uri");
1035 if (!uri) {
1036 return c.json({ error: "MissingRequiredParameter", message: "uri is required" }, 400);
1037 }
1038
1039 const depth = Math.min(parseInt(c.req.query("depth") || "2"), 3);
1040 const limit = Math.min(parseInt(c.req.query("limit") || "50"), 100);
1041 const types = c.req.query("types")?.split(",").map((t) => t.trim());
1042
1043 // Reuse island detection for the subgraph
1044 const islands = await detectIslands(db);
1045 const island = islands.find(i => i.vertices.includes(uri));
1046
1047 if (!island) {
1048 return c.json({ connections: [], resources: [], depth, uri });
1049 }
1050
1051 // Convert to graph format
1052 const resources = island.vertices.map(v => ({
1053 uri: v,
1054 title: v,
1055 type: "unknown",
1056 }));
1057
1058 return c.json({
1059 connections: island.edges,
1060 resources,
1061 depth,
1062 uri,
1063 });
1064 });
1065
1066 // ── OAuth client metadata ──────────────────────────────────────
1067
1068 app.get("/oauth-client-metadata.json", (c) => {
1069 const host = new URL(c.req.url).host;
1070 return c.json({
1071 client_id: `https://${host}/oauth-client-metadata.json`,
1072 client_name: "Stigmergic",
1073 client_uri: `https://${host}`,
1074 redirect_uris: [`https://${host}/`],
1075 scope: "atproto transition:generic",
1076 grant_types: ["authorization_code", "refresh_token"],
1077 response_types: ["code"],
1078 token_endpoint_auth_method: "none",
1079 application_type: "web",
1080 dpop_bound_access_tokens: true,
1081 });
1082 });
1083
1084 // ── SPA fallback: serve landing page for client-side routes ──────
1085 for (const path of ["/island", "/strata", "/connect"]) {
1086 app.get(path, async (c) => {
1087 let islandsJson = "[]";
1088 try {
1089 await ensureContrailReady(db);
1090 const rows = await db
1091 .prepare("SELECT uri, record FROM records_strata ORDER BY time_us DESC LIMIT 50")
1092 .all<{ uri: string; record: string }>();
1093 const islands: any[] = [];
1094 for (const row of rows.results || []) {
1095 try {
1096 const rec = JSON.parse(row.record);
1097 const lexmin = rec.source?.uri;
1098 if (!lexmin) continue;
1099 const id = row.uri.split("/").pop() || row.uri;
1100 islands.push({ id, lexmin, title: rec.analysis?.title || null, recordUri: row.uri });
1101 } catch {}
1102 }
1103 islandsJson = JSON.stringify(islands);
1104 } catch {}
1105
1106 const page = LANDING_PAGE.replace(
1107 "</head>",
1108 `<script>window.__ISLANDS__=${islandsJson};</script></head>`,
1109 );
1110 return c.html(page);
1111 });
1112 }
1113
1114 // ── All other routes pass through to contrail ──────────────────
1115
1116 app.all("*", async (c) => {
1117 const response = await contrailWorker.fetch(
1118 c.req.raw,
1119 c.env as unknown as Record<string, unknown>,
1120 );
1121 return response;
1122 });
1123
1124 return app;
1125}
1126
1127// ─── Export ─────────────────────────────────────────────────────────
1128
1129export { StrataContainer };
1130
1131export default {
1132 fetch(request: Request, env: Env): Response | Promise<Response> {
1133 app ??= buildApp(env);
1134 return app.fetch(request, env);
1135 },
1136 async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) {
1137 // Run Contrail indexing
1138 await contrailWorker.scheduled(
1139 event,
1140 env as unknown as Record<string, unknown>,
1141 ctx,
1142 );
1143 },
1144};
1145