diff --git a/browser-host/relay-broker.mjs b/browser-host/relay-broker.mjs index 4e0c25e50f..abfb1bd5a1 100644 --- a/browser-host/relay-broker.mjs +++ b/browser-host/relay-broker.mjs @@ -338,6 +338,34 @@ async function relayAuthority(fetch, relay) { // NIP-IA snapshots require the explicit relay signing identity. The NIP-11 // contact-key fallback used by older channel reads cannot grant this authority. ...(nip11.self === author ? { archiveAuthority: author } : {}), + // Server-owned read state (`/buzz/v1`). Absent means NIP-RS stays in use. + ...(readStateV1(nip11.buzz_v1) + ? { readStateV1: readStateV1(nip11.buzz_v1) } + : {}), + }; +} +/** The advertised `buzz_v1` descriptor the client needs, or undefined. */ +function readStateV1(value) { + if ( + !value || + typeof value !== "object" || + value.version !== 1 || + value.base_path !== "/buzz/v1" || + !Array.isArray(value.capabilities) || + !value.capabilities.includes("ping") + ) + return undefined; + return { + capabilities: value.capabilities.filter((name) => typeof name === "string"), + eligibleKinds: Array.isArray(value.eligible_kinds) + ? value.eligible_kinds.filter(Number.isSafeInteger) + : [9, 40002, 45001, 45003], + maxChannels: Number.isSafeInteger(value.max_channels) + ? value.max_channels + : 20, + maxIntents: Number.isSafeInteger(value.max_intents) + ? value.max_intents + : 100, }; } /** @@ -1418,6 +1446,115 @@ export function relayBrokerPlugin({ }); } } + if ( + route === "/api/relay/read-v1-sidebar" || + route === "/api/relay/read-v1-threads" || + route === "/api/relay/read-v1-write" || + route === "/api/relay/read-v1-coverage" + ) { + // Fixed `/buzz/v1/me` routes only; NIP-98 signing stays in this process. + if (!(await getAuthority(relay)).readStateV1) + return json(res, 404, { error: "Read state API unavailable" }); + const coverage = route === "/api/relay/read-v1-coverage"; + const write = coverage || route === "/api/relay/read-v1-write"; + if (req.method !== (write ? "POST" : "GET")) + return json(res, 405, { error: "Method not allowed" }); + let body; + if (write) { + let raw = ""; + req.setEncoding("utf8"); + for await (const part of req) { + raw += part; + if (Buffer.byteLength(raw) > 64 * 1024) + return json(res, 413, { error: "Read intents too large" }); + } + try { + const parsed = JSON.parse(raw); + const items = coverage ? parsed.message_ids : parsed.intents; + if (!Array.isArray(items) || !items.length) throw 0; + body = JSON.stringify( + coverage ? { message_ids: items } : { intents: items }, + ); + } catch { + return json(res, 400, { error: "Invalid read intents" }); + } + } + const query = new URLSearchParams(); + for (const name of ["limit", "cursor", "channel_ids"]) { + const value = url.searchParams.get(name); + if (value !== null) query.set(name, value); + } + const path = `/buzz/v1/me/${coverage ? "coverage" : write ? "read-state" : route === "/api/relay/read-v1-threads" ? "threads" : "sidebar"}${query.size ? `?${query}` : ""}`; + const requestSignal = AbortSignal.any([ + cancel.signal, + AbortSignal.timeout(UPSTREAM_TIMEOUT_MS), + ]); + try { + const response = await admittedApiRequest( + admissions(relay, viewer).api, + () => { + requestSignal.throwIfAborted(); + const auth = finalizeEvent( + { + kind: 27235, + created_at: Math.floor(Date.now() / 1000), + content: "", + tags: [ + ["u", `${relay}${path}`], + ["method", write ? "POST" : "GET"], + ...(body === undefined + ? [] + : [ + [ + "payload", + createHash("sha256").update(body).digest("hex"), + ], + ]), + ["nonce", randomBytes(16).toString("hex")], + ], + }, + key, + ); + return fetchUpstream(`${relay}${path}`, { + method: write ? "POST" : "GET", + headers: { + ...(body === undefined + ? {} + : { "Content-Type": "application/json" }), + Authorization: + "Nostr " + + Buffer.from(JSON.stringify(auth)).toString("base64"), + }, + body, + redirect: "error", + signal: requestSignal, + }); + }, + requestSignal, + write ? "foreground" : "background", + ); + const text = await response.text(); + if (Buffer.byteLength(text) > 1024 * 1024) + return json(res, 502, { + error: "Read state response too large", + }); + res.writeHead(response.status, { + "Content-Type": "application/json", + "Cache-Control": "no-store", + ...(response.headers.get("retry-after") + ? { "Retry-After": response.headers.get("retry-after") } + : {}), + }); + return res.end(text); + } catch (error) { + if (cancel.signal.aborted) return; + return json(res, 502, { + error: + error instanceof Error ? error.message : "Read state failed", + sent: false, + }); + } + } if (route === "/api/relay/session" && req.method === "GET") { return json(res, 200, { viewer, diff --git a/src-tauri/build.rs b/src-tauri/build.rs index 05737b10b8..fea77e9b51 100644 --- a/src-tauri/build.rs +++ b/src-tauri/build.rs @@ -79,6 +79,7 @@ fn main() { "relay_sign_read_state", "relay_publish_read_state", "relay_http", + "relay_read_v1", "relay_workflow_runs", "relay_admin_discover", "relay_admin_request", diff --git a/src-tauri/capabilities/default.json b/src-tauri/capabilities/default.json index 018db8def2..a8772ee80e 100644 --- a/src-tauri/capabilities/default.json +++ b/src-tauri/capabilities/default.json @@ -41,6 +41,7 @@ "allow-relay-sign-read-state", "allow-relay-publish-read-state", "allow-relay-http", + "allow-relay-read-v1", "allow-relay-workflow-runs", "allow-relay-admin-discover", "allow-relay-admin-request", diff --git a/src-tauri/src/browser_permissions_tests.rs b/src-tauri/src/browser_permissions_tests.rs index e7be18869e..f4b8e17b5a 100644 --- a/src-tauri/src/browser_permissions_tests.rs +++ b/src-tauri/src/browser_permissions_tests.rs @@ -94,6 +94,7 @@ fn native_command_permissions_isolate_webviews() { "relay_sign_read_state", "relay_publish_read_state", "relay_http", + "relay_read_v1", "relay_workflow_runs", "relay_project_git", "relay_project_git_cancel", diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 64aeea4f1f..b3b22ab5ca 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -75,9 +75,9 @@ use relay::{ relay_channel_publish, relay_channel_sign, relay_decode_read_state, relay_decode_reminders, relay_decode_sidebar, relay_direct_message, relay_git_authorization, relay_http, relay_kit_decode, relay_kit_prepare, relay_kit_sign, relay_project_git, - relay_project_git_cancel, relay_publish_read_state, relay_sign, relay_sign_read_state, - relay_sign_reminder, relay_sign_sidebar, relay_upload, relay_upload_begin, relay_upload_cancel, - relay_upload_chunk, relay_workflow_runs, + relay_project_git_cancel, relay_publish_read_state, relay_read_v1, relay_sign, + relay_sign_read_state, relay_sign_reminder, relay_sign_sidebar, relay_upload, + relay_upload_begin, relay_upload_cancel, relay_upload_chunk, relay_workflow_runs, }; use relay::{huddle_close, huddle_open, huddle_pcm, huddle_touch, Huddles}; mod terminal; @@ -515,6 +515,7 @@ fn commands() -> impl Fn(tauri::ipc::Invoke) -> bool + Sen relay_sign_read_state, relay_publish_read_state, relay_http, + relay_read_v1, relay_workflow_runs, relay_admin::relay_admin_discover, relay_admin::relay_admin_request, diff --git a/src-tauri/src/relay.rs b/src-tauri/src/relay.rs index 13ad7004b4..c7154e2362 100644 --- a/src-tauri/src/relay.rs +++ b/src-tauri/src/relay.rs @@ -37,6 +37,8 @@ mod media_stream; pub(crate) use media_stream::{media_stream_base, MediaStream}; mod project_git; pub(crate) use project_git::{relay_project_git, relay_project_git_cancel}; +mod read_v1; +pub(crate) use read_v1::relay_read_v1; type Result = std::result::Result; const MAX_BODY: usize = 1024 * 1024; const MAX_RESPONSE: usize = 16 * 1024 * 1024; diff --git a/src-tauri/src/relay/read_v1.rs b/src-tauri/src/relay/read_v1.rs new file mode 100644 index 0000000000..556dad921a --- /dev/null +++ b/src-tauri/src/relay/read_v1.rs @@ -0,0 +1,220 @@ +//! `/buzz/v1/me` server-owned read state: fixed routes, NIP-98 signed by the +//! person's key. Mirrors the web broker's read-v1 routes (browser-host/relay-broker.mjs). +use super::{origin, send, workflow_uuid, IdentityHost, RelayResponse, Result}; +use url::Url; + +/// Matches the relay's per-request channel cap (`max_channels`). +const MAX_CHANNEL_IDS: usize = 20; +/// A batch of read intents or coverage IDs never needs more. +const MAX_BODY: usize = 64 * 1024; +/// Sidebar pages and outcomes are compact JSON. +const MAX_RESPONSE: usize = 1024 * 1024; + +#[derive(Clone, Copy, serde::Deserialize)] +#[serde(rename_all = "lowercase")] +pub(crate) enum ReadV1Operation { + Sidebar, + Write, + Coverage, +} + +impl ReadV1Operation { + fn route(self) -> &'static str { + match self { + Self::Sidebar => "sidebar", + Self::Write => "read-state", + Self::Coverage => "coverage", + } + } +} + +fn read_v1_url( + community: &str, + op: ReadV1Operation, + cursor: Option<&str>, + channel_ids: Option<&[String]>, +) -> Result { + let mut url = origin(community)? + .join(&format!("/buzz/v1/me/{}", op.route())) + .map_err(|_| "Invalid relay path".to_string())?; + match op { + ReadV1Operation::Sidebar => { + let cursor = cursor + .map(|cursor| { + if cursor.is_empty() + || cursor.len() > 512 + || !cursor.bytes().all(|b| b.is_ascii_graphic()) + { + return Err("Invalid read-state cursor".to_string()); + } + Ok(cursor) + }) + .transpose()?; + if let Some(ids) = channel_ids { + if ids.is_empty() + || ids.len() > MAX_CHANNEL_IDS + || ids.iter().any(|id| !workflow_uuid(id)) + { + return Err("Invalid read-state channels".into()); + } + } + if cursor.is_some() || channel_ids.is_some() { + let mut query = url.query_pairs_mut(); + if let Some(cursor) = cursor { + query.append_pair("cursor", cursor); + } + if let Some(ids) = channel_ids { + query.append_pair("channel_ids", &ids.join(",")); + } + } + } + _ => { + if cursor.is_some() || channel_ids.is_some() { + return Err("Invalid read-state request".into()); + } + } + } + Ok(url) +} + +/// The relay reads one non-empty array; reserializing drops anything else. +fn read_v1_body(op: ReadV1Operation, body: Option) -> Result { + let body = body.ok_or("Invalid read-state request")?; + if body.len() > MAX_BODY { + return Err("Read-state request is too large".into()); + } + let key = match op { + ReadV1Operation::Write => "intents", + ReadV1Operation::Coverage => "message_ids", + ReadV1Operation::Sidebar => return Err("Invalid read-state request".into()), + }; + let value: serde_json::Value = + serde_json::from_str(&body).map_err(|_| "Invalid read-state request")?; + let items = value + .get(key) + .and_then(serde_json::Value::as_array) + .filter(|items| !items.is_empty()) + .ok_or("Invalid read-state request")?; + serde_json::to_string(&serde_json::json!({ key: items })) + .map_err(|_| "Invalid read-state request".into()) +} + +#[tauri::command] +pub(crate) async fn relay_read_v1( + host: tauri::State<'_, IdentityHost>, + community: String, + op: ReadV1Operation, + cursor: Option, + channel_ids: Option>, + body: Option, +) -> Result { + let url = read_v1_url(&community, op, cursor.as_deref(), channel_ids.as_deref())?; + let (method, body) = match op { + ReadV1Operation::Sidebar => { + if body.is_some() { + return Err("Invalid read-state request".into()); + } + ("GET", None) + } + _ => ("POST", Some(read_v1_body(op, body)?)), + }; + send(host.inner(), url, method, body, true, MAX_RESPONSE).await +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn sidebar_url_bounds_its_query() { + let url = read_v1_url("https://relay.test", ReadV1Operation::Sidebar, None, None).unwrap(); + assert_eq!(url.as_str(), "https://relay.test/buzz/v1/me/sidebar"); + + let id = "550e8400-e29b-41d4-a716-446655440000"; + let url = read_v1_url( + "https://relay.test", + ReadV1Operation::Sidebar, + Some("cursor=evil&x=1"), + Some(&[id.into()]), + ) + .unwrap(); + assert_eq!( + url.as_str(), + "https://relay.test/buzz/v1/me/sidebar?cursor=cursor%3Devil%26x%3D1&channel_ids=550e8400-e29b-41d4-a716-446655440000" + ); + + let long = "x".repeat(513); + let rejected_cursors = ["", long.as_str(), "line\nbreak"]; + for cursor in rejected_cursors { + assert!( + read_v1_url( + "https://relay.test", + ReadV1Operation::Sidebar, + Some(cursor), + None + ) + .is_err(), + "cursor {cursor:?}" + ); + } + let rejected_ids: Vec> = vec![ + vec![], + vec!["not-a-uuid".into()], + vec!["550E8400-E29B-41D4-A716-446655440000".into()], + vec![id.into(); MAX_CHANNEL_IDS + 1], + ]; + for ids in rejected_ids { + assert!( + read_v1_url( + "https://relay.test", + ReadV1Operation::Sidebar, + None, + Some(&ids) + ) + .is_err(), + "ids {ids:?}" + ); + } + } + + #[test] + fn writes_take_no_query_and_keep_only_their_array() { + for op in [ReadV1Operation::Write, ReadV1Operation::Coverage] { + assert!(read_v1_url("https://relay.test", op, Some("c"), None).is_err()); + assert!(read_v1_url("https://relay.test", op, None, Some(&["x".into()])).is_err()); + } + assert_eq!( + read_v1_body( + ReadV1Operation::Write, + Some(r#"{"intents":[{"type":"follow"}],"extra":1}"#.into()) + ) + .unwrap(), + r#"{"intents":[{"type":"follow"}]}"# + ); + assert_eq!( + read_v1_body( + ReadV1Operation::Coverage, + Some(r#"{"message_ids":["ab"]}"#.into()) + ) + .unwrap(), + r#"{"message_ids":["ab"]}"# + ); + for (op, body) in [ + (ReadV1Operation::Write, None), + (ReadV1Operation::Write, Some("not json".into())), + (ReadV1Operation::Write, Some(r#"{"intents":[]}"#.into())), + ( + ReadV1Operation::Write, + Some(r#"{"message_ids":["ab"]}"#.into()), + ), + ( + ReadV1Operation::Coverage, + Some(r#"{"intents":[{}]}"#.into()), + ), + (ReadV1Operation::Coverage, Some("x".repeat(MAX_BODY + 1))), + (ReadV1Operation::Sidebar, Some(r#"{"intents":[{}]}"#.into())), + ] { + assert!(read_v1_body(op, body).is_err()); + } + } +} diff --git a/src/features/relay/live.ts b/src/features/relay/live.ts index 6128046b3b..a1d1bdc8af 100644 --- a/src/features/relay/live.ts +++ b/src/features/relay/live.ts @@ -11,6 +11,10 @@ import { type ObserverFrame, } from "../agents/observer.ts"; import type { EventTemplate, VerifiedEvent } from "nostr-tools"; + +/** Relay-signed `/buzz/v1` ping (`buzz_v1.ping_kind`): the p-tagged viewer's + * read state changed. Content names the changed channels and threads. */ +export const READ_STATE_CHANGED_KIND = 24300; import { eventDto } from "./events.ts"; import { EMOJI_SET } from "./emoji.ts"; import type { @@ -627,6 +631,9 @@ export function subscribeRelayTraffic( })), ...(route.id === "membership" ? [ + // `/buzz/v1` ping: another connection changed this viewer's + // read state. Relays without it never match. + { kinds: [READ_STATE_CHANGED_KIND], "#p": [viewer], limit: 0 }, { kinds: [30078], authors: [viewer], diff --git a/src/features/relay/native.ts b/src/features/relay/native.ts index 4585c6f45f..1d8aab6fd4 100644 --- a/src/features/relay/native.ts +++ b/src/features/relay/native.ts @@ -62,6 +62,12 @@ import { } from "./transport"; import { nativeSidebar } from "./native-sidebar"; import { readApiFailure } from "./http-admission"; +import type { + ReadApiHost, + ReadOutcome, + ServerChannelRow, + SidebarPage, +} from "./read-api"; async function nativeReadInvoke( command: string, @@ -320,6 +326,45 @@ export async function nativeRelayInfo(community: string, signal?: AbortSignal) { return info as Record; } +/** The advertised `buzz_v1` descriptor the client needs, or undefined. Mirrors + * the web broker's `readStateV1()` gate (browser-host/relay-broker.mjs): when it + * does not hold, `readApi` stays off and NIP-RS behaves as before. */ +export function readV1Descriptor(value: unknown): + | { + capabilities: string[]; + eligibleKinds: number[]; + maxChannels: number; + maxIntents: number; + } + | undefined { + if (!value || typeof value !== "object" || Array.isArray(value)) + return undefined; + const descriptor = value as Record; + if ( + descriptor.version !== 1 || + descriptor.base_path !== "/buzz/v1" || + !Array.isArray(descriptor.capabilities) || + !descriptor.capabilities.includes("ping") + ) + return undefined; + return { + capabilities: descriptor.capabilities.filter( + (name): name is string => typeof name === "string", + ), + eligibleKinds: Array.isArray(descriptor.eligible_kinds) + ? descriptor.eligible_kinds.filter((kind): kind is number => + Number.isSafeInteger(kind), + ) + : [9, 40002, 45001, 45003], + maxChannels: Number.isSafeInteger(descriptor.max_channels) + ? (descriptor.max_channels as number) + : 20, + maxIntents: Number.isSafeInteger(descriptor.max_intents) + ? (descriptor.max_intents as number) + : 100, + }; +} + export async function connectNativeTransport( community: string, signal?: AbortSignal, @@ -342,6 +387,7 @@ export async function connectNativeTransport( | { sync?: { version?: unknown } } | undefined; const artifactSync = artifacts?.sync?.version === 1; + const readV1 = readV1Descriptor(info.buzz_v1); const transport = await connectSignedTransport( nativeRelaySigner(origin), origin, @@ -729,6 +775,95 @@ export async function connectNativeTransport( await acceptReadStatePublish(response, event.id); }, }, + // Server-owned read state, when the relay advertises `buzz_v1`. The host + // signs NIP-98 in Rust; the page never holds the key. + ...(readV1 + ? { + readApi: ((): ReadApiHost => { + const call = async ( + op: "sidebar" | "write" | "coverage", + args: { + cursor?: string | undefined; + channelIds?: readonly string[] | undefined; + body?: string | undefined; + }, + signal: AbortSignal, + priority: "foreground" | "background", + ): Promise => { + const response = await admitSignedRequest( + origin, + transport.viewer, + async () => + nativeResponse( + await nativeReadInvoke<{ + status: number; + headers: Record; + body: string; + }>("relay_read_v1", { + community: origin, + op, + cursor: args.cursor ?? null, + channelIds: args.channelIds ?? null, + body: args.body ?? null, + }), + ), + signal, + priority, + ); + signal.throwIfAborted(); + if (!response.ok) + throw new Error((await readApiFailure(response)).error); + return (await response.json()) as T; + }; + return { + capabilities: readV1.capabilities, + eligibleKinds: readV1.eligibleKinds, + maxChannels: readV1.maxChannels, + maxIntents: readV1.maxIntents, + sidebar: (query, signal) => + call( + "sidebar", + { cursor: query.cursor, channelIds: query.channelIds }, + signal, + "background", + ), + write: (intents, signal) => + call< + Readonly<{ + outcomes: readonly ReadOutcome[]; + channels?: readonly ServerChannelRow[]; + ping_id?: string; + }> + >( + "write", + { body: JSON.stringify({ intents }) }, + signal, + "foreground", + ), + ...(readV1.capabilities.includes("coverage") + ? { + coverage: ( + messageIds: readonly string[], + signal: AbortSignal, + ) => + call< + Readonly<{ + messages: Readonly< + Record + >; + }> + >( + "coverage", + { body: JSON.stringify({ message_ids: messageIds }) }, + signal, + "foreground", + ), + } + : {}), + }; + })(), + } + : {}), ...(readCommunity ? { async readStateSnapshot(signal: AbortSignal, _requestId, priority) { diff --git a/src/features/relay/read-api.test.ts b/src/features/relay/read-api.test.ts new file mode 100644 index 0000000000..4d76e74c1a --- /dev/null +++ b/src/features/relay/read-api.test.ts @@ -0,0 +1,843 @@ +// @vitest-environment jsdom +import { afterEach, beforeEach, expect, it, vi } from "vitest"; +import { + createReadApi, + type ReadApiHost, + type ReadIntent, + type ServerChannelRow, + type SidebarPage, +} from "./read-api"; +import type { RelayEvent } from "./events"; + +const viewer = "a".repeat(64); +const other = "b".repeat(64); +const channel = "00000000-0000-4000-8000-000000000001"; +const second = "00000000-0000-4000-8000-000000000002"; +const id = (n: number) => n.toString(16).padStart(64, "0"); + +const row = ( + patch: Partial & { channel_id?: string } = {}, +): ServerChannelRow => ({ + channel_id: channel, + unread: false, + mentions: 0, + read_through_id: id(1), + latest_id: id(1), + threads: [], + ...patch, +}); +const page = (...channels: ServerChannelRow[]): SidebarPage => ({ + channels, + next_cursor: null, +}); +function deferred() { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((res, rej) => { + resolve = res; + reject = rej; + }); + return { promise, resolve, reject }; +} +function fakeHost() { + const sidebar = vi.fn(); + const write = vi.fn(); + const host: ReadApiHost = { + capabilities: ["follow", "threads", "ping"], + eligibleKinds: [9], + maxChannels: 20, + maxIntents: 100, + sidebar, + write, + }; + return { host, sidebar, write }; +} +const message = ( + n: number, + patch: Partial = {}, + tags: string[][] = [], +): RelayEvent => + ({ + id: id(n), + pubkey: other, + kind: 9, + created_at: 1_700_000_000 + n, + content: "", + sig: "", + tags: [["h", channel], ...tags], + ...patch, + }) as RelayEvent; +const create = (host: ReadApiHost, storageKey?: string) => + createReadApi({ + host, + viewer, + storageKey, + isDm: () => false, + rootAuthor: () => undefined, + }); + +beforeEach(() => { + vi.useFakeTimers(); + localStorage.clear(); +}); +afterEach(() => { + vi.useRealTimers(); +}); + +it("an older targeted response cannot restore unread a newer one cleared", async () => { + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValueOnce(page(row())); + const api = create(host, "k"); + await api.load(); + const older = deferred(); + const newer = deferred(); + sidebar.mockReturnValueOnce(older.promise).mockReturnValueOnce(newer.promise); + api.refresh([channel]); + await vi.advanceTimersByTimeAsync(50); + api.refresh([channel]); + await vi.advanceTimersByTimeAsync(50); + newer.resolve(page(row({ read_through_id: id(3), latest_id: id(3) }))); + await vi.advanceTimersByTimeAsync(0); + older.resolve(page(row({ unread: true, mentions: 1, latest_id: id(3) }))); + await vi.advanceTimersByTimeAsync(0); + expect(api.row(channel)).toMatchObject({ unread: false, mentions: 0 }); + api.dispose(); +}); + +it("a full load issued before a targeted refresh keeps the refreshed channel", async () => { + const { host, sidebar } = fakeHost(); + const full = deferred(); + sidebar.mockReturnValueOnce(full.promise); + const api = create(host, "k"); + const loading = api.load(); + sidebar.mockResolvedValueOnce(page(row())); + api.refresh([channel]); + await vi.advanceTimersByTimeAsync(50); + full.resolve( + page( + row({ unread: true, latest_id: id(2) }), + row({ channel_id: second, unread: true }), + ), + ); + await loading; + expect(api.row(channel)?.unread).toBe(false); + expect(api.row(second)?.unread).toBe(true); + expect(api.complete()).toBe(true); + api.dispose(); +}); + +it("a live arrival newer than a response survives it and refetches", async () => { + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValueOnce(page(row())); + const api = create(host, "k"); + await api.load(); + const held = deferred(); + sidebar.mockReturnValueOnce(held.promise); + api.refresh([channel]); + await vi.advanceTimersByTimeAsync(50); + api.observe(message(5)); + expect(api.row(channel)).toMatchObject({ unread: true, latest_id: id(5) }); + sidebar.mockResolvedValueOnce(page(row({ unread: true, latest_id: id(5) }))); + held.resolve(page(row())); + await vi.advanceTimersByTimeAsync(0); + expect(api.row(channel)?.unread).toBe(true); + await vi.advanceTimersByTimeAsync(50); + expect(sidebar).toHaveBeenLastCalledWith( + { channelIds: [channel] }, + expect.anything(), + ); + api.dispose(); +}); + +it("a failed targeted refresh repairs with a full load", async () => { + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValueOnce(page(row())); + const api = create(host, "k"); + await api.load(); + sidebar.mockRejectedValueOnce(new Error("offline")); + api.refresh([channel]); + await vi.advanceTimersByTimeAsync(50); + sidebar.mockResolvedValueOnce(page(row({ unread: true, latest_id: id(2) }))); + await vi.advanceTimersByTimeAsync(2000); + expect(sidebar).toHaveBeenLastCalledWith({}, expect.anything()); + expect(api.row(channel)?.unread).toBe(true); + api.dispose(); +}); + +it("first launch starts the account once per install", async () => { + const { host, sidebar, write } = fakeHost(); + sidebar.mockResolvedValue(page(row())); + write.mockResolvedValue({ outcomes: [{ status: "applied" }] }); + const api = create(host, "k"); + await api.load(); + await vi.advanceTimersByTimeAsync(150); + expect(write).toHaveBeenCalledTimes(1); + expect(write.mock.calls[0]?.[0]).toEqual([ + { + type: "mark_through", + target: { channel_id: channel }, + message_id: id(1), + }, + ]); + api.dispose(); + const again = create(host, "k"); + await again.load(); + await vi.advanceTimersByTimeAsync(150); + expect(write).toHaveBeenCalledTimes(1); + again.dispose(); +}); + +it("keeps unresolved intents across restarts and retries with backoff", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar, write } = fakeHost(); + sidebar.mockResolvedValue(page(row({ unread: true, latest_id: id(2) }))); + write.mockRejectedValue(new Error("offline")); + const intent: ReadIntent = { + type: "mark_through", + target: { channel_id: channel }, + message_id: id(2), + }; + const api = create(host, "k"); + await api.load(); + api.enqueue([intent]); + await vi.advanceTimersByTimeAsync(150); + expect(write).toHaveBeenCalledTimes(1); + api.dispose(); + expect(JSON.parse(localStorage.getItem("k") ?? "[]")).toEqual([intent]); + + write.mockReset(); + write + .mockRejectedValueOnce(new Error("offline")) + .mockResolvedValue({ outcomes: [{ status: "applied" }] }); + const restarted = create(host, "k"); + await restarted.load(); + expect(write).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(1000); + expect(write).toHaveBeenCalledTimes(2); + expect(write.mock.calls[1]?.[0]).toEqual([intent]); + expect(localStorage.getItem("k")).toBeNull(); + restarted.dispose(); +}); + +it("skips its own ping and refreshes the channels another install's ping names", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar, write } = fakeHost(); + sidebar.mockResolvedValue(page(row({ unread: true, latest_id: id(2) }))); + write.mockResolvedValue({ + outcomes: [{ status: "applied" }], + channels: [row({ read_through_id: id(2), latest_id: id(2) })], + ping_id: id(99), + }); + const api = create(host, "k"); + await api.load(); + api.readLocally(channel, undefined, id(2)); + api.enqueue([ + { + type: "mark_through", + target: { channel_id: channel }, + message_id: id(2), + }, + ]); + await vi.advanceTimersByTimeAsync(150); + const loads = sidebar.mock.calls.length; + api.changed(JSON.stringify({ channels: [channel], threads: [] }), id(99)); + await vi.advanceTimersByTimeAsync(50); + expect(sidebar.mock.calls.length).toBe(loads); + api.changed( + JSON.stringify({ + channels: [], + threads: [{ channel_id: second, root_id: id(7) }], + }), + id(100), + ); + await vi.advanceTimersByTimeAsync(50); + expect(sidebar).toHaveBeenLastCalledWith( + { channelIds: [second] }, + expect.anything(), + ); + api.dispose(); +}); + +it("counts live mentions and follows replies only in listed threads", async () => { + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValue( + page( + row({ + threads: [ + { + root_id: id(10), + unread: false, + mentions: 0, + read_through_id: id(11), + latest_id: id(11), + }, + ], + }), + ), + ); + const api = create(host, "k"); + await api.load(); + api.observe(message(2, {}, [["p", viewer]])); + expect(api.row(channel)).toMatchObject({ unread: true, mentions: 1 }); + api.observe(message(12, {}, [["e", id(10), "", "reply"]])); + expect(api.row(channel)?.threads[0]).toMatchObject({ + root_id: id(10), + unread: true, + latest_id: id(12), + }); + api.observe(message(21, {}, [["e", id(20), "", "reply"]])); + expect(api.row(channel)?.threads).toHaveLength(1); + api.observe(message(3, { pubkey: viewer })); + expect(api.row(channel)).toMatchObject({ + unread: false, + mentions: 0, + read_through_id: id(3), + }); + api.dispose(); +}); + +it("drops a response read at an older revision than one already seen and asks again", async () => { + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValueOnce({ ...page(row()), read_revision: 4 }); + const api = create(host, "k"); + await api.load(); + sidebar + .mockResolvedValueOnce({ + ...page(row({ unread: true, latest_id: id(2) })), + read_revision: 3, + }) + .mockResolvedValueOnce({ + ...page(row({ read_through_id: id(2), latest_id: id(2) })), + read_revision: 5, + }); + api.refresh([channel]); + await vi.advanceTimersByTimeAsync(50); + expect(api.row(channel)?.unread).toBe(false); + await vi.advanceTimersByTimeAsync(50); + expect(sidebar).toHaveBeenCalledTimes(3); + expect(api.row(channel)?.read_through_id).toBe(id(2)); + expect(api.stale(4)).toBe(true); + expect(api.stale(5)).toBe(false); + api.dispose(); +}); + +it("on wake, probes one channel and loads everything only when the revision moved", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValueOnce({ ...page(row()), read_revision: 7 }); + const api = create(host, "k"); + await api.load(); + const hide = (state: "hidden" | "visible") => { + Object.defineProperty(document, "visibilityState", { + configurable: true, + value: state, + }); + document.dispatchEvent(new Event("visibilitychange")); + }; + sidebar.mockResolvedValueOnce({ ...page(row()), read_revision: 7 }); + hide("hidden"); + await vi.advanceTimersByTimeAsync(31000); + hide("visible"); + await vi.advanceTimersByTimeAsync(0); + expect(sidebar).toHaveBeenCalledTimes(2); + expect(sidebar).toHaveBeenLastCalledWith( + { channelIds: [channel] }, + expect.anything(), + ); + + sidebar + .mockResolvedValueOnce({ ...page(row()), read_revision: 8 }) + .mockResolvedValueOnce({ + ...page(row(), row({ channel_id: second, unread: true })), + read_revision: 8, + }); + hide("hidden"); + await vi.advanceTimersByTimeAsync(31000); + hide("visible"); + await vi.advanceTimersByTimeAsync(0); + expect(sidebar).toHaveBeenCalledTimes(4); + expect(sidebar).toHaveBeenLastCalledWith({}, expect.anything()); + expect(api.row(second)?.unread).toBe(true); + api.dispose(); +}); + +it("a ping that starts past the reconciled revision reloads everything", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValue({ + ...page(row(), row({ channel_id: second })), + read_revision: 45, + }); + const api = create(host, "k"); + await api.load(); + const ping = (from: number, to: number, channelId: string) => + JSON.stringify({ + read_revision_from: from, + read_revision: to, + channels: [channelId], + threads: [], + }); + // Contiguous: only the named channel. + sidebar.mockResolvedValueOnce({ ...page(row()), read_revision: 46 }); + api.changed(ping(45, 46, channel), id(200)); + await vi.advanceTimersByTimeAsync(50); + expect(sidebar).toHaveBeenLastCalledWith( + { channelIds: [channel] }, + expect.anything(), + ); + const calls = sidebar.mock.calls.length; + // 47 was dropped; 48 starts past what was reconciled. + sidebar.mockResolvedValue({ + ...page(row(), row({ channel_id: second })), + read_revision: 48, + }); + api.changed(ping(47, 48, second), id(201)); + await vi.advanceTimersByTimeAsync(50); + // The missing ping gets a moment to arrive before everything reloads. + expect(sidebar.mock.calls.slice(calls).map(([query]) => query)).toEqual([ + { channelIds: [second] }, + ]); + await vi.advanceTimersByTimeAsync(1500); + expect(sidebar.mock.calls.slice(calls).map(([query]) => query)).toEqual([ + { channelIds: [second] }, + {}, + ]); + api.dispose(); +}); + +it("waits out pings held behind a concurrent own write instead of reloading", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar, write } = fakeHost(); + sidebar.mockResolvedValue({ ...page(row()), read_revision: 431 }); + const api = create(host, "k"); + await api.load(); + sidebar.mockResolvedValue({ ...page(row()), read_revision: 434 }); + const calls = sidebar.mock.calls.length; + const ping = (from: number, to: number) => + JSON.stringify({ + read_revision_from: from, + read_revision: to, + channels: [channel], + threads: [], + }); + // Two other installs mark the same message first; their pings land while + // this install's write is in flight and wait for its answer, 433..434. + let answer!: () => void; + write.mockImplementationOnce( + (intents) => + new Promise((resolve) => { + answer = () => + resolve({ + outcomes: intents.map(() => ({ status: "applied" })), + read_revision_from: 433, + read_revision: 434, + ping_id: id(302), + }); + }), + ); + api.enqueue([ + { + type: "mark_through", + target: { channel_id: channel }, + message_id: id(1), + }, + ]); + await vi.advanceTimersByTimeAsync(150); + api.changed(ping(431, 432), id(300)); + api.changed(ping(432, 433), id(301)); + api.changed(ping(433, 434), id(302)); + answer(); + await vi.advanceTimersByTimeAsync(3000); + const queries = sidebar.mock.calls.slice(calls).map(([query]) => query); + expect(queries).not.toContainEqual({}); + expect(queries.length).toBeGreaterThan(0); + api.dispose(); +}); + +it("keeps current rows when every traversal mixes revisions, then repairs", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValueOnce({ ...page(row()), read_revision: 5 }); + const api = create(host, "k"); + await api.load(); + // Each traversal: page one is older than page two, and nothing covers it. + let revision = 5; + sidebar.mockImplementation(async (query) => + query.cursor + ? { ...page(row({ channel_id: second })), read_revision: ++revision } + : { + channels: [row({ unread: true, latest_id: id(9) })], + next_cursor: "next", + read_revision: revision, + }, + ); + await api.load(); + expect(sidebar).toHaveBeenCalledTimes(7); + expect(api.row(channel)?.unread).toBe(false); + expect(api.row(second)).toBeUndefined(); + expect(api.status()).toMatchObject({ status: "error" }); + sidebar.mockResolvedValue({ + ...page(row({ unread: true, latest_id: id(9) })), + read_revision: 20, + }); + await vi.advanceTimersByTimeAsync(2000); + expect(api.row(channel)?.unread).toBe(true); + expect(api.status()).toMatchObject({ status: "ready" }); + api.dispose(); +}); + +it("accepts a traversal whose newer revisions arrived as pings while it ran", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar } = fakeHost(); + const second_page = deferred(); + sidebar + .mockResolvedValueOnce({ + channels: [row({ unread: true, latest_id: id(9) })], + next_cursor: "next", + read_revision: 5, + }) + .mockReturnValueOnce(second_page.promise); + const api = create(host, "k"); + const loading = api.load(); + await vi.advanceTimersByTimeAsync(0); + sidebar.mockResolvedValue({ + ...page(row({ read_through_id: id(9), latest_id: id(9) })), + read_revision: 6, + }); + api.changed( + JSON.stringify({ + read_revision_from: 5, + read_revision: 6, + channels: [channel], + threads: [], + }), + id(300), + ); + await vi.advanceTimersByTimeAsync(50); + second_page.resolve({ + ...page(row({ channel_id: second })), + read_revision: 6, + }); + await loading; + expect(sidebar).toHaveBeenCalledTimes(3); + expect(api.row(channel)?.unread).toBe(false); + expect(api.row(second)).toBeDefined(); + expect(api.status()).toMatchObject({ status: "ready" }); + api.dispose(); +}); + +it("without followed_threads, a reply in a remembered thread asks the relay whether it is still followed", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar, write } = fakeHost(); + const thread = (latest: number) => ({ + root_id: id(10), + unread: true, + mentions: 0, + read_through_id: null, + latest_id: id(latest), + }); + sidebar.mockResolvedValueOnce(page(row({ threads: [thread(11)] }))); + write.mockResolvedValue({ outcomes: [{ status: "applied" }] }); + const api = create(host, "k"); + await api.load(); + api.readLocally(channel, id(10), id(11)); + expect(api.following(channel, id(10))).toBe(true); + // Still followed: the relay lists it. + sidebar.mockResolvedValueOnce(page(row({ threads: [thread(12)] }))); + api.observe(message(12, {}, [["e", id(10), "", "reply"]])); + expect(api.row(channel)?.threads).toEqual([]); + await vi.advanceTimersByTimeAsync(1050); + expect(api.row(channel)?.threads[0]).toMatchObject({ latest_id: id(12) }); + // Unfollowed on another install: the relay does not list it. + api.readLocally(channel, id(10), id(12)); + sidebar.mockResolvedValueOnce(page(row())); + api.observe(message(13, {}, [["e", id(10), "", "reply"]])); + await vi.advanceTimersByTimeAsync(1050); + expect(api.row(channel)?.threads).toEqual([]); + expect(api.following(channel, id(10))).toBe(false); + // Known unfollowed: no request at all. + const calls = sidebar.mock.calls.length; + api.observe(message(14, {}, [["e", id(10), "", "reply"]])); + await vi.advanceTimersByTimeAsync(1050); + expect(sidebar.mock.calls.length).toBe(calls); + expect(api.row(channel)?.threads).toEqual([]); + api.dispose(); +}); + +const reply = (n: number, root: number, patch: Partial = {}) => + message(n, patch, [["e", id(root), "", "reply"]]); + +it("followed_threads decides live replies; queued intents win until applied", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar, write } = fakeHost(); + sidebar.mockResolvedValueOnce({ + ...page(row({ followed_threads: [id(10)] })), + read_revision: 3, + }); + const api = create(host, "k"); + await api.load(); + expect(api.following(channel, id(10))).toBe(true); + expect(api.following(channel, id(20))).toBe(false); + // Listed: lights at once, no request. + api.observe(reply(11, 10)); + expect(api.row(channel)?.threads[0]).toMatchObject({ latest_id: id(11) }); + // Unlisted and not truncated: not followed, no request. + api.observe(reply(21, 20)); + await vi.advanceTimersByTimeAsync(1050); + expect(sidebar).toHaveBeenCalledTimes(1); + expect(api.row(channel)?.threads).toHaveLength(1); + // Unfollowed on another install: the refreshed row decides. + sidebar.mockResolvedValueOnce({ + ...page(row({ followed_threads: [] })), + read_revision: 4, + }); + api.changed( + JSON.stringify({ + read_revision_from: 3, + read_revision: 4, + channels: [channel], + }), + id(400), + ); + await vi.advanceTimersByTimeAsync(50); + expect(api.following(channel, id(10))).toBe(false); + api.observe(reply(12, 10)); + expect(api.row(channel)?.threads).toEqual([]); + // A follow intent counts while queued, even over a row read before it applied. + const applied = deferred>>(); + write.mockReturnValueOnce(applied.promise); + api.enqueue([ + { type: "follow", target: { channel_id: channel, root_id: id(10) } }, + ]); + api.refresh([channel]); + await vi.advanceTimersByTimeAsync(200); + expect(api.following(channel, id(10))).toBe(true); + api.observe(reply(13, 10)); + expect(api.row(channel)?.threads[0]).toMatchObject({ latest_id: id(13) }); + sidebar.mockResolvedValueOnce({ + ...page(row({ followed_threads: [id(10)] })), + read_revision: 5, + }); + applied.resolve({ + outcomes: [{ status: "applied" }], + read_revision: 5, + read_revision_from: 4, + }); + await vi.advanceTimersByTimeAsync(50); + expect(api.following(channel, id(10))).toBe(true); + api.dispose(); +}); + +it("past the cap, an unlisted reply asks once per root until the revision moves", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar } = fakeHost(); + const truncated = (revision: number, followed: number[] = []) => ({ + ...page( + row({ + followed_threads: followed.map(id), + followed_threads_truncated: true, + }), + ), + read_revision: revision, + }); + sidebar.mockResolvedValueOnce(truncated(3)); + const api = create(host, "k"); + await api.load(); + expect(api.following(channel, id(20))).toBeUndefined(); + sidebar.mockResolvedValueOnce(truncated(3)); + api.observe(reply(21, 20)); + api.observe(reply(22, 20)); + await vi.advanceTimersByTimeAsync(999); + expect(sidebar).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(51); + expect(sidebar).toHaveBeenCalledTimes(2); + expect(api.row(channel)?.threads).toEqual([]); + expect(api.following(channel, id(20))).toBe(false); + // Not followed as of revision 3: further replies ask nothing. + api.observe(reply(23, 20)); + await vi.advanceTimersByTimeAsync(1050); + expect(sidebar).toHaveBeenCalledTimes(2); + // The row moved past it: ask again; now listed. + sidebar.mockResolvedValueOnce(truncated(4)); + api.refresh([channel]); + await vi.advanceTimersByTimeAsync(50); + expect(api.following(channel, id(20))).toBeUndefined(); + sidebar.mockResolvedValueOnce(truncated(4, [20])); + api.observe(reply(24, 20)); + await vi.advanceTimersByTimeAsync(1050); + expect(sidebar).toHaveBeenCalledTimes(4); + expect(api.following(channel, id(20))).toBe(true); + api.dispose(); +}); + +it("an own new root counts as followed until the relay's next row", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValueOnce(page(row({ followed_threads: [] }))); + const api = create(host, "k"); + await api.load(); + api.observe(message(5, { pubkey: viewer })); + expect(api.following(channel, id(5))).toBe(true); + api.observe(reply(6, 5)); + expect(api.row(channel)?.threads[0]).toMatchObject({ root_id: id(5) }); + api.dispose(); +}); + +it("sends follow, unfollow, follow in order instead of dropping the repeat", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar, write } = fakeHost(); + sidebar.mockResolvedValue(page(row())); + write.mockImplementation(async (intents) => ({ + outcomes: intents.map(() => ({ status: "applied" })), + })); + const api = create(host, "k"); + await api.load(); + const target = { channel_id: channel, root_id: id(10) }; + api.enqueue([{ type: "follow", target }]); + api.enqueue([{ type: "unfollow", target }]); + api.enqueue([{ type: "follow", target }]); + await vi.advanceTimersByTimeAsync(150); + expect( + write.mock.calls.flatMap(([intents]) => intents.map((i) => i.type)), + ).toEqual(["follow", "unfollow", "follow"]); + expect(api.following(channel, id(10))).toBe(true); + api.dispose(); +}); + +it("counts a repeated live delivery once and asks about an unknown channel once", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValue(page(row())); + const api = create(host, "k"); + await api.load(); + const mention = message(2, {}, [["p", viewer]]); + api.observe(mention); + api.observe(mention); + expect(api.row(channel)?.mentions).toBe(1); + const elsewhere = (n: number) => message(n, { tags: [["h", second]] }); + api.observe(elsewhere(3)); + api.observe(elsewhere(4)); + await vi.advanceTimersByTimeAsync(50); + api.observe(elsewhere(5)); + await vi.advanceTimersByTimeAsync(50); + expect(sidebar.mock.calls.filter(([query]) => query.channelIds)).toHaveLength( + 1, + ); + api.dispose(); +}); + +it("drops a write batch the relay refuses as malformed", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar, write } = fakeHost(); + sidebar.mockResolvedValue(page(row())); + write + .mockRejectedValueOnce( + Object.assign(new Error("Read write failed (400)"), { status: 400 }), + ) + .mockResolvedValue({ outcomes: [{ status: "applied" }] }); + const api = create(host, "k"); + await api.load(); + api.enqueue([ + { + type: "mark_through", + target: { channel_id: channel }, + message_id: id(1), + }, + ]); + await vi.advanceTimersByTimeAsync(150); + expect(api.status().pending).toBe(0); + api.enqueue([ + { + type: "mark_through", + target: { channel_id: channel }, + message_id: id(2), + }, + ]); + await vi.advanceTimersByTimeAsync(1150); + expect(write).toHaveBeenCalledTimes(2); + expect(write.mock.calls[1]?.[0]).toHaveLength(1); + api.dispose(); +}); + +it("a targeted refresh that keeps coming back stale falls back to a full load", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar } = fakeHost(); + sidebar.mockResolvedValueOnce({ ...page(row()), read_revision: 5 }); + const api = create(host, "k"); + await api.load(); + sidebar.mockResolvedValue({ ...page(row()), read_revision: 4 }); + api.refresh([channel]); + await vi.advanceTimersByTimeAsync(1000); + // One ask plus three re-asks, then the full-load repair after its backoff. + expect(sidebar.mock.calls.filter(([query]) => query.channelIds)).toHaveLength( + 4, + ); + await vi.advanceTimersByTimeAsync(2000); + expect( + sidebar.mock.calls.filter(([query]) => !query.channelIds).length, + ).toBeGreaterThan(1); + api.dispose(); +}); + +it("a write answer without the follow list keeps the last one, and the applied intent still wins", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar, write } = fakeHost(); + sidebar.mockResolvedValueOnce({ + ...page(row({ followed_threads: [id(10)] })), + read_revision: 3, + }); + write.mockResolvedValue({ + outcomes: [{ status: "applied" }], + channels: [row()], + read_revision: 4, + read_revision_from: 3, + }); + const api = create(host, "k"); + await api.load(); + api.enqueue([ + { type: "unfollow", target: { channel_id: channel, root_id: id(10) } }, + ]); + await vi.advanceTimersByTimeAsync(150); + expect(api.status().pending).toBe(0); + expect(api.row(channel)?.followed_threads).toEqual([id(10)]); + expect(api.following(channel, id(10))).toBe(false); + expect(api.following(channel, id(20))).toBe(false); + api.observe(reply(11, 10)); + expect(api.row(channel)?.threads).toEqual([]); + api.dispose(); +}); + +it("a periodic probe after this install's own marks stays one targeted read", async () => { + localStorage.setItem("k:started", "1"); + const { host, sidebar, write } = fakeHost(); + sidebar.mockResolvedValueOnce({ ...page(row()), read_revision: 7 }); + const api = create(host, "k"); + await api.load(); + write.mockResolvedValueOnce({ + outcomes: [{ status: "applied" }], + channels: [row({ read_through_id: id(2), latest_id: id(2) })], + read_revision: 8, + read_revision_from: 7, + }); + api.enqueue([ + { + type: "mark_through", + target: { channel_id: channel }, + message_id: id(2), + }, + ]); + await vi.advanceTimersByTimeAsync(150); + sidebar.mockResolvedValue({ ...page(row()), read_revision: 8 }); + await api.probe(); + await api.probe(); + expect(sidebar).toHaveBeenCalledTimes(3); + expect(sidebar).toHaveBeenLastCalledWith( + { channelIds: [channel] }, + expect.anything(), + ); + // Another install's mark whose ping was missed: the probe sees it. + sidebar.mockResolvedValue({ ...page(row()), read_revision: 9 }); + await api.probe(); + expect(sidebar).toHaveBeenLastCalledWith({}, expect.anything()); + api.dispose(); +}); diff --git a/src/features/relay/read-api.ts b/src/features/relay/read-api.ts new file mode 100644 index 0000000000..a43c6cca2b --- /dev/null +++ b/src/features/relay/read-api.ts @@ -0,0 +1,905 @@ +import type { RelayEvent } from "./events"; +import { threadReference } from "./thread-reference"; + +/** Server-owned read state (`/buzz/v1`, docs/buzz-v1-read-state.md in block/buzz). + * The relay decides unread and mentions; the client replaces rows with each + * response and keeps them current from live events between loads. */ +export type ServerThreadRow = Readonly<{ + root_id: string; + unread: boolean; + mentions: number; + read_through_id: string | null; + latest_id: string | null; +}>; +export type ServerChannelRow = Readonly<{ + channel_id: string; + unread: boolean; + mentions: number; + read_through_id: string | null; + latest_id: string | null; + threads: readonly ServerThreadRow[]; + /** Present when the relay cut thread rows at its cap. */ + threads_truncated?: boolean; + /** Every live follow in the channel, read or unread, newest reply first; + * a root not listed is not followed unless the list was cut at its cap. */ + followed_threads?: readonly string[]; + followed_threads_truncated?: boolean; +}>; +export type ReadIntent = + | Readonly<{ + type: "mark_through"; + target: Readonly<{ channel_id: string; root_id?: string }>; + message_id: string; + }> + | Readonly<{ + type: "follow" | "unfollow"; + target: Readonly<{ channel_id: string; root_id: string }>; + }>; +export type ReadOutcome = Readonly<{ status: string; retryable?: boolean }>; +export type SidebarPage = Readonly<{ + channels: readonly ServerChannelRow[]; + next_cursor: string | null; + /** The viewer's read-state revision in the same snapshot as the rows. */ + read_revision?: number; +}>; +/** Host boundary: the host signs NIP-98; the page never holds the key. */ +export interface ReadApiHost { + readonly capabilities: readonly string[]; + readonly eligibleKinds: readonly number[]; + readonly maxChannels: number; + readonly maxIntents: number; + sidebar( + query: Readonly<{ cursor?: string; channelIds?: readonly string[] }>, + signal: AbortSignal, + ): Promise; + /** Exact per-message answers in relay arrival order, when advertised. */ + coverage?( + messageIds: readonly string[], + signal: AbortSignal, + ): Promise< + Readonly<{ + messages: Readonly>; + read_revision?: number; + }> + >; + write( + intents: readonly ReadIntent[], + signal: AbortSignal, + ): Promise< + Readonly<{ + outcomes: readonly ReadOutcome[]; + channels?: readonly ServerChannelRow[]; + /** The ping this write sent, so its own echo can be skipped. */ + ping_id?: string; + /** Omitted when the relay could not read it in time. */ + read_revision?: number; + /** The revision just before this write's first applied intent. */ + read_revision_from?: number; + }> + >; +} + +const intentKey = (intent: ReadIntent) => + `${intent.type}:${intent.target.channel_id}:${intent.target.root_id ?? ""}:${"message_id" in intent ? intent.message_id : ""}`; +/** Repeating a mark changes nothing; a repeated follow may undo an unfollow. */ +const repeatable = (intent: ReadIntent) => intent.type !== "mark_through"; +/** How long a revision gap may stay open for its ping before a full load. */ +const GAP_MS = 1500; + +export type ReadApiStatus = Readonly<{ + status: "loading" | "ready" | "error"; + /** Intents saved here and not yet resolved by the relay. */ + pending: number; + error?: string | undefined; + /** Wall ms of the last complete sidebar load. */ + loadedAt?: number | undefined; +}>; + +export type ReadApi = ReturnType; + +/** + * One owner for the relay's sidebar rows. Loads on start, reconnect and wake; + * otherwise rows change only from live events this client already receives, + * write responses and targeted refreshes after a ping. No polling. + */ +export function createReadApi({ + host, + viewer, + storageKey, + notify = (listener) => listener(), + isDm, + rootAuthor, +}: { + host: ReadApiHost; + viewer: string; + /** localStorage key for unresolved intents; omit for memory only. */ + storageKey?: string | undefined; + notify?: (listener: () => void) => void; + isDm: (channelId: string) => boolean; + /** Author of a loaded root event, to follow replies to the viewer's roots. */ + rootAuthor: (rootId: string) => string | undefined; +}) { + let closed = false; + const lifetime = new AbortController(); + const listeners = new Set<() => void>(); + const rows = new Map(); + let complete = false; + let state: ReadApiStatus = Object.freeze({ status: "loading", pending: 0 }); + let loading: Promise | undefined; + let revision = 0; + const refreshQueue = new Set(); + let refreshTimer: ReturnType | undefined; + let staleRefreshes = 0; + const askQueue = new Set(); + let askTimer: ReturnType | undefined; + let flushTimer: ReturnType | undefined; + let flushing: Promise | undefined; + let retryMs = 1000; + let repairMs = 2000; + let repairTimer: ReturnType | undefined; + const eligible = new Set(host.eligibleKinds); + const ownPings = new Set(); + + const storage = + storageKey && typeof localStorage !== "undefined" + ? localStorage + : undefined; + let queue: ReadIntent[] = (() => { + try { + const saved = storage?.getItem(storageKey ?? ""); + const parsed = saved ? (JSON.parse(saved) as unknown) : []; + return Array.isArray(parsed) ? (parsed as ReadIntent[]) : []; + } catch { + return []; + } + })(); + const persist = () => { + try { + if (queue.length) + storage?.setItem(storageKey ?? "", JSON.stringify(queue)); + else storage?.removeItem(storageKey ?? ""); + } catch { + // A full or disabled store keeps the queue in memory for this session. + } + }; + const emit = () => { + revision++; + if (!closed) for (const listener of listeners) notify(listener); + }; + const health = (patch: Partial) => { + state = Object.freeze({ ...state, pending: queue.length, ...patch }); + emit(); + }; + // The relay's per-viewer revision moves with every applied intent from any + // install. A response older than the highest seen is dropped and asked again. + let highWater = 0; + const behind = (revision: number | undefined) => + revision !== undefined && revision < highWater; + const raise = (revision: number | undefined) => { + if (revision !== undefined && revision > highWater) highWater = revision; + }; + // Rows reflect every intent through `reconciled`. A full load sets it to + // its oldest page's revision; each ping or write answer (`from`..`to`) + // extends it when it starts there, since its channels are refetched or + // replaced. A range starting later means a missed ping: a full load repairs. + let reconciled = 0; + const ranges = new Map(); + const through = (start: number) => { + let at = start; + for (let moved = true; moved; ) { + moved = false; + for (const [from, to] of ranges) + if (from <= at && to > at) { + at = to; + moved = true; + } + } + return at; + }; + const reconcile = (at: number) => { + reconciled = through(Math.max(reconciled, at)); + for (const [from, to] of ranges) if (to <= reconciled) ranges.delete(from); + }; + function advance(from: number | undefined, to: number | undefined) { + if (to === undefined || to <= reconciled) return; + const start = from ?? to - 1; + ranges.set(start, Math.max(ranges.get(start) ?? 0, to)); + reconcile(reconciled); + if (ranges.size) awaitGap(); + } + // A range past `reconciled` is often early, not missed: pings wait for an + // own write in flight, and concurrent writes answer out of order. Give the + // missing ping a moment before repairing with a full load. + let gapTimer: ReturnType | undefined; + function awaitGap() { + if (gapTimer || closed) return; + gapTimer = setTimeout(() => { + gapTimer = undefined; + // A load in flight decides once it knows its own revision. + if (ranges.size && !loading) void load(); + }, GAP_MS); + } + // Responses can land out of order. A channel takes a response's row unless + // it already holds one read at a later revision or, at the same or unknown + // revision, from a later-issued request or local change. A local change + // newer than a response may be missing from it, so the channel is refetched. + let sequence = 0; + const written = new Map< + string, + Readonly<{ at: number; local: boolean; revision?: number }> + >(); + const issue = () => ++sequence; + // Follow state. A relay row with `followed_threads` decides for its + // channel; changes this install made since, or still has queued, win until + // the next one. Without the field, a follow is remembered from rows that + // listed the thread unread and from this install's own intents. + const follows = new Map< + string, + Readonly<{ following: boolean; at: number; revision?: number }> + >(); + const noteFollow = ( + channelId: string, + rootId: string, + following: boolean, + at: number, + revision?: number, + ) => { + const key = `${channelId}:${rootId}`; + if ((follows.get(key)?.at ?? 0) <= at) + follows.set(key, { + following, + at, + ...(revision === undefined ? {} : { revision }), + }); + }; + /** When each channel's row last came from the relay, and at what revision. */ + const fromRelay = new Map< + string, + Readonly<{ at: number; revision?: number }> + >(); + // Replies whose follow the relay was asked about: a relay row read after + // the reply that does not list the thread means it is not followed. + const probes = new Map(); + const noteListed = ( + row: ServerChannelRow, + at: number, + relay: boolean, + revision?: number, + ) => { + const followed = row.followed_threads; + if (!followed) + for (const thread of row.threads) + noteFollow(row.channel_id, thread.root_id, true, at); + if (!relay) return; + fromRelay.set(row.channel_id, { + at, + ...(revision === undefined ? {} : { revision }), + }); + for (const [key, asked] of probes) + if (key.startsWith(`${row.channel_id}:`) && asked < at) { + probes.delete(key); + const rootId = key.slice(row.channel_id.length + 1); + if ( + followed + ? !followed.includes(rootId) + : !row.threads_truncated && + !row.threads.some((thread) => thread.root_id === rootId) + ) + noteFollow(row.channel_id, rootId, false, at, revision); + } + }; + /** Whether the relay follows this thread for the viewer, when known. */ + function followOf(channelId: string, rootId: string) { + for (let index = queue.length - 1; index >= 0; index--) { + const intent = queue[index]; + if ( + intent?.type !== "mark_through" && + intent?.target.channel_id === channelId && + intent.target.root_id === rootId + ) + return intent.type === "follow"; + } + const known = follows.get(`${channelId}:${rootId}`); + const row = rows.get(channelId); + const followed = row?.followed_threads; + if (!row || !followed) return known?.following; + const relay = fromRelay.get(channelId); + if (known && known.at > (relay?.at ?? 0)) return known.following; + if ( + followed.includes(rootId) || + row.threads.some((thread) => thread.root_id === rootId) + ) + return true; + if (!row.followed_threads_truncated) return false; + // Past the cap, only a refresh at the row's revision that omitted the + // thread says it is not followed. + return known?.revision !== undefined && known.revision === relay?.revision + ? known.following + : undefined; + } + const settle = ( + issued: number, + revision: (channelId: string) => number | undefined, + next: readonly ServerChannelRow[], + absent: readonly string[] = [], + ) => { + const stale: string[] = []; + const take = (channelId: string, row?: ServerChannelRow) => { + const last = written.get(channelId); + const at = revision(channelId); + const newer = + at !== undefined && last?.revision !== undefined && at !== last.revision + ? at > last.revision + : !last || last.at <= issued; + if (!newer) { + if (last?.local) stale.push(channelId); + return; + } + written.set(channelId, { + at: issued, + local: false, + ...(at === undefined ? {} : { revision: at }), + }); + if (row) { + // Write answers omit the follow list: keep the last read's, which + // still predates this install's intents. + const previous = rows.get(channelId)?.followed_threads; + const kept: ServerChannelRow = + row.followed_threads || !previous + ? row + : { + ...row, + followed_threads: previous, + ...(rows.get(channelId)?.followed_threads_truncated + ? { followed_threads_truncated: true } + : {}), + }; + rows.set(channelId, Object.freeze(kept)); + noteListed(kept, issued, kept === row, at); + } else { + rows.delete(channelId); + fromRelay.delete(channelId); + } + }; + for (const row of next) take(row.channel_id, row); + for (const id of absent) take(id); + if (stale.length) refresh(stale); + }; + const writeLocally = (channelId: string, row: ServerChannelRow) => { + const at = issue(); + written.set(channelId, { at, local: true }); + rows.set(channelId, Object.freeze(row)); + noteListed(row, at, false); + }; + + let reload = false; + /** `again`: a load already in flight may predate the change; run another. */ + async function load(again = false): Promise { + if (closed) return; + if (loading) { + reload ||= again; + return loading; + } + const signal = AbortSignal.any([ + lifetime.signal, + AbortSignal.timeout(30000), + ]); + loading = (async () => { + try { + // Pages are separate snapshots. The traversal holds only if every + // intent after its oldest page is a ping or write answer seen while + // it ran, whose channels were refetched; otherwise read it again. + for (let attempt = 0; ; attempt++) { + const issued = issue(); + const next = new Map(); + const revisions = new Map(); + let oldest: number | undefined; + let cursor: string | undefined; + for (let page = 0; page < 500; page++) { + const result = await host.sidebar(cursor ? { cursor } : {}, signal); + for (const row of result.channels) { + next.set(row.channel_id, row); + if (result.read_revision !== undefined) + revisions.set(row.channel_id, result.read_revision); + } + if (result.read_revision !== undefined) + oldest = Math.min(oldest ?? Infinity, result.read_revision); + raise(result.read_revision); + if (!result.next_cursor) break; + cursor = result.next_cursor; + } + if (closed) return; + if (oldest !== undefined && through(oldest) < highWater) { + if (attempt === 2) + throw new Error("Read state kept changing during the load"); + continue; + } + settle( + issued, + (channelId) => revisions.get(channelId) ?? oldest, + [...next.values()], + [...rows.keys()].filter((id) => !next.has(id)), + ); + if (oldest !== undefined) reconcile(oldest); + if (ranges.size) awaitGap(); + break; + } + complete = true; + unknownChannels.clear(); + repairMs = 2000; + health({ status: "ready", error: undefined, loadedAt: Date.now() }); + start(); + void flush(); + } catch (error) { + if (closed) return; + health({ + status: "error", + error: error instanceof Error ? error.message : String(error), + }); + repair(); + } finally { + loading = undefined; + if (reload && !closed) { + reload = false; + void load(); + } + } + })(); + return loading; + } + + /** The relay counts nothing as unread until the account's first applied + * intent. Marking a read row through its latest changes no read state but + * starts the account. Once per install; another install's start is harmless. */ + function start() { + const key = storageKey && `${storageKey}:started`; + try { + if (key && storage?.getItem(key)) return; + } catch { + // Without storage, starting again each load costs one no-op write. + } + const row = [...rows.values()].find( + (candidate) => !candidate.unread && candidate.latest_id, + ); + if (!row?.latest_id) return; + enqueue([ + { + type: "mark_through", + target: { channel_id: row.channel_id }, + message_id: row.latest_id, + }, + ]); + try { + if (key) storage?.setItem(key, "1"); + } catch { + // As above. + } + } + /** A failed read leaves rows stale until a full load succeeds. */ + function repair() { + if (repairTimer || closed) return; + repairTimer = setTimeout(() => { + repairTimer = undefined; + void load(); + }, repairMs); + repairMs = Math.min(repairMs * 2, 60000); + } + // Wake: a hidden window may have slept through pings and reconnects. + let hiddenAt: number | undefined; + const onVisibility = () => { + if (document.visibilityState === "hidden") hiddenAt = Date.now(); + else if (hiddenAt !== undefined && Date.now() - hiddenAt > 30000) { + hiddenAt = undefined; + void probe(); + } + }; + /** Wake on a connected socket: a full load only if the relay's revision + * moved past ours, which means a ping was missed. Reconnects always load. */ + async function probe() { + const channelId = rows.keys().next().value; + if (!complete || !channelId) return load(); + try { + const issued = issue(); + const result = await host.sidebar( + { channelIds: [channelId] }, + lifetime.signal, + ); + if (closed) return; + raise(result.read_revision); + if ( + result.read_revision === undefined || + result.read_revision > reconciled + ) + return load(); + if (behind(result.read_revision)) return; + settle( + issued, + () => result.read_revision, + result.channels, + result.channels.length ? [] : [channelId], + ); + emit(); + } catch { + if (!closed) repair(); + } + } + if (typeof document !== "undefined") + document.addEventListener("visibilitychange", onVisibility); + // Back online: the socket may never have noticed, so pings were missed and + // queued intents are waiting out their backoff. + const onOnline = () => { + if (repairTimer) clearTimeout(repairTimer); + repairTimer = undefined; + repairMs = 2000; + retryMs = 1000; + void load(); + if (flushTimer) { + clearTimeout(flushTimer); + flushNow(); + } + }; + if (typeof window !== "undefined") + window.addEventListener("online", onOnline); + /** Targeted reload of some channels, batched to the advertised maximum. */ + function refresh(channelIds: readonly string[]) { + for (const id of channelIds) refreshQueue.add(id); + if (refreshTimer || closed) return; + refreshTimer = setTimeout(() => { + refreshTimer = undefined; + const ids = [...refreshQueue]; + refreshQueue.clear(); + void (async () => { + for (let index = 0; index < ids.length; index += host.maxChannels) { + const batch = ids.slice(index, index + host.maxChannels); + try { + const issued = issue(); + const result = await host.sidebar( + { channelIds: batch }, + lifetime.signal, + ); + if (closed) return; + if (behind(result.read_revision)) { + // Older than an answer already applied: ask again, a few times. + if (++staleRefreshes > 3) { + staleRefreshes = 0; + repair(); + return; + } + refresh(batch); + continue; + } + staleRefreshes = 0; + raise(result.read_revision); + const present = new Set( + result.channels.map((row) => row.channel_id), + ); + settle( + issued, + () => result.read_revision, + result.channels, + batch.filter((id) => !present.has(id)), + ); + emit(); + } catch { + // A ping is only a hint, but its change is real: reload everything. + if (!closed) repair(); + return; + } + } + })(); + }, 50); + } + + /** Replies the row cannot decide: ask the relay about their channels. */ + function ask(channelId: string) { + askQueue.add(channelId); + askTimer ??= setTimeout(() => { + askTimer = undefined; + const ids = [...askQueue]; + askQueue.clear(); + refresh(ids); + }, 1000); + } + + function enqueue(intents: readonly ReadIntent[]) { + if (!intents.length || closed) return; + const seen = new Set(queue.map(intentKey)); + for (const intent of intents) { + const key = intentKey(intent); + if (!repeatable(intent) && seen.has(key)) continue; + seen.add(key); + queue.push(intent); + if (intent.type !== "mark_through") + noteFollow( + intent.target.channel_id, + intent.target.root_id, + intent.type === "follow", + issue(), + ); + } + persist(); + health({}); + if (!flushTimer && !flushing) flushTimer = setTimeout(flushNow, 150); + } + function flushNow() { + flushTimer = undefined; + void flush(); + } + async function flush(): Promise { + if (closed || !queue.length) return; + if (flushing) return flushing; + flushing = (async () => { + try { + while (queue.length && !closed) { + const batch = queue.slice(0, host.maxIntents); + const issued = issue(); + const result = await host.write(batch, lifetime.signal); + if (result.ping_id) ownPings.add(result.ping_id); + raise(result.read_revision); + advance( + result.read_revision_from ?? + (result.read_revision === undefined + ? undefined + : result.read_revision - + result.outcomes.filter((o) => o.status === "applied").length), + result.read_revision, + ); + const retry: ReadIntent[] = []; + batch.forEach((intent, index) => { + const outcome = result.outcomes[index]; + if (!outcome || outcome.retryable) retry.push(intent); + }); + const done = new Set(batch.map(intentKey)); + for (const intent of retry) done.delete(intentKey(intent)); + queue = [ + ...retry, + ...queue + .slice(batch.length) + .filter((i) => repeatable(i) || !done.has(intentKey(i))), + ]; + persist(); + if (result.channels) + settle(issued, () => result.read_revision, result.channels); + else + refresh([ + ...new Set(batch.map((intent) => intent.target.channel_id)), + ]); + retryMs = 1000; + health({ error: undefined }); + if (retry.length === batch.length) + throw new Error("Read intents unresolved"); + } + } catch (error) { + if (closed) return; + health({ + error: error instanceof Error ? error.message : String(error), + }); + // A refused request will be refused again: drop it so the rest can go. + const status = (error as { status?: unknown } | null)?.status; + if (status === 400 || status === 422) { + queue = queue.slice(host.maxIntents); + persist(); + health({ pending: queue.length }); + } + flushTimer = setTimeout(flushNow, retryMs); + retryMs = Math.min(retryMs * 2, 60000); + } finally { + flushing = undefined; + } + })(); + return flushing; + } + + const mentionsViewer = (event: RelayEvent) => + event.tags.some(([name, value]) => name === "p" && value === viewer); + const broadcast = (event: RelayEvent) => + event.tags.some(([name, value]) => name === "broadcast" && value === "1"); + const channelOf = (event: RelayEvent) => { + const tags = event.tags.filter(([name]) => name === "h"); + return tags.length === 1 ? tags[0]?.[1] : undefined; + }; + + // Live delivery can repeat an event; channels the relay did not know are + // asked about once per full load. + const observed = new Set(); + const unknownChannels = new Set(); + + /** Keep rows current from a verified live arrival, as the relay would count it. */ + function observe(event: RelayEvent) { + if (closed || !eligible.has(event.kind) || observed.has(event.id)) return; + observed.add(event.id); + for (const oldest of observed) { + if (observed.size <= 2000) break; + observed.delete(oldest); + } + const channelId = channelOf(event); + const row = channelId ? rows.get(channelId) : undefined; + if (!channelId || !row) { + // A channel the snapshot does not know yet (just joined): ask for it. + if (channelId && complete && !unknownChannels.has(channelId)) { + unknownChannels.add(channelId); + refresh([channelId]); + } + return; + } + const own = event.pubkey === viewer; + const reference = threadReference(event); + const rootId = reference?.rootId; + const dm = isDm(channelId); + const counts = dm || mentionsViewer(event); + let next: ServerChannelRow = row; + if (!rootId || broadcast(event)) { + next = own + ? { + ...next, + unread: false, + mentions: 0, + read_through_id: event.id, + latest_id: event.id, + } + : { + ...next, + unread: true, + mentions: Math.min(100, next.mentions + (counts ? 1 : 0)), + latest_id: event.id, + }; + } + // A new root follows its author; the relay's next row lists it. + if (own && !rootId) noteFollow(channelId, event.id, true, issue()); + if (rootId) { + const listed = next.threads.find((thread) => thread.root_id === rootId); + const known = followOf(channelId, rootId); + const decides = row.followed_threads !== undefined; + const followed = !!listed || counts || (decides && known === true); + // Posting reads (and follows) the thread through the post. + if (own) noteFollow(channelId, rootId, true, issue()); + else if ( + !followed && + (decides + ? known === undefined + : known === true || + (known === undefined && rootAuthor(rootId) === viewer)) + ) { + // The row cannot tell, or a remembered follow may have been dropped + // on another device: the relay's next row decides. + probes.set(`${channelId}:${rootId}`, issue()); + ask(channelId); + } + const others = next.threads.filter((thread) => thread.root_id !== rootId); + if (own && listed) next = { ...next, threads: others }; + else if (!own && followed) + next = { + ...next, + threads: [ + { + root_id: rootId, + unread: true, + mentions: Math.min( + 100, + (listed?.mentions ?? 0) + (counts ? 1 : 0), + ), + read_through_id: listed?.read_through_id ?? null, + latest_id: event.id, + }, + ...others, + ], + }; + } + if (next !== row) { + writeLocally(channelId, next); + emit(); + } + } + + /** A ping's content: refresh just the channels it names. */ + function apply(content: string) { + try { + const parsed = JSON.parse(content) as { + channels?: unknown; + threads?: unknown; + read_revision?: unknown; + read_revision_from?: unknown; + }; + const ids = new Set(); + if (Array.isArray(parsed.channels)) + for (const id of parsed.channels) + if (typeof id === "string") ids.add(id); + if (Array.isArray(parsed.threads)) + for (const thread of parsed.threads) + if (typeof thread?.channel_id === "string") + ids.add(thread.channel_id); + if (ids.size) refresh([...ids]); + else void load(true); + const revision = (value: unknown) => + typeof value === "number" && Number.isSafeInteger(value) + ? value + : undefined; + advance( + revision(parsed.read_revision_from), + revision(parsed.read_revision), + ); + } catch { + void load(); + } + } + return { + host, + status: () => state, + revision: () => revision, + /** Whether a response at this read revision is older than one already seen; + * a current one raises the mark. */ + stale(readRevision: number | undefined) { + if (behind(readRevision)) return true; + raise(readRevision); + return false; + }, + /** Whether a complete sidebar load has finished; until then rows are unknown. */ + complete: () => complete, + row: (channelId: string) => rows.get(channelId), + following: followOf, + rows: () => rows, + load, + /** One targeted read; a full load only if the relay's revision moved + * past what the rows reflect. */ + probe, + refresh, + /** A relay ping (`READ_STATE_CHANGED_KIND`) for this viewer. */ + changed(content: string, id: string) { + if (ownPings.delete(id)) return; + // The relay pings before answering the write that caused it: while a + // write is in flight, wait for its answer to recognize the own echo. + const writing = flushing; + if (!writing) return apply(content); + void writing.then(() => { + if (!closed && !ownPings.delete(id)) apply(content); + }); + }, + enqueue, + flush, + observe, + subscribe(listener: () => void) { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + /** Optimistic local read until the relay's row replaces it. */ + readLocally( + channelId: string, + rootId: string | undefined, + messageId: string, + ) { + const row = rows.get(channelId); + if (!row) return; + const next: ServerChannelRow = rootId + ? { + ...row, + threads: row.threads.filter( + (thread) => + thread.root_id !== rootId || thread.latest_id !== messageId, + ), + } + : row.latest_id === messageId + ? { ...row, unread: false, mentions: 0, read_through_id: messageId } + : row; + if (next !== row) { + writeLocally(channelId, next); + emit(); + } + }, + dispose() { + closed = true; + lifetime.abort(); + if (typeof document !== "undefined") + document.removeEventListener("visibilitychange", onVisibility); + if (typeof window !== "undefined") + window.removeEventListener("online", onOnline); + if (repairTimer) clearTimeout(repairTimer); + if (refreshTimer) clearTimeout(refreshTimer); + if (askTimer) clearTimeout(askTimer); + if (gapTimer) clearTimeout(gapTimer); + if (flushTimer) clearTimeout(flushTimer); + listeners.clear(); + }, + }; +} diff --git a/src/features/relay/read-state.ts b/src/features/relay/read-state.ts index 747c14b553..db813099a3 100644 --- a/src/features/relay/read-state.ts +++ b/src/features/relay/read-state.ts @@ -118,6 +118,7 @@ export function createReadState({ now = () => Math.floor(Date.now() / 1000), debounceMs = 5000, broadcastName, + localFrontiers = false, }: { viewer: string; reader: RelayReader; @@ -128,6 +129,9 @@ export function createReadState({ now?: () => number; debounceMs?: number; broadcastName?: string | undefined; + /** Record frontiers on this device without NIP-RS: another owner + * (server read state) syncs them. */ + localFrontiers?: boolean; }) { let closed = false; const lifetime = new AbortController(); @@ -150,7 +154,9 @@ export function createReadState({ { event: RelayEvent; parsed: ParsedReadBlob } >(); const capability = !host - ? "unsupported" + ? localFrontiers + ? "frontier-sync" + : "unsupported" : !host.sign || !host.publish || !lock ? "read-only" : "frontier-sync"; @@ -298,7 +304,14 @@ export function createReadState({ } } function schedule() { - if (closed || capability !== "frontier-sync" || timer || publishing) return; + if ( + closed || + !host || + capability !== "frontier-sync" || + timer || + publishing + ) + return; timer = setTimeout(() => { timer = undefined; void flush(); diff --git a/src/features/relay/server-unread.test.ts b/src/features/relay/server-unread.test.ts new file mode 100644 index 0000000000..162a961b17 --- /dev/null +++ b/src/features/relay/server-unread.test.ts @@ -0,0 +1,361 @@ +// @vitest-environment jsdom +import { afterEach, beforeEach, expect, it, vi } from "vitest"; +import type { RelayEvent } from "./events"; +import { + createReadApi, + type ReadApiHost, + type ServerChannelRow, +} from "./read-api"; +import type { RelayReader } from "./reader"; +import { serverReadDecision, withServerReadState } from "./server-unread"; +import type { UnreadCapability, UnreadSnapshot } from "./unread"; + +const viewer = "a".repeat(64); +const channel = "00000000-0000-4000-8000-000000000001"; +const id = (n: number) => n.toString(16).padStart(64, "0"); +const root = id(10); +const message = (n: number, tags: string[][] = []): RelayEvent => + ({ + id: id(n), + pubkey: "b".repeat(64), + kind: 9, + created_at: 1_700_000_000 + n, + content: "", + sig: "", + tags: [["h", channel], ...tags], + }) as RelayEvent; +const reply = (n: number) => message(n, [["e", root, "", "reply"]]); + +function setup( + channelRow: ServerChannelRow, + coverage?: ReadApiHost["coverage"], +) { + const write = vi.fn(async (intents) => ({ + outcomes: intents.map(() => ({ status: "applied" })), + })); + const host: ReadApiHost = { + capabilities: ["follow", "threads", "ping"], + eligibleKinds: [9], + maxChannels: 20, + maxIntents: 100, + sidebar: async () => ({ channels: [channelRow], next_cursor: null }), + write, + ...(coverage ? { coverage } : {}), + }; + localStorage.setItem("k:started", "1"); + const api = createReadApi({ + host, + viewer, + storageKey: "k", + isDm: () => false, + rootAuthor: () => undefined, + }); + const events = new Map(); + const base = (target: UnreadSnapshot["target"]): UnreadSnapshot => ({ + target, + observedCount: 0, + attentionCount: 0, + coverage: "observed", + freshness: "observed", + manual: "none", + }); + const attention = new Map(); + const local = { + snapshot: base, + attention: (_channelId: string, messageId: string) => + attention.get(messageId) ?? { unread: false, mentioned: false }, + reading: () => ({ + view: () => undefined, + observe: async () => undefined, + catchUp: async () => undefined, + dispose: () => undefined, + }), + following: () => false, + follow: () => undefined, + markMessageRead: async () => ({}), + } as unknown as UnreadCapability; + const unread = withServerReadState({ + local, + api, + event: (eventId) => events.get(eventId), + isDm: () => false, + canFollow: true, + }); + const decision = serverReadDecision({ + api, + reader: { read: async () => [] } as unknown as RelayReader, + event: (eventId) => events.get(eventId), + }); + return { api, unread, decision, events, attention, write }; +} + +beforeEach(() => { + vi.useFakeTimers(); + localStorage.clear(); +}); +afterEach(() => { + vi.useRealTimers(); +}); + +it("takes channel and thread counts from the relay's rows", async () => { + const { api, unread } = setup({ + channel_id: channel, + unread: true, + mentions: 2, + read_through_id: id(1), + latest_id: id(3), + threads: [ + { + root_id: root, + unread: true, + mentions: 0, + read_through_id: null, + latest_id: id(11), + }, + ], + }); + await api.load(); + expect( + unread.snapshot({ kind: "channel", channelId: channel }), + ).toMatchObject({ observedCount: 3, attentionCount: 3 }); + expect( + unread.snapshot({ kind: "thread", channelId: channel, rootId: root }), + ).toMatchObject({ observedCount: 1, attentionCount: 1 }); + expect( + unread.snapshot({ kind: "thread", channelId: channel, rootId: id(20) }), + ).toMatchObject({ observedCount: 0 }); + api.dispose(); +}); + +it("answers per message from coverage and asks again after a stale answer", async () => { + const coverage = vi.fn>(); + const { api, decision } = setup( + { + channel_id: channel, + unread: true, + mentions: 0, + read_through_id: id(1), + latest_id: id(4), + threads: [], + }, + coverage, + ); + await api.load(); + expect(api.stale(9)).toBe(false); + expect(decision.unread(message(4), channel, undefined)).toBe(true); + expect(decision.unread(message(1), channel, undefined)).toBe(false); + coverage + .mockResolvedValueOnce({ messages: { [id(2)]: "read" }, read_revision: 8 }) + .mockResolvedValueOnce({ + messages: { [id(2)]: "unread" }, + read_revision: 9, + }); + expect(decision.unread(message(2), channel, undefined)).toBeUndefined(); + await vi.advanceTimersByTimeAsync(20); + expect(decision.unread(message(2), channel, undefined)).toBeUndefined(); + await vi.advanceTimersByTimeAsync(1); + expect(coverage).toHaveBeenCalledTimes(2); + expect(decision.unread(message(2), channel, undefined)).toBe(true); + // A reply in a thread the relay does not list is read. + expect(decision.unread(reply(12), channel, root)).toBe(false); + api.dispose(); +}); + +it("sends mark_through for viewed mentions and unread catch-up only", async () => { + const { api, unread, events, attention, write } = setup({ + channel_id: channel, + unread: true, + mentions: 1, + read_through_id: id(1), + latest_id: id(3), + threads: [], + }); + await api.load(); + for (const event of [message(2), message(3, [["p", viewer]])]) + events.set(event.id, event); + attention.set(id(2), { unread: true, mentioned: false }); + attention.set(id(3), { unread: true, mentioned: true }); + const reading = unread.reading(channel); + await reading.observe([id(2), id(3)]); + await vi.advanceTimersByTimeAsync(150); + expect(write.mock.calls.at(-1)?.[0]).toEqual([ + { + type: "mark_through", + target: { channel_id: channel }, + message_id: id(3), + }, + ]); + expect(api.row(channel)).toMatchObject({ unread: false, mentions: 0 }); + // Already read: catching up again writes nothing. + const writes = write.mock.calls.length; + await reading.catchUp(id(3)); + await vi.advanceTimersByTimeAsync(150); + expect(write.mock.calls.length).toBe(writes); + api.dispose(); +}); + +it("sends follow and unfollow intents", async () => { + const { api, unread, write } = setup({ + channel_id: channel, + unread: false, + mentions: 0, + read_through_id: id(1), + latest_id: id(1), + threads: [], + }); + await api.load(); + unread.follow(channel, root, false); + await vi.advanceTimersByTimeAsync(150); + expect(write.mock.calls.at(-1)?.[0]).toEqual([ + { type: "unfollow", target: { channel_id: channel, root_id: root } }, + ]); + api.dispose(); +}); + +it("a thread stays followed after it is read, until unfollowed", async () => { + const { api, unread } = setup({ + channel_id: channel, + unread: false, + mentions: 0, + read_through_id: id(1), + latest_id: id(1), + threads: [ + { + root_id: root, + unread: true, + mentions: 1, + read_through_id: null, + latest_id: id(11), + }, + ], + }); + await api.load(); + expect(unread.following(channel, root)).toBe(true); + api.readLocally(channel, root, id(11)); + expect(api.row(channel)?.threads).toEqual([]); + expect(unread.following(channel, root)).toBe(true); + unread.follow(channel, root, false); + expect(unread.following(channel, root)).toBe(false); + api.dispose(); +}); + +it("asks coverage once for a broadcast reply that sits in two scopes", async () => { + const coverage = vi.fn>(async (ids) => ({ + messages: Object.fromEntries(ids.map((one) => [one, "unread"])), + })); + const { api, decision } = setup( + { + channel_id: channel, + unread: true, + mentions: 0, + read_through_id: id(1), + latest_id: id(4), + threads: [ + { + root_id: root, + unread: true, + mentions: 0, + read_through_id: id(11), + latest_id: id(14), + }, + ], + }, + coverage, + ); + await api.load(); + const both = message(12, [ + ["e", root, "", "reply"], + ["broadcast", "1"], + ]); + expect(decision.unread(both, channel, root)).toBeUndefined(); + await vi.advanceTimersByTimeAsync(20); + expect(decision.unread(both, channel, root)).toBe(true); + await vi.advanceTimersByTimeAsync(100); + expect(coverage).toHaveBeenCalledTimes(1); + expect(coverage.mock.calls[0]?.[0]).toEqual([id(12)]); + api.dispose(); +}); + +it("sends no mark for a message whose scope it cannot tell", async () => { + const { api, unread, write } = setup({ + channel_id: channel, + unread: true, + mentions: 0, + read_through_id: id(1), + latest_id: id(3), + threads: [], + }); + await api.load(); + await unread.markMessageRead(channel, id(3)); + await vi.advanceTimersByTimeAsync(150); + expect(write).not.toHaveBeenCalled(); + api.dispose(); +}); + +it("asks no coverage again when a later message arrives", async () => { + const coverage = vi.fn>(async (ids) => ({ + messages: Object.fromEntries(ids.map((one) => [one, "unread"])), + })); + const { api, decision } = setup( + { + channel_id: channel, + unread: true, + mentions: 0, + read_through_id: id(1), + latest_id: id(4), + threads: [], + }, + coverage, + ); + await api.load(); + expect(decision.unread(message(2), channel, undefined)).toBeUndefined(); + expect(decision.unread(message(4), channel, undefined)).toBe(true); + await vi.advanceTimersByTimeAsync(20); + expect(coverage).toHaveBeenCalledTimes(1); + api.observe(message(5)); + expect(decision.unread(message(2), channel, undefined)).toBe(true); + // The replaced latest stays unread under the same anchor. + expect(decision.unread(message(4), channel, undefined)).toBe(true); + // Shown before its row caught up: the arrival is the latest by the time + // the batch goes out. + expect(decision.unread(message(6), channel, undefined)).toBeUndefined(); + api.observe(message(6)); + await vi.advanceTimersByTimeAsync(100); + expect(decision.unread(message(6), channel, undefined)).toBe(true); + expect(coverage).toHaveBeenCalledTimes(1); + api.dispose(); +}); + +it("keeps a message read once its scope's anchor moves on", async () => { + const coverage = vi.fn>(async (ids) => ({ + messages: Object.fromEntries(ids.map((one) => [one, "read"])), + })); + const { api, decision } = setup( + { + channel_id: channel, + unread: true, + mentions: 0, + read_through_id: id(3), + latest_id: id(4), + threads: [], + }, + coverage, + ); + await api.load(); + expect(decision.unread(message(2), channel, undefined)).toBeUndefined(); + expect(decision.unread(message(3), channel, undefined)).toBe(false); + await vi.advanceTimersByTimeAsync(20); + expect(coverage).toHaveBeenCalledTimes(1); + // Read through an own post, then another arrival: the anchor moved. + api.observe({ ...message(5), pubkey: viewer }); + api.observe(message(6)); + expect(api.row(channel)).toMatchObject({ + unread: true, + read_through_id: id(5), + }); + for (const n of [2, 3]) + expect(decision.unread(message(n), channel, undefined)).toBe(false); + await vi.advanceTimersByTimeAsync(100); + expect(coverage).toHaveBeenCalledTimes(1); + api.dispose(); +}); diff --git a/src/features/relay/server-unread.ts b/src/features/relay/server-unread.ts new file mode 100644 index 0000000000..2d0940daf3 --- /dev/null +++ b/src/features/relay/server-unread.ts @@ -0,0 +1,536 @@ +import type { RelayEvent } from "./events"; +import type { ReadApi, ReadIntent, ServerChannelRow } from "./read-api"; +import type { ReadTarget } from "./read-state-model"; +import type { ReadMutationResult, ReadSyncSnapshot } from "./read-state"; +import type { RelayReader } from "./reader"; +import { threadReference } from "./thread-reference"; +import type { + RemoteReadDecision, + UnreadCapability, + UnreadSnapshot, +} from "./unread"; + +const broadcast = (event: RelayEvent) => + event.tags.some(([name, value]) => name === "broadcast" && value === "1"); + +/** + * Per-message answers from the relay's rows. A read row reads every message + * in its scope; in an unread row, the latest is unread and the rest are asked + * of `/me/coverage`, which orders by arrival. Without coverage, comparing + * author time with the anchor's is an approximation: a backdated message + * after the anchor reads as read. + */ +export function serverReadDecision({ + api, + reader, + event, +}: { + api: ReadApi; + reader: RelayReader; + /** A retained verified event, from unread evidence. */ + event: (id: string) => RelayEvent | undefined; +}): RemoteReadDecision { + const anchors = new Map(); + const wanted = new Set(); + const failed = new Set(); + const listeners = new Set<() => void>(); + let timer: ReturnType | undefined; + const fetchAnchors = () => { + timer = undefined; + const ids = [...wanted].slice(0, 100); + for (const id of ids) wanted.delete(id); + if (!ids.length) return; + void reader + .read([{ ids, limit: ids.length }], { priority: "background" }) + .then((events) => { + for (const found of events) anchors.set(found.id, found.created_at); + for (const id of ids) if (!anchors.has(id)) failed.add(id); + for (const listener of listeners) listener(); + }) + .catch(() => { + for (const id of ids) failed.add(id); + }) + .finally(() => { + if (wanted.size && !timer) timer = setTimeout(fetchAnchors, 0); + }); + }; + const anchorTime = (id: string) => { + const known = anchors.get(id) ?? event(id)?.created_at; + if (known !== undefined || failed.has(id)) return known; + wanted.add(id); + timer ??= setTimeout(fetchAnchors, 0); + return undefined; + }; + // Exact answers, valid while the scope's anchor stays the same: a later + // arrival moves only the latest, which reads no earlier message and unreads + // no read one. A broadcast reply sits in two scopes, so answers key by + // message and scope. + const covered = new Map(); + // A scope's anchor only moves forward, so a message read there once stays + // read under every later anchor. + const settled = new Set(); + const settle = (key: string) => { + settled.add(key); + for (const oldest of settled) { + if (settled.size <= 5000) break; + settled.delete(oldest); + } + }; + const asking = new Map< + string, + { id: string; channelId: string; rootId: string | undefined } + >(); + const cover = (key: string, answer: boolean | undefined) => { + covered.delete(key); + covered.set(key, answer); + for (const oldest of covered.keys()) { + if (covered.size <= 5000) break; + covered.delete(oldest); + } + }; + let coverTimer: ReturnType | undefined; + let staleAnswers = 0; + const latestOf = (channelId: string, rootId: string | undefined) => { + const row = api.row(channelId); + const scoped = rootId + ? row?.threads.find((thread) => thread.root_id === rootId) + : row?.unread + ? row + : undefined; + return ( + scoped && { anchor: scoped.read_through_id, latest: scoped.latest_id } + ); + }; + const askCoverage = () => { + coverTimer = undefined; + const batch = [...asking].slice(0, 100); + for (const [key] of batch) asking.delete(key); + // A live message asked about before its row caught up may now be the + // scope's latest, which is unread without asking. + const ask = batch.filter(([key, { id, channelId, rootId }]) => { + const now = latestOf(channelId, rootId); + if ( + !now || + now.latest !== id || + key !== `${id}|${rootId ?? ""}|${now.anchor}` + ) + return true; + cover(key, true); + return false; + }); + if (ask.length < batch.length) for (const listener of listeners) listener(); + if (!ask.length || !api.host.coverage) return; + void api.host + .coverage( + [...new Set(ask.map(([, { id }]) => id))], + AbortSignal.timeout(10000), + ) + .then(({ messages, read_revision }) => { + // Older than a response already applied: ask again, a few times. + if (api.stale(read_revision) && ++staleAnswers <= 3) { + for (const [key, value] of ask) asking.set(key, value); + return; + } + staleAnswers = 0; + for (const [key, { id, rootId }] of ask) { + const answer = messages[id]; + if (answer === "read") settle(`${id}|${rootId ?? ""}`); + cover( + key, + answer === "read" || answer === "unread" + ? answer === "unread" + : undefined, + ); + } + for (const listener of listeners) listener(); + }) + .catch(() => { + // Unanswered messages keep the local decision until the next change. + for (const [key] of ask) cover(key, undefined); + }) + .finally(() => { + if (asking.size && !coverTimer) coverTimer = setTimeout(askCoverage, 0); + }); + }; + const scope = ( + message: RelayEvent, + unread: boolean, + anchor: string | null, + latest: string | null, + channelId: string, + rootId: string | undefined, + ) => { + if (!unread) return false; + const scoped = `${message.id}|${rootId ?? ""}`; + if (message.id === anchor) { + settle(scoped); + return false; + } + if (settled.has(scoped)) return false; + const key = `${scoped}|${anchor}`; + if (message.id === latest) { + // Still unread once a later arrival replaces it. + if (api.host.coverage) cover(key, true); + return true; + } + if (api.host.coverage) { + if (covered.has(key)) return covered.get(key); + if (!asking.has(key)) { + asking.set(key, { id: message.id, channelId, rootId }); + coverTimer ??= setTimeout(askCoverage, 20); + } + return undefined; + } + if (!anchor) return undefined; + const time = anchorTime(anchor); + return time === undefined ? undefined : message.created_at > time; + }; + return { + unread(message, channelId, threadRootId) { + if (!api.complete()) return undefined; + const row = api.row(channelId); + if (!row) return undefined; + const timeline = + !threadRootId || broadcast(message) + ? scope( + message, + row.unread, + row.read_through_id, + row.latest_id, + channelId, + undefined, + ) + : false; + if (!threadRootId) return timeline; + const thread = row.threads.find( + (candidate) => candidate.root_id === threadRootId, + ); + const replies = thread + ? scope( + message, + true, + thread.read_through_id, + thread.latest_id, + channelId, + threadRootId, + ) + : row.threads_truncated + ? undefined + : false; + if (timeline === true || replies === true) return true; + return timeline === undefined || replies === undefined + ? undefined + : false; + }, + following: (channelId, rootId) => + api.complete() ? api.following(channelId, rootId) : undefined, + subscribe(listener) { + const stop = api.subscribe(listener); + listeners.add(listener); + return () => { + stop(); + listeners.delete(listener); + }; + }, + }; +} + +const sameSnapshot = (a: UnreadSnapshot, b: UnreadSnapshot) => + a.observedCount === b.observedCount && + a.attentionCount === b.attentionCount && + a.coverage === b.coverage && + a.freshness === b.freshness && + a.manual === b.manual && + a.error === b.error && + a.latestMessage?.id === b.latestMessage?.id; + +/** + * The unread capability with sidebar and thread rows from the relay, and + * reading intent sent to it. Consumers keep the same contract. Per-message + * state, Inbox and thread activity stay local evidence decided by the rows + * (`serverReadDecision`). Manual unread stays on this device. + */ +export function withServerReadState({ + local, + api, + event, + isDm, + canFollow, +}: { + local: UnreadCapability; + api: ReadApi; + event: (id: string) => RelayEvent | undefined; + isDm: (channelId: string) => boolean; + canFollow: boolean; +}): UnreadCapability { + const cache = new Map(); + let syncSnapshot: ReadSyncSnapshot | undefined; + + const threadAttention = (row: ServerChannelRow) => + row.threads.reduce( + (sum, thread) => sum + (thread.unread ? Math.max(1, thread.mentions) : 0), + 0, + ); + function snapshot(target: ReadTarget): UnreadSnapshot { + const base = local.snapshot(target); + const row = api.complete() ? api.row(target.channelId) : undefined; + if (target.kind === "message" || !row) return base; + let observed: number; + let attention: number; + if (target.kind === "thread") { + const thread = row.threads.find((t) => t.root_id === target.rootId); + if (!thread && row.threads_truncated) return base; + observed = attention = thread?.unread ? Math.max(1, thread.mentions) : 0; + } else { + attention = row.mentions + threadAttention(row); + // The relay says whether the timeline is unread, not whether it is all + // mentions; local evidence (decided by the same rows) tells them apart. + const localOrdinary = Math.max( + 0, + (base.observedCount ?? 0) - (base.attentionCount ?? 0), + ); + const ordinary = row.unread + ? isDm(target.channelId) + ? 0 + : Math.max(localOrdinary, row.mentions ? 0 : 1) + : 0; + observed = ordinary + attention; + } + if (base.manual !== "none") observed = Math.max(observed, 1); + const status = api.status(); + const next: UnreadSnapshot = Object.freeze({ + target, + ...(base.latestMessage ? { latestMessage: base.latestMessage } : {}), + observedCount: observed, + attentionCount: attention, + coverage: "observed", + freshness: status.status === "error" ? "stale" : "observed", + manual: base.manual, + error: status.error, + }); + const key = `${target.kind}:${target.channelId}:${target.kind === "thread" ? target.rootId : ""}`; + const old = cache.get(key); + if (old && sameSnapshot(old, next)) return old; + cache.set(key, next); + return next; + } + + const markThrough = ( + target: ReadIntent["target"], + messageId: string, + ): ReadIntent => ({ type: "mark_through", target, message_id: messageId }); + /** Mark a message's scope on the relay through it. Without the retained + * event its scope is unknown, so nothing is sent. */ + const markMessage = (channelId: string, messageId: string): ReadIntent[] => { + const found = event(messageId); + if (!found) return []; + const rootId = threadReference(found)?.rootId; + return [ + markThrough( + rootId && !broadcast(found) + ? { channel_id: channelId, root_id: rootId } + : { channel_id: channelId }, + messageId, + ), + ]; + }; + /** Explicit "read everything here": each listed scope through its latest. */ + function channelIntents(channelId: string): ReadIntent[] { + const row = api.row(channelId); + if (!row) { + const latest = local.snapshot({ kind: "channel", channelId }) + .latestMessage?.id; + return latest ? markMessage(channelId, latest) : []; + } + return [ + ...(row.latest_id && row.unread + ? [markThrough({ channel_id: channelId }, row.latest_id)] + : []), + ...row.threads.flatMap((thread) => + thread.latest_id + ? [ + markThrough( + { channel_id: channelId, root_id: thread.root_id }, + thread.latest_id, + ), + ] + : [], + ), + ]; + } + function send(intents: readonly ReadIntent[]) { + for (const intent of intents) + if (intent.type === "mark_through") + api.readLocally( + intent.target.channel_id, + intent.target.root_id, + intent.message_id, + ); + api.enqueue(intents); + } + const withIntents = async ( + result: Promise, + intents: readonly ReadIntent[], + ) => { + send(intents); + return result; + }; + + return Object.freeze({ + ...local, + snapshot, + subscribe(target, listener) { + const stopLocal = local.subscribe(target, listener); + const stopApi = api.subscribe(listener); + return () => { + stopLocal(); + stopApi(); + }; + }, + sync() { + const status = api.status(); + const next: ReadSyncSnapshot = Object.freeze({ + capability: "frontier-sync", + status: + status.status === "loading" + ? "loading" + : status.error + ? "error" + : status.pending + ? "pending" + : "reconciled", + completeness: api.complete() ? "snapshot" : "unknown", + ...(status.error ? { error: status.error } : {}), + }); + if ( + syncSnapshot && + syncSnapshot.status === next.status && + syncSnapshot.completeness === next.completeness && + syncSnapshot.error === next.error + ) + return syncSnapshot; + syncSnapshot = next; + return next; + }, + subscribeSync(listener) { + const stopLocal = local.subscribeSync(listener); + const stopApi = api.subscribe(listener); + return () => { + stopLocal(); + stopApi(); + }; + }, + async ensure() { + await Promise.all([ + local.ensure(), + api.complete() ? undefined : api.load(), + ]); + }, + async refresh() { + await Promise.all([local.refresh(), api.load()]); + }, + async retrySync() { + await Promise.all([local.retrySync(), api.flush()]); + }, + reading(channelId) { + const inner = local.reading(channelId); + const dm = () => isDm(channelId); + return Object.freeze({ + view: inner.view, + async observe(messageIds) { + // Mentions, broadcasts and DMs read on every device: each viewed + // one moves its scope through it. Ordinary rows stay on this device. + const intents = messageIds.flatMap((id) => { + const attention = local.attention(channelId, id); + if (!attention.unread) return []; + const found = event(id); + if (!dm() && !attention.mentioned && !(found && broadcast(found))) + return []; + return markMessage(channelId, id); + }); + await inner.observe(messageIds); + send(intents); + }, + async catchUp(messageId, rootId) { + await inner.catchUp(messageId, rootId); + // Catch-up repeats at every dwell; only an unread scope needs a write. + const row = api.row(channelId); + if ( + row && + (rootId + ? !row.threads_truncated && + !row.threads.some((thread) => thread.root_id === rootId) + : !row.unread) + ) + return; + send( + rootId + ? [ + markThrough( + { channel_id: channelId, root_id: rootId }, + messageId, + ), + ] + : markMessage(channelId, messageId), + ); + }, + dispose: inner.dispose, + }); + }, + following(channelId, rootId) { + return ( + api.following(channelId, rootId) ?? local.following(channelId, rootId) + ); + }, + follow(channelId, rootId, following) { + local.follow(channelId, rootId, following); + if (canFollow) + api.enqueue([ + { + type: following ? "follow" : "unfollow", + target: { channel_id: channelId, root_id: rootId }, + }, + ]); + }, + markThrough(target, messageId) { + return withIntents( + local.markThrough(target, messageId), + target.kind === "thread" + ? [ + markThrough( + { channel_id: target.channelId, root_id: target.rootId }, + messageId, + ), + ] + : target.kind === "channel" + ? [markThrough({ channel_id: target.channelId }, messageId)] + : markMessage(target.channelId, messageId), + ); + }, + markMessageRead(channelId, messageId) { + return withIntents( + local.markMessageRead(channelId, messageId), + markMessage(channelId, messageId), + ); + }, + markChannelRead(channelId) { + return withIntents( + local.markChannelRead(channelId), + channelIntents(channelId), + ); + }, + prepareChannelRead(channelId) { + const inner = local.prepareChannelRead(channelId); + const intents = channelIntents(channelId); + return () => withIntents(inner(), intents); + }, + async markAllChannelsRead() { + const intents = [...api.rows().values()] + .filter((row) => row.unread || row.threads.length) + .flatMap((row) => channelIntents(row.channel_id)); + const results = await local.markAllChannelsRead(); + send(intents); + return results; + }, + }); +} diff --git a/src/features/relay/session.ts b/src/features/relay/session.ts index 167b1c9ea8..e791b3d2c0 100644 --- a/src/features/relay/session.ts +++ b/src/features/relay/session.ts @@ -56,6 +56,8 @@ import { createTyping } from "./typing"; import { validTypingTemplate } from "./typing-protocol"; import { workflowOwner } from "./workflow-attribution"; import { createUnread } from "./unread"; +import { createReadApi } from "./read-api"; +import { serverReadDecision, withServerReadState } from "./server-unread"; import type { SnapshotRead } from "../agents/agent-config-artifacts"; import { createInboxFeed } from "./inbox-feed"; @@ -84,6 +86,7 @@ import { import { PRODUCT_FEEDBACK_KIND } from "./product-feedback"; import type { ReadTransport } from "./transport"; import type { LiveSnapshot, LiveSubscription } from "./live"; +import { READ_STATE_CHANGED_KIND } from "./live"; import { hasTag, type ReadFilter, @@ -903,10 +906,27 @@ export function createRelaySession( const readScope = transport ? transportPartition(transport) : relayPartition("offline", ""); + // Server-owned read state replaces NIP-RS entirely when the relay offers it. + const readApiHost = transport?.readApi; + const isDmChannel = (channelId: string) => + channels.queries.list().channels.find((channel) => channel.id === channelId) + ?.channelType === "dm"; + let readEvidence: ((id: string) => RelayEvent | undefined) | undefined; + const readApi = readApiHost + ? createReadApi({ + host: readApiHost, + viewer: transport?.viewer ?? "", + storageKey: `buzz.read-v1.pending:${readScope}:${transport?.viewer ?? ""}`, + notify, + isDm: isDmChannel, + rootAuthor: (rootId) => readEvidence?.(rootId)?.pubkey, + }) + : undefined; const reads = createReadState({ viewer: transport?.viewer ?? "", reader: requests.reader, - host: transport?.readState, + host: readApi ? undefined : transport?.readState, + localFrontiers: !!readApi, storage: options.readStateStorage ?? browserReadStateStorage(readScope, transport?.viewer ?? ""), @@ -931,7 +951,24 @@ export function createRelaySession( : {}), notify, follows: browserThreadFollows(readScope), + remote: + readApi && + serverReadDecision({ + api: readApi, + reader: requests.reader, + event: (id) => readEvidence?.(id), + }), }); + readEvidence = unread.event; + const unreadCapability = readApi + ? withServerReadState({ + local: unread.capability, + api: readApi, + event: unread.event, + isDm: isDmChannel, + canFollow: !!readApiHost?.capabilities.includes("follow"), + }) + : unread.capability; const inboxFeed = createInboxFeed({ // A withheld auxiliary event is not proof of an exhausted history page. // Fail closed at raw/admitted admission; do not relax reference visibility. @@ -1958,7 +1995,7 @@ export function createRelaySession( }, searchProfiles, ), - unread: unread.capability, + unread: unreadCapability, sidebarPreferences: sidebarPreferences.queries, soloPreferences: soloPreferences.queries, reminders: reminders?.capability, @@ -2820,9 +2857,21 @@ export function createRelaySession( ), ), ); + if (readApi) + for (const event of events) + if ( + event.kind === READ_STATE_CHANGED_KIND && + event.pubkey === transport.relayAuthor && + event.tags.some( + ([name, value]) => name === "p" && value === transport.viewer, + ) + ) + readApi.changed(event.content, event.id); const epoch = accessEpoch; const generation = liveGeneration; const visible = accept(events); + if (readApi && provenance?.phase === "live") + for (const event of visible) readApi.observe(event); // Completion subscribers can synchronously clear, revoke or retire this // live delivery. Do not admit its remaining pulses into the new lifetime. if ( @@ -2989,6 +3038,7 @@ export function createRelaySession( emoji.reconnect(); statuses.reconnect(); unread.reconnect(); + void readApi?.load(true); inboxFeed.reconnect(); // Only while this same connection is still up. if (generation === liveGeneration && memoryConnected) @@ -3023,6 +3073,8 @@ export function createRelaySession( if (closed) return; refreshRoster(); unread.reconnect(); + // The socket stayed up: a missed ping shows as a moved revision. + void readApi?.probe(); }, denied(channelId, reason) { if (!closed) channels.denyChannel(channelId, new Error(reason)); @@ -3147,6 +3199,7 @@ export function createRelaySession( for (const timer of timers) clearTimeout(timer); for (const dispose of [...views.keys()]) dispose(); unread.dispose(); + readApi?.dispose(); inboxFeed.dispose(); writes?.dispose(); requests.dispose(); diff --git a/src/features/relay/transport.ts b/src/features/relay/transport.ts index dea971ce04..4122a52285 100644 --- a/src/features/relay/transport.ts +++ b/src/features/relay/transport.ts @@ -17,6 +17,7 @@ import { } from "../repos/git"; import type { WorkflowHost } from "../workflows/host"; import { readReceiptText } from "./receipt"; +import type { ReadApiHost } from "./read-api"; import { ReadStateTimestampRejected, readStateRefusal, @@ -122,6 +123,8 @@ export interface ReadTransport { readonly writeSidebarSort?: SidebarSortMutator; readonly writeSidebarFilter?: SidebarFilterMutator; readonly readState?: ReadStateHost; + /** Server-owned read state, when the relay advertises `buzz_v1`. */ + readonly readApi?: ReadApiHost; /** Purpose-bound NIP-ER codec; never a general NIP-44 or signing primitive. */ readonly reminders?: ReminderHost; readonly channelKit?: ChannelKitHost; @@ -472,6 +475,12 @@ export async function connectBrokerTransport( agentActivity?: boolean; readState?: boolean; readStateCommunity?: string; + readStateV1?: { + capabilities: string[]; + eligibleKinds: number[]; + maxChannels: number; + maxIntents: number; + }; }; if ( typeof session.viewer !== "string" || @@ -828,6 +837,68 @@ export async function connectBrokerTransport( }, } : {}), + ...(session.readStateV1 + ? { + readApi: { + capabilities: session.readStateV1.capabilities, + eligibleKinds: session.readStateV1.eligibleKinds, + maxChannels: session.readStateV1.maxChannels, + maxIntents: session.readStateV1.maxIntents, + async sidebar(query, signal) { + const params = new URLSearchParams(); + if (query.cursor) params.set("cursor", query.cursor); + if (query.channelIds) + params.set("channel_ids", query.channelIds.join(",")); + const response = await fetch( + `${endpoint}/read-v1-sidebar${params.size ? `?${params}` : ""}`, + { credentials: "same-origin", signal }, + ); + if (!response.ok) + throw new Error(`Sidebar read failed (${response.status})`); + return response.json(); + }, + async write(intents, signal) { + const response = await fetch(`${endpoint}/read-v1-write`, { + method: "POST", + credentials: "same-origin", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ intents }), + signal, + }); + if (!response.ok) + throw Object.assign( + new Error(`Read write failed (${response.status})`), + { status: response.status }, + ); + return response.json(); + }, + ...(session.readStateV1.capabilities.includes("coverage") + ? { + async coverage( + messageIds: readonly string[], + signal: AbortSignal, + ) { + const response = await fetch( + `${endpoint}/read-v1-coverage`, + { + method: "POST", + credentials: "same-origin", + headers: { "Content-Type": "application/json" }, + body: JSON.stringify({ message_ids: messageIds }), + signal, + }, + ); + if (!response.ok) + throw new Error( + `Read coverage failed (${response.status})`, + ); + return response.json(); + }, + } + : {}), + } satisfies ReadApiHost, + } + : {}), ...(session.readState ? { readState: { diff --git a/src/features/relay/unread.test.ts b/src/features/relay/unread.test.ts index 0758a9ac1b..30819636c3 100644 --- a/src/features/relay/unread.test.ts +++ b/src/features/relay/unread.test.ts @@ -9,6 +9,7 @@ import { } from "./read-state-storage"; import { eventDto, type RelayEvent } from "./events"; import type { ThreadActivitySnapshot } from "./unread"; +import type { ReadApiHost, ServerChannelRow } from "./read-api"; import type { ChannelStoreOptions } from "./store"; import type { SavedHead } from "./persistence"; import type { ReadStateSigning } from "./read-state-host"; @@ -42,6 +43,7 @@ function setup( signer = true, preloaded?: (journal: ReadJournal) => ReadJournal, workflowAuthority = false, + readApi?: ReadApiHost, ) { const viewer = keypair(), relay = keypair(), @@ -105,6 +107,7 @@ function setup( query, media: () => undefined, readState: signer ? host : { decode: host.decode }, + ...(readApi ? { readApi } : {}), subscribe(callbacks) { incoming = callbacks.receive; return { update() {}, retry() {}, dispose() {} }; @@ -4009,3 +4012,73 @@ it("catch-up and explicit unread intent still apply to agent-audience rows", asy await unread.markMessageRead("room", progress.id); expect(h.snapshot()).toMatchObject({ observedCount: 0 }); }); + +it("with server read state, the relay's follow list decides Inbox thread replies over a local choice", async () => { + const stored = new Map(); + vi.stubGlobal("localStorage", { + getItem: (key: string) => stored.get(key) ?? null, + setItem: (key: string, value: string) => stored.set(key, value), + removeItem: (key: string) => stored.delete(key), + }); + const rooms: ServerChannelRow[] = []; + const write = vi.fn(async (intents) => ({ + outcomes: intents.map(() => ({ status: "applied" })), + })); + const h = setup({}, true, undefined, false, { + capabilities: ["follow", "threads", "ping"], + eligibleKinds: [9], + maxChannels: 20, + maxIntents: 100, + sidebar: async () => ({ channels: rooms, next_cursor: null }), + write, + }); + h.grant("room"); + const root = message(h.viewer, "room", "My thread", 20); + const first = message(h.alice, "room", "First reply", 21, [ + ["e", root.id, "", "reply"], + ]); + h.emit([root, first]); + const row = ( + followed: string[], + unreadThread?: string, + ): ServerChannelRow => ({ + channel_id: "room", + unread: false, + mentions: 0, + read_through_id: root.id, + latest_id: root.id, + threads: unreadThread + ? [ + { + root_id: root.id, + unread: true, + mentions: 0, + read_through_id: first.id, + latest_id: unreadThread, + }, + ] + : [], + followed_threads: followed, + }); + rooms.splice(0, rooms.length, row([root.id])); + const unread = h.session.unread; + await unread.refresh(); + // Unfollowed here, then refollowed on another install. + unread.follow("room", root.id, false); + await vi.waitFor(() => expect(write).toHaveBeenCalled()); + const thread = () => + unread.inbox().items.find((item) => item.rootId === root.id); + await vi.waitFor(() => expect(unread.following("room", root.id)).toBe(false)); + const next = message(h.alice, "room", "Second reply", 22, [ + ["e", root.id, "", "reply"], + ]); + rooms.splice(0, rooms.length, row([root.id], next.id)); + await unread.refresh(); + h.emit([next]); + expect(unread.following("room", root.id)).toBe(true); + expect(thread()).toMatchObject({ + unreadCount: 1, + preview: "Second reply", + }); + vi.unstubAllGlobals(); +}); diff --git a/src/features/relay/unread.ts b/src/features/relay/unread.ts index ad5e7ba67e..9fe5298b21 100644 --- a/src/features/relay/unread.ts +++ b/src/features/relay/unread.ts @@ -145,6 +145,17 @@ export interface UnreadCapability { clearUnreadLocal(target: ReadTarget): Promise; readonly syncedManualUnread: false; } +/** Server read state's per-message answer; undefined leaves it to local evidence. */ +export type RemoteReadDecision = Readonly<{ + unread( + event: RelayEvent, + channelId: string, + threadRootId: string | undefined, + ): boolean | undefined; + /** Whether the viewer follows this thread, when the server knows. */ + following?(channelId: string, rootId: string): boolean | undefined; + subscribe(listener: () => void): () => void; +}>; /** Rendered message order: effective ms, then event id. */ const order = (a: RelayEvent, b: RelayEvent) => eventMs(a) - eventMs(b) || a.id.localeCompare(b.id); @@ -209,6 +220,7 @@ export function createUnread({ profiles, notify = (listener) => listener(), follows = memoryThreadFollows(), + remote, }: { reads: ReturnType; channels: ChannelQueries; @@ -220,6 +232,9 @@ export function createUnread({ profiles?: ProfileQueries; notify?: (listener: () => void) => void; follows?: ThreadFollowStorage; + /** Server-owned read state: when it decides a message, it replaces the + * local frontiers and conversation rules for it. */ + remote?: RemoteReadDecision | undefined; }) { let closed = false, epoch = 0; @@ -495,11 +510,13 @@ export function createUnread({ // Read on first use: constructing a session touches no storage. let saved: ThreadFollows | undefined; const choices = () => (saved ??= follows.read()); - /** An explicit follow choice for the reply's whole thread, if any. */ + /** The server's follow decision for the reply's whole thread, or else an + * explicit choice made here, if any. */ const chosen = ({ channelId, threadRootId }: Evidence) => threadRootId === undefined ? undefined - : choices().get(conversationKey(channelId, threadRootId)); + : (remote?.following?.(channelId, threadRootId) ?? + choices().get(conversationKey(channelId, threadRootId))); /** The reply is in a conversation the viewer is part of in its own channel: * the viewer follows its thread, or (without an explicit choice) wrote or * replied anywhere in that thread, wrote the parent, or also replied to the @@ -546,6 +563,8 @@ export function createUnread({ reads.localUnread(`msg:${event.id}`) ) return true; + const decided = remote?.unread(event, channelId, entry.threadRootId); + if (decided !== undefined) return decided && !entry.quiet; return relevant(entry, dm) && afterFrontier(entry, state, dm); } /** The read-state half of unread, before conversation relevance. */ @@ -1183,6 +1202,7 @@ export function createUnread({ notify(listener); } const stopRead = reads.subscribe(publish); + const stopRemote = remote?.subscribe(() => publish()); const stopProfiles = profiles?.subscribe(() => publish()); /** A choice changes relevance in every projection and held live alert. */ function followsChanged(channelIds?: ReadonlySet) { @@ -2518,6 +2538,7 @@ export function createUnread({ membershipListeners.clear(); for (const stop of [...handles]) stop(); stopRead(); + stopRemote?.(); stopProfiles?.(); stopFollows(); stopChannels();