refactor: remove island_cache, query records_strata directly · nandi.uk/stigmergic@bdf84a3

+38 -117

1 changed file

Expand all

src

worker.ts

+38 -117

src/worker.ts

··· 363 363 return { id, vertices, edges }; 364 364 } 365 365 366 - // ─── Island cache ────────────────────────────────────────────────── 367 - // 368 - // The scheduled handler computes islands and writes them to the 369 - // `island_cache` table. The read path just queries this table 370 - // instead of recomputing on every request. 366 + // ─── List strata islands from records_strata ────────────────────── 371 367 372 - async function rebuildIslandCache(db: D1Database): Promise<void> { 368 + async function listStrataIslands(db: D1Database): Promise<any[]> { 373 369 await ensureContrailReady(db); 374 - const islands = await detectIslands(db); 375 370 376 - // Enrich with vertex metadata (batched) 377 - const enriched = []; 378 - for (const island of islands.slice(0, 50)) { 379 - const meta = await resolveVertexMeta(db, island.vertices); 380 - enriched.push({ 381 - ...island, 382 - vertexMeta: Object.fromEntries(meta), 383 - }); 384 - } 371 + const rows = await db 372 + .prepare("SELECT uri, record FROM records_strata ORDER BY time_us DESC LIMIT 50") 373 + .all<{ uri: string; record: string }>(); 385 374 386 - // Preserve existing summaries + strata across rebuild 387 - // Match by lexmin vertex: since the ID is sha256(lexmin)[:16], we can 388 - // match old rows to new islands by computing the new ID from the old data. 389 - const existing = await db 390 - .prepare("SELECT id, data, summary, strata FROM island_cache") 391 - .all<{ id: string; data: string; summary: string | null; strata: string | null }>(); 392 - 393 - const preserved = new Map<string, { summary: string | null; strata: string | null }>(); 394 - for (const row of existing.results || []) { 375 + const results: any[] = []; 376 + for (const row of rows.results || []) { 395 377 try { 396 - const oldIsland = JSON.parse(row.data); 397 - const oldLexmin = oldIsland.vertices?.sort()?.[0]; 398 - if (!oldLexmin) continue; 399 - // Compute the new ID for this old island 400 - const hash = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(oldLexmin)); 401 - const newId = Array.from(new Uint8Array(hash)).map(b => b.toString(16).padStart(2, "0")).join("").slice(0, 16); 402 - if (row.summary || row.strata) { 403 - preserved.set(newId, { summary: row.summary, strata: row.strata }); 404 - } 405 - } catch {} 406 - } 378 + const rec = JSON.parse(row.record); 379 + const lexmin = rec.source?.uri; 380 + if (!lexmin) continue; 407 381 408 - const now = Date.now() * 1000; 382 + const island = await deriveIsland(db, lexmin); 383 + if (!island) continue; 409 384 410 - // Clear old cache and write new 411 - await db.prepare("DELETE FROM island_cache").run(); 385 + const meta = await resolveVertexMeta(db, island.vertices); 386 + const analysis = rec.analysis || {}; 412 387 413 - const stmts = enriched.map(island => { 414 - const saved = preserved.get(island.id); 415 - return db 416 - .prepare( 417 - `INSERT INTO island_cache (id, vertex_count, edge_count, data, summary, strata, updated_at) 418 - VALUES (?, ?, ?, ?, ?, ?, ?)`, 419 - ) 420 - .bind( 421 - island.id, 422 - island.vertices.length, 423 - island.edges.length, 424 - JSON.stringify(island), 425 - saved?.summary || null, 426 - saved?.strata || null, 427 - now, 428 - ); 429 - }); 430 - 431 - // D1 supports batched statements 432 - if (stmts.length > 0) { 433 - await db.batch(stmts); 388 + results.push({ 389 + ...island, 390 + vertexMeta: Object.fromEntries(meta), 391 + summary: analysis.title || null, 392 + strata: { 393 + prose: analysis.synthesis || "", 394 + title: analysis.title || "", 395 + themes: analysis.themes || [], 396 + relationships: (analysis.connections || []).map((c: any) => c.description || ""), 397 + tensions: analysis.tensions || [], 398 + open_questions: analysis.openQuestions || [], 399 + synthesis: analysis.synthesis || "", 400 + }, 401 + recordUri: row.uri, 402 + }); 403 + } catch {} 434 404 } 435 - } 436 405 437 - async function readIslandCache(db: D1Database): Promise<any[]> { 438 - const rows = await db 439 - .prepare( 440 - `SELECT data, summary, strata FROM island_cache ORDER BY vertex_count DESC LIMIT 50`, 441 - ) 442 - .all<{ data: string; summary: string | null; strata: string | null }>(); 443 - 444 - return (rows.results || []).map(r => { 445 - const island = JSON.parse(r.data); 446 - island.summary = r.summary || null; 447 - island.strata = r.strata ? JSON.parse(r.strata) : null; 448 - return island; 449 - }); 406 + return results; 450 407 } 451 408 409 + // ─── Island cache ────────────────────────────────────────────────── 410 + // 452 411 // ─── Hono app ─────────────────────────────────────────────────────── 453 412 454 413 let app: Hono<{ Bindings: Env }> | null = null; ··· 459 418 460 419 app.use("*", cors()); 461 420 462 - // Landing page — islands feed from cache 421 + // Landing page — strata records feed 463 422 app.get("/", async (c) => { 464 423 let islandsJson = "[]"; 465 424 try { 466 - const islands = await readIslandCache(db); 425 + const islands = await listStrataIslands(db); 467 426 islandsJson = JSON.stringify(islands); 468 427 } catch (e) { 469 428 console.error("Failed to load islands:", e); ··· 479 438 // ── Islands API ────────────────────────────────────────────────── 480 439 481 440 app.get("/xrpc/org.latha.strata.getIslands", async (c) => { 482 - const islands = await readIslandCache(db); 441 + const islands = await listStrataIslands(db); 483 442 return c.json({ islands }); 484 443 }); 485 444 ··· 508 467 return c.json({ records }); 509 468 }); 510 469 511 - // Update an island's summary 512 - app.put("/xrpc/org.latha.strata.putSummary", async (c) => { 513 - const id = c.req.query("id"); 514 - if (!id) { 515 - return c.json({ error: "MissingRequiredParameter", message: "id is required" }, 400); 516 - } 517 - const body = await c.req.json<{ summary?: string }>().catch(() => ({ summary: undefined })); 518 - if (!body.summary) { 519 - return c.json({ error: "MissingSummary" }, 400); 520 - } 521 - await db 522 - .prepare("UPDATE island_cache SET summary = ? WHERE id = ?") 523 - .bind(body.summary, id) 524 - .run(); 525 - return c.json({ ok: true }); 526 - }); 527 - 528 - // Upload strata analysis for an island 529 - app.put("/xrpc/org.latha.strata.putAnalysis", async (c) => { 530 - const id = c.req.query("id"); 531 - if (!id) { 532 - return c.json({ error: "MissingRequiredParameter", message: "id is required" }, 400); 533 - } 534 - const body = await c.req.json().catch(() => ({})); 535 - if (!body || !body.prose) { 536 - return c.json({ error: "MissingRequiredField", message: "prose is required" }, 400); 537 - } 538 - const strataJson = JSON.stringify(body); 539 - await db 540 - .prepare("UPDATE island_cache SET strata = ? WHERE id = ?") 541 - .bind(strataJson, id) 542 - .run(); 543 - return c.json({ ok: true }); 544 - }); 545 - 546 470 // ── Strata record endpoint (AT URI-based) ────────────────────── 547 471 548 472 app.get("/xrpc/org.latha.strata.getRecord", async (c) => { ··· 698 622 app.get(path, async (c) => { 699 623 let islandsJson = "[]"; 700 624 try { 701 - const islands = await readIslandCache(db); 625 + const islands = await listStrataIslands(db); 702 626 islandsJson = JSON.stringify(islands); 703 627 } catch {} 704 628 ··· 731 655 return app.fetch(request, env); 732 656 }, 733 657 async scheduled(event: ScheduledEvent, env: Env, ctx: ExecutionContext) { 734 - // Run Contrail indexing first 658 + // Run Contrail indexing 735 659 await contrailWorker.scheduled( 736 660 event, 737 661 env as unknown as Record<string, unknown>, 738 662 ctx, 739 663 ); 740 - 741 - // Rebuild island cache in the background after indexing 742 - ctx.waitUntil(rebuildIslandCache(env.DB)); 743 664 }, 744 665 }; 745 666

Configure Feed

Configure Feed