getIsland: resolve 12-char hex IDs as centroid hashes · nandi.uk/stigmergic@5febe8a

+255 -44

1 changed file

Expand all

src

worker.ts

+255 -44

src/worker.ts

··· 41 41 42 42 interface Island { 43 43 id: string; 44 + centroid: string; 44 45 vertices: string[]; 45 46 edges: Array<{ 46 47 uri: string; ··· 53 54 }>; 54 55 } 55 56 57 + /** Compute centroid: vertex with highest closeness centrality (smallest avg BFS distance). */ 58 + function computeCentroid(adj: Map<string, Set<string>>, nodes: Set<string>): string { 59 + if (nodes.size <= 1) return [...nodes][0] ?? ''; 60 + 61 + let bestNode = ''; 62 + let bestAvg = Infinity; 63 + 64 + for (const start of nodes) { 65 + const dist = new Map<string, number>(); 66 + dist.set(start, 0); 67 + const queue = [start]; 68 + while (queue.length > 0) { 69 + const current = queue.shift()!; 70 + for (const neighbor of adj.get(current) || []) { 71 + if (!dist.has(neighbor) && nodes.has(neighbor)) { 72 + dist.set(neighbor, dist.get(current)! + 1); 73 + queue.push(neighbor); 74 + } 75 + } 76 + } 77 + 78 + let totalDist = 0; 79 + let reachable = 0; 80 + for (const [node, d] of dist) { 81 + if (nodes.has(node)) { 82 + totalDist += d; 83 + reachable++; 84 + } 85 + } 86 + 87 + const avg = reachable > 0 ? totalDist / reachable : Infinity; 88 + if (avg < bestAvg) { 89 + bestAvg = avg; 90 + bestNode = start; 91 + } 92 + } 93 + 94 + return bestNode; 95 + } 96 + 97 + /** Hash a URL to a 12-char hex island ID (matches CLI's islandId()). */ 98 + async function islandHash(url: string): Promise<string> { 99 + const hash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(url)); 100 + return Array.from(new Uint8Array(hash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 12); 101 + } 102 + 56 103 async function detectIslands(db: D1Database): Promise<Island[]> { 57 104 await ensureContrailReady(db); 58 105 ··· 120 167 e => componentSet.has(e.source) && componentSet.has(e.target), 121 168 ); 122 169 123 - // Stable ID: truncated SHA-256 of the lexicographic minimum vertex 124 - const lexmin = component.sort()[0]; 125 - const lexminHash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(lexmin)); 126 - const id = Array.from(new Uint8Array(lexminHash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 16); 170 + // Stable ID: truncated SHA-256 of the centroid vertex (highest closeness centrality) 171 + const centroid = computeCentroid(adj, componentSet); 172 + const id = await islandHash(centroid); 127 173 128 174 islands.push({ 129 175 id, 176 + centroid, 130 177 vertices: component, 131 178 edges: componentEdges, 132 179 }); ··· 269 316 return meta; 270 317 } 271 318 272 - // ─── Derive island from lexmin ──────────────────────────────────── 319 + // ─── Derive island from centroid ──────────────────────────────────── 273 320 // 274 - // Given a lexmin vertex (canonical component reference), derive the 321 + // Given a centroid vertex (canonical component reference), derive the 275 322 // connected component by traversing network.cosmik.connection records. 276 323 277 324 async function deriveIsland( ··· 362 409 }); 363 410 } 364 411 365 - // Compute stable ID from lexmin 366 - const sortedVertices = [...vertices].sort(); 367 - const actualLexmin = sortedVertices[0]; 368 - const lexminHash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(actualLexmin)); 369 - const id = Array.from(new Uint8Array(lexminHash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 16); 412 + // Compute stable ID from centroid (highest closeness centrality) 413 + const adj = new Map<string, Set<string>>(); 414 + for (const e of edges) { 415 + if (!adj.has(e.source)) adj.set(e.source, new Set()); 416 + if (!adj.has(e.target)) adj.set(e.target, new Set()); 417 + adj.get(e.source)!.add(e.target); 418 + adj.get(e.target)!.add(e.source); 419 + } 420 + const centroid = computeCentroid(adj, vertexSet); 421 + const id = await islandHash(centroid); 370 422 371 - return { id, vertices, edges }; 423 + return { id, centroid, vertices, edges }; 372 424 } 373 425 374 426 // ─── List strata islands from records_strata ────────────────────── 375 427 428 + // ─── Resolve edge URIs to full edge objects ────────────────────── 429 + 430 + /** Given a list of connections (either full objects or just {uri}), resolve URIs to full edges from D1. */ 431 + async function resolveEdgeUris(db: D1Database, connections: any[]): Promise<any[]> { 432 + const resolved: any[] = []; 433 + const uriIndices: Map<string, number[]> = new Map(); // uri → indices in connections that need it 434 + 435 + for (let i = 0; i < connections.length; i++) { 436 + const conn = connections[i]; 437 + // Already has source/target — use as-is 438 + if (typeof conn === "object" && conn.source && conn.target) { 439 + resolved.push(conn); 440 + continue; 441 + } 442 + // Just a URI reference — needs lookup 443 + const uri = conn.uri; 444 + if (!uri) continue; 445 + resolved.push(null); // placeholder 446 + if (!uriIndices.has(uri)) uriIndices.set(uri, []); 447 + uriIndices.get(uri)!.push(i); 448 + } 449 + 450 + if (uriIndices.size === 0) return resolved.filter(Boolean); 451 + 452 + // Batch-fetch connection records from D1 453 + const uris = [...uriIndices.keys()]; 454 + const placeholders = uris.map(() => "?").join(","); 455 + const rows = await db 456 + .prepare( 457 + `SELECT r.uri, r.record, r.did, r.rkey, i.handle 458 + FROM records_connection r 459 + LEFT JOIN identities i ON r.did = i.did 460 + WHERE r.uri IN (${placeholders})`, 461 + ) 462 + .bind(...uris) 463 + .all<{ uri: string; record: string; did: string; rkey: string; handle: string | null }>(); 464 + 465 + for (const row of rows.results || []) { 466 + try { 467 + const value = JSON.parse(row.record); 468 + const edge = { 469 + uri: row.uri, 470 + did: row.did, 471 + source: value.source, 472 + target: value.target, 473 + connectionType: value.connectionType || "relates", 474 + note: value.note || "", 475 + handle: row.handle, 476 + }; 477 + for (const idx of uriIndices.get(row.uri) || []) { 478 + resolved[idx] = edge; 479 + } 480 + } catch {} 481 + } 482 + 483 + return resolved.filter(Boolean); 484 + } 485 + 376 486 // ─── Hono app ─────────────────────────────────────────────────────── 377 487 378 488 let app: Hono<{ Bindings: Env }> | null = null; ··· 390 500 let islandsJson = "[]"; 391 501 try { 392 502 const rows = await db 393 - .prepare("SELECT uri, json_extract(record, '$.source.uri') as lexmin, json_extract(record, '$.analysis.title') as title FROM records_strata ORDER BY time_us DESC LIMIT 10") 394 - .all<{ uri: string; lexmin: string | null; title: string | null }>(); 503 + .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") 504 + .all<{ uri: string; centroid: string | null; title: string | null }>(); 395 505 396 506 const islands: any[] = []; 397 507 for (const row of rows.results || []) { 398 - if (!row.lexmin) continue; 508 + if (!row.centroid) continue; 399 509 const id = row.uri.split("/").pop() || row.uri; 400 - islands.push({ id, lexmin: row.lexmin, title: row.title || null, recordUri: row.uri }); 510 + islands.push({ id, centroid: row.centroid, title: row.title || null, recordUri: row.uri }); 401 511 } 402 512 islandsJson = JSON.stringify(islands); 403 513 } catch (e) { ··· 427 537 for (const row of rows.results || []) { 428 538 try { 429 539 const rec = JSON.parse(row.record); 430 - const lexmin = rec.source?.uri; 431 - if (!lexmin) continue; 540 + const centroid = rec.source?.uri; 541 + if (!centroid) continue; 432 542 433 543 // ID is the record rkey — unique per strata record even if same graph 434 544 const id = row.uri.split("/").pop() || row.uri; 435 545 436 - // Build graph directly from embedded connections — no D1 lookups needed 546 + // Build graph from connections — resolve URIs if edges are just {uri} references 437 547 const connections: any[] = rec.connections || []; 548 + const resolvedEdges = await resolveEdgeUris(db, connections); 438 549 const vertexSet = new Set<string>(); 439 550 const edges: any[] = []; 440 551 441 - for (const conn of connections) { 442 - if (typeof conn === "object" && conn.source && conn.target) { 552 + for (const conn of resolvedEdges) { 553 + if (conn.source && conn.target) { 443 554 vertexSet.add(conn.source); 444 555 vertexSet.add(conn.target); 445 556 edges.push(conn); ··· 457 568 : edges; 458 569 459 570 islands.push({ 460 - id, lexmin, vertices, edges: trimmedEdges, vertexMeta, 571 + id, centroid, vertices, edges: trimmedEdges, vertexMeta, 461 572 title: analysis.title || null, 462 573 strata: analysis.synthesis ? { 463 574 prose: analysis.synthesis || "", title: analysis.title || "", ··· 486 597 487 598 await ensureContrailReady(db); 488 599 489 - // If it's an AT URI, look up the strata record and derive from its lexmin 600 + // If it's an AT URI, look up the strata record and derive from its centroid 490 601 if (id.startsWith("at://")) { 491 602 const row = await db 492 603 .prepare("SELECT uri, record FROM records_strata WHERE uri = ?") ··· 498 609 } 499 610 500 611 const rec = JSON.parse(row.record); 501 - const lexmin = rec.source?.uri; 502 - if (!lexmin) { 612 + const centroid = rec.source?.uri; 613 + if (!centroid) { 503 614 return c.json({ error: "InvalidRecord" }, 400); 504 615 } 505 616 506 - const island = await deriveIsland(db, lexmin); 617 + const island = await deriveIsland(db, centroid); 507 618 if (!island) { 508 619 return c.json({ error: "IslandNotFound" }, 404); 509 620 } ··· 514 625 return c.json({ 515 626 island: { 516 627 ...island, 517 - lexmin: island.vertices.slice().sort()[0], 628 + centroid: island.centroid, 518 629 vertexMeta: Object.fromEntries(meta), 519 630 summary: analysis.title || null, 520 631 strata: { ··· 542 653 if (row) { 543 654 try { 544 655 const rec = JSON.parse(row.record); 545 - const lexmin = rec.source?.uri; 656 + const centroid = rec.source?.uri; 546 657 547 - // Build graph from embedded connections 658 + // Build graph from connections — resolve URIs if edges are just {uri} references 548 659 const connections: any[] = rec.connections || []; 660 + const resolvedEdges = await resolveEdgeUris(db, connections); 549 661 const vertexSet = new Set<string>(); 550 662 const edges: any[] = []; 551 - for (const conn of connections) { 552 - if (typeof conn === "object" && conn.source && conn.target) { 663 + for (const conn of resolvedEdges) { 664 + if (conn.source && conn.target) { 553 665 vertexSet.add(conn.source); 554 666 vertexSet.add(conn.target); 555 667 edges.push(conn); ··· 564 676 id, 565 677 vertices, 566 678 edges, 567 - lexmin: lexmin || vertices[0], 679 + centroid: centroid || vertices[0], 568 680 vertexMeta: Object.fromEntries(meta), 569 681 summary: analysis.title || null, 570 682 strata: { ··· 583 695 } catch {} 584 696 } 585 697 586 - // No strata record found — try deriving island by treating the ID as a lexmin seed 698 + // No strata record found — try deriving island by treating the ID as a centroid seed 699 + // If the ID is a 12-char hex string (centroid hash), scan strata records for a match 700 + if (/^[0-9a-f]{12}$/.test(id)) { 701 + const rows = await db 702 + .prepare("SELECT uri, record FROM records_strata ORDER BY time_us DESC LIMIT 500") 703 + .all<{ uri: string; record: string }>(); 704 + for (const row of rows.results || []) { 705 + try { 706 + const rec = JSON.parse(row.record); 707 + const centroid = rec.source?.uri; 708 + if (!centroid) continue; 709 + const hash = await islandHash(centroid); 710 + if (hash === id) { 711 + // Found the record — redirect to the AT URI path 712 + const connections: any[] = rec.connections || []; 713 + const resolvedEdges = await resolveEdgeUris(db, connections); 714 + const vertexSet = new Set<string>(); 715 + const edges: any[] = []; 716 + for (const conn of resolvedEdges) { 717 + if (conn.source && conn.target) { 718 + vertexSet.add(conn.source); 719 + vertexSet.add(conn.target); 720 + edges.push(conn); 721 + } 722 + } 723 + const vertices = [...vertexSet].sort(); 724 + const meta = await resolveVertexMeta(db, vertices); 725 + const analysis = rec.analysis || {}; 726 + return c.json({ 727 + island: { 728 + id, 729 + vertices, 730 + edges, 731 + centroid, 732 + vertexMeta: Object.fromEntries(meta), 733 + summary: analysis.title || null, 734 + strata: { 735 + prose: analysis.synthesis || "", 736 + title: analysis.title || "", 737 + themes: analysis.themes || [], 738 + relationships: [], 739 + highlights: analysis.highlights || [], 740 + tensions: analysis.tensions || [], 741 + open_questions: analysis.openQuestions || [], 742 + synthesis: analysis.synthesis || "", 743 + }, 744 + }, 745 + recordUri: row.uri, 746 + }); 747 + } 748 + } catch {} 749 + } 750 + } 751 + 752 + // Fall back: treat ID as a vertex URL for BFS derivation 587 753 const island = await deriveIsland(db, id); 588 754 if (!island) { 589 755 return c.json({ error: "IslandNotFound" }, 404); ··· 593 759 return c.json({ 594 760 island: { 595 761 ...island, 596 - lexmin: island.vertices.slice().sort()[0], 762 + centroid: island.centroid, 597 763 vertexMeta: Object.fromEntries(meta), 598 764 summary: null, 599 765 strata: null, ··· 652 818 return c.json({ error: "InvalidRecord" }, 500); 653 819 } 654 820 655 - // Derive island from lexmin (source.uri) 656 - const lexmin = strataRecord.source?.uri; 657 - if (!lexmin) { 821 + // Derive island from centroid (source.uri) 822 + const centroid = strataRecord.source?.uri; 823 + if (!centroid) { 658 824 return c.json({ error: "MissingLexmin", message: "Strata record has no source.uri" }, 400); 659 825 } 660 826 661 - const island = await deriveIsland(db, lexmin); 827 + const island = await deriveIsland(db, centroid); 662 828 if (!island) { 663 - return c.json({ error: "IslandNotFound", message: "Could not derive island from lexmin" }, 404); 829 + return c.json({ error: "IslandNotFound", message: "Could not derive island from centroid" }, 404); 664 830 } 665 831 666 832 // Resolve vertex metadata ··· 813 979 } catch (e: any) { 814 980 return c.json({ error: "ContainerError", message: e.message }, 500); 815 981 } 982 + }); 983 + 984 + // Notify appview of a newly published record — crawl the DID's PDS and upsert into D1 985 + app.post("/xrpc/org.latha.island.notifyOfUpdate", async (c) => { 986 + const body = await c.req.json().catch(() => ({})) as { did?: string; uri?: string }; 987 + const did = body.did; 988 + if (!did) return c.json({ error: "InvalidRequest", message: "did is required" }, 400); 989 + 990 + await ensureContrailReady(db); 991 + 992 + // Look up the DID's PDS 993 + const identity = await db 994 + .prepare("SELECT pds FROM identities WHERE did = ?") 995 + .bind(did) 996 + .first<{ pds: string }>(); 997 + 998 + const pds = identity?.pds; 999 + if (!pds) return c.json({ error: "IdentityNotFound", message: `DID ${did} not in identities table` }, 404); 1000 + 1001 + // Fetch org.latha.island records from the DID's PDS 1002 + const url = `${pds}/xrpc/com.atproto.repo.listRecords?repo=${did}&collection=org.latha.island&limit=50`; 1003 + const resp = await fetch(url); 1004 + if (!resp.ok) return c.json({ error: "PdsError", message: `PDS returned ${resp.status}` }, 502); 1005 + 1006 + const data = await resp.json() as { records: Array<{ uri: string; cid: string; value: Record<string, unknown> }> }; 1007 + const records = data.records || []; 1008 + if (records.length === 0) return c.json({ ingested: 0 }); 1009 + 1010 + // Upsert each record into records_strata 1011 + const now = Date.now() * 1000; // time_us 1012 + const indexedAt = Date.now(); 1013 + const stmt = db.prepare( 1014 + `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` 1015 + ); 1016 + 1017 + let ingested = 0; 1018 + const batch: D1PreparedStatement[] = []; 1019 + for (const r of records) { 1020 + const rkey = r.uri.split("/").pop() || ""; 1021 + batch.push(stmt.bind(r.uri, did, rkey, r.cid || "", JSON.stringify(r.value), now + ingested, indexedAt)); 1022 + ingested++; 1023 + } 1024 + 1025 + await db.batch(batch); 1026 + return c.json({ ingested }); 816 1027 }); 817 1028 818 1029 // Container health + data status ··· 1121 1332 for (const row of rows.results || []) { 1122 1333 try { 1123 1334 const rec = JSON.parse(row.record); 1124 - const lexmin = rec.source?.uri; 1125 - if (!lexmin) continue; 1335 + const centroid = rec.source?.uri; 1336 + if (!centroid) continue; 1126 1337 const id = row.uri.split("/").pop() || row.uri; 1127 - // Compute island hash from lexmin (same as island-shared.ts) 1338 + // Compute island hash from centroid (same as island-shared.ts) 1128 1339 const encoder = new TextEncoder(); 1129 - const data = encoder.encode(lexmin); 1340 + const data = encoder.encode(centroid); 1130 1341 const hashBuffer = await crypto.subtle.digest("SHA-256", data); 1131 1342 const hashArray = Array.from(new Uint8Array(hashBuffer)); 1132 1343 const islandHash = hashArray.map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 12); 1133 - islands.push({ id, islandHash, lexmin, title: rec.analysis?.title || null, recordUri: row.uri }); 1344 + islands.push({ id, islandHash, centroid, title: rec.analysis?.title || null, recordUri: row.uri }); 1134 1345 } catch {} 1135 1346 } 1136 1347 islandsJson = JSON.stringify(islands);

Configure Feed

Configure Feed