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"; // ─── Env ──────────────────────────────────────────────────────────── interface Env { DB: D1Database; ASSETS: { fetch(request: Request): Promise }; VAULT_BUCKET: R2Bucket; CARRY_BUCKET: R2Bucket; } // ─── 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; centroid: string; vertices: string[]; edges: Array<{ uri: string; did: string; source: string; target: string; connectionType: string; note: string; handle?: string | null; }>; } /** Compute centroid: vertex with highest closeness centrality (smallest avg BFS distance). */ function computeCentroid(adj: Map>, nodes: Set): string { if (nodes.size <= 1) return [...nodes][0] ?? ''; let bestNode = ''; let bestAvg = Infinity; for (const start of nodes) { const dist = new Map(); dist.set(start, 0); const queue = [start]; while (queue.length > 0) { const current = queue.shift()!; for (const neighbor of adj.get(current) || []) { if (!dist.has(neighbor) && nodes.has(neighbor)) { dist.set(neighbor, dist.get(current)! + 1); queue.push(neighbor); } } } let totalDist = 0; let reachable = 0; for (const [node, d] of dist) { if (nodes.has(node)) { totalDist += d; reachable++; } } const avg = reachable > 0 ? totalDist / reachable : Infinity; if (avg < bestAvg) { bestAvg = avg; bestNode = start; } } return bestNode; } /** Hash a URL to a 12-char hex island ID (matches CLI's islandId()). */ async function islandHash(url: string): Promise { const hash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(url)); return Array.from(new Uint8Array(hash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 12); } 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 centroid vertex (highest closeness centrality) const centroid = computeCentroid(adj, componentSet); const id = await islandHash(centroid); islands.push({ id, centroid, 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; // Build all queries upfront, then batch-execute in one D1 round trip const stmts: D1PreparedStatement[] = []; // HTTP URI queries — cards const httpChunks: string[][] = []; for (let i = 0; i < httpUris.length; i += CHUNK) { const chunk = httpUris.slice(i, i + CHUNK); httpChunks.push(chunk); const placeholders = chunk.map(() => "?").join(","); stmts.push( db.prepare( `SELECT record FROM records_card WHERE json_extract(record, '$.content.url') IN (${placeholders})` ).bind(...chunk) ); } // HTTP URI queries — citations for (let i = 0; i < httpUris.length; i += CHUNK) { const chunk = httpUris.slice(i, i + CHUNK); const placeholders = chunk.map(() => "?").join(","); stmts.push( db.prepare( `SELECT record FROM records_citation WHERE json_extract(record, '$.url') IN (${placeholders})` ).bind(...chunk) ); } // AT URI queries — cards const atChunks: string[][] = []; for (let i = 0; i < atUris.length; i += CHUNK) { const chunk = atUris.slice(i, i + CHUNK); atChunks.push(chunk); const placeholders = chunk.map(() => "?").join(","); stmts.push( db.prepare(`SELECT uri, record FROM records_card WHERE uri IN (${placeholders})`).bind(...chunk) ); } // AT URI queries — collections for (let i = 0; i < atUris.length; i += CHUNK) { const chunk = atUris.slice(i, i + CHUNK); const placeholders = chunk.map(() => "?").join(","); stmts.push( db.prepare(`SELECT uri, record FROM records_collection WHERE uri IN (${placeholders})`).bind(...chunk) ); } // Execute all queries in one batch const results = stmts.length > 0 ? await db.batch(stmts) : []; let resultIdx = 0; // Process card results (HTTP) for (const chunk of httpChunks) { const rows = (results[resultIdx++] as { results: { record: string }[] } | null)?.results || []; for (const row of rows) { 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 {} } } // Process citation results (HTTP) — overrides card data for (let i = 0; i < httpChunks.length; i++) { const rows = (results[resultIdx++] as { results: { record: string }[] } | null)?.results || []; for (const row of rows) { 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 {} } } // Process card results (AT) for (const chunk of atChunks) { const rows = (results[resultIdx++] as { results: { uri: string; record: string }[] } | null)?.results || []; for (const row of rows) { 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 {} } } // Process collection results (AT) for (let i = 0; i < atChunks.length; i++) { const rows = (results[resultIdx++] as { results: { uri: string; record: string }[] } | null)?.results || []; for (const row of rows) { 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 centroid ──────────────────────────────────── // // Given a centroid 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 centroid (highest closeness centrality) const adj = new Map>(); for (const e of edges) { if (!adj.has(e.source)) adj.set(e.source, new Set()); if (!adj.has(e.target)) adj.set(e.target, new Set()); adj.get(e.source)!.add(e.target); adj.get(e.target)!.add(e.source); } const centroid = computeCentroid(adj, vertexSet); const id = await islandHash(centroid); return { id, centroid, vertices, edges }; } // ─── List islands from records_island ─────────────────────────── // ─── Resolve edge URIs to full edge objects ────────────────────── /** Given a list of connections (either full objects or just {uri}), resolve URIs to full edges from D1. */ async function resolveEdgeUris(db: D1Database, connections: any[]): Promise { const resolved: any[] = []; const uriIndices: Map = new Map(); // uri → indices in connections that need it for (let i = 0; i < connections.length; i++) { const conn = connections[i]; // Already has source/target — use as-is if (typeof conn === "object" && conn.source && conn.target) { resolved.push(conn); continue; } // Just a URI reference — needs lookup const uri = conn.uri; if (!uri) continue; resolved.push(null); // placeholder if (!uriIndices.has(uri)) uriIndices.set(uri, []); uriIndices.get(uri)!.push(i); } // Batch-fetch connection records from D1 (for URI-only references) if (uriIndices.size > 0) { const uris = [...uriIndices.keys()]; const placeholders = uris.map(() => "?").join(","); const rows = await db .prepare( `SELECT r.uri, r.record, r.did, r.rkey, i.handle FROM records_connection r LEFT JOIN identities i ON r.did = i.did WHERE r.uri IN (${placeholders})`, ) .bind(...uris) .all<{ uri: string; record: string; did: string; rkey: string; handle: string | null }>(); for (const row of rows.results || []) { try { const value = JSON.parse(row.record); const edge = { uri: row.uri, did: row.did, source: value.source, target: value.target, connectionType: value.connectionType || "relates", note: value.note || "", handle: row.handle, }; for (const idx of uriIndices.get(row.uri) || []) { resolved[idx] = edge; } } catch {} } } // Resolve handles for inline edges (those with source/target but no did/handle) // Extract DID from AT URI (at://did:plc:.../collection/rkey) and batch-lookup handles const inlineEdges = resolved.filter((e: any) => e && !e.did); if (inlineEdges.length > 0) { const dids = new Set(); for (const edge of inlineEdges) { const uri = (edge as any).uri as string | undefined; if (!uri) continue; const match = uri.match(/^at:\/\/(did:[^/]+)\//); if (match) dids.add(match[1]); } if (dids.size > 0) { const ph = [...dids].map(() => "?").join(","); const rows = await db .prepare(`SELECT did, handle FROM identities WHERE did IN (${ph})`) .bind(...dids) .all<{ did: string; handle: string | null }>(); const handleMap = new Map(); for (const row of rows.results || []) { handleMap.set(row.did, row.handle); } for (const edge of inlineEdges) { const uri = (edge as any).uri as string | undefined; if (!uri) continue; const match = uri.match(/^at:\/\/(did:[^/]+)\//); if (match) { const did = match[1]; (edge as any).did = did; (edge as any).handle = handleMap.get(did) ?? null; } } } } return resolved.filter(Boolean); } // ─── 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 centroid, json_extract(record, '$.analysis.title') as title FROM records_island ORDER BY time_us DESC LIMIT 10") .all<{ uri: string; centroid: string | null; title: string | null }>(); const islands: any[] = []; for (const row of rows.results || []) { if (!row.centroid) continue; const id = row.uri.split("/").pop() || row.uri; islands.push({ id, centroid: row.centroid, 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.island.getIslands", async (c) => { const limit = Math.min(parseInt(c.req.query("limit") || "50"), 100); const summary = c.req.query("summary") === "true"; // No ensureContrailReady — just D1 queries const rows = await db .prepare("SELECT uri, record FROM records_island 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 centroid = rec.source?.uri; if (!centroid) continue; // ID is the record rkey — unique per strata record even if same graph const id = row.uri.split("/").pop() || row.uri; // Build graph from connections — resolve URIs if edges are just {uri} references const connections: any[] = rec.connections || []; const resolvedEdges = await resolveEdgeUris(db, connections); const vertexSet = new Set(); const edges: any[] = []; for (const conn of resolvedEdges) { if (conn.source && conn.target) { vertexSet.add(conn.source); vertexSet.add(conn.target); edges.push(conn); } } const vertices = [...vertexSet].sort(); const analysis = rec.analysis || {}; // Summary mode: minimal payload for explore page canvas // Skip vertexMeta (122KB) and trim edges to just source/target const vertexMeta = summary ? {} : Object.fromEntries(await resolveVertexMeta(db, vertices)); const trimmedEdges = summary ? edges.map(e => ({ source: e.source, target: e.target, connectionType: e.connectionType })) : edges; islands.push({ id, centroid, vertices, edges: trimmedEdges, vertexMeta, title: analysis.title || null, strata: analysis.synthesis || analysis.synthesisDoc ? { prose: analysis.synthesis || "", title: analysis.title || "", themes: analysis.themes || [], relationships: [], highlights: analysis.highlights || [], tensions: analysis.tensions || [], open_questions: analysis.openQuestions || [], synthesis: analysis.synthesis || "", synthesisDoc: analysis.synthesisDoc || null, } : null, recordUri: row.uri, handle: null as string | null, }); } catch {} } // Batch-resolve handles for all island DIDs const dids = new Set(); for (const island of islands) { const parts = island.recordUri.split("/"); if (parts[2]?.startsWith("did:")) dids.add(parts[2]); } if (dids.size > 0) { const placeholders = [...dids].map(() => "?").join(","); const handleRows = await db .prepare(`SELECT did, handle FROM identities WHERE did IN (${placeholders})`) .bind(...dids) .all<{ did: string; handle: string | null }>(); const handleMap = new Map(); for (const r of handleRows.results || []) { if (r.handle) handleMap.set(r.did, r.handle); } for (const island of islands) { const parts = island.recordUri.split("/"); const did = parts[2]; if (did?.startsWith("did:")) island.handle = handleMap.get(did) || null; } } return c.json({ islands }); }); // ── Get single island by ID or AT URI ──────────────────────────── app.get("/xrpc/org.latha.island.getIsland", async (c) => { const id = c.req.query("id"); if (!id) { return c.json({ error: "MissingRequiredParameter", message: "id is required" }, 400); } // Normalize handle-based AT URIs to DID-based URIs // e.g. at://nandi.latha.org/org.latha.island/rkey → at://did:plc:.../org.latha.island/rkey let normalizedId = id; let resolvedHandle: string | null = null; if (id.startsWith("at://")) { const match = id.match(/^at:\/\/([^/]+)\/(.+)$/); if (match && !match[1].startsWith("did:")) { const handle = match[1]; const rest = match[2]; const identity = await db .prepare("SELECT did, handle FROM identities WHERE handle = ?") .bind(handle) .first<{ did: string; handle: string }>(); if (!identity) { return c.json({ error: "IdentityNotFound", message: `Handle ${handle} not found` }, 404); } normalizedId = `at://${identity.did}/${rest}`; resolvedHandle = identity.handle; } } // If it's an AT URI, look up the strata record if (normalizedId.startsWith("at://")) { // Batch: fetch record + resolve handle in one D1 round trip const did = normalizedId.split("/")[2]; const stmts: D1PreparedStatement[] = [ db.prepare("SELECT uri, record FROM records_island WHERE uri = ?").bind(normalizedId), ]; if (!resolvedHandle && did?.startsWith("did:")) { stmts.push(db.prepare("SELECT handle FROM identities WHERE did = ?").bind(did)); } const batchResults = await db.batch(stmts); const row = (batchResults[0] as { results: { uri: string; record: string }[] }).results?.[0]; if (!row) { return c.json({ error: "RecordNotFound" }, 404); } if (!resolvedHandle && batchResults[1]) { const identity = (batchResults[1] as { results: { handle: string }[] }).results?.[0]; resolvedHandle = identity?.handle || null; } const rec = JSON.parse(row.record); const centroid = rec.source?.uri; if (!centroid) { return c.json({ error: "InvalidRecord" }, 400); } // Build graph from record's connections — no BFS needed const connections: any[] = rec.connections || []; const resolvedEdges = await resolveEdgeUris(db, connections); const vertexSet = new Set(); const edges: any[] = []; for (const conn of resolvedEdges) { if (conn.source && conn.target) { vertexSet.add(conn.source); vertexSet.add(conn.target); edges.push(conn); } } const vertices = [...vertexSet].sort(); // Compute centroid and island ID const adj = new Map>(); for (const e of edges) { if (!adj.has(e.source)) adj.set(e.source, new Set()); if (!adj.has(e.target)) adj.set(e.target, new Set()); adj.get(e.source)!.add(e.target); adj.get(e.target)!.add(e.source); } const actualCentroid = computeCentroid(adj, vertexSet); const islandId = await islandHash(actualCentroid); const meta = await resolveVertexMeta(db, vertices); const analysis = rec.analysis || {}; return c.json({ island: { id: islandId, centroid: actualCentroid, vertices, edges, vertexMeta: Object.fromEntries(meta), summary: analysis.title || null, handle: resolvedHandle, rawRecord: rec, strata: analysis.synthesis ? { prose: analysis.synthesis || "", title: analysis.title || "", themes: analysis.themes || [], relationships: [], highlights: analysis.highlights || [], tensions: analysis.tensions || [], open_questions: analysis.openQuestions || [], synthesis: analysis.synthesis || "", synthesisDoc: analysis.synthesisDoc || null, } : null, }, recordUri: row.uri, }); } // ID is the record rkey — look up directly by URI const recordUri = `at://did:plc:ngokl2gnmpbvuvrfckja3g7p/org.latha.island/${id}`; const row = await db .prepare("SELECT uri, record FROM records_island WHERE uri = ?") .bind(recordUri) .first<{ uri: string; record: string }>(); if (row) { try { const rec = JSON.parse(row.record); const centroid = rec.source?.uri; // Build graph from connections — resolve URIs if edges are just {uri} references const connections: any[] = rec.connections || []; const resolvedEdges = await resolveEdgeUris(db, connections); const vertexSet = new Set(); const edges: any[] = []; for (const conn of resolvedEdges) { if (conn.source && conn.target) { vertexSet.add(conn.source); vertexSet.add(conn.target); edges.push(conn); } } const vertices = [...vertexSet].sort(); const meta = await resolveVertexMeta(db, vertices); const analysis = rec.analysis || {}; return c.json({ island: { id, vertices, edges, centroid: centroid || vertices[0], vertexMeta: Object.fromEntries(meta), summary: analysis.title || null, strata: { prose: analysis.synthesis || "", title: analysis.title || "", themes: analysis.themes || [], relationships: [], highlights: analysis.highlights || [], tensions: analysis.tensions || [], open_questions: analysis.openQuestions || [], synthesis: analysis.synthesis || "", synthesisDoc: analysis.synthesisDoc || null, }, recordUri: row.uri, }, }); } catch {} } // No strata record found — try deriving island by treating the ID as a centroid seed // If the ID is a 12-char hex string (centroid hash), scan strata records for a match if (/^[0-9a-f]{12}$/.test(id)) { const rows = await db .prepare("SELECT uri, record FROM records_island ORDER BY time_us DESC LIMIT 500") .all<{ uri: string; record: string }>(); for (const row of rows.results || []) { try { const rec = JSON.parse(row.record); const centroid = rec.source?.uri; if (!centroid) continue; const hash = await islandHash(centroid); if (hash === id) { // Found the record — redirect to the AT URI path const connections: any[] = rec.connections || []; const resolvedEdges = await resolveEdgeUris(db, connections); const vertexSet = new Set(); const edges: any[] = []; for (const conn of resolvedEdges) { if (conn.source && conn.target) { vertexSet.add(conn.source); vertexSet.add(conn.target); edges.push(conn); } } const vertices = [...vertexSet].sort(); const meta = await resolveVertexMeta(db, vertices); const analysis = rec.analysis || {}; return c.json({ island: { id, vertices, edges, centroid, vertexMeta: Object.fromEntries(meta), summary: analysis.title || null, strata: { prose: analysis.synthesis || "", title: analysis.title || "", themes: analysis.themes || [], relationships: [], highlights: analysis.highlights || [], tensions: analysis.tensions || [], open_questions: analysis.openQuestions || [], synthesis: analysis.synthesis || "", synthesisDoc: analysis.synthesisDoc || null, }, }, recordUri: row.uri, }); } } catch {} } } // Fall back: treat ID as a vertex URL for BFS derivation 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, centroid: island.centroid, vertexMeta: Object.fromEntries(meta), summary: null, strata: null, }, recordUri: null, }); }); // ── List strata records (AT URI-based) ────────────────────────── app.get("/xrpc/org.latha.island.listRecords", async (c) => { const rows = await db .prepare("SELECT uri, record FROM records_island 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 }); }); // ── Search islands by URL ──────────────────────────────────────── app.get("/xrpc/org.latha.island.searchIslands", async (c) => { const url = c.req.query("url"); if (!url) { return c.json({ error: "MissingRequiredParameter", message: "url is required" }, 400); } // Normalize the query URL let normalizedUrl: string; let queryDomain: string; try { const u = new URL(url); u.hash = ""; if (u.hostname.startsWith("www.")) u.hostname = u.hostname.slice(4); if (u.pathname.endsWith("/")) u.pathname = u.pathname.slice(0, -1); normalizedUrl = u.toString(); queryDomain = u.hostname.toLowerCase(); } catch { return c.json({ error: "InvalidUrl", message: "Could not parse URL" }, 400); } const limit = Math.min(parseInt(c.req.query("limit") || "50"), 100); const rows = await db .prepare("SELECT uri, record FROM records_island ORDER BY time_us DESC LIMIT ?") .bind(limit) .all<{ uri: string; record: string }>(); const results: any[] = []; for (const row of rows.results || []) { try { const rec = JSON.parse(row.record); const centroid = rec.source?.uri; if (!centroid) continue; const connections: any[] = rec.connections || []; const resolvedEdges = await resolveEdgeUris(db, connections); const vertexSet = new Set(); const edges: any[] = []; for (const conn of resolvedEdges) { if (conn.source && conn.target) { vertexSet.add(conn.source); vertexSet.add(conn.target); edges.push(conn); } } const vertices = [...vertexSet].sort(); // Check for exact and domain matches let matchType: "exact" | "domain" | null = null; const matchedUrls: string[] = []; for (const v of vertices) { let vNorm: string; try { const vu = new URL(v); vu.hash = ""; if (vu.hostname.startsWith("www.")) vu.hostname = vu.hostname.slice(4); if (vu.pathname.endsWith("/")) vu.pathname = vu.pathname.slice(0, -1); vNorm = vu.toString(); } catch { vNorm = v; } if (vNorm === normalizedUrl) { matchType = "exact"; matchedUrls.push(v); } else if (!matchType) { try { const vDomain = new URL(v).hostname.toLowerCase(); if (vDomain === queryDomain) { matchType = "domain"; matchedUrls.push(v); } } catch {} } } if (!matchType) continue; const id = row.uri.split("/").pop() || row.uri; const analysis = rec.analysis || {}; const did = row.uri.split("/")[2] || ""; const handleRow = await db .prepare("SELECT handle FROM identities WHERE did = ?") .bind(did) .first<{ handle: string | null }>(); results.push({ id, centroid, title: analysis.title || null, recordUri: row.uri, handle: handleRow?.handle || null, matchType, matchedUrls, nodeCount: vertices.length, edgeCount: edges.length, edges: edges.map(e => ({ source: e.source, target: e.target, connectionType: e.connectionType })), strata: analysis.synthesis ? { title: analysis.title || "", themes: analysis.themes || [], synthesis: analysis.synthesis || "", tensions: analysis.tensions || [], openQuestions: analysis.openQuestions || [], highlights: analysis.highlights || [], } : null, }); } catch {} } // Sort: exact before domain results.sort((a, b) => (a.matchType === "exact" ? -1 : 1) - (b.matchType === "exact" ? -1 : 1)); return c.json({ url, results }); }); // ── Strata record endpoint (AT URI-based) ────────────────────── app.get("/xrpc/org.latha.island.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_island 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 centroid (source.uri) const centroid = strataRecord.source?.uri; if (!centroid) { return c.json({ error: "MissingLexmin", message: "Strata record has no source.uri" }, 400); } const island = await deriveIsland(db, centroid); if (!island) { return c.json({ error: "IslandNotFound", message: "Could not derive island from centroid" }, 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: [], highlights: analysis.highlights || [], tensions: analysis.tensions || [], open_questions: analysis.openQuestions || [], synthesis: analysis.synthesis || "", synthesisDoc: analysis.synthesisDoc || null, }; return c.json({ island: { ...island, vertexMeta: Object.fromEntries(meta), summary: analysis.title || null, strata: strataData, rawRecord: strataRecord, }, record: strataRecord, recordUri: uri, }); }); // Notify appview of a newly published record — crawl the DID's PDS and upsert into D1 app.post("/xrpc/org.latha.island.notifyOfUpdate", async (c) => { const body = await c.req.json().catch(() => ({})) as { did?: string; uri?: string }; const did = body.did; if (!did) return c.json({ error: "InvalidRequest", message: "did is required" }, 400); await ensureContrailReady(db); // Look up the DID's PDS const identity = await db .prepare("SELECT pds FROM identities WHERE did = ?") .bind(did) .first<{ pds: string }>(); const pds = identity?.pds; if (!pds) return c.json({ error: "IdentityNotFound", message: `DID ${did} not in identities table` }, 404); // Fetch org.latha.island records from the DID's PDS const url = `${pds}/xrpc/com.atproto.repo.listRecords?repo=${did}&collection=org.latha.island&limit=50`; const resp = await fetch(url); if (!resp.ok) return c.json({ error: "PdsError", message: `PDS returned ${resp.status}` }, 502); const data = await resp.json() as { records: Array<{ uri: string; cid: string; value: Record }> }; const records = data.records || []; if (records.length === 0) return c.json({ ingested: 0 }); // Upsert each record into records_island const now = Date.now() * 1000; // time_us const indexedAt = Date.now(); const stmt = db.prepare( `INSERT INTO records_island (uri, did, rkey, cid, record, time_us, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?) ON CONFLICT(uri) DO UPDATE SET cid = excluded.cid, record = excluded.record, time_us = excluded.time_us, indexed_at = excluded.indexed_at` ); let ingested = 0; const batch: D1PreparedStatement[] = []; for (const r of records) { const rkey = r.uri.split("/").pop() || ""; batch.push(stmt.bind(r.uri, did, rkey, r.cid || "", JSON.stringify(r.value), now + ingested, indexedAt)); ingested++; } await db.batch(batch); return c.json({ ingested }); }); // ── 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.island.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}/`, `https://${host}/oauth/callback`], 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, }); }); app.get("/oauth/callback", (c) => c.html(landingPage())); app.all("/debug/echo/*", (c) => c.json({ path: c.req.path, url: c.req.url })); app.all("/debug/echo", (c) => c.json({ path: c.req.path, url: c.req.url })); // Static assets are files only. App routing stays in this Worker. app.get("/app.js", (c) => c.env.ASSETS.fetch(c.req.raw)); app.get("/app.js.map", (c) => c.env.ASSETS.fetch(c.req.raw)); // ── SPA fallback: serve landing page for client-side routes ────── // /island// — inject island data for instant hydration app.get("/island/:repo/:rkey", async (c) => { let repo = c.req.param("repo"); const rkey = c.req.param("rkey"); let islandsJson = "[]"; let islandDataJson = "null"; try { // Batch: island list + specific island record const atUri = repo.startsWith("did:") ? `at://${repo}/org.latha.island/${rkey}` : `at://${repo}/org.latha.island/${rkey}`; // Resolve handle → DID if needed let did = repo; let handle: string | null = null; if (!repo.startsWith("did:")) { const identity = await db .prepare("SELECT did, handle FROM identities WHERE handle = ?") .bind(repo) .first<{ did: string; handle: string }>(); if (identity) { did = identity.did; handle = identity.handle; } } else { const identity = await db .prepare("SELECT handle FROM identities WHERE did = ?") .bind(repo) .first<{ handle: string }>(); handle = identity?.handle || null; } const fullAtUri = `at://${did}/org.latha.island/${rkey}`; // Batch: fetch record + island list const stmts = [ db.prepare("SELECT uri, record FROM records_island WHERE uri = ?").bind(fullAtUri), db.prepare("SELECT uri, json_extract(record, '$.source.uri') as centroid, json_extract(record, '$.analysis.title') as title FROM records_island ORDER BY time_us DESC LIMIT 50"), ]; const [recordResult, listResult] = await db.batch(stmts); // Build island list const listRows = (listResult as { results: { uri: string; centroid: string | null; title: string | null }[] }).results || []; const islands: any[] = []; for (const row of listRows) { if (!row.centroid) continue; const id = row.uri.split("/").pop() || row.uri; const listDid = row.uri.split("/")[2]; // Resolve handle for list items lazily — skip for now, use DID islands.push({ id, centroid: row.centroid, title: row.title || null, recordUri: row.uri, handle: listDid === did ? handle : null }); } islandsJson = JSON.stringify(islands); // Build island data from record const recordRow = (recordResult as { results: { uri: string; record: string }[] }).results?.[0]; if (recordRow) { const rec = JSON.parse(recordRow.record); const centroid = rec.source?.uri; if (centroid) { const connections: any[] = rec.connections || []; const resolvedEdges = await resolveEdgeUris(db, connections); const vertexSet = new Set(); const edges: any[] = []; for (const conn of resolvedEdges) { if (conn.source && conn.target) { vertexSet.add(conn.source); vertexSet.add(conn.target); edges.push(conn); } } const vertices = [...vertexSet].sort(); const adj = new Map>(); for (const e of edges) { if (!adj.has(e.source)) adj.set(e.source, new Set()); if (!adj.has(e.target)) adj.set(e.target, new Set()); adj.get(e.source)!.add(e.target); adj.get(e.target)!.add(e.source); } const actualCentroid = computeCentroid(adj, vertexSet); const islandId = await islandHash(actualCentroid); const meta = await resolveVertexMeta(db, vertices); const analysis = rec.analysis || {}; islandDataJson = JSON.stringify({ island: { id: islandId, centroid: actualCentroid, vertices, edges, vertexMeta: Object.fromEntries(meta), summary: analysis.title || null, handle, strata: analysis.synthesis || analysis.synthesisDoc ? { prose: analysis.synthesis || "", title: analysis.title || "", themes: analysis.themes || [], relationships: [], highlights: analysis.highlights || [], tensions: analysis.tensions || [], open_questions: analysis.openQuestions || [], synthesis: analysis.synthesis || "", synthesisDoc: analysis.synthesisDoc || null, } : null, }, recordUri: recordRow.uri, }); } } } catch (e) { console.error("Failed to load island data:", e); } const page = LANDING_PAGE.replace( "", ``, ); return c.html(page); }); // Other SPA routes — just inject island list for (const path of ["/island", "/strata", "/connect", "/search"]) { app.get(path, async (c) => { let islandsJson = "[]"; try { const rows = await db .prepare("SELECT uri, json_extract(record, '$.source.uri') as centroid, json_extract(record, '$.analysis.title') as title FROM records_island ORDER BY time_us DESC LIMIT 50") .all<{ uri: string; centroid: string | null; title: string | null }>(); const islands: any[] = []; for (const row of rows.results || []) { if (!row.centroid) continue; const id = row.uri.split("/").pop() || row.uri; islands.push({ id, centroid: row.centroid, title: row.title || null, recordUri: row.uri }); } 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, ); }, };