diff --git a/api/stream/data.ts b/api/stream/data.ts new file mode 100644 index 000000000..35241cd11 --- /dev/null +++ b/api/stream/data.ts @@ -0,0 +1,166 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { createHash } from "node:crypto"; + +import { extractPayloadTokenFromUrlString, decodeFeedPayload } from "../../src/utils/feedPayload"; + +export type StreamRow = { + token: string; + url: string; + pulse: number; + updatedAt: number; + preview: { + title: string; + author: string; + pulse: number; + kind: string; + shortBody: string; + links: string[]; + }; +}; + +type PersistedIndex = { + sourceDigest: string; + rows: StreamRow[]; + seals: string[]; + builtAt: number; +}; + +type StreamIndex = { + sourceDigest: string; + rows: StreamRow[]; + latestSeal: string; + latestPulse: number; + sealByCount: string[]; + prefixIndexBySeal: Map; +}; + +const JOURNAL_PATH = path.resolve(process.cwd(), ".cache/stream-journal.json"); +let memIndex: PersistedIndex | null = null; + +function bodyText(payload: ReturnType): string { + if (!payload) return ""; + const body = payload.body; + const raw = + body?.kind === "text" + ? body.text + : body?.kind === "md" + ? body.md + : body?.kind === "code" + ? body.code + : body?.kind === "html" + ? body.html + : payload.caption ?? ""; + return String(raw).replace(/\s+/g, " ").slice(0, 180); +} + +function hashString(input: string): string { + return createHash("sha256").update(input).digest("hex"); +} + +async function loadPersistedIndex(): Promise { + if (memIndex) return memIndex; + try { + const raw = await fs.readFile(JOURNAL_PATH, "utf8"); + const parsed = JSON.parse(raw) as PersistedIndex; + if (!Array.isArray(parsed.rows) || !Array.isArray(parsed.seals) || typeof parsed.sourceDigest !== "string") { + return null; + } + memIndex = parsed; + return parsed; + } catch { + return null; + } +} + +async function persistIndex(index: PersistedIndex): Promise { + memIndex = index; + await fs.mkdir(path.dirname(JOURNAL_PATH), { recursive: true }); + await fs.writeFile(JOURNAL_PATH, JSON.stringify(index), "utf8"); +} + +function computePrefixSeals(rows: StreamRow[]): string[] { + const h = createHash("sha256"); + const seals: string[] = ["0"]; + for (const row of rows) { + h.update(`${row.token}:${row.pulse}:${row.updatedAt}`); + seals.push(h.copy().digest("hex").slice(0, 24)); + } + return seals; +} + +async function scanRowsFromLinks(rawLinks: string): Promise { + const json = JSON.parse(rawLinks) as Array<{ url?: string }>; + const out: StreamRow[] = []; + for (const row of json) { + const url = row.url?.trim(); + if (!url) continue; + const token = extractPayloadTokenFromUrlString(url); + if (!token) continue; + const payload = decodeFeedPayload(token); + if (!payload) continue; + out.push({ + token, + url, + pulse: Number(payload.pulse) || 0, + updatedAt: Number(payload.ts) || 0, + preview: { + title: payload.caption?.slice(0, 72) || payload.body?.kind || "memory", + author: payload.author || "@unknown", + pulse: Number(payload.pulse) || 0, + kind: payload.body?.kind || "text", + shortBody: bodyText(payload), + links: [payload.url], + }, + }); + } + out.sort((a, b) => a.updatedAt - b.updatedAt || a.pulse - b.pulse || a.token.localeCompare(b.token)); + return out; +} + +export async function getStreamIndex(): Promise { + const linksPath = path.resolve(process.cwd(), "public/links.json"); + const raw = await fs.readFile(linksPath, "utf8"); + const sourceDigest = hashString(raw); + + let persisted = await loadPersistedIndex(); + if (!persisted || persisted.sourceDigest !== sourceDigest) { + const rows = await scanRowsFromLinks(raw); + const seals = computePrefixSeals(rows); + persisted = { sourceDigest, rows, seals, builtAt: Date.now() }; + await persistIndex(persisted); + } + + const prefixIndexBySeal = new Map(); + for (let i = 0; i < persisted.seals.length; i += 1) { + prefixIndexBySeal.set(persisted.seals[i], i); + } + + return { + sourceDigest: persisted.sourceDigest, + rows: persisted.rows, + latestSeal: persisted.seals[persisted.seals.length - 1] ?? "0", + latestPulse: persisted.rows[persisted.rows.length - 1]?.pulse ?? 0, + sealByCount: persisted.seals, + prefixIndexBySeal, + }; +} + +export type CursorState = { offset: number; anchorPulse: number }; + +export function encodeCursor(state: CursorState): string { + return Buffer.from(JSON.stringify(state)).toString("base64url"); +} + +export function decodeCursor(raw: string | null, fallbackPulse: number): CursorState { + if (!raw) return { offset: 0, anchorPulse: fallbackPulse }; + try { + const parsed = JSON.parse(Buffer.from(raw, "base64url").toString("utf8")) as CursorState; + return { + offset: Math.max(0, Number(parsed.offset) || 0), + anchorPulse: Number(parsed.anchorPulse) || fallbackPulse, + }; + } catch { + return { offset: 0, anchorPulse: fallbackPulse }; + } +} diff --git a/api/stream/delta.ts b/api/stream/delta.ts new file mode 100644 index 000000000..84c0faddc --- /dev/null +++ b/api/stream/delta.ts @@ -0,0 +1,39 @@ +import { decodeCursor, encodeCursor, getStreamIndex } from "./data"; + +type Req = { url?: string }; +type Res = { statusCode: number; setHeader: (k: string, v: string) => void; end: (body: string) => void }; + +export default async function handler(req: Req, res: Res): Promise { + const requestUrl = new URL(req.url ?? "/api/stream/delta", "http://localhost"); + const limit = Math.max(1, Math.min(1000, Number(requestUrl.searchParams.get("limit") ?? "200"))); + const after = requestUrl.searchParams.get("after") ?? ""; + const index = await getStreamIndex(); + + const cursor = decodeCursor(requestUrl.searchParams.get("cursor"), index.latestPulse); + const baseCount = after ? index.prefixIndexBySeal.get(after) ?? 0 : 0; + const changedRows = index.rows + .slice(baseCount) + .filter((row) => row.pulse <= cursor.anchorPulse); + + const slice = changedRows.slice(cursor.offset, cursor.offset + limit); + const nextOffset = cursor.offset + slice.length; + const nextCursor = nextOffset < changedRows.length + ? encodeCursor({ offset: nextOffset, anchorPulse: cursor.anchorPulse }) + : null; + + const deliveredCount = baseCount + nextOffset; + const seal = index.sealByCount[Math.min(deliveredCount, index.rows.length)] ?? index.latestSeal; + + res.statusCode = 200; + res.setHeader("Content-Type", "application/json"); + res.end( + JSON.stringify({ + seal, + latestSeal: index.latestSeal, + latestPulse: index.latestPulse, + anchorPulse: cursor.anchorPulse, + nextCursor, + rows: slice.map((r) => ({ token: r.token, url: r.url, pulse: r.pulse, preview: r.preview })), + }), + ); +} diff --git a/api/stream/head.ts b/api/stream/head.ts new file mode 100644 index 000000000..f8ecc79c7 --- /dev/null +++ b/api/stream/head.ts @@ -0,0 +1,11 @@ +import { getStreamIndex } from "./data"; + +type Res = { statusCode: number; setHeader: (k: string, v: string) => void; end: (body: string) => void }; + +export default async function handler(_req: unknown, res: Res): Promise { + const index = await getStreamIndex(); + const body = { seal: index.latestSeal, latestPulse: index.latestPulse, total: index.rows.length, sourceDigest: index.sourceDigest }; + res.statusCode = 200; + res.setHeader("Content-Type", "application/json"); + res.end(JSON.stringify(body)); +} diff --git a/api/stream/snapshot.ts b/api/stream/snapshot.ts new file mode 100644 index 000000000..552a2ec2a --- /dev/null +++ b/api/stream/snapshot.ts @@ -0,0 +1,33 @@ +import { decodeCursor, encodeCursor, getStreamIndex } from "./data"; + +type Req = { url?: string }; +type Res = { statusCode: number; setHeader: (k: string, v: string) => void; end: (body: string) => void }; + +export default async function handler(req: Req, res: Res): Promise { + const requestUrl = new URL(req.url ?? "/api/stream/snapshot", "http://localhost"); + const compact = requestUrl.searchParams.get("compact") === "1"; + const limit = Math.max(1, Math.min(1000, Number(requestUrl.searchParams.get("limit") ?? "200"))); + const index = await getStreamIndex(); + + const cursor = decodeCursor(requestUrl.searchParams.get("cursor"), index.latestPulse); + const rows = index.rows.filter((row) => row.pulse <= cursor.anchorPulse); + const slice = rows.slice(cursor.offset, cursor.offset + limit); + const nextOffset = cursor.offset + slice.length; + const nextCursor = nextOffset < rows.length ? encodeCursor({ offset: nextOffset, anchorPulse: cursor.anchorPulse }) : null; + + const bodyRows = compact + ? slice.map((r) => ({ token: r.token, url: r.url, pulse: r.pulse, preview: r.preview })) + : slice; + + const body = { + seal: index.latestSeal, + latestSeal: index.latestSeal, + latestPulse: index.latestPulse, + anchorPulse: cursor.anchorPulse, + nextCursor, + rows: bodyRows, + }; + res.statusCode = 200; + res.setHeader("Content-Type", "application/json"); + res.end(JSON.stringify(body)); +} diff --git a/scripts/dev-api.mjs b/scripts/dev-api.mjs index 61bb551b8..0960d3c12 100644 --- a/scripts/dev-api.mjs +++ b/scripts/dev-api.mjs @@ -1,58 +1,222 @@ +import fs from "node:fs/promises"; import http from "node:http"; +import path from "node:path"; +import { createHash } from "node:crypto"; import { URL } from "node:url"; import { generateSigilProof } from "../api/proof/sigil.js"; const PORT = Number(process.env.API_PORT ?? 8787); +const JOURNAL_PATH = path.resolve(process.cwd(), ".cache/stream-journal.dev.json"); +let cachedIndex = null; const readJson = async (req) => { const chunks = []; - for await (const chunk of req) { - chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); - } + for await (const chunk of req) chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); if (!chunks.length) return {}; const raw = Buffer.concat(chunks).toString("utf8"); if (!raw) return {}; return JSON.parse(raw); }; +const payloadTokenFromUrl = (rawUrl) => { + const value = String(rawUrl || "").trim(); + if (!value) return null; + + let parsed; + try { + parsed = new URL(value); + } catch { + parsed = new URL(value, "http://localhost"); + } + + const pathName = parsed.pathname || ""; + const fromPath = + pathName.match(/\/p\/([^/?#]+)/u)?.[1] + ?? pathName.match(/\/(?:stream|feed)\/p\/([^/?#]+)/u)?.[1] + ?? pathName.match(/\/p(?:~|%7[Ee])\/?([^/?#]+)/u)?.[1]; + if (fromPath) return decodeURIComponent(fromPath).trim() || null; + + const hashStr = parsed.hash && parsed.hash.startsWith("#") ? parsed.hash.slice(1) : ""; + const hashParams = new URLSearchParams(hashStr); + for (const key of ["t", "p", "token", "capsule"]) { + const candidate = hashParams.get(key) ?? parsed.searchParams.get(key); + if (candidate?.trim()) return candidate.trim(); + } + return null; +}; + +const encodeCursor = (state) => Buffer.from(JSON.stringify(state)).toString("base64url"); +const decodeCursor = (raw, fallbackPulse) => { + if (!raw) return { offset: 0, anchorPulse: fallbackPulse }; + try { + const parsed = JSON.parse(Buffer.from(raw, "base64url").toString("utf8")); + return { + offset: Math.max(0, Number(parsed.offset) || 0), + anchorPulse: Number(parsed.anchorPulse) || fallbackPulse, + }; + } catch { + return { offset: 0, anchorPulse: fallbackPulse }; + } +}; + +const hashString = (input) => createHash("sha256").update(input).digest("hex"); + +const computePrefixSeals = (rows) => { + const h = createHash("sha256"); + const seals = ["0"]; + for (const row of rows) { + h.update(`${row.token}:${row.pulse}:${row.updatedAt}`); + seals.push(h.copy().digest("hex").slice(0, 24)); + } + return seals; +}; + +const getStreamIndex = async () => { + const linksPath = path.resolve(process.cwd(), "public/links.json"); + let raw; + try { + raw = await fs.readFile(linksPath, "utf8"); + } catch { + raw = "[]"; + } + const sourceDigest = hashString(raw); + + if (!cachedIndex) { + try { + cachedIndex = JSON.parse(await fs.readFile(JOURNAL_PATH, "utf8")); + } catch { + cachedIndex = null; + } + } + + if (!cachedIndex || cachedIndex.sourceDigest !== sourceDigest) { + const json = JSON.parse(raw); + const rows = []; + for (const row of json) { + const itemUrl = typeof row?.url === "string" ? row.url.trim() : ""; + if (!itemUrl) continue; + const token = payloadTokenFromUrl(itemUrl); + if (!token) continue; + rows.push({ + token, + url: itemUrl, + pulse: Number(row?.pulse) || 0, + updatedAt: Number(row?.updatedAt) || 0, + preview: { + title: String(row?.title || "memory").slice(0, 72), + author: String(row?.author || "@unknown"), + pulse: Number(row?.pulse) || 0, + kind: String(row?.kind || "text"), + shortBody: String(row?.shortBody || "").replace(/\s+/g, " ").slice(0, 180), + links: [itemUrl], + }, + }); + } + rows.sort((a, b) => a.updatedAt - b.updatedAt || a.pulse - b.pulse || a.token.localeCompare(b.token)); + cachedIndex = { + sourceDigest, + rows, + seals: computePrefixSeals(rows), + }; + await fs.mkdir(path.dirname(JOURNAL_PATH), { recursive: true }); + await fs.writeFile(JOURNAL_PATH, JSON.stringify(cachedIndex), "utf8"); + } + + const prefixIndexBySeal = new Map(); + for (let i = 0; i < cachedIndex.seals.length; i += 1) prefixIndexBySeal.set(cachedIndex.seals[i], i); + + return { + sourceDigest: cachedIndex.sourceDigest, + rows: cachedIndex.rows, + latestSeal: cachedIndex.seals[cachedIndex.seals.length - 1] ?? "0", + latestPulse: cachedIndex.rows[cachedIndex.rows.length - 1]?.pulse ?? 0, + sealByCount: cachedIndex.seals, + prefixIndexBySeal, + }; +}; + +const sendJson = (res, statusCode, body) => { + res.statusCode = statusCode; + res.setHeader("Content-Type", "application/json"); + res.end(JSON.stringify(body)); +}; + const server = http.createServer(async (req, res) => { const url = new URL(req.url ?? "/", `http://${req.headers.host}`); res.setHeader("Access-Control-Allow-Origin", "*"); - res.setHeader("Access-Control-Allow-Methods", "POST, OPTIONS"); + res.setHeader("Access-Control-Allow-Methods", "GET, POST, OPTIONS"); res.setHeader("Access-Control-Allow-Headers", "Content-Type"); - if (req.method === "OPTIONS") { - res.statusCode = 204; - res.end(); + if (req.method === "OPTIONS") return void res.writeHead(204).end(); + + if (url.pathname === "/api/proof/sigil") { + if (req.method !== "POST") return void res.writeHead(405).end("Method not allowed"); + try { + const body = await readJson(req); + const result = await generateSigilProof(body); + sendJson(res, 200, result); + } catch (err) { + const message = err instanceof Error ? err.message : "Proof generation failed"; + sendJson(res, 400, { error: message }); + } return; } - if (url.pathname !== "/api/proof/sigil") { - res.statusCode = 404; - res.end("Not found"); + if (req.method === "GET" && url.pathname === "/api/stream/head") { + const index = await getStreamIndex(); + sendJson(res, 200, { seal: index.latestSeal, latestPulse: index.latestPulse, total: index.rows.length, sourceDigest: index.sourceDigest }); return; } - if (req.method !== "POST") { - res.statusCode = 405; - res.end("Method not allowed"); + if (req.method === "GET" && url.pathname === "/api/stream/snapshot") { + const compact = url.searchParams.get("compact") === "1"; + const limit = Math.max(1, Math.min(1000, Number(url.searchParams.get("limit") ?? "200"))); + const index = await getStreamIndex(); + const cursor = decodeCursor(url.searchParams.get("cursor"), index.latestPulse); + const rows = index.rows.filter((row) => row.pulse <= cursor.anchorPulse); + const slice = rows.slice(cursor.offset, cursor.offset + limit); + const nextOffset = cursor.offset + slice.length; + const nextCursor = nextOffset < rows.length ? encodeCursor({ offset: nextOffset, anchorPulse: cursor.anchorPulse }) : null; + + sendJson(res, 200, { + seal: index.latestSeal, + latestSeal: index.latestSeal, + latestPulse: index.latestPulse, + anchorPulse: cursor.anchorPulse, + nextCursor, + rows: compact ? slice.map((r) => ({ token: r.token, url: r.url, pulse: r.pulse, preview: r.preview })) : slice, + }); return; } - try { - const body = await readJson(req); - const result = await generateSigilProof(body); - res.statusCode = 200; - res.setHeader("Content-Type", "application/json"); - res.end(JSON.stringify(result)); - } catch (err) { - const message = err instanceof Error ? err.message : "Proof generation failed"; - res.statusCode = 400; - res.setHeader("Content-Type", "application/json"); - res.end(JSON.stringify({ error: message })); + if (req.method === "GET" && url.pathname === "/api/stream/delta") { + const limit = Math.max(1, Math.min(1000, Number(url.searchParams.get("limit") ?? "200"))); + const after = url.searchParams.get("after") ?? ""; + const index = await getStreamIndex(); + const cursor = decodeCursor(url.searchParams.get("cursor"), index.latestPulse); + + const baseCount = after ? index.prefixIndexBySeal.get(after) ?? 0 : 0; + const changedRows = index.rows.slice(baseCount).filter((row) => row.pulse <= cursor.anchorPulse); + const slice = changedRows.slice(cursor.offset, cursor.offset + limit); + const nextOffset = cursor.offset + slice.length; + const nextCursor = nextOffset < changedRows.length ? encodeCursor({ offset: nextOffset, anchorPulse: cursor.anchorPulse }) : null; + const seal = index.sealByCount[Math.min(baseCount + nextOffset, index.rows.length)] ?? index.latestSeal; + + sendJson(res, 200, { + seal, + latestSeal: index.latestSeal, + latestPulse: index.latestPulse, + anchorPulse: cursor.anchorPulse, + nextCursor, + rows: slice.map((r) => ({ token: r.token, url: r.url, pulse: r.pulse, preview: r.preview })), + }); + return; } + + res.statusCode = 404; + res.end("Not found"); }); server.listen(PORT, () => { diff --git a/src/lib/stream/streamStore.ts b/src/lib/stream/streamStore.ts new file mode 100644 index 000000000..f598d9c4c --- /dev/null +++ b/src/lib/stream/streamStore.ts @@ -0,0 +1,316 @@ +import { decodeFeedPayload, extractPayloadTokenFromUrlString, type FeedPostPayload } from "../../utils/feedPayload"; + +export type StreamPreview = { + token: string; + url: string; + title: string; + author: string; + pulse: number; + kind: string; + shortBody: string; + links: string[]; + updatedAt: number; +}; + +type StreamRecord = StreamPreview & { + parentToken?: string; + payload?: FeedPostPayload; +}; + +export type StreamDeltaRow = { + token: string; + url: string; + pulse?: number; + payload?: FeedPostPayload; + preview?: Partial; + deleted?: boolean; +}; + +export type StreamDelta = { seal?: string; latestSeal?: string; sourceDigest?: string; rows: StreamDeltaRow[] }; +export type StreamFeedPage = { rows: StreamPreview[]; nextCursor: string | null }; + +const DB_NAME = "kai-stream-v1"; +const DB_VERSION = 1; +const STORE_ITEMS = "items"; +const STORE_META = "meta"; + +function shortBody(payload: FeedPostPayload): string { + const body = payload.body; + const raw = + body?.kind === "text" + ? body.text + : body?.kind === "md" + ? body.md + : body?.kind === "code" + ? body.code + : body?.kind === "html" + ? body.html + : payload.caption ?? ""; + return String(raw).replace(/\s+/g, " ").slice(0, 180); +} + +function parseLinks(payload: FeedPostPayload): string[] { + const links = new Set(); + for (const item of payload.attachments?.items ?? []) { + if (item.kind === "url" && item.url) links.add(item.url); + if (item.kind === "file-ref" && item.url) links.add(item.url); + } + if (payload.url) links.add(payload.url); + return [...links].slice(0, 6); +} + +function previewFromPayload(token: string, url: string, payload: FeedPostPayload): StreamPreview { + return { + token, + url, + title: payload.caption?.slice(0, 72) || payload.body?.kind || "memory", + author: payload.author || "@unknown", + pulse: Number(payload.pulse) || 0, + kind: payload.body?.kind || (payload.seal ? "sealed" : "text"), + shortBody: shortBody(payload), + links: parseLinks(payload), + updatedAt: Date.now(), + }; +} + +function decodeFromUrl(url: string): { token: string; payload: FeedPostPayload; preview: StreamPreview } | null { + const token = extractPayloadTokenFromUrlString(url); + if (!token) return null; + const payload = decodeFeedPayload(token); + if (!payload) return null; + return { token, payload, preview: previewFromPayload(token, url, payload) }; +} + +function openDb(): Promise { + return new Promise((resolve, reject) => { + const req = indexedDB.open(DB_NAME, DB_VERSION); + req.onupgradeneeded = () => { + const db = req.result; + if (!db.objectStoreNames.contains(STORE_ITEMS)) { + const s = db.createObjectStore(STORE_ITEMS, { keyPath: "token" }); + s.createIndex("pulse", "pulse"); + s.createIndex("updatedAt", "updatedAt"); + } + if (!db.objectStoreNames.contains(STORE_META)) { + db.createObjectStore(STORE_META, { keyPath: "k" }); + } + }; + req.onsuccess = () => resolve(req.result); + req.onerror = () => reject(req.error); + }); +} + +async function idbGetAll(): Promise { + const db = await openDb(); + return new Promise((resolve, reject) => { + const tx = db.transaction(STORE_ITEMS, "readonly"); + const req = tx.objectStore(STORE_ITEMS).getAll(); + req.onsuccess = () => resolve((req.result as StreamRecord[]) || []); + req.onerror = () => reject(req.error); + }); +} + +async function idbPutMany(rows: StreamRecord[]): Promise { + const db = await openDb(); + await new Promise((resolve, reject) => { + const tx = db.transaction([STORE_ITEMS], "readwrite"); + const store = tx.objectStore(STORE_ITEMS); + for (const row of rows) store.put(row); + tx.oncomplete = () => resolve(); + tx.onerror = () => reject(tx.error); + }); +} + +async function idbDeleteMany(tokens: string[]): Promise { + if (!tokens.length) return; + const db = await openDb(); + await new Promise((resolve, reject) => { + const tx = db.transaction([STORE_ITEMS], "readwrite"); + const store = tx.objectStore(STORE_ITEMS); + for (const token of tokens) store.delete(token); + tx.oncomplete = () => resolve(); + tx.onerror = () => reject(tx.error); + }); +} + +async function idbGetMeta(k: string): Promise { + const db = await openDb(); + return new Promise((resolve, reject) => { + const tx = db.transaction(STORE_META, "readonly"); + const req = tx.objectStore(STORE_META).get(k); + req.onsuccess = () => resolve((req.result?.v as T) ?? null); + req.onerror = () => reject(req.error); + }); +} + +async function idbSetMeta(k: string, v: unknown): Promise { + const db = await openDb(); + await new Promise((resolve, reject) => { + const tx = db.transaction(STORE_META, "readwrite"); + tx.objectStore(STORE_META).put({ k, v }); + tx.oncomplete = () => resolve(); + tx.onerror = () => reject(tx.error); + }); +} + +const mem = new Map(); +let workerFailed = false; + +function notNull(value: T | null): value is T { + return value !== null; +} + +export function setStreamWorkerFailedForTests(value: boolean): void { + workerFailed = value; +} + +export function applyDeltaRowsToRecords(existing: Map, delta: StreamDelta): Map { + const next = new Map(existing); + for (const row of delta.rows) { + if (row.deleted) { + next.delete(row.token); + continue; + } + const base = row.payload + ? previewFromPayload(row.token, row.url, row.payload) + : next.get(row.token) ?? { + token: row.token, + url: row.url, + title: "memory", + author: "@unknown", + pulse: row.pulse ?? 0, + kind: "text", + shortBody: "", + links: [], + updatedAt: Date.now(), + }; + const merged = { ...base, ...row.preview, url: row.url, token: row.token, updatedAt: Date.now() }; + next.set(row.token, { + ...merged, + payload: row.payload, + parentToken: row.payload?.parentUrl ? extractPayloadTokenFromUrlString(row.payload.parentUrl) ?? undefined : undefined, + }); + } + return next; +} + +async function decodeBatch(urls: string[]): Promise { + if (typeof window === "undefined") return []; + if (workerFailed || typeof Worker === "undefined") { + return urls.map(decodeFromUrl).filter(notNull).map((r) => ({ ...r.preview, payload: r.payload, parentToken: r.payload.parentUrl ? extractPayloadTokenFromUrlString(r.payload.parentUrl) ?? undefined : undefined })); + } + + try { + const worker = new Worker(new URL("../../workers/streamWorker.ts", import.meta.url), { type: "module" }); + const rows = await new Promise((resolve, reject) => { + const timer = window.setTimeout(() => reject(new Error("stream worker timeout")), 2500); + worker.onmessage = (evt: MessageEvent) => { + window.clearTimeout(timer); + resolve(evt.data); + }; + worker.onerror = (err) => { + window.clearTimeout(timer); + reject(err); + }; + worker.postMessage(urls); + }); + worker.terminate(); + return rows; + } catch { + workerFailed = true; + return urls.map(decodeFromUrl).filter(notNull).map((r) => ({ ...r.preview, payload: r.payload, parentToken: r.payload.parentUrl ? extractPayloadTokenFromUrlString(r.payload.parentUrl) ?? undefined : undefined })); + } +} + +export const streamStore = { + async ingestUrls(urls: string[]): Promise { + const rows = await decodeBatch(urls); + for (const row of rows) mem.set(row.token, row); + await idbPutMany(rows); + }, + + async getFeedPage(cursor: string | null, limit: number): Promise { + const all = (await idbGetAll()).sort((a, b) => (b.pulse - a.pulse) || (b.updatedAt - a.updatedAt)); + const offset = Number(cursor || 0); + const rows = all.slice(offset, offset + limit).map(({ payload: _p, parentToken: _t, ...preview }) => preview); + return { rows, nextCursor: offset + limit < all.length ? String(offset + limit) : null }; + }, + + async getThread(token: string, depth: number): Promise { + const all = await idbGetAll(); + const byToken = new Map(all.map((r) => [r.token, r])); + const out: StreamPreview[] = []; + const seen = new Set(); + const walk = (t: string, d: number) => { + if (d < 0 || seen.has(t)) return; + seen.add(t); + const row = byToken.get(t); + if (!row) return; + const { payload: _p, parentToken: _pt, ...preview } = row; + out.push(preview); + if (row.parentToken) walk(row.parentToken, d - 1); + for (const child of all) { + if (child.parentToken === t) walk(child.token, d - 1); + } + }; + walk(token, depth); + return out.sort((a, b) => a.pulse - b.pulse); + }, + + async applyDelta(delta: StreamDelta): Promise { + const persisted = await idbGetAll(); + const baseline = new Map(persisted.map((row) => [row.token, row])); + for (const [token, row] of mem.entries()) { + baseline.set(token, row); + } + + const next = applyDeltaRowsToRecords(baseline, delta); + const upserts: StreamRecord[] = []; + const deletes: string[] = []; + + for (const [token, record] of next.entries()) { + upserts.push(record); + mem.set(token, record); + } + + for (const token of baseline.keys()) { + if (!next.has(token)) { + mem.delete(token); + deletes.push(token); + } + } + + await idbPutMany(upserts); + await idbDeleteMany(deletes); + if (delta.seal) await idbSetMeta("seal", delta.seal); + if (delta.sourceDigest) await idbSetMeta("sourceDigest", delta.sourceDigest); + }, + + async getPreview(urlOrToken: string): Promise { + const token = extractPayloadTokenFromUrlString(urlOrToken) ?? urlOrToken; + const inMem = mem.get(token); + if (inMem) { + const { payload: _p, parentToken: _pt, ...preview } = inMem; + return preview; + } + const rows = await idbGetAll(); + const row = rows.find((r) => r.token === token); + if (!row) return null; + const { payload: _p, parentToken: _pt, ...preview } = row; + return preview; + }, + + async prefetchAround(token: string): Promise { + const thread = await this.getThread(token, 2); + const urls = thread.map((t) => t.url); + await this.ingestUrls(urls); + }, + + async getSeal(): Promise { + return idbGetMeta("seal"); + }, + + async getSourceDigest(): Promise { + return idbGetMeta("sourceDigest"); + }, +}; diff --git a/src/lib/stream/streamSync.ts b/src/lib/stream/streamSync.ts new file mode 100644 index 000000000..27e7bf44e --- /dev/null +++ b/src/lib/stream/streamSync.ts @@ -0,0 +1,65 @@ +import { streamStore, type StreamDelta } from "./streamStore"; + +export type StreamHead = { seal: string; latestPulse: number; total: number; sourceDigest?: string }; + +export function buildSyncUrl(localSeal: string | null, limit: number, cursor: string | null = null): string { + const encodedLimit = encodeURIComponent(String(limit)); + const cursorPart = cursor ? `&cursor=${encodeURIComponent(cursor)}` : ""; + return localSeal + ? `/api/stream/delta?after=${encodeURIComponent(localSeal)}&limit=${encodedLimit}${cursorPart}` + : `/api/stream/snapshot?compact=1&limit=${encodedLimit}${cursorPart}`; +} + +async function fetchJson(url: string): Promise { + try { + const res = await fetch(url); + if (!res.ok) return null; + return (await res.json()) as T; + } catch { + return null; + } +} + +export async function syncStreamDelta(limit = 200): Promise { + const head = await fetchJson("/api/stream/head"); + if (!head) return 0; + + let [localSeal, localSourceDigest] = await Promise.all([streamStore.getSeal(), streamStore.getSourceDigest()]); + + const sourceDigestChanged = + Boolean(head.sourceDigest) && Boolean(localSourceDigest) && head.sourceDigest !== localSourceDigest; + + if (!sourceDigestChanged && localSeal && localSeal === head.seal) return 0; + + if (sourceDigestChanged) localSeal = null; + + let totalChanged = 0; + let cursor: string | null = null; + + for (let page = 0; page < 40; page += 1) { + const url = buildSyncUrl(localSeal, limit, cursor); + const delta = await fetchJson(url); + if (!delta) return totalChanged; + + if (!delta.rows.length) { + if (delta.latestSeal && delta.latestSeal !== localSeal) { + await streamStore.applyDelta({ + rows: [], + seal: delta.latestSeal, + latestSeal: delta.latestSeal, + sourceDigest: head.sourceDigest, + }); + } + break; + } + + await streamStore.applyDelta({ ...delta, sourceDigest: head.sourceDigest }); + totalChanged += delta.rows.length; + localSeal = delta.seal ?? localSeal; + cursor = delta.nextCursor ?? null; + + if (!cursor && delta.latestSeal && delta.seal === delta.latestSeal) break; + } + + return totalChanged; +} diff --git a/src/pages/sigilstream/SigilStreamRoot.tsx b/src/pages/sigilstream/SigilStreamRoot.tsx index 3db71b32e..b9fceca5c 100644 --- a/src/pages/sigilstream/SigilStreamRoot.tsx +++ b/src/pages/sigilstream/SigilStreamRoot.tsx @@ -125,6 +125,8 @@ import { /* PhiStream auto-add */ import { autoAddVisitedPayloadToPhiStream } from "./core/phiStreamAutoAdd"; +import { syncStreamDelta } from "../../lib/stream/streamSync"; +import { streamStore } from "../../lib/stream/streamStore"; /** Simple source shape */ type Source = { url: string }; @@ -1569,6 +1571,28 @@ function SigilStreamInner(): React.JSX.Element { }; }, [ms2IngestMany, rehydrateUsernameClaimsFromHistory]); + useEffect(() => { + if (typeof window === "undefined") return; + let dead = false; + void (async () => { + const changed = await syncStreamDelta(300); + if (!changed || dead) return; + const page = await streamStore.getFeedPage(null, 120); + if (dead) return; + const urls = page.rows.map((row) => row.url); + setSources((prev) => { + const seen = new Set(prev.map((row) => row.url)); + const fresh = urls.filter((url) => !seen.has(url)); + if (!fresh.length) return prev; + prependUniqueToStorage(fresh); + return [...fresh.map((url) => ({ url })), ...prev]; + }); + })(); + return () => { + dead = true; + }; + }, []); + /** * Ingest add= links from CURRENT location every time it changes. */ @@ -2369,4 +2393,4 @@ function SigilStreamInner(): React.JSX.Element { ); } -export default SigilStreamRoot; \ No newline at end of file +export default SigilStreamRoot; diff --git a/src/pages/sigilstream/list/StreamList.tsx b/src/pages/sigilstream/list/StreamList.tsx index b035ba05c..6e0eaf31d 100644 --- a/src/pages/sigilstream/list/StreamList.tsx +++ b/src/pages/sigilstream/list/StreamList.tsx @@ -1,35 +1,107 @@ -// src/pages/sigilstream/list/StreamList.tsx "use client"; -import type React from "react"; +import React, { useEffect, useMemo, useRef, useState } from "react"; import FeedCard from "../../../components/FeedCard"; +import { streamStore, type StreamPreview } from "../../../lib/stream/streamStore"; -type Props = { - /** Ordered list of canonical item URLs to render */ - urls: string[]; -}; +import { BATCH, ROW_H, computeVirtualWindow } from "./virtualWindow"; + +type Props = { urls: string[] }; + +function useVirtualWindow(count: number): { start: number; end: number; topPad: number; bottomPad: number } { + const [scrollTop, setScrollTop] = useState(0); + const [vh, setVh] = useState(900); + + useEffect(() => { + const onScroll = () => setScrollTop(window.scrollY || 0); + const onResize = () => setVh(window.innerHeight || 900); + onResize(); + window.addEventListener("scroll", onScroll, { passive: true }); + window.addEventListener("resize", onResize); + return () => { + window.removeEventListener("scroll", onScroll); + window.removeEventListener("resize", onResize); + }; + }, []); + + const { start, end } = computeVirtualWindow(scrollTop, vh, count); + return { start, end, topPad: start * ROW_H, bottomPad: Math.max(0, (count - end) * ROW_H) }; +} + +function PreviewCard({ preview, onOpen }: { preview: StreamPreview; onOpen: () => void }): React.JSX.Element { + return ( +
+
{preview.title}
+

{preview.shortBody}

+
+ +
+
+ ); +} -/** - * StreamList — maps an ordered array of URLs to . - * Mobile-first, minimal DOM, safe to embed anywhere. - */ export function StreamList({ urls }: Props): React.JSX.Element { - if (!urls || urls.length === 0) { - return ( -
-
- No items yet. Paste a link above or open a /stream/p/<payload> link and reply to start a thread. -
-
- ); - } + const [hydrated, setHydrated] = useState(false); + const [visibleCount, setVisibleCount] = useState(BATCH); + const [expanded, setExpanded] = useState>({}); + const [previewMap, setPreviewMap] = useState>({}); + const ingestedRef = useRef>(new Set()); + + useEffect(() => { + let dead = false; + void (async () => { + const fresh = urls.filter((u) => !ingestedRef.current.has(u)); + if (fresh.length === 0) { + if (!dead) setHydrated(true); + return; + } + + await streamStore.ingestUrls(fresh); + const map: Record = {}; + for (const u of fresh) { + const p = await streamStore.getPreview(u); + if (p) map[u] = p; + } + + if (!dead) { + setPreviewMap((prev) => ({ ...prev, ...map })); + for (const u of fresh) ingestedRef.current.add(u); + setHydrated(true); + } + })(); + return () => { + dead = true; + }; + }, [urls]); + + useEffect(() => { + const t = window.setInterval(() => setVisibleCount((v) => Math.min(urls.length, v + BATCH)), 120); + return () => window.clearInterval(t); + }, [urls.length]); + + const progressive = useMemo(() => urls.slice(0, visibleCount), [urls, visibleCount]); + const vw = useVirtualWindow(progressive.length); + const windowed = progressive.slice(vw.start, vw.end); + + if (!urls.length) return <>; return ( -
- {urls.map((u) => ( - - ))} -
+
+
+ {windowed.map((u) => { + const isOpen = Boolean(expanded[u]); + const preview = previewMap[u]; + return isOpen || !hydrated || !preview ? ( + + ) : ( + { + setExpanded((prev) => ({ ...prev, [u]: true })); + void streamStore.prefetchAround(preview.token); + }} /> + ); + })} +
+
); } diff --git a/src/pages/sigilstream/list/virtualWindow.ts b/src/pages/sigilstream/list/virtualWindow.ts new file mode 100644 index 000000000..782b4838a --- /dev/null +++ b/src/pages/sigilstream/list/virtualWindow.ts @@ -0,0 +1,9 @@ +export const ROW_H = 240; +export const OVERSCAN = 4; +export const BATCH = 18; + +export function computeVirtualWindow(scrollTop: number, vh: number, count: number): { start: number; end: number } { + const start = Math.max(0, Math.floor(scrollTop / ROW_H) - OVERSCAN); + const end = Math.min(count, Math.ceil((scrollTop + vh) / ROW_H) + OVERSCAN); + return { start, end }; +} diff --git a/src/workers/streamWorker.ts b/src/workers/streamWorker.ts new file mode 100644 index 000000000..a0ee5e73a --- /dev/null +++ b/src/workers/streamWorker.ts @@ -0,0 +1,54 @@ +import { decodeFeedPayload, extractPayloadTokenFromUrlString } from "../utils/feedPayload"; + +type StreamRecord = { + token: string; + url: string; + title: string; + author: string; + pulse: number; + kind: string; + shortBody: string; + links: string[]; + updatedAt: number; + parentToken?: string; +}; + +function project(url: string): StreamRecord | null { + const token = extractPayloadTokenFromUrlString(url); + if (!token) return null; + const payload = decodeFeedPayload(token); + if (!payload) return null; + const bodyText = + payload.body?.kind === "text" + ? payload.body.text + : payload.body?.kind === "md" + ? payload.body.md + : payload.body?.kind === "code" + ? payload.body.code + : payload.body?.kind === "html" + ? payload.body.html + : payload.caption ?? ""; + const links = payload.attachments?.items.flatMap((item) => { + if (item.kind === "url") return item.url ? [item.url] : []; + if (item.kind === "file-ref") return item.url ? [item.url] : []; + return []; + }) ?? []; + if (payload.url) links.unshift(payload.url); + return { + token, + url, + title: payload.caption?.slice(0, 72) || payload.body?.kind || "memory", + author: payload.author || "@unknown", + pulse: Number(payload.pulse) || 0, + kind: payload.body?.kind || (payload.seal ? "sealed" : "text"), + shortBody: bodyText.replace(/\s+/g, " ").slice(0, 180), + links: links.slice(0, 6), + updatedAt: Date.now(), + parentToken: payload.parentUrl ? extractPayloadTokenFromUrlString(payload.parentUrl) ?? undefined : undefined, + }; +} + +self.onmessage = (event: MessageEvent) => { + const rows = event.data.map(project).filter((v): v is StreamRecord => Boolean(v)); + self.postMessage(rows); +}; diff --git a/tests/dev_api_stream_loader.test.mjs b/tests/dev_api_stream_loader.test.mjs new file mode 100644 index 000000000..d9a2c0aed --- /dev/null +++ b/tests/dev_api_stream_loader.test.mjs @@ -0,0 +1,67 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; +import { join, resolve } from "node:path"; +import { tmpdir } from "node:os"; +import { spawn } from "node:child_process"; +import { test } from "node:test"; + +const repoRoot = resolve(process.cwd()); +const devApiScript = resolve(repoRoot, "scripts/dev-api.mjs"); + +function waitForServer(proc, timeoutMs = 8000) { + return new Promise((resolveReady, rejectReady) => { + const timeout = setTimeout(() => { + rejectReady(new Error("dev api did not start in time")); + }, timeoutMs); + + const onData = (chunk) => { + if (String(chunk).includes("Sigil proof API listening")) { + clearTimeout(timeout); + proc.stdout.off("data", onData); + resolveReady(); + } + }; + + proc.stdout.on("data", onData); + proc.once("exit", (code) => { + clearTimeout(timeout); + rejectReady(new Error(`dev api exited before ready (code ${code ?? "null"})`)); + }); + }); +} + +test("dev api stream loader accepts /p~ URLs and keeps seal stable without updatedAt", async () => { + const tempRoot = mkdtempSync(join(tmpdir(), "phi-dev-api-")); + const publicDir = join(tempRoot, "public"); + mkdirSync(publicDir, { recursive: true }); + + const token = "A".repeat(20); + writeFileSync( + join(publicDir, "links.json"), + JSON.stringify([{ url: `https://example.com/p~${token}` }]), + "utf8", + ); + + const port = 18787; + const proc = spawn(process.execPath, [devApiScript], { + cwd: tempRoot, + env: { ...process.env, API_PORT: String(port) }, + stdio: ["ignore", "pipe", "pipe"], + }); + + try { + await waitForServer(proc); + + const first = await fetch(`http://127.0.0.1:${port}/api/stream/head`).then((r) => r.json()); + const second = await fetch(`http://127.0.0.1:${port}/api/stream/head`).then((r) => r.json()); + const delta = await fetch(`http://127.0.0.1:${port}/api/stream/delta?after=${encodeURIComponent(first.seal)}`).then((r) => r.json()); + + assert.equal(first.total, 1); + assert.equal(second.total, 1); + assert.equal(first.seal, second.seal); + assert.equal(delta.rows.length, 0); + } finally { + proc.kill("SIGTERM"); + rmSync(tempRoot, { recursive: true, force: true }); + } +}); diff --git a/tests/stream_delta_roundtrip.integration.test.mjs b/tests/stream_delta_roundtrip.integration.test.mjs new file mode 100644 index 000000000..96cf88b60 --- /dev/null +++ b/tests/stream_delta_roundtrip.integration.test.mjs @@ -0,0 +1,108 @@ +import assert from "node:assert/strict"; +import { mkdtempSync, mkdirSync, rmSync, writeFileSync } from "node:fs"; +import { join, resolve } from "node:path"; +import { tmpdir } from "node:os"; +import { spawn } from "node:child_process"; +import { test } from "node:test"; + +const repoRoot = resolve(process.cwd()); +const devApiScript = resolve(repoRoot, "scripts/dev-api.mjs"); + +function waitForServer(proc, timeoutMs = 8000) { + return new Promise((resolveReady, rejectReady) => { + const timeout = setTimeout(() => rejectReady(new Error("dev api did not start in time")), timeoutMs); + const onData = (chunk) => { + if (String(chunk).includes("Sigil proof API listening")) { + clearTimeout(timeout); + proc.stdout.off("data", onData); + resolveReady(); + } + }; + proc.stdout.on("data", onData); + proc.once("exit", (code) => { + clearTimeout(timeout); + rejectReady(new Error(`dev api exited before ready (code ${code ?? "null"})`)); + }); + }); +} + +const makeRow = (id, pulse) => ({ + url: `https://example.com/p~${String(id).repeat(20).slice(0, 20)}`, + pulse, + updatedAt: pulse, + title: `Title-${id}`, + author: "@sync", + kind: "text", + shortBody: `Body ${id}`, +}); + +async function fetchJson(url) { + const res = await fetch(url); + assert.equal(res.ok, true, `request failed: ${url}`); + return res.json(); +} + +test("delta roundtrip syncs paginated snapshot->delta->apply against live API", async () => { + const tempRoot = mkdtempSync(join(tmpdir(), "phi-stream-roundtrip-")); + const publicDir = join(tempRoot, "public"); + mkdirSync(publicDir, { recursive: true }); + + const initial = [makeRow("A", 1), makeRow("B", 2), makeRow("C", 3), makeRow("D", 4), makeRow("E", 5)]; + writeFileSync(join(publicDir, "links.json"), JSON.stringify(initial), "utf8"); + + const port = 18788; + const proc = spawn(process.execPath, [devApiScript], { + cwd: tempRoot, + env: { ...process.env, API_PORT: String(port) }, + stdio: ["ignore", "pipe", "pipe"], + }); + + try { + await waitForServer(proc); + + const base = `http://127.0.0.1:${port}`; + const snapshotRecords = new Map(); + let cursor = null; + + for (let i = 0; i < 10; i += 1) { + const page = await fetchJson(`${base}/api/stream/snapshot?compact=1&limit=2${cursor ? `&cursor=${encodeURIComponent(cursor)}` : ""}`); + for (const row of page.rows) snapshotRecords.set(row.token, row.pulse); + cursor = page.nextCursor; + if (!cursor) break; + } + + const startHead = await fetchJson(`${base}/api/stream/head`); + assert.equal(snapshotRecords.size, 5); + assert.equal(startHead.total, 5); + + const updated = [...initial, makeRow("F", 6), makeRow("G", 7)]; + writeFileSync(join(publicDir, "links.json"), JSON.stringify(updated), "utf8"); + + const deltaRecords = new Map(snapshotRecords); + let deltaCursor = null; + let currentSeal = startHead.seal; + + for (let i = 0; i < 10; i += 1) { + const page = await fetchJson( + `${base}/api/stream/delta?after=${encodeURIComponent(startHead.seal)}&limit=1${deltaCursor ? `&cursor=${encodeURIComponent(deltaCursor)}` : ""}`, + ); + for (const row of page.rows) deltaRecords.set(row.token, row.pulse); + currentSeal = page.seal; + deltaCursor = page.nextCursor; + if (!deltaCursor) { + assert.equal(page.latestSeal, currentSeal); + break; + } + } + + const finalHead = await fetchJson(`${base}/api/stream/head`); + assert.equal(finalHead.total, 7); + assert.equal(finalHead.seal, currentSeal); + assert.equal(deltaRecords.size, 7); + assert.equal(deltaRecords.get("FFFFFFFFFFFFFFFFFFFF"), 6); + assert.equal(deltaRecords.get("GGGGGGGGGGGGGGGGGGGG"), 7); + } finally { + proc.kill("SIGTERM"); + rmSync(tempRoot, { recursive: true, force: true }); + } +}); diff --git a/tests/stream_upgrade.test.mjs b/tests/stream_upgrade.test.mjs new file mode 100644 index 000000000..cf3f05421 --- /dev/null +++ b/tests/stream_upgrade.test.mjs @@ -0,0 +1,220 @@ +import assert from "node:assert/strict"; +import { existsSync, lstatSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { basename, dirname, join, resolve } from "node:path"; +import { fileURLToPath, pathToFileURL } from "node:url"; +import { test } from "node:test"; +import ts from "typescript"; + +const tempRoot = mkdtempSync(join(process.cwd(), ".tmp-stream-upgrade-")); +const moduleCache = new Map(); + +const IMPORT_FROM_RE = /from\s+["']([^"']+)["']/g; +const IMPORT_CALL_RE = /import\(\s*["']([^"']+)["']\s*\)/g; + +function resolveImport(spec, baseFile) { + if (!spec.startsWith(".")) return null; + const baseDir = dirname(baseFile); + const candidates = [spec, `${spec}.ts`, `${spec}.tsx`, `${spec}.js`, `${spec}.jsx`]; + for (const candidate of candidates) { + const full = resolve(baseDir, candidate); + if (existsSync(full) && lstatSync(full).isFile()) return full; + } + return null; +} + +function gatherImports(source) { + const specs = new Set(); + for (const match of source.matchAll(IMPORT_FROM_RE)) specs.add(match[1]); + for (const match of source.matchAll(IMPORT_CALL_RE)) specs.add(match[1]); + return [...specs]; +} + +function rewriteImports(code, replacements) { + let out = code; + for (const [spec, replacement] of replacements) { + out = out.replaceAll(`"${spec}"`, `"${replacement}"`); + out = out.replaceAll(`'${spec}'`, `'${replacement}'`); + } + return out; +} + +function transpileRecursive(fileUrl) { + const filePath = fileURLToPath(fileUrl); + if (moduleCache.has(filePath)) return moduleCache.get(filePath); + moduleCache.set(filePath, filePath); + + const source = readFileSync(filePath, "utf8"); + const imports = gatherImports(source); + const replacements = new Map(); + + for (const spec of imports) { + const resolved = resolveImport(spec, filePath); + if (!resolved) continue; + const depUrl = pathToFileURL(resolved).href; + const compiledPath = transpileRecursive(depUrl); + replacements.set(spec, pathToFileURL(compiledPath).href); + } + + const transpiled = ts.transpileModule(source, { + compilerOptions: { + module: ts.ModuleKind.ES2022, + target: ts.ScriptTarget.ES2022, + moduleResolution: ts.ModuleResolutionKind.Bundler, + jsx: ts.JsxEmit.ReactJSX, + }, + }).outputText; + + const rewritten = rewriteImports(transpiled, replacements); + const tempFile = join( + tempRoot, + `${basename(filePath).replace(/\W+/g, "_")}-${Date.now()}-${Math.random().toString(16).slice(2)}.mjs`, + ); + writeFileSync(tempFile, rewritten, "utf8"); + moduleCache.set(filePath, tempFile); + return tempFile; +} + +process.on("exit", () => { + rmSync(tempRoot, { recursive: true, force: true }); +}); + +const streamStorePath = new URL("../src/lib/stream/streamStore.ts", import.meta.url); +const streamSyncPath = new URL("../src/lib/stream/streamSync.ts", import.meta.url); +const streamListPath = new URL("../src/pages/sigilstream/list/virtualWindow.ts", import.meta.url); + +const streamStore = await import(pathToFileURL(transpileRecursive(streamStorePath.href)).href); +const streamSync = await import(pathToFileURL(transpileRecursive(streamSyncPath.href)).href); +const streamList = await import(pathToFileURL(transpileRecursive(streamListPath.href)).href); + +test("delta apply correctness updates and deletes records", () => { + const start = new Map([ + [ + "tokA", + { + token: "tokA", + url: "https://x/p/tokA", + title: "old", + author: "@a", + pulse: 10, + kind: "text", + shortBody: "old", + links: [], + updatedAt: 1, + }, + ], + ]); + + const next = streamStore.applyDeltaRowsToRecords(start, { + rows: [ + { token: "tokA", url: "https://x/p/tokA", preview: { title: "new", pulse: 22 } }, + { token: "tokB", url: "https://x/p/tokB", preview: { title: "insert", pulse: 30 } }, + { token: "tokA", url: "https://x/p/tokA", deleted: true }, + ], + }); + + assert.equal(next.has("tokA"), false); + assert.equal(next.get("tokB")?.title, "insert"); + assert.equal(next.get("tokB")?.pulse, 30); +}); + +test("offline-first sync path uses snapshot for empty local seal", () => { + assert.equal(streamSync.buildSyncUrl(null, 200), "/api/stream/snapshot?compact=1&limit=200"); + assert.equal(streamSync.buildSyncUrl("abc", 200), "/api/stream/delta?after=abc&limit=200"); + assert.equal(streamSync.buildSyncUrl("abc", 200, "cur"), "/api/stream/delta?after=abc&limit=200&cursor=cur"); +}); + +test("worker fallback control flag can be toggled", () => { + streamStore.setStreamWorkerFailedForTests(false); + streamStore.setStreamWorkerFailedForTests(true); + streamStore.setStreamWorkerFailedForTests(false); + assert.equal(typeof streamStore.setStreamWorkerFailedForTests, "function"); +}); + + + +test("syncStreamDelta paginates until latest seal", async () => { + const originalFetch = globalThis.fetch; + const originalGetSeal = streamStore.streamStore.getSeal; + const originalApplyDelta = streamStore.streamStore.applyDelta; + const originalGetSourceDigest = streamStore.streamStore.getSourceDigest; + + let localSeal = "old-seal"; + const applied = []; + const responses = [ + { seal: "page-1", latestPulse: 10, total: 3 }, + { seal: "page-1", latestSeal: "page-3", rows: [{ token: "t1", url: "https://x/p/t1" }] }, + { seal: "page-2", latestSeal: "page-3", rows: [{ token: "t2", url: "https://x/p/t2" }] }, + { seal: "page-3", latestSeal: "page-3", rows: [{ token: "t3", url: "https://x/p/t3" }] }, + ]; + + globalThis.fetch = async () => ({ ok: true, json: async () => responses.shift() }); + streamStore.streamStore.getSeal = async () => localSeal; + streamStore.streamStore.getSourceDigest = async () => "digest-a"; + streamStore.streamStore.applyDelta = async (delta) => { + applied.push(delta); + if (delta.seal) localSeal = delta.seal; + }; + + try { + const changed = await streamSync.syncStreamDelta(1); + assert.equal(changed, 3); + assert.equal(applied.length, 3); + assert.equal(applied[0].seal, "page-1"); + assert.equal(applied[2].seal, "page-3"); + assert.equal(localSeal, "page-3"); + } finally { + globalThis.fetch = originalFetch; + streamStore.streamStore.getSeal = originalGetSeal; + streamStore.streamStore.applyDelta = originalApplyDelta; + streamStore.streamStore.getSourceDigest = originalGetSourceDigest; + } +}); + + +test("syncStreamDelta forces snapshot when source digest changes", async () => { + const originalFetch = globalThis.fetch; + const originalGetSeal = streamStore.streamStore.getSeal; + const originalGetSourceDigest = streamStore.streamStore.getSourceDigest; + const originalApplyDelta = streamStore.streamStore.applyDelta; + + const seenUrls = []; + const applied = []; + const responses = [ + { seal: "head-seal", latestPulse: 10, total: 2, sourceDigest: "digest-b" }, + { seal: "next-seal", latestSeal: "next-seal", rows: [{ token: "t1", url: "https://x/p/t1" }] }, + ]; + + globalThis.fetch = async (url) => { + seenUrls.push(String(url)); + return { ok: true, json: async () => responses.shift() }; + }; + + streamStore.streamStore.getSeal = async () => "head-seal"; + streamStore.streamStore.getSourceDigest = async () => "digest-a"; + streamStore.streamStore.applyDelta = async (delta) => { + applied.push(delta); + }; + + try { + const changed = await streamSync.syncStreamDelta(50); + assert.equal(changed, 1); + assert.equal(seenUrls[1], "/api/stream/snapshot?compact=1&limit=50"); + assert.equal(applied[0].sourceDigest, "digest-b"); + } finally { + globalThis.fetch = originalFetch; + streamStore.streamStore.getSeal = originalGetSeal; + streamStore.streamStore.getSourceDigest = originalGetSourceDigest; + streamStore.streamStore.applyDelta = originalApplyDelta; + } +}); + +test("virtual list windowing computes bounded start/end", () => { + const nearTop = streamList.computeVirtualWindow(0, 900, 100); + assert.equal(nearTop.start, 0); + assert.ok(nearTop.end > 0); + + const mid = streamList.computeVirtualWindow(2400, 900, 100); + assert.ok(mid.start > 0); + assert.ok(mid.end <= 100); + assert.ok(mid.end > mid.start); +});