murmer 0.4.1

//! Core actor traits and types.
//!
//! This module defines the foundational traits that every murmer actor implements:
//!
//! - [`Actor`] — the stateful message processor trait
//! - [`Message`] — defines a request type and its associated response
//! - [`RemoteMessage`] — extends `Message` with serialization for wire transport
//! - [`Handler<M>`] — declares that an actor can handle message type `M`
//! - [`RemoteDispatch`] — generated by `#[handlers]`, routes serialized messages by type ID
//! - [`ActorContext<A>`] — per-invocation context providing identity, self-endpoint, and receptionist access
//! - [`ActorRef<A>`] — serializable actor identity that can be embedded in messages and sent across nodes
//!
//! # Defining an actor
//!
//! ```rust,ignore
//! use murmer::prelude::*;
//!
//! #[derive(Debug)]
//! struct Counter;
//!
//! struct CounterState { count: i64 }
//!
//! impl Actor for Counter {
//!     type State = CounterState;
//! }
//! ```
//!
//! The actor struct itself is typically a zero-sized type — all mutable state
//! lives in `Actor::State`. This separation enables the supervisor to manage
//! state across restarts independently of the actor's handler logic.

use std::any::Any;
use std::fmt::Debug;
use std::future::Future;
use std::marker::PhantomData;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;

use crate::endpoint::Endpoint;
use crate::lifecycle::{ActorTerminated, SystemSignal};
use crate::receptionist::Receptionist;
use crate::runtime::SpawnHandle;
use crate::wire::EnvelopeProxy;

// =============================================================================
// CORE TRAITS
// =============================================================================

/// A message that can be sent to an actor. Defines its response type.
///
/// Implement this directly or use `#[derive(Message)]` from `murmer_macros`:
///
/// ```rust,ignore
/// // Manual implementation
/// impl Message for Increment {
///     type Result = i64;
/// }
///
/// // Or with the derive macro
/// #[derive(Debug, Clone, Serialize, Deserialize, Message)]
/// #[message(result = i64)]
/// struct Increment { amount: i64 }
/// ```
pub trait Message: Debug + Send + 'static {
    type Result: Send + 'static;
}

/// A message that can cross the wire. Requires serialization for both
/// the message and its result. `TYPE_ID` is the dispatch key on the receiver.
///
/// Use `#[derive(Message)]` with `remote = "..."` to implement both `Message`
/// and `RemoteMessage` at once:
///
/// ```rust,ignore
/// #[derive(Debug, Clone, Serialize, Deserialize, Message)]
/// #[message(result = i64, remote = "counter::Increment")]
/// struct Increment { amount: i64 }
/// ```
///
/// The `TYPE_ID` must be stable across versions — it's the string used to
/// match incoming wire messages to the correct deserializer. Choose something
/// descriptive like `"module::MessageName"`.
pub trait RemoteMessage: Message + Serialize + DeserializeOwned
where
    Self::Result: Serialize + DeserializeOwned,
{
    const TYPE_ID: &'static str;
}

/// An actor — a stateful message processor.
///
/// Actors in murmer separate identity (the struct) from state (the `State` type).
/// The actor struct defines behavior through [`Handler`] implementations,
/// while the state type holds all mutable data.
///
/// # Lifecycle
///
/// Override [`on_actor_terminated`](Actor::on_actor_terminated) to react when
/// a watched actor terminates (set up watches via [`ActorContext::watch`]).
///
/// # Examples
///
/// ```rust,ignore
/// use murmer::prelude::*;
///
/// #[derive(Debug)]
/// struct Counter;
///
/// struct CounterState { count: i64 }
///
/// impl Actor for Counter {
///     type State = CounterState;
/// }
///
/// // Define a message and handler
/// #[derive(Debug, Clone, Serialize, Deserialize, Message)]
/// #[message(result = i64, remote = "counter::Increment")]
/// struct Increment { amount: i64 }
///
/// impl Handler<Increment> for Counter {
///     fn handle(&self, _ctx: &ActorContext<Self>, state: &mut CounterState, msg: Increment) -> i64 {
///         state.count += msg.amount;
///         state.count
///     }
/// }
/// ```
pub trait Actor: Send + 'static {
    type State: Send + 'static;

    /// Called when a watched actor terminates. Override to react to failures.
    fn on_actor_terminated(&mut self, _state: &mut Self::State, _terminated: &ActorTerminated) {
        // default: no-op — actors opt in by overriding
    }
}

// =============================================================================
// SCHEDULE HANDLE
// =============================================================================

/// Handle to a scheduled timer. Cancels the timer when dropped.
///
/// Returned by [`ActorContext::schedule_once`] and [`ActorContext::schedule_repeat`].
/// Store it in actor state to keep the schedule alive; drop it to cancel.
///
/// # Examples
///
/// ```rust,ignore
/// struct MyState {
///     // Dropping the state cancels the schedule automatically.
///     maintenance: Option<ScheduleHandle>,
/// }
/// ```
pub struct ScheduleHandle {
    handle: SpawnHandle,
    /// Set to request cancellation. The scheduled future checks this
    /// cooperatively, so cancellation works even on a runtime (sim) whose
    /// tasks cannot be aborted from outside.
    cancel: Arc<AtomicBool>,
    /// Set by the scheduled future when it finishes (once-fired or loop-exited).
    done: Arc<AtomicBool>,
}

impl ScheduleHandle {
    /// Cancel the scheduled timer explicitly.
    pub fn cancel(self) {
        self.cancel.store(true, Ordering::Relaxed);
        self.handle.abort();
    }

    /// Returns `true` if the timer has not yet fired/exited and has not been
    /// cancelled.
    pub fn is_active(&self) -> bool {
        !self.cancel.load(Ordering::Relaxed) && !self.done.load(Ordering::Relaxed)
    }
}

impl Drop for ScheduleHandle {
    fn drop(&mut self) {
        self.cancel.store(true, Ordering::Relaxed);
        self.handle.abort();
    }
}

impl std::fmt::Debug for ScheduleHandle {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ScheduleHandle")
            .field("active", &self.is_active())
            .finish()
    }
}

// =============================================================================
// ACTOR CONTEXT
// =============================================================================

/// Intrinsic context available to every actor handler invocation.
///
/// Provides:
/// - [`label()`](ActorContext::label) — this actor's label in the receptionist
/// - [`node_id()`](ActorContext::node_id) — which node this actor runs on
/// - [`endpoint()`](ActorContext::endpoint) — a self-endpoint for passing identity
/// - [`receptionist()`](ActorContext::receptionist) — for looking up other actors
/// - [`watch()`](ActorContext::watch) — Erlang-style actor monitoring
/// - [`watch_with_tag()`](ActorContext::watch_with_tag) — actor monitoring with role tag
/// - [`schedule_once()`](ActorContext::schedule_once) — delayed self-message
/// - [`schedule_repeat()`](ActorContext::schedule_repeat) — repeating self-message
/// - [`actor_ref()`](ActorContext::actor_ref) — serializable reference for embedding in messages
///
/// # Examples
///
/// ```rust,ignore
/// impl Handler<Ping> for MyActor {
///     fn handle(&self, ctx: &ActorContext<Self>, state: &mut MyState, _msg: Ping) -> String {
///         // Access actor identity
///         let label = ctx.label();
///         let node = ctx.node_id();
///
///         // Look up other actors
///         if let Some(peer) = ctx.receptionist().lookup::<OtherActor>("peer/0") {
///             ctx.spawn(async move { peer.send(Notify).await.ok(); });
///         }
///
///         // Watch another actor for termination
///         ctx.watch("worker/0");
///
///         format!("{label}@{node}")
///     }
/// }
/// ```
pub struct ActorContext<A: Actor> {
    pub(crate) label: String,
    pub(crate) node_id: String,
    pub(crate) mailbox_tx: mpsc::UnboundedSender<Box<dyn EnvelopeProxy<A>>>,
    pub(crate) receptionist: Receptionist,
    pub(crate) system_tx: mpsc::UnboundedSender<SystemSignal>,
    /// Per-message reply token. Set by the envelope before each handler call,
    /// cleared after. Interior mutability because ctx is passed by `&`.
    pub(crate) reply_token: Mutex<Option<Box<dyn Any + Send>>>,
}

impl<A: Actor + 'static> ActorContext<A> {
    /// This actor's label in the receptionist.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// fn handle(&self, ctx: &ActorContext<Self>, _state: &mut MyState, _msg: Ping) -> String {
    ///     format!("Hello from {}", ctx.label())
    /// }
    /// ```
    pub fn label(&self) -> &str {
        &self.label
    }

    /// The node ID this actor is running on.
    pub fn node_id(&self) -> &str {
        &self.node_id
    }

    /// Get an endpoint to this actor (useful for self-sends or passing identity).
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// fn handle(&self, ctx: &ActorContext<Self>, _state: &mut MyState, msg: RegisterSelf) -> () {
    ///     let self_ep = ctx.endpoint();
    ///     msg.coordinator.send(Register { worker: self_ep }).await.ok();
    /// }
    /// ```
    pub fn endpoint(&self) -> Endpoint<A> {
        Endpoint::local(self.mailbox_tx.clone())
    }

    /// Access the receptionist for lookups or subscriptions.
    pub fn receptionist(&self) -> &Receptionist {
        &self.receptionist
    }

    /// Watch another actor for termination. Erlang-style one-shot monitor.
    ///
    /// When the watched actor terminates (for any reason), this actor's
    /// [`Actor::on_actor_terminated`] method is called with the termination details.
    ///
    /// If the watched actor doesn't exist at the time of the call, the
    /// termination notification fires immediately (Erlang semantics).
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// fn handle(&self, ctx: &ActorContext<Self>, _state: &mut MyState, msg: WatchWorker) -> () {
    ///     ctx.watch(&msg.worker_label);
    /// }
    ///
    /// // Then handle termination in the Actor trait:
    /// fn on_actor_terminated(&mut self, state: &mut MyState, event: &ActorTerminated) {
    ///     state.dead_workers.push(event.label.clone());
    /// }
    /// ```
    pub fn watch(&self, label: &str) {
        self.receptionist
            .add_watch(label, &self.label, self.system_tx.clone());
    }

    /// Watch another actor with a role tag.
    ///
    /// Same as [`watch`](Self::watch) but attaches `tag` to the notification.
    /// When the watched actor terminates, `terminated.tag` is `Some(tag)`,
    /// letting the watcher identify the role without parsing label strings.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// // In a supervisor:
    /// ctx.watch_with_tag("writer/ns/container", "writer");
    /// ctx.watch_with_tag("reader/ns/container/0", "reader");
    ///
    /// fn on_actor_terminated(&mut self, state: &mut State, event: &ActorTerminated) {
    ///     match event.tag.as_deref() {
    ///         Some("writer") => { /* restart writer */ }
    ///         Some("reader") => { /* restart reader */ }
    ///         _ => {}
    ///     }
    /// }
    /// ```
    pub fn watch_with_tag(&self, label: &str, tag: impl Into<String>) {
        self.receptionist.add_watch_with_tag(
            label,
            &self.label,
            self.system_tx.clone(),
            tag.into(),
        );
    }

    /// Schedule a message to be delivered once after `delay`.
    ///
    /// The returned [`ScheduleHandle`] cancels the timer when dropped. If you
    /// don't need to cancel, you can discard it with `let _ = ...`.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// fn handle(&self, ctx: &ActorContext<Self>, state: &mut MyState, _msg: StartTimer) -> () {
    ///     state.timeout = Some(ctx.schedule_once(Duration::from_secs(5), Timeout));
    /// }
    /// ```
    pub fn schedule_once<M>(&self, delay: Duration, message: M) -> ScheduleHandle
    where
        M: Message,
        A: Handler<M>,
    {
        let tx = self.mailbox_tx.clone();
        let runtime = self.receptionist.runtime().clone();
        let timer = runtime.clone();
        let cancel = Arc::new(AtomicBool::new(false));
        let done = Arc::new(AtomicBool::new(false));
        let (c, d) = (cancel.clone(), done.clone());
        let handle = runtime.spawn(Box::pin(async move {
            timer.sleep(delay).await;
            if !c.load(Ordering::Relaxed) {
                let (reply_tx, _reply_rx) = tokio::sync::oneshot::channel();
                let envelope = crate::wire::TypedEnvelope {
                    message,
                    respond_to: reply_tx,
                };
                tx.send(Box::new(envelope)).ok();
            }
            d.store(true, Ordering::Relaxed);
        }));
        ScheduleHandle {
            handle,
            cancel,
            done,
        }
    }

    /// Schedule a message to be delivered repeatedly at `interval`.
    ///
    /// The timer fires immediately after the first `interval` elapses (no
    /// initial skip). The returned [`ScheduleHandle`] cancels the timer when
    /// dropped — store it in actor state to keep the schedule alive.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// fn on_start(&self, ctx: &ActorContext<Self>, state: &mut MyState) {
    ///     state.heartbeat = Some(ctx.schedule_repeat(Duration::from_secs(1), Heartbeat));
    /// }
    /// ```
    pub fn schedule_repeat<M>(&self, interval: Duration, message: M) -> ScheduleHandle
    where
        M: Message + Clone,
        A: Handler<M>,
    {
        let tx = self.mailbox_tx.clone();
        let runtime = self.receptionist.runtime().clone();
        let timer = runtime.clone();
        let cancel = Arc::new(AtomicBool::new(false));
        let done = Arc::new(AtomicBool::new(false));
        let (c, d) = (cancel.clone(), done.clone());
        let handle = runtime.spawn(Box::pin(async move {
            // Fire after each `interval` elapses (no immediate first tick),
            // matching the previous tokio-interval behavior.
            loop {
                timer.sleep(interval).await;
                if c.load(Ordering::Relaxed) {
                    break;
                }
                let (reply_tx, _reply_rx) = tokio::sync::oneshot::channel();
                let envelope = crate::wire::TypedEnvelope {
                    message: message.clone(),
                    respond_to: reply_tx,
                };
                if tx.send(Box::new(envelope)).is_err() {
                    break;
                }
            }
            d.store(true, Ordering::Relaxed);
        }));
        ScheduleHandle {
            handle,
            cancel,
            done,
        }
    }

    /// Request this actor to stop itself.
    ///
    /// The stop is asynchronous: the current handler completes normally,
    /// and the supervisor exits on its next loop iteration. The actor
    /// terminates with [`TerminationReason::Stopped`](crate::TerminationReason::Stopped) — a `Transient`
    /// restart policy will *not* restart it, but `Permanent` will.
    ///
    /// Idempotent: calling `stop()` multiple times is harmless (the
    /// channel may already be closed).
    pub fn stop(&self) {
        let _ = self.system_tx.send(SystemSignal::Stop);
    }

    /// Spawn a fire-and-forget task on the actor's runtime.
    ///
    /// Use this for background work that doesn't need to block the handler.
    /// The spawned future runs independently of the actor's lifecycle, on the
    /// same [`Runtime`](crate::runtime::Runtime) the system was built with — so
    /// under a deterministic sim runtime it is scheduled deterministically too.
    ///
    /// The future must be fire-and-forget (`Output = ()`); to surface a result,
    /// send it back through an endpoint. Returns a [`SpawnHandle`] that can
    /// request cancellation.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// fn handle(&self, ctx: &ActorContext<Self>, _state: &mut MyState, msg: Notify) -> () {
    ///     let ep = ctx.endpoint();
    ///     ctx.spawn(async move {
    ///         ep.send(CheckTimeout).await.ok();
    ///     });
    /// }
    /// ```
    pub fn spawn<F>(&self, future: F) -> SpawnHandle
    where
        F: Future<Output = ()> + Send + 'static,
    {
        self.receptionist.runtime().spawn(Box::pin(future))
    }

    /// Get a serializable reference to this actor.
    ///
    /// The returned [`ActorRef<A>`] can be embedded in messages and sent to
    /// other actors or nodes. Resolve it back to a live endpoint via
    /// [`ActorRef::resolve`].
    pub fn actor_ref(&self) -> ActorRef<A> {
        ActorRef {
            label: self.label.clone(),
            node_id: self.node_id.clone(),
            _phantom: PhantomData,
        }
    }

    /// Reply to the current message immediately. The handler's return value
    /// will be silently discarded.
    ///
    /// Returns `true` if the reply was sent, `false` if already consumed
    /// (e.g., by a previous `reply()` call or `forward()`).
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// fn handle(&self, ctx: &ActorContext<Self>, state: &mut MyState, msg: GetCached) -> String {
    ///     if let Some(cached) = state.cache.get(&msg.key) {
    ///         ctx.reply(cached.clone()); // reply early
    ///         return String::new();      // return value is discarded
    ///     }
    ///     "miss".to_string() // normal return-value reply
    /// }
    /// ```
    ///
    /// # Panics
    ///
    /// Panics if `R` doesn't match the current message's result type.
    pub fn reply<R: Send + 'static>(&self, value: R) -> bool {
        let mut token = self.reply_token.lock().unwrap();
        if let Some(any) = token.take() {
            let sender = any
                .downcast::<crate::wire::ReplySender<R>>()
                .expect("ctx.reply() called with wrong type — R must match M::Result");
            sender.send(value)
        } else {
            tracing::debug!("ctx.reply() called but reply already consumed");
            false
        }
    }

    /// Extract the raw reply channel for advanced patterns (scatter-gather,
    /// deferred reply from a spawned task, etc.).
    ///
    /// After calling this, the handler's return value will be discarded.
    /// The caller is responsible for eventually calling `sender.send(value)`.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// fn handle(&self, ctx: &ActorContext<Self>, _state: &mut MyState, msg: SlowQuery) -> String {
    ///     let sender = ctx.reply_sender::<String>();
    ///     ctx.spawn(async move {
    ///         let result = expensive_io(msg.query).await;
    ///         sender.send(result); // reply from background task
    ///     });
    ///     String::new() // discarded
    /// }
    /// ```
    ///
    /// # Panics
    ///
    /// Panics if `R` doesn't match the current message's result type,
    /// or if the reply was already consumed.
    pub fn reply_sender<R: Send + 'static>(&self) -> crate::wire::ReplySender<R> {
        let mut token = self.reply_token.lock().unwrap();
        if let Some(any) = token.take() {
            *any.downcast::<crate::wire::ReplySender<R>>()
                .expect("ctx.reply_sender() called with wrong type — R must match M::Result")
        } else {
            panic!("ctx.reply_sender() called but reply already consumed");
        }
    }

    /// Forward the current message's reply responsibility to another local actor.
    ///
    /// The target actor handles `msg` and its response goes directly to the
    /// original caller. This actor's return value is discarded.
    ///
    /// Only works for local endpoints. Returns `false` if the reply was already
    /// consumed or the target's mailbox is closed.
    ///
    /// # Examples
    ///
    /// ```rust,ignore
    /// fn handle(&self, ctx: &ActorContext<Self>, state: &mut RouterState, msg: Request) -> Response {
    ///     // Route to the appropriate backend — reply goes directly to caller
    ///     let backend = &state.backends[msg.shard % state.backends.len()];
    ///     ctx.forward(backend, msg);
    ///     Response::default() // discarded
    /// }
    /// ```
    pub fn forward<B, M>(&self, endpoint: &Endpoint<B>, message: M) -> bool
    where
        B: Handler<M> + 'static,
        M: Message + Send + 'static,
        M::Result: Send + 'static,
    {
        let mut token = self.reply_token.lock().unwrap();
        if let Some(any) = token.take() {
            let sender = *any
                .downcast::<crate::wire::ReplySender<M::Result>>()
                .expect("ctx.forward() called with wrong result type");
            endpoint.send_with_reply_sender(message, sender)
        } else {
            tracing::debug!("ctx.forward() called but reply already consumed");
            false
        }
    }
}

// =============================================================================
// ACTOR REF — serializable distributed actor identity
// =============================================================================

/// A serializable, typed reference to an actor. Can be embedded in messages
/// and sent across the wire to other nodes.
///
/// # Usage
///
/// ```rust,ignore
/// // Create from context inside a handler
/// let my_ref = ctx.actor_ref();
///
/// // Or create directly
/// let actor_ref = ActorRef::<CounterActor>::new("counter/main", "node-1");
///
/// // Serialize and send across the wire (e.g., in a message field)
/// let encoded = bincode::serde::encode_to_vec(&actor_ref, bincode::config::standard())?;
///
/// // On the receiving side, resolve to a live endpoint
/// let endpoint = actor_ref.resolve(&receptionist).unwrap();
/// let count = endpoint.send(GetCount).await?;
/// ```
#[derive(Debug)]
pub struct ActorRef<A: Actor> {
    pub label: String,
    pub node_id: String,
    _phantom: PhantomData<A>,
}

impl<A: Actor> Clone for ActorRef<A> {
    fn clone(&self) -> Self {
        Self {
            label: self.label.clone(),
            node_id: self.node_id.clone(),
            _phantom: PhantomData,
        }
    }
}

/// Wire format for ActorRef serialization (no PhantomData).
#[derive(Serialize, Deserialize)]
struct ActorRefWire {
    label: String,
    node_id: String,
}

impl<A: Actor> Serialize for ActorRef<A> {
    fn serialize<S: serde::Serializer>(&self, serializer: S) -> Result<S::Ok, S::Error> {
        ActorRefWire {
            label: self.label.clone(),
            node_id: self.node_id.clone(),
        }
        .serialize(serializer)
    }
}

impl<'de, A: Actor> Deserialize<'de> for ActorRef<A> {
    fn deserialize<D: serde::Deserializer<'de>>(deserializer: D) -> Result<Self, D::Error> {
        let wire = ActorRefWire::deserialize(deserializer)?;
        Ok(ActorRef {
            label: wire.label,
            node_id: wire.node_id,
            _phantom: PhantomData,
        })
    }
}

impl<A: Actor + 'static> ActorRef<A> {
    /// Create a new ActorRef with the given label and node ID.
    pub fn new(label: impl Into<String>, node_id: impl Into<String>) -> Self {
        Self {
            label: label.into(),
            node_id: node_id.into(),
            _phantom: PhantomData,
        }
    }

    /// Resolve this reference to a live Endpoint via the receptionist.
    /// Returns `None` if the actor is not registered or has a different type.
    pub fn resolve(&self, receptionist: &Receptionist) -> Option<Endpoint<A>> {
        receptionist.lookup::<A>(&self.label)
    }
}

// =============================================================================
// MIGRATABLE ACTOR — opt-in for remote spawn and migration
// =============================================================================

/// Marker trait for actors whose state can be serialized and sent to another node.
///
/// Implementing this trait opts your actor into the orchestrator's placement and
/// migration capabilities. Without it, the actor is pinned to whatever node
/// starts it — it cannot be remotely spawned or moved.
///
/// # Requirements
///
/// Your actor's `State` type must implement `Serialize` and `DeserializeOwned`.
/// The actor struct itself is typically zero-sized (no serialization needed).
///
/// # Example
///
/// ```rust,ignore
/// use murmer::prelude::*;
/// use serde::{Serialize, Deserialize};
///
/// #[derive(Debug)]
/// struct Worker;
///
/// #[derive(Serialize, Deserialize)]
/// struct WorkerState { queue: Vec<String> }
///
/// impl Actor for Worker {
///     type State = WorkerState;
/// }
///
/// // Opt-in: this actor's state can cross the wire
/// impl MigratableActor for Worker {}
/// ```
pub trait MigratableActor: Actor
where
    Self::State: Serialize + DeserializeOwned,
{
}

/// Declares that actor `A` can handle message `M`.
///
/// This is typically generated by the `#[handlers]` + `#[handler]` macro.
/// You can also implement it manually:
///
/// ```rust,ignore
/// impl Handler<Increment> for CounterActor {
///     fn handle(&mut self, ctx: &ActorContext<Self>, state: &mut CounterState, msg: Increment) -> i64 {
///         state.count += msg.amount;
///         state.count
///     }
/// }
/// ```
pub trait Handler<M: Message>: Actor + Sized {
    fn handle(
        &mut self,
        ctx: &ActorContext<Self>,
        state: &mut Self::State,
        message: M,
    ) -> M::Result;
}

/// Declares that actor `A` can handle message `M` asynchronously.
///
/// Like [`Handler`], but the handler returns a future, allowing `.await`
/// inside handler logic without blocking the tokio runtime.
///
/// ```rust,ignore
/// impl AsyncHandler<FetchData> for MyActor {
///     fn handle(&mut self, ctx: &ActorContext<Self>, state: &mut MyState, msg: FetchData)
///         -> impl Future<Output = String> + Send + '_ {
///         async move {
///             let data = some_async_call().await;
///             format!("got: {data}")
///         }
///     }
/// }
/// ```
pub trait AsyncHandler<M: Message>: Actor + Sized {
    fn handle<'a>(
        &'a mut self,
        ctx: &'a ActorContext<Self>,
        state: &'a mut Self::State,
        message: M,
    ) -> impl Future<Output = M::Result> + Send + 'a;
}

/// Generated by the `#[handlers]` proc macro: dispatches a remote message by type tag.
///
/// Each match arm in the generated implementation deserializes the payload to the
/// concrete message type, calls the corresponding [`Handler`] or [`AsyncHandler`],
/// and serializes the result.
///
/// You should not need to implement this manually — use `#[handlers]` instead.
///
/// # Examples
///
/// ```rust,ignore
/// // Generated by #[handlers] — you don't write this by hand:
/// #[handlers]
/// impl Counter {
///     #[handler]
///     fn handle_increment(&self, ctx: &ActorContext<Self>, state: &mut CounterState, msg: Increment) -> i64 {
///         state.count += msg.amount;
///         state.count
///     }
/// }
/// // The macro generates a RemoteDispatch impl that routes
/// // "counter::Increment" → handle_increment
/// ```
pub trait RemoteDispatch: Actor + Sized {
    fn dispatch_remote<'a>(
        &'a mut self,
        ctx: &'a ActorContext<Self>,
        state: &'a mut Self::State,
        message_type: &'a str,
        payload: &'a [u8],
    ) -> impl Future<Output = Result<Vec<u8>, DispatchError>> + Send + 'a;
}

/// Errors from remote message dispatch.
///
/// # Examples
///
/// ```rust,ignore
/// match actor.dispatch_remote(ctx, state, "unknown::Msg", &payload).await {
///     Ok(bytes) => { /* send response */ }
///     Err(DispatchError::UnknownMessageType(t)) => {
///         tracing::warn!("No handler for message type: {t}");
///     }
///     Err(e) => tracing::error!("Dispatch failed: {e}"),
/// }
/// ```
#[derive(Debug, thiserror::Error)]
pub enum DispatchError {
    #[error("unknown message type: {0}")]
    UnknownMessageType(String),
    #[error("deserialize failed: {0}")]
    DeserializeFailed(String),
    #[error("serialize failed: {0}")]
    SerializeFailed(String),
}