offline_protocol_services/
services.rs

1use crate::error::ServiceError;
2use crate::events::ServiceEvent;
3use crate::payloads::{
4    ServiceDiscoveryQueryPayload, ServiceDiscoveryResponsePayload, ServiceRequestPayload,
5    ServiceResponsePayload, DISCOVERY_GOSSIP_MAX_FANOUT, DISCOVERY_INITIAL_BROADCAST_MAX,
6    DISCOVERY_QUERY_DEDUP_TTL_SECS, DISCOVERY_QUERY_DEFAULT_MAX_HOPS,
7    DISCOVERY_QUERY_MAX_DEDUP_ENTRIES, MAX_SERVICE_BODY_SIZE, MAX_SERVICE_METHOD_LEN,
8    MAX_SERVICE_PAYLOAD_SIZE, SVC_DISCOVER_QUERY, SVC_DISCOVER_RESPONSE, SVC_REQUEST, SVC_RESPONSE,
9    VALID_SERVICE_STATUSES,
10};
11use offline_protocol_core::{MessagePriority, ServiceDescriptor};
12use std::collections::{HashMap, VecDeque};
13use std::time::{Duration, Instant};
14use tracing::{debug, info, warn};
15
16/// A message to send to a peer.
17#[derive(Debug, Clone)]
18pub struct OutboundMessage {
19    /// The recipient peer ID.
20    pub recipient: String,
21    /// The serialized message content.
22    pub content: String,
23    /// The message priority.
24    pub priority: MessagePriority,
25}
26
27/// Result of initiating a service discovery broadcast.
28#[must_use]
29pub struct DiscoverResult {
30    /// Unique query identifier for correlating responses.
31    pub query_id: String,
32    /// Messages to send to peers.
33    pub messages: Vec<OutboundMessage>,
34}
35
36/// Result of sending a service request.
37#[must_use]
38pub struct SendRequestResult {
39    /// Unique request identifier for correlating the response.
40    pub request_id: String,
41    /// Message to send to the provider.
42    pub message: OutboundMessage,
43}
44
45/// Result of responding to a service request.
46#[must_use]
47pub struct SendResponseResult {
48    /// Message to send to the requester.
49    pub message: OutboundMessage,
50}
51
52/// Action returned from handling an incoming service message.
53pub enum ServiceAction {
54    /// The message was not a service message; caller should continue processing.
55    NotHandled,
56    /// The message was consumed as a service message.
57    Consumed {
58        /// Messages to send to peers.
59        messages_to_send: Vec<OutboundMessage>,
60        /// Events to emit to the application.
61        events_to_emit: Vec<ServiceEvent>,
62    },
63}
64
65/// Mesh service registry and message handler.
66///
67/// Manages local service registration, discovery query generation and handling,
68/// and service request/response routing. All methods return **actions** (messages
69/// to send, events to emit) rather than performing I/O directly.
70///
71/// ## Multi-hop discovery
72///
73/// Discovery **queries** are forwarded across multiple hops via gossip.
74/// Discovery **responses** are sent to the direct sender of the query, never
75/// to the `originator` the payload names, so that a spoofed originator cannot
76/// make a provider leak its service list to an arbitrary peer. A node that
77/// receives a response it did not originate forwards it as a unicast toward
78/// the originator (see `try_handle_discover_response`) and emits no event of
79/// its own; only the originator emits `ServiceDiscovered`.
80pub struct MeshServices {
81    local_services: HashMap<String, ServiceDescriptor>,
82    seen_discovery_queries: HashMap<String, Instant>,
83    /// Insertion-order index for O(1) oldest-entry eviction.
84    seen_discovery_order: VecDeque<String>,
85}
86
87impl MeshServices {
88    /// Creates a new empty service registry.
89    pub fn new() -> Self {
90        Self {
91            local_services: HashMap::new(),
92            seen_discovery_queries: HashMap::new(),
93            seen_discovery_order: VecDeque::new(),
94        }
95    }
96
97    /// Registers a local service.
98    pub fn register_service(&mut self, descriptor: ServiceDescriptor) -> Result<(), ServiceError> {
99        let key = descriptor.service_id.as_str().to_string();
100        self.local_services.insert(key, descriptor);
101        Ok(())
102    }
103
104    /// Unregisters a local service. Returns `true` if it was found and removed.
105    pub fn unregister_service(&mut self, service_id: &str) -> Result<bool, ServiceError> {
106        Ok(self.local_services.remove(service_id).is_some())
107    }
108
109    /// Returns `true` if a service with the given ID is registered locally.
110    pub fn has_service(&self, service_id: &str) -> bool {
111        self.local_services.contains_key(service_id)
112    }
113
114    /// Generates a discovery broadcast to known peers, capped at
115    /// [`DISCOVERY_INITIAL_BROADCAST_MAX`] recipients.
116    ///
117    /// See the [multi-hop discovery](MeshServices#multi-hop-discovery) note on
118    /// `MeshServices` for where the responses go.
119    pub fn discover_services(
120        &mut self,
121        user_id: &str,
122        known_peers: &[String],
123        service_id: Option<&str>,
124    ) -> Result<DiscoverResult, ServiceError> {
125        let query_id = uuid::Uuid::new_v4().to_string();
126
127        let payload = ServiceDiscoveryQueryPayload {
128            query_id: query_id.clone(),
129            originator: user_id.to_string(),
130            service_id: service_id.map(|s| s.to_string()),
131            remaining_hops: DISCOVERY_QUERY_DEFAULT_MAX_HOPS,
132        };
133        let serialized = serde_json::to_string(&payload)
134            .map_err(|e| ServiceError::Serialization(e.to_string()))?;
135        let content = format!("{}{}", SVC_DISCOVER_QUERY, serialized);
136
137        self.record_seen_query(query_id.clone());
138
139        // Apply fanout limit to the initial broadcast to bound message generation.
140        let selected_peers = select_fanout_peers(
141            known_peers,
142            DISCOVERY_INITIAL_BROADCAST_MAX,
143            &query_id,
144            user_id,
145        );
146
147        let messages: Vec<OutboundMessage> = selected_peers
148            .into_iter()
149            .map(|peer| OutboundMessage {
150                recipient: peer,
151                content: content.clone(),
152                priority: MessagePriority::Medium,
153            })
154            .collect();
155
156        info!(query_id = %query_id, service_id = ?service_id, peer_count = messages.len(), "Broadcast service discovery query");
157        Ok(DiscoverResult { query_id, messages })
158    }
159
160    /// Generates a service request message to a specific provider peer.
161    pub fn send_service_request(
162        &self,
163        provider: &str,
164        service_id: &str,
165        method: &str,
166        body: &str,
167    ) -> Result<SendRequestResult, ServiceError> {
168        if body.len() > MAX_SERVICE_BODY_SIZE {
169            return Err(ServiceError::PayloadTooLarge(format!(
170                "request body size {} exceeds maximum {}",
171                body.len(),
172                MAX_SERVICE_BODY_SIZE
173            )));
174        }
175        if method.len() > MAX_SERVICE_METHOD_LEN {
176            return Err(ServiceError::PayloadTooLarge(format!(
177                "method length {} exceeds maximum {}",
178                method.len(),
179                MAX_SERVICE_METHOD_LEN
180            )));
181        }
182
183        let request_id = uuid::Uuid::new_v4().to_string();
184
185        let payload = ServiceRequestPayload {
186            request_id: request_id.clone(),
187            service_id: service_id.to_string(),
188            method: method.to_string(),
189            body: body.to_string(),
190        };
191        let serialized = serde_json::to_string(&payload)
192            .map_err(|e| ServiceError::Serialization(e.to_string()))?;
193        let content = format!("{}{}", SVC_REQUEST, serialized);
194
195        info!(request_id = %request_id, provider = %provider, service_id = %service_id, method = %method, "Sent service request");
196        Ok(SendRequestResult {
197            request_id,
198            message: OutboundMessage {
199                recipient: provider.to_string(),
200                content,
201                priority: MessagePriority::High,
202            },
203        })
204    }
205
206    /// Generates a service response message.
207    pub fn respond_to_service_request(
208        &self,
209        request_id: &str,
210        requester: &str,
211        service_id: &str,
212        status: &str,
213        body: &str,
214    ) -> Result<SendResponseResult, ServiceError> {
215        if !VALID_SERVICE_STATUSES.contains(&status) {
216            return Err(ServiceError::InvalidStatus(status.to_string()));
217        }
218        if body.len() > MAX_SERVICE_BODY_SIZE {
219            return Err(ServiceError::PayloadTooLarge(format!(
220                "response body size {} exceeds maximum {}",
221                body.len(),
222                MAX_SERVICE_BODY_SIZE
223            )));
224        }
225
226        let payload = ServiceResponsePayload {
227            request_id: request_id.to_string(),
228            service_id: service_id.to_string(),
229            status: status.to_string(),
230            body: body.to_string(),
231        };
232        let serialized = serde_json::to_string(&payload)
233            .map_err(|e| ServiceError::Serialization(e.to_string()))?;
234        let content = format!("{}{}", SVC_RESPONSE, serialized);
235
236        info!(request_id = %request_id, requester = %requester, status = %status, "Sent service response");
237        Ok(SendResponseResult {
238            message: OutboundMessage {
239                recipient: requester.to_string(),
240                content,
241                priority: MessagePriority::High,
242            },
243        })
244    }
245
246    /// Handles an incoming message, returning a `ServiceAction`.
247    ///
248    /// `hop_count` is the hop count from the incoming message (used in discovery responses).
249    /// `known_peers` should exclude the sender.
250    pub fn handle_incoming_message(
251        &mut self,
252        content: &str,
253        sender: &str,
254        hop_count: u8,
255        user_id: &str,
256        known_peers: &[String],
257    ) -> ServiceAction {
258        if let Some(action) =
259            self.try_handle_discover_query(content, sender, hop_count, user_id, known_peers)
260        {
261            return action;
262        }
263        if let Some(action) = self.try_handle_discover_response(content, sender, user_id) {
264            return action;
265        }
266        if let Some(action) = self.try_handle_request(content, sender) {
267            return action;
268        }
269        if let Some(action) = self.try_handle_response(content, sender) {
270            return action;
271        }
272        ServiceAction::NotHandled
273    }
274
275    /// Removes expired discovery query dedup entries.
276    pub fn cleanup_expired(&mut self) {
277        // Ages are compared with `saturating_duration_since` rather than a
278        // precomputed `now - TTL` cutoff: on platforms where the monotonic
279        // clock starts at boot, that subtraction underflows (and panics)
280        // when the process is younger than the TTL.
281        let now = Instant::now();
282        let ttl = Duration::from_secs(DISCOVERY_QUERY_DEDUP_TTL_SECS);
283        self.seen_discovery_queries
284            .retain(|_, ts| now.saturating_duration_since(*ts) < ttl);
285        self.seen_discovery_order
286            .retain(|id| self.seen_discovery_queries.contains_key(id));
287    }
288
289    /// Provides read access to `seen_discovery_queries` for testing / inspection.
290    pub fn seen_discovery_queries(&self) -> &HashMap<String, Instant> {
291        &self.seen_discovery_queries
292    }
293
294    /// Inserts a query ID into the dedup map, evicting the oldest entry if at capacity.
295    fn record_seen_query(&mut self, query_id: String) {
296        if self.seen_discovery_queries.len() >= DISCOVERY_QUERY_MAX_DEDUP_ENTRIES {
297            // Evict the oldest entry using insertion-order index (O(1)).
298            if let Some(oldest) = self.seen_discovery_order.pop_front() {
299                self.seen_discovery_queries.remove(&oldest);
300            }
301        }
302        self.seen_discovery_queries
303            .insert(query_id.clone(), Instant::now());
304        self.seen_discovery_order.push_back(query_id);
305    }
306
307    // --- private handlers ---
308
309    fn try_handle_discover_query(
310        &mut self,
311        content: &str,
312        sender: &str,
313        hop_count: u8,
314        user_id: &str,
315        known_peers: &[String],
316    ) -> Option<ServiceAction> {
317        let data = content.strip_prefix(SVC_DISCOVER_QUERY)?;
318
319        if data.len() > MAX_SERVICE_PAYLOAD_SIZE {
320            warn!(sender = %sender, size = data.len(), "Discovery query payload too large, dropping");
321            return Some(ServiceAction::Consumed {
322                messages_to_send: vec![],
323                events_to_emit: vec![],
324            });
325        }
326
327        let payload: ServiceDiscoveryQueryPayload = match serde_json::from_str(data) {
328            Ok(p) => p,
329            Err(_) => {
330                warn!(sender = %sender, "Failed to parse service discovery query payload");
331                return Some(ServiceAction::Consumed {
332                    messages_to_send: vec![],
333                    events_to_emit: vec![],
334                });
335            }
336        };
337
338        // Dedup
339        if self.seen_discovery_queries.contains_key(&payload.query_id) {
340            debug!(query_id = %payload.query_id, "Duplicate discovery query, skipping");
341            return Some(ServiceAction::Consumed {
342                messages_to_send: vec![],
343                events_to_emit: vec![],
344            });
345        }
346        self.record_seen_query(payload.query_id.clone());
347
348        let mut messages = Vec::new();
349
350        // Check local services for matches
351        let matches: Vec<_> = self
352            .local_services
353            .values()
354            .filter(|svc| {
355                payload
356                    .service_id
357                    .as_ref()
358                    .is_none_or(|q| q == svc.service_id.as_str())
359            })
360            .cloned()
361            .collect();
362
363        // Send discovery responses back to the *sender* (not the originator) to
364        // prevent a spoofed originator field from leaking our service list to an
365        // arbitrary peer. The sender relays the response back along the path.
366        for svc in &matches {
367            let response = ServiceDiscoveryResponsePayload {
368                query_id: payload.query_id.clone(),
369                service_id: svc.service_id.as_str().to_string(),
370                version: svc.version.clone(),
371                provider_peer_id: user_id.to_string(),
372                capabilities: svc.capabilities.clone(),
373                hop_count,
374                originator: payload.originator.clone(),
375            };
376            if let Ok(serialized) = serde_json::to_string(&response) {
377                let resp_content = format!("{}{}", SVC_DISCOVER_RESPONSE, serialized);
378                messages.push(OutboundMessage {
379                    recipient: sender.to_string(),
380                    content: resp_content,
381                    priority: MessagePriority::Medium,
382                });
383            }
384        }
385
386        // Forward query to other known peers (gossip) if hops remain
387        let forwarded_count = if payload.remaining_hops > 0 {
388            let mut fwd_payload = payload.clone();
389            fwd_payload.remaining_hops -= 1;
390            let fwd_content = match serde_json::to_string(&fwd_payload) {
391                Ok(serialized) => format!("{}{}", SVC_DISCOVER_QUERY, serialized),
392                Err(e) => {
393                    warn!("Failed to re-serialize discovery query for forwarding: {e}");
394                    return Some(ServiceAction::Consumed {
395                        messages_to_send: messages,
396                        events_to_emit: vec![],
397                    });
398                }
399            };
400            let eligible_peers: Vec<String> = known_peers
401                .iter()
402                .filter(|p| p.as_str() != sender && p.as_str() != payload.originator)
403                .cloned()
404                .collect();
405            let forward_peers = select_fanout_peers(
406                &eligible_peers,
407                DISCOVERY_GOSSIP_MAX_FANOUT,
408                &payload.query_id,
409                user_id,
410            );
411            for peer in &forward_peers {
412                messages.push(OutboundMessage {
413                    recipient: peer.clone(),
414                    content: fwd_content.clone(),
415                    priority: MessagePriority::Medium,
416                });
417            }
418            forward_peers.len()
419        } else {
420            debug!(query_id = %payload.query_id, "Discovery query reached max hops, not forwarding");
421            0
422        };
423
424        debug!(
425            query_id = %payload.query_id,
426            matches = matches.len(),
427            forwarded_to = forwarded_count,
428            "Processed service discovery query"
429        );
430
431        Some(ServiceAction::Consumed {
432            messages_to_send: messages,
433            events_to_emit: vec![],
434        })
435    }
436
437    fn try_handle_discover_response(
438        &self,
439        content: &str,
440        sender: &str,
441        user_id: &str,
442    ) -> Option<ServiceAction> {
443        let data = content.strip_prefix(SVC_DISCOVER_RESPONSE)?;
444
445        if data.len() > MAX_SERVICE_PAYLOAD_SIZE {
446            warn!(sender = %sender, size = data.len(), "Discovery response payload too large, dropping");
447            return Some(ServiceAction::Consumed {
448                messages_to_send: vec![],
449                events_to_emit: vec![],
450            });
451        }
452
453        let payload: ServiceDiscoveryResponsePayload = match serde_json::from_str(data) {
454            Ok(p) => p,
455            Err(_) => {
456                warn!(sender = %sender, "Failed to parse service discovery response payload");
457                return Some(ServiceAction::Consumed {
458                    messages_to_send: vec![],
459                    events_to_emit: vec![],
460                });
461            }
462        };
463
464        info!(
465            query_id = %payload.query_id,
466            service_id = %payload.service_id,
467            provider = %payload.provider_peer_id,
468            "Service discovered"
469        );
470
471        // If we are not the originator, forward the response toward them
472        if !payload.originator.is_empty() && payload.originator != user_id {
473            debug!(
474                query_id = %payload.query_id,
475                originator = %payload.originator,
476                "Forwarding discovery response toward originator"
477            );
478            // Re-serialize and forward as unicast to the originator
479            let fwd_content = content.to_string();
480            return Some(ServiceAction::Consumed {
481                messages_to_send: vec![OutboundMessage {
482                    recipient: payload.originator,
483                    content: fwd_content,
484                    priority: MessagePriority::Medium,
485                }],
486                events_to_emit: vec![],
487            });
488        }
489
490        Some(ServiceAction::Consumed {
491            messages_to_send: vec![],
492            events_to_emit: vec![ServiceEvent::ServiceDiscovered {
493                query_id: payload.query_id,
494                service_id: payload.service_id,
495                version: payload.version,
496                provider_peer_id: payload.provider_peer_id,
497                capabilities: payload.capabilities,
498                hop_count: payload.hop_count,
499            }],
500        })
501    }
502
503    fn try_handle_request(&self, content: &str, sender: &str) -> Option<ServiceAction> {
504        let data = content.strip_prefix(SVC_REQUEST)?;
505
506        if data.len() > MAX_SERVICE_PAYLOAD_SIZE {
507            warn!(sender = %sender, size = data.len(), "Service request payload too large, dropping");
508            return Some(ServiceAction::Consumed {
509                messages_to_send: vec![],
510                events_to_emit: vec![],
511            });
512        }
513
514        let payload: ServiceRequestPayload = match serde_json::from_str(data) {
515            Ok(p) => p,
516            Err(_) => {
517                warn!(sender = %sender, "Failed to parse service request payload");
518                return Some(ServiceAction::Consumed {
519                    messages_to_send: vec![],
520                    events_to_emit: vec![],
521                });
522            }
523        };
524
525        // Validate field sizes from untrusted input
526        if payload.body.len() > MAX_SERVICE_BODY_SIZE {
527            warn!(sender = %sender, size = payload.body.len(), "Service request body too large, dropping");
528            return Some(ServiceAction::Consumed {
529                messages_to_send: vec![],
530                events_to_emit: vec![],
531            });
532        }
533        if payload.method.len() > MAX_SERVICE_METHOD_LEN {
534            warn!(sender = %sender, len = payload.method.len(), "Service request method too long, dropping");
535            return Some(ServiceAction::Consumed {
536                messages_to_send: vec![],
537                events_to_emit: vec![],
538            });
539        }
540
541        if !self.local_services.contains_key(&payload.service_id) {
542            // Auto-respond with not_found
543            let response = ServiceResponsePayload {
544                request_id: payload.request_id.clone(),
545                service_id: payload.service_id.clone(),
546                status: "not_found".to_string(),
547                body: String::new(),
548            };
549            let mut messages = Vec::new();
550            if let Ok(serialized) = serde_json::to_string(&response) {
551                let resp_content = format!("{}{}", SVC_RESPONSE, serialized);
552                messages.push(OutboundMessage {
553                    recipient: sender.to_string(),
554                    content: resp_content,
555                    priority: MessagePriority::High,
556                });
557            }
558            debug!(
559                request_id = %payload.request_id,
560                service_id = %payload.service_id,
561                "Service not found, auto-responded not_found"
562            );
563            return Some(ServiceAction::Consumed {
564                messages_to_send: messages,
565                events_to_emit: vec![],
566            });
567        }
568
569        info!(
570            request_id = %payload.request_id,
571            service_id = %payload.service_id,
572            method = %payload.method,
573            "Service request received"
574        );
575
576        Some(ServiceAction::Consumed {
577            messages_to_send: vec![],
578            events_to_emit: vec![ServiceEvent::ServiceRequestReceived {
579                request_id: payload.request_id,
580                service_id: payload.service_id,
581                method: payload.method,
582                body: payload.body,
583                sender: sender.to_string(),
584            }],
585        })
586    }
587
588    fn try_handle_response(&self, content: &str, sender: &str) -> Option<ServiceAction> {
589        let data = content.strip_prefix(SVC_RESPONSE)?;
590
591        if data.len() > MAX_SERVICE_PAYLOAD_SIZE {
592            warn!(sender = %sender, size = data.len(), "Service response payload too large, dropping");
593            return Some(ServiceAction::Consumed {
594                messages_to_send: vec![],
595                events_to_emit: vec![],
596            });
597        }
598
599        let payload: ServiceResponsePayload = match serde_json::from_str(data) {
600            Ok(p) => p,
601            Err(_) => {
602                warn!(sender = %sender, "Failed to parse service response payload");
603                return Some(ServiceAction::Consumed {
604                    messages_to_send: vec![],
605                    events_to_emit: vec![],
606                });
607            }
608        };
609
610        // Validate body size from untrusted input
611        if payload.body.len() > MAX_SERVICE_BODY_SIZE {
612            warn!(sender = %sender, size = payload.body.len(), "Service response body too large, dropping");
613            return Some(ServiceAction::Consumed {
614                messages_to_send: vec![],
615                events_to_emit: vec![],
616            });
617        }
618
619        // Validate status from untrusted input
620        if !VALID_SERVICE_STATUSES.contains(&payload.status.as_str()) {
621            warn!(sender = %sender, status = %payload.status, "Service response with invalid status, dropping");
622            return Some(ServiceAction::Consumed {
623                messages_to_send: vec![],
624                events_to_emit: vec![],
625            });
626        }
627
628        info!(
629            request_id = %payload.request_id,
630            service_id = %payload.service_id,
631            status = %payload.status,
632            "Service response received"
633        );
634
635        Some(ServiceAction::Consumed {
636            messages_to_send: vec![],
637            events_to_emit: vec![ServiceEvent::ServiceResponseReceived {
638                request_id: payload.request_id,
639                service_id: payload.service_id,
640                status: payload.status,
641                body: payload.body,
642                provider_peer_id: sender.to_string(),
643            }],
644        })
645    }
646}
647
648/// Selects up to `max` peers from `peers` using a deterministic pseudo-random
649/// subset selection. When `peers.len() <= max`, all peers are returned.
650///
651/// The selection uses a simple hash-based seed derived from `query_id` and
652/// `user_id` so that different nodes select different subsets for the same query,
653/// improving mesh coverage. The distribution has some bias because we use a
654/// lightweight hash rather than a full PRNG — this is intentional as uniform
655/// randomness is not critical for gossip fanout and avoids adding a `rand`
656/// dependency.
657fn select_fanout_peers(peers: &[String], max: usize, query_id: &str, user_id: &str) -> Vec<String> {
658    if peers.len() <= max {
659        return peers.to_vec();
660    }
661
662    let seed = query_id
663        .bytes()
664        .chain(user_id.bytes())
665        .fold(0u64, |acc, b| acc.wrapping_mul(31).wrapping_add(b as u64))
666        ^ peers.len() as u64;
667
668    let mut indices: Vec<usize> = (0..peers.len()).collect();
669    // Partial Fisher-Yates shuffle for the first `max` elements.
670    for i in 0..max {
671        let remaining = peers.len() - i;
672        let j = i + (seed.wrapping_mul((i as u64).wrapping_add(7)) as usize % remaining);
673        indices.swap(i, j);
674    }
675
676    indices[..max]
677        .iter()
678        .map(|&idx| peers[idx].clone())
679        .collect()
680}
681
682impl Default for MeshServices {
683    fn default() -> Self {
684        Self::new()
685    }
686}
687
688#[cfg(test)]
689mod tests {
690    use super::*;
691    use offline_protocol_core::ServiceId;
692
693    fn make_descriptor(id: &str) -> ServiceDescriptor {
694        ServiceDescriptor {
695            service_id: ServiceId::new(id).unwrap(),
696            version: "1.0".to_string(),
697            capabilities: HashMap::new(),
698        }
699    }
700
701    #[test]
702    fn test_register_and_unregister() {
703        let mut svc = MeshServices::new();
704        svc.register_service(make_descriptor("echo.v1")).unwrap();
705        assert!(svc.has_service("echo.v1"));
706
707        assert!(svc.unregister_service("echo.v1").unwrap());
708        assert!(!svc.has_service("echo.v1"));
709
710        assert!(!svc.unregister_service("echo.v1").unwrap());
711    }
712
713    #[test]
714    fn test_discover_services_generates_messages() {
715        let mut svc = MeshServices::new();
716        let peers = vec!["alice".to_string(), "bob".to_string()];
717        let result = svc.discover_services("user1", &peers, None).unwrap();
718        assert!(!result.query_id.is_empty());
719        assert_eq!(result.messages.len(), 2);
720        assert!(svc.seen_discovery_queries().contains_key(&result.query_id));
721    }
722
723    #[test]
724    fn test_discover_services_no_peers() {
725        let mut svc = MeshServices::new();
726        let result = svc
727            .discover_services("user1", &[], Some("weather"))
728            .unwrap();
729        assert!(result.messages.is_empty());
730    }
731
732    #[test]
733    fn test_discover_services_initial_broadcast_fanout_limit() {
734        let mut svc = MeshServices::new();
735        let peers: Vec<String> = (0..50).map(|i| format!("peer-{i}")).collect();
736        let result = svc.discover_services("user1", &peers, None).unwrap();
737        assert_eq!(result.messages.len(), DISCOVERY_INITIAL_BROADCAST_MAX);
738    }
739
740    #[test]
741    fn test_send_service_request() {
742        let svc = MeshServices::new();
743        let result = svc
744            .send_service_request("bob", "echo.v1", "ping", "{}")
745            .unwrap();
746        assert!(!result.request_id.is_empty());
747        assert_eq!(result.message.recipient, "bob");
748        assert!(result.message.content.starts_with(SVC_REQUEST));
749    }
750
751    #[test]
752    fn test_send_service_request_body_too_large() {
753        let svc = MeshServices::new();
754        let large_body = "x".repeat(MAX_SERVICE_BODY_SIZE + 1);
755        let result = svc.send_service_request("bob", "echo.v1", "ping", &large_body);
756        assert!(matches!(result, Err(ServiceError::PayloadTooLarge(_))));
757    }
758
759    #[test]
760    fn test_send_service_request_method_too_long() {
761        let svc = MeshServices::new();
762        let long_method = "m".repeat(MAX_SERVICE_METHOD_LEN + 1);
763        let result = svc.send_service_request("bob", "echo.v1", &long_method, "{}");
764        assert!(matches!(result, Err(ServiceError::PayloadTooLarge(_))));
765    }
766
767    #[test]
768    fn test_respond_to_service_request() {
769        let svc = MeshServices::new();
770        let result = svc
771            .respond_to_service_request("req-1", "alice", "echo.v1", "ok", "pong")
772            .unwrap();
773        assert_eq!(result.message.recipient, "alice");
774        assert!(result.message.content.starts_with(SVC_RESPONSE));
775    }
776
777    #[test]
778    fn test_respond_to_service_request_invalid_status() {
779        let svc = MeshServices::new();
780        let result = svc.respond_to_service_request("req-1", "alice", "echo.v1", "invalid", "");
781        assert!(matches!(result, Err(ServiceError::InvalidStatus(_))));
782    }
783
784    #[test]
785    fn test_respond_to_service_request_body_too_large() {
786        let svc = MeshServices::new();
787        let large_body = "x".repeat(MAX_SERVICE_BODY_SIZE + 1);
788        let result = svc.respond_to_service_request("req-1", "alice", "echo.v1", "ok", &large_body);
789        assert!(matches!(result, Err(ServiceError::PayloadTooLarge(_))));
790    }
791
792    #[test]
793    fn test_respond_valid_statuses() {
794        let svc = MeshServices::new();
795        for status in VALID_SERVICE_STATUSES {
796            let result =
797                svc.respond_to_service_request("req-1", "alice", "echo.v1", status, "body");
798            assert!(result.is_ok(), "status '{status}' should be valid");
799        }
800    }
801
802    #[test]
803    fn test_handle_discover_query_with_match() {
804        let mut svc = MeshServices::new();
805        svc.register_service(ServiceDescriptor {
806            service_id: ServiceId::new("weather").unwrap(),
807            version: "2.0".to_string(),
808            capabilities: {
809                let mut m = HashMap::new();
810                m.insert("format".to_string(), "json".to_string());
811                m
812            },
813        })
814        .unwrap();
815
816        let payload = ServiceDiscoveryQueryPayload {
817            query_id: "q-001".to_string(),
818            originator: "alice".to_string(),
819            service_id: Some("weather".to_string()),
820            remaining_hops: 5,
821        };
822        let content = format!(
823            "{}{}",
824            SVC_DISCOVER_QUERY,
825            serde_json::to_string(&payload).unwrap()
826        );
827
828        let peers = vec!["charlie".to_string()];
829        let action = svc.handle_incoming_message(&content, "alice", 1, "user1", &peers);
830        match action {
831            ServiceAction::Consumed {
832                messages_to_send,
833                events_to_emit,
834            } => {
835                assert!(!messages_to_send.is_empty());
836                assert!(events_to_emit.is_empty());
837                // Response should go to the sender ("alice"), not the originator field
838                assert!(messages_to_send.iter().any(
839                    |m| m.recipient == "alice" && m.content.starts_with(SVC_DISCOVER_RESPONSE)
840                ));
841            }
842            ServiceAction::NotHandled => panic!("Should have been handled"),
843        }
844        assert!(svc.seen_discovery_queries().contains_key("q-001"));
845    }
846
847    #[test]
848    fn test_handle_discover_query_dedup() {
849        let mut svc = MeshServices::new();
850
851        let payload = ServiceDiscoveryQueryPayload {
852            query_id: "q-dedup".to_string(),
853            originator: "alice".to_string(),
854            service_id: None,
855            remaining_hops: 5,
856        };
857        let content = format!(
858            "{}{}",
859            SVC_DISCOVER_QUERY,
860            serde_json::to_string(&payload).unwrap()
861        );
862
863        // First time: processes
864        let a1 = svc.handle_incoming_message(&content, "alice", 0, "user1", &[]);
865        assert!(matches!(a1, ServiceAction::Consumed { .. }));
866
867        // Second time: deduped
868        let a2 = svc.handle_incoming_message(&content, "alice", 0, "user1", &[]);
869        assert!(matches!(a2, ServiceAction::Consumed { .. }));
870    }
871
872    #[test]
873    fn test_handle_discover_response() {
874        let mut svc = MeshServices::new();
875
876        let payload = ServiceDiscoveryResponsePayload {
877            query_id: "q-123".to_string(),
878            service_id: "weather".to_string(),
879            version: "2.0".to_string(),
880            provider_peer_id: "bob".to_string(),
881            capabilities: HashMap::new(),
882            hop_count: 1,
883            originator: "user1".to_string(),
884        };
885        let content = format!(
886            "{}{}",
887            SVC_DISCOVER_RESPONSE,
888            serde_json::to_string(&payload).unwrap()
889        );
890
891        let action = svc.handle_incoming_message(&content, "bob", 0, "user1", &[]);
892        match action {
893            ServiceAction::Consumed { events_to_emit, .. } => {
894                assert_eq!(events_to_emit.len(), 1);
895                match &events_to_emit[0] {
896                    ServiceEvent::ServiceDiscovered {
897                        query_id,
898                        service_id,
899                        ..
900                    } => {
901                        assert_eq!(query_id, "q-123");
902                        assert_eq!(service_id, "weather");
903                    }
904                    other => panic!("Wrong event: {:?}", other),
905                }
906            }
907            ServiceAction::NotHandled => panic!("Should have been handled"),
908        }
909    }
910
911    #[test]
912    fn test_handle_request_unregistered() {
913        let mut svc = MeshServices::new();
914
915        let payload = ServiceRequestPayload {
916            request_id: "req-001".to_string(),
917            service_id: "nonexistent".to_string(),
918            method: "get".to_string(),
919            body: "{}".to_string(),
920        };
921        let content = format!(
922            "{}{}",
923            SVC_REQUEST,
924            serde_json::to_string(&payload).unwrap()
925        );
926
927        let action = svc.handle_incoming_message(&content, "alice", 0, "user1", &[]);
928        match action {
929            ServiceAction::Consumed {
930                messages_to_send,
931                events_to_emit,
932            } => {
933                assert_eq!(messages_to_send.len(), 1);
934                assert!(messages_to_send[0].content.contains("not_found"));
935                assert!(events_to_emit.is_empty());
936            }
937            ServiceAction::NotHandled => panic!("Should have been handled"),
938        }
939    }
940
941    #[test]
942    fn test_handle_request_registered() {
943        let mut svc = MeshServices::new();
944        svc.register_service(make_descriptor("echo")).unwrap();
945
946        let payload = ServiceRequestPayload {
947            request_id: "req-002".to_string(),
948            service_id: "echo".to_string(),
949            method: "ping".to_string(),
950            body: "hello".to_string(),
951        };
952        let content = format!(
953            "{}{}",
954            SVC_REQUEST,
955            serde_json::to_string(&payload).unwrap()
956        );
957
958        let action = svc.handle_incoming_message(&content, "alice", 0, "user1", &[]);
959        match action {
960            ServiceAction::Consumed {
961                messages_to_send,
962                events_to_emit,
963            } => {
964                assert!(messages_to_send.is_empty());
965                assert_eq!(events_to_emit.len(), 1);
966                match &events_to_emit[0] {
967                    ServiceEvent::ServiceRequestReceived {
968                        request_id,
969                        service_id,
970                        method,
971                        ..
972                    } => {
973                        assert_eq!(request_id, "req-002");
974                        assert_eq!(service_id, "echo");
975                        assert_eq!(method, "ping");
976                    }
977                    other => panic!("Wrong event: {:?}", other),
978                }
979            }
980            ServiceAction::NotHandled => panic!("Should have been handled"),
981        }
982    }
983
984    #[test]
985    fn test_handle_response() {
986        let mut svc = MeshServices::new();
987
988        let payload = ServiceResponsePayload {
989            request_id: "req-003".to_string(),
990            service_id: "echo".to_string(),
991            status: "ok".to_string(),
992            body: "pong".to_string(),
993        };
994        let content = format!(
995            "{}{}",
996            SVC_RESPONSE,
997            serde_json::to_string(&payload).unwrap()
998        );
999
1000        let action = svc.handle_incoming_message(&content, "bob", 0, "user1", &[]);
1001        match action {
1002            ServiceAction::Consumed { events_to_emit, .. } => {
1003                assert_eq!(events_to_emit.len(), 1);
1004                match &events_to_emit[0] {
1005                    ServiceEvent::ServiceResponseReceived {
1006                        request_id, status, ..
1007                    } => {
1008                        assert_eq!(request_id, "req-003");
1009                        assert_eq!(status, "ok");
1010                    }
1011                    other => panic!("Wrong event: {:?}", other),
1012                }
1013            }
1014            ServiceAction::NotHandled => panic!("Should have been handled"),
1015        }
1016    }
1017
1018    #[test]
1019    fn test_not_handled() {
1020        let mut svc = MeshServices::new();
1021        let action = svc.handle_incoming_message("Hello, world!", "alice", 0, "user1", &[]);
1022        assert!(matches!(action, ServiceAction::NotHandled));
1023    }
1024
1025    #[test]
1026    fn test_cleanup_expired() {
1027        let mut svc = MeshServices::new();
1028        svc.seen_discovery_queries.insert(
1029            "old-query".to_string(),
1030            Instant::now() - Duration::from_secs(120),
1031        );
1032        svc.seen_discovery_order.push_back("old-query".to_string());
1033        svc.seen_discovery_queries
1034            .insert("fresh-query".to_string(), Instant::now());
1035        svc.seen_discovery_order
1036            .push_back("fresh-query".to_string());
1037
1038        svc.cleanup_expired();
1039
1040        assert!(!svc.seen_discovery_queries().contains_key("old-query"));
1041        assert!(svc.seen_discovery_queries().contains_key("fresh-query"));
1042    }
1043
1044    #[test]
1045    fn test_gossip_fanout_limit() {
1046        let mut svc = MeshServices::new();
1047
1048        let payload = ServiceDiscoveryQueryPayload {
1049            query_id: "q-fanout".to_string(),
1050            originator: "alice".to_string(),
1051            service_id: None,
1052            remaining_hops: 5,
1053        };
1054        let content = format!(
1055            "{}{}",
1056            SVC_DISCOVER_QUERY,
1057            serde_json::to_string(&payload).unwrap()
1058        );
1059
1060        // Create more peers than the fanout limit
1061        let peers: Vec<String> = (0..20).map(|i| format!("peer-{i}")).collect();
1062        let action = svc.handle_incoming_message(&content, "alice", 1, "user1", &peers);
1063        match action {
1064            ServiceAction::Consumed {
1065                messages_to_send, ..
1066            } => {
1067                // Should be capped at DISCOVERY_GOSSIP_MAX_FANOUT forwards
1068                assert_eq!(messages_to_send.len(), DISCOVERY_GOSSIP_MAX_FANOUT);
1069            }
1070            ServiceAction::NotHandled => panic!("Should have been handled"),
1071        }
1072    }
1073
1074    #[test]
1075    fn test_dedup_cap_evicts_oldest() {
1076        let mut svc = MeshServices::new();
1077
1078        // Fill to capacity using record_seen_query for consistent state
1079        let base = Instant::now() - Duration::from_secs(30);
1080        for i in 0..DISCOVERY_QUERY_MAX_DEDUP_ENTRIES {
1081            let key = format!("q-{i}");
1082            svc.seen_discovery_queries
1083                .insert(key.clone(), base + Duration::from_millis(i as u64));
1084            svc.seen_discovery_order.push_back(key);
1085        }
1086        assert_eq!(
1087            svc.seen_discovery_queries().len(),
1088            DISCOVERY_QUERY_MAX_DEDUP_ENTRIES
1089        );
1090
1091        // Insert one more — should evict the oldest (q-0)
1092        svc.record_seen_query("q-new".to_string());
1093        assert_eq!(
1094            svc.seen_discovery_queries().len(),
1095            DISCOVERY_QUERY_MAX_DEDUP_ENTRIES
1096        );
1097        assert!(!svc.seen_discovery_queries().contains_key("q-0"));
1098        assert!(svc.seen_discovery_queries().contains_key("q-new"));
1099    }
1100
1101    #[test]
1102    fn test_remaining_hops_zero_no_forwarding() {
1103        let mut svc = MeshServices::new();
1104
1105        let payload = ServiceDiscoveryQueryPayload {
1106            query_id: "q-nohop".to_string(),
1107            originator: "alice".to_string(),
1108            service_id: None,
1109            remaining_hops: 0,
1110        };
1111        let content = format!(
1112            "{}{}",
1113            SVC_DISCOVER_QUERY,
1114            serde_json::to_string(&payload).unwrap()
1115        );
1116
1117        let peers = vec!["bob".to_string(), "charlie".to_string()];
1118        let action = svc.handle_incoming_message(&content, "alice", 1, "user1", &peers);
1119        match action {
1120            ServiceAction::Consumed {
1121                messages_to_send, ..
1122            } => {
1123                // No local services, no forwarding — should be empty
1124                assert!(
1125                    messages_to_send.is_empty(),
1126                    "Should not forward when remaining_hops=0"
1127                );
1128            }
1129            ServiceAction::NotHandled => panic!("Should have been handled"),
1130        }
1131    }
1132
1133    #[test]
1134    fn test_oversized_payload_dropped() {
1135        let mut svc = MeshServices::new();
1136
1137        // Build a request with oversized body
1138        let huge_body = "x".repeat(MAX_SERVICE_PAYLOAD_SIZE + 1);
1139        let content = format!("{}{}", SVC_REQUEST, huge_body);
1140
1141        let action = svc.handle_incoming_message(&content, "alice", 0, "user1", &[]);
1142        assert!(matches!(
1143            action,
1144            ServiceAction::Consumed {
1145                messages_to_send,
1146                events_to_emit,
1147            } if messages_to_send.is_empty() && events_to_emit.is_empty()
1148        ));
1149    }
1150
1151    #[test]
1152    fn test_oversized_request_body_dropped() {
1153        let mut svc = MeshServices::new();
1154        svc.register_service(make_descriptor("echo")).unwrap();
1155
1156        let large_body = "x".repeat(MAX_SERVICE_BODY_SIZE + 1);
1157        let payload = ServiceRequestPayload {
1158            request_id: "req-big".to_string(),
1159            service_id: "echo".to_string(),
1160            method: "ping".to_string(),
1161            body: large_body,
1162        };
1163        let content = format!(
1164            "{}{}",
1165            SVC_REQUEST,
1166            serde_json::to_string(&payload).unwrap()
1167        );
1168
1169        let action = svc.handle_incoming_message(&content, "alice", 0, "user1", &[]);
1170        match action {
1171            ServiceAction::Consumed {
1172                messages_to_send,
1173                events_to_emit,
1174            } => {
1175                assert!(messages_to_send.is_empty());
1176                assert!(
1177                    events_to_emit.is_empty(),
1178                    "Should not emit event for oversized body"
1179                );
1180            }
1181            ServiceAction::NotHandled => panic!("Should have been handled"),
1182        }
1183    }
1184
1185    #[test]
1186    fn test_oversized_method_dropped() {
1187        let mut svc = MeshServices::new();
1188        svc.register_service(make_descriptor("echo")).unwrap();
1189
1190        let long_method = "m".repeat(MAX_SERVICE_METHOD_LEN + 1);
1191        let payload = ServiceRequestPayload {
1192            request_id: "req-longmethod".to_string(),
1193            service_id: "echo".to_string(),
1194            method: long_method,
1195            body: "{}".to_string(),
1196        };
1197        let content = format!(
1198            "{}{}",
1199            SVC_REQUEST,
1200            serde_json::to_string(&payload).unwrap()
1201        );
1202
1203        let action = svc.handle_incoming_message(&content, "alice", 0, "user1", &[]);
1204        match action {
1205            ServiceAction::Consumed {
1206                messages_to_send,
1207                events_to_emit,
1208            } => {
1209                assert!(messages_to_send.is_empty());
1210                assert!(
1211                    events_to_emit.is_empty(),
1212                    "Should not emit event for oversized method"
1213                );
1214            }
1215            ServiceAction::NotHandled => panic!("Should have been handled"),
1216        }
1217    }
1218
1219    #[test]
1220    fn test_handle_response_invalid_status_dropped() {
1221        let mut svc = MeshServices::new();
1222
1223        let payload = ServiceResponsePayload {
1224            request_id: "req-bad".to_string(),
1225            service_id: "echo".to_string(),
1226            status: "invalid_status".to_string(),
1227            body: "data".to_string(),
1228        };
1229        let content = format!(
1230            "{}{}",
1231            SVC_RESPONSE,
1232            serde_json::to_string(&payload).unwrap()
1233        );
1234
1235        let action = svc.handle_incoming_message(&content, "bob", 0, "user1", &[]);
1236        match action {
1237            ServiceAction::Consumed {
1238                messages_to_send,
1239                events_to_emit,
1240            } => {
1241                assert!(messages_to_send.is_empty());
1242                assert!(
1243                    events_to_emit.is_empty(),
1244                    "Should not emit event for invalid status"
1245                );
1246            }
1247            ServiceAction::NotHandled => panic!("Should have been handled"),
1248        }
1249    }
1250
1251    #[test]
1252    fn test_handle_discover_query_with_match_and_forwarding() {
1253        let mut svc = MeshServices::new();
1254        svc.register_service(ServiceDescriptor {
1255            service_id: ServiceId::new("weather").unwrap(),
1256            version: "2.0".to_string(),
1257            capabilities: HashMap::new(),
1258        })
1259        .unwrap();
1260
1261        let payload = ServiceDiscoveryQueryPayload {
1262            query_id: "q-combined".to_string(),
1263            originator: "alice".to_string(),
1264            service_id: Some("weather".to_string()),
1265            remaining_hops: 5,
1266        };
1267        let content = format!(
1268            "{}{}",
1269            SVC_DISCOVER_QUERY,
1270            serde_json::to_string(&payload).unwrap()
1271        );
1272
1273        // Provide peers to forward to (excluding sender "alice" and originator "alice")
1274        let peers = vec!["bob".to_string(), "charlie".to_string(), "dave".to_string()];
1275        let action = svc.handle_incoming_message(&content, "alice", 1, "user1", &peers);
1276        match action {
1277            ServiceAction::Consumed {
1278                messages_to_send,
1279                events_to_emit,
1280            } => {
1281                // Should have a discovery response back to sender
1282                let responses: Vec<_> = messages_to_send
1283                    .iter()
1284                    .filter(|m| m.content.starts_with(SVC_DISCOVER_RESPONSE))
1285                    .collect();
1286                assert_eq!(responses.len(), 1, "Should have one discovery response");
1287                assert_eq!(
1288                    responses[0].recipient, "alice",
1289                    "Response should go to sender"
1290                );
1291
1292                // Should also have forwarded query messages to other peers
1293                let forwards: Vec<_> = messages_to_send
1294                    .iter()
1295                    .filter(|m| m.content.starts_with(SVC_DISCOVER_QUERY))
1296                    .collect();
1297                assert_eq!(forwards.len(), 3, "Should forward to all 3 eligible peers");
1298                // Forwarded queries should NOT go to the sender
1299                assert!(
1300                    forwards.iter().all(|m| m.recipient != "alice"),
1301                    "Should not forward back to sender"
1302                );
1303
1304                // No events emitted for queries — events only on discovery responses
1305                assert!(events_to_emit.is_empty());
1306            }
1307            ServiceAction::NotHandled => panic!("Should have been handled"),
1308        }
1309    }
1310
1311    #[test]
1312    fn test_select_fanout_peers_all_when_under_limit() {
1313        let peers: Vec<String> = vec!["a".into(), "b".into(), "c".into()];
1314        let selected = select_fanout_peers(&peers, 5, "q-1", "user1");
1315        assert_eq!(selected.len(), 3);
1316    }
1317
1318    #[test]
1319    fn test_select_fanout_peers_caps_at_max() {
1320        let peers: Vec<String> = (0..50).map(|i| format!("p-{i}")).collect();
1321        let selected = select_fanout_peers(&peers, 5, "q-1", "user1");
1322        assert_eq!(selected.len(), 5);
1323    }
1324
1325    #[test]
1326    fn test_select_fanout_peers_deterministic() {
1327        let peers: Vec<String> = (0..20).map(|i| format!("p-{i}")).collect();
1328        let a = select_fanout_peers(&peers, 5, "q-1", "user1");
1329        let b = select_fanout_peers(&peers, 5, "q-1", "user1");
1330        assert_eq!(a, b, "Same inputs should produce same selection");
1331    }
1332
1333    #[test]
1334    fn test_select_fanout_peers_varies_by_user() {
1335        let peers: Vec<String> = (0..20).map(|i| format!("p-{i}")).collect();
1336        let a = select_fanout_peers(&peers, 5, "q-1", "user1");
1337        let b = select_fanout_peers(&peers, 5, "q-1", "user2");
1338        // Different users should (usually) select different subsets
1339        // This is probabilistic but with 20 peers and 5 selected, collision is unlikely
1340        assert_ne!(
1341            a, b,
1342            "Different users should typically select different peers"
1343        );
1344    }
1345
1346    #[test]
1347    fn test_discovery_response_forwarded_to_originator() {
1348        let mut svc = MeshServices::new();
1349
1350        // Response with originator != user_id → should forward
1351        let payload = ServiceDiscoveryResponsePayload {
1352            query_id: "q-200".to_string(),
1353            service_id: "svc1".to_string(),
1354            version: "1.0".to_string(),
1355            provider_peer_id: "provider1".to_string(),
1356            capabilities: HashMap::new(),
1357            hop_count: 2,
1358            originator: "original_requester".to_string(),
1359        };
1360        let content = format!(
1361            "{}{}",
1362            SVC_DISCOVER_RESPONSE,
1363            serde_json::to_string(&payload).unwrap()
1364        );
1365
1366        // We are "relay_node", not the originator
1367        let action = svc.handle_incoming_message(&content, "provider1", 0, "relay_node", &[]);
1368        match action {
1369            ServiceAction::Consumed {
1370                messages_to_send,
1371                events_to_emit,
1372            } => {
1373                // Should forward to originator, not emit event
1374                assert!(
1375                    events_to_emit.is_empty(),
1376                    "Relay node should not emit ServiceDiscovered"
1377                );
1378                assert_eq!(messages_to_send.len(), 1);
1379                assert_eq!(messages_to_send[0].recipient, "original_requester");
1380            }
1381            ServiceAction::NotHandled => panic!("Should have been handled"),
1382        }
1383    }
1384
1385    #[test]
1386    fn test_discovery_response_consumed_by_originator() {
1387        let mut svc = MeshServices::new();
1388
1389        // Response where originator == user_id → should emit event
1390        let payload = ServiceDiscoveryResponsePayload {
1391            query_id: "q-300".to_string(),
1392            service_id: "svc2".to_string(),
1393            version: "1.0".to_string(),
1394            provider_peer_id: "provider2".to_string(),
1395            capabilities: HashMap::new(),
1396            hop_count: 1,
1397            originator: "me".to_string(),
1398        };
1399        let content = format!(
1400            "{}{}",
1401            SVC_DISCOVER_RESPONSE,
1402            serde_json::to_string(&payload).unwrap()
1403        );
1404
1405        let action = svc.handle_incoming_message(&content, "relay_node", 0, "me", &[]);
1406        match action {
1407            ServiceAction::Consumed {
1408                messages_to_send,
1409                events_to_emit,
1410            } => {
1411                assert!(messages_to_send.is_empty(), "Originator should not forward");
1412                assert_eq!(events_to_emit.len(), 1);
1413                match &events_to_emit[0] {
1414                    ServiceEvent::ServiceDiscovered { service_id, .. } => {
1415                        assert_eq!(service_id, "svc2");
1416                    }
1417                    other => panic!("Wrong event: {:?}", other),
1418                }
1419            }
1420            ServiceAction::NotHandled => panic!("Should have been handled"),
1421        }
1422    }
1423}
offline_protocol_services/services.rs

offline_protocol_services/
services.rs
