mirror of
https://github.com/block/buzz.git
synced 2026-08-18 06:50:31 +02:00
perf(desktop): batch unread catch-up in Rust behind one IPC call
Problem: on wake or channel-list refresh the renderer issued one relay round-trip per not-yet-caught-up channel (N REQs, N awaits, ~175 lines of per-channel JS), and the two-pass notify classification ran per channel as promises resolved, so pass-one roots discovered in one channel were visible to another channel's pass two only if that channel's fetch happened to resolve later. Design: a single `unread_catch_up` Tauri command on the shared native relay session fetches all channels concurrently (Semaphore(8) finite REQs via fetch_events, JoinSet) and classifies globally in two strict passes — gather ALL history first, then classify — so cross-channel pass-one visibility no longer depends on resolution order. The Promise.all completion-order race is unrepresentable, not handled. Per-channel Success/Error results keep retry semantics: the error arm releases the channel's caught-up claim (its only identity) so the next effect run retries it. DiscoveredRoots deltas flow back to the renderer, which still owns membership stores and scope fencing — the command re-checks pubkey+relay after fetch; the renderer keeps isScopeLoaded/isCancelled for effect cleanup the command can't see. Both fences are load-bearing. ACTIVITY_LIMIT 100 is applied as a global pre-cap (newest-100-of-union ⊆ newest-100-of-batch, proven). Wire contract: `ChannelResult` is an internally tagged enum, and serde's `rename_all` on such an enum renames VARIANTS, not variant fields — variant fields need `rename_all_fields`. The initial cut emitted snake_case fields against the camelCase TS contract, breaking every catch-up while all gates ran green (the e2e bridge hand-wrote the intended shape; no test compared emitted bytes to declared types). Both IPC DTO surfaces in the pack are now pinned by whole-value wire tests asserting serde OUTPUT against the TS contract: catch-up (red-run proven at the defective bytes; variant-rename, nested-struct, and tag-key mutants all killed) and the persona catalog (rename_all mutants on both DTOs killed). The e2e bridge result is typed `UnreadCatchUpChannelResult[]`, so a drifting mock fails typecheck instead of silently certifying a stale contract. Interim cost, named where it is paid: membership stays renderer-owned until the native observed-unread store lands, so five root-id sets cross IPC on every catch-up — bounded at 5x1000 ids ~= 332 KiB worst case, linear fetch-body cost with no size cliff. Retiring this is the follow-up store's job, not an interim optimization. Co-authored-by: Tyler Longwell <tlongwell@squareup.com> Signed-off-by: Tyler Longwell <tlongwell@squareup.com>
This commit is contained in:
Perci
@@ -45,6 +45,7 @@ mod terminal_runtime;
|
||||
mod terminal_transport;
|
||||
#[cfg(target_os = "macos")]
|
||||
mod tray_menu;
|
||||
mod unread_catch_up;
|
||||
mod util;
|
||||
#[cfg(target_os = "linux")]
|
||||
pub mod webkit_rendering;
|
||||
@@ -716,6 +717,7 @@ pub fn run() {
|
||||
discover_backend_providers,
|
||||
probe_backend_provider,
|
||||
persona_catalog::fetch_persona_catalog,
|
||||
unread_catch_up::unread_catch_up,
|
||||
list_personas,
|
||||
create_persona,
|
||||
update_persona,
|
||||
|
||||
@@ -190,3 +190,46 @@ fn exact_tags_reject_duplicates_and_extra_fields() {
|
||||
assert_eq!(coordinate_tag(&extended, "d").as_deref(), Some("reviewer"));
|
||||
assert_eq!(exact_tag(&extended, "shared"), None);
|
||||
}
|
||||
|
||||
/// Pins the serialized DTO output against the renderer's catalog contract.
|
||||
/// The Tauri generic is only a TypeScript assertion; serde's bytes are the
|
||||
/// actual boundary, so populate every optional field and compare the value.
|
||||
#[test]
|
||||
fn serialized_catalog_matches_the_typescript_contract() {
|
||||
let publication = PersonaCatalogPublication {
|
||||
event_id: "ev1".into(),
|
||||
owner_pubkey: "owner".into(),
|
||||
source_persona_id: "persona-1".into(),
|
||||
created_at: 42,
|
||||
agent: CatalogAgentProjection {
|
||||
display_name: "Ada".into(),
|
||||
avatar_url: Some("https://example.com/a.png".into()),
|
||||
system_prompt: "be kind".into(),
|
||||
runtime: Some("acp".into()),
|
||||
model: Some("m1".into()),
|
||||
provider: Some("p1".into()),
|
||||
name_pool: vec!["Ada".into(), "Lin".into()],
|
||||
respond_to: Some("mentions".into()),
|
||||
parallelism: Some(2),
|
||||
},
|
||||
};
|
||||
let actual = serde_json::to_value(vec![publication]).unwrap();
|
||||
let expected = serde_json::json!([{
|
||||
"eventId": "ev1",
|
||||
"ownerPubkey": "owner",
|
||||
"sourcePersonaId": "persona-1",
|
||||
"createdAt": 42,
|
||||
"agent": {
|
||||
"displayName": "Ada",
|
||||
"avatarUrl": "https://example.com/a.png",
|
||||
"systemPrompt": "be kind",
|
||||
"runtime": "acp",
|
||||
"model": "m1",
|
||||
"provider": "p1",
|
||||
"namePool": ["Ada", "Lin"],
|
||||
"respondTo": "mentions",
|
||||
"parallelism": 2,
|
||||
},
|
||||
}]);
|
||||
assert_eq!(actual, expected);
|
||||
}
|
||||
|
||||
@@ -0,0 +1,647 @@
|
||||
//! Batched native unread catch-up.
|
||||
//!
|
||||
//! The renderer supplies its history-derived notification membership because
|
||||
//! those sets are still renderer-owned until the native observed-unread store
|
||||
//! lands. Rust performs every channel REQ over the shared authenticated session,
|
||||
//! then classifies the complete successful batch in two passes so a root learned
|
||||
//! anywhere in pass one is visible everywhere in pass two.
|
||||
|
||||
use std::{collections::HashSet, time::Duration};
|
||||
|
||||
use buzz_core_pkg::kind::{
|
||||
KIND_FORUM_COMMENT, KIND_FORUM_POST, KIND_HUDDLE_STARTED, KIND_STREAM_MESSAGE,
|
||||
KIND_STREAM_MESSAGE_V2,
|
||||
};
|
||||
use nostr::Event;
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tauri::State;
|
||||
use tokio::{sync::Semaphore, task::JoinSet};
|
||||
|
||||
use crate::{app_state::AppState, native_relay_client::NativeRelayClient};
|
||||
|
||||
const CATCH_UP_LIMIT: usize = 1_000;
|
||||
const ACTIVITY_LIMIT: usize = 100;
|
||||
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
|
||||
#[derive(Clone, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub(crate) struct UnreadCatchUpRequest {
|
||||
channels: Vec<CatchUpChannel>,
|
||||
self_pubkey: String,
|
||||
participated_root_ids: HashSet<String>,
|
||||
authored_root_ids: HashSet<String>,
|
||||
mentioned_root_ids: HashSet<String>,
|
||||
followed_root_ids: HashSet<String>,
|
||||
muted_root_ids: HashSet<String>,
|
||||
muted_channel_ids: HashSet<String>,
|
||||
}
|
||||
|
||||
#[derive(Clone, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct CatchUpChannel {
|
||||
id: String,
|
||||
#[serde(rename = "type")]
|
||||
channel_type: String,
|
||||
name: String,
|
||||
read_at: Option<u64>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub(crate) struct UnreadCatchUpResponse {
|
||||
channels: Vec<ChannelResult>,
|
||||
}
|
||||
|
||||
#[derive(Serialize)]
|
||||
#[serde(
|
||||
tag = "status",
|
||||
rename_all = "camelCase",
|
||||
rename_all_fields = "camelCase"
|
||||
)]
|
||||
enum ChannelResult {
|
||||
Success {
|
||||
channel_id: String,
|
||||
observed_events: Vec<ObservedUnreadEvent>,
|
||||
max_trigger: u64,
|
||||
activity_rows: Vec<ActivityRow>,
|
||||
discovered: DiscoveredRoots,
|
||||
},
|
||||
Error {
|
||||
channel_id: String,
|
||||
error: String,
|
||||
},
|
||||
}
|
||||
|
||||
#[derive(Clone, PartialEq, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct ObservedUnreadEvent {
|
||||
id: String,
|
||||
created_at: u64,
|
||||
root_id: Option<String>,
|
||||
high_priority: bool,
|
||||
counts_toward_badge: bool,
|
||||
counts_toward_app_badge: bool,
|
||||
}
|
||||
|
||||
#[derive(Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct ActivityRow {
|
||||
id: String,
|
||||
kind: u16,
|
||||
pubkey: String,
|
||||
content: String,
|
||||
created_at: u64,
|
||||
channel_id: String,
|
||||
channel_name: String,
|
||||
tags: Vec<Vec<String>>,
|
||||
}
|
||||
|
||||
#[derive(Default, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct DiscoveredRoots {
|
||||
participated: Vec<String>,
|
||||
authored: Vec<String>,
|
||||
mentioned: Vec<String>,
|
||||
}
|
||||
|
||||
struct FetchedChannel {
|
||||
order: usize,
|
||||
channel: CatchUpChannel,
|
||||
events: Vec<EventView>,
|
||||
}
|
||||
|
||||
#[derive(Clone)]
|
||||
struct EventView {
|
||||
id: String,
|
||||
kind: u16,
|
||||
pubkey: String,
|
||||
content: String,
|
||||
created_at: u64,
|
||||
tags: Vec<Vec<String>>,
|
||||
}
|
||||
|
||||
impl From<Event> for EventView {
|
||||
fn from(event: Event) -> Self {
|
||||
Self {
|
||||
id: event.id.to_hex(),
|
||||
kind: event.kind.as_u16(),
|
||||
pubkey: event.pubkey.to_hex(),
|
||||
content: event.content,
|
||||
created_at: event.created_at.as_secs(),
|
||||
tags: event
|
||||
.tags
|
||||
.iter()
|
||||
.map(|tag| tag.as_slice().to_vec())
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tauri::command]
|
||||
pub(crate) async fn unread_catch_up(
|
||||
request: UnreadCatchUpRequest,
|
||||
state: State<'_, AppState>,
|
||||
relay_client: State<'_, NativeRelayClient>,
|
||||
) -> Result<UnreadCatchUpResponse, String> {
|
||||
let keys = state.signing_keys()?;
|
||||
let owner = keys.public_key().to_hex();
|
||||
if !owner.eq_ignore_ascii_case(&request.self_pubkey) {
|
||||
return Err("unread catch-up identity does not match active scope".to_string());
|
||||
}
|
||||
let relay_url = crate::relay::relay_ws_url_with_override(&state);
|
||||
let session = relay_client.session(relay_url.clone(), keys).await;
|
||||
|
||||
let concurrency = std::sync::Arc::new(Semaphore::new(8));
|
||||
let mut pending = JoinSet::new();
|
||||
// One command replaces N renderer invokes while the shared session still
|
||||
// multiplexes bounded finite REQs on one authenticated socket.
|
||||
for (order, channel) in request.channels.iter().cloned().enumerate() {
|
||||
let permit = concurrency
|
||||
.clone()
|
||||
.acquire_owned()
|
||||
.await
|
||||
.map_err(|error| error.to_string())?;
|
||||
let session = session.clone();
|
||||
pending.spawn(async move {
|
||||
let _permit = permit;
|
||||
let kinds: &[u32] = if channel.channel_type == "dm" {
|
||||
&[
|
||||
KIND_STREAM_MESSAGE,
|
||||
KIND_STREAM_MESSAGE_V2,
|
||||
KIND_FORUM_POST,
|
||||
KIND_FORUM_COMMENT,
|
||||
KIND_HUDDLE_STARTED,
|
||||
]
|
||||
} else {
|
||||
&[
|
||||
KIND_STREAM_MESSAGE,
|
||||
KIND_STREAM_MESSAGE_V2,
|
||||
KIND_FORUM_POST,
|
||||
KIND_FORUM_COMMENT,
|
||||
]
|
||||
};
|
||||
let filter = serde_json::json!({
|
||||
"kinds": kinds,
|
||||
"#h": [channel.id],
|
||||
"since": channel.read_at.map_or(0, |value| value.saturating_add(1)),
|
||||
"limit": CATCH_UP_LIMIT,
|
||||
});
|
||||
let result = session.fetch_events(filter, REQUEST_TIMEOUT).await;
|
||||
(order, channel, result)
|
||||
});
|
||||
}
|
||||
|
||||
let mut fetched = Vec::new();
|
||||
let mut failures = Vec::new();
|
||||
while let Some(joined) = pending.join_next().await {
|
||||
let (order, channel, result) =
|
||||
joined.map_err(|error| format!("unread catch-up task failed: {error}"))?;
|
||||
match result {
|
||||
Ok(events) => fetched.push(FetchedChannel {
|
||||
order,
|
||||
channel,
|
||||
events: events
|
||||
.into_iter()
|
||||
.take(CATCH_UP_LIMIT)
|
||||
.map(EventView::from)
|
||||
.collect(),
|
||||
}),
|
||||
Err(error) => failures.push(ChannelResult::Error {
|
||||
channel_id: channel.id,
|
||||
error,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
fetched.sort_by_key(|item| item.order);
|
||||
|
||||
let current_keys = state.signing_keys()?;
|
||||
if current_keys.public_key().to_hex() != owner
|
||||
|| crate::relay::relay_ws_url_with_override(&state) != relay_url
|
||||
{
|
||||
return Err("unread catch-up scope changed while fetching".to_string());
|
||||
}
|
||||
|
||||
let mut channels = classify_batch(&request, fetched);
|
||||
channels.extend(failures);
|
||||
Ok(UnreadCatchUpResponse { channels })
|
||||
}
|
||||
|
||||
fn classify_batch(
|
||||
request: &UnreadCatchUpRequest,
|
||||
fetched: Vec<FetchedChannel>,
|
||||
) -> Vec<ChannelResult> {
|
||||
let self_pubkey = request.self_pubkey.to_lowercase();
|
||||
let mut participated = request.participated_root_ids.clone();
|
||||
let mut authored = request.authored_root_ids.clone();
|
||||
let mut mentioned = request.mentioned_root_ids.clone();
|
||||
|
||||
// Pass one is deliberately global, not per-channel: notification validity
|
||||
// depends on roots learned from history, while the command observes a batch.
|
||||
// Deltas remain attributed to the channel that first discovered each root.
|
||||
let mut discoveries = Vec::with_capacity(fetched.len());
|
||||
for item in &fetched {
|
||||
let mut discovered = DiscoveredRoots::default();
|
||||
for event in &item.events {
|
||||
if event.pubkey.eq_ignore_ascii_case(&self_pubkey) {
|
||||
let reference = thread_reference(&event.tags);
|
||||
if let Some(root_id) = reference.root_id {
|
||||
if participated.insert(root_id.clone()) {
|
||||
discovered.participated.push(root_id);
|
||||
}
|
||||
} else if authored.insert(event.id.clone()) {
|
||||
discovered.authored.push(event.id.clone());
|
||||
}
|
||||
} else if has_tag_value(&event.tags, "p", &self_pubkey) {
|
||||
if let Some(root_id) = thread_reference(&event.tags).root_id {
|
||||
if mentioned.insert(root_id.clone()) {
|
||||
discovered.mentioned.push(root_id);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
discoveries.push(discovered);
|
||||
}
|
||||
|
||||
let mut outputs = Vec::new();
|
||||
let mut all_activity = Vec::new();
|
||||
for (item, discovered) in fetched.into_iter().zip(discoveries) {
|
||||
let mut observed_events = Vec::new();
|
||||
let mut activity_rows = Vec::new();
|
||||
let mut max_trigger = 0;
|
||||
for event in item.events {
|
||||
if event.pubkey.eq_ignore_ascii_case(&self_pubkey)
|
||||
|| item
|
||||
.channel
|
||||
.read_at
|
||||
.is_some_and(|read_at| event.created_at <= read_at)
|
||||
|| !should_notify(&event, &self_pubkey, request, &participated, &authored)
|
||||
{
|
||||
continue;
|
||||
}
|
||||
let reference = thread_reference(&event.tags);
|
||||
let broadcast = has_exact_tag(&event.tags, "broadcast", "1");
|
||||
let threaded = reference.parent_id.is_some() && !broadcast;
|
||||
let high_priority = item.channel.channel_type == "dm"
|
||||
|| broadcast
|
||||
|| has_tag_value(&event.tags, "p", &self_pubkey);
|
||||
max_trigger = max_trigger.max(event.created_at);
|
||||
observed_events.push(ObservedUnreadEvent {
|
||||
id: event.id.clone(),
|
||||
created_at: event.created_at,
|
||||
root_id: if broadcast {
|
||||
None
|
||||
} else {
|
||||
reference.root_id.clone()
|
||||
},
|
||||
high_priority,
|
||||
counts_toward_badge: item.channel.channel_type == "dm" || threaded || high_priority,
|
||||
counts_toward_app_badge: item.channel.channel_type == "dm"
|
||||
|| (!threaded && high_priority),
|
||||
});
|
||||
if threaded {
|
||||
activity_rows.push(ActivityRow {
|
||||
id: event.id,
|
||||
kind: event.kind,
|
||||
pubkey: event.pubkey,
|
||||
content: event.content,
|
||||
created_at: event.created_at,
|
||||
channel_id: item.channel.id.clone(),
|
||||
channel_name: item.channel.name.clone(),
|
||||
tags: event.tags,
|
||||
});
|
||||
}
|
||||
}
|
||||
all_activity.extend(activity_rows.iter().cloned());
|
||||
outputs.push((
|
||||
item.channel.id,
|
||||
observed_events,
|
||||
max_trigger,
|
||||
activity_rows,
|
||||
discovered,
|
||||
));
|
||||
}
|
||||
|
||||
all_activity.sort_by_key(|row| row.created_at);
|
||||
let mut seen = HashSet::new();
|
||||
all_activity.retain(|row| seen.insert(row.id.clone()));
|
||||
if all_activity.len() > ACTIVITY_LIMIT {
|
||||
all_activity.drain(..all_activity.len() - ACTIVITY_LIMIT);
|
||||
}
|
||||
let allowed: HashSet<_> = all_activity.into_iter().map(|row| row.id).collect();
|
||||
|
||||
outputs
|
||||
.into_iter()
|
||||
.map(
|
||||
|(channel_id, observed_events, max_trigger, mut activity_rows, discovered)| {
|
||||
activity_rows.retain(|row| allowed.contains(&row.id));
|
||||
ChannelResult::Success {
|
||||
channel_id,
|
||||
observed_events,
|
||||
max_trigger,
|
||||
activity_rows,
|
||||
discovered,
|
||||
}
|
||||
},
|
||||
)
|
||||
.collect()
|
||||
}
|
||||
|
||||
struct ThreadReference {
|
||||
parent_id: Option<String>,
|
||||
root_id: Option<String>,
|
||||
}
|
||||
|
||||
fn thread_reference(tags: &[Vec<String>]) -> ThreadReference {
|
||||
let event_tags: Vec<_> = tags
|
||||
.iter()
|
||||
.filter(|tag| tag.first().is_some_and(|v| v == "e") && tag.get(1).is_some())
|
||||
.collect();
|
||||
let root = event_tags
|
||||
.iter()
|
||||
.find(|tag| tag.get(3).is_some_and(|v| v == "root"));
|
||||
let reply = event_tags
|
||||
.iter()
|
||||
.rev()
|
||||
.find(|tag| tag.get(3).is_some_and(|v| v == "reply"));
|
||||
let Some(reply) = reply else {
|
||||
return ThreadReference {
|
||||
parent_id: None,
|
||||
root_id: None,
|
||||
};
|
||||
};
|
||||
let parent_id = reply.get(1).cloned();
|
||||
ThreadReference {
|
||||
root_id: root
|
||||
.and_then(|tag| tag.get(1).cloned())
|
||||
.or_else(|| parent_id.clone()),
|
||||
parent_id,
|
||||
}
|
||||
}
|
||||
|
||||
fn should_notify(
|
||||
event: &EventView,
|
||||
self_pubkey: &str,
|
||||
request: &UnreadCatchUpRequest,
|
||||
participated: &HashSet<String>,
|
||||
authored: &HashSet<String>,
|
||||
) -> bool {
|
||||
if has_exact_tag(&event.tags, "broadcast", "1") || has_tag_value(&event.tags, "p", self_pubkey)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
let event_channel_id = event
|
||||
.tags
|
||||
.iter()
|
||||
.find(|tag| tag.first().is_some_and(|part| part == "h"))
|
||||
.and_then(|tag| tag.get(1));
|
||||
if event_channel_id.is_some_and(|id| request.muted_channel_ids.contains(id)) {
|
||||
return false;
|
||||
}
|
||||
let reference = thread_reference(&event.tags);
|
||||
if reference.parent_id.is_none() {
|
||||
return true;
|
||||
}
|
||||
let Some(root_id) = reference.root_id else {
|
||||
return false;
|
||||
};
|
||||
if request.muted_root_ids.contains(&root_id) {
|
||||
return false;
|
||||
}
|
||||
participated.contains(&root_id)
|
||||
|| request.followed_root_ids.contains(&root_id)
|
||||
|| authored.contains(&root_id)
|
||||
}
|
||||
|
||||
fn has_exact_tag(tags: &[Vec<String>], name: &str, value: &str) -> bool {
|
||||
tags.iter().any(|tag| {
|
||||
tag.first().is_some_and(|part| part == name) && tag.get(1).is_some_and(|part| part == value)
|
||||
})
|
||||
}
|
||||
|
||||
fn has_tag_value(tags: &[Vec<String>], name: &str, value: &str) -> bool {
|
||||
tags.iter().any(|tag| {
|
||||
tag.first().is_some_and(|part| part == name)
|
||||
&& tag
|
||||
.get(1)
|
||||
.is_some_and(|part| part.eq_ignore_ascii_case(value))
|
||||
})
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn event(id: &str, pubkey: &str, created_at: u64, tags: &[&[&str]]) -> EventView {
|
||||
EventView {
|
||||
id: id.into(),
|
||||
kind: 9,
|
||||
pubkey: pubkey.into(),
|
||||
content: id.into(),
|
||||
created_at,
|
||||
tags: tags
|
||||
.iter()
|
||||
.map(|tag| tag.iter().map(|part| (*part).to_string()).collect())
|
||||
.collect(),
|
||||
}
|
||||
}
|
||||
|
||||
fn request() -> UnreadCatchUpRequest {
|
||||
UnreadCatchUpRequest {
|
||||
channels: vec![],
|
||||
self_pubkey: "self".into(),
|
||||
participated_root_ids: HashSet::new(),
|
||||
authored_root_ids: HashSet::new(),
|
||||
mentioned_root_ids: HashSet::new(),
|
||||
followed_root_ids: HashSet::new(),
|
||||
muted_root_ids: HashSet::new(),
|
||||
muted_channel_ids: HashSet::new(),
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pass_one_history_changes_later_classification() {
|
||||
let req = request();
|
||||
let channel = CatchUpChannel {
|
||||
id: "ch".into(),
|
||||
channel_type: "stream".into(),
|
||||
name: "Ch".into(),
|
||||
read_at: Some(9),
|
||||
};
|
||||
let fetched = vec![FetchedChannel {
|
||||
order: 0,
|
||||
channel,
|
||||
events: vec![
|
||||
event(
|
||||
"self-reply",
|
||||
"self",
|
||||
10,
|
||||
&[&["e", "root", "", "reply"], &["h", "ch"]],
|
||||
),
|
||||
event(
|
||||
"external-reply",
|
||||
"other",
|
||||
11,
|
||||
&[&["e", "root", "", "reply"], &["h", "ch"]],
|
||||
),
|
||||
],
|
||||
}];
|
||||
let result = classify_batch(&req, fetched);
|
||||
let ChannelResult::Success {
|
||||
observed_events,
|
||||
discovered,
|
||||
..
|
||||
} = &result[0]
|
||||
else {
|
||||
panic!("expected success")
|
||||
};
|
||||
assert_eq!(
|
||||
observed_events
|
||||
.iter()
|
||||
.map(|event| event.id.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
["external-reply"]
|
||||
);
|
||||
assert_eq!(discovered.participated, ["root"]);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn same_second_marker_and_mutes_match_renderer_rules() {
|
||||
let mut req = request();
|
||||
req.muted_root_ids.insert("muted".into());
|
||||
let channel = CatchUpChannel {
|
||||
id: "ch".into(),
|
||||
channel_type: "stream".into(),
|
||||
name: "Ch".into(),
|
||||
read_at: Some(10),
|
||||
};
|
||||
let fetched = vec![FetchedChannel {
|
||||
order: 0,
|
||||
channel,
|
||||
events: vec![
|
||||
event("boundary", "other", 10, &[&["h", "ch"]]),
|
||||
event(
|
||||
"muted",
|
||||
"other",
|
||||
11,
|
||||
&[&["e", "muted", "", "reply"], &["h", "ch"]],
|
||||
),
|
||||
event(
|
||||
"broadcast",
|
||||
"other",
|
||||
12,
|
||||
&[&["broadcast", "1"], &["h", "ch"]],
|
||||
),
|
||||
],
|
||||
}];
|
||||
let result = classify_batch(&req, fetched);
|
||||
let ChannelResult::Success {
|
||||
observed_events,
|
||||
max_trigger,
|
||||
..
|
||||
} = &result[0]
|
||||
else {
|
||||
panic!("expected success")
|
||||
};
|
||||
assert_eq!(
|
||||
observed_events
|
||||
.iter()
|
||||
.map(|event| event.id.as_str())
|
||||
.collect::<Vec<_>>(),
|
||||
["broadcast"]
|
||||
);
|
||||
assert_eq!(*max_trigger, 12);
|
||||
}
|
||||
|
||||
/// Pins the SERIALIZED wire contract against `tauriUnreadCatchUp.ts`.
|
||||
///
|
||||
/// Asserts on serde's OUTPUT, not on `ChannelResult`: the renderer never
|
||||
/// sees the Rust type, it sees bytes, through an `invokeTauri<T>` cast
|
||||
/// that validates nothing. Every other test here inspects the enum before
|
||||
/// serialization and the e2e bridge hand-writes the intended shape, so
|
||||
/// without this nothing compares what Rust emits to what TypeScript
|
||||
/// declares.
|
||||
///
|
||||
/// Whole-value rather than a key list, deliberately: a key-set assertion
|
||||
/// passes a mutant that drops the variant rename and emits `"Success"`,
|
||||
/// which the renderer's `status === "error"` branch silently misreads.
|
||||
/// Failure here means the merge loop throws on the first success row and
|
||||
/// catch-up yields nothing, silently.
|
||||
#[test]
|
||||
fn serialized_response_matches_the_typescript_contract() {
|
||||
let channels = vec![
|
||||
ChannelResult::Success {
|
||||
channel_id: "ch".into(),
|
||||
observed_events: vec![ObservedUnreadEvent {
|
||||
id: "evt".into(),
|
||||
created_at: 11,
|
||||
root_id: Some("root".into()),
|
||||
high_priority: true,
|
||||
counts_toward_badge: true,
|
||||
counts_toward_app_badge: false,
|
||||
}],
|
||||
max_trigger: 11,
|
||||
activity_rows: vec![ActivityRow {
|
||||
id: "evt".into(),
|
||||
kind: 9,
|
||||
pubkey: "other".into(),
|
||||
content: "hi".into(),
|
||||
created_at: 11,
|
||||
channel_id: "ch".into(),
|
||||
channel_name: "Ch".into(),
|
||||
tags: vec![vec!["h".into(), "ch".into()]],
|
||||
}],
|
||||
discovered: DiscoveredRoots {
|
||||
participated: vec!["root".into()],
|
||||
authored: Vec::new(),
|
||||
mentioned: Vec::new(),
|
||||
},
|
||||
},
|
||||
ChannelResult::Error {
|
||||
channel_id: "ch-2".into(),
|
||||
error: "relay request timed out".into(),
|
||||
},
|
||||
];
|
||||
|
||||
let actual = serde_json::to_value(UnreadCatchUpResponse { channels }).unwrap();
|
||||
let expected = serde_json::json!({
|
||||
"channels": [
|
||||
{
|
||||
"status": "success",
|
||||
"channelId": "ch",
|
||||
"observedEvents": [{
|
||||
"id": "evt",
|
||||
"createdAt": 11,
|
||||
"rootId": "root",
|
||||
"highPriority": true,
|
||||
"countsTowardBadge": true,
|
||||
"countsTowardAppBadge": false,
|
||||
}],
|
||||
"maxTrigger": 11,
|
||||
"activityRows": [{
|
||||
"id": "evt",
|
||||
"kind": 9,
|
||||
"pubkey": "other",
|
||||
"content": "hi",
|
||||
"createdAt": 11,
|
||||
"channelId": "ch",
|
||||
"channelName": "Ch",
|
||||
"tags": [["h", "ch"]],
|
||||
}],
|
||||
"discovered": {
|
||||
"participated": ["root"],
|
||||
"authored": [],
|
||||
"mentioned": [],
|
||||
},
|
||||
},
|
||||
{
|
||||
"status": "error",
|
||||
"channelId": "ch-2",
|
||||
"error": "relay request timed out",
|
||||
},
|
||||
]
|
||||
});
|
||||
|
||||
assert_eq!(actual, expected);
|
||||
}
|
||||
}
|
||||
@@ -29,7 +29,6 @@ import {
|
||||
import {
|
||||
hasMentionForEvent,
|
||||
isHighPriorityEventForUser,
|
||||
shouldNotifyForEvent,
|
||||
} from "@/features/notifications/lib/shouldNotify";
|
||||
import type { RelayClient } from "@/shared/api/relayClientSession";
|
||||
import type { Channel, RelayEvent } from "@/shared/api/types";
|
||||
@@ -53,6 +52,7 @@ export {
|
||||
} from "@/features/channels/threadActivityStorage";
|
||||
import { useObservedUnreadPersistence } from "@/features/channels/useObservedUnreadPersistence";
|
||||
import { useThreadActivityPersistence } from "@/features/channels/useThreadActivityPersistence";
|
||||
import { unreadCatchUp } from "@/shared/api/tauriUnreadCatchUp";
|
||||
|
||||
type UseUnreadChannelsOptions = UseLiveChannelUpdatesOptions & {
|
||||
pubkey?: string;
|
||||
@@ -590,189 +590,114 @@ export function useUnreadChannels(
|
||||
const authoredSizeBefore = authoredRootIdsRef.current.size;
|
||||
const mentionedSizeBefore = mentionedRootIdsRef.current.size;
|
||||
|
||||
type CatchUpResult =
|
||||
| {
|
||||
channelId: string;
|
||||
ok: true;
|
||||
maxExternal: number;
|
||||
unreadEvents: ObservedUnreadEvent[];
|
||||
threadReplies: ThreadActivityItem[];
|
||||
}
|
||||
| { channelId: string; ok: false };
|
||||
|
||||
void Promise.all(
|
||||
toFetch.map(async (channelId): Promise<CatchUpResult> => {
|
||||
try {
|
||||
const readAt = getEffectiveTimestamp(channelId);
|
||||
const channel = channels.find((c) => c.id === channelId);
|
||||
// NIP-01 `since` is inclusive of `created_at >= since`. The +1
|
||||
// makes the relay-side filter strict-newer; the client-side
|
||||
// `> readAt` check below is the belt to the suspenders.
|
||||
const sinceParam = readAt === null ? 0 : readAt + 1;
|
||||
|
||||
const events = await relayClient.fetchEvents({
|
||||
kinds: [...channelCatchUpEventKinds(channel?.channelType)],
|
||||
"#h": [channelId],
|
||||
since: sinceParam,
|
||||
limit: CATCH_UP_LIMIT,
|
||||
});
|
||||
|
||||
// Pass 1: build participation from self-authored thread replies,
|
||||
// track self-authored top-level messages for author notifications,
|
||||
// and capture external mentions so their threads gate a badge.
|
||||
for (const event of events) {
|
||||
const isSelf =
|
||||
normalizedPubkey !== null &&
|
||||
event.pubkey.toLowerCase() === normalizedPubkey;
|
||||
if (isSelf) {
|
||||
const ref = getThreadReference(event.tags);
|
||||
if (ref.rootId !== null) {
|
||||
participatedRootIdsRef.current.add(ref.rootId);
|
||||
} else {
|
||||
authoredRootIdsRef.current.add(event.id);
|
||||
}
|
||||
} else {
|
||||
recordMentionedRoot(event);
|
||||
}
|
||||
}
|
||||
|
||||
if (normalizedPubkey !== null) {
|
||||
participationStore.write(
|
||||
normalizedPubkey,
|
||||
participatedRootIdsRef.current,
|
||||
);
|
||||
authoredStore.write(normalizedPubkey, authoredRootIdsRef.current);
|
||||
}
|
||||
|
||||
// Pass 2: compute maxExternal and collect thread reply activity,
|
||||
// applying the notification filter to both.
|
||||
let maxExternal = 0;
|
||||
const unreadEvents: ObservedUnreadEvent[] = [];
|
||||
const threadReplies: ThreadActivityItem[] = [];
|
||||
const chType = channel?.channelType;
|
||||
const chName = channel?.name ?? "";
|
||||
for (const event of events) {
|
||||
if (
|
||||
normalizedPubkey !== null &&
|
||||
event.pubkey.toLowerCase() === normalizedPubkey
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
if (readAt !== null && event.created_at <= readAt) continue;
|
||||
const eventChannelId =
|
||||
event.tags.find((t) => t[0] === "h")?.[1] ?? null;
|
||||
if (
|
||||
!shouldNotifyForEvent(event, normalizedPubkey ?? "", {
|
||||
participatedRootIds: participatedRootIdsRef.current,
|
||||
followedRootIds: options.followedRootIds ?? EMPTY_SET,
|
||||
authoredRootIds: authoredRootIdsRef.current,
|
||||
mutedRootIds: mutedRootIdsRef.current,
|
||||
mutedChannelIds: mutedChannelIdsRef.current,
|
||||
channelId: eventChannelId,
|
||||
})
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
const evtRef = getThreadReference(event.tags);
|
||||
const isThreadedReply =
|
||||
evtRef.parentId !== null && !isBroadcastReply(event.tags);
|
||||
if (event.created_at > maxExternal) {
|
||||
maxExternal = event.created_at;
|
||||
}
|
||||
const isHighPriority =
|
||||
chType === "dm" ||
|
||||
(normalizedPubkey !== null &&
|
||||
isHighPriorityEventForUser(event, normalizedPubkey));
|
||||
unreadEvents.push(
|
||||
makeObservedUnreadEvent({
|
||||
id: event.id,
|
||||
createdAt: event.created_at,
|
||||
rootId: resolveObservedUnreadRootId(event.tags),
|
||||
highPriority: isHighPriority,
|
||||
channelType: chType,
|
||||
isThreadedReply,
|
||||
}),
|
||||
);
|
||||
if (isThreadedReply) {
|
||||
threadReplies.push({
|
||||
id: event.id,
|
||||
kind: event.kind,
|
||||
pubkey: event.pubkey,
|
||||
content: event.content,
|
||||
createdAt: event.created_at,
|
||||
channelId,
|
||||
channelName: chName,
|
||||
tags: [...event.tags],
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
channelId,
|
||||
ok: true,
|
||||
maxExternal,
|
||||
unreadEvents,
|
||||
threadReplies,
|
||||
};
|
||||
} catch {
|
||||
// Transient relay failure for this channel — release the claim
|
||||
// so we retry on the next effect run instead of staying stuck
|
||||
// until identity reset.
|
||||
return { channelId, ok: false };
|
||||
}
|
||||
// Membership remains renderer-owned until E's native observed-unread store,
|
||||
// so unchanged sets cross IPC on every catch-up. The five 1,000-entry
|
||||
// stores bound that interim cost at roughly 332 KiB per request. Command
|
||||
// arguments use a fetch body, so this cost is linear with no size cliff.
|
||||
void unreadCatchUp({
|
||||
channels: toFetch.map((channelId) => {
|
||||
const channel = channels.find(
|
||||
(candidate) => candidate.id === channelId,
|
||||
);
|
||||
return {
|
||||
id: channelId,
|
||||
type: channel?.channelType ?? "stream",
|
||||
name: channel?.name ?? "",
|
||||
readAt: getEffectiveTimestamp(channelId),
|
||||
};
|
||||
}),
|
||||
).then((results) => {
|
||||
if (isCancelled) return;
|
||||
// Guard: don't merge catch-up results into a ref whose scope has drifted
|
||||
// (relay/pubkey changed while this async fetch was in flight). Use the
|
||||
// observed owner's loaded-scope predicate — one scope authority, not two.
|
||||
if (!observedPersistence.isScopeLoaded()) return;
|
||||
let didAdvance = false;
|
||||
const allThreadReplies: ThreadActivityItem[] = [];
|
||||
for (const result of results) {
|
||||
if (!result.ok) {
|
||||
caughtUpChannelsRef.current.delete(result.channelId);
|
||||
continue;
|
||||
}
|
||||
const { channelId, maxExternal, unreadEvents, threadReplies } = result;
|
||||
allThreadReplies.push(...threadReplies);
|
||||
if (unreadEvents.length > 0) {
|
||||
for (const event of unreadEvents) {
|
||||
recordUnreadEvent(channelId, event);
|
||||
selfPubkey: normalizedPubkey ?? "",
|
||||
participatedRootIds: [...participatedRootIdsRef.current],
|
||||
authoredRootIds: [...authoredRootIdsRef.current],
|
||||
mentionedRootIds: [...mentionedRootIdsRef.current],
|
||||
followedRootIds: [...(options.followedRootIds ?? EMPTY_SET)],
|
||||
mutedRootIds: [...mutedRootIdsRef.current],
|
||||
mutedChannelIds: [...mutedChannelIdsRef.current],
|
||||
})
|
||||
.then(({ channels: results }) => {
|
||||
if (isCancelled) return;
|
||||
// The command rejects if relay/pubkey scope changes in flight. Keep the
|
||||
// renderer fence too: it also covers effect cleanup before merge.
|
||||
if (!observedPersistence.isScopeLoaded()) return;
|
||||
|
||||
let didAdvance = false;
|
||||
let didDiscover = false;
|
||||
const allThreadReplies: ThreadActivityItem[] = [];
|
||||
for (const result of results) {
|
||||
if (result.status === "error") {
|
||||
// The error arm carries only this identity; releasing its claim is
|
||||
// what lets a failed channel retry on the next effect run.
|
||||
caughtUpChannelsRef.current.delete(result.channelId);
|
||||
continue;
|
||||
}
|
||||
didAdvance = true;
|
||||
}
|
||||
if (maxExternal > 0) {
|
||||
const readAtNow = getEffectiveTimestamp(channelId) ?? 0;
|
||||
if (maxExternal > readAtNow) {
|
||||
const current = latestByChannelRef.current.get(channelId) ?? 0;
|
||||
if (maxExternal > current) {
|
||||
latestByChannelRef.current.set(channelId, maxExternal);
|
||||
for (const rootId of result.discovered.participated) {
|
||||
const before = participatedRootIdsRef.current.size;
|
||||
participatedRootIdsRef.current.add(rootId);
|
||||
didDiscover ||= participatedRootIdsRef.current.size !== before;
|
||||
}
|
||||
for (const rootId of result.discovered.authored) {
|
||||
const before = authoredRootIdsRef.current.size;
|
||||
authoredRootIdsRef.current.add(rootId);
|
||||
didDiscover ||= authoredRootIdsRef.current.size !== before;
|
||||
}
|
||||
for (const rootId of result.discovered.mentioned) {
|
||||
const before = mentionedRootIdsRef.current.size;
|
||||
mentionedRootIdsRef.current.add(rootId);
|
||||
didDiscover ||= mentionedRootIdsRef.current.size !== before;
|
||||
}
|
||||
allThreadReplies.push(...result.activityRows);
|
||||
for (const event of result.observedEvents) {
|
||||
recordUnreadEvent(result.channelId, event);
|
||||
didAdvance = true;
|
||||
}
|
||||
if (
|
||||
result.maxTrigger > (getEffectiveTimestamp(result.channelId) ?? 0)
|
||||
) {
|
||||
const current =
|
||||
latestByChannelRef.current.get(result.channelId) ?? 0;
|
||||
if (result.maxTrigger > current) {
|
||||
latestByChannelRef.current.set(
|
||||
result.channelId,
|
||||
result.maxTrigger,
|
||||
);
|
||||
didAdvance = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
if (allThreadReplies.length > 0) {
|
||||
const added = addThreadActivityItems(
|
||||
threadActivityRef.current,
|
||||
allThreadReplies,
|
||||
);
|
||||
if (added.didAdd) {
|
||||
threadActivityRef.current = added.items;
|
||||
activityPersistence.schedule(currentActivityScope);
|
||||
didAdvance = true;
|
||||
|
||||
if (normalizedPubkey !== null && didDiscover) {
|
||||
participationStore.write(
|
||||
normalizedPubkey,
|
||||
participatedRootIdsRef.current,
|
||||
);
|
||||
authoredStore.write(normalizedPubkey, authoredRootIdsRef.current);
|
||||
mentionedStore.write(normalizedPubkey, mentionedRootIdsRef.current);
|
||||
}
|
||||
}
|
||||
if (didAdvance) bumpLatestVersion();
|
||||
if (
|
||||
participatedRootIdsRef.current.size !== participatedSizeBefore ||
|
||||
authoredRootIdsRef.current.size !== authoredSizeBefore ||
|
||||
mentionedRootIdsRef.current.size !== mentionedSizeBefore
|
||||
) {
|
||||
bumpMembershipVersion();
|
||||
}
|
||||
});
|
||||
if (allThreadReplies.length > 0) {
|
||||
const added = addThreadActivityItems(
|
||||
threadActivityRef.current,
|
||||
allThreadReplies,
|
||||
);
|
||||
if (added.didAdd) {
|
||||
threadActivityRef.current = added.items;
|
||||
activityPersistence.schedule(currentActivityScope);
|
||||
didAdvance = true;
|
||||
}
|
||||
}
|
||||
if (didAdvance) bumpLatestVersion();
|
||||
if (
|
||||
didDiscover ||
|
||||
participatedRootIdsRef.current.size !== participatedSizeBefore ||
|
||||
authoredRootIdsRef.current.size !== authoredSizeBefore ||
|
||||
mentionedRootIdsRef.current.size !== mentionedSizeBefore
|
||||
) {
|
||||
bumpMembershipVersion();
|
||||
}
|
||||
})
|
||||
.catch(() => {
|
||||
if (isCancelled) return;
|
||||
for (const id of toFetch) caughtUpChannelsRef.current.delete(id);
|
||||
});
|
||||
|
||||
return () => {
|
||||
isCancelled = true;
|
||||
|
||||
@@ -0,0 +1,42 @@
|
||||
import { invokeTauri } from "@/shared/api/tauri";
|
||||
import type { ObservedUnreadEvent } from "@/features/channels/unreadChannelCounts";
|
||||
import type { ThreadActivityItem } from "@/features/channels/threadActivityStorage";
|
||||
|
||||
export type UnreadCatchUpChannel = {
|
||||
id: string;
|
||||
type: string;
|
||||
name: string;
|
||||
readAt: number | null;
|
||||
};
|
||||
|
||||
export type UnreadCatchUpRequest = {
|
||||
channels: UnreadCatchUpChannel[];
|
||||
selfPubkey: string;
|
||||
participatedRootIds: string[];
|
||||
authoredRootIds: string[];
|
||||
mentionedRootIds: string[];
|
||||
followedRootIds: string[];
|
||||
mutedRootIds: string[];
|
||||
mutedChannelIds: string[];
|
||||
};
|
||||
|
||||
export type UnreadCatchUpChannelResult =
|
||||
| {
|
||||
status: "success";
|
||||
channelId: string;
|
||||
observedEvents: ObservedUnreadEvent[];
|
||||
maxTrigger: number;
|
||||
activityRows: ThreadActivityItem[];
|
||||
discovered: {
|
||||
participated: string[];
|
||||
authored: string[];
|
||||
mentioned: string[];
|
||||
};
|
||||
}
|
||||
| { status: "error"; channelId: string; error: string };
|
||||
|
||||
export function unreadCatchUp(
|
||||
request: UnreadCatchUpRequest,
|
||||
): Promise<{ channels: UnreadCatchUpChannelResult[] }> {
|
||||
return invokeTauri("unread_catch_up", { request });
|
||||
}
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
handleDeleteCustomHarness,
|
||||
} from "./e2eBridgeCustomHarnesses.ts";
|
||||
|
||||
import type { UnreadCatchUpChannelResult } from "@/shared/api/tauriUnreadCatchUp";
|
||||
import { relayClient } from "@/shared/api/relayClient";
|
||||
import { activateRateLimit } from "@/shared/api/relayRateLimitGate";
|
||||
import { resolveAgentParallelism } from "@/features/agents/lib/agentParallelism";
|
||||
@@ -13376,6 +13377,23 @@ export function maybeInstallE2eTauriMocks() {
|
||||
case "start_archive_sync":
|
||||
case "stop_archive_sync":
|
||||
return null;
|
||||
case "unread_catch_up": {
|
||||
const request = payload as {
|
||||
request: { channels: Array<{ id: string }> };
|
||||
};
|
||||
const results: UnreadCatchUpChannelResult[] =
|
||||
request.request.channels.map((channel) => ({
|
||||
status: "success",
|
||||
channelId: channel.id,
|
||||
observedEvents: [],
|
||||
maxTrigger: 0,
|
||||
activityRows: [],
|
||||
discovered: { participated: [], authored: [], mentioned: [] },
|
||||
}));
|
||||
// Keep this mock aligned with the complete Rust serde shape pinned by
|
||||
// `serialized_response_matches_the_typescript_contract`.
|
||||
return { channels: results };
|
||||
}
|
||||
case "agent_metric_archive_default_enabled":
|
||||
return activeConfig?.mock?.agentMetricArchiveDefaultEnabled ?? true;
|
||||
case "set_prevent_sleep_active":
|
||||
|
||||
Reference in New Issue
Block a user