Skip to main content

vector_core/community/
realtime.rs

1//! Realtime Community (Concord) subscription, routing, dispatch, and control-follow.
2//!
3//! The per-event cryptographic engine ([`inbound::process_incoming`]) already lives in core;
4//! this module is the realtime *plumbing* around it — the relay subscription, the pseudonym→channel
5//! route maps, the dispatch of typed [`inbound::IncomingEvent`]s to an [`InboundEventHandler`], and
6//! the control/rekey realtime-follow. It is consumed by [`crate::VectorCore::listen`] so headless
7//! clients (SDK, CLI, bots) get realtime Community delivery through the same handler as DMs.
8
9use std::collections::{HashMap, HashSet};
10use std::sync::{LazyLock, Mutex as StdMutex};
11use std::sync::Arc;
12use std::time::Duration;
13
14use nostr_sdk::prelude::*;
15use tokio::sync::Mutex;
16
17use crate::community::{derive, inbound, roster, service, Channel, CommunityId, Epoch};
18use crate::community::transport::LiveTransport;
19use crate::event_handler::InboundEventHandler;
20use crate::state::SessionGuard;
21use crate::stored_event::event_kind;
22use crate::ClientRelayExt;
23
24/// Realtime channel-follow retry budget: a re-founding publishes the channel rekey under the NEW
25/// root right after the base rekey, so the base-3303-triggered follow can race its propagation.
26const CHANNEL_FOLLOW_MAX_ATTEMPTS: usize = 5;
27const CHANNEL_FOLLOW_BACKOFF_MS: u64 = 700;
28
29/// Current Community subscription id — single subscription scoped to the epoch pseudonyms of
30/// every channel we hold; refreshed on join/leave/rekey.
31static COMMUNITY_SUB_ID: LazyLock<Mutex<Option<SubscriptionId>>> = LazyLock::new(|| Mutex::new(None));
32
33/// Sorted pseudonym set of the CURRENTLY live Community subscription. Lets `refresh_subscription` skip
34/// a redundant unsubscribe+resubscribe when nothing changed (e.g. a metadata/avatar edit, which fires
35/// a control re-fold but doesn't alter the pseudonym set). The relay pool auto-re-applies the existing
36/// subscription on every reconnect, so leaving it in place is strictly more reliable than rebuilding it
37/// (a rebuild that lands mid-reconnect silently fails to register and kills realtime delivery).
38static COMMUNITY_SUB_SET: LazyLock<Mutex<Vec<String>>> = LazyLock::new(|| Mutex::new(Vec::new()));
39
40/// Pool-wide `subscribe` of the community filter, opened alongside the targeted `subscribe_to`. On
41/// Android the targeted sub registers but never streams; the pool-wide one does (it rides the same
42/// auto-managed path the DM sub uses). On desktop the targeted sub streams. We keep BOTH so every
43/// platform gets live delivery; `process_incoming` dedups by outer-event id, so an event seen on both
44/// folds exactly once.
45static COMMUNITY_POOLWIDE_SUB_ID: LazyLock<Mutex<Option<SubscriptionId>>> = LazyLock::new(|| Mutex::new(None));
46
47/// `z` pseudonym (hex) → the Channel it belongs to. Lets the notification loop open an arriving
48/// event against the right channel key (and read its `banned` set for the inbound drop-filter).
49static COMMUNITY_ROUTES: LazyLock<Mutex<HashMap<String, Channel>>> =
50    LazyLock::new(|| Mutex::new(HashMap::new()));
51
52/// Control-plane `z` pseudonym (hex) → community id (hex). Routes a CONTROL edition (3308) or a
53/// base/channel REKEY coordinate (3303) to a realtime control refresh of that community.
54static CONTROL_ROUTES: LazyLock<Mutex<HashMap<String, String>>> =
55    LazyLock::new(|| Mutex::new(HashMap::new()));
56
57/// Per-community in-flight guard for [`refresh_control`] — one concurrent fold per community.
58static REFRESH_CONTROL_INFLIGHT: LazyLock<StdMutex<HashSet<String>>> =
59    LazyLock::new(|| StdMutex::new(HashSet::new()));
60
61/// The subscription id of the live Community subscription, if any. Lets a notification loop test
62/// whether an arriving event belongs to the Community sub.
63pub async fn subscription_id() -> Option<SubscriptionId> {
64    COMMUNITY_SUB_ID.lock().await.clone()
65}
66
67/// The pool-wide subscription id (the path that streams on Android, where the
68/// targeted `subscribe_to` registers but never delivers). The listen loop must OR
69/// this with [`subscription_id`], else events arriving under the pool-wide sub
70/// match no branch and are silently dropped.
71pub async fn poolwide_subscription_id() -> Option<SubscriptionId> {
72    COMMUNITY_POOLWIDE_SUB_ID.lock().await.clone()
73}
74
75/// Clear all realtime route/subscription state. Call from `swap_session` so a swapped-in account
76/// can't read the prior account's channel keys / banned sets.
77pub async fn clear() {
78    *COMMUNITY_SUB_ID.lock().await = None;
79    *COMMUNITY_POOLWIDE_SUB_ID.lock().await = None;
80    COMMUNITY_SUB_SET.lock().await.clear();
81    COMMUNITY_ROUTES.lock().await.clear();
82    CONTROL_ROUTES.lock().await.clear();
83    REFRESH_CONTROL_INFLIGHT.lock().unwrap_or_else(|e| e.into_inner()).clear();
84    super::migration::clear_drive_inflight();
85}
86
87/// Rebuild ONLY the in-memory route maps from the persisted communities, WITHOUT touching the live
88/// relay subscription. Refreshes each cached `Channel` (incl. its `banned` set) so a received
89/// ban/unban takes live effect without a full resubscribe. Returns the pseudonyms + relays a caller
90/// also resubscribing needs.
91pub async fn rebuild_routes() -> (Vec<String>, HashSet<String>) {
92    let mut routes: HashMap<String, Channel> = HashMap::new();
93    let mut control_routes: HashMap<String, String> = HashMap::new();
94    let mut pseudonyms: Vec<String> = Vec::new();
95    let mut relays: HashSet<String> = HashSet::new();
96
97    if let Ok(ids) = crate::db::community::list_community_ids() {
98        for id in ids {
99            if let Ok(Some(community)) = crate::db::community::load_community(&id) {
100                for r in &community.relays {
101                    relays.insert(r.clone());
102                }
103                for ch in &community.channels {
104                    // Subscribe to EVERY held epoch pseudonym (not just the head) so a straggler posting
105                    // under a retained older epoch still arrives in realtime; the inbound router opens each
106                    // against the channel's full keyset.
107                    for (epoch, key) in ch.read_epoch_keys() {
108                        let pseudonym = derive::channel_pseudonym(&key, &ch.id, epoch).to_hex();
109                        pseudonyms.push(pseudonym.clone());
110                        routes.insert(pseudonym, ch.clone());
111                    }
112                    // The NEXT channel-rekey coordinate (3303) under the CURRENT server root — follow a
113                    // channel rotation event-driven the instant it lands.
114                    let next_chan = derive::rekey_pseudonym(
115                        &community.server_root_key, &ch.id, Epoch(ch.epoch.0 + 1),
116                    ).to_hex();
117                    pseudonyms.push(next_chan.clone());
118                    control_routes.insert(next_chan, community.id.to_hex());
119                }
120                // Control-plane pseudonym at the current server-root epoch (banlist/roles/metadata/invites).
121                let ctrl = roster::control_pseudonym(
122                    &community.server_root_key, &community.id, community.server_root_epoch,
123                );
124                pseudonyms.push(ctrl.clone());
125                control_routes.insert(ctrl, community.id.to_hex());
126                // The NEXT base re-founding coordinate (3303) — follow a privatize / private-ban in realtime.
127                let next_base = derive::base_rekey_pseudonym(
128                    &community.server_root_key, &community.id,
129                    Epoch(community.server_root_epoch.0 + 1),
130                ).to_hex();
131                pseudonyms.push(next_base.clone());
132                control_routes.insert(next_base, community.id.to_hex());
133            }
134        }
135    }
136
137    *COMMUNITY_ROUTES.lock().await = routes;
138    *CONTROL_ROUTES.lock().await = control_routes;
139    (pseudonyms, relays)
140}
141
142/// The coalesced control-probe coordinate set: every held v1 community's
143/// control pseudonym + next-base-rekey + next-channel-rekey `z` coordinate,
144/// mapped back to its community id. A single change-detector fetch over these
145/// (kinds 3308/3303) tells the boot sweep which communities actually have new
146/// control/rotation editions, so the rest skip the per-community catch-up chain.
147/// Returns `(coordinates, coordinate → community_id_hex, relay_union)`.
148pub async fn control_probe_coordinates() -> (Vec<String>, HashMap<String, String>, HashSet<String>) {
149    let mut coords: Vec<String> = Vec::new();
150    let mut map: HashMap<String, String> = HashMap::new();
151    let mut relays: HashSet<String> = HashSet::new();
152    let Ok(ids) = crate::db::community::list_community_ids() else {
153        return (coords, map, relays);
154    };
155    for id in ids {
156        // v1 only — v2 control/rekey planes are author-addressed giftwraps (a
157        // separate, auth-gated probe; see B1b).
158        if matches!(
159            crate::db::community::community_protocol(&id).ok().flatten(),
160            Some(crate::community::ConcordProtocol::V2)
161        ) {
162            continue;
163        }
164        let Ok(Some(community)) = crate::db::community::load_community(&id) else {
165            continue;
166        };
167        let cid = community.id.to_hex();
168        for r in &community.relays {
169            relays.insert(r.clone());
170        }
171        let mut add = |coord: String| {
172            if !map.contains_key(&coord) {
173                coords.push(coord.clone());
174                map.insert(coord, cid.clone());
175            }
176        };
177        add(roster::control_pseudonym(&community.server_root_key, &community.id, community.server_root_epoch));
178        add(derive::base_rekey_pseudonym(&community.server_root_key, &community.id, Epoch(community.server_root_epoch.0 + 1)).to_hex());
179        for ch in &community.channels {
180            add(derive::rekey_pseudonym(&community.server_root_key, &ch.id, Epoch(ch.epoch.0 + 1)).to_hex());
181        }
182    }
183    (coords, map, relays)
184}
185
186/// (Re)build the Community subscription: rebuild the route maps, then open TWO subscriptions over the
187/// same filter — a targeted `subscribe_to` (streams on desktop) and a pool-wide `subscribe` (streams on
188/// Android). `process_incoming` dedups by outer-event id, so overlap folds exactly once.
189pub async fn refresh_subscription(client: &Client) {
190    let (pseudonyms, relays) = rebuild_routes().await;
191
192    // Hold COMMUNITY_SUB_ID across the unsubscribe+subscribe ON PURPOSE: it serializes concurrent
193    // refreshers (Monitor, health-probe reconnect, refresh_control, accept_invite) so they can't
194    // race into a duplicate subscription. Narrowing this lock would reintroduce that double-sub race.
195    let mut new_set = pseudonyms.clone();
196    new_set.sort();
197
198    let mut sub_guard = COMMUNITY_SUB_ID.lock().await;
199    let mut set_guard = COMMUNITY_SUB_SET.lock().await;
200
201    // Idempotent: unchanged pseudonym set + a live sub → keep it (the pool auto-re-applies it across
202    // reconnects, verified). Rebuilding would be pure churn and a rebuild landing mid-reconnect silently
203    // fails to register. rebuild_routes() above already refreshed routes/banlist — all a no-change needs.
204    if sub_guard.is_some() && *set_guard == new_set {
205        return;
206    }
207
208    if let Some(old_id) = sub_guard.take() {
209        let _ = client.unsubscribe(&old_id).await;
210    }
211    *set_guard = new_set;
212
213    if pseudonyms.is_empty() {
214        if let Some(old_pw) = COMMUNITY_POOLWIDE_SUB_ID.lock().await.take() {
215            let _ = client.unsubscribe(&old_pw).await;
216        }
217        return;
218    }
219
220    // Community events live on the community's relays, which may differ from the user's DM relays.
221    // Add them GOSSIP|PING (warm, but excluded from pool-wide DM/profile ops) and subscribe by TARGET.
222    for r in &relays {
223        let _ = client.add_managed_relay(r.as_str()).capabilities(crate::community_relay_capabilities()).await;
224    }
225    client.connect().await;
226
227    // Wait (briefly) for at least one community relay to actually CONNECT before subscribing. A
228    // subscribe_to/subscribe issued against a still-connecting relay silently fails to register the
229    // live sub (seen right after the Android bg-sync churn / on create-with-avatar, which fires extra
230    // control re-folds → re-subscribes mid-reconnect). Polling until a socket is live makes the sub
231    // land reliably regardless of churn timing. Dead/slow relays (e.g. a timing-out one) are ignored —
232    // one connected relay is enough to stream.
233    {
234        let wanted: Vec<RelayUrl> = relays.iter().filter_map(|r| RelayUrl::parse(r).ok()).collect();
235        for _ in 0..24 {
236            let pool = client.relays().all().await;
237            let any_live = wanted.iter().any(|u| {
238                pool.get(u).map(|r| r.status() == RelayStatus::Connected).unwrap_or(false)
239            });
240            if any_live {
241                break;
242            }
243            tokio::time::sleep(Duration::from_millis(250)).await;
244        }
245    }
246
247    let filter = Filter::new()
248        .kinds([
249            Kind::Custom(event_kind::COMMUNITY_MESSAGE),
250            Kind::Custom(event_kind::COMMUNITY_REACTION),
251            Kind::Custom(event_kind::COMMUNITY_EDIT),
252            Kind::Custom(event_kind::COMMUNITY_DELETE),
253            Kind::Custom(event_kind::COMMUNITY_PRESENCE),
254            Kind::Custom(event_kind::COMMUNITY_KICK),
255            Kind::Custom(event_kind::COMMUNITY_TYPING),
256            Kind::Custom(event_kind::COMMUNITY_WEBXDC),
257            Kind::Custom(event_kind::COMMUNITY_CONTROL),
258            Kind::Custom(event_kind::COMMUNITY_REKEY),
259        ])
260        .custom_tags(SingleLetterTag::LOWERCASE_Z, pseudonyms)
261        .limit(0);
262
263    // Pool-wide subscribe — the path that streams on Android (replaces any prior one).
264    {
265        let mut pw = COMMUNITY_POOLWIDE_SUB_ID.lock().await;
266        if let Some(old) = pw.take() {
267            let _ = client.unsubscribe(&old).await;
268        }
269        if let Ok(out) = client.subscribe(filter.clone()).await {
270            *pw = Some(out.value);
271        }
272    }
273    // Targeted subscribe — the path that streams on desktop.
274    if let Ok(output) = client
275        .subscribe(nostr_sdk::prelude::ReqTarget::manual(
276            relays.iter().cloned().map(|u| (u, vec![filter.clone()])),
277        ))
278        .await
279    {
280        *sub_guard = Some(output.value);
281    }
282}
283
284/// Route an arriving Community event: a CONTROL/REKEY edition triggers a realtime control refresh;
285/// any other (message/reaction/edit/delete/presence/typing/webxdc/kick) is opened against its
286/// channel via [`inbound::process_incoming`], persisted, and dispatched to `handler`. `session`
287/// straddles the relay I/O so a mid-flight account swap can't write into the swapped-in account.
288pub async fn dispatch_event(
289    session: &SessionGuard,
290    event: Event,
291    handler: Arc<dyn InboundEventHandler>,
292) {
293    let Some(my_pk) = crate::my_public_key() else { return; };
294    let Some(pseudonym) = event.tags.iter().find_map(|t| {
295        let s = t.as_slice();
296        (s.len() >= 2 && s[0] == "z").then(|| s[1].clone())
297    }) else { return; };
298
299    // A CONTROL edition (3308) or base/channel REKEY (3303) → follow the control plane in realtime.
300    let kind = event.kind.as_u16();
301    if kind == event_kind::COMMUNITY_CONTROL || kind == event_kind::COMMUNITY_REKEY {
302        let community_id = CONTROL_ROUTES.lock().await.get(&pseudonym).cloned();
303        if let Some(community_id) = community_id {
304            if session.is_valid() {
305                // Spawn off the loop — the refresh runs several relay fetches (seconds) and must not
306                // head-of-line-block other event consumption. It self-captures a guard + re-checks.
307                tokio::spawn(refresh_control(community_id, handler.clone()));
308            }
309        }
310        return;
311    }
312
313    let Some(channel) = COMMUNITY_ROUTES.lock().await.get(&pseudonym).cloned() else {
314        return;
315    };
316    if !session.is_valid() {
317        return;
318    }
319
320    let outcome = {
321        let mut state = crate::state::STATE.lock().await;
322        inbound::process_incoming(&mut state, &event, &channel, &my_pk)
323    };
324    let chat_id = channel.id.to_hex();
325    match outcome {
326        Some(inbound::IncomingEvent::NewMessage(mut msg)) => {
327            // Resolve the reply preview (content/npub) from the DB before emitting,
328            // mirroring the DM realtime path. The replied-to message is often an
329            // older one that's persisted but outside the in-memory window; without
330            // this the recipient's live render finds no in-memory target and the
331            // reply shows as a plain message with no context.
332            if !msg.replied_to.is_empty() {
333                let _ = crate::db::events::populate_reply_context(&mut msg).await;
334            }
335            let _ = crate::db::events::save_message(&chat_id, &msg).await;
336            // This is the LIVE stream (limit-0 subscription) — back-paged history arrives via a
337            // separate one-shot batch, never here. So these are always genuinely new; surface them.
338            handler.on_community_message(&chat_id, &msg, true);
339        }
340        Some(inbound::IncomingEvent::Updated { target_id, mut message, edit_event }) => {
341            // Edits are event-sourced (folded on reload); reactions re-save the message row.
342            if let Some(ev) = edit_event {
343                let mut ev = (*ev).clone();
344                if let Ok(cid) = crate::db::id_cache::get_chat_id_by_identifier(&chat_id) {
345                    ev.chat_id = cid;
346                }
347                let _ = crate::db::events::save_event(&ev).await;
348            } else {
349                let _ = crate::db::events::save_message(&chat_id, &message).await;
350            }
351            // Re-resolve before emitting: this message came back out of STATE, whose
352            // compact form drops the quoted attachment's extension.
353            let _ = crate::db::events::populate_reply_context(&mut message).await;
354            handler.on_community_update(&chat_id, &target_id, &message);
355        }
356        Some(inbound::IncomingEvent::Removed { target_id }) => {
357            let _ = crate::db::events::delete_event(&target_id).await;
358            handler.on_community_removed(&chat_id, &target_id);
359        }
360        Some(inbound::IncomingEvent::ReactionRemoved { message_id, reaction_id, mut message }) => {
361            // Drop the reaction's kind-7 row (save is additive) and refresh the parent's chips.
362            let _ = crate::db::events::delete_event(&reaction_id).await;
363            let _ = crate::db::events::populate_reply_context(&mut message).await;
364            handler.on_community_update(&chat_id, &message_id, &message);
365        }
366        Some(inbound::IncomingEvent::Presence { npub, joined, event_id, created_at, invited_by, invited_label }) => {
367            handler.on_community_presence(
368                &chat_id, &npub, joined, &event_id, created_at,
369                invited_by.as_deref(), invited_label.as_deref(),
370            );
371        }
372        Some(inbound::IncomingEvent::WebxdcPeer { npub, topic_id, node_addr, event_id, created_at }) => {
373            handler.on_community_webxdc(
374                &chat_id, &npub, &topic_id, node_addr.as_deref(), &event_id, created_at,
375            );
376        }
377        Some(inbound::IncomingEvent::Typing { npub, until }) => {
378            handler.on_community_typing(&chat_id, &npub, until);
379        }
380        Some(inbound::IncomingEvent::Kicked { community_id })
381        | Some(inbound::IncomingEvent::SelfLeft { community_id }) => {
382            crate::log_warn!("[v1:teardown {}] INBOUND Kicked/SelfLeft on the v1 channel plane", &community_id[..8.min(community_id.len())]);
383            // The handler owns teardown (the GUI prunes chats/relays + republishes the list; a
384            // headless consumer can call `teardown_local`). Core only routes + notifies here.
385            handler.on_community_self_removed(&community_id);
386        }
387        None => {}
388    }
389}
390
391/// Realtime control-plane follow: walk a re-founding, fold the control plane (banlist/roles/
392/// metadata/invites), follow channel rekeys, then resubscribe at the new pseudonyms if any epoch
393/// advanced (else just refresh the route maps so the new banlist takes live effect). Mirrors the
394/// Tauri `refresh_community_control` orchestration over the already-core `catch_up_*` primitives.
395pub async fn refresh_control(community_id: String, handler: Arc<dyn InboundEventHandler>) {
396    // Claim the in-flight slot or bail (a concurrent refresh is already folding this community).
397    {
398        let mut inflight = REFRESH_CONTROL_INFLIGHT.lock().unwrap_or_else(|e| e.into_inner());
399        if !inflight.insert(community_id.clone()) {
400            return;
401        }
402    }
403    struct RefreshClaim(String);
404    impl Drop for RefreshClaim {
405        fn drop(&mut self) {
406            REFRESH_CONTROL_INFLIGHT.lock().unwrap_or_else(|e| e.into_inner()).remove(&self.0);
407        }
408    }
409    let _claim = RefreshClaim(community_id.clone());
410
411    let session = SessionGuard::capture();
412    let Some(id_bytes) = hex_to_id32(&community_id) else { return; };
413    let Some(community) = crate::db::community::load_community(&CommunityId(id_bytes)).ok().flatten() else { return; };
414    let bt = LiveTransport::with_timeout(Duration::from_secs(20));
415    let pre_server_epoch = community.server_root_epoch.0;
416    let pre_channel_epochs: Vec<(String, u64)> =
417        community.channels.iter().map(|c| (c.id.to_hex(), c.epoch.0)).collect();
418
419    // FOLLOW FIRST: a privatize / private-ban re-founds the base under a NEW epoch + re-anchors the
420    // control plane there, so walk the rotation BEFORE folding control. An AUTHORIZED rotation that
421    // excluded us is a removal → tear down locally.
422    if let Ok(c) = service::catch_up_server_root(&bt, &community).await {
423        if !session.is_valid() { return; }
424        if c.removed {
425            crate::log_warn!("[v1:teardown {}] catch_up_server_root says REMOVED (v1 rekey walk)", &community_id[..8.min(community_id.len())]);
426            handler.on_community_self_removed(&community_id); return;
427        }
428    }
429    if !session.is_valid() { return; }
430    let community = crate::db::community::load_community(&CommunityId(id_bytes)).ok().flatten().unwrap_or(community);
431    let _ = service::fetch_and_apply_control(&bt, &community).await;
432    if !session.is_valid() { return; }
433    // Banned by the just-folded banlist → torn down, nothing more to do.
434    if let Some(c) = crate::db::community::load_community(&CommunityId(id_bytes)).ok().flatten() {
435        if service::am_i_banned(&c) {
436            crate::log_warn!("[v1:teardown {}] am_i_banned on the v1 refresh path", &community_id[..8.min(community_id.len())]);
437            handler.on_community_self_removed(&community_id);
438            return;
439        }
440    }
441    if !session.is_valid() { return; }
442
443    // Walk each channel's rekey chain. A re-founding rotates base AND every channel once; the channel
444    // rekey publishes right after the base rekey, so a single fetch can race propagation — retry with a
445    // short backoff until every channel reaches the expected epoch (next sync is the backstop).
446    let base_delta = crate::db::community::load_community(&CommunityId(id_bytes)).ok().flatten()
447        .map(|c| c.server_root_epoch.0).unwrap_or(pre_server_epoch).saturating_sub(pre_server_epoch);
448    for attempt in 0..CHANNEL_FOLLOW_MAX_ATTEMPTS {
449        if !session.is_valid() { return; }
450        let Some(cur) = crate::db::community::load_community(&CommunityId(id_bytes)).ok().flatten() else { break; };
451        for ch in &cur.channels {
452            let _ = service::catch_up_channel_rekeys(&bt, &cur, &ch.id).await;
453        }
454        let caught = base_delta == 0 || crate::db::community::load_community(&CommunityId(id_bytes)).ok().flatten()
455            .map(|c| c.channels.iter().all(|ch| {
456                let pre = pre_channel_epochs.iter().find(|(id, _)| id == &ch.id.to_hex()).map(|(_, e)| *e).unwrap_or(ch.epoch.0);
457                ch.epoch.0 >= pre.saturating_add(base_delta)
458            }))
459            .unwrap_or(true);
460        if caught { break; }
461        if attempt + 1 < CHANNEL_FOLLOW_MAX_ATTEMPTS {
462            tokio::time::sleep(Duration::from_millis(CHANNEL_FOLLOW_BACKOFF_MS)).await;
463        }
464    }
465    if !session.is_valid() { return; }
466    let community = crate::db::community::load_community(&CommunityId(id_bytes)).ok().flatten().unwrap_or(community);
467    let _ = service::retry_pending_read_cut(&bt, &community).await;
468    if !session.is_valid() { return; }
469    let community = crate::db::community::load_community(&CommunityId(id_bytes)).ok().flatten().unwrap_or(community);
470
471    // If an epoch advanced, rebuild the FULL subscription so realtime delivery resumes at the new
472    // pseudonyms; else just refresh the route maps so the inbound drop-filter sees the new banlist.
473    let advanced = community.server_root_epoch.0 != pre_server_epoch
474        || community.channels.iter().any(|c| {
475            pre_channel_epochs.iter().find(|(id, _)| id == &c.id.to_hex()).map(|(_, e)| *e != c.epoch.0).unwrap_or(true)
476        });
477    if advanced && session.is_valid() {
478        if let Some(client) = crate::state::nostr_client() {
479            refresh_subscription(&client).await;
480        }
481    } else {
482        let _ = rebuild_routes().await;
483    }
484    if !session.is_valid() { return; }
485    crate::community::list::refresh_membership_current(&community);
486    handler.on_community_refreshed(&community_id);
487}
488
489/// Tear down a community locally on a received removal (kick/leave/ban), RETAINING the held epoch
490/// keys (so a self-scrub republish/erase still works), then refresh the subscription so we stop
491/// listening on its pseudonyms. Headless consumers can call this from `on_community_self_removed`;
492/// the GUI does a richer teardown (prune chats/relays + republish the list) in its handler instead.
493pub async fn teardown_local(community_id: &str) {
494    let _ = crate::db::community::delete_community_retain_keys(community_id);
495    if let Some(client) = crate::state::nostr_client() {
496        refresh_subscription(&client).await;
497    }
498}
499
500/// hex (64 chars) → 32-byte id.
501fn hex_to_id32(hex: &str) -> Option<[u8; 32]> {
502    (hex.len() == 64).then(|| crate::simd::hex::hex_to_bytes_32(hex))
503}