From 16c2ef7beaae3d9d355e10ce8f8dcc70536b13fb Mon Sep 17 00:00:00 2001 From: Wes Date: Tue, 28 Apr 2026 14:06:53 -0700 Subject: [PATCH] Instant unread badges: remove event buffering, add kind:40002 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Remove all event buffering from the relay client on both platforms so unread badges appear the instant a WebSocket event arrives. Desktop: remove MessageChannel-based event buffer from relayClientSession.ts — handleEvent now calls subscription.onEvent() directly. Mobile: remove Timer-based 16ms event buffer from relay_session.dart — _handleEvent now calls liveSub.onEvent() directly with inline dedup. Add KIND_STREAM_MESSAGE_V2 (40002) to CHANNEL_EVENT_KINDS on desktop and channelEventKinds on mobile so rich-format messages trigger instant badge updates instead of waiting for the 60-second polling refetch. Also includes NIP-RS cross-device read state sync (useReadStateSync hook on desktop, ReadStateSyncNotifier on mobile), read state seeding on first load, and a sync toggle in settings. Co-Authored-By: Claude Opus 4.6 --- crates/sprout-core/src/kind.rs | 4 + crates/sprout-relay/src/handlers/ingest.rs | 9 +- desktop/scripts/check-file-sizes.mjs | 4 +- desktop/src-tauri/Cargo.toml | 2 +- desktop/src-tauri/src/commands/identity.rs | 35 +- desktop/src-tauri/src/lib.rs | 2 + desktop/src/app/AppShell.tsx | 9 +- .../features/channels/lib/readStateSync.ts | 223 +++++++++ .../src/features/channels/useReadStateSync.ts | 419 +++++++++++++++++ .../features/channels/useUnreadChannels.ts | 204 +++------ .../settings/ui/NotificationSettingsCard.tsx | 24 + .../features/settings/ui/SettingsPanels.tsx | 4 + .../features/settings/ui/SettingsScreen.tsx | 6 + .../src/features/settings/ui/SettingsView.tsx | 4 + desktop/src/main.tsx | 21 + desktop/src/shared/api/relayClientSession.ts | 82 ++-- desktop/src/shared/api/relayClientShared.ts | 1 + desktop/src/shared/api/tauri.ts | 9 + desktop/src/shared/constants/kinds.ts | 2 + .../channels/channel_detail_page.dart | 15 + .../lib/features/channels/channels_page.dart | 40 ++ .../features/channels/channels_provider.dart | 40 ++ .../channels/read_state_provider.dart | 432 ++++++++++++++++++ .../features/channels/read_state_sync.dart | 174 +++++++ .../lib/features/settings/settings_page.dart | 36 ++ mobile/lib/shared/relay/nostr_models.dart | 6 + mobile/lib/shared/relay/relay_session.dart | 113 ++--- 27 files changed, 1681 insertions(+), 239 deletions(-) create mode 100644 desktop/src/features/channels/lib/readStateSync.ts create mode 100644 desktop/src/features/channels/useReadStateSync.ts create mode 100644 mobile/lib/features/channels/read_state_provider.dart create mode 100644 mobile/lib/features/channels/read_state_sync.dart diff --git a/crates/sprout-core/src/kind.rs b/crates/sprout-core/src/kind.rs index 27507c471d..c0dfa85290 100644 --- a/crates/sprout-core/src/kind.rs +++ b/crates/sprout-core/src/kind.rs @@ -25,6 +25,9 @@ pub const KIND_FILE_METADATA: u32 = 1063; /// Parameterized replaceable (NIP-33, 30000–39999 range) — keyed by `(pubkey, kind, d_tag)`. /// Stored globally (channel_id = NULL); author-owned, not channel-scoped. pub const KIND_LONG_FORM: u32 = 30023; +/// NIP-78: Arbitrary custom app data (parameterized replaceable). +/// Used by NIP-RS for cross-device read state sync. +pub const KIND_READ_STATE: u32 = 30078; /// NIP-42 auth event — never stored (carries bearer tokens). pub const KIND_AUTH: u32 = 22242; @@ -286,6 +289,7 @@ pub const ALL_KINDS: &[u32] = &[ KIND_MEMBER_ADDED_NOTIFICATION, KIND_MEMBER_REMOVED_NOTIFICATION, KIND_LONG_FORM, + KIND_READ_STATE, KIND_FORUM_POST, KIND_FORUM_VOTE, KIND_FORUM_COMMENT, diff --git a/crates/sprout-relay/src/handlers/ingest.rs b/crates/sprout-relay/src/handlers/ingest.rs index 72fbe6c2a0..82f9f91edc 100644 --- a/crates/sprout-relay/src/handlers/ingest.rs +++ b/crates/sprout-relay/src/handlers/ingest.rs @@ -20,9 +20,10 @@ use sprout_core::kind::{ KIND_MEMBER_REMOVED_NOTIFICATION, KIND_NIP29_CREATE_GROUP, KIND_NIP29_DELETE_EVENT, KIND_NIP29_DELETE_GROUP, KIND_NIP29_EDIT_METADATA, KIND_NIP29_JOIN_REQUEST, KIND_NIP29_LEAVE_REQUEST, KIND_NIP29_PUT_USER, KIND_NIP29_REMOVE_USER, KIND_PRESENCE_UPDATE, - KIND_PROFILE, KIND_REACTION, KIND_STREAM_MESSAGE, KIND_STREAM_MESSAGE_BOOKMARKED, - KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, KIND_STREAM_MESSAGE_PINNED, - KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, KIND_STREAM_REMINDER, KIND_TEXT_NOTE, + KIND_PROFILE, KIND_REACTION, KIND_READ_STATE, KIND_STREAM_MESSAGE, + KIND_STREAM_MESSAGE_BOOKMARKED, KIND_STREAM_MESSAGE_DIFF, KIND_STREAM_MESSAGE_EDIT, + KIND_STREAM_MESSAGE_PINNED, KIND_STREAM_MESSAGE_SCHEDULED, KIND_STREAM_MESSAGE_V2, + KIND_STREAM_REMINDER, KIND_TEXT_NOTE, }; use sprout_core::verification::verify_event; @@ -149,7 +150,7 @@ pub enum IngestError { /// Returns `Err` for unknown kinds — the relay rejects them. fn required_scope_for_kind(kind: u32, event: &Event) -> Result { match kind { - KIND_PROFILE => Ok(Scope::UsersWrite), + KIND_PROFILE | KIND_READ_STATE => Ok(Scope::UsersWrite), KIND_TEXT_NOTE | KIND_LONG_FORM => Ok(Scope::MessagesWrite), KIND_CONTACT_LIST => Ok(Scope::UsersWrite), KIND_DELETION diff --git a/desktop/scripts/check-file-sizes.mjs b/desktop/scripts/check-file-sizes.mjs index 79551b90e5..4e855b5135 100644 --- a/desktop/scripts/check-file-sizes.mjs +++ b/desktop/scripts/check-file-sizes.mjs @@ -40,7 +40,7 @@ const overrides = new Map([ ["src/features/settings/ui/SettingsView.tsx", 600], ["src/features/sidebar/ui/AppSidebar.tsx", 860], // channels + forums creation forms + Pulse nav ["src/features/tokens/ui/TokenSettingsCard.tsx", 800], - ["src/shared/api/relayClientSession.ts", 890], // durable websocket session manager with reconnect/replay/recovery state + sendTypingIndicator + fetchChannelHistoryBefore + subscribeToChannelLive (huddle TTS) + subscribeToHuddleEvents (huddle indicator) + disconnect() for workspace switch teardown + ["src/shared/api/relayClientSession.ts", 910], // durable websocket session manager with reconnect/replay/recovery state + sendTypingIndicator + fetchChannelHistoryBefore + subscribeToChannelLive (huddle TTS) + subscribeToHuddleEvents (huddle indicator) + disconnect() for workspace switch teardown + NIP-RS read state fetch/subscribe/publish + fetchOwnReadStateBlob ["src/shared/api/tauri.ts", 1100], // remote agent provider API bindings + canvas API functions ["src-tauri/src/lib.rs", 710], // sprout-media:// proxy + Range headers + Sprout nest init (ensure_nest) in setup() + huddle command registration + PTT global shortcut handler + persona pack commands + app_handle storage for event emission ["src-tauri/src/commands/media.rs", 720], // ffmpeg video transcode + poster frame extraction + run_ffmpeg_with_timeout (find_ffmpeg, is_video_file, transcode_to_mp4, extract_poster_frame, transcode_and_extract_poster) + spawn_blocking wrappers + tests @@ -68,7 +68,7 @@ const overrides = new Map([ ["src-tauri/src/relay.rs", 510], // +4 lines for NIP-OA auth tag injection in profile sync (build_profile_event) + verification test ["src-tauri/src/commands/pairing.rs", 550], // NIP-AB pairing actor: 3 Tauri commands + background WS task + NIP-42 auth + event parsing helpers ["src-tauri/src/lib.rs", 715], // +4 lines for PairingHandle managed state + 3 pairing command registrations - ["src/shared/api/tauri.ts", 1125], // +14 lines for 3 NIP-AB pairing command wrappers + applyWorkspace + ["src/shared/api/tauri.ts", 1140], // +14 lines for 3 NIP-AB pairing command wrappers + applyWorkspace + NIP-44 encrypt/decrypt wrappers ]); async function walkFiles(directory) { diff --git a/desktop/src-tauri/Cargo.toml b/desktop/src-tauri/Cargo.toml index e65f611dcf..87601b01fb 100644 --- a/desktop/src-tauri/Cargo.toml +++ b/desktop/src-tauri/Cargo.toml @@ -42,7 +42,7 @@ futures-util = "0.3" opus = "0.3" serde = { version = "1", features = ["derive"] } serde_json = "1" -nostr = "0.37" +nostr = { version = "0.37", features = ["nip44"] } nostr-compat = { package = "nostr", version = "0.36" } zeroize = "1" reqwest = { version = "0.13", features = ["json", "query", "stream"] } diff --git a/desktop/src-tauri/src/commands/identity.rs b/desktop/src-tauri/src/commands/identity.rs index 4844d7c98a..142d15207c 100644 --- a/desktop/src-tauri/src/commands/identity.rs +++ b/desktop/src-tauri/src/commands/identity.rs @@ -1,4 +1,4 @@ -use nostr::{EventBuilder, JsonUtil, Kind, Tag, ToBech32}; +use nostr::{nips::nip44, EventBuilder, JsonUtil, Kind, Tag, Timestamp, ToBech32}; use tauri::State; use crate::{ @@ -54,6 +54,7 @@ pub fn sign_event( kind: u16, content: String, tags: Vec>, + created_at: Option, state: State<'_, AppState>, ) -> Result { let keys = state.keys.lock().map_err(|error| error.to_string())?; @@ -63,8 +64,11 @@ pub fn sign_event( .map(|tag| Tag::parse(tag).map_err(|error| format!("invalid tag: {error}"))) .collect::, _>>()?; - let event = EventBuilder::new(Kind::Custom(kind), content) - .tags(nostr_tags) + let mut builder = EventBuilder::new(Kind::Custom(kind), content).tags(nostr_tags); + if let Some(ts) = created_at { + builder = builder.custom_created_at(Timestamp::from(ts)); + } + let event = builder .sign_with_keys(&keys) .map_err(|error| format!("sign failed: {error}"))?; @@ -79,6 +83,31 @@ pub fn get_nsec(state: State<'_, AppState>) -> Result { .map_err(|error| format!("encode nsec: {error}")) } +#[tauri::command] +pub fn nip44_encrypt_to_self( + plaintext: String, + state: State<'_, AppState>, +) -> Result { + let keys = state.keys.lock().map_err(|e| e.to_string())?; + nip44::encrypt( + keys.secret_key(), + &keys.public_key(), + plaintext, + nip44::Version::V2, + ) + .map_err(|e| format!("nip44 encrypt failed: {e}")) +} + +#[tauri::command] +pub fn nip44_decrypt_self( + ciphertext: String, + state: State<'_, AppState>, +) -> Result { + let keys = state.keys.lock().map_err(|e| e.to_string())?; + nip44::decrypt(keys.secret_key(), &keys.public_key(), ciphertext) + .map_err(|e| format!("nip44 decrypt failed: {e}")) +} + #[tauri::command] pub fn create_auth_event( challenge: String, diff --git a/desktop/src-tauri/src/lib.rs b/desktop/src-tauri/src/lib.rs index 69fb657e08..b16157c5a7 100644 --- a/desktop/src-tauri/src/lib.rs +++ b/desktop/src-tauri/src/lib.rs @@ -372,6 +372,8 @@ pub fn run() { discover_acp_providers, discover_managed_agent_prereqs, sign_event, + nip44_encrypt_to_self, + nip44_decrypt_self, create_auth_event, get_channels, create_channel, diff --git a/desktop/src/app/AppShell.tsx b/desktop/src/app/AppShell.tsx index 753b6b8c25..364bc9f552 100644 --- a/desktop/src/app/AppShell.tsx +++ b/desktop/src/app/AppShell.tsx @@ -207,7 +207,12 @@ export function AppShell() { [channels, selectedChannelId], ); - const { markChannelRead, unreadChannelIds } = useUnreadChannels( + const { + markChannelRead, + unreadChannelIds, + syncEnabled: readStateSyncEnabled, + setSyncEnabled: setReadStateSyncEnabled, + } = useUnreadChannels( channels, activeChannel, // Wait for ChannelScreen to report the latest loaded message before @@ -640,6 +645,8 @@ export function AppShell() { onSetNeedsActionNotificationsEnabled={ notificationSettings.setNeedsActionEnabled } + readStateSyncEnabled={readStateSyncEnabled} + onSetReadStateSyncEnabled={setReadStateSyncEnabled} section={settingsSection} /> diff --git a/desktop/src/features/channels/lib/readStateSync.ts b/desktop/src/features/channels/lib/readStateSync.ts new file mode 100644 index 0000000000..3c231a9acd --- /dev/null +++ b/desktop/src/features/channels/lib/readStateSync.ts @@ -0,0 +1,223 @@ +import { nip44DecryptSelf } from "@/shared/api/tauri"; +import type { RelayEvent } from "@/shared/api/types"; + +// ── Storage keys ───────────────────────────────────────────────────────────── + +export const CLIENT_ID_STORAGE_KEY = "sprout.read-state-client-id.v1"; +export const SLOT_ID_STORAGE_KEY = "sprout.read-state-slot-id.v1"; +export const SYNC_ENABLED_STORAGE_KEY = "sprout.read-state-sync-enabled.v1"; +export const CACHE_STORAGE_KEY = "sprout.channel-read-state.v1"; + +// ── Types ──────────────────────────────────────────────────────────────────── + +export type ReadStateBlob = { + v: 1; + client_id: string; + contexts: Record; +}; + +export type DecodedBlob = { + blob: ReadStateBlob; + event: RelayEvent; +}; + +// ── Random / ID helpers ───────────────────────────────────────────────────── + +export function randomHex(length: number): string { + const bytes = new Uint8Array(length / 2); + crypto.getRandomValues(bytes); + return Array.from(bytes, (b) => b.toString(16).padStart(2, "0")).join(""); +} + +export function getOrCreateStorageValue(key: string): string { + const existing = localStorage.getItem(key); + if (existing && existing.length > 0) { + return existing; + } + const value = randomHex(32); + localStorage.setItem(key, value); + return value; +} + +// ── Sync-enabled storage ──────────────────────────────────────────────────── + +export function readSyncEnabled(): boolean { + const stored = localStorage.getItem(SYNC_ENABLED_STORAGE_KEY); + // NIP-RS: MUST NOT publish without explicit user opt-in — default OFF + return stored === "true"; +} + +export function writeSyncEnabled(enabled: boolean): void { + localStorage.setItem(SYNC_ENABLED_STORAGE_KEY, String(enabled)); +} + +// ── Cached read state ─────────────────────────────────────────────────────── + +/** Read the localStorage cache of merged read state (unix timestamps). */ +export function readCachedReadState(): Record { + try { + const raw = localStorage.getItem(CACHE_STORAGE_KEY); + if (!raw) return {}; + const parsed = JSON.parse(raw); + if (!parsed || typeof parsed !== "object") return {}; + + // The old format stored ISO strings; the new format stores unix seconds. + // Support both: if a value is a string, parse it to unix seconds. + const result: Record = {}; + for (const [key, value] of Object.entries(parsed)) { + if (typeof value === "number" && Number.isFinite(value)) { + result[key] = value; + } else if (typeof value === "string") { + const ms = Date.parse(value); + if (!Number.isNaN(ms)) { + result[key] = Math.floor(ms / 1000); + } + } + } + return result; + } catch { + return {}; + } +} + +/** Write merged read state to localStorage cache. */ +export function writeCachedReadState(state: Record): void { + localStorage.setItem(CACHE_STORAGE_KEY, JSON.stringify(state)); +} + +// ── Validation per NIP-RS Content Validation ───────────────────────────────── + +export function validateDTag(event: RelayEvent): string | null { + const dTags = event.tags.filter((tag) => tag[0] === "d" && tag.length >= 2); + if (dTags.length !== 1) return null; + const dTag = dTags[0]; + const dValue = dTag?.[1]; + if (!dValue) return null; + if (!dValue.startsWith("read-state:")) return null; + const slotId = dValue.slice("read-state:".length); + if (slotId.length === 0 || slotId.length > 64) return null; + // ASCII check + for (let i = 0; i < slotId.length; i++) { + if (slotId.charCodeAt(i) > 127) return null; + } + return dValue; +} + +export function validateTTag(event: RelayEvent): boolean { + const tTags = event.tags.filter( + (tag) => tag[0] === "t" && tag[1] === "read-state", + ); + return tTags.length === 1; +} + +export async function decryptAndValidateBlob( + event: RelayEvent, +): Promise { + // Validate d tag + if (!validateDTag(event)) return null; + + // Validate t tag + if (!validateTTag(event)) return null; + + // Decrypt + let plaintext: string; + try { + plaintext = await nip44DecryptSelf(event.content); + } catch { + return null; + } + + // Parse JSON + let parsed: unknown; + try { + parsed = JSON.parse(plaintext); + } catch { + return null; + } + + if (!parsed || typeof parsed !== "object") return null; + const obj = parsed as Record; + + // Validate v + if (!("v" in obj) || typeof obj.v !== "number" || !Number.isInteger(obj.v)) { + return null; + } + if (obj.v !== 1) return null; // Unknown version — ignore + + // Validate client_id + if ( + !("client_id" in obj) || + typeof obj.client_id !== "string" || + obj.client_id.length < 1 || + obj.client_id.length > 64 + ) { + return null; + } + + // Validate contexts + if ( + !("contexts" in obj) || + typeof obj.contexts !== "object" || + obj.contexts === null || + Array.isArray(obj.contexts) + ) { + return null; + } + + const rawContexts = obj.contexts as Record; + + // Reject if > 10,000 entries + if (Object.keys(rawContexts).length > 10_000) return null; + + // Validate individual entries + const contexts: Record = {}; + for (const [contextId, timestamp] of Object.entries(rawContexts)) { + // Context ID must not exceed 256 bytes + if (new TextEncoder().encode(contextId).length > 256) continue; + // Timestamp must be integer in 0–4294967295 + if ( + typeof timestamp !== "number" || + !Number.isInteger(timestamp) || + timestamp < 0 || + timestamp > 4294967295 + ) { + continue; + } + contexts[contextId] = timestamp; + } + + return { + blob: { + v: 1, + client_id: obj.client_id as string, + contexts, + }, + event, + }; +} + +// ── Merge logic (CvRDT max-register) ──────────────────────────────────────── + +export function mergeContexts( + base: Record, + incoming: Record, +): Record { + const result = { ...base }; + for (const [ctx, ts] of Object.entries(incoming)) { + result[ctx] = Math.max(result[ctx] ?? 0, ts); + } + return result; +} + +export function contextsEqual( + a: Record, + b: Record, +): boolean { + const keysA = Object.keys(a); + const keysB = Object.keys(b); + if (keysA.length !== keysB.length) return false; + for (const key of keysA) { + if (a[key] !== b[key]) return false; + } + return true; +} diff --git a/desktop/src/features/channels/useReadStateSync.ts b/desktop/src/features/channels/useReadStateSync.ts new file mode 100644 index 0000000000..1d4c8d2e3b --- /dev/null +++ b/desktop/src/features/channels/useReadStateSync.ts @@ -0,0 +1,419 @@ +import * as React from "react"; + +import { relayClient } from "@/shared/api/relayClient"; +import { nip44EncryptToSelf, signRelayEvent } from "@/shared/api/tauri"; +import type { RelayEvent } from "@/shared/api/types"; +import { KIND_READ_STATE } from "@/shared/constants/kinds"; + +import { + CLIENT_ID_STORAGE_KEY, + SLOT_ID_STORAGE_KEY, + type ReadStateBlob, + type DecodedBlob, + randomHex, + getOrCreateStorageValue, + readSyncEnabled, + writeSyncEnabled, + readCachedReadState, + writeCachedReadState, + validateDTag, + decryptAndValidateBlob, + mergeContexts, + contextsEqual, +} from "@/features/channels/lib/readStateSync"; + +// ── The hook ───────────────────────────────────────────────────────────────── + +export function useReadStateSync(userPubkey: string | undefined) { + // Merged effective read state: contextId → unix seconds + const [mergedState, setMergedState] = + React.useState>(readCachedReadState); + const [syncEnabled, setSyncEnabledState] = React.useState(readSyncEnabled); + + // Refs for the sync engine (mutable, not triggering re-renders) + const clientIdRef = React.useRef(""); + const slotIdRef = React.useRef(""); + const ownBlobRef = React.useRef>({}); + const lastPublishedRef = React.useRef>({}); + const maxFetchedCreatedAtRef = React.useRef(0); + const debounceTimerRef = React.useRef | null>( + null, + ); + const isInitializedRef = React.useRef(false); + const isPublishingRef = React.useRef(false); + const needsRepublishRef = React.useRef(false); + const republishTimerRef = React.useRef | null>( + null, + ); + + // Update cache whenever merged state changes + React.useEffect(() => { + writeCachedReadState(mergedState); + }, [mergedState]); + + // Settings toggle + const setSyncEnabled = React.useCallback((enabled: boolean) => { + writeSyncEnabled(enabled); + setSyncEnabledState(enabled); + }, []); + + // ── Publish own blob ─────────────────────────────────────────────────── + + const publishOwnBlob = React.useCallback( + async (retryDepth = 0) => { + if (!userPubkey || !syncEnabled || !isInitializedRef.current) return; + + // Publish serialization: skip if already in-flight, flag for retry after + if (isPublishingRef.current) { + needsRepublishRef.current = true; + return; + } + isPublishingRef.current = true; + + try { + // Read-before-write: fetch own blob by slot-ID (NIP-RS §Read-before-write) + const slotId = slotIdRef.current; + + const events = await relayClient.fetchOwnReadStateBlob( + userPubkey, + slotId, + ); + + // Validate and merge fetched blob(s) + let fetchedOwnContexts: Record = {}; + let maxCreatedAt = maxFetchedCreatedAtRef.current; + + for (const event of events) { + maxCreatedAt = Math.max(maxCreatedAt, event.created_at); + + const decoded = await decryptAndValidateBlob(event); + if (!decoded) continue; + + // Slot-ID conflict check + if (decoded.blob.client_id !== clientIdRef.current) { + // Another client has claimed our slot — regenerate + const newSlotId = randomHex(32); + localStorage.setItem(SLOT_ID_STORAGE_KEY, newSlotId); + slotIdRef.current = newSlotId; + // Retry with new slot, up to 3 attempts + if (retryDepth >= 3) { + console.error( + "[ReadStateSync] Slot-ID conflict retry limit reached (3 attempts)", + ); + return; + } + isPublishingRef.current = false; + return publishOwnBlob(retryDepth + 1); + } + + fetchedOwnContexts = mergeContexts( + fetchedOwnContexts, + decoded.blob.contexts, + ); + } + + maxFetchedCreatedAtRef.current = maxCreatedAt; + + // Merge fetched with local + const merged = mergeContexts(fetchedOwnContexts, ownBlobRef.current); + ownBlobRef.current = merged; + + // Suppress publish if unchanged + if (contextsEqual(merged, lastPublishedRef.current)) return; + + // Build the blob payload + const payload: ReadStateBlob = { + v: 1, + client_id: clientIdRef.current, + contexts: merged, + }; + + const encrypted = await nip44EncryptToSelf(JSON.stringify(payload)); + + // NIP-RS §Clock Skew: created_at MUST be max(now, max_fetched + 1) + const now = Math.floor(Date.now() / 1000); + const createdAt = Math.max(now, maxFetchedCreatedAtRef.current + 1); + + const event = await signRelayEvent({ + kind: KIND_READ_STATE, + content: encrypted, + tags: [ + ["d", `read-state:${slotIdRef.current}`], + ["t", "read-state"], + ], + created_at: createdAt, + }); + + await relayClient.publishReadStateEvent(event); + lastPublishedRef.current = { ...merged }; + } catch (error) { + console.error("[ReadStateSync] Failed to publish read state:", error); + } finally { + isPublishingRef.current = false; + // If a publish was requested while we were in-flight, retry now + if (needsRepublishRef.current) { + needsRepublishRef.current = false; + void publishOwnBlob(); + } + } + }, + [userPubkey, syncEnabled], + ); + + // ── Mark a channel as read ───────────────────────────────────────────── + + const markContextRead = React.useCallback( + (contextId: string, timestamp: number) => { + // Update local state immediately + setMergedState((current) => { + const existing = current[contextId] ?? 0; + if (timestamp <= existing) return current; + return { ...current, [contextId]: timestamp }; + }); + + // Update own blob + const existingOwn = ownBlobRef.current[contextId] ?? 0; + if (timestamp > existingOwn) { + ownBlobRef.current = { + ...ownBlobRef.current, + [contextId]: timestamp, + }; + } + + // Schedule debounced publish + if (debounceTimerRef.current) { + clearTimeout(debounceTimerRef.current); + } + debounceTimerRef.current = setTimeout(() => { + debounceTimerRef.current = null; + void publishOwnBlob(); + }, 5_000); + }, + [publishOwnBlob], + ); + + // ── Initialization: fetch + merge + subscribe ────────────────────────── + + React.useEffect(() => { + if (!userPubkey) return; + // Capture narrowed type for closures below + const pubkey: string = userPubkey; + + const clientId = getOrCreateStorageValue(CLIENT_ID_STORAGE_KEY); + const slotId = getOrCreateStorageValue(SLOT_ID_STORAGE_KEY); + clientIdRef.current = clientId; + slotIdRef.current = slotId; + + let disposed = false; + let subscribeDispose: (() => Promise) | undefined; + + async function initialize() { + const since = Math.floor(Date.now() / 1000) - 7 * 86400; + + // Fetch all read-state blobs within horizon + let events: RelayEvent[]; + try { + events = await relayClient.fetchReadStateEvents(pubkey, since); + } catch (error) { + console.error("[ReadStateSync] Failed to fetch read state:", error); + isInitializedRef.current = true; + return; + } + + if (disposed) return; + + // Decode all blobs + const decoded: DecodedBlob[] = []; + for (const event of events) { + const result = await decryptAndValidateBlob(event); + if (result) decoded.push(result); + } + + if (disposed) return; + + // Find own blob(s) + const ownBlobs = decoded.filter((d) => d.blob.client_id === clientId); + + // Handle duplicate client_id: use highest created_at + if (ownBlobs.length > 1) { + ownBlobs.sort((a, b) => b.event.created_at - a.event.created_at); + // Delete stale duplicates via kind:5 + for (let i = 1; i < ownBlobs.length; i++) { + const stale = ownBlobs[i]; + if (!stale) continue; + const staleD = validateDTag(stale.event); + if (staleD) { + try { + const deleteEvent = await signRelayEvent({ + kind: 5, + content: "", + tags: [["a", `${KIND_READ_STATE}:${pubkey}:${staleD}`]], + }); + await relayClient.publishReadStateEvent(deleteEvent); + } catch { + // Best-effort deletion + } + } + } + } + + // Slot-ID conflict detection + const ownDTag = `read-state:${slotId}`; + const conflictingBlob = decoded.find((d) => { + const dTag = validateDTag(d.event); + return dTag === ownDTag && d.blob.client_id !== clientId; + }); + if (conflictingBlob) { + const newSlotId = randomHex(32); + localStorage.setItem(SLOT_ID_STORAGE_KEY, newSlotId); + slotIdRef.current = newSlotId; + } + + // Track max created_at + let maxCreatedAt = 0; + for (const d of decoded) { + maxCreatedAt = Math.max(maxCreatedAt, d.event.created_at); + } + maxFetchedCreatedAtRef.current = maxCreatedAt; + + // Set own blob from highest-priority own blob + const primaryOwnBlob = ownBlobs[0]; + if (primaryOwnBlob) { + ownBlobRef.current = { ...primaryOwnBlob.blob.contexts }; + lastPublishedRef.current = { ...primaryOwnBlob.blob.contexts }; + } + + // Merge ALL blobs (own + others) using max() per context + let effective: Record = {}; + for (const d of decoded) { + effective = mergeContexts(effective, d.blob.contexts); + } + + // Merge with cached localStorage state (may have local reads not yet published) + const cached = readCachedReadState(); + effective = mergeContexts(effective, cached); + + // Also merge own blob ref (may have writes since cache was read) + effective = mergeContexts(effective, ownBlobRef.current); + + if (!disposed) { + // Use functional update to merge with any marks that arrived while + // initialize() was running (e.g. from useUnreadChannels init effect). + // A plain `setMergedState(effective)` would overwrite those marks. + setMergedState((current) => mergeContexts(current, effective)); + isInitializedRef.current = true; + } + + // ── Live subscription ──────────────────────────────────────────── + if (disposed) return; + + void relayClient + .subscribeToReadState(pubkey, (event: RelayEvent) => { + if (disposed) return; + void handleIncomingEvent(event); + }) + .then((dispose) => { + if (disposed) { + void dispose?.(); + } else { + subscribeDispose = dispose; + } + }) + .catch((error) => { + console.error( + "[ReadStateSync] Failed to subscribe to read state:", + error, + ); + }); + } + + async function handleIncomingEvent(event: RelayEvent) { + const decoded = await decryptAndValidateBlob(event); + if (!decoded) return; + + // Update max created_at + maxFetchedCreatedAtRef.current = Math.max( + maxFetchedCreatedAtRef.current, + event.created_at, + ); + + if (decoded.blob.client_id === clientIdRef.current) { + // Our own blob echoed back — update own ref + ownBlobRef.current = mergeContexts( + ownBlobRef.current, + decoded.blob.contexts, + ); + setMergedState((current) => + mergeContexts(current, decoded.blob.contexts), + ); + return; + } + + // Another client's blob — merge into state + let needsRepublish = false; + + setMergedState((current) => { + const merged = mergeContexts(current, decoded.blob.contexts); + + // Check if any context advanced beyond our last-published state + for (const [ctx, ts] of Object.entries(decoded.blob.contexts)) { + if (ts > (lastPublishedRef.current[ctx] ?? 0)) { + needsRepublish = true; + break; + } + } + + return merged; + }); + + // Schedule re-publish if another device advanced a context beyond our + // last publish — but do NOT absorb their data into ownBlobRef. + if (needsRepublish && syncEnabled) { + if (republishTimerRef.current) { + clearTimeout(republishTimerRef.current); + } + republishTimerRef.current = setTimeout(() => { + republishTimerRef.current = null; + void publishOwnBlob(); + }, 5_000); + } + } + + void initialize(); + + return () => { + disposed = true; + if (debounceTimerRef.current) { + clearTimeout(debounceTimerRef.current); + debounceTimerRef.current = null; + } + if (republishTimerRef.current) { + clearTimeout(republishTimerRef.current); + republishTimerRef.current = null; + } + // Handle fast-unmount: if subscribe resolved, dispose immediately; + // otherwise the .then() handler will dispose when it resolves. + if (subscribeDispose) { + void subscribeDispose(); + } + // Reset refs so a re-mount starts fresh + isInitializedRef.current = false; + isPublishingRef.current = false; + needsRepublishRef.current = false; + ownBlobRef.current = {}; + lastPublishedRef.current = {}; + maxFetchedCreatedAtRef.current = 0; + }; + }, [userPubkey, publishOwnBlob, syncEnabled]); + + return { + /** Merged effective read state: contextId → unix seconds (0 = unknown). */ + mergedState, + /** Mark a context as read up to a unix timestamp (seconds). */ + markContextRead, + /** Whether cross-device sync publishing is enabled. */ + syncEnabled, + /** Toggle cross-device sync publishing. */ + setSyncEnabled, + }; +} diff --git a/desktop/src/features/channels/useUnreadChannels.ts b/desktop/src/features/channels/useUnreadChannels.ts index 6b1b0258d5..00ee40cf95 100644 --- a/desktop/src/features/channels/useUnreadChannels.ts +++ b/desktop/src/features/channels/useUnreadChannels.ts @@ -3,67 +3,16 @@ import { useLiveChannelUpdates, type UseLiveChannelUpdatesOptions, } from "@/features/channels/useLiveChannelUpdates"; +import { useReadStateSync } from "@/features/channels/useReadStateSync"; import type { Channel } from "@/shared/api/types"; -const CHANNEL_READ_STATE_STORAGE_KEY = "sprout.channel-read-state.v1"; - -type ChannelReadState = Record; type UseUnreadChannelsOptions = UseLiveChannelUpdatesOptions; -function parseTimestamp(value: string | null | undefined) { - if (!value) { - return null; - } - - const timestamp = Date.parse(value); - return Number.isNaN(timestamp) ? null : timestamp; -} - -function normalizeTimestamp(value: string | null | undefined) { - const timestamp = parseTimestamp(value); - return timestamp === null ? null : new Date(timestamp).toISOString(); -} - -function isNewerTimestamp( - candidate: string | null | undefined, - current: string | null | undefined, -) { - const candidateTimestamp = parseTimestamp(candidate); - if (candidateTimestamp === null) { - return false; - } - - const currentTimestamp = parseTimestamp(current); - return currentTimestamp === null || candidateTimestamp > currentTimestamp; -} - -function readStoredChannelReadState(): ChannelReadState { - if (typeof window === "undefined") { - return {}; - } - - const rawState = window.localStorage.getItem(CHANNEL_READ_STATE_STORAGE_KEY); - if (!rawState) { - return {}; - } - - try { - const parsed = JSON.parse(rawState); - if (!parsed || typeof parsed !== "object") { - return {}; - } - - return Object.fromEntries( - Object.entries(parsed).map(([channelId, value]) => [ - channelId, - typeof value === "string" || value === null - ? normalizeTimestamp(value) - : null, - ]), - ); - } catch { - return {}; - } +/** Parse an ISO timestamp string to unix seconds, or null. */ +function isoToUnixSeconds(value: string | null | undefined): number | null { + if (!value) return null; + const ms = Date.parse(value); + return Number.isNaN(ms) ? null : Math.floor(ms / 1000); } export function useUnreadChannels( @@ -72,9 +21,6 @@ export function useUnreadChannels( activeReadAt?: string | null, options: UseUnreadChannelsOptions = {}, ) { - const [lastReadByChannel, setLastReadByChannel] = - React.useState(readStoredChannelReadState); - const hasInitializedChannelsRef = React.useRef(false); const activeChannelId = activeChannel?.id ?? null; const activeChannelLastMessageAt = activeChannel?.lastMessageAt ?? null; // Let callers pass `null` to intentionally suppress the optimistic @@ -82,109 +28,83 @@ export function useUnreadChannels( const effectiveActiveReadAt = activeReadAt === undefined ? activeChannelLastMessageAt : activeReadAt; + const { mergedState, markContextRead, syncEnabled, setSyncEnabled } = + useReadStateSync(options.currentPubkey); + + const hasInitializedChannelsRef = React.useRef(false); + const mergedStateRef = React.useRef(mergedState); + mergedStateRef.current = mergedState; + const markContextReadRef = React.useRef(markContextRead); + markContextReadRef.current = markContextRead; + + // ── markChannelRead: preserve the same external API (ISO string) ───── + const markChannelRead = React.useCallback( (channelId: string, readAt: string | null | undefined) => { - const normalizedReadAt = normalizeTimestamp(readAt); - - setLastReadByChannel((current) => { - const previousReadAt = current[channelId] ?? null; - - if (normalizedReadAt === null) { - if (channelId in current) { - return current; - } - - return { - ...current, - [channelId]: null, - }; - } - - if (!isNewerTimestamp(normalizedReadAt, previousReadAt)) { - return current; - } - - return { - ...current, - [channelId]: normalizedReadAt, - }; - }); + const unixSeconds = isoToUnixSeconds(readAt); + if (unixSeconds === null) return; + markContextRead(channelId, unixSeconds); }, - [], + [markContextRead], ); - React.useEffect(() => { - if (typeof window === "undefined") { - return; - } - - window.localStorage.setItem( - CHANNEL_READ_STATE_STORAGE_KEY, - JSON.stringify(lastReadByChannel), - ); - }, [lastReadByChannel]); + // ── Initialize new channels: mark as read on first load ────────────── React.useEffect(() => { - if (channels.length === 0) { - return; - } - - setLastReadByChannel((current) => { - const knownChannelIds = new Set(channels.map((channel) => channel.id)); - const nextReadState: ChannelReadState = {}; - let didChange = false; + if (channels.length === 0) return; + if (!hasInitializedChannelsRef.current) { + // On first channel load, mark channels that are unknown to the merged + // read state as read at their lastMessageAt, so everything starts as + // "read" (same as the old localStorage-only behaviour). + const currentMerged = mergedStateRef.current; + const currentMarkRead = markContextReadRef.current; for (const channel of channels) { - if (channel.id in current) { - nextReadState[channel.id] = current[channel.id] ?? null; - continue; - } - - nextReadState[channel.id] = hasInitializedChannelsRef.current - ? null - : normalizeTimestamp(channel.lastMessageAt); - didChange = true; - } - - for (const channelId of Object.keys(current)) { - if (!knownChannelIds.has(channelId)) { - didChange = true; + if (currentMerged[channel.id] === undefined) { + const ts = isoToUnixSeconds(channel.lastMessageAt); + if (ts !== null) { + currentMarkRead(channel.id, ts); + } } } - - return didChange ? nextReadState : current; - }); - - hasInitializedChannelsRef.current = true; + hasInitializedChannelsRef.current = true; + } }, [channels]); - React.useEffect(() => { - if (!activeChannelId) { - return; - } + // ── Auto-mark active channel as read ───────────────────────────────── + React.useEffect(() => { + if (!activeChannelId) return; markChannelRead(activeChannelId, effectiveActiveReadAt); }, [activeChannelId, effectiveActiveReadAt, markChannelRead]); + + // ── Live channel updates (message cache, callbacks) ────────────────── + useLiveChannelUpdates(channels, activeChannelId, options); - const unreadChannelIds = React.useMemo( - () => - new Set( - channels - .filter((channel) => channel.id !== activeChannelId) - .filter((channel) => - isNewerTimestamp( - channel.lastMessageAt, - lastReadByChannel[channel.id], - ), - ) - .map((channel) => channel.id), - ), - [activeChannelId, channels, lastReadByChannel], - ); + // ── Compute unread set ─────────────────────────────────────────────── + + const unreadChannelIds = React.useMemo(() => { + return new Set( + channels + .filter((channel) => channel.id !== activeChannelId) + .filter((channel) => { + if (!channel.lastMessageAt) return false; + const lastMessageMs = Date.parse(channel.lastMessageAt); + if (Number.isNaN(lastMessageMs)) return false; + // Read state stores unix seconds (NIP-RS). Convert to ms for + // sub-second–accurate comparison against the ISO lastMessageAt. + const lastReadMs = (mergedState[channel.id] ?? 0) * 1000; + return lastMessageMs > lastReadMs; + }) + .map((channel) => channel.id), + ); + }, [activeChannelId, channels, mergedState]); return { unreadChannelIds, markChannelRead, + syncEnabled, + setSyncEnabled, }; } diff --git a/desktop/src/features/settings/ui/NotificationSettingsCard.tsx b/desktop/src/features/settings/ui/NotificationSettingsCard.tsx index dd6735e09c..5763be673c 100644 --- a/desktop/src/features/settings/ui/NotificationSettingsCard.tsx +++ b/desktop/src/features/settings/ui/NotificationSettingsCard.tsx @@ -3,6 +3,7 @@ import { BellRing, CircleAlert, Home as HomeIcon, + RefreshCw, type LucideIcon, } from "lucide-react"; @@ -91,6 +92,8 @@ export function NotificationSettingsCard({ onSetHomeBadgeEnabled, onSetMentionNotificationsEnabled, onSetNeedsActionNotificationsEnabled, + readStateSyncEnabled, + onSetReadStateSyncEnabled, }: { isUpdatingDesktopNotifications: boolean; notificationErrorMessage: string | null; @@ -100,6 +103,8 @@ export function NotificationSettingsCard({ onSetHomeBadgeEnabled: (enabled: boolean) => void; onSetMentionNotificationsEnabled: (enabled: boolean) => void; onSetNeedsActionNotificationsEnabled: (enabled: boolean) => void; + readStateSyncEnabled: boolean; + onSetReadStateSyncEnabled: (enabled: boolean) => void; }) { return (
@@ -184,6 +189,25 @@ export function NotificationSettingsCard({ /> +
+

Read state

+

+ Sync which channels you've read across all your devices. +

+
+ { + onSetReadStateSyncEnabled(!readStateSyncEnabled); + }} + testId="read-state-sync-toggle" + title="Sync read state across devices" + /> +
+
+ {notificationErrorMessage ? (

{notificationErrorMessage} diff --git a/desktop/src/features/settings/ui/SettingsPanels.tsx b/desktop/src/features/settings/ui/SettingsPanels.tsx index 046c34b0a7..59d48bd2dd 100644 --- a/desktop/src/features/settings/ui/SettingsPanels.tsx +++ b/desktop/src/features/settings/ui/SettingsPanels.tsx @@ -58,6 +58,8 @@ export type SettingsPanelProps = { onSetHomeBadgeEnabled: (enabled: boolean) => void; onSetMentionNotificationsEnabled: (enabled: boolean) => void; onSetNeedsActionNotificationsEnabled: (enabled: boolean) => void; + readStateSyncEnabled: boolean; + onSetReadStateSyncEnabled: (enabled: boolean) => void; }; export const settingsSections: SettingsSectionDescriptor[] = [ @@ -246,6 +248,8 @@ export function renderSettingsSection( onSetNeedsActionNotificationsEnabled={ props.onSetNeedsActionNotificationsEnabled } + readStateSyncEnabled={props.readStateSyncEnabled} + onSetReadStateSyncEnabled={props.onSetReadStateSyncEnabled} /> ); case "appearance": diff --git a/desktop/src/features/settings/ui/SettingsScreen.tsx b/desktop/src/features/settings/ui/SettingsScreen.tsx index 03bbffcc49..982df30e23 100644 --- a/desktop/src/features/settings/ui/SettingsScreen.tsx +++ b/desktop/src/features/settings/ui/SettingsScreen.tsx @@ -16,6 +16,8 @@ type SettingsScreenProps = { onSetHomeBadgeEnabled: (enabled: boolean) => void; onSetMentionNotificationsEnabled: (enabled: boolean) => void; onSetNeedsActionNotificationsEnabled: (enabled: boolean) => void; + readStateSyncEnabled: boolean; + onSetReadStateSyncEnabled: (enabled: boolean) => void; section: SettingsSection; }; @@ -32,6 +34,8 @@ export function SettingsScreen({ onSetHomeBadgeEnabled, onSetMentionNotificationsEnabled, onSetNeedsActionNotificationsEnabled, + readStateSyncEnabled, + onSetReadStateSyncEnabled, section, }: SettingsScreenProps) { return ( @@ -50,6 +54,8 @@ export function SettingsScreen({ onSetNeedsActionNotificationsEnabled={ onSetNeedsActionNotificationsEnabled } + readStateSyncEnabled={readStateSyncEnabled} + onSetReadStateSyncEnabled={onSetReadStateSyncEnabled} section={section} /> ); diff --git a/desktop/src/features/settings/ui/SettingsView.tsx b/desktop/src/features/settings/ui/SettingsView.tsx index b54d5fe591..dbfc166731 100644 --- a/desktop/src/features/settings/ui/SettingsView.tsx +++ b/desktop/src/features/settings/ui/SettingsView.tsx @@ -74,6 +74,8 @@ export function SettingsView({ onSetHomeBadgeEnabled, onSetMentionNotificationsEnabled, onSetNeedsActionNotificationsEnabled, + readStateSyncEnabled, + onSetReadStateSyncEnabled, section, }: SettingsViewProps) { const [isLoaded, setIsLoaded] = React.useState(false); @@ -192,6 +194,8 @@ export function SettingsView({ onSetHomeBadgeEnabled, onSetMentionNotificationsEnabled, onSetNeedsActionNotificationsEnabled, + readStateSyncEnabled, + onSetReadStateSyncEnabled, })}

diff --git a/desktop/src/main.tsx b/desktop/src/main.tsx index ff8988551b..2781c63db0 100644 --- a/desktop/src/main.tsx +++ b/desktop/src/main.tsx @@ -1,6 +1,7 @@ import React from "react"; import ReactDOM from "react-dom/client"; import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { notifyManager } from "@tanstack/react-query"; import { App } from "@/app/App"; import "@/shared/styles/globals.css"; import { WorkspacesProvider } from "@/features/workspaces/useWorkspaces"; @@ -8,6 +9,26 @@ import { ThemeProvider } from "@/shared/theme/ThemeProvider"; import { Toaster } from "@/shared/ui/sonner"; import { TooltipProvider } from "@/shared/ui/tooltip"; +// Override React Query's default setTimeout(cb, 0) scheduler with +// MessageChannel. WebKit throttles setTimeout to 1s boundaries when +// the window is occluded; MessageChannel macrotasks bypass this. +const notifyChannel = new MessageChannel(); +const notifyQueue: Array<() => void> = []; +notifyChannel.port1.onmessage = () => { + const fns = notifyQueue.splice(0); + for (const fn of fns) { + try { + fn(); + } catch (e) { + console.error(e); + } + } +}; +notifyManager.setScheduler((cb) => { + notifyQueue.push(cb); + notifyChannel.port2.postMessage(null); +}); + type E2eWindow = Window & { __SPROUT_E2E__?: unknown; }; diff --git a/desktop/src/shared/api/relayClientSession.ts b/desktop/src/shared/api/relayClientSession.ts index c8bee22695..acef53ffcb 100644 --- a/desktop/src/shared/api/relayClientSession.ts +++ b/desktop/src/shared/api/relayClientSession.ts @@ -23,8 +23,7 @@ import { buildThreadReferenceTags } from "@/features/messages/lib/threading"; const RECONNECT_BASE_DELAY_MS = 1_000, RECONNECT_MAX_DELAY_MS = 30_000; -const RECONNECT_REPLAY_SKEW_SECS = 5, - EVENT_BATCH_MS = 16; +const RECONNECT_REPLAY_SKEW_SECS = 5; export class RelayClient { private wsId: number | null = null; @@ -41,8 +40,6 @@ export class RelayClient { } | null = null; private subscriptions = new Map(); private pendingEvents = new Map(); - private eventBuffer: Array<{ subId: string; event: RelayEvent }> = []; - private flushTimeout: number | null = null; private reconnectListeners = new Set<() => void>(); private hasConnectedOnce = false; private notifyReconnectListeners = false; @@ -94,11 +91,6 @@ export class RelayClient { this.pendingEvents.delete(eventId); } - if (this.flushTimeout !== null) { - window.clearTimeout(this.flushTimeout); - this.flushTimeout = null; - } - this.eventBuffer = []; this.reconnectListeners.clear(); this.onMessageChannel = null; this.reconnectDelayMs = RECONNECT_BASE_DELAY_MS; @@ -292,6 +284,55 @@ export class RelayClient { ); } + async fetchReadStateEvents( + userPubkey: string, + since: number, + ): Promise { + return this.fetchHistory({ + kinds: [30078], + authors: [userPubkey], + "#t": ["read-state"], + since, + limit: 100, + }); + } + + async fetchOwnReadStateBlob( + userPubkey: string, + slotId: string, + ): Promise { + return this.fetchHistory({ + kinds: [30078], + authors: [userPubkey], + "#d": [`read-state:${slotId}`], + limit: 10, + }); + } + + async subscribeToReadState( + userPubkey: string, + onEvent: (event: RelayEvent) => void, + ) { + return this.subscribe( + { + kinds: [30078], + authors: [userPubkey], + "#t": ["read-state"], + limit: 100, + }, + onEvent, + ); + } + + async publishReadStateEvent(event: RelayEvent): Promise { + await this.ensureConnected(); + return this.publishEvent( + event, + "Timed out while publishing read state.", + "Failed to publish read state.", + ); + } + async preconnect() { this.keepAliveRequested = true; await this.ensureConnected(); @@ -661,25 +702,7 @@ export class RelayClient { event.created_at, ); - this.eventBuffer.push({ subId, event }); - this.flushTimeout ??= window.setTimeout( - () => this.flushEventBuffer(), - EVENT_BATCH_MS, - ); - } - - private flushEventBuffer() { - this.flushTimeout = null; - const buffer = this.eventBuffer; - this.eventBuffer = []; - - // Re-lookup: subscriptions removed during batch window are intentionally skipped. - for (const { subId, event } of buffer) { - const subscription = this.subscriptions.get(subId); - if (subscription?.mode === "live") { - subscription.onEvent(event); - } - } + subscription.onEvent(event); } private handleEose(subId: string) { @@ -834,9 +857,6 @@ export class RelayClient { }, ) { this.onMessageChannel = null; - if (this.flushTimeout !== null) window.clearTimeout(this.flushTimeout); - this.flushTimeout = null; - this.eventBuffer = []; if (options?.reconnect !== false && this.hasConnectedOnce) { this.notifyReconnectListeners = true; diff --git a/desktop/src/shared/api/relayClientShared.ts b/desktop/src/shared/api/relayClientShared.ts index 037d4428cf..f2dac401ad 100644 --- a/desktop/src/shared/api/relayClientShared.ts +++ b/desktop/src/shared/api/relayClientShared.ts @@ -5,6 +5,7 @@ export type RelaySubscriptionFilter = { limit: number; since?: number; until?: number; + authors?: string[]; } & Partial>; type HistorySubscription = { diff --git a/desktop/src/shared/api/tauri.ts b/desktop/src/shared/api/tauri.ts index 266a2c9c87..3c02f26706 100644 --- a/desktop/src/shared/api/tauri.ts +++ b/desktop/src/shared/api/tauri.ts @@ -792,10 +792,19 @@ export async function removeReaction( await invokeTauri("remove_reaction", { eventId, emoji }); } +export async function nip44EncryptToSelf(plaintext: string): Promise { + return invokeTauri("nip44_encrypt_to_self", { plaintext }); +} + +export async function nip44DecryptSelf(ciphertext: string): Promise { + return invokeTauri("nip44_decrypt_self", { ciphertext }); +} + export async function signRelayEvent(input: { kind: number; content: string; tags: string[][]; + created_at?: number; }): Promise { const eventJson = await invokeTauri("sign_event", input); return JSON.parse(eventJson) as RelayEvent; diff --git a/desktop/src/shared/constants/kinds.ts b/desktop/src/shared/constants/kinds.ts index 7366938466..9882800802 100644 --- a/desktop/src/shared/constants/kinds.ts +++ b/desktop/src/shared/constants/kinds.ts @@ -16,6 +16,7 @@ export const KIND_FORUM_POST = 45001; export const KIND_FORUM_COMMENT = 45003; export const KIND_APPROVAL_REQUEST = 46010; export const KIND_TYPING_INDICATOR = 20002; +export const KIND_READ_STATE = 30078; // Keep this in sync with the Home-feed mention query in sprout-db. export const HOME_MENTION_EVENT_KINDS = [ @@ -30,6 +31,7 @@ export const CHANNEL_EVENT_KINDS = [ KIND_REACTION, // 7 — NIP-25 reactions KIND_STREAM_MESSAGE, // 9 — NIP-29 group chat messages 40001, // legacy: pre-migration stream messages + KIND_STREAM_MESSAGE_V2, // 40002 — v2 stream messages KIND_STREAM_MESSAGE_EDIT, // 40003 — message edits KIND_STREAM_MESSAGE_DIFF, // 40008 — message diffs KIND_SYSTEM_MESSAGE, // 40099 — system messages (join, leave, etc.) diff --git a/mobile/lib/features/channels/channel_detail_page.dart b/mobile/lib/features/channels/channel_detail_page.dart index 29193cc5ae..ec90ac25df 100644 --- a/mobile/lib/features/channels/channel_detail_page.dart +++ b/mobile/lib/features/channels/channel_detail_page.dart @@ -19,6 +19,7 @@ import 'channel_management_provider.dart'; import 'channel_messages_provider.dart'; import 'channel_typing_provider.dart'; import 'channels_provider.dart'; +import 'read_state_provider.dart'; import 'compose_bar.dart'; import 'date_formatters.dart'; import 'day_divider.dart'; @@ -96,6 +97,20 @@ class ChannelDetailPage extends HookConsumerWidget { return null; }, [channel.id]); + // Mark channel as read when opened and when new messages arrive. + final lastMessageAt = resolvedChannel.lastMessageAt; + useEffect(() { + if (lastMessageAt != null) { + ref + .read(readStateSyncProvider.notifier) + .markContextRead( + channel.id, + lastMessageAt.millisecondsSinceEpoch ~/ 1000, + ); + } + return null; + }, [channel.id, lastMessageAt]); + return FrostedScaffold( appBar: FrostedAppBar( title: resolvedChannel.isDm diff --git a/mobile/lib/features/channels/channels_page.dart b/mobile/lib/features/channels/channels_page.dart index 137b8dc5c3..57ae83c3ac 100644 --- a/mobile/lib/features/channels/channels_page.dart +++ b/mobile/lib/features/channels/channels_page.dart @@ -22,6 +22,7 @@ import 'channel.dart'; import 'channel_detail_page.dart'; import 'channel_management_provider.dart'; import 'channels_provider.dart'; +import 'read_state_provider.dart'; enum _QuickAction { createChannel, createForum, newDm } @@ -36,6 +37,8 @@ class ChannelsPage extends HookConsumerWidget { Widget build(BuildContext context, WidgetRef ref) { final channelsAsync = ref.watch(channelsProvider); final sessionState = ref.watch(relaySessionProvider); + // Activate read-state seeding so new channels start as read. + ref.watch(readStateSeedProvider); final currentPubkey = ref .watch(profileProvider) .whenData((value) => value?.pubkey) @@ -62,11 +65,33 @@ class ChannelsPage extends HookConsumerWidget { Future openChannel(Channel channel) async { if (!context.mounted) return; + // Mark channel as read at its latest message timestamp. + if (channel.lastMessageAt != null) { + ref + .read(readStateSyncProvider.notifier) + .markContextRead( + channel.id, + channel.lastMessageAt!.millisecondsSinceEpoch ~/ 1000, + ); + } await Navigator.of(context).push( MaterialPageRoute( builder: (_) => ChannelDetailPage(channel: channel), ), ); + // Mark read again on return in case new messages arrived while viewing. + final updatedChannels = ref.read(channelsProvider).asData?.value ?? []; + final updated = updatedChannels + .where((c) => c.id == channel.id) + .firstOrNull; + if (updated?.lastMessageAt != null) { + ref + .read(readStateSyncProvider.notifier) + .markContextRead( + channel.id, + updated!.lastMessageAt!.millisecondsSinceEpoch ~/ 1000, + ); + } } Future openQuickActions() async { @@ -443,6 +468,8 @@ class _ChannelTile extends ConsumerWidget { @override Widget build(BuildContext context, WidgetRef ref) { final hasActivity = channel.lastMessageAt != null; + final unreadIds = ref.watch(unreadChannelIdsProvider); + final isUnread = unreadIds.contains(channel.id); return InkWell( borderRadius: BorderRadius.circular(Radii.md), @@ -479,11 +506,24 @@ class _ChannelTile extends ConsumerWidget { color: hasActivity ? context.colors.onSurface : context.colors.onSurfaceVariant, + fontWeight: isUnread ? FontWeight.w600 : null, ), ), ], ), ), + if (isUnread) + Padding( + padding: const EdgeInsets.only(right: Grid.xxs), + child: Container( + width: 8, + height: 8, + decoration: BoxDecoration( + color: context.colors.primary, + shape: BoxShape.circle, + ), + ), + ), if (!channel.isMember && !channel.isDm) Padding( padding: const EdgeInsets.only(right: Grid.xxs), diff --git a/mobile/lib/features/channels/channels_provider.dart b/mobile/lib/features/channels/channels_provider.dart index 744dc3a1cf..7461baf27f 100644 --- a/mobile/lib/features/channels/channels_provider.dart +++ b/mobile/lib/features/channels/channels_provider.dart @@ -3,6 +3,7 @@ import 'package:hooks_riverpod/hooks_riverpod.dart'; import '../../shared/relay/relay.dart'; import 'channel.dart'; +import 'read_state_provider.dart'; const _channelTypeOrder = {'stream': 0, 'forum': 1, 'dm': 2}; @@ -100,3 +101,42 @@ class ChannelsNotifier extends AsyncNotifier> { final channelsProvider = AsyncNotifierProvider>( ChannelsNotifier.new, ); + +/// Set of channel IDs that have unread messages based on read-state sync. +final unreadChannelIdsProvider = Provider>((ref) { + final readState = ref.watch(readStateSyncProvider); + final channelsAsync = ref.watch(channelsProvider); + final channels = channelsAsync.asData?.value ?? []; + + return channels + .where((ch) { + if (ch.lastMessageAt == null) return false; + final lastMessageSecs = + ch.lastMessageAt!.millisecondsSinceEpoch ~/ 1000; + final lastReadSecs = readState.mergedState[ch.id] ?? 0; + return lastMessageSecs > lastReadSecs; + }) + .map((ch) => ch.id) + .toSet(); +}); + +/// Seeds channels as read on first load so everything doesn't appear unread. +/// Watch this provider from the top-level widget (e.g. ChannelsPage) to +/// activate it. +final readStateSeedProvider = Provider((ref) { + final readState = ref.watch(readStateSyncProvider); + final channelsAsync = ref.watch(channelsProvider); + + // Only seed once read state has initialized and channels have loaded. + if (!readState.isInitialized) return; + final channels = channelsAsync.asData?.value; + if (channels == null) return; + + ref + .read(readStateSyncProvider.notifier) + .seedChannelsAsRead( + channels + .map((ch) => (id: ch.id, lastMessageAt: ch.lastMessageAt)) + .toList(), + ); +}); diff --git a/mobile/lib/features/channels/read_state_provider.dart b/mobile/lib/features/channels/read_state_provider.dart new file mode 100644 index 0000000000..119e048d5e --- /dev/null +++ b/mobile/lib/features/channels/read_state_provider.dart @@ -0,0 +1,432 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:math'; +import 'dart:typed_data'; +import 'package:flutter/widgets.dart'; +import 'package:hooks_riverpod/hooks_riverpod.dart'; +import 'package:nostr/nostr.dart' as nostr; +import 'package:shared_preferences/shared_preferences.dart'; + +import '../../shared/crypto/nip44.dart'; +import '../../shared/relay/app_lifecycle_provider.dart'; +import '../../shared/relay/nostr_models.dart'; +import '../../shared/relay/relay_provider.dart'; +import '../../shared/relay/relay_session.dart'; +import '../../shared/theme/theme_provider.dart'; +import 'read_state_sync.dart'; + +// Re-export so existing importers of ReadSyncState keep working. +export 'read_state_sync.dart' show ReadSyncState; + +// ── Notifier ───────────────────────────────────────────────────────────────── + +class ReadStateSyncNotifier extends Notifier { + // Mutable engine state (not in the immutable state object). + String _clientId = ''; + String _slotId = ''; + Map _ownBlob = {}; + Map _lastPublished = {}; + int _maxFetchedCreatedAt = 0; + Timer? _debounceTimer; + Timer? _republishTimer; + bool _isPublishing = false; + bool _needsRepublish = false; + bool _disposed = false; + void Function()? _unsubscribe; + Uint8List? _conversationKey; + String? _privkeyHex; + String? _pubkeyHex; + + SharedPreferences get _prefs => ref.read(savedPrefsProvider); + + @override + ReadSyncState build() { + // Watch relay config so notifier rebuilds on workspace switch, + // clearing stale keys/crypto state per identity. + ref.watch(relayConfigProvider); + + // Watch session status so we re-initialize (re-subscribe + catch-up + // fetch) after websocket reconnect (phone wake, network restore). + final sessionState = ref.watch(relaySessionProvider); + + // Flush on background, re-initialize on foreground resume. + ref.listen(appLifecycleProvider, (prev, next) { + if (next == AppLifecycleState.paused || + next == AppLifecycleState.detached) { + flushNow(); + } + }); + + ref.onDispose(_cleanup); + + // Read cached state synchronously for first paint. + // Full initialization happens asynchronously when connected. + if (sessionState.status == SessionStatus.connected) { + Future.microtask(_initialize); + } + + return const ReadSyncState(); + } + + Future _initialize() async { + _disposed = false; + + // Get keypair from relay config. + final config = ref.read(relayConfigProvider); + final nsec = config.nsec; + if (nsec == null || nsec.isEmpty) { + state = state.copyWith(isInitialized: true); + return; + } + + _privkeyHex = nostr.Nip19.decodePrivkey(nsec); + if (_privkeyHex == null || _privkeyHex!.isEmpty) { + state = state.copyWith(isInitialized: true); + return; + } + _pubkeyHex = nostr.Keychain(_privkeyHex!).public; + _conversationKey = getConversationKey(_privkeyHex!, _pubkeyHex!); + + // Read sync enabled preference (scoped by pubkey). + final syncEnabled = _prefs.getString(syncEnabledKey(_pubkeyHex!)) == 'true'; + + // Read cached read state from disk (scoped by pubkey). + final cached = _readCachedState(); + + state = state.copyWith(mergedState: cached, syncEnabled: syncEnabled); + + // Get or create persistent IDs (scoped by pubkey). + _clientId = _getOrCreateStorageValue(clientIdKey(_pubkeyHex!)); + _slotId = _getOrCreateStorageValue(slotIdKey(_pubkeyHex!)); + + // Fetch 7-day horizon of read-state events. + final since = DateTime.now().millisecondsSinceEpoch ~/ 1000 - 7 * 86400; + final session = ref.read(relaySessionProvider.notifier); + + List events; + try { + events = await session.fetchReadStateEvents(_pubkeyHex!, since: since); + } catch (_) { + state = state.copyWith(isInitialized: true); + return; + } + + if (_disposed) return; + + // Decode all blobs. + final decoded = []; + for (final event in events) { + final result = decryptAndValidateBlob(event, _conversationKey!); + if (result != null) decoded.add(result); + } + + if (_disposed) return; + + // Find our own blobs (matching client_id). + final ownBlobs = decoded.where((d) => d.clientId == _clientId).toList(); + + // Handle duplicate client_id: keep highest created_at, delete rest. + if (ownBlobs.length > 1) { + ownBlobs.sort((a, b) => b.event.createdAt.compareTo(a.event.createdAt)); + for (var i = 1; i < ownBlobs.length; i++) { + final stale = ownBlobs[i]; + final staleD = validateDTag(stale.event); + if (staleD != null) { + try { + _publishDeleteEvent(staleD); + } catch (_) { + // Best-effort cleanup. + } + } + } + } + + // Slot-ID conflict detection: another client owns our slot. + final ownDTag = 'read-state:$_slotId'; + final conflict = decoded.any((d) { + final dTag = validateDTag(d.event); + return dTag == ownDTag && d.clientId != _clientId; + }); + if (conflict) { + _slotId = randomHex(32); + _prefs.setString(slotIdKey(_pubkeyHex!), _slotId); + } + + // Track max created_at for clock-skew protection. + for (final d in decoded) { + _maxFetchedCreatedAt = max(_maxFetchedCreatedAt, d.event.createdAt); + } + + // Set own blob from primary (highest created_at). + if (ownBlobs.isNotEmpty) { + _ownBlob = Map.from(ownBlobs.first.contexts); + _lastPublished = Map.from(ownBlobs.first.contexts); + } + + // Merge ALL blobs (ours + other clients) + local cache. + var effective = {}; + for (final d in decoded) { + effective = mergeContexts(effective, d.contexts); + } + effective = mergeContexts(effective, cached); + effective = mergeContexts(effective, _ownBlob); + + if (!_disposed) { + // Functional merge with any marks that arrived during init. + state = state.copyWith( + mergedState: mergeContexts(state.mergedState, effective), + isInitialized: true, + ); + _writeCachedState(state.mergedState); + } + + // Subscribe to live read-state events. + if (_disposed) return; + try { + _unsubscribe = await session.subscribeToReadState( + _pubkeyHex!, + _handleIncomingEvent, + ); + } catch (_) { + // Non-fatal: sync still works via fetch. + } + } + + void _handleIncomingEvent(NostrEvent event) { + if (_disposed || _conversationKey == null) return; + + final decoded = decryptAndValidateBlob(event, _conversationKey!); + if (decoded == null) return; + + _maxFetchedCreatedAt = max(_maxFetchedCreatedAt, event.createdAt); + + if (decoded.clientId == _clientId) { + // Own blob echoed back from relay. + _ownBlob = mergeContexts(_ownBlob, decoded.contexts); + state = state.copyWith( + mergedState: mergeContexts(state.mergedState, decoded.contexts), + ); + _writeCachedState(state.mergedState); + return; + } + + // Another client's blob — merge into state. + final merged = mergeContexts(state.mergedState, decoded.contexts); + state = state.copyWith(mergedState: merged); + _writeCachedState(merged); + + // Check if we need to re-publish our blob (another client advanced us). + var needsRepublish = false; + for (final entry in decoded.contexts.entries) { + if (entry.value > (_lastPublished[entry.key] ?? 0)) { + needsRepublish = true; + break; + } + } + + if (needsRepublish && state.syncEnabled) { + _republishTimer?.cancel(); + _republishTimer = Timer(const Duration(seconds: 5), _publishOwnBlob); + } + } + + /// Mark a context (channel ID) as read at the given unix timestamp (seconds). + void markContextRead(String contextId, int timestamp) { + final existing = state.mergedState[contextId] ?? 0; + if (timestamp <= existing) return; + + final newMerged = Map.from(state.mergedState); + newMerged[contextId] = timestamp; + state = state.copyWith(mergedState: newMerged); + _writeCachedState(newMerged); + + // Update own blob. + final existingOwn = _ownBlob[contextId] ?? 0; + if (timestamp > existingOwn) { + _ownBlob = Map.from(_ownBlob)..[contextId] = timestamp; + } + + // Debounced publish. + _debounceTimer?.cancel(); + _debounceTimer = Timer(const Duration(seconds: 5), _publishOwnBlob); + } + + /// Flush pending state immediately (call on app background). + Future flushNow() async { + _debounceTimer?.cancel(); + _debounceTimer = null; + if (!contextsEqual(_ownBlob, _lastPublished)) { + await _publishOwnBlob(); + } + } + + /// Seed channels that have no read state yet. Call once after channels + /// load to prevent everything showing as unread on first install. + void seedChannelsAsRead( + List<({String id, DateTime? lastMessageAt})> channels, + ) { + var changed = false; + final newMerged = Map.from(state.mergedState); + for (final ch in channels) { + if (ch.lastMessageAt == null) continue; + if (newMerged.containsKey(ch.id)) continue; + final ts = ch.lastMessageAt!.millisecondsSinceEpoch ~/ 1000; + newMerged[ch.id] = ts; + _ownBlob = Map.from(_ownBlob)..[ch.id] = ts; + changed = true; + } + if (changed) { + state = state.copyWith(mergedState: newMerged); + _writeCachedState(newMerged); + } + } + + /// Toggle sync on/off. + void setSyncEnabled(bool enabled) { + if (_pubkeyHex != null) { + _prefs.setString(syncEnabledKey(_pubkeyHex!), enabled.toString()); + } + state = state.copyWith(syncEnabled: enabled); + } + + Future _publishOwnBlob({int retryDepth = 0}) async { + if (_pubkeyHex == null || !state.syncEnabled || !state.isInitialized) { + return; + } + + if (_isPublishing) { + _needsRepublish = true; + return; + } + _isPublishing = true; + + try { + final session = ref.read(relaySessionProvider.notifier); + final events = await session.fetchOwnReadStateBlob(_pubkeyHex!, _slotId); + + var fetchedOwnContexts = {}; + var localMaxCreatedAt = _maxFetchedCreatedAt; + + for (final event in events) { + localMaxCreatedAt = max(localMaxCreatedAt, event.createdAt); + final decoded = decryptAndValidateBlob(event, _conversationKey!); + if (decoded == null) continue; + + // Slot-ID conflict: another client claimed our slot. + if (decoded.clientId != _clientId) { + _slotId = randomHex(32); + _prefs.setString(slotIdKey(_pubkeyHex!), _slotId); + if (retryDepth >= 3) return; + _isPublishing = false; + return _publishOwnBlob(retryDepth: retryDepth + 1); + } + + fetchedOwnContexts = mergeContexts( + fetchedOwnContexts, + decoded.contexts, + ); + } + + _maxFetchedCreatedAt = localMaxCreatedAt; + + final merged = mergeContexts(fetchedOwnContexts, _ownBlob); + _ownBlob = merged; + + if (contextsEqual(merged, _lastPublished)) return; + + final payload = jsonEncode({ + 'v': 1, + 'client_id': _clientId, + 'contexts': merged, + }); + + final encrypted = nip44Encrypt(_conversationKey!, payload); + + // Clock skew: created_at = max(now, maxFetched + 1). + final now = DateTime.now().millisecondsSinceEpoch ~/ 1000; + final createdAt = max(now, _maxFetchedCreatedAt + 1); + + final signedEvent = nostr.Event.from( + kind: EventKind.readState, + content: encrypted, + tags: [ + ['d', 'read-state:$_slotId'], + ['t', 'read-state'], + ], + privkey: _privkeyHex!, + createdAt: createdAt, + ); + + // Convert nostr.Event to our NostrEvent for the session publish API. + final nostrEvent = NostrEvent.fromJson(signedEvent.toJson()); + + await session.publish(nostrEvent); + _lastPublished = Map.from(merged); + } catch (_) { + // Log error — non-fatal. + } finally { + _isPublishing = false; + if (_needsRepublish) { + _needsRepublish = false; + _publishOwnBlob(); + } + } + } + + void _publishDeleteEvent(String dTag) { + if (_privkeyHex == null || _pubkeyHex == null) return; + final signedEvent = nostr.Event.from( + kind: 5, + content: '', + tags: [ + ['a', '${EventKind.readState}:$_pubkeyHex:$dTag'], + ], + privkey: _privkeyHex!, + ); + try { + final nostrEvent = NostrEvent.fromJson(signedEvent.toJson()); + final session = ref.read(relaySessionProvider.notifier); + session.publish(nostrEvent); + } catch (_) { + // Best-effort. + } + } + + String _getOrCreateStorageValue(String key) { + final existing = _prefs.getString(key); + if (existing != null && existing.isNotEmpty) return existing; + final value = randomHex(32); + _prefs.setString(key, value); + return value; + } + + Map _readCachedState() { + if (_pubkeyHex == null) return {}; + return parseCachedReadState(_prefs.getString(cacheKey(_pubkeyHex!))); + } + + void _writeCachedState(Map s) { + if (_pubkeyHex == null) return; + _prefs.setString(cacheKey(_pubkeyHex!), jsonEncode(s)); + } + + void _cleanup() { + _disposed = true; + _debounceTimer?.cancel(); + _debounceTimer = null; + _republishTimer?.cancel(); + _republishTimer = null; + _unsubscribe?.call(); + _unsubscribe = null; + _ownBlob = {}; + _lastPublished = {}; + _maxFetchedCreatedAt = 0; + _isPublishing = false; + _needsRepublish = false; + } +} + +final readStateSyncProvider = + NotifierProvider( + ReadStateSyncNotifier.new, + ); diff --git a/mobile/lib/features/channels/read_state_sync.dart b/mobile/lib/features/channels/read_state_sync.dart new file mode 100644 index 0000000000..5daeea605b --- /dev/null +++ b/mobile/lib/features/channels/read_state_sync.dart @@ -0,0 +1,174 @@ +import 'dart:convert'; +import 'dart:math'; + +import 'package:flutter/foundation.dart'; + +import '../../shared/crypto/nip44.dart'; +import '../../shared/relay/nostr_models.dart'; + +// ── Storage keys (scoped by pubkey for multi-workspace isolation) ──────────── + +String clientIdKey(String pubkey) => 'sprout.read-state-client-id.v1.$pubkey'; +String slotIdKey(String pubkey) => 'sprout.read-state-slot-id.v1.$pubkey'; +String syncEnabledKey(String pubkey) => + 'sprout.read-state-sync-enabled.v1.$pubkey'; +String cacheKey(String pubkey) => 'sprout.channel-read-state.v1.$pubkey'; + +// ── State ──────────────────────────────────────────────────────────────────── + +@immutable +class ReadSyncState { + final Map mergedState; + final bool syncEnabled; + final bool isInitialized; + + const ReadSyncState({ + this.mergedState = const {}, + this.syncEnabled = false, + this.isInitialized = false, + }); + + ReadSyncState copyWith({ + Map? mergedState, + bool? syncEnabled, + bool? isInitialized, + }) => ReadSyncState( + mergedState: mergedState ?? this.mergedState, + syncEnabled: syncEnabled ?? this.syncEnabled, + isInitialized: isInitialized ?? this.isInitialized, + ); +} + +// ── Helpers ────────────────────────────────────────────────────────────────── + +String randomHex(int length) { + final rng = Random.secure(); + final bytes = List.generate(length ~/ 2, (_) => rng.nextInt(256)); + return bytes.map((b) => b.toRadixString(16).padLeft(2, '0')).join(); +} + +Map mergeContexts( + Map base, + Map incoming, +) { + final result = Map.from(base); + for (final entry in incoming.entries) { + result[entry.key] = max(result[entry.key] ?? 0, entry.value); + } + return result; +} + +bool contextsEqual(Map a, Map b) { + if (a.length != b.length) return false; + for (final key in a.keys) { + if (a[key] != b[key]) return false; + } + return true; +} + +// ── Validation per NIP-RS Content Validation ───────────────────────────────── + +String? validateDTag(NostrEvent event) { + final dTags = event.tags.where((t) => t.length >= 2 && t[0] == 'd').toList(); + if (dTags.length != 1) return null; + final dValue = dTags[0][1]; + if (!dValue.startsWith('read-state:')) return null; + final slotId = dValue.substring('read-state:'.length); + if (slotId.isEmpty || slotId.length > 64) return null; + for (var i = 0; i < slotId.length; i++) { + if (slotId.codeUnitAt(i) > 127) return null; + } + return dValue; +} + +bool validateTTag(NostrEvent event) { + final tTags = event.tags + .where((t) => t.length >= 2 && t[0] == 't' && t[1] == 'read-state') + .toList(); + return tTags.length == 1; +} + +/// Decoded read-state blob from a kind:30078 event. +class DecodedBlob { + final Map contexts; + final String clientId; + final NostrEvent event; + + DecodedBlob({ + required this.contexts, + required this.clientId, + required this.event, + }); +} + +DecodedBlob? decryptAndValidateBlob( + NostrEvent event, + Uint8List conversationKey, +) { + if (validateDTag(event) == null) return null; + if (!validateTTag(event)) return null; + + String plaintext; + try { + plaintext = nip44Decrypt(conversationKey, event.content); + } catch (_) { + return null; + } + + Map parsed; + try { + parsed = jsonDecode(plaintext) as Map; + } catch (_) { + return null; + } + + // Validate v + final v = parsed['v']; + if (v is! int || v != 1) return null; + + // Validate client_id + final clientId = parsed['client_id']; + if (clientId is! String || clientId.isEmpty || clientId.length > 64) { + return null; + } + + // Validate contexts + final rawContexts = parsed['contexts']; + if (rawContexts is! Map) return null; + if (rawContexts.length > 10000) return null; + + final contexts = {}; + for (final entry in rawContexts.entries) { + if (utf8.encode(entry.key).length > 256) continue; + final ts = entry.value; + if (ts is! int || ts < 0 || ts > 4294967295) continue; + contexts[entry.key] = ts; + } + + return DecodedBlob(contexts: contexts, clientId: clientId, event: event); +} + +/// Parse a cached read-state JSON string back into a `Map`. +/// Handles both the old ISO-string format and the new unix-seconds format. +Map parseCachedReadState(String? raw) { + if (raw == null || raw.isEmpty) return {}; + try { + final parsed = jsonDecode(raw); + if (parsed is! Map) return {}; + + final result = {}; + for (final entry in parsed.entries) { + if (entry.value is int) { + result[entry.key] = entry.value as int; + } else if (entry.value is String) { + final ms = DateTime.tryParse( + entry.value as String, + )?.millisecondsSinceEpoch; + if (ms != null) result[entry.key] = ms ~/ 1000; + } + } + return result; + } catch (_) { + return {}; + } +} diff --git a/mobile/lib/features/settings/settings_page.dart b/mobile/lib/features/settings/settings_page.dart index 9a5c3ba9b4..cfd276c45a 100644 --- a/mobile/lib/features/settings/settings_page.dart +++ b/mobile/lib/features/settings/settings_page.dart @@ -9,6 +9,7 @@ import '../../shared/relay/relay.dart'; import '../../shared/theme/theme.dart'; import '../../shared/widgets/frosted_app_bar.dart'; import '../../shared/widgets/frosted_scaffold.dart'; +import '../channels/read_state_provider.dart'; import 'theme_picker_page.dart'; class SettingsPage extends HookConsumerWidget { @@ -100,6 +101,13 @@ class SettingsPage extends HookConsumerWidget { const SizedBox(height: Grid.sm), + // Sync + Text('Sync', style: context.textTheme.titleMedium), + const SizedBox(height: Grid.twelve), + _ReadStateSyncToggle(), + + const SizedBox(height: Grid.sm), + // Appearance Text('Appearance', style: context.textTheme.titleMedium), const SizedBox(height: Grid.twelve), @@ -191,6 +199,34 @@ class SettingsPage extends HookConsumerWidget { } } +class _ReadStateSyncToggle extends ConsumerWidget { + @override + Widget build(BuildContext context, WidgetRef ref) { + final syncEnabled = ref.watch( + readStateSyncProvider.select((s) => s.syncEnabled), + ); + + return SwitchListTile( + secondary: const Icon(LucideIcons.refreshCw), + title: const Text('Cross-device read sync'), + subtitle: Text( + 'Sync read status across your devices', + style: context.textTheme.bodySmall?.copyWith( + color: context.colors.onSurfaceVariant, + ), + ), + value: syncEnabled, + onChanged: (value) { + ref.read(readStateSyncProvider.notifier).setSyncEnabled(value); + }, + shape: RoundedRectangleBorder( + borderRadius: BorderRadius.circular(Radii.md), + side: BorderSide(color: context.colors.outlineVariant), + ), + ); + } +} + class _AccentSwatch extends StatelessWidget { const _AccentSwatch({ required this.color, diff --git a/mobile/lib/shared/relay/nostr_models.dart b/mobile/lib/shared/relay/nostr_models.dart index d0f35f7565..662d44e997 100644 --- a/mobile/lib/shared/relay/nostr_models.dart +++ b/mobile/lib/shared/relay/nostr_models.dart @@ -13,6 +13,7 @@ abstract final class EventKind { static const streamMessageEdit = 40003; static const streamMessageDiff = 40008; static const systemMessage = 40099; + static const readState = 30078; static const forumPost = 45001; static const forumComment = 45003; @@ -23,6 +24,7 @@ abstract final class EventKind { reaction, // 7 streamMessage, // 9 40001, // legacy pre-migration stream messages + streamMessageV2, // 40002 streamMessageEdit, // 40003 streamMessageDiff, // 40008 systemMessage, // 40099 @@ -136,6 +138,7 @@ class NostrFilter { final int limit; final int? since; final int? until; + final List? authors; /// Tag filters, e.g. `{'#h': ['channel-id']}`. final Map> tags; @@ -145,6 +148,7 @@ class NostrFilter { this.limit = 100, this.since, this.until, + this.authors, this.tags = const {}, }); @@ -154,6 +158,7 @@ class NostrFilter { limit: limit, since: since, until: until, + authors: authors, tags: tags, ); @@ -161,6 +166,7 @@ class NostrFilter { final json = {'kinds': kinds, 'limit': limit}; if (since != null) json['since'] = since; if (until != null) json['until'] = until; + if (authors != null) json['authors'] = authors; for (final entry in tags.entries) { json[entry.key] = entry.value; } diff --git a/mobile/lib/shared/relay/relay_session.dart b/mobile/lib/shared/relay/relay_session.dart index 9dedb808e6..73aa80a3f0 100644 --- a/mobile/lib/shared/relay/relay_session.dart +++ b/mobile/lib/shared/relay/relay_session.dart @@ -55,33 +55,23 @@ class _PendingEvent { _PendingEvent({required this.completer, required this.timeout}); } -class _BufferedEvent { - final String subId; - final NostrEvent event; - - _BufferedEvent(this.subId, this.event); -} - // --------------------------------------------------------------------------- // RelaySession — the core session manager // --------------------------------------------------------------------------- -/// Manages websocket subscriptions, event batching, reconnection with replay, +/// Manages websocket subscriptions, reconnection with replay, /// and pending event tracking. Equivalent to the desktop's RelayClientSession. class RelaySessionNotifier extends Notifier { static const _baseReconnectDelayMs = 1000; static const _maxReconnectDelayMs = 30000; - static const _eventBatchMs = 16; static const _reconnectReplaySkewSeconds = 5; RelaySocket? _socket; final Map _historySubscriptions = {}; final Map _liveSubscriptions = {}; final Map _pendingEvents = {}; - final List<_BufferedEvent> _eventBuffer = []; final Set _recentEventIds = {}; Timer? _reconnectTimer; - Timer? _flushTimer; Timer? _backgroundGraceTimer; int _reconnectDelayMs = _baseReconnectDelayMs; int _subIdCounter = 0; @@ -199,6 +189,56 @@ class RelaySessionNotifier extends Notifier { _socket?.send(payload); } + /// Fetch all kind:30078 read-state events for [pubkey] within [since] horizon. + Future> fetchReadStateEvents( + String pubkey, { + required int since, + }) { + return fetchHistory( + NostrFilter( + kinds: const [EventKind.readState], + authors: [pubkey], + tags: const { + '#t': ['read-state'], + }, + since: since, + limit: 100, + ), + ); + } + + /// Fetch own blob by slot-ID (read-before-write, NIP-RS §Read-Before-Write). + Future> fetchOwnReadStateBlob(String pubkey, String slotId) { + return fetchHistory( + NostrFilter( + kinds: const [EventKind.readState], + authors: [pubkey], + tags: { + '#d': ['read-state:$slotId'], + }, + limit: 10, + ), + ); + } + + /// Subscribe to live read-state events for [pubkey]. + Future subscribeToReadState( + String pubkey, + void Function(NostrEvent) onEvent, + ) { + return subscribe( + NostrFilter( + kinds: const [EventKind.readState], + authors: [pubkey], + tags: const { + '#t': ['read-state'], + }, + limit: 0, + ), + onEvent, + ); + } + /// Force a reconnect (e.g., returning from background). Future reconnect() async { await _socket?.disconnect(); @@ -267,9 +307,6 @@ class RelaySessionNotifier extends Notifier { if (_disposed) return; _cancelAllHistory(error); _rejectAllPending(error); - _eventBuffer.clear(); - _flushTimer?.cancel(); - _flushTimer = null; _scheduleReconnect(); } @@ -333,23 +370,25 @@ class RelaySessionNotifier extends Notifier { final eventJson = data[2] as Map; final event = NostrEvent.fromJson(eventJson); - // History subscriptions accumulate immediately. final historySub = _historySubscriptions[subId]; if (historySub != null) { historySub.events.add(event); return; } - // Live subscriptions get batched. final liveSub = _liveSubscriptions[subId]; if (liveSub != null) { - // Track last seen timestamp for reconnect replay. if (liveSub.lastSeenCreatedAt == null || event.createdAt > liveSub.lastSeenCreatedAt!) { liveSub.lastSeenCreatedAt = event.createdAt; } - _eventBuffer.add(_BufferedEvent(subId, event)); - _scheduleFlush(); + + // Dedup: skip recently seen events. + if (_recentEventIds.contains(event.id)) return; + _recentEventIds.add(event.id); + if (_recentEventIds.length > 5000) _recentEventIds.clear(); + + liveSub.onEvent(event); } } @@ -411,41 +450,6 @@ class RelaySessionNotifier extends Notifier { } } - // ------------------------------------------------------------------------- - // Event batching (16ms flush interval) - // ------------------------------------------------------------------------- - - void _scheduleFlush() { - _flushTimer ??= Timer( - const Duration(milliseconds: _eventBatchMs), - _flushEventBuffer, - ); - } - - void _flushEventBuffer() { - _flushTimer = null; - if (_eventBuffer.isEmpty) return; - - final batch = List<_BufferedEvent>.from(_eventBuffer); - _eventBuffer.clear(); - - for (final buffered in batch) { - // Deduplication: skip events we've already seen recently. - if (_recentEventIds.contains(buffered.event.id)) continue; - _recentEventIds.add(buffered.event.id); - - // Cap the dedup set to prevent unbounded memory growth. - if (_recentEventIds.length > 5000) { - _recentEventIds.clear(); - } - - final sub = _liveSubscriptions[buffered.subId]; - if (sub != null) { - sub.onEvent(buffered.event); - } - } - } - // ------------------------------------------------------------------------- // Helpers // ------------------------------------------------------------------------- @@ -491,7 +495,6 @@ class RelaySessionNotifier extends Notifier { void _dispose() { _disposed = true; _reconnectTimer?.cancel(); - _flushTimer?.cancel(); _backgroundGraceTimer?.cancel(); _cancelAllHistory(null); _rejectAllPending(null);