apcore 0.20.0

// APCore Protocol — Event emitter
// Spec reference: Event types and emission

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

use serde::{Deserialize, Serialize};

use super::subscribers::EventSubscriber;
use crate::errors::ModuleError;

/// An event emitted by the `APCore` system.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ApCoreEvent {
    pub event_type: String,
    /// ISO 8601 timestamp string.
    pub timestamp: String,
    pub data: serde_json::Value,
    #[serde(skip_serializing_if = "Option::is_none")]
    pub module_id: Option<String>,
    pub severity: String,
}

impl ApCoreEvent {
    /// Create a new event with "info" severity.
    pub fn new(event_type: impl Into<String>, data: serde_json::Value) -> Self {
        Self {
            event_type: event_type.into(),
            timestamp: chrono::Utc::now().to_rfc3339(),
            data,
            module_id: None,
            severity: "info".to_string(),
        }
    }

    /// Create a new event with explicit `module_id` and severity.
    pub fn with_module(
        event_type: impl Into<String>,
        data: serde_json::Value,
        module_id: impl Into<String>,
        severity: impl Into<String>,
    ) -> Self {
        Self {
            event_type: event_type.into(),
            timestamp: chrono::Utc::now().to_rfc3339(),
            data,
            module_id: Some(module_id.into()),
            severity: severity.into(),
        }
    }
}

/// Manages event subscribers and dispatches events.
///
/// Subscribers are stored as `Arc<dyn EventSubscriber>` so they can be
/// cheaply cloned into spawned tasks for [`Self::emit_spawn`]'s
/// fire-and-forget dispatch model (sync finding A-D-501).
#[derive(Debug)]
pub struct EventEmitter {
    subscribers: Vec<Arc<dyn EventSubscriber>>,
    pub max_workers: usize,
    /// Set to `true` by [`Self::shutdown`]. Once set, all `emit*` methods
    /// drop incoming events as no-ops (sync finding A-D-502).
    is_shutdown: Arc<AtomicBool>,
}

impl EventEmitter {
    /// Create a new event emitter.
    #[must_use]
    pub fn new() -> Self {
        Self {
            subscribers: vec![],
            max_workers: 4,
            is_shutdown: Arc::new(AtomicBool::new(false)),
        }
    }

    /// Add a subscriber (matching Python's void return signature).
    pub fn subscribe(&mut self, subscriber: Box<dyn EventSubscriber>) {
        // Convert Box -> Arc so subscribers can be cloned into spawned
        // tasks by `emit_spawn` (sync finding A-D-501).
        self.subscribers.push(Arc::from(subscriber));
    }

    /// Remove the first subscriber whose `subscriber_id()` matches the given
    /// subscriber's ID, matching Python's identity-based removal semantics.
    pub fn unsubscribe(&mut self, subscriber: &dyn EventSubscriber) -> bool {
        let target_id = subscriber.subscriber_id();
        self.unsubscribe_by_id(target_id)
    }

    /// Remove the first subscriber whose `subscriber_id()` matches the given ID string.
    pub fn unsubscribe_by_id(&mut self, subscriber_id: &str) -> bool {
        let pos = self
            .subscribers
            .iter()
            .position(|s| s.subscriber_id() == subscriber_id);
        if let Some(i) = pos {
            self.subscribers.remove(i);
            true
        } else {
            false
        }
    }

    /// Remove all subscribers whose `event_type_filter()` equals `event_type`.
    ///
    /// Returns the number of subscribers removed. Matches Python/TypeScript
    /// `off(event_type)` semantics where passing an event-type string removes
    /// all handlers bound to that type.
    pub fn unsubscribe_by_event_type(&mut self, event_type: &str) -> usize {
        let before = self.subscribers.len();
        self.subscribers
            .retain(|s| s.event_type_filter().is_none_or(|t| t != event_type));
        before - self.subscribers.len()
    }

    /// Emit an event to all subscribers whose pattern matches the event type.
    ///
    /// Errors from individual subscribers are logged but not propagated
    /// (error isolation), matching Python's behaviour.
    ///
    /// **Post-shutdown behaviour:** if [`Self::shutdown`] has been called,
    /// this method returns immediately as a no-op (sync finding A-D-502).
    ///
    /// Spec event-system.md:448 declares `No errors raised`. The body is
    /// infallible — subscriber errors are caught and logged internally —
    /// so the return type is unit, not `Result<(), ModuleError>` (D10-008).
    pub async fn emit(&self, event: &ApCoreEvent) {
        if self.is_shutdown.load(Ordering::SeqCst) {
            return;
        }
        for subscriber in &self.subscribers {
            if Self::matches_pattern(subscriber.event_pattern(), &event.event_type) {
                if let Err(e) = subscriber.on_event(event).await {
                    tracing::warn!(
                        subscriber_id = %subscriber.subscriber_id(),
                        event_type = %event.event_type,
                        error = %e,
                        "event subscriber failed"
                    );
                }
            }
        }
    }

    /// Emit an event to subscribers matching both the caller's filter pattern
    /// AND the subscriber's own `event_pattern`.
    ///
    /// **Post-shutdown behaviour:** drops the event as a no-op (sync
    /// finding A-D-502).
    pub async fn emit_filtered(
        &self,
        event: &ApCoreEvent,
        pattern: &str,
    ) -> Result<(), ModuleError> {
        if self.is_shutdown.load(Ordering::SeqCst) {
            return Ok(());
        }
        for subscriber in &self.subscribers {
            if Self::matches_pattern(pattern, &event.event_type)
                && Self::matches_pattern(subscriber.event_pattern(), &event.event_type)
            {
                if let Err(e) = subscriber.on_event(event).await {
                    tracing::warn!(
                        subscriber_id = %subscriber.subscriber_id(),
                        event_type = %event.event_type,
                        error = %e,
                        "event subscriber failed"
                    );
                }
            }
        }
        Ok(())
    }

    /// Flush all pending events, waiting up to `timeout_ms` milliseconds.
    ///
    /// This implementation uses a synchronous dispatch model — all events are
    /// dispatched inline during `emit()`, so there is nothing to flush.
    pub fn flush(&self, _timeout_ms: u64) -> Result<(), ModuleError> {
        // Synchronous dispatch model — nothing to flush.
        Ok(())
    }

    /// Fire-and-forget dispatch: spawns one `tokio::task` per matching
    /// subscriber and returns immediately.
    ///
    /// Use this when the caller cannot wait for subscriber completion (the
    /// canonical "fire-and-forget" path called out by the spec, sync finding
    /// A-D-501). Errors from subscribers are logged via `tracing::warn!` and
    /// never propagated. Subscribers run **concurrently** because each is
    /// driven on its own task.
    ///
    /// `emit_spawn` is the preferred dispatch path for runtime modules
    /// (system modules, circuit breaker, etc.); the sequential `emit` is
    /// retained for tests that need deterministic ordering.
    ///
    /// **Post-shutdown behaviour:** drops the event as a no-op once
    /// [`Self::shutdown`] has been called (sync finding A-D-502).
    // Takes `event: ApCoreEvent` by value because each spawned task needs an
    // owned `ApCoreEvent`; passing by reference would force the caller to
    // clone, hiding the cost. The spec's pseudocode signature also
    // matches by-value (sync finding A-D-501).
    #[allow(clippy::needless_pass_by_value)]
    pub fn emit_spawn(&self, event: ApCoreEvent) {
        if self.is_shutdown.load(Ordering::SeqCst) {
            return;
        }
        for subscriber in &self.subscribers {
            if !Self::matches_pattern(subscriber.event_pattern(), &event.event_type) {
                continue;
            }
            let sub = Arc::clone(subscriber);
            let evt = event.clone();
            tokio::spawn(async move {
                if let Err(e) = sub.on_event(&evt).await {
                    tracing::warn!(
                        subscriber_id = %sub.subscriber_id(),
                        event_type = %evt.event_type,
                        error = %e,
                        "emit_spawn: subscriber on_event failed"
                    );
                }
            });
        }
    }

    /// Mark this emitter as shut down and flush any pending work.
    ///
    /// After `shutdown` returns:
    /// - Subsequent calls to `emit`, `emit_filtered`, and `emit_spawn` are
    ///   no-ops — the event is dropped.
    /// - Subsequent calls to `shutdown` return `Ok(())` immediately
    ///   (idempotent).
    ///
    /// `timeout_ms` bounds the wait for any in-flight flush. The current
    /// implementation dispatches events synchronously / via spawned tasks
    /// that the emitter does not own, so flush is a no-op pass-through —
    /// the parameter is reserved for future buffering implementations and
    /// kept for API parity with apcore-python and apcore-typescript
    /// (sync finding A-D-502).
    ///
    /// `async fn` is preserved for cross-language API parity even though
    /// the current body has no `.await` points; future buffered/spawned
    /// implementations will need to await flush completion.
    #[allow(clippy::unused_async)]
    pub async fn shutdown(&mut self, timeout_ms: u64) -> Result<(), ModuleError> {
        if self.is_shutdown.swap(true, Ordering::SeqCst) {
            // Already shut down — idempotent.
            return Ok(());
        }
        self.flush(timeout_ms)
    }

    /// Returns `true` if this emitter has been shut down.
    #[must_use]
    pub fn is_shutdown(&self) -> bool {
        self.is_shutdown.load(Ordering::SeqCst)
    }

    /// Simple glob-style pattern matching with `*` wildcard.
    ///
    /// - `"*"` matches everything.
    /// - `"foo.*"` matches `"foo.bar"`, `"foo.baz"`, etc.
    /// - An exact string matches only itself.
    fn matches_pattern(pattern: &str, event_type: &str) -> bool {
        if pattern == "*" {
            return true;
        }
        // Split pattern by '*' and check that all parts appear in order.
        let parts: Vec<&str> = pattern.split('*').collect();
        let mut remaining = event_type;
        for (i, part) in parts.iter().enumerate() {
            if part.is_empty() {
                continue;
            }
            if i == 0 {
                // First part must be a prefix.
                if let Some(rest) = remaining.strip_prefix(part) {
                    remaining = rest;
                } else {
                    return false;
                }
            } else if let Some(pos) = remaining.find(part) {
                remaining = &remaining[pos + part.len()..];
            } else {
                return false;
            }
        }
        // If pattern doesn't end with *, remaining must be empty.
        if !pattern.ends_with('*') && !remaining.is_empty() {
            return false;
        }
        true
    }
}

impl Default for EventEmitter {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use async_trait::async_trait;
    use parking_lot::Mutex;
    use serde_json::json;
    use std::sync::Arc;

    #[derive(Debug, Clone)]
    struct RecordingSubscriber {
        id: String,
        pattern: String,
        received: Arc<Mutex<Vec<String>>>,
    }

    impl RecordingSubscriber {
        fn new(id: &str, pattern: &str) -> Self {
            Self {
                id: id.to_string(),
                pattern: pattern.to_string(),
                received: Arc::new(Mutex::new(Vec::new())),
            }
        }
    }

    #[async_trait]
    impl EventSubscriber for RecordingSubscriber {
        fn subscriber_id(&self) -> &str {
            &self.id
        }
        fn event_pattern(&self) -> &str {
            &self.pattern
        }
        async fn on_event(&self, event: &ApCoreEvent) -> Result<(), ModuleError> {
            self.received.lock().push(event.event_type.clone());
            Ok(())
        }
    }

    #[test]
    fn test_event_new_defaults() {
        let event = ApCoreEvent::new("test.event", json!({"key": "val"}));
        assert_eq!(event.event_type, "test.event");
        assert_eq!(event.severity, "info");
        assert!(event.module_id.is_none());
        assert!(!event.timestamp.is_empty());
    }

    #[test]
    fn test_event_with_module() {
        let event = ApCoreEvent::with_module("err.event", json!({}), "mod.a", "error");
        assert_eq!(event.event_type, "err.event");
        assert_eq!(event.severity, "error");
        assert_eq!(event.module_id.as_deref(), Some("mod.a"));
    }

    #[test]
    fn test_event_serialization_skips_none_module_id() {
        let event = ApCoreEvent::new("test", json!(null));
        let serialized = serde_json::to_value(&event).unwrap();
        assert!(serialized.get("module_id").is_none());
    }

    #[test]
    fn test_emitter_default_max_workers() {
        let emitter = EventEmitter::new();
        assert_eq!(emitter.max_workers, 4);
    }

    #[tokio::test]
    async fn test_emit_to_matching_subscriber() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "test.*");
        let received = sub.received.clone();
        emitter.subscribe(Box::new(sub));

        let event = ApCoreEvent::new("test.hello", json!({}));
        emitter.emit(&event).await;
        assert_eq!(received.lock().len(), 1);
        assert_eq!(received.lock()[0], "test.hello");
    }

    #[tokio::test]
    async fn test_emit_skips_non_matching_subscriber() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "other.*");
        let received = sub.received.clone();
        emitter.subscribe(Box::new(sub));

        let event = ApCoreEvent::new("test.hello", json!({}));
        emitter.emit(&event).await;
        assert!(received.lock().is_empty());
    }

    #[tokio::test]
    async fn test_emit_wildcard_matches_all() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "*");
        let received = sub.received.clone();
        emitter.subscribe(Box::new(sub));

        let event = ApCoreEvent::new("anything.at.all", json!({}));
        emitter.emit(&event).await;
        assert_eq!(received.lock().len(), 1);
    }

    #[tokio::test]
    async fn test_unsubscribe_by_id() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "*");
        emitter.subscribe(Box::new(sub));
        assert!(emitter.unsubscribe_by_id("sub1"));
        assert!(!emitter.unsubscribe_by_id("sub1"));
    }

    #[tokio::test]
    async fn test_unsubscribe_removes_subscriber() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "*");
        let received = sub.received.clone();
        emitter.subscribe(Box::new(sub.clone()));
        emitter.unsubscribe(&sub);

        let event = ApCoreEvent::new("test", json!({}));
        emitter.emit(&event).await;
        assert!(received.lock().is_empty());
    }

    #[tokio::test]
    async fn test_emit_filtered() {
        let mut emitter = EventEmitter::new();
        let sub = RecordingSubscriber::new("sub1", "test.*");
        let received = sub.received.clone();
        emitter.subscribe(Box::new(sub));

        let event = ApCoreEvent::new("test.hello", json!({}));
        emitter.emit_filtered(&event, "test.*").await.unwrap();
        assert_eq!(received.lock().len(), 1);

        emitter.emit_filtered(&event, "other.*").await.unwrap();
        assert_eq!(received.lock().len(), 1);
    }

    #[test]
    fn test_flush_succeeds() {
        let emitter = EventEmitter::new();
        emitter.flush(1000).unwrap();
    }

    #[test]
    fn test_matches_pattern_wildcard() {
        assert!(EventEmitter::matches_pattern("*", "anything"));
    }

    #[test]
    fn test_matches_pattern_exact() {
        assert!(EventEmitter::matches_pattern("test.event", "test.event"));
        assert!(!EventEmitter::matches_pattern("test.event", "test.other"));
    }

    #[test]
    fn test_matches_pattern_prefix_wildcard() {
        assert!(EventEmitter::matches_pattern("test.*", "test.hello"));
        assert!(EventEmitter::matches_pattern("test.*", "test."));
        assert!(!EventEmitter::matches_pattern("test.*", "other.hello"));
    }

    #[test]
    fn test_matches_pattern_suffix_wildcard() {
        assert!(EventEmitter::matches_pattern("*.event", "test.event"));
        assert!(!EventEmitter::matches_pattern("*.event", "test.other"));
    }

    #[test]
    fn test_matches_pattern_middle_wildcard() {
        assert!(EventEmitter::matches_pattern("a.*.z", "a.b.z"));
        assert!(EventEmitter::matches_pattern("a.*.z", "a.anything.z"));
        assert!(!EventEmitter::matches_pattern("a.*.z", "a.b.c"));
    }

    #[tokio::test]
    async fn test_emit_error_isolation() {
        #[derive(Debug)]
        struct FailingSub;

        #[async_trait]
        impl EventSubscriber for FailingSub {
            fn subscriber_id(&self) -> &'static str {
                "fail"
            }
            fn event_pattern(&self) -> &'static str {
                "*"
            }
            async fn on_event(&self, _event: &ApCoreEvent) -> Result<(), ModuleError> {
                Err(ModuleError::new(
                    crate::errors::ErrorCode::GeneralInternalError,
                    "boom",
                ))
            }
        }

        let mut emitter = EventEmitter::new();
        emitter.subscribe(Box::new(FailingSub));
        let good_sub = RecordingSubscriber::new("good", "*");
        let received = good_sub.received.clone();
        emitter.subscribe(Box::new(good_sub));

        let event = ApCoreEvent::new("test", json!({}));
        emitter.emit(&event).await;
        assert_eq!(received.lock().len(), 1);
    }
}