From 63dfcc6821694b51b42d1d6cf69a3d201d00a14c Mon Sep 17 00:00:00 2001 From: Tom Brow Date: Sun, 18 Oct 2026 16:58:15 -0700 Subject: [PATCH] feat(push): notify authors about replies in their threads Signed-off-by: Tom Brow --- CONTEXT.md | 14 +- crates/buzz-db/src/store/event.rs | 21 ++ crates/buzz-db/src/store/thread.rs | 159 ++++++++++++++ crates/buzz-relay/src/api/bridge.rs | 195 +++++++++++++++++- crates/buzz-relay/src/handlers/push_lease.rs | 52 ++++- crates/buzz-relay/src/nip11.rs | 7 +- crates/buzz-relay/src/push_runtime.rs | 112 +++++++++- docs/authored-thread-reply-push.md | 74 +++++++ .../BuzzPushNotificationResolver.swift | 48 ++++- .../Sources/BuzzPushKit/PushLease.swift | 22 +- .../BuzzPushNotificationResolverTests.swift | 56 ++++- .../BuzzPushKitTests/PushLeaseTests.swift | 17 ++ mobile/lib/shared/push/dev_push_lease.dart | 5 + mobile/lib/shared/push/push_subscription.dart | 63 +++++- .../push/push_subscription_provider.dart | 17 ++ .../push/push_subscription_provider_test.dart | 33 +++ .../shared/push/push_subscription_test.dart | 31 +++ 17 files changed, 904 insertions(+), 22 deletions(-) create mode 100644 docs/authored-thread-reply-push.md diff --git a/CONTEXT.md b/CONTEXT.md index c9f74be0df..52afc344fa 100644 --- a/CONTEXT.md +++ b/CONTEXT.md @@ -1,6 +1,6 @@ # Buzz domain language -Shared terminology for Buzz product behavior. This glossary currently covers mobile [push suppression](#push-suppression); it is not a complete model of Buzz. +Shared terminology for Buzz product behavior. This glossary currently covers mobile notification eligibility and [push suppression](#push-suppression); it is not a complete model of Buzz. ## Notifications @@ -51,3 +51,15 @@ A [suppression activity signal](#suppression-activity-signal) that extends how l ### Inactivity cutoff The time of the last observed [desktop activity](#desktop-activity) plus the duration specified in the spec, when [desktop inactivity](#desktop-inactivity) begins. New desktop activity moves this cutoff forward. + +### Thread root + +The message identifying the outer thread containing a reply. A nested reply can have a different immediate parent while retaining the same thread root. + +### Authored-thread reply + +A message by another author in a thread whose root was authored by the notification recipient. This includes nested replies and does not imply an explicit follow or a prior mention. + +### Push eligibility + +Whether a message qualifies for a recipient's mobile push notification. Eligibility does not guarantee delivery or presentation, which remain subject to notification controls, suppression, and platform behavior. diff --git a/crates/buzz-db/src/store/event.rs b/crates/buzz-db/src/store/event.rs index 209f09e573..8b92d403d9 100644 --- a/crates/buzz-db/src/store/event.rs +++ b/crates/buzz-db/src/store/event.rs @@ -125,6 +125,8 @@ pub struct EventQuery { /// containment alone would also admit decoys like `["x","shared","true"]`. /// `event_visible_to_reader` remains the post-filter. pub shared_gated_reader: Option>, + /// Restrict replies to threads whose root has this author, before pagination. + pub thread_root_author: Option>, } impl EventQuery { @@ -158,6 +160,7 @@ impl EventQuery { channel_ids_include_global: true, max_limit: None, shared_gated_reader: None, + thread_root_author: None, } } } @@ -649,6 +652,15 @@ fn build_query_events_sql(q: &EventQuery) -> QueryBuilder { // Use unqualified column names when no join, qualified when joined. let col_prefix = if q.p_tag_hex.is_some() { "e." } else { "" }; + if let Some(author) = &q.thread_root_author { + let table = if q.p_tag_hex.is_some() { "e" } else { "events" }; + qb.push(" AND EXISTS (") + .push(crate::thread::root_author_sql(table)) + .push(" AND root.pubkey = ") + .push_bind(author.clone()) + .push(")"); + } + // Generic reads return only current artifact revisions, including delete // tombstones; explicit revision IDs also read earlier revisions. if q.ids.is_none() { @@ -1024,6 +1036,15 @@ pub(crate) async fn count_events_on(conn: &mut sqlx::PgConnection, q: &EventQuer }; let col_prefix = if q.p_tag_hex.is_some() { "e." } else { "" }; + if let Some(author) = &q.thread_root_author { + let table = if q.p_tag_hex.is_some() { "e" } else { "events" }; + qb.push(" AND EXISTS (") + .push(crate::thread::root_author_sql(table)) + .push(" AND root.pubkey = ") + .push_bind(author.clone()) + .push(")"); + } + // Generic reads return only current artifact revisions, including delete // tombstones; explicit revision IDs also read earlier revisions. if q.ids.is_none() { diff --git a/crates/buzz-db/src/store/thread.rs b/crates/buzz-db/src/store/thread.rs index 9a18554efe..ab8250441c 100644 --- a/crates/buzz-db/src/store/thread.rs +++ b/crates/buzz-db/src/store/thread.rs @@ -18,6 +18,62 @@ use crate::{ ReadSessionInner, RouteDecision, RoutePredicate, }; +/// Correlated root-author lookup shared by historical retrieval and push matching. +/// `event_table` is a trusted SQL alias, never client input. Exact timestamps keep +/// joins partition-aware; community and channel equality prevent cross-scope roots. +pub(crate) fn root_author_sql(event_table: &str) -> String { + format!( + "SELECT root.pubkey FROM thread_metadata tm \ + JOIN events root ON root.community_id = tm.community_id \ + AND root.id = tm.root_event_id \ + AND root.created_at = tm.root_event_created_at \ + AND root.channel_id = tm.channel_id \ + WHERE tm.community_id = {event_table}.community_id \ + AND tm.event_id = {event_table}.id \ + AND tm.event_created_at = {event_table}.created_at \ + AND tm.channel_id = {event_table}.channel_id \ + AND tm.depth > 0 AND root.deleted_at IS NULL" + ) +} + +impl Db { + /// Resolve root authors for a bounded batch of reply events in one community. + /// Missing/deleted roots and non-replies have no entry. + #[datastore_span(name = "thread_root_authors", system = "postgresql")] + pub async fn thread_root_authors( + &self, + community: CommunityId, + event_ids: &[Vec], + ) -> Result, Vec>> { + if event_ids.len() > 500 { + return Err(crate::error::DbError::InvalidData( + "too many reply IDs".into(), + )); + } + let mut query = sqlx::QueryBuilder::new("SELECT e.id, ("); + query + .push(root_author_sql("e")) + .push(") AS root_author FROM events e WHERE e.community_id = ") + .push_bind(community.as_uuid()) + .push(" AND e.id = ANY(") + .push_bind(event_ids) + .push(") AND e.deleted_at IS NULL"); + let mut connection = crate::observability::acquire_writer( + &self.pool, + crate::observability::WriterOperation::Maintenance, + ) + .await?; + let rows = query.build().fetch_all(&mut *connection).await?; + let mut authors = std::collections::HashMap::new(); + for row in rows { + if let Some(author) = row.try_get::>, _>("root_author")? { + authors.insert(row.try_get("id")?, author); + } + } + Ok(authors) + } +} + // -- Structs ------------------------------------------------------------------ /// A single reply within a thread, joined with event content. @@ -1248,6 +1304,109 @@ mod postgres_tests { .map(|channel| (channel, buzz_core::CommunityId::from_uuid(community_id))) } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn authored_thread_query_and_matcher_agree_for_nested_replies() { + let pool = setup_pool().await; + let db = Db::from_pool(pool.clone()); + let owner = Keys::generate(); + let other = Keys::generate(); + let (channel, community) = create_test_channel( + &pool, + "authored", + ChannelType::Stream, + ChannelVisibility::Open, + None, + owner.public_key().to_bytes().as_slice(), + None, + ) + .await + .expect("channel"); + let root = make_stream_event(&owner, "root"); + let reply = make_stream_event(&other, "direct"); + let nested = make_stream_event(&other, "nested"); + let unrelated = make_stream_event(&other, "unrelated root"); + for (event, parent, depth) in [ + (&root, None, 0), + (&reply, Some(&root), 1), + (&nested, Some(&reply), 2), + (&unrelated, None, 0), + ] { + insert_event_with_thread_metadata( + &pool, + community, + event, + Some(channel.id), + Some(ThreadMetadataParams { + event_id: event.id.as_bytes(), + event_created_at: event_created_at(event), + channel_id: channel.id, + parent_event_id: parent.map(|e| e.id.as_bytes().as_slice()), + parent_event_created_at: parent.map(event_created_at), + root_event_id: parent.map(|_| root.id.as_bytes().as_slice()), + root_event_created_at: parent.map(|_| event_created_at(&root)), + depth, + broadcast: false, + }), + ) + .await + .expect("insert event"); + } + let ids = [root.id, reply.id, nested.id, unrelated.id].map(|id| id.as_bytes().to_vec()); + let authors = db + .thread_root_authors(community, &ids) + .await + .expect("batch roots"); + assert_eq!(authors.len(), 2); + for event in [&reply, &nested] { + assert_eq!( + authors.get(event.id.as_bytes().as_slice()), + Some(&owner.public_key().to_bytes().to_vec()) + ); + } + let mut query = crate::EventQuery::for_community(community); + query.thread_root_author = Some(owner.public_key().to_bytes().to_vec()); + query.kinds = Some(vec![9]); + query.channel_ids = Some(vec![channel.id]); + let events = db.query_events(&query).await.expect("authored query"); + assert_eq!(events.len(), 2); + assert!(events + .iter() + .all(|event| authors.contains_key(event.event.id.as_bytes().as_slice()))); + query.thread_root_author = Some(other.public_key().to_bytes().to_vec()); + assert!(db + .query_events(&query) + .await + .expect("unrelated author") + .is_empty()); + query.thread_root_author = Some(owner.public_key().to_bytes().to_vec()); + query.channel_ids = Some(vec![]); + assert!(db.query_events(&query).await.expect("no access").is_empty()); + let foreign = CommunityId::from_uuid(make_test_community(&pool).await); + assert!(db + .thread_root_authors(foreign, &ids) + .await + .expect("foreign community") + .is_empty()); + sqlx::query("UPDATE events SET deleted_at = NOW() WHERE community_id = $1 AND id = $2") + .bind(community.as_uuid()) + .bind(root.id.as_bytes().as_slice()) + .execute(&pool) + .await + .expect("delete root"); + assert!(db + .thread_root_authors(community, &ids) + .await + .expect("deleted root") + .is_empty()); + query.channel_ids = Some(vec![channel.id]); + assert!(db + .query_events(&query) + .await + .expect("deleted root query") + .is_empty()); + } + #[tokio::test] #[ignore = "requires Postgres"] async fn resolve_deleted_roots_preserves_community_scope_for_colliding_ids() { diff --git a/crates/buzz-relay/src/api/bridge.rs b/crates/buzz-relay/src/api/bridge.rs index 9bc2940a81..2e05502e94 100644 --- a/crates/buzz-relay/src/api/bridge.rs +++ b/crates/buzz-relay/src/api/bridge.rs @@ -1409,7 +1409,14 @@ pub async fn query_events( ); } } - Ok(result.into_response()) + let mut response = result.into_response(); + if response.status().is_success() { + response.headers_mut().insert( + "x-buzz-query-extensions", + axum::http::HeaderValue::from_static("buzz-authored-thread-replies-v1"), + ); + } + Ok(response) } /// Filter execution for [`query_events`], run once NIP-98 auth succeeds. @@ -1442,6 +1449,52 @@ async fn query_events_authed( // depth_limit, feed_types) that nostr::Filter silently drops. let raw_filters: Vec = serde_json::from_slice(body) .map_err(|e| api_error(StatusCode::BAD_REQUEST, &format!("invalid filters: {e}")))?; + if raw_filters + .iter() + .any(|raw| raw.get("authored_thread_replies").is_some()) + && raw_filters + .iter() + .any(|raw| raw.get("authored_thread_replies").is_none()) + { + return Err(api_error( + StatusCode::BAD_REQUEST, + "authored_thread_replies cannot mix with other query modes", + )); + } + // Validate before any specialized query dispatcher can discard the extension. + for raw in &raw_filters { + if let Some(value) = raw.get("authored_thread_replies") { + let allowed = [ + "kinds", + "authored_thread_replies", + "since", + "until", + "limit", + "#h", + ]; + let valid = value == &Value::Bool(true) + && raw + .as_object() + .is_some_and(|object| object.keys().all(|key| allowed.contains(&key.as_str()))) + && raw + .get("kinds") + .and_then(Value::as_array) + .is_some_and(|kinds| { + !kinds.is_empty() + && kinds.iter().all(|kind| { + kind.as_u64().is_some_and(|kind| { + crate::handlers::push_lease::PUSH_KINDS.contains(&kind) + }) + }) + }); + if !valid { + return Err(api_error( + StatusCode::BAD_REQUEST, + "invalid authored_thread_replies query", + )); + } + } + } if let Some(result) = super::artifact::query(state, tenant, &pubkey, &raw_filters, false).await { return result; @@ -1850,6 +1903,9 @@ async fn query_events_authed( extract_channel_from_filter(filter), &accessible_channels, ); + if raw.get("authored_thread_replies") == Some(&Value::Bool(true)) { + query.thread_root_author = Some(pubkey_bytes.clone()); + } if let Some(channel) = extract_buzz_channel(raw) { query.custom_tag = Some(("buzz-channel".into(), channel.into())); } @@ -7859,6 +7915,131 @@ pub(crate) mod postgres_tests { ); } + #[tokio::test] + #[ignore = "requires Postgres"] + async fn authored_thread_bridge_uses_authenticated_root_author_before_limit() { + use buzz_db::channel::{ChannelType, ChannelVisibility}; + use uuid::Uuid; + let state = bridge_handler_test_state() + .await + .expect("test infrastructure"); + let host = format!("authored-{}.local", Uuid::new_v4().simple()); + let community = state + .db + .ensure_configured_community(&host) + .await + .expect("community") + .id; + let owner = Keys::generate(); + let other = Keys::generate(); + let (channel, _) = state + .db + .create_channel_with_id( + community, + Uuid::new_v4(), + "authored", + ChannelType::Stream, + ChannelVisibility::Open, + None, + owner.public_key().as_bytes(), + None, + ) + .await + .expect("channel"); + let now = nostr::Timestamp::now().as_secs(); + let root = EventBuilder::new(Kind::Custom(9), "owner root") + .tag(Tag::parse(["h", &channel.id.to_string()]).unwrap()) + .custom_created_at(nostr::Timestamp::from(now - 100)) + .sign_with_keys(&owner) + .unwrap(); + state + .db + .insert_event(community, &root, Some(channel.id)) + .await + .unwrap(); + let reply = EventBuilder::new(Kind::Custom(9), "reply") + .tag(Tag::parse(["h", &channel.id.to_string()]).unwrap()) + .tag(Tag::parse(["e", &root.id.to_hex(), "", "root"]).unwrap()) + .custom_created_at(nostr::Timestamp::from(now - 5)) + .sign_with_keys(&other) + .unwrap(); + state + .db + .insert_event_with_thread_metadata( + community, + &reply, + Some(channel.id), + Some(buzz_db::event::ThreadMetadataParams { + event_id: reply.id.as_bytes(), + event_created_at: chrono::DateTime::from_timestamp(now as i64 - 5, 0).unwrap(), + channel_id: channel.id, + parent_event_id: Some(root.id.as_bytes()), + parent_event_created_at: chrono::DateTime::from_timestamp(now as i64 - 100, 0), + root_event_id: Some(root.id.as_bytes()), + root_event_created_at: chrono::DateTime::from_timestamp(now as i64 - 100, 0), + depth: 1, + broadcast: false, + }), + ) + .await + .unwrap(); + // A newer non-reply must not consume the one-event result page. + let unrelated = EventBuilder::new(Kind::Custom(9), "newer unrelated") + .tag(Tag::parse(["h", &channel.id.to_string()]).unwrap()) + .sign_with_keys(&other) + .unwrap(); + state + .db + .insert_event(community, &unrelated, Some(channel.id)) + .await + .unwrap(); + let body = br#"[{"kinds":[9],"authored_thread_replies":true,"limit":1}]"#; + let (status, bytes) = + post_query(state.clone(), &host, &owner.public_key().to_hex(), body).await; + assert_eq!( + status, + StatusCode::OK, + "{}", + String::from_utf8_lossy(&bytes) + ); + let events: Vec = serde_json::from_slice(&bytes).unwrap(); + assert_eq!( + events.iter().map(|e| e.id).collect::>(), + vec![reply.id] + ); + let (status, bytes) = + post_query(state.clone(), &host, &other.public_key().to_hex(), body).await; + assert_eq!(status, StatusCode::OK); + assert_eq!( + serde_json::from_slice::>(&bytes) + .unwrap() + .len(), + 0 + ); + for filter in [ + serde_json::json!({"kinds": [9], "authored_thread_replies": false}), + serde_json::json!({"kinds": [9], "authored_thread_replies": true, "search": "reply"}), + serde_json::json!({"authored_thread_replies": true}), + ] { + let body = serde_json::to_vec(&vec![filter]).unwrap(); + assert_eq!( + post_query(state.clone(), &host, &owner.public_key().to_hex(), &body) + .await + .0, + StatusCode::BAD_REQUEST + ); + } + let body = serde_json::to_vec(&serde_json::json!([ + {"kinds": [9], "authored_thread_replies": true, "since": now} + ])) + .unwrap(); + let (status, bytes) = post_query(state, &host, &owner.public_key().to_hex(), &body).await; + assert_eq!(status, StatusCode::OK); + assert!(serde_json::from_slice::>(&bytes) + .unwrap() + .is_empty()); + } + /// Drive a single POST /query request through the router and return the /// HTTP status code + body bytes. async fn post_query( @@ -7885,6 +8066,18 @@ pub(crate) mod postgres_tests { .await .expect("router oneshot"); let status = resp.status(); + if status.is_success() + && serde_json::from_slice::>(body) + .unwrap() + .iter() + .any(|filter| filter.get("authored_thread_replies") == Some(&Value::Bool(true))) + { + assert_eq!( + resp.headers().get("x-buzz-query-extensions").unwrap(), + "buzz-authored-thread-replies-v1" + ); + } + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) .await .expect("read body"); diff --git a/crates/buzz-relay/src/handlers/push_lease.rs b/crates/buzz-relay/src/handlers/push_lease.rs index 4f5cea8aa5..7a6f017f8d 100644 --- a/crates/buzz-relay/src/handlers/push_lease.rs +++ b/crates/buzz-relay/src/handlers/push_lease.rs @@ -246,10 +246,42 @@ fn validate_filter( limits: &LeaseLimits<'_>, require_narrowing: bool, ) -> Result<(), String> { - const ALLOWED: &[&str] = &["kinds", "authors", "#p", "#h", "#e"]; + const ALLOWED: &[&str] = &[ + "kinds", + "authors", + "#p", + "#h", + "#e", + "authored_thread_replies", + "since", + ]; if let Some(key) = filter.keys().find(|key| !ALLOWED.contains(&key.as_str())) { return Err(format!("filter member not permitted: {key}")); } + let authored_replies = match filter.get("authored_thread_replies") { + None => false, + Some(Value::Bool(true)) if require_narrowing => true, + Some(_) => { + return Err( + "authored_thread_replies must be true and is not permitted in ignore filters" + .into(), + ) + } + }; + if authored_replies + && ["authors", "#p", "#e"] + .iter() + .any(|key| filter.contains_key(*key)) + { + return Err("authored_thread_replies only combines with kinds, #h and since".into()); + } + if let Some(since) = filter.get("since") { + if !authored_replies || since.as_u64().is_none() { + return Err( + "since requires authored_thread_replies and a nonnegative timestamp".into(), + ); + } + } let kinds = string_or_integer_array(filter, "kinds", limits.max_kinds, true)?; let kinds: Vec = kinds .iter() @@ -266,7 +298,7 @@ fn validate_filter( let p = optional_string_array(filter, "#p", limits.max_tag_values)?; let h = optional_string_array(filter, "#h", limits.max_h)?; let e = optional_string_array(filter, "#e", limits.max_tag_values)?; - if require_narrowing && authors.is_none() && p.is_none() && h.is_none() { + if require_narrowing && !authored_replies && authors.is_none() && p.is_none() && h.is_none() { return Err("lease filter not narrowed".into()); } for value in authors.into_iter().flatten() { @@ -640,6 +672,22 @@ mod tests { ); } + #[test] + fn authored_thread_filter_is_self_scoped_and_not_an_ignore_rule() { + let raw = serde_json::json!({"kinds": [9], "authored_thread_replies": true, "since": 1000}); + let filter = raw.as_object().unwrap(); + assert!(validate_filter(filter, &limits(), true).is_ok()); + assert!(validate_filter(filter, &limits(), false).is_err()); + for raw in [ + serde_json::json!({"kinds": [9], "authored_thread_replies": false}), + serde_json::json!({"kinds": [9], "authored_thread_replies": "true"}), + serde_json::json!({"kinds": [9], "authored_thread_replies": true, "since": -1}), + serde_json::json!({"kinds": [9], "since": 1000}), + ] { + assert!(validate_filter(raw.as_object().unwrap(), &limits(), true).is_err()); + } + } + fn limits<'a>() -> LeaseLimits<'a> { LeaseLimits { expected_origin: "o", diff --git a/crates/buzz-relay/src/nip11.rs b/crates/buzz-relay/src/nip11.rs index d61566efbb..b088f2e910 100644 --- a/crates/buzz-relay/src/nip11.rs +++ b/crates/buzz-relay/src/nip11.rs @@ -421,9 +421,10 @@ pub(crate) async fn nip11_document(state: &crate::state::AppState, raw_host: &st &state.relay_keypair, tenant_host.as_deref(), ) { - info.supported_extensions - .get_or_insert_default() - .push("buzz-push-v1".to_string()); + info.supported_extensions.get_or_insert_default().extend([ + "buzz-push-v1".to_string(), + "buzz-authored-thread-replies-v1".to_string(), + ]); info.push = Some(push); } info diff --git a/crates/buzz-relay/src/push_runtime.rs b/crates/buzz-relay/src/push_runtime.rs index c89284f948..754c080441 100644 --- a/crates/buzz-relay/src/push_runtime.rs +++ b/crates/buzz-relay/src/push_runtime.rs @@ -71,8 +71,9 @@ enum DeliveryResponse { /// Continuously claim accepted events in per-community batches and match /// them against active leases (T2b). One batch costs one claim statement, /// one event load, one lease scan, one membership scan, and one wake-enqueue -/// transaction — plus one complete/retry statement each — regardless of -/// batch size or how many (event, lease) pairs match. +/// transaction, plus one complete/retry statement each, regardless of +/// batch size or how many (event, lease) pairs match. Leases requesting authored +/// thread replies add one bounded root-author lookup for the whole batch. pub async fn run_matcher(state: Arc) { run_push_task( state.config.push_enabled, @@ -121,6 +122,7 @@ async fn matcher_loop(state: &AppState) { struct MatchContext { leases: Vec, memberships: std::collections::HashSet<(uuid::Uuid, Vec)>, + root_authors: std::collections::HashMap, Vec>, } async fn load_match_context( @@ -151,9 +153,29 @@ async fn load_match_context( .into_iter() .filter(|(channel, _)| !system.contains(channel)) .collect(); + let ids: Vec> = batch + .jobs + .iter() + .map(|job| job.event.event.id.as_bytes().to_vec()) + .collect(); + let needs_roots = leases.iter().any(|lease| { + lease.subscriptions.as_array().is_some_and(|subscriptions| { + subscriptions.iter().any(|sub| { + sub.get("filter") + .and_then(|filter| filter.get("authored_thread_replies")) + == Some(&serde_json::Value::Bool(true)) + }) + }) + }); + let root_authors = if needs_roots { + state.db.thread_root_authors(batch.community, &ids).await? + } else { + std::collections::HashMap::new() + }; Ok(MatchContext { leases, memberships, + root_authors, }) } @@ -298,6 +320,15 @@ fn match_job( let subscriptions: Vec = serde_json::from_value(lease.subscriptions.clone())?; let mut class: Option<&str> = None; for sub in &subscriptions { + if sub.filter.get("authored_thread_replies") == Some(&serde_json::Value::Bool(true)) + && (context + .root_authors + .get(job.event.event.id.as_bytes().as_slice()) + != Some(&lease.author) + || job.event.event.pubkey.to_bytes().as_slice() == lease.author.as_slice()) + { + continue; + } let filter: Filter = serde_json::from_value(serde_json::Value::Object(sub.filter.clone()))?; if !push_filter_authorized_for_event(&filter, &job.event.event, &author_hex) @@ -731,6 +762,83 @@ mod tests { use std::{future::IntoFuture, sync::Arc}; use tokio::sync::Mutex; + #[test] + fn authored_reply_matching_preserves_membership_mutes_self_and_dedup() { + let owner = nostr::Keys::generate(); + let sender = nostr::Keys::generate(); + let channel = uuid::Uuid::new_v4(); + let event = EventBuilder::new(Kind::Custom(9), "reply") + .tag(Tag::parse(["h", &channel.to_string()]).unwrap()) + .tag(Tag::public_key(owner.public_key())) + .sign_with_keys(&sender) + .unwrap(); + let owner_bytes = owner.public_key().to_bytes().to_vec(); + let mut context = MatchContext { + leases: vec![buzz_db::push::MatchLease { + author: owner_bytes.clone(), + installation_id: "installation".into(), + generation: 1, + expires_at: Utc::now().timestamp() + 3600, + subscriptions: serde_json::json!([ + {"filter": {"kinds": [9], "authored_thread_replies": true}, "class": "default"}, + {"filter": {"kinds": [9], "#p": [owner.public_key().to_hex()]}, "class": "default"} + ]), + }], + memberships: [(channel, owner_bytes.clone())].into(), + root_authors: [(event.id.as_bytes().to_vec(), owner_bytes.clone())].into(), + }; + let job = buzz_db::push::BatchedMatch { + event: buzz_core::StoredEvent::new(event, Some(channel)), + attempt: 1, + }; + assert_eq!( + match_job(&job, &context).unwrap().len(), + 1, + "overlap emits one wake" + ); + context.leases[0] + .subscriptions + .as_array_mut() + .unwrap() + .truncate(1); + assert_eq!(match_job(&job, &context).unwrap().len(), 1); + context.root_authors.clear(); + assert!( + match_job(&job, &context).unwrap().is_empty(), + "no root evidence" + ); + context + .root_authors + .insert(job.event.event.id.as_bytes().to_vec(), owner_bytes); + context.memberships.clear(); + assert!( + match_job(&job, &context).unwrap().is_empty(), + "membership is required" + ); + context + .memberships + .insert((channel, context.leases[0].author.clone())); + context.leases[0].subscriptions[0]["ignore"] = + serde_json::json!([{"kinds": [9], "#h": [channel.to_string()]}]); + assert!( + match_job(&job, &context).unwrap().is_empty(), + "channel mute" + ); + context.leases[0].subscriptions[0]["ignore"] = serde_json::json!([]); + context.leases[0].author = sender.public_key().to_bytes().to_vec(); + context + .memberships + .insert((channel, context.leases[0].author.clone())); + context.root_authors.insert( + job.event.event.id.as_bytes().to_vec(), + context.leases[0].author.clone(), + ); + assert!( + match_job(&job, &context).unwrap().is_empty(), + "self authored reply" + ); + } + #[test] fn gift_wrap_match_requires_self_p_filter_and_recipient() { let recipient = nostr::Keys::generate(); diff --git a/docs/authored-thread-reply-push.md b/docs/authored-thread-reply-push.md new file mode 100644 index 0000000000..7b66f940bd --- /dev/null +++ b/docs/authored-thread-reply-push.md @@ -0,0 +1,74 @@ +# Authored-thread reply push notifications + +Status: implemented; repository checks, physical delivery validation, and review +are in progress. + +## Eligibility + +A newly arriving message from another author qualifies when its thread root was +authored by the recipient. Nested replies retaining that root qualify, matching +desktop's root-authorship eligibility rule. Explicit follows, prior mentions, +and participation are not required. + +The rule applies to new replies in both existing and newly created threads, +even when the receiving phone has never opened the thread. Enabling the feature +does not generate notifications for historical replies. + +Existing direct-mention and DM behavior, access checks, notification controls, +channel mute behavior, and push suppression remain in force. Eligibility does +not guarantee OS delivery or lock-screen presentation. + +## Delivery and scope + +Preserve existing push grouping and coalescing. A burst of eligible replies does +not require a separate alert for each message; the current iOS resolver selects +the newest qualifying message after a wake. Overlapping eligibility reasons must +not introduce duplicate alerts. + +This increment changes push eligibility only. Inbox parity, desktop changes, +participation-based eligibility, prior-mention automatic subscriptions, explicit +follow/unfollow behavior, and cross-device follow synchronization are out of scope. + +## Acceptance + +- Another author's direct reply to the recipient's thread root is eligible. +- A nested reply retaining that root is eligible. +- A reply to an unrelated root does not qualify through this new rule, even if + the recipient authored an intermediate message or was previously mentioned. +- New replies to old threads qualify without prior phone viewing or fetching. +- Activation does not backfill historical notifications. +- Self-authored replies do not notify; overlapping mention/DM/thread eligibility + does not introduce duplicate alerts. +- Authorization, notification controls, channel mutes, and suppression are + exercised through real matching and native presentation paths. +- Tests distinguish eligibility from platform delivery and presentation. + +Implementation and PR validation, including physical-device testing, are authorized. +Do not merge, roll out to production, or include in the conference release. + +## Implementation contract + +The relay advertises `buzz-authored-thread-replies-v1` alongside its existing +push capability. Supporting mobile clients add a single self-scoped +`authored_thread_replies: true` lease filter, with a persisted `since` timestamp +for the first activation. They do not enumerate or synchronize thread follows. +Old relays retain the existing mention/DM policy. + +The authenticated `/query` bridge accepts the same predicate and derives the +recipient from the authenticated identity. Historical query filtering and batch +wake matching share the database root-author relation, including community, +channel, exact event timestamp, and live-root checks. The query applies that +predicate before pagination and retains existing access controls. + +The extension queries this relational policy separately from ordinary policies. +It trusts the authenticated relay response for the database relationship, while +still verifying event signatures, kind, activation time, ignore filters, +suppression, and recipient identity locally. An ordinary query response cannot +satisfy the relational policy. The extension also requires the relay response +header `x-buzz-query-extensions: buzz-authored-thread-replies-v1`, so an old +relay silently dropping the query field cannot broaden eligibility after a +rollback. Authored-thread queries cannot mix with other query modes. Existing +newest-message coalescing remains. + +Validation still requires the full repository gates, physical-device testing, +and review approval before this work is complete. diff --git a/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/BuzzPushNotificationResolver.swift b/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/BuzzPushNotificationResolver.swift index 1428d3fa5f..cbb40d0b21 100644 --- a/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/BuzzPushNotificationResolver.swift +++ b/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/BuzzPushNotificationResolver.swift @@ -135,20 +135,52 @@ public final class BuzzPushNotificationResolver: BuzzPushNotificationResolving { return key } + // Keep relational queries separate from ordinary filters. Their authorized + // response carries the relay's root-authorship decision; a response to an + // ordinary mention/DM query must never satisfy that relational policy. private func query( _ community: PushLeaseCommunity, completion: @escaping ((VerifiedNostrEvent, PushLeaseCommunity)?) -> Void + ) { + let groups = [false, true].map { authored in + community.policies.filter { ($0.filter.authoredThreadReplies == true) == authored } + }.filter { !$0.isEmpty } + let group = DispatchGroup() + let lock = NSLock() + var candidates: [(VerifiedNostrEvent, PushLeaseCommunity)] = [] + for policies in groups { + group.enter() + queryPolicies(community, policies: policies) { candidate in + if let candidate { + lock.lock() + candidates.append(candidate) + lock.unlock() + } + group.leave() + } + } + group.notify(queue: .global(qos: .userInitiated)) { + completion(candidates.max { + $0.0.createdAt == $1.0.createdAt ? $0.0.id > $1.0.id : $0.0.createdAt < $1.0.createdAt + }) + } + } + + private func queryPolicies( + _ community: PushLeaseCommunity, + policies: [PushResolutionPolicy], + completion: @escaping ((VerifiedNostrEvent, PushLeaseCommunity)?) -> Void ) { guard let privateKey = accountPrivateKey(for: community), community.pubkey?.isEmpty == false else { completion(nil) return } guard - !community.policies.isEmpty, + !policies.isEmpty, let relayURL = community.relayURL, let url = URL(string: "/query", relativeTo: relayURL), let body = try? JSONSerialization.data( - withJSONObject: community.policies.map { $0.filter.queryFilter(since: nil, limit: 10) } + withJSONObject: policies.map { $0.filter.queryFilter(since: nil, limit: 10) } ) else { completion(nil) @@ -176,11 +208,19 @@ public final class BuzzPushNotificationResolver: BuzzPushNotificationResolving { completion(nil) return } + let authoredQuery = policies.allSatisfy { $0.filter.authoredThreadReplies == true } + // Older relays silently discard unknown filter members. Require an + // explicit response capability before trusting relational selection. + if authoredQuery && response.value(forHTTPHeaderField: "x-buzz-query-extensions") != "buzz-authored-thread-replies-v1" { + completion(nil) + return + } let candidate = Self.newestMessage( events: events.filter { event in event.hasValidIDAndSignature() - && community.policies.contains { policy in - PushLeaseMatcher.matches(event: event, policy: policy) + && policies.contains { policy in + PushLeaseMatcher.matches(event: event, policy: policy, + authoredThreadQuery: authoredQuery) } }, community: community diff --git a/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushLease.swift b/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushLease.swift index 969520c2cd..479d97900c 100644 --- a/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushLease.swift +++ b/mobile/ios/BuzzPushKit/Sources/BuzzPushKit/PushLease.swift @@ -72,13 +72,17 @@ public struct PushLeaseFilter: Codable, Equatable, Sendable { public let pTags: [String]? public let hTags: [String]? public let eTags: [String]? + public let authoredThreadReplies: Bool? + public let since: Int? enum CodingKeys: String, CodingKey { case kinds + case since case authors case pTags = "#p" case hTags = "#h" case eTags = "#e" + case authoredThreadReplies = "authored_thread_replies" } public init( @@ -86,13 +90,17 @@ public struct PushLeaseFilter: Codable, Equatable, Sendable { authors: [String]? = nil, pTags: [String]? = nil, hTags: [String]? = nil, - eTags: [String]? = nil + eTags: [String]? = nil, + authoredThreadReplies: Bool? = nil, + since: Int? = nil ) { self.kinds = kinds self.authors = authors self.pTags = pTags self.hTags = hTags self.eTags = eTags + self.authoredThreadReplies = authoredThreadReplies + self.since = since } public func queryFilter(since: Int?, limit: Int) -> [String: Any] { @@ -101,11 +109,14 @@ public struct PushLeaseFilter: Codable, Equatable, Sendable { if let pTags { filter["#p"] = pTags } if let hTags { filter["#h"] = hTags } if let eTags { filter["#e"] = eTags } - if let since { filter["since"] = since } + if authoredThreadReplies == true { filter["authored_thread_replies"] = true } + if let since = [self.since, since].compactMap({ $0 }).max() { filter["since"] = since } return filter } - public func matches(_ event: VerifiedNostrEvent) -> Bool { + public func matches(_ event: VerifiedNostrEvent, authoredThreadQuery: Bool = false) -> Bool { + if authoredThreadReplies == true && !authoredThreadQuery { return false } + if let since, event.createdAt < since { return false } guard kinds.contains(event.kind) else { return false } if let authors, !authors.contains(event.pubkey.lowercased()) { return false } if let pTags, !event.hasAnyTag(named: "p", values: pTags) { return false } @@ -118,9 +129,10 @@ public struct PushLeaseFilter: Codable, Equatable, Sendable { public enum PushLeaseMatcher { public static func matches( event: VerifiedNostrEvent, - policy: PushResolutionPolicy + policy: PushResolutionPolicy, + authoredThreadQuery: Bool = false ) -> Bool { - guard policy.filter.matches(event) else { return false } + guard policy.filter.matches(event, authoredThreadQuery: authoredThreadQuery) else { return false } if policy.ignore.contains(where: { $0.matches(event) }) { return false } if let maximum = policy.suppress?.pTagsMax, event.tagCount(named: "p") > maximum diff --git a/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushNotificationResolverTests.swift b/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushNotificationResolverTests.swift index 366812bf91..374d65b7e0 100644 --- a/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushNotificationResolverTests.swift +++ b/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/BuzzPushNotificationResolverTests.swift @@ -32,6 +32,57 @@ final class BuzzPushNotificationResolverTests: XCTestCase { super.tearDown() } + func testAuthoredThreadQueryCannotMatchUnrelatedOrdinaryResults() throws { + let authored = PushResolutionPolicy(filter: PushLeaseFilter( + kinds: [9], authoredThreadReplies: true, since: Self.now - 10)) + let mention = PushResolutionPolicy(filter: PushLeaseFilter(kinds: [9], pTags: [Self.ownPubkey])) + let snapshot = PushLeaseCommunity(id: "community-id", name: "Community", + relayUrl: "https://relay.example", pubkey: Self.ownPubkey, + policies: [mention, authored], notificationSessionID: "test-session") + let reply = try Self.signedEvent(privateKey: Self.profilePrivateKey, + createdAt: Self.now, kind: 9, + tags: [["h", Self.channelID], ["e", String(repeating: "a", count: 64), "", "root"]], + content: "Authored thread reply") + let unrelated = try Self.signedEvent(privateKey: Self.profilePrivateKey, + createdAt: Self.now + 1, kind: 9, tags: [["h", Self.channelID]], content: "Unrelated") + URLProtocolStub.handler = { request in + var body = request.httpBody ?? Data() + if let stream = request.httpBodyStream { + stream.open() + defer { stream.close() } + var buffer = [UInt8](repeating: 0, count: 4096) + while stream.hasBytesAvailable { + let count = stream.read(&buffer, maxLength: buffer.count) + if count <= 0 { break } + body.append(contentsOf: buffer.prefix(count)) + } + } + let filters = try XCTUnwrap(JSONSerialization.jsonObject(with: body) as? [[String: Any]]) + let isAuthored = filters.first?["authored_thread_replies"] as? Bool == true + if isAuthored { XCTAssertEqual(filters.first?["since"] as? Int, Self.now - 10) } + return Self.response(request, status: 200, + data: try JSONEncoder().encode(isAuthored ? [reply] : [unrelated]), + headers: isAuthored ? ["x-buzz-query-extensions": "buzz-authored-thread-replies-v1"] : [:]) + } + let result = resolve(makeResolver(communitiesData: try snapshotData([snapshot]))) + XCTAssertEqual(result?.body, "Authored thread reply") + // Two eligibility requests and the existing presentation-enrichment request. + XCTAssertEqual(URLProtocolStub.requests.count, 3) + } + + func testAuthoredThreadQueryFailsClosedOnOlderRelay() throws { + let snapshot = PushLeaseCommunity(id: "community-id", name: "Community", + relayUrl: "https://relay.example", pubkey: Self.ownPubkey, + policies: [PushResolutionPolicy(filter: PushLeaseFilter(kinds: [9], authoredThreadReplies: true))], + notificationSessionID: "test-session") + let unrelated = try Self.signedEvent(privateKey: Self.profilePrivateKey, + createdAt: Self.now, kind: 9, tags: [["h", Self.channelID]], content: "Unrelated") + URLProtocolStub.handler = { request in + Self.response(request, status: 200, data: try JSONEncoder().encode([unrelated])) + } + XCTAssertNil(resolve(makeResolver(communitiesData: try snapshotData([snapshot])))) + } + func testResolveReturnsNilWhenCommunitiesDataIsMissing() { let result = resolve(makeResolver(communitiesData: nil)) @@ -1026,13 +1077,14 @@ final class BuzzPushNotificationResolverTests: XCTestCase { static func response( _ request: URLRequest, status: Int, - data: Data + data: Data, + headers: [String: String] = [:] ) -> (HTTPURLResponse, Data) { let response = HTTPURLResponse( url: request.url!, statusCode: status, httpVersion: "HTTP/1.1", - headerFields: ["Content-Type": "application/json"] + headerFields: ["Content-Type": "application/json"].merging(headers) { _, value in value } )! return (response, data) } diff --git a/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushLeaseTests.swift b/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushLeaseTests.swift index 096737d535..2a4df5db5c 100644 --- a/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushLeaseTests.swift +++ b/mobile/ios/BuzzPushKit/Tests/BuzzPushKitTests/PushLeaseTests.swift @@ -6,6 +6,23 @@ final class PushLeaseTests: XCTestCase { private let mine = String(repeating: "a", count: 64) private let other = String(repeating: "b", count: 64) + func testAuthoredRepliesRequireRelationalQueryAndPreserveRestrictions() throws { + let filter = PushLeaseFilter(kinds: [9], authoredThreadReplies: true, since: 900) + let policy = PushResolutionPolicy(filter: filter, + ignore: [PushLeaseFilter(kinds: [9], hTags: ["muted"])]) + let event = makeEvent(kind: 9) + XCTAssertFalse(PushLeaseMatcher.matches(event: event, policy: policy)) + XCTAssertTrue(PushLeaseMatcher.matches(event: event, policy: policy, authoredThreadQuery: true)) + XCTAssertFalse(PushLeaseMatcher.matches( + event: makeEvent(kind: 9, tags: [["h", "muted"]]), policy: policy, authoredThreadQuery: true)) + let future = PushResolutionPolicy(filter: PushLeaseFilter(kinds: [9], authoredThreadReplies: true, since: 1001)) + XCTAssertFalse(PushLeaseMatcher.matches(event: event, policy: future, authoredThreadQuery: true)) + XCTAssertEqual(filter.queryFilter(since: nil, limit: 10)["since"] as? Int, 900) + XCTAssertEqual(filter.queryFilter(since: 950, limit: 10)["since"] as? Int, 950) + XCTAssertEqual(filter.queryFilter(since: nil, limit: 10)["authored_thread_replies"] as? Bool, true) + XCTAssertEqual(try JSONDecoder().decode(PushLeaseFilter.self, from: JSONEncoder().encode(filter)), filter) + } + func testFilterBuildsQueryFromLeaseWithoutHardcodedKinds() { let filter = PushLeaseFilter( kinds: [7, 1059], diff --git a/mobile/lib/shared/push/dev_push_lease.dart b/mobile/lib/shared/push/dev_push_lease.dart index 26d69538c5..1e0f2f89f9 100644 --- a/mobile/lib/shared/push/dev_push_lease.dart +++ b/mobile/lib/shared/push/dev_push_lease.dart @@ -19,6 +19,7 @@ const _installationIdPattern = r'^[0-9a-f]{32}$'; class BuzzPushLeaseDescriptor { final String origin; + final bool authoredThreadReplies; final String executorKeyId; final String executorPubkey; final String transport; @@ -30,6 +31,7 @@ class BuzzPushLeaseDescriptor { const BuzzPushLeaseDescriptor({ required this.origin, + this.authoredThreadReplies = false, required this.executorKeyId, required this.executorPubkey, required this.transport, @@ -185,6 +187,9 @@ class BuzzPushLeaseDescriptor { return BuzzPushLeaseDescriptor( origin: origin, + authoredThreadReplies: extensions.contains( + 'buzz-authored-thread-replies-v1', + ), executorKeyId: currentKey['id'] as String, executorPubkey: currentKey['pubkey'] as String, transport: transport, diff --git a/mobile/lib/shared/push/push_subscription.dart b/mobile/lib/shared/push/push_subscription.dart index 41ef1808fe..90dee1b5e0 100644 --- a/mobile/lib/shared/push/push_subscription.dart +++ b/mobile/lib/shared/push/push_subscription.dart @@ -13,7 +13,15 @@ const buzzPushMaxIgnoreFilters = 8; const buzzPushHellthreadParticipantLimit = 20; const _supportedNotificationClasses = {'default'}; -const _filterKeys = {'kinds', 'authors', '#p', '#h', '#e'}; +const _filterKeys = { + 'kinds', + 'authors', + '#p', + '#h', + '#e', + 'authored_thread_replies', + 'since', +}; final _exactHexPattern = RegExp(r'^[0-9a-f]{64}$'); final _channelIdPattern = RegExp( r'^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$', @@ -27,6 +35,8 @@ class BuzzPushFilter { final List? pTags; final List? hTags; final List? eTags; + final bool authoredThreadReplies; + final int? since; BuzzPushFilter({ required Iterable kinds, @@ -34,6 +44,8 @@ class BuzzPushFilter { Iterable? pTags, Iterable? hTags, Iterable? eTags, + this.authoredThreadReplies = false, + this.since, }) : kinds = List.unmodifiable(kinds), authors = _optionalList(authors), pTags = _optionalList(pTags), @@ -48,11 +60,19 @@ class BuzzPushFilter { if (pTags != null) '#p': pTags, if (hTags != null) '#h': hTags, if (eTags != null) '#e': eTags, + if (authoredThreadReplies) 'authored_thread_replies': true, + if (since != null) 'since': since, }; factory BuzzPushFilter.fromJson(Map json) { _rejectUnknownKeys(json, _filterKeys, 'push filter'); + if (json.containsKey('authored_thread_replies') && + json['authored_thread_replies'] != true) { + throw const FormatException('authored_thread_replies must be true.'); + } return BuzzPushFilter( + authoredThreadReplies: json['authored_thread_replies'] == true, + since: json['since'] as int?, kinds: _intList(json, 'kinds'), authors: _optionalStringList(json, 'authors'), pTags: _optionalStringList(json, '#p'), @@ -62,6 +82,17 @@ class BuzzPushFilter { } void _validate() { + if (authoredThreadReplies && + (authors != null || pTags != null || eTags != null)) { + throw const FormatException( + 'Authored-thread eligibility only combines with kinds, channels and since.', + ); + } + if (since != null && (!authoredThreadReplies || since! < 0)) { + throw const FormatException( + 'since requires authored-thread eligibility and a nonnegative timestamp.', + ); + } if (kinds.isEmpty || kinds.any((kind) => !buzzPushEligibleKinds.contains(kind))) { throw const FormatException('Push filter contains invalid kinds.'); @@ -117,9 +148,15 @@ class BuzzPushSubscription { } if (filter.authors == null && filter.pTags == null && - filter.hTags == null) { + filter.hTags == null && + !filter.authoredThreadReplies) { throw const FormatException('Push subscription filter is not narrowed.'); } + if (this.ignore.any((filter) => filter.authoredThreadReplies)) { + throw const FormatException( + 'Thread eligibility is not an ignore filter.', + ); + } if (this.ignore.length > buzzPushMaxIgnoreFilters) { throw const FormatException( 'Push subscription has too many ignore filters.', @@ -449,6 +486,8 @@ List buildDesiredBuzzPushSubscriptions({ required String myPubkey, Iterable channelIds = const [], Iterable mutedChannelIds = const [], + bool authoredThreadReplies = false, + int? authoredThreadRepliesSince, }) { final normalizedPubkey = myPubkey.toLowerCase(); if (!_exactHexPattern.hasMatch(normalizedPubkey)) { @@ -489,6 +528,26 @@ List buildDesiredBuzzPushSubscriptions({ ), ]; + if (authoredThreadReplies) { + final muted = normalizedMuted.toList()..sort(); + subscriptions.add( + BuzzPushSubscription( + filter: BuzzPushFilter( + kinds: buzzPushEligibleKinds, + authoredThreadReplies: true, + since: authoredThreadRepliesSince, + ), + notificationClass: 'default', + ignore: [ + selfAuthored, + for (final chunk in _chunks(muted, buzzPushChannelChunkSize)) + BuzzPushFilter(kinds: buzzPushEligibleKinds, hTags: chunk), + ], + suppress: suppression, + ), + ); + } + final channels = normalizedChannelIds.difference(normalizedMuted).toList() ..sort(); for (final chunk in _chunks(channels, buzzPushChannelChunkSize)) { diff --git a/mobile/lib/shared/push/push_subscription_provider.dart b/mobile/lib/shared/push/push_subscription_provider.dart index 9d24e7cebc..1832c8a966 100644 --- a/mobile/lib/shared/push/push_subscription_provider.dart +++ b/mobile/lib/shared/push/push_subscription_provider.dart @@ -9,6 +9,7 @@ import '../community/community.dart'; import '../community/community_provider.dart'; import '../relay/relay_provider.dart'; import 'push_subscription.dart'; +import 'push_relay_capability_provider.dart'; /// Keeps the persisted desired lease and the App Group snapshot aligned with /// the active identity, joined channels, and mute state. This is desired client @@ -17,6 +18,7 @@ final pushSubscriptionSyncProvider = Provider((ref) { final active = ref.watch(activeCommunityProvider).value; final channels = ref.watch(channelsProvider).value; final mutes = ref.watch(channelMutesProvider); + final descriptor = ref.watch(currentRelayPushDescriptorProvider).value; if (active == null || !active.pushNotificationsEnabled || channels == null || @@ -26,6 +28,13 @@ final pushSubscriptionSyncProvider = Provider((ref) { final subscriptions = desiredBuzzPushSubscriptions( community: active, + // Retain an already discovered capability while offline so mute changes + // still update the native snapshot. A fresh old-relay descriptor disables it. + authoredThreadReplies: + descriptor?.authoredThreadReplies ?? + active.pushSubscriptionState.desired.any( + (s) => s.filter.authoredThreadReplies, + ), channels: channels, mutedChannelIds: [ for (final entry in mutes.store.channels.entries) @@ -44,11 +53,19 @@ List? desiredBuzzPushSubscriptions({ required Community community, required Iterable channels, required Iterable mutedChannelIds, + bool authoredThreadReplies = false, }) { final pubkey = community.pubkey ?? pubkeyFromNsec(community.nsec); if (pubkey == null || pubkey.isEmpty) return null; return buildDesiredBuzzPushSubscriptions( myPubkey: pubkey, + authoredThreadReplies: authoredThreadReplies, + authoredThreadRepliesSince: + community.pushSubscriptionState.desired + .where((subscription) => subscription.filter.authoredThreadReplies) + .map((subscription) => subscription.filter.since) + .firstOrNull ?? + DateTime.now().millisecondsSinceEpoch ~/ 1000, channelIds: [ // Activity surfaces channel-wide traffic only for joined DM channels. // Other message kinds enter the inbox through an exact #p mention or diff --git a/mobile/test/shared/push/push_subscription_provider_test.dart b/mobile/test/shared/push/push_subscription_provider_test.dart index d757106594..5ed39556a7 100644 --- a/mobile/test/shared/push/push_subscription_provider_test.dart +++ b/mobile/test/shared/push/push_subscription_provider_test.dart @@ -10,6 +10,39 @@ void main() { const archivedID = '123e4567-e89b-42d3-a456-426614174001'; const nonMemberID = '123e4567-e89b-42d3-a456-426614174002'; + test('authored reply activation time survives desired policy rebuilding', () { + final community = Community.create( + name: 'Team', + relayUrl: 'https://relay.example.com', + nsec: nostr.Keys.generate().nsec, + ); + final first = desiredBuzzPushSubscriptions( + community: community, + channels: [], + mutedChannelIds: [], + authoredThreadReplies: true, + )!; + final persisted = community.copyWith( + pushSubscriptionState: BuzzPushLeaseSubscriptionState.desired( + desired: first, + ), + ); + final rebuilt = desiredBuzzPushSubscriptions( + community: persisted, + channels: [], + mutedChannelIds: [activeID], + authoredThreadReplies: true, + )!; + final before = first + .singleWhere((s) => s.filter.authoredThreadReplies) + .filter; + final after = rebuilt + .singleWhere((s) => s.filter.authoredThreadReplies) + .filter; + expect(before.since, isNotNull); + expect(after.since, before.since); + }); + test( 'derives desired subscriptions from nsec, membership, and mute state', () { diff --git a/mobile/test/shared/push/push_subscription_test.dart b/mobile/test/shared/push/push_subscription_test.dart index 3ef21938e2..54e061fb3f 100644 --- a/mobile/test/shared/push/push_subscription_test.dart +++ b/mobile/test/shared/push/push_subscription_test.dart @@ -8,6 +8,37 @@ void main() { const channelA = '123e4567-e89b-42d3-a456-426614174000'; const channelB = '123e4567-e89b-42d3-a456-426614174001'; + test( + 'authored replies opt in without enumerating roots and honor all mutes', + () { + final subscriptions = buildDesiredBuzzPushSubscriptions( + myPubkey: me, + authoredThreadReplies: true, + authoredThreadRepliesSince: 1000, + mutedChannelIds: [channelA], + ); + final rule = subscriptions.singleWhere( + (s) => s.filter.authoredThreadReplies, + ); + expect(rule.filter.eTags, isNull); + expect(rule.filter.since, 1000); + expect(rule.ignore.first.authors, [me]); + expect(rule.ignore.last.hTags, [channelA]); + expect(rule.ignore.last.kinds, buzzPushEligibleKinds); + expect(rule.suppress!.pTagsMax, buzzPushHellthreadParticipantLimit); + expect( + BuzzPushSubscription.fromJson(rule.toJson()).toJson(), + rule.toJson(), + ); + expect( + buildDesiredBuzzPushSubscriptions( + myPubkey: me, + ).any((s) => s.filter.authoredThreadReplies), + isFalse, + ); + }, + ); + test( 'desired subscription state round-trips with an accepted-state seam', () {