theta 0.1.0-alpha.56

//! Actor monitoring and observation capabilities.
//!
//! This module provides a comprehensive monitoring system for observing actor
//! state changes, lifecycle events, and behavior patterns. Monitors enable
//! debugging, metrics collection, and reactive programming patterns.
//!
//! # Core Concepts
//!
//! ## Monitors
//!
//! A **monitor** is an monitor that receives notifications about actor activity.
//! Monitors can monitor:
//! - **State changes**: When actors modify their internal state
//! - **Lifecycle events**: Start, stop, restart, and error conditions
//! - **Message processing**: Timing and throughput metrics
//! - **Remote activity**: Cross-network actor communication
//!
//! ## Updates
//!
//! **Updates** are structured data sent from actors to monitors. Two main types:
//!
//! ### State Updates (`Update::State`)
//! - Contain snapshots of actor state at observation time
//! - Generated from the actor's `View` associated type
//! - Automatically created via `From<&Actor>` implementation
//! - Used for debugging, metrics, and state visualization
//!
//! ### Status Updates (`Update::Status`)
//! - Contain lifecycle and operational information
//! - Include timing data, message counts, and error states
//! - Generated automatically by the framework
//! - Used for health monitoring and performance analysis
//!
//! ## Hashing and Optimization
//!
//! ### State Hash Optimization
//! Actors can implement custom `hash_code()` methods to optimize monitoring:
//! - **Efficient change detection**: Only send updates when hash changes
//! - **Auto-generation**: When using the `#[actor]` macro with a custom `type View`,
//!   and self is `Hash`, a `hash_code()` implementation is automatically generated using `AHasher`
//!
//! ```
//! # use theta::prelude::*;
//! # use theta::__private::ahash::AHasher;
//! # use std::hash::{Hash, Hasher};
//! # #[derive(Debug, Clone, ActorArgs)]
//! # struct MyActor { critical_value: u64 }
//! # #[actor("aaaaaaaa-bbbb-cccc-dddd-111111111111")]
//! impl Actor for MyActor {
//!     fn hash_code(&self) -> u64 {
//!         // Custom hash based on significant state changes
//!         let mut hasher = AHasher::default();
//!         self.critical_value.hash(&mut hasher);
//!         hasher.finish()
//!     }
//! }
//! ```
//!
//! ### Observation Frequency
//! - **Hash-based**: Update only when state hash changes (most efficient)
//!
//! # Usage Patterns
//!
//! ## Local Actor Observation
//!
//! ```no_run
//! use theta::prelude::*;
//! use theta_flume::unbounded_anonymous;
//! # use serde::{Serialize, Deserialize};
//! # #[derive(Debug, Clone, ActorArgs)]
//! # struct MyActor;
//! # #[actor("aaaaaaaa-bbbb-cccc-dddd-222222222222")]
//! # impl Actor for MyActor {}
//!
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! // Set up monitoring channel
//! let (tx, rx) = unbounded_anonymous();
//!
//! // Start observing an actor by name
//! monitor_local::<MyActor>("my_actor", tx)?;
//!
//! // Process incoming updates
//! while let Some(update) = rx.recv().await {
//!     match update {
//!         Update::State(state) => {
//!             println!("State changed: {state:?}");
//!         },
//!         Update::Status(status) => {
//!             println!("Status: {status:?}");
//!         },
//!     }
//! }
//! # Ok(())
//! # }
//! ```
//!
//! ## Remote Actor Observation
//!
//! ```no_run
//! # use theta::prelude::*;
//! # use serde::{Serialize, Deserialize};
//! # #[derive(Debug, Clone, ActorArgs)]
//! # struct MyActor;
//! # #[actor("aaaaaaaa-bbbb-cccc-dddd-444444444444")]
//! # impl Actor for MyActor {}
//! # async fn example() -> Result<(), Box<dyn std::error::Error>> {
//! # let (tx, _rx) = theta_flume::unbounded_anonymous::<Update<MyActor>>();
//! // Monitor actor on remote peer using iroh:// URL
//! let url = "iroh://my_actor@peer_public_key";
//! monitor::<MyActor>(url, tx).await?;
//! # Ok(())
//! # }
//! ```
//!
//! ## Custom State Updateing
//!
//! ```
//! # use theta::prelude::*;
//! # use serde::{Serialize, Deserialize};
//! # use std::time::Duration;
//! # #[derive(Debug, Clone, ActorArgs)]
//! # struct DatabaseActor {
//! #     active_connections: u32,
//! #     query_count: u64,
//! #     avg_response_time: Duration,
//! # }
//! #[derive(Debug, Clone, Serialize, Deserialize)]
//! struct DatabaseStats {
//!     active_connections: u32,
//!     query_count: u64,
//!     avg_response_time: Duration,
//! }
//!
//! impl From<&DatabaseActor> for DatabaseStats {
//!     fn from(actor: &DatabaseActor) -> Self {
//!         DatabaseStats {
//!             active_connections: actor.active_connections,
//!             query_count: actor.query_count,
//!             avg_response_time: actor.avg_response_time,
//!         }
//!     }
//! }
//!
//! # #[actor("aaaaaaaa-bbbb-cccc-dddd-555555555555")]
//! impl Actor for DatabaseActor {
//!     type View = DatabaseStats;
//!
//!     fn hash_code(&self) -> u64 {
//!         // Only update when significant metrics change
//!         (self.query_count / 100) ^ self.active_connections as u64
//!     }
//! }
//! ```
use std::{any::Any, sync::LazyLock};

use theta_flume::{Receiver, Sender};
use uuid::Uuid;

#[cfg(feature = "monitor")]
use crate::base::{BindingError, MonitorError};
use crate::{
    actor::{Actor, ActorId},
    actor_instance::Cont,
    actor_ref::ActorHdl,
    compat,
    message::Escalation,
    prelude::ActorRef,
};
#[cfg(feature = "remote")]
use {
    crate::{
        base::Ident,
        remote::{
            base::{RemoteError, parse_url},
            peer::LocalPeer,
        },
    },
    iroh::PublicKey,
    serde::{Deserialize, Serialize},
    url::Url,
};

/// Global registry of active actor handles indexed by actor ID backed by a `ConcurrentMap`
/// for lock-free concurrent read/write access across monitoring, lookup, and lifecycle paths.
pub(crate) static HDLS: LazyLock<compat::ConcurrentMap<ActorId, ActorHdl>> =
    LazyLock::new(|| compat::ConcurrentMap::with_capacity(1024));

/// Type-erased update transmitter for internal use.
pub type AnyUpdateTx = Box<dyn Any + Send>;

/// Actor lifecycle and processing status information.
///
/// `Status` represents the current operational state of an actor, including
/// both normal operations and exceptional conditions. This information is
/// essential for monitoring actor health and debugging issues.
///
/// # Lifecycle States
///
/// ## Normal Operations
/// - `Processing` - Actor is actively handling messages
/// - `Paused` - Actor is temporarily suspended
/// - `WaitingSignal` - Actor is idle, waiting for new messages
/// - `Resuming` - Actor is transitioning from paused to active state
///
/// ## Supervision States
/// - `Supervising(ActorId, Escalation)` - Actor is handling a child failure
/// - `CleanupChildren` - Actor is cleaning up terminated child references
///
/// ## Error States
/// - `Panic(Escalation)` - Actor has panicked and is updateing the error
/// - `Restarting` - Actor is being restarted after a failure
/// - `Terminating` - Actor is shutting down gracefully
/// - `Terminated` - Actor has completed shutdown
///
/// # Usage in Monitoring
///
/// Status information is automatically generated during actor lifecycle events:
///
/// ```
/// # use theta::prelude::*;
/// # fn example(status: Status) {
/// match status {
///     Status::Processing => {
///         // Actor is healthy and processing messages
///     },
///     Status::Panic(escalation) => {
///         // Actor has failed, may need intervention
///         eprintln!("Actor failed: {escalation:?}");
///     },
///     Status::Supervising(child_id, escalation) => {
///         // Actor is handling child failure
///         eprintln!("Child {child_id} escalated: {escalation:?}");
///     },
///     _ => {}
/// }
/// # }
/// ```
#[derive(Debug, Clone)]
#[cfg_attr(feature = "remote", derive(Serialize, Deserialize))]
pub enum Status {
    /// Actor is processing messages
    Processing,
    /// Actor is paused
    Paused,
    /// Actor is waiting for signals
    WaitingSignal,
    /// Actor is resuming from pause
    Resuming,
    /// Actor is supervising a failed child
    Supervising(ActorId, Escalation),
    /// Actor is cleaning up child references
    CleanupChildren,
    /// Actor has panicked
    Panic(Escalation),
    /// Actor is restarting
    Restarting,
    /// Actor is being dropped
    Dropping,
    /// Actor is terminating
    Terminating,
}

/// Updates sent from actors to monitors containing state and status information.
///
/// `Update` is the primary data structure for actor monitoring. It contains
/// either a state snapshot or lifecycle status information about an actor.
///
/// # Variants
///
/// - `State(A::View)` - Contains actor state data for debugging and metrics
/// - `Status(Status)` - Contains lifecycle and operational status information
///
/// # Usage
///
/// Updates are automatically generated by the framework and sent to registered
/// monitors. The frequency and content depend on the actor's configuration:
///
/// ```no_run
/// # use theta::prelude::*;
/// # use serde::{Serialize, Deserialize};
/// # #[derive(Debug, Clone, ActorArgs)]
/// # struct MyActor;
/// # #[actor("aaaaaaaa-bbbb-cccc-dddd-666666666666")]
/// # impl Actor for MyActor {}
/// # async fn example(monitor_rx: UpdateRx<MyActor>) {
/// while let Some(update) = monitor_rx.recv().await {
///     match update {
///         Update::State(state) => {
///             // Process state data for metrics/debugging
///             println!("Actor state: {state:?}");
///         },
///         Update::Status(status) => {
///             // Handle lifecycle events
///             match status {
///                 Status::Processing => println!("Actor is healthy"),
///                 Status::Panic(escalation) => println!("Actor failed: {escalation:?}"),
///                 _ => {}
///             }
///         },
///     }
/// }
/// # }
/// ```
#[derive(Debug)]
#[cfg_attr(feature = "remote", derive(Serialize, Deserialize))]
pub enum Update<A: Actor> {
    /// Actor state snapshot
    State(A::View),
    /// Actor lifecycle status
    Status(Status),
}

/// Channel for sending actor updates to monitors.
pub type UpdateTx<A> = Sender<Update<A>>;
/// Channel for receiving actor updates from observations.
pub type UpdateRx<A> = Receiver<Update<A>>;

/// Internal monitor structure for managing monitors of an actor.
pub(crate) struct Monitor<A: Actor> {
    pub(crate) monitors: Vec<UpdateTx<A>>,
}

impl<A: Actor> Monitor<A> {
    pub fn add_monitor(&mut self, tx: UpdateTx<A>) {
        self.monitors.push(tx);
    }

    pub fn update(&mut self, update: &Update<A>) {
        self.monitors
            .retain(|tx| tx.send((*update).clone()).is_ok());
    }

    pub const fn is_monitor(&self) -> bool {
        !self.monitors.is_empty()
    }
}

impl<A: Actor> Default for Monitor<A> {
    fn default() -> Self {
        Self {
            monitors: Vec::new(),
        }
    }
}

impl<A: Actor> Clone for Update<A> {
    fn clone(&self) -> Self {
        match self {
            Self::State(state) => Self::State(state.clone()),
            Self::Status(status) => Self::Status(status.clone()),
        }
    }
}

impl From<&Cont> for Status {
    fn from(cont: &Cont) -> Self {
        match cont {
            Cont::Process => Self::Processing,
            Cont::Pause(_) => Self::Paused,
            Cont::WaitSignal => Self::WaitingSignal,
            Cont::Resume(_) => Self::Resuming,
            Cont::Supervise(_, e) => Self::Supervising(ActorId::nil(), e.clone()),
            Cont::CleanupChildren => Self::CleanupChildren,
            Cont::Panic(e) => Self::Panic(e.clone()),
            Cont::Restart(_) => Self::Restarting,
            Cont::Drop => Self::Dropping,
            Cont::Terminate(_) => Self::Terminating,
        }
    }
}

/// Monitor an actor by name or URL for both local and remote actors.
///
/// # Arguments
///
/// * `ident_or_url` - Actor name (local) or iroh:// URL (remote)
/// * `tx` - Channel to send updates to
///
/// # Return
///
/// `Result<(), RemoteError>` - Success or error during observation setup
///
/// # Errors
///
/// Returns `RemoteError` if:
/// - URL parsing fails for remote actors
/// - Network connection to remote peer fails
/// - Actor lookup fails
#[cfg(feature = "remote")]
pub async fn monitor<A: Actor>(
    ident_or_url: impl AsRef<str>,
    tx: UpdateTx<A>,
) -> Result<(), RemoteError> {
    let ident_or_url = ident_or_url.as_ref();

    match Url::parse(ident_or_url) {
        Err(_) => match Uuid::try_parse(ident_or_url) {
            Err(_) => Ok(monitor_local::<A>(ident_or_url, tx)?),
            Ok(actor_id) => match LocalPeer::inst().get_imported_peer(&actor_id) {
                None => Ok(monitor_local_id::<A>(actor_id, tx)?),
                Some(peer) => peer.monitor(*actor_id.as_bytes(), tx).await,
            },
        },
        Ok(url) => {
            let (ident, public_key) = parse_url(&url)?;

            monitor_remote::<A>(ident, public_key, tx).await
        }
    }
}

/// Monitor a local actor by name or UUID.
///
/// # Arguments
///
/// * `ident` - Actor name (as bytes) or UUID string
/// * `tx` - Channel to send updates to
///
/// # Return
///
/// `Result<(), MonitorError>` - Success or error during local observation setup
///
/// # Errors
///
/// Returns `MonitorError` if:
/// - Actor not found by the given identifier
/// - Failed to send observation signal to actor
#[cfg(feature = "monitor")]
pub fn monitor_local<A: Actor>(
    ident: impl AsRef<str>,
    tx: UpdateTx<A>,
) -> Result<(), MonitorError> {
    let ident = ident.as_ref();

    match Uuid::try_parse(ident) {
        Err(_) => {
            let actor = ActorRef::<A>::lookup_local(ident)?;

            monitor_local_id::<A>(actor.id(), tx)
        }
        Ok(actor_id) => monitor_local_id::<A>(actor_id, tx),
    }
}

/// Monitor a local actor by its unique ID.
///
/// # Arguments
///
/// * `actor_id` - The unique ID of the actor to observe
/// * `tx` - Channel to send updates to
///
/// # Return
///
/// `Result<(), MonitorError>` - Success or error during observation setup
///
/// # Errors
///
/// Returns `MonitorError` if:
/// - Actor with the given ID is not found
/// - Failed to send observation signal to actor
#[cfg(feature = "monitor")]
pub fn monitor_local_id<A: Actor>(actor_id: ActorId, tx: UpdateTx<A>) -> Result<(), MonitorError> {
    let result = HDLS
        .with(&actor_id, |hdl| hdl.monitor(Box::new(tx)))
        .ok_or(MonitorError::BindingError(BindingError::NotFound))?;

    result.map_err(|_| MonitorError::SigSendError)?;

    Ok(())
}

/// Monitor a remote actor by identifier and public key.
///
/// # Arguments
///
/// * `ident` - Actor identifier on the remote peer
/// * `public_key` - Public key of the remote peer
/// * `tx` - Channel to send updates to
///
/// # Return
///
/// `Result<(), RemoteError>` - Success or error during remote observation setup
///
/// # Errors
///
/// Returns `RemoteError` if:
/// - Connection to remote peer fails
/// - Remote actor lookup fails
#[cfg(all(feature = "monitor", feature = "remote"))]
pub async fn monitor_remote<A: Actor>(
    ident: Ident,
    public_key: PublicKey,
    tx: UpdateTx<A>,
) -> Result<(), RemoteError> {
    let peer = LocalPeer::inst().get_or_connect_peer(public_key);

    peer.monitor(ident, tx).await
}

/// Monitor a remote actor by its unique ID and public key.
///
/// This function establishes monitoring for a remote actor when you know both
/// the actor's unique ID and the public key of the peer hosting it. It connects
/// to the specified peer and sets up observation of the target actor.
///
/// # Arguments
///
/// * `actor_id` - The unique ID of the remote actor to observe
/// * `public_key` - Public key of the remote peer hosting the actor
/// * `tx` - Channel to send updates to
///
/// # Return
///
/// `Result<(), RemoteError>` - Success or error during remote observation setup
///
/// # Errors
///
/// Returns `RemoteError` if:
/// - Connection to remote peer fails
/// - Remote actor with the given ID is not found
/// - Network communication errors occur
#[cfg(all(feature = "remote", feature = "remote"))]
pub async fn monitor_remote_id<A: Actor>(
    actor_id: ActorId,
    public_key: PublicKey,
    tx: UpdateTx<A>,
) -> Result<(), RemoteError> {
    let peer = LocalPeer::inst().get_or_connect_peer(public_key);

    peer.monitor(*actor_id.as_bytes(), tx).await
}