pmat 3.30.1

#[cfg_attr(coverage_nightly, coverage(off))]
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_wildcard_matching() {
        let matcher = WildcardMatcher::new();

        assert!(matcher.pattern_matches("logs.*", "logs.error"));
        assert!(matcher.pattern_matches("logs.*", "logs.info"));
        assert!(!matcher.pattern_matches("logs.*", "metrics.cpu"));
        assert!(matcher.pattern_matches("*.*", "logs.error"));
    }

    // NOTE: the end-to-end publish test lives in `pubsub_delivery_tests` below.
    // It was disabled for years behind a TODO claiming `recipient()` was "an Actix
    // method not available on tokio::sync::mpsc::Sender" — the messaging layer is
    // Actix-native (`PubSubBroker::subscribe` takes `Recipient<AgentMessage>`), so
    // the defect was the test reaching for a tokio channel, not the layer.

    #[test]
    fn test_event_store() {
        let store = EventStore::new(100);
        let topic = Topic("test".to_string());

        for i in 0..10 {
            let event = Event {
                topic: "test".to_string(),
                data: serde_json::json!({"index": i}),
                timestamp: i,
            };
            store.store(topic.clone(), event);
        }

        let replayed = store.replay(&topic, 5);
        assert_eq!(replayed.len(), 5);
    }
}

/// End-to-end delivery tests for `PubSubBroker` against the real Actix
/// `Recipient` API (issue #968). These are the tests the old commented-out
/// `test_pubsub` was trying to be.
#[cfg_attr(coverage_nightly, coverage(off))]
#[cfg(test)]
mod pubsub_delivery_tests {
    use super::*;
    use tokio::sync::mpsc::{unbounded_channel, UnboundedReceiver, UnboundedSender};

    /// Minimal Actix actor that records every `AgentMessage` it is handed.
    struct RecordingAgent {
        seen: UnboundedSender<AgentMessage>,
    }

    impl Actor for RecordingAgent {
        type Context = Context<Self>;
    }

    impl Handler<AgentMessage> for RecordingAgent {
        type Result = Result<crate::agents::AgentResponse, crate::agents::AgentError>;

        fn handle(&mut self, msg: AgentMessage, _ctx: &mut Context<Self>) -> Self::Result {
            let _ = self.seen.send(msg);
            Ok(crate::agents::AgentResponse::Success(
                serde_json::json!({"ack": true}),
            ))
        }
    }

    #[derive(actix::Message)]
    #[rtype(result = "()")]
    struct StopAgent;

    impl Handler<StopAgent> for RecordingAgent {
        type Result = ();

        fn handle(&mut self, _msg: StopAgent, ctx: &mut Context<Self>) {
            ctx.stop();
        }
    }

    fn spawn_agent() -> (
        Addr<RecordingAgent>,
        Recipient<AgentMessage>,
        UnboundedReceiver<AgentMessage>,
    ) {
        let (tx, rx) = unbounded_channel();
        let addr = RecordingAgent { seen: tx }.start();
        let recipient = addr.clone().recipient();
        (addr, recipient, rx)
    }

    fn test_event(topic: &str, value: &str) -> Event {
        Event {
            topic: topic.to_string(),
            data: serde_json::json!({ "test": value }),
            timestamp: 0,
        }
    }

    /// Await the next delivered message, failing loudly instead of hanging.
    async fn next_message(rx: &mut UnboundedReceiver<AgentMessage>) -> AgentMessage {
        actix_rt::time::timeout(std::time::Duration::from_secs(5), rx.recv())
            .await
            .expect("timed out waiting for a published message")
            .expect("recording agent dropped its channel")
    }

    /// The restored `test_pubsub`: a real subscriber receives the published event.
    #[actix_rt::test]
    async fn test_pubsub_delivers_event_to_subscriber() {
        let broker = PubSubBroker::new();
        let topic = Topic("test.topic".to_string());
        let (_addr, recipient, mut rx) = spawn_agent();

        let agent_id = Uuid::new_v4();
        broker.subscribe(agent_id, topic.clone(), recipient);

        let sent = broker
            .publish(topic, test_event("test.topic", "data"))
            .await
            .unwrap();
        assert_eq!(sent, 1, "one live subscriber must be reported as one send");

        let delivered = next_message(&mut rx).await;
        assert_eq!(
            delivered.header.to, agent_id,
            "message must be addressed to the subscribing agent"
        );
        let event: Event = delivered.deserialize_payload().unwrap();
        assert_eq!(event.data, serde_json::json!({"test": "data"}));
    }

    /// REGRESSION (#968): `publish` used to return `subscribers.len()` — the
    /// number of ids registered on the topic — without ever checking that the
    /// message reached anything. A subscriber whose actor has stopped is an
    /// absence; reporting it as a successful send is absence-as-success.
    #[actix_rt::test]
    async fn test_publish_does_not_count_dead_subscribers() {
        let broker = PubSubBroker::new();
        let topic = Topic("dead.topic".to_string());
        let (addr, recipient, _rx) = spawn_agent();

        let agent_id = Uuid::new_v4();
        broker.subscribe(agent_id, topic.clone(), recipient.clone());

        // Stop the actor and wait until its mailbox is provably closed.
        addr.do_send(StopAgent);
        drop(addr);
        for _ in 0..500 {
            if !recipient.connected() {
                break;
            }
            actix_rt::time::sleep(std::time::Duration::from_millis(10)).await;
        }
        assert!(
            !recipient.connected(),
            "test precondition: subscriber actor must be stopped"
        );

        let sent = broker
            .publish(topic, test_event("dead.topic", "data"))
            .await
            .unwrap();
        assert_eq!(
            sent, 0,
            "a stopped subscriber received nothing — publish must report 0, not 1"
        );
    }

    /// REGRESSION (#968): subscribing the same agent to the same topic twice
    /// pushed the id twice, so the agent got the event twice and `publish`
    /// reported 2 sends to 1 subscriber.
    #[actix_rt::test]
    async fn test_duplicate_subscribe_delivers_once() {
        let broker = PubSubBroker::new();
        let topic = Topic("dup.topic".to_string());
        let (_addr, recipient, mut rx) = spawn_agent();

        let agent_id = Uuid::new_v4();
        broker.subscribe(agent_id, topic.clone(), recipient.clone());
        broker.subscribe(agent_id, topic.clone(), recipient);

        assert_eq!(
            broker
                .get_topic_stats()
                .get("dup.topic")
                .unwrap()
                .subscriber_count,
            1,
            "re-subscribing one agent must not inflate the subscriber count"
        );

        let sent = broker
            .publish(topic, test_event("dup.topic", "once"))
            .await
            .unwrap();
        assert_eq!(sent, 1, "one agent must be counted once");

        let first = next_message(&mut rx).await;
        assert_eq!(first.header.to, agent_id);

        // Nothing else may arrive. Give the runtime a chance to deliver a
        // second copy before concluding there is none.
        actix_rt::time::sleep(std::time::Duration::from_millis(50)).await;
        assert!(
            rx.try_recv().is_err(),
            "a doubly-subscribed agent must receive the event exactly once"
        );
    }

    /// REGRESSION (#968): `unsubscribe` dropped the agent from the topic list
    /// but never from `subscribers`, so every recipient (and the actor `Addr`
    /// it holds alive) leaked for the lifetime of the broker.
    #[actix_rt::test]
    async fn test_unsubscribe_releases_recipient() {
        let broker = PubSubBroker::new();
        let topic = Topic("bye.topic".to_string());
        let (_addr, recipient, _rx) = spawn_agent();

        let agent_id = Uuid::new_v4();
        broker.subscribe(agent_id, topic.clone(), recipient);
        assert_eq!(broker.subscribers.len(), 1);

        broker.unsubscribe(agent_id, &topic);

        assert!(
            broker.subscribers.is_empty(),
            "the last unsubscribe must release the recipient, not leak it"
        );
        let sent = broker
            .publish(topic, test_event("bye.topic", "data"))
            .await
            .unwrap();
        assert_eq!(sent, 0, "nobody is subscribed any more");
    }

    /// An agent subscribed to two topics keeps its recipient after leaving one.
    #[actix_rt::test]
    async fn test_unsubscribe_keeps_recipient_for_remaining_topics() {
        let broker = PubSubBroker::new();
        let topic_a = Topic("a.topic".to_string());
        let topic_b = Topic("b.topic".to_string());
        let (_addr, recipient, mut rx) = spawn_agent();

        let agent_id = Uuid::new_v4();
        broker.subscribe(agent_id, topic_a.clone(), recipient.clone());
        broker.subscribe(agent_id, topic_b.clone(), recipient);

        broker.unsubscribe(agent_id, &topic_a);

        let sent = broker
            .publish(topic_b, test_event("b.topic", "still-here"))
            .await
            .unwrap();
        assert_eq!(sent, 1, "the remaining subscription must still deliver");
        let delivered = next_message(&mut rx).await;
        let event: Event = delivered.deserialize_payload().unwrap();
        assert_eq!(event.data, serde_json::json!({"test": "still-here"}));
    }
}

#[cfg_attr(coverage_nightly, coverage(off))]
#[cfg(test)]
mod coverage_tests {
    use super::*;

    // Topic struct tests
    #[test]
    fn test_topic_creation_and_traits() {
        let topic1 = Topic("test.topic".to_string());
        let topic2 = Topic("test.topic".to_string());
        let topic3 = Topic("other.topic".to_string());

        // Test Clone
        let topic1_clone = topic1.clone();
        assert_eq!(topic1_clone.0, topic1.0);

        // Test Eq and PartialEq
        assert_eq!(topic1, topic2);
        assert_ne!(topic1, topic3);

        // Test Hash (indirectly via HashMap)
        let mut map = HashMap::new();
        map.insert(topic1.clone(), 1);
        assert_eq!(map.get(&topic2), Some(&1));
        assert_eq!(map.get(&topic3), None);

        // Test Debug
        let debug_str = format!("{:?}", topic1);
        assert!(debug_str.contains("test.topic"));
    }

    // Event struct tests
    #[test]
    fn test_event_creation_and_traits() {
        let event = Event {
            topic: "test.topic".to_string(),
            data: serde_json::json!({"key": "value", "number": 42}),
            timestamp: 1234567890,
        };

        // Test Clone
        let event_clone = event.clone();
        assert_eq!(event_clone.topic, event.topic);
        assert_eq!(event_clone.data, event.data);
        assert_eq!(event_clone.timestamp, event.timestamp);

        // Test Debug
        let debug_str = format!("{:?}", event);
        assert!(debug_str.contains("test.topic"));

        // Test Serialize/Deserialize
        let json = serde_json::to_string(&event).unwrap();
        let deserialized: Event = serde_json::from_str(&json).unwrap();
        assert_eq!(deserialized.topic, event.topic);
        assert_eq!(deserialized.timestamp, event.timestamp);
    }

    // PubSubBroker tests
    #[test]
    fn test_pubsub_broker_default() {
        let broker = PubSubBroker::default();
        let stats = broker.get_topic_stats();
        assert!(stats.is_empty());
    }

    #[test]
    fn test_pubsub_broker_unsubscribe_from_nonexistent_topic() {
        let broker = PubSubBroker::new();
        let agent_id = Uuid::new_v4();
        let topic = Topic("nonexistent".to_string());

        // Should not panic when unsubscribing from non-existent topic
        broker.unsubscribe(agent_id, &topic);
        let stats = broker.get_topic_stats();
        assert!(stats.is_empty());
    }

    #[test]
    fn test_pubsub_broker_get_topic_stats_with_subscribers() {
        let broker = PubSubBroker::new();
        let topic1 = Topic("topic.one".to_string());
        let topic2 = Topic("topic.two".to_string());

        // We can't easily add subscribers without Actix recipients,
        // but we can test the empty stats path
        let stats = broker.get_topic_stats();
        assert!(stats.is_empty());

        // Manually add to topics map for stats testing
        broker.topics.entry(topic1.clone()).or_default();
        broker.topics.entry(topic2.clone()).or_default();

        let stats = broker.get_topic_stats();
        assert_eq!(stats.len(), 2);
        assert!(stats.contains_key("topic.one"));
        assert!(stats.contains_key("topic.two"));
    }

    // TopicStats tests
    #[test]
    fn test_topic_stats_traits() {
        let stats = TopicStats {
            topic_name: "test.topic".to_string(),
            subscriber_count: 5,
        };

        // Test Clone
        let stats_clone = stats.clone();
        assert_eq!(stats_clone.topic_name, stats.topic_name);
        assert_eq!(stats_clone.subscriber_count, stats.subscriber_count);

        // Test Debug
        let debug_str = format!("{:?}", stats);
        assert!(debug_str.contains("test.topic"));
        assert!(debug_str.contains("5"));
    }

    // PubSubError tests
    #[test]
    fn test_pubsub_error_display() {
        let err = PubSubError::NoSubscribers;
        let display_str = format!("{}", err);
        assert!(display_str.contains("No subscribers"));

        // Test Debug
        let debug_str = format!("{:?}", err);
        assert!(debug_str.contains("NoSubscribers"));
    }

    #[test]
    fn test_pubsub_error_serialization_variant() {
        // Create a serialization error by serializing a known-bad value
        // Use a size hint that will trigger a bincode error during deserialization
        let bad_data = vec![0xFFu8; 16]; // Invalid serialized data
        let result: Result<String, _> = rmp_serde::from_slice(&bad_data);
        if let Err(e) = result {
            let pubsub_err = PubSubError::Serialization(e);
            let display_str = format!("{}", pubsub_err);
            assert!(display_str.contains("Serialization error"));
        }
    }

    // WildcardMatcher tests
    #[test]
    fn test_wildcard_matcher_default() {
        let matcher = WildcardMatcher::default();
        assert!(matcher.matches("any.topic").is_empty());
    }

    #[test]
    fn test_wildcard_matcher_add_pattern() {
        let mut matcher = WildcardMatcher::new();
        let agent1 = Uuid::new_v4();
        let agent2 = Uuid::new_v4();

        matcher.add_pattern("logs.*".to_string(), agent1);
        matcher.add_pattern("metrics.*".to_string(), agent2);
        matcher.add_pattern("logs.*".to_string(), agent2); // Duplicate pattern

        // Test matches returns correct agents
        let logs_agents = matcher.matches("logs.error");
        assert!(logs_agents.contains(&agent1));
        assert!(logs_agents.contains(&agent2));
        assert_eq!(logs_agents.len(), 2);

        let metrics_agents = matcher.matches("metrics.cpu");
        assert!(metrics_agents.contains(&agent2));
        assert_eq!(metrics_agents.len(), 1);

        // No match
        let no_match = matcher.matches("other.topic");
        assert!(no_match.is_empty());
    }

    #[test]
    fn test_wildcard_pattern_matching_edge_cases() {
        let matcher = WildcardMatcher::new();

        // Test length mismatch
        assert!(!matcher.pattern_matches("a.b.c", "a.b"));
        assert!(!matcher.pattern_matches("a.b", "a.b.c"));

        // Test exact match
        assert!(matcher.pattern_matches("exact.match", "exact.match"));

        // Test wildcard in different positions
        assert!(matcher.pattern_matches("*.second", "first.second"));
        assert!(matcher.pattern_matches("first.*", "first.anything"));

        // Test multiple wildcards
        assert!(matcher.pattern_matches("*.*", "any.thing"));
        assert!(matcher.pattern_matches("*.*.*", "a.b.c"));

        // Test no wildcards - exact match required
        assert!(!matcher.pattern_matches("exact.match", "exact.other"));
    }

    // EventStore tests
    #[test]
    fn test_event_store_trim_on_max_exceeded() {
        let store = EventStore::new(5);
        let topic = Topic("test".to_string());

        // Add more events than max
        for i in 0..10 {
            let event = Event {
                topic: "test".to_string(),
                data: serde_json::json!({"index": i}),
                timestamp: i as u64,
            };
            store.store(topic.clone(), event);
        }

        // Should only have 5 most recent events (timestamps 5-9)
        let all_events = store.replay(&topic, 0);
        assert_eq!(all_events.len(), 5);

        // Verify we have the most recent events
        let timestamps: Vec<u64> = all_events.iter().map(|e| e.timestamp).collect();
        assert!(timestamps.iter().all(|&t| t >= 5));
    }

    #[test]
    fn test_event_store_replay_with_different_topics() {
        let store = EventStore::new(100);
        let topic1 = Topic("topic1".to_string());
        let topic2 = Topic("topic2".to_string());

        // Store events for different topics
        for i in 0..5 {
            let event1 = Event {
                topic: "topic1".to_string(),
                data: serde_json::json!({"topic": 1, "index": i}),
                timestamp: i as u64,
            };
            store.store(topic1.clone(), event1);

            let event2 = Event {
                topic: "topic2".to_string(),
                data: serde_json::json!({"topic": 2, "index": i}),
                timestamp: i as u64,
            };
            store.store(topic2.clone(), event2);
        }

        // Replay should filter by topic
        let topic1_events = store.replay(&topic1, 0);
        assert_eq!(topic1_events.len(), 5);
        assert!(topic1_events.iter().all(|e| e.topic == "topic1"));

        let topic2_events = store.replay(&topic2, 0);
        assert_eq!(topic2_events.len(), 5);
        assert!(topic2_events.iter().all(|e| e.topic == "topic2"));
    }

    #[test]
    fn test_event_store_replay_with_since_filter() {
        let store = EventStore::new(100);
        let topic = Topic("test".to_string());

        for i in 0..10 {
            let event = Event {
                topic: "test".to_string(),
                data: serde_json::json!({"index": i}),
                timestamp: i as u64 * 100, // 0, 100, 200, ..., 900
            };
            store.store(topic.clone(), event);
        }

        // Replay since timestamp 500 should return events with timestamps >= 500
        let recent_events = store.replay(&topic, 500);
        assert_eq!(recent_events.len(), 5); // timestamps 500, 600, 700, 800, 900

        // Replay since 0 should return all
        let all_events = store.replay(&topic, 0);
        assert_eq!(all_events.len(), 10);

        // Replay since future should return none
        let future_events = store.replay(&topic, 1000);
        assert_eq!(future_events.len(), 0);
    }

    #[test]
    fn test_event_store_empty_replay() {
        let store = EventStore::new(100);
        let topic = Topic("empty".to_string());

        let events = store.replay(&topic, 0);
        assert!(events.is_empty());
    }

    #[test]
    fn test_stored_event_clone() {
        let event = Event {
            topic: "test".to_string(),
            data: serde_json::json!({"key": "value"}),
            timestamp: 12345,
        };

        let stored = StoredEvent {
            event: event.clone(),
            topic: Topic("test".to_string()),
            timestamp: 12345,
        };

        // Test Clone
        let stored_clone = stored.clone();
        assert_eq!(stored_clone.timestamp, stored.timestamp);
        assert_eq!(stored_clone.topic, stored.topic);
    }
}