1use 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
24const CHANNEL_FOLLOW_MAX_ATTEMPTS: usize = 5;
27const CHANNEL_FOLLOW_BACKOFF_MS: u64 = 700;
28
29static COMMUNITY_SUB_ID: LazyLock<Mutex<Option<SubscriptionId>>> = LazyLock::new(|| Mutex::new(None));
32
33static COMMUNITY_SUB_SET: LazyLock<Mutex<Vec<String>>> = LazyLock::new(|| Mutex::new(Vec::new()));
39
40static COMMUNITY_POOLWIDE_SUB_ID: LazyLock<Mutex<Option<SubscriptionId>>> = LazyLock::new(|| Mutex::new(None));
46
47static COMMUNITY_ROUTES: LazyLock<Mutex<HashMap<String, Channel>>> =
50 LazyLock::new(|| Mutex::new(HashMap::new()));
51
52static CONTROL_ROUTES: LazyLock<Mutex<HashMap<String, String>>> =
55 LazyLock::new(|| Mutex::new(HashMap::new()));
56
57static REFRESH_CONTROL_INFLIGHT: LazyLock<StdMutex<HashSet<String>>> =
59 LazyLock::new(|| StdMutex::new(HashSet::new()));
60
61pub async fn subscription_id() -> Option<SubscriptionId> {
64 COMMUNITY_SUB_ID.lock().await.clone()
65}
66
67pub async fn poolwide_subscription_id() -> Option<SubscriptionId> {
72 COMMUNITY_POOLWIDE_SUB_ID.lock().await.clone()
73}
74
75pub 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
87pub 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 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 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 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 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
142pub 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 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
186pub async fn refresh_subscription(client: &Client) {
190 let (pseudonyms, relays) = rebuild_routes().await;
191
192 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 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 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 {
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 {
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 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
284pub 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 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 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 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 handler.on_community_message(&chat_id, &msg, true);
339 }
340 Some(inbound::IncomingEvent::Updated { target_id, mut message, edit_event }) => {
341 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 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 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 handler.on_community_self_removed(&community_id);
386 }
387 None => {}
388 }
389}
390
391pub async fn refresh_control(community_id: String, handler: Arc<dyn InboundEventHandler>) {
396 {
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 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 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 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 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
489pub 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
500fn hex_to_id32(hex: &str) -> Option<[u8; 32]> {
502 (hex.len() == 64).then(|| crate::simd::hex::hex_to_bytes_32(hex))
503}