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;
  PDS_APP_PASSWORD: string;
  CARRY_GITHUB_TOKEN: string;
  OBSIDIAN_AUTH_TOKEN: string;
  OBSIDIAN_ENCRYPTION_KEY: string;
  OBSIDIAN_ENCRYPTION_SALT: string;
}

// ─── Contrail setup ─────────────────────────────────────────────────

const contrailWorker = createWorker(config, { lexicons });
const contrail = new Contrail(config);
let contrailReady = false;

async function ensureContrailReady(db: D1Database): Promise<void> {
  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<string, Set<string>>, nodes: Set<string>): string {
  if (nodes.size <= 1) return [...nodes][0] ?? '';

  let bestNode = '';
  let bestAvg = Infinity;

  for (const start of nodes) {
    const dist = new Map<string, number>();
    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<string> {
  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<Island[]> {
  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<string, Set<string>>();
  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<string>();
  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<Map<string, { title: string; description: string; type: string }>> {
  const meta = new Map<string, { title: string; description: string; type: string }>();
  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<Island | null> {
  await ensureContrailReady(db);

  // BFS in application code — recursive CTE OOMs in D1 due to cross join
  const visited = new Set<string>();
  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<string>();
  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<string, Set<string>>();
  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 strata islands from records_strata ──────────────────────

// ─── 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<any[]> {
  const resolved: any[] = [];
  const uriIndices: Map<string, number[]> = 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);
  }

  if (uriIndices.size === 0) return resolved.filter(Boolean);

  // Batch-fetch connection records from D1
  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 {}
  }

  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_strata 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(
      "</head>",
      `<script>window.__ISLANDS__=${islandsJson};</script></head>`,
    );
    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_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 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<string>();
        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 ? {
            prose: analysis.synthesis || "", title: analysis.title || "",
            themes: analysis.themes || [],
            relationships: [],
            highlights: analysis.highlights || [],
            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.island.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 centroid
    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 centroid = rec.source?.uri;
      if (!centroid) {
        return c.json({ error: "InvalidRecord" }, 400);
      }

      const island = await deriveIsland(db, centroid);
      if (!island) {
        return c.json({ error: "IslandNotFound" }, 404);
      }

      const meta = await resolveVertexMeta(db, island.vertices);
      const analysis = rec.analysis || {};

      return c.json({
        island: {
          ...island,
        centroid: island.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 || "",
          },
        },
        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_strata 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<string>();
        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 || "",
            },
            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_strata 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<string>();
            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 || "",
                },
              },
              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_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.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_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 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 || "",
    };

    return c.json({
      island: {
        ...island,
        vertexMeta: Object.fromEntries(meta),
        summary: analysis.title || null,
        strata: strataData,
      },
      record: strataRecord,
      recordUri: uri,
    });
  });

  // ── Container-backed xrpc methods ────────────────────────────────

  // Helper: read carry data from D1 for analysis
  async function getCarryData(db: D1Database, did?: string): Promise<{
    entities: Array<{ atUri: string; name: string; entityType: string; description?: string; url?: string }>;
    citations: Array<{ atUri: string; url: string; title: string; takeaway: string; confidence?: string }>;
    reasonings: Array<{ atUri: string; claim: string; evidence: string; reasoningType?: string; url?: string }>;
    connections: Array<{ atUri: string; source: string; target: string; relation: string; sourceUrl: string; targetUrl: string; context: string }>;
  }> {
    const entities: Array<{ atUri: string; name: string; entityType: string; description?: string; url?: string }> = [];
    const citations: Array<{ atUri: string; url: string; title: string; takeaway: string; confidence?: string }> = [];
    const reasonings: Array<{ atUri: string; claim: string; evidence: string; reasoningType?: string; url?: string }> = [];
    const connections: Array<{ atUri: string; source: string; target: string; relation: string; sourceUrl: string; targetUrl: string; context: string }> = [];

    try {
      // Read entities
      const entityRows = await db
        .prepare("SELECT uri, record FROM records_entity ORDER BY time_us DESC LIMIT 100")
        .all<{ uri: string; record: string }>();
      for (const row of entityRows.results || []) {
        try {
          const r = JSON.parse(row.record);
          entities.push({
            atUri: row.uri,
            name: r.name || "",
            entityType: r.entityType || "",
            description: r.description || "",
            url: r.url || "",
          });
        } catch {}
      }

      // Read citations
      const citeRows = await db
        .prepare("SELECT uri, record FROM records_citation ORDER BY time_us DESC LIMIT 100")
        .all<{ uri: string; record: string }>();
      for (const row of citeRows.results || []) {
        try {
          const r = JSON.parse(row.record);
          citations.push({
            atUri: row.uri,
            url: r.url || "",
            title: r.title || "",
            takeaway: r.takeaway || "",
            confidence: r.confidence || "",
          });
        } catch {}
      }

      // Read reasonings
      const reasonRows = await db
        .prepare("SELECT uri, record FROM records_reasoning ORDER BY time_us DESC LIMIT 100")
        .all<{ uri: string; record: string }>();
      for (const row of reasonRows.results || []) {
        try {
          const r = JSON.parse(row.record);
          reasonings.push({
            atUri: row.uri,
            claim: r.claim || "",
            evidence: r.evidence || "",
            reasoningType: r.reasoningType || "",
            url: r.url || "",
          });
        } catch {}
      }
    } catch (e: any) {
      console.error("getCarryData error:", e.message);
    }

    // Read connections from D1
    try {
      const connRows = await db
        .prepare("SELECT uri, record FROM records_connection ORDER BY time_us DESC LIMIT 100")
        .all<{ uri: string; record: string }>();
      for (const row of connRows.results || []) {
        try {
          const r = JSON.parse(row.record);
          connections.push({
            atUri: row.uri,
            source: r.source || "",
            target: r.target || "",
            relation: r.relation || "",
            sourceUrl: r.sourceUrl || "",
            targetUrl: r.targetUrl || "",
            context: r.context || "",
          });
        } catch {}
      }
    } catch (e: any) {
      console.error("getCarryData connections error:", e.message);
    }

    return { entities, citations, reasonings, connections };
  }

  // Helper: call the strata container
  async function callContainer(path: string, method: string, body?: any): Promise<any> {
    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));
    const text = await resp.text();
    try {
      return JSON.parse(text);
    } catch {
      throw new Error(`Container returned non-JSON: ${text.slice(0, 500)}`);
    }
  }

  // Sync container data (vault + carry)
  app.post("/xrpc/org.latha.island.syncContainer", async (c) => {
    try {
      const result = await callContainer("/sync", "POST");
      return c.json(result);
    } catch (e: any) {
      return c.json({ error: "ContainerError", message: e.message }, 500);
    }
  });

  // 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<string, unknown> }> };
    const records = data.records || [];
    if (records.length === 0) return c.json({ ingested: 0 });

    // Upsert each record into records_strata
    const now = Date.now() * 1000; // time_us
    const indexedAt = Date.now();
    const stmt = db.prepare(
      `INSERT INTO records_strata (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 });
  });

  // Container health + data status
  app.get("/xrpc/org.latha.island.containerHealth", async (c) => {
    try {
      const result = await callContainer("/health", "GET");
      return c.json(result);
    } catch (e: any) {
      return c.json({ error: "ContainerError", message: e.message }, 500);
    }
  });

  // Derive islands — POST, delegates to container
  // With seed: single island. Without: all islands.
  app.post("/xrpc/org.latha.island.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.island.analyze", async (c) => {
    const body = await c.req.json().catch(() => ({}));
    if (!body.island) {
      return c.json({ error: "MissingRequiredField", message: "island is required" }, 400);
    }

    try {
      // Read carry data from D1 for the user's DID
      const carryData = await getCarryData(db, body.did);

      // Pass carry data to container alongside island
      const payload = { ...body, carryData };
      const result = await callContainer("/analyze", "POST", payload);

      // Create new connection records on PDS
      const newConnections: Array<{ source: string; target: string; connectionType: string; note: string }> = result.newConnections || [];
      const createdUris: string[] = [];

      if (newConnections.length > 0) {
        const appPassword = env.PDS_APP_PASSWORD;
        if (appPassword && body.did) {
          const pdsHost = "amanita.us-east.host.bsky.network";
          const authResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.server.createSession`, {
            method: "POST",
            headers: { "Content-Type": "application/json" },
            body: JSON.stringify({ identifier: body.did, password: appPassword }),
          });
          if (authResp.ok) {
            const { accessJwt } = await authResp.json() as { accessJwt: string };

            for (const conn of newConnections) {
              const rkey = crypto.randomUUID().replace(/-/g, "").slice(0, 13); // TID format
              const record = {
                "$type": "network.cosmik.connection",
                source: conn.source,
                target: conn.target,
                connectionType: conn.connectionType,
                note: conn.note,
                createdAt: new Date().toISOString(),
              };

              const createResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.repo.createRecord`, {
                method: "POST",
                headers: {
                  "Content-Type": "application/json",
                  "Authorization": `Bearer ${accessJwt}`,
                },
                body: JSON.stringify({
                  repo: body.did,
                  collection: "network.cosmik.connection",
                  rkey,
                  record,
                }),
              });

              if (createResp.ok) {
                const createResult = await createResp.json() as { uri: string };
                createdUris.push(createResult.uri);
              }
            }
          }
        }
      }

      // Add new connection URIs to highlights
      result.highlights = [...(result.highlights || []), ...createdUris];
      result.createdConnections = createdUris;

      return c.json(result);
    } catch (e: any) {
      return c.json({ error: "ContainerError", message: e.message }, 500);
    }
  });

  // ── Update strata record (write back to PDS) ────────────────────
  app.put("/xrpc/org.latha.island.updateRecord", async (c) => {
    const body = await c.req.json().catch(() => ({})) as {
      uri?: string;
      analysis?: Record<string, any>;
    };
    if (!body.uri || !body.analysis) {
      return c.json({ error: "MissingRequiredField", message: "uri and analysis are required" }, 400);
    }

    const appPassword = env.PDS_APP_PASSWORD;
    if (!appPassword) {
      return c.json({ error: "NotConfigured", message: "PDS_APP_PASSWORD not set" }, 500);
    }

    // Parse the AT URI to extract repo DID, collection, rkey
    const parts = body.uri.replace("at://", "").split("/");
    const repoDid = parts[0];
    const collection = parts[1];
    const rkey = parts[2];
    if (!repoDid || !collection || !rkey) {
      return c.json({ error: "InvalidUri", message: "Expected at://did:.../org.latha.island/rkey" }, 400);
    }

    // Read existing record from D1
    const row = await db
      .prepare("SELECT record FROM records_strata WHERE uri = ?")
      .bind(body.uri)
      .first<{ record: string }>();
    if (!row) {
      return c.json({ error: "RecordNotFound" }, 404);
    }

    // Merge updated analysis into existing record
    const record = JSON.parse(row.record);
    record.analysis = { ...record.analysis, ...body.analysis };
    // Remove stale fields that are no longer in the lexicon
    delete record.analysis.connections;
    delete record.connections;
    record.updatedAt = new Date().toISOString();

    // Write back to PDS via com.atproto.repo.putRecord
    const pdsHost = "amanita.us-east.host.bsky.network";
    const authResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.server.createSession`, {
      method: "POST",
      headers: { "Content-Type": "application/json" },
      body: JSON.stringify({ identifier: repoDid, password: appPassword }),
    });
    if (!authResp.ok) {
      const err = await authResp.text();
      return c.json({ error: "AuthFailed", message: err }, 500);
    }
    const { accessJwt } = await authResp.json() as { accessJwt: string };

    const putResp = await fetch(`https://${pdsHost}/xrpc/com.atproto.repo.putRecord`, {
      method: "POST",
      headers: {
        "Content-Type": "application/json",
        "Authorization": `Bearer ${accessJwt}`,
      },
      body: JSON.stringify({
        repo: repoDid,
        collection,
        rkey,
        record,
      }),
    });
    if (!putResp.ok) {
      const err = await putResp.text();
      return c.json({ error: "PutRecordFailed", message: err }, 500);
    }

    // Update D1 cache
    await db
      .prepare("UPDATE records_strata SET record = ? WHERE uri = ?")
      .bind(JSON.stringify(record), body.uri)
      .run();

    return c.json({ ok: true, uri: body.uri });
  });

  // ── 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}/`],
      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", "/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 centroid = rec.source?.uri;
            if (!centroid) continue;
            const id = row.uri.split("/").pop() || row.uri;
            // Compute island hash from centroid (same as island-shared.ts)
            const encoder = new TextEncoder();
            const data = encoder.encode(centroid);
            const hashBuffer = await crypto.subtle.digest("SHA-256", data);
            const hashArray = Array.from(new Uint8Array(hashBuffer));
            const islandHash = hashArray.map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 12);
            islands.push({ id, islandHash, centroid, title: rec.analysis?.title || null, recordUri: row.uri });
          } catch {}
        }
        islandsJson = JSON.stringify(islands);
      } catch {}

      const page = LANDING_PAGE.replace(
        "</head>",
        `<script>window.__ISLANDS__=${islandsJson};</script></head>`,
      );
      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<string, unknown>,
    );
    return response;
  });

  return app;
}

// ─── Export ─────────────────────────────────────────────────────────

export { StrataContainer };

export default {
  fetch(request: Request, env: Env): Response | Promise<Response> {
    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<string, unknown>,
      ctx,
    );
  },
};