1use std::collections::HashMap;
29use std::net::{SocketAddr, SocketAddrV4, SocketAddrV6};
30use std::sync::Arc;
31#[cfg(test)]
32use std::sync::atomic::{AtomicBool, Ordering};
33use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
34
35use mdns_sd::{IfKind, ScopedIp, ServiceDaemon, ServiceEvent, ServiceInfo};
36use thiserror::Error;
37use tokio::sync::Mutex;
38use tracing::{debug, info, warn};
39
40use crate::Identity;
41
42pub const SERVICE_TYPE: &str = "_fips._udp.local.";
48
49const IP_CHECK_INTERVAL_SECS: u32 = 30;
52const BROWSE_WINDOW: Duration = Duration::from_secs(5);
55const BROWSE_PERIOD_MS: u128 = 60_000;
56
57pub const TXT_KEY_NPUB: &str = "npub";
59
60pub const TXT_KEY_SCOPE: &str = "scope";
62
63pub const TXT_KEY_VERSION: &str = "v";
66
67#[derive(Debug, Error)]
68pub enum LanDiscoveryError {
69 #[error("mDNS daemon init failed: {0}")]
70 Daemon(String),
71 #[error("mDNS register failed: {0}")]
72 Register(String),
73 #[error("mDNS browse failed: {0}")]
74 Browse(String),
75 #[error("no advertised UDP port — start a UDP transport first")]
76 NoAdvertisedPort,
77 #[error("LAN discovery disabled in config")]
78 Disabled,
79}
80
81#[derive(Debug, Clone)]
85pub struct LanDiscoveredPeer {
86 pub npub: String,
87 pub scope: Option<String>,
88 pub addr: SocketAddr,
89 pub observed_at: Instant,
90}
91
92#[derive(Debug, Clone)]
94pub enum LanEvent {
95 Discovered(LanDiscoveredPeer),
96}
97
98#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
100pub struct LanDiscoveryConfig {
101 #[serde(default = "LanDiscoveryConfig::default_enabled")]
105 pub enabled: bool,
106 #[serde(default = "LanDiscoveryConfig::default_service_type")]
109 pub service_type: String,
110 #[serde(default, skip_serializing_if = "Option::is_none")]
117 pub scope: Option<String>,
118}
119
120impl Default for LanDiscoveryConfig {
121 fn default() -> Self {
122 Self {
123 enabled: Self::default_enabled(),
124 service_type: Self::default_service_type(),
125 scope: None,
126 }
127 }
128}
129
130impl LanDiscoveryConfig {
131 fn default_enabled() -> bool {
132 true
133 }
134 fn default_service_type() -> String {
135 SERVICE_TYPE.to_string()
136 }
137}
138
139pub struct LanDiscovery {
141 daemon: ServiceDaemon,
142 own_npub: String,
143 instance_fullname: String,
144 events_rx: Mutex<tokio::sync::mpsc::UnboundedReceiver<LanEvent>>,
145 event_pump: tokio::task::JoinHandle<()>,
146 #[cfg(test)]
147 browse_paused: Arc<AtomicBool>,
148}
149
150impl LanDiscovery {
151 pub async fn start(
158 identity: &Identity,
159 scope: Option<String>,
160 advertised_port: u16,
161 config: LanDiscoveryConfig,
162 ) -> Result<Arc<Self>, LanDiscoveryError> {
163 if !config.enabled {
164 return Err(LanDiscoveryError::Disabled);
165 }
166 if advertised_port == 0 {
167 return Err(LanDiscoveryError::NoAdvertisedPort);
168 }
169
170 let daemon = ServiceDaemon::new().map_err(|e| LanDiscoveryError::Daemon(e.to_string()))?;
171 daemon
172 .set_ip_check_interval(IP_CHECK_INTERVAL_SECS)
173 .map_err(|e| LanDiscoveryError::Daemon(e.to_string()))?;
174
175 let npub = identity.npub();
176 let label_npub = &npub[..16.min(npub.len())];
180 let instance_name = format!("fips-{label_npub}");
181 let host_name = format!("{instance_name}.local.");
182
183 let mut props: HashMap<String, String> = HashMap::new();
184 props.insert(TXT_KEY_NPUB.to_string(), npub.clone());
185 if let Some(s) = scope.as_deref()
186 && !s.is_empty()
187 {
188 props.insert(TXT_KEY_SCOPE.to_string(), s.to_string());
189 }
190 props.insert(
191 TXT_KEY_VERSION.to_string(),
192 super::nostr::PROTOCOL_VERSION.to_string(),
193 );
194
195 let service_info = ServiceInfo::new(
202 &config.service_type,
203 &instance_name,
204 &host_name,
205 "127.0.0.1",
206 advertised_port,
207 Some(props),
208 )
209 .map_err(|e| LanDiscoveryError::Register(e.to_string()))?
210 .enable_addr_auto();
211
212 let instance_fullname = service_info.get_fullname().to_string();
213
214 daemon
215 .register(service_info)
216 .map_err(|e| LanDiscoveryError::Register(e.to_string()))?;
217
218 let browse_rx = daemon
219 .browse(&config.service_type)
220 .map_err(|e| LanDiscoveryError::Browse(e.to_string()))?;
221
222 let (events_tx, events_rx) = tokio::sync::mpsc::unbounded_channel();
223 let own_npub = npub.clone();
224 let scope_filter = scope.clone().filter(|s| !s.is_empty());
225 let browse_daemon = daemon.clone();
226 let service_type = config.service_type.clone();
227 #[cfg(test)]
228 let browse_paused = Arc::new(AtomicBool::new(false));
229 #[cfg(test)]
230 let event_pump_browse_paused = Arc::clone(&browse_paused);
231 let event_pump = tokio::spawn(async move {
232 let mut browse_rx = browse_rx;
233 loop {
234 let window = tokio::time::sleep(BROWSE_WINDOW);
235 tokio::pin!(window);
236 loop {
237 let event = tokio::select! {
238 event = browse_rx.recv_async() => match event {
239 Ok(event) => event,
240 Err(_) => return,
241 },
242 () = &mut window => break,
243 };
244 match event {
245 ServiceEvent::ServiceResolved(info) => {
246 let mut peer_npub: Option<String> = None;
247 let mut peer_scope: Option<String> = None;
248 for prop in info.get_properties().iter() {
249 match prop.key() {
250 TXT_KEY_NPUB => {
251 peer_npub = Some(prop.val_str().to_string());
252 }
253 TXT_KEY_SCOPE => {
254 peer_scope = Some(prop.val_str().to_string());
255 }
256 _ => {}
257 }
258 }
259 let Some(peer_npub) = peer_npub else {
260 debug!(
261 instance = info.get_fullname(),
262 "lan: skip advert without npub TXT"
263 );
264 continue;
265 };
266 if peer_npub == own_npub {
267 continue;
270 }
271 if scope_filter.is_some() && scope_filter != peer_scope {
272 debug!(
273 npub = %short(&peer_npub),
274 their_scope = ?peer_scope,
275 our_scope = ?scope_filter,
276 "lan: skip cross-scope advert"
277 );
278 continue;
279 }
280 let port = info.get_port();
281 if port == 0 {
282 continue;
283 }
284 let observed_at = Instant::now();
285 for scoped in info.get_addresses() {
294 let Some(addr) = socket_addr_from_scoped_ip(scoped, port) else {
295 debug!(
296 npub = %short(&peer_npub),
297 addr = %scoped.to_ip_addr(),
298 "lan: skip scope-less IPv6 link-local advert"
299 );
300 continue;
301 };
302 if events_tx
303 .send(LanEvent::Discovered(LanDiscoveredPeer {
304 npub: peer_npub.clone(),
305 scope: peer_scope.clone(),
306 addr,
307 observed_at,
308 }))
309 .is_err()
310 {
311 return;
312 }
313 }
314 }
315 ServiceEvent::ServiceRemoved(_, fullname) => {
316 debug!(fullname = %fullname, "lan: service removed");
317 }
318 other => {
319 debug!(?other, "lan: mDNS event");
320 }
321 }
322 }
323 if let Err(error) = browse_daemon.stop_browse(&service_type) {
324 warn!(%error, "lan: stop mDNS browse failed");
325 return;
326 }
327 if let Err(error) = browse_daemon.disable_interface(IfKind::All) {
328 warn!(%error, "lan: suspend mDNS interfaces failed");
329 return;
330 }
331 #[cfg(test)]
332 event_pump_browse_paused.store(true, Ordering::Release);
333 tokio::time::sleep(next_browse_window_delay()).await;
334 #[cfg(test)]
335 event_pump_browse_paused.store(false, Ordering::Release);
336 if let Err(error) = browse_daemon.enable_interface(IfKind::All) {
337 warn!(%error, "lan: resume mDNS interfaces failed");
338 return;
339 }
340 browse_rx = match browse_daemon.browse(&service_type) {
341 Ok(receiver) => receiver,
342 Err(error) => {
343 warn!(%error, "lan: restart mDNS browse failed");
344 return;
345 }
346 };
347 }
348 });
349
350 info!(
351 instance = %instance_fullname,
352 port = advertised_port,
353 scope = ?scope,
354 "lan: mDNS discovery started"
355 );
356 Ok(Arc::new(Self {
357 daemon,
358 own_npub: npub,
359 instance_fullname,
360 events_rx: Mutex::new(events_rx),
361 event_pump,
362 #[cfg(test)]
363 browse_paused,
364 }))
365 }
366
367 pub fn own_npub(&self) -> &str {
369 &self.own_npub
370 }
371
372 pub async fn drain_events(&self) -> Vec<LanEvent> {
374 let mut rx = self.events_rx.lock().await;
375 let mut events = Vec::new();
376 while let Ok(event) = rx.try_recv() {
377 events.push(event);
378 }
379 events
380 }
381
382 pub async fn shutdown(self: &Arc<Self>) {
384 if let Err(e) = self.daemon.unregister(&self.instance_fullname) {
385 warn!(error = %e, "lan: unregister failed");
386 }
387 if let Err(e) = self.daemon.shutdown() {
388 warn!(error = %e, "lan: daemon shutdown failed");
389 }
390 self.event_pump.abort();
391 }
392}
393
394fn next_browse_window_delay() -> Duration {
395 let now_ms = SystemTime::now()
396 .duration_since(UNIX_EPOCH)
397 .unwrap_or_default()
398 .as_millis();
399 browse_window_delay_at(now_ms)
400}
401
402fn browse_window_delay_at(now_ms: u128) -> Duration {
403 Duration::from_millis((BROWSE_PERIOD_MS - now_ms % BROWSE_PERIOD_MS) as u64)
404}
405
406fn short(npub: &str) -> &str {
407 let end = 16.min(npub.len());
408 &npub[..end]
409}
410
411fn socket_addr_from_scoped_ip(scoped: &ScopedIp, port: u16) -> Option<SocketAddr> {
412 match scoped {
413 ScopedIp::V4(v4) => Some(SocketAddr::V4(SocketAddrV4::new(*v4.addr(), port))),
414 ScopedIp::V6(v6) => {
415 let ip = *v6.addr();
416 let scope_id = v6.scope_id().index;
417 if ipv6_is_unicast_link_local(ip) && scope_id == 0 {
418 return None;
419 }
420 Some(SocketAddr::V6(SocketAddrV6::new(ip, port, 0, scope_id)))
421 }
422 _ => None,
423 }
424}
425
426fn ipv6_is_unicast_link_local(ip: std::net::Ipv6Addr) -> bool {
427 (ip.segments()[0] & 0xffc0) == 0xfe80
428}
429
430#[cfg(test)]
431mod tests;