swactor/src/actor.rs

use std::any::Any;
use std::sync::Arc;

use crate::Error;

/// The primary trait defining data that can be passed to and from actor processes
pub trait Message: 'static + Sized + Clone + Send + Sync {}
impl<T: 'static + Sized + Clone + Send + Sync> Message for T {}

pub trait ActorInterface: 'static + Send {
    type Incoming: Message;
    type Response: Message;
    fn handle(&mut self, ctx: &Ctx, msg: Self::Incoming);

    /// Called once after the actor is added to a worker, before the first message.
    /// Receives `&Ctx` so the actor can send messages or spawn children during init.
    ///
    /// If `on_start` panics, the actor is immediately poisoned (no restart attempted).
    fn on_start(&mut self, _ctx: &Ctx) {}

    /// Called when the actor is being gracefully stopped (via `ctx.stop_self()` or
    /// `Runtime::stop_actor()`), before removal from the worker pool.
    ///
    /// NOT called when an actor is poisoned by panic — panicked actors may have
    /// corrupt state and calling methods on them is unsafe.
    fn on_stop(&mut self, _ctx: &Ctx) {}
}

/// A unique address for this actor. 32 bytes is overkill for a small application,
/// but most systems are powerful, and this allows us to create a global map of
/// actor processes in the future, without worrying about collision.
#[derive(Debug, Default, Clone, Copy, PartialEq, Eq, Hash)]
#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
pub struct ActorAddress(pub [u8; 32]);

impl std::fmt::Display for ActorAddress {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        for b in &self.0[..8] {
            write!(f, "{:02x}", b)?;
        }
        write!(f, "\u{2026}")
    }
}
impl ActorAddress {
    pub fn new_random() -> Self {
        let mut bytes = [0u8; 32];
        crate::get_random(&mut bytes);
        Self(bytes)
    }
}

/// The actor process as represented in the Runtime — thin wrapper around user state.
pub struct Actor<A: ActorInterface> {
    inner: A,
    /// Factory for creating fresh instances on restart. None = not restartable.
    restart_factory: Option<Arc<dyn Fn() -> A + Send + Sync>>,
    max_restarts: u32,
    restart_count: u32,
}

impl<A: ActorInterface> Actor<A> {
    pub fn new(inner: A) -> Self {
        Self {
            inner,
            restart_factory: None,
            max_restarts: 0,
            restart_count: 0,
        }
    }

    pub fn new_restartable(
        inner: A,
        factory: Arc<dyn Fn() -> A + Send + Sync>,
        max_restarts: u32,
    ) -> Self {
        Self {
            inner,
            restart_factory: Some(factory),
            max_restarts,
            restart_count: 0,
        }
    }
}

/// Trait for type-erased actors — single-message handler.
///
/// Returns `Some(type_name)` if handled, `None` on type mismatch.
pub trait AnyActor: Send {
    fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>) -> Option<&'static str>;

    /// Attempt to create a fresh instance for restart after panic.
    /// Returns `None` if restart is not supported or restart limit exceeded.
    fn try_restart(&self) -> Option<Box<dyn AnyActor>> {
        None
    }

    /// Called once after spawn, before first message. See [`ActorInterface::on_start`].
    fn on_start(&mut self, _ctx: &Ctx) {}

    /// Called on graceful stop, before removal. See [`ActorInterface::on_stop`].
    fn on_stop(&mut self, _ctx: &Ctx) {}
}

impl<A> AnyActor for Actor<A>
where
    A: ActorInterface,
{
    fn handle_any(&mut self, ctx: &Ctx, msg: Box<dyn Any + Send>) -> Option<&'static str> {
        if let Ok(typed) = msg.downcast::<A::Incoming>() {
            self.inner.handle(ctx, *typed);
            Some(std::any::type_name::<A::Incoming>())
        } else {
            None
        }
    }

    fn try_restart(&self) -> Option<Box<dyn AnyActor>> {
        let factory = self.restart_factory.as_ref()?;
        if self.restart_count >= self.max_restarts {
            return None;
        }
        let fresh = factory();
        Some(Box::new(Actor {
            inner: fresh,
            restart_factory: Some(factory.clone()),
            max_restarts: self.max_restarts,
            restart_count: self.restart_count + 1,
        }))
    }

    fn on_start(&mut self, ctx: &Ctx) {
        self.inner.on_start(ctx);
    }

    fn on_stop(&mut self, ctx: &Ctx) {
        self.inner.on_stop(ctx);
    }
}

/// Internal sentinel message for graceful actor stop.
/// Not a `Message` — intercepted in `tick_all` before reaching `handle_any`.
pub(crate) struct StopSignal;

/// Type-erased cloneable message for interval timers.
/// Since `Message: Clone`, all actor messages can implement this.
pub(crate) trait CloneMsg: Send {
    fn clone_boxed(&self) -> Box<dyn Any + Send>;
}

impl<M: Message> CloneMsg for M {
    fn clone_boxed(&self) -> Box<dyn Any + Send> {
        Box::new(self.clone())
    }
}

/// Timer request from a handler, queued for processing after tick_all.
pub(crate) enum TimerRequest {
    /// One-shot: deliver `msg` to `dest` after `ticks` worker ticks.
    Once {
        dest: ActorAddress,
        msg: Box<dyn Any + Send>,
        ticks: u64,
    },
    /// Repeating: deliver a clone of `msg` to `dest` every `period` ticks.
    Interval {
        dest: ActorAddress,
        msg: Box<dyn CloneMsg>,
        period: u64,
    },
}

/// Object-safe inner trait for sending type-erased messages.
#[allow(private_interfaces)]
pub trait ContextInner {
    fn send_any(&self, addr: ActorAddress, msg: Box<dyn Any + Send>) -> Result<(), Error>;
    fn spawn_any(&self, addr: ActorAddress, actor: Box<dyn AnyActor>);
    /// Request graceful stop for an actor. Takes effect after the current message.
    fn request_stop(&self, addr: ActorAddress);
    /// Schedule a timer (one-shot or interval).
    fn schedule_timer(&self, request: TimerRequest);
    /// Look up an actor address by registered name.
    fn where_is(&self, name: &str) -> Option<ActorAddress>;
    /// Register a name → address mapping. Returns `Err` if the name is taken.
    fn register_name(&self, name: String, addr: ActorAddress) -> Result<(), Error>;
}

/// Actor syscall interface — passed to `ActorInterface::handle()`.
///
/// Wraps a `&dyn ContextInner` to solve the object-safety problem while
/// providing a typed public API.
pub struct Ctx<'a> {
    inner: &'a dyn ContextInner,
    self_addr: ActorAddress,
}

impl<'a> Ctx<'a> {
    pub(crate) fn new(inner: &'a dyn ContextInner, self_addr: ActorAddress) -> Self {
        Self { inner, self_addr }
    }

    pub fn raw_inner(&self) -> &dyn ContextInner {
        self.inner
    }

    /// Returns the address of the actor currently being ticked.
    pub fn self_addr(&self) -> ActorAddress {
        self.self_addr
    }

    /// Send a typed message to an actor address.
    pub fn send<M: Message>(&self, addr: ActorAddress, msg: M) -> Result<(), Error> {
        self.inner.send_any(addr, Box::new(msg))
    }

    /// Spawn a new actor, returning its address.
    pub fn spawn<A: ActorInterface>(&self, actor: A) -> Result<ActorAddress, Error> {
        let addr = ActorAddress::new_random();
        let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
        self.inner.spawn_any(addr, boxed);
        Ok(addr)
    }

    /// Request graceful stop for this actor after the current message completes.
    ///
    /// The actor's `on_stop()` hook is called and the actor is removed from the
    /// worker pool. Pending messages in the mailbox are discarded.
    pub fn stop_self(&self) {
        self.inner.request_stop(self.self_addr);
    }

    /// Schedule a one-shot timer: deliver `msg` to `addr` after `ticks` worker ticks.
    ///
    /// The message is delivered as a normal mailbox message during the fire tick,
    /// before `tick_all` processes messages. The timer is tick-counted (deterministic),
    /// not wall-clock based.
    pub fn send_after_ticks<M: Message>(&self, addr: ActorAddress, msg: M, ticks: u64) {
        self.inner.schedule_timer(TimerRequest::Once {
            dest: addr,
            msg: Box::new(msg),
            ticks,
        });
    }

    /// Schedule a repeating timer: deliver a clone of `msg` to `addr` every `period` ticks.
    ///
    /// The first delivery happens after `period` ticks. The message is cloned for each
    /// delivery. The timer continues until the target actor is stopped/poisoned.
    pub fn send_interval_ticks<M: Message>(&self, addr: ActorAddress, msg: M, period: u64) {
        self.inner.schedule_timer(TimerRequest::Interval {
            dest: addr,
            msg: Box::new(msg),
            period,
        });
    }

    /// Look up an actor address by its registered name.
    ///
    /// Returns `None` if no actor is registered under that name.
    pub fn where_is(&self, name: &str) -> Option<ActorAddress> {
        self.inner.where_is(name)
    }

    /// Spawn a new actor with a registered name.
    ///
    /// The name is reserved immediately (before the actor starts processing).
    /// Returns `Err` if the name is already taken.
    pub fn spawn_named<A: ActorInterface>(
        &self,
        name: impl Into<String>,
        actor: A,
    ) -> Result<ActorAddress, Error> {
        let addr = ActorAddress::new_random();
        self.inner.register_name(name.into(), addr)?;
        let boxed: Box<dyn AnyActor> = Box::new(Actor::new(actor));
        self.inner.spawn_any(addr, boxed);
        Ok(addr)
    }

    /// Spawn a restartable actor. On panic, recreated via `factory` up to
    /// `max_restarts` times before permanent poisoning.
    pub fn spawn_restartable<A, F>(
        &self,
        actor: A,
        factory: F,
        max_restarts: u32,
    ) -> Result<ActorAddress, Error>
    where
        A: ActorInterface,
        F: Fn() -> A + Send + Sync + 'static,
    {
        let addr = ActorAddress::new_random();
        let boxed: Box<dyn AnyActor> = Box::new(Actor::new_restartable(
            actor,
            Arc::new(factory),
            max_restarts,
        ));
        self.inner.spawn_any(addr, boxed);
        Ok(addr)
    }
}