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