import { Hono } from "hono"; import { cors } from "hono/cors"; import { createWorker } from "@atmo-dev/contrail/worker"; import { Contrail } from "@atmo-dev/contrail"; import { config } from "./contrail.config.js"; import { lexicons } from "../lexicons/generated/index.js"; import { LANDING_PAGE } from "./landing-page.js"; import { StrataContainer } from "./container.js"; // ─── Env ──────────────────────────────────────────────────────────── interface Env { DB: D1Database; VAULT_BUCKET: R2Bucket; CARRY_BUCKET: R2Bucket; STRATA_CONTAINER: DurableObjectNamespace; } // ─── Contrail setup ───────────────────────────────────────────────── const contrailWorker = createWorker(config, { lexicons }); const contrail = new Contrail(config); let contrailReady = false; async function ensureContrailReady(db: D1Database): Promise { if (contrailReady) return; await contrail.init(db); contrailReady = true; } // ─── Island detection (connected components) ──────────────────────── // // Load all connections, build an undirected graph, find connected // components via BFS. Each component is an "island" — a cluster of // linked URLs that form a constellation in the knowledge graph. interface Island { id: string; vertices: string[]; edges: Array<{ uri: string; did: string; source: string; target: string; connectionType: string; note: string; handle?: string | null; }>; } async function detectIslands(db: D1Database): Promise { await ensureContrailReady(db); // Load all connections const rows = await db .prepare( `SELECT r.record, r.did, r.rkey, i.handle FROM records_connection r LEFT JOIN identities i ON r.did = i.did ORDER BY r.time_us DESC`, ) .all<{ record: string; did: string; rkey: string; handle: string | null }>(); // Build adjacency list const adj = new Map>(); const edges: Island["edges"] = []; for (const row of rows.results || []) { try { const value = JSON.parse(row.record); const source = value.source as string; const target = value.target as string; if (!source || !target) continue; if (!adj.has(source)) adj.set(source, new Set()); if (!adj.has(target)) adj.set(target, new Set()); adj.get(source)!.add(target); adj.get(target)!.add(source); edges.push({ uri: `at://${row.did}/network.cosmik.connection/${row.rkey}`, did: row.did, source, target, connectionType: value.connectionType || "relates", note: value.note || "", handle: row.handle, }); } catch {} } // BFS to find connected components const visited = new Set(); const islands: Island[] = []; for (const node of adj.keys()) { if (visited.has(node)) continue; const component: string[] = []; const queue = [node]; while (queue.length > 0) { const current = queue.shift()!; if (visited.has(current)) continue; visited.add(current); component.push(current); for (const neighbor of adj.get(current) || []) { if (!visited.has(neighbor)) queue.push(neighbor); } } // Only include edges where both endpoints are in this component const componentSet = new Set(component); const componentEdges = edges.filter( e => componentSet.has(e.source) && componentSet.has(e.target), ); // Stable ID: truncated SHA-256 of the lexicographic minimum vertex const lexmin = component.sort()[0]; const lexminHash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(lexmin)); const id = Array.from(new Uint8Array(lexminHash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 16); islands.push({ id, vertices: component, edges: componentEdges, }); } // Sort by size descending islands.sort((a, b) => b.vertices.length - a.vertices.length); return islands; } // ─── Resolve vertex metadata (batched) ────────────────────────────── // // Bulk queries instead of N+1 per vertex. Loads all cards and citations // matching the given URIs in a few chunked queries. async function resolveVertexMeta( db: D1Database, uris: string[], ): Promise> { const meta = new Map(); for (const uri of uris) { meta.set(uri, { title: uri, description: "", type: "unknown" }); } const httpUris = uris.filter(u => u.startsWith("http")); const atUris = uris.filter(u => u.startsWith("at://")); const CHUNK = 50; // ── Resolve HTTP URIs (existing logic) ── // Batch card lookup for (let i = 0; i < httpUris.length; i += CHUNK) { const chunk = httpUris.slice(i, i + CHUNK); const placeholders = chunk.map(() => "?").join(","); const rows = await db .prepare( `SELECT record FROM records_card WHERE json_extract(record, '$.content.url') IN (${placeholders})`, ) .bind(...chunk) .all<{ record: string }>(); for (const row of rows.results || []) { try { const card = JSON.parse(row.record); const url = card?.content?.url; if (!url || !meta.has(url)) continue; const m = card.content.metadata; if (m?.title || m?.description) { meta.set(url, { title: m.title || url, description: m.description || "", type: card.type === "URL" ? "url" : "note", }); } } catch {} } } // Batch citation lookup (overrides card data if found) for (let i = 0; i < httpUris.length; i += CHUNK) { const chunk = httpUris.slice(i, i + CHUNK); const placeholders = chunk.map(() => "?").join(","); const rows = await db .prepare( `SELECT record FROM records_citation WHERE json_extract(record, '$.url') IN (${placeholders})`, ) .bind(...chunk) .all<{ record: string }>(); for (const row of rows.results || []) { try { const citation = JSON.parse(row.record); const url = citation.url; if (!url || !meta.has(url)) continue; if (citation.title) { const existing = meta.get(url)!; meta.set(url, { title: citation.title, description: citation.takeaway || existing.description, type: "citation", }); } } catch {} } } // ── Resolve AT URIs ── // Batch card lookup by AT URI for (let i = 0; i < atUris.length; i += CHUNK) { const chunk = atUris.slice(i, i + CHUNK); const placeholders = chunk.map(() => "?").join(","); const rows = await db .prepare(`SELECT uri, record FROM records_card WHERE uri IN (${placeholders})`) .bind(...chunk) .all<{ uri: string; record: string }>(); for (const row of rows.results || []) { try { const card = JSON.parse(row.record); const m = card?.content?.metadata; const url = card?.content?.url; if (meta.has(row.uri)) { meta.set(row.uri, { title: m?.title || url || row.uri, description: m?.description || "", type: "card", }); } } catch {} } } // Batch collection lookup by AT URI for (let i = 0; i < atUris.length; i += CHUNK) { const chunk = atUris.slice(i, i + CHUNK); const placeholders = chunk.map(() => "?").join(","); const rows = await db .prepare(`SELECT uri, record FROM records_collection WHERE uri IN (${placeholders})`) .bind(...chunk) .all<{ uri: string; record: string }>(); for (const row of rows.results || []) { try { const coll = JSON.parse(row.record); if (meta.has(row.uri)) { meta.set(row.uri, { title: coll.name || row.uri, description: coll.description || "", type: "collection", }); } } catch {} } } return meta; } // ─── Derive island from lexmin ──────────────────────────────────── // // Given a lexmin vertex (canonical component reference), derive the // connected component by traversing network.cosmik.connection records. async function deriveIsland( db: D1Database, seed: string, ): Promise { await ensureContrailReady(db); // BFS in application code — recursive CTE OOMs in D1 due to cross join const visited = new Set(); const queue = [seed]; const allEdges: Array<{ uri: string; did: string; rkey: string; source: string; target: string; connectionType: string; note: string; handle: string | null; }> = []; while (queue.length > 0) { // Drain the current queue into a batch const batch = queue.splice(0, 50); const toQuery = batch.filter(v => !visited.has(v)); if (toQuery.length === 0) continue; // Mark visited for (const v of toQuery) visited.add(v); // Query connections where any of these vertices are source or target const placeholders = toQuery.map(() => "?").join(","); const rows = await db .prepare( `SELECT r.record, r.did, r.rkey, i.handle FROM records_connection r LEFT JOIN identities i ON r.did = i.did WHERE json_extract(r.record, '$.source') IN (${placeholders}) OR json_extract(r.record, '$.target') IN (${placeholders})`, ) .bind(...toQuery, ...toQuery) .all<{ record: string; did: string; rkey: string; handle: string | null }>(); for (const row of rows.results || []) { try { const value = JSON.parse(row.record); const source = value.source as string; const target = value.target as string; if (!source || !target) continue; // Collect edge const edgeKey = `${source}|${target}|${row.rkey}`; allEdges.push({ uri: `at://${row.did}/network.cosmik.connection/${row.rkey}`, did: row.did, rkey: row.rkey, source, target, connectionType: value.connectionType || "relates", note: value.note || "", handle: row.handle, }); // Enqueue unvisited neighbors if (!visited.has(source)) queue.push(source); if (!visited.has(target)) queue.push(target); } catch {} } } if (visited.size === 0) return null; const vertices = [...visited]; // Filter edges to only those with both endpoints in the component const vertexSet = new Set(vertices); const edges: Island["edges"] = []; const seenEdges = new Set(); for (const e of allEdges) { if (!vertexSet.has(e.source) || !vertexSet.has(e.target)) continue; const edgeKey = `${e.source}|${e.target}|${e.rkey}`; if (seenEdges.has(edgeKey)) continue; seenEdges.add(edgeKey); edges.push({ uri: e.uri, did: e.did, source: e.source, target: e.target, connectionType: e.connectionType, note: e.note, handle: e.handle, }); } // Compute stable ID from lexmin const sortedVertices = [...vertices].sort(); const actualLexmin = sortedVertices[0]; const lexminHash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(actualLexmin)); const id = Array.from(new Uint8Array(lexminHash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 16); return { id, vertices, edges }; } // ─── List strata islands from records_strata ────────────────────── // ─── Hono app ─────────────────────────────────────────────────────── let app: Hono<{ Bindings: Env }> | null = null; function buildApp(env: Env): Hono<{ Bindings: Env }> { const app = new Hono<{ Bindings: Env }>(); const db = env.DB; app.use("*", cors()); // Landing page — strata records feed app.get("/", async (c) => { // Load only the fields we need — records are 100KB+ with embedded connections // No ensureContrailReady needed — just D1 let islandsJson = "[]"; try { const rows = await db .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") .all<{ uri: string; lexmin: string | null; title: string | null }>(); const islands: any[] = []; for (const row of rows.results || []) { if (!row.lexmin) continue; const lexminHash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(row.lexmin)); const id = Array.from(new Uint8Array(lexminHash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 16); islands.push({ id, lexmin: row.lexmin, title: row.title || null, recordUri: row.uri }); } islandsJson = JSON.stringify(islands); } catch (e) { console.error("Failed to load islands:", e); } const page = LANDING_PAGE.replace( "", ``, ); return c.html(page); }); // ── Islands API ────────────────────────────────────────────────── app.get("/xrpc/org.latha.strata.getIslands", async (c) => { const limit = Math.min(parseInt(c.req.query("limit") || "50"), 100); await ensureContrailReady(db); const rows = await db .prepare("SELECT uri, record FROM records_strata ORDER BY time_us DESC LIMIT ?") .bind(limit) .all<{ uri: string; record: string }>(); const islands: any[] = []; for (const row of rows.results || []) { try { const rec = JSON.parse(row.record); const lexmin = rec.source?.uri; if (!lexmin) continue; const lexminHash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(lexmin)); const id = Array.from(new Uint8Array(lexminHash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 16); // Build graph directly from embedded connections — no D1 lookups needed const connections: any[] = rec.connections || []; const vertexSet = new Set(); const edges: any[] = []; for (const conn of connections) { if (typeof conn === "object" && conn.source && conn.target) { vertexSet.add(conn.source); vertexSet.add(conn.target); edges.push(conn); } } const vertices = [...vertexSet].sort(); const vertexMeta = Object.fromEntries(await resolveVertexMeta(db, vertices)); const analysis = rec.analysis || {}; islands.push({ id, lexmin, vertices, edges, vertexMeta, title: analysis.title || null, strata: analysis.synthesis ? { prose: analysis.synthesis || "", title: analysis.title || "", themes: analysis.themes || [], relationships: (analysis.connections || []).map((conn: any) => conn.description || ""), tensions: analysis.tensions || [], open_questions: analysis.openQuestions || [], synthesis: analysis.synthesis || "", } : null, recordUri: row.uri, }); } catch {} } return c.json({ islands }); }); // ── Get single island by ID or AT URI ──────────────────────────── app.get("/xrpc/org.latha.strata.getIsland", async (c) => { const id = c.req.query("id"); if (!id) { return c.json({ error: "MissingRequiredParameter", message: "id is required" }, 400); } await ensureContrailReady(db); // If it's an AT URI, look up the strata record and derive from its lexmin if (id.startsWith("at://")) { const row = await db .prepare("SELECT uri, record FROM records_strata WHERE uri = ?") .bind(id) .first<{ uri: string; record: string }>(); if (!row) { return c.json({ error: "RecordNotFound" }, 404); } const rec = JSON.parse(row.record); const lexmin = rec.source?.uri; if (!lexmin) { return c.json({ error: "InvalidRecord" }, 400); } const island = await deriveIsland(db, lexmin); if (!island) { return c.json({ error: "IslandNotFound" }, 404); } const meta = await resolveVertexMeta(db, island.vertices); const analysis = rec.analysis || {}; return c.json({ island: { ...island, lexmin: island.vertices.slice().sort()[0], vertexMeta: Object.fromEntries(meta), summary: analysis.title || null, strata: { prose: analysis.synthesis || "", title: analysis.title || "", themes: analysis.themes || [], relationships: (analysis.connections || []).map((conn: any) => conn.description || ""), tensions: analysis.tensions || [], open_questions: analysis.openQuestions || [], synthesis: analysis.synthesis || "", }, }, recordUri: row.uri, }); } // Stable ID — derive from lexmin by reverse-lookup // Find the strata record whose lexmin hashes to this ID const strataRows = await db .prepare("SELECT uri, record FROM records_strata") .all<{ uri: string; record: string }>(); for (const row of strataRows.results || []) { try { const rec = JSON.parse(row.record); const lexmin = rec.source?.uri; if (!lexmin) continue; const lexminHash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(lexmin)); const candidateId = Array.from(new Uint8Array(lexminHash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 16); if (candidateId === id) { const island = await deriveIsland(db, lexmin); if (!island) continue; const meta = await resolveVertexMeta(db, island.vertices); const analysis = rec.analysis || {}; return c.json({ island: { ...island, lexmin: island.vertices.slice().sort()[0], vertexMeta: Object.fromEntries(meta), summary: analysis.title || null, strata: { prose: analysis.synthesis || "", title: analysis.title || "", themes: analysis.themes || [], relationships: (analysis.connections || []).map((conn: any) => conn.description || ""), tensions: analysis.tensions || [], open_questions: analysis.openQuestions || [], synthesis: analysis.synthesis || "", }, }, recordUri: row.uri, }); } } catch {} } // No strata record — try deriving island by treating the ID as a lexmin seed // (for islands that haven't been analyzed yet) const island = await deriveIsland(db, id); if (!island) { return c.json({ error: "IslandNotFound" }, 404); } const meta = await resolveVertexMeta(db, island.vertices); return c.json({ island: { ...island, lexmin: island.vertices.slice().sort()[0], vertexMeta: Object.fromEntries(meta), summary: null, strata: null, }, recordUri: null, }); }); // ── List strata records (AT URI-based) ────────────────────────── app.get("/xrpc/org.latha.strata.listRecords", async (c) => { const rows = await db .prepare("SELECT uri, record FROM records_strata ORDER BY time_us DESC LIMIT 50") .all<{ uri: string; record: string }>(); const records = (rows.results || []).map(r => { try { const rec = JSON.parse(r.record); return { uri: r.uri, source: rec.source?.uri || "", title: rec.analysis?.title || "", themes: rec.analysis?.themes || [], createdAt: rec.createdAt || "", }; } catch { return null; } }).filter(Boolean); return c.json({ records }); }); // ── Strata record endpoint (AT URI-based) ────────────────────── app.get("/xrpc/org.latha.strata.getRecord", async (c) => { const uri = c.req.query("uri"); if (!uri) { return c.json({ error: "MissingRequiredParameter", message: "uri is required" }, 400); } // Fetch the strata record from D1 (indexed by Contrail) const row = await db .prepare("SELECT record FROM records_strata WHERE uri = ?") .bind(uri) .first<{ record: string }>(); if (!row) { return c.json({ error: "NotFound", message: "Strata record not found" }, 404); } let strataRecord: any; try { strataRecord = JSON.parse(row.record); } catch { return c.json({ error: "InvalidRecord" }, 500); } // Derive island from lexmin (source.uri) const lexmin = strataRecord.source?.uri; if (!lexmin) { return c.json({ error: "MissingLexmin", message: "Strata record has no source.uri" }, 400); } const island = await deriveIsland(db, lexmin); if (!island) { return c.json({ error: "IslandNotFound", message: "Could not derive island from lexmin" }, 404); } // Resolve vertex metadata const meta = await resolveVertexMeta(db, island.vertices); // Build the response in the same format as the island cache const analysis = strataRecord.analysis || {}; const strataData = { prose: analysis.synthesis || "", title: analysis.title || "", themes: analysis.themes || [], relationships: (analysis.connections || []).map((c: any) => c.description || ""), tensions: analysis.tensions || [], open_questions: analysis.openQuestions || [], synthesis: analysis.synthesis || "", }; return c.json({ island: { ...island, vertexMeta: Object.fromEntries(meta), summary: analysis.title || null, strata: strataData, }, record: strataRecord, recordUri: uri, }); }); // ── Container-backed xrpc methods ──────────────────────────────── // Helper: call the strata container async function callContainer(path: string, method: string, body?: any): Promise { const id = env.STRATA_CONTAINER.idFromName("strata"); const stub = env.STRATA_CONTAINER.get(id); const url = new URL(path, "http://container"); const init: RequestInit = { method }; if (body) { init.body = JSON.stringify(body); init.headers = { "Content-Type": "application/json" }; } const resp = await stub.fetch(new Request(url.toString(), init)); return resp.json(); } // Derive islands — POST, delegates to container // With seed: single island. Without: all islands. app.post("/xrpc/org.latha.strata.deriveIsland", async (c) => { const body = await c.req.json().catch(() => ({})) as { seed?: string }; const seed = body.seed; // Load all connections from D1 for the container to traverse await ensureContrailReady(db); const rows = await db .prepare( `SELECT r.record, r.did, r.rkey, i.handle FROM records_connection r LEFT JOIN identities i ON r.did = i.did`, ) .all<{ record: string; did: string; rkey: string; handle: string | null }>(); const connections = (rows.results || []).map(r => { try { const value = JSON.parse(r.record); return { source: value.source, target: value.target, connectionType: value.connectionType || "relates", note: value.note || "", did: r.did, rkey: r.rkey, }; } catch { return null; } }).filter(Boolean); try { const result = await callContainer("/deriveIsland", "POST", { connections, seed: seed || undefined }); return c.json(result); } catch (e: any) { return c.json({ error: "ContainerError", message: e.message }, 500); } }); // Run strata analysis — delegates to container app.post("/xrpc/org.latha.strata.analyze", async (c) => { const body = await c.req.json().catch(() => ({})); if (!body.island) { return c.json({ error: "MissingRequiredField", message: "island is required" }, 400); } try { const result = await callContainer("/analyze", "POST", body); return c.json(result); } catch (e: any) { return c.json({ error: "ContainerError", message: e.message }, 500); } }); // ── R2 sync endpoints ─────────────────────────────────────────── app.put("/api/sync/vault/:did/*", async (c) => { const did = c.req.param("did"); const path = c.req.param("path"); if (!did || !path) { return c.json({ error: "Missing path" }, 400); } const key = `${did}/${path}`; const body = await c.req.raw.arrayBuffer(); await env.VAULT_BUCKET.put(key, body, { httpMetadata: { contentType: c.req.header("content-type") || "text/markdown" }, }); return c.json({ ok: true, key }); }); app.put("/api/sync/carry/:did/*", async (c) => { const did = c.req.param("did"); const path = c.req.param("path"); if (!did || !path) { return c.json({ error: "Missing path" }, 400); } const key = `${did}/${path}`; const body = await c.req.raw.arrayBuffer(); await env.CARRY_BUCKET.put(key, body, { httpMetadata: { contentType: c.req.header("content-type") || "application/json" }, }); return c.json({ ok: true, key }); }); // ── Graph neighborhood query ──────────────────────────────────── app.get("/xrpc/org.latha.strata.connection.getGraph", async (c) => { const uri = c.req.query("uri"); if (!uri) { return c.json({ error: "MissingRequiredParameter", message: "uri is required" }, 400); } const depth = Math.min(parseInt(c.req.query("depth") || "2"), 3); const limit = Math.min(parseInt(c.req.query("limit") || "50"), 100); const types = c.req.query("types")?.split(",").map((t) => t.trim()); // Reuse island detection for the subgraph const islands = await detectIslands(db); const island = islands.find(i => i.vertices.includes(uri)); if (!island) { return c.json({ connections: [], resources: [], depth, uri }); } // Convert to graph format const resources = island.vertices.map(v => ({ uri: v, title: v, type: "unknown", })); return c.json({ connections: island.edges, resources, depth, uri, }); }); // ── OAuth client metadata ────────────────────────────────────── app.get("/oauth-client-metadata.json", (c) => { const host = new URL(c.req.url).host; return c.json({ client_id: `https://${host}/oauth-client-metadata.json`, client_name: "Stigmergic", client_uri: `https://${host}`, redirect_uris: [`https://${host}/`], scope: "atproto transition:generic", grant_types: ["authorization_code", "refresh_token"], response_types: ["code"], token_endpoint_auth_method: "none", application_type: "web", dpop_bound_access_tokens: true, }); }); // ── SPA fallback: serve landing page for client-side routes ────── for (const path of ["/island", "/strata", "/connect"]) { app.get(path, async (c) => { let islandsJson = "[]"; try { await ensureContrailReady(db); const rows = await db .prepare("SELECT uri, record FROM records_strata ORDER BY time_us DESC LIMIT 50") .all<{ uri: string; record: string }>(); const islands: any[] = []; for (const row of rows.results || []) { try { const rec = JSON.parse(row.record); const lexmin = rec.source?.uri; if (!lexmin) continue; const lexminHash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(lexmin)); const id = Array.from(new Uint8Array(lexminHash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 16); islands.push({ id, lexmin, title: rec.analysis?.title || null, recordUri: row.uri }); } catch {} } islandsJson = JSON.stringify(islands); } catch {} const page = LANDING_PAGE.replace( "", ``, ); return c.html(page); }); } // ── All other routes pass through to contrail ────────────────── app.all("*", async (c) => { const response = await contrailWorker.fetch( c.req.raw, c.env as unknown as Record, ); return response; }); return app; } // ─── Export ───────────────────────────────────────────────────────── export { StrataContainer }; export default { fetch(request: Request, env: Env): Response | Promise { app ??= buildApp(env); return app.fetch(request, env); }, async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) { // Run Contrail indexing await contrailWorker.scheduled( event, env as unknown as Record, ctx, ); }, };