use core::mem::discriminant;
use parking_lot::Mutex;
use smol_str::SmolStr;
use std::{
    any::Any,
    fmt::{self, Debug, Display},
    future::{poll_fn, Future},
    hash::Hash,
    marker::PhantomData,
    pin::Pin,
    sync::{
        atomic::{AtomicBool, AtomicUsize, Ordering},
        Arc,
    },
    task::{Context, Poll, Waker},
};

/// Every actor has an ID that is generated by `acto`, independent of the [`ActoRuntime`] used.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ActoId(usize);

impl ActoId {
    fn next() -> Self {
        static COUNTER: AtomicUsize = AtomicUsize::new(0);
        let id = COUNTER.fetch_add(1, Ordering::Relaxed);
        if id == usize::MAX {
            panic!("ActoId wrap-around! Cannot create more than {} actors", id)
        }
        Self(id)
    }
}

/// A handle for sending messages to an actor.
///
/// You may freely clone or share this handle and store it in collections.
pub struct ActoRef<M>(Arc<ActoRefInner<dyn Sender<M>>>);

struct ActoRefInner<S: ?Sized> {
    fields: Inner,
    id: ActoId,
    sender: S,
}

enum Inner {
    Straight(Straight),
    Mapped(Mapped),
    Blackhole,
}
unsafe impl Send for Inner {}
unsafe impl Sync for Inner {}

struct Straight {
    name: SmolStr,
    count: AtomicUsize,
    dead: AtomicBool,
    waker: Mutex<Option<Waker>>,
}

impl From<&Straight> for Mapped {
    fn from(s: &Straight) -> Self {
        Mapped {
            name: s.name.as_str(),
            count: &s.count,
            dead: &s.dead,
            waker: &s.waker,
        }
    }
}

/// safety: the pointers are guaranteed to not be dangling because the
/// associated `sender` holds a strong reference to the Arc providing them.
#[derive(Clone, Copy)]
struct Mapped {
    name: *const str,
    count: *const AtomicUsize,
    dead: *const AtomicBool,
    waker: *const Mutex<Option<Waker>>,
}

impl<T, U> PartialEq<ActoRef<U>> for ActoRef<T> {
    fn eq(&self, other: &ActoRef<U>) -> bool {
        self.0.id == other.0.id
    }
}

impl<T> Eq for ActoRef<T> {}

impl<T, U> PartialOrd<ActoRef<U>> for ActoRef<T> {
    fn partial_cmp(&self, other: &ActoRef<U>) -> Option<std::cmp::Ordering> {
        self.0.id.partial_cmp(&other.0.id)
    }
}

impl<T> Ord for ActoRef<T> {
    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
        self.0.id.cmp(&other.0.id)
    }
}

impl<T> Hash for ActoRef<T> {
    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
        self.0.id.hash(state);
    }
}

impl<M> ActoRef<M> {
    /// Create a dummy reference that drops all messages
    pub fn blackhole() -> Self {
        Self(Arc::new(ActoRefInner {
            fields: Inner::Blackhole,
            id: ActoId::next(),
            sender: |_| true,
        }))
    }

    /// The [`ActoId`] of the referenced actor.
    pub fn id(&self) -> ActoId {
        self.0.id
    }

    /// The actor’s given name plus ActoRuntime name and ActoId.
    pub fn name(&self) -> &str {
        match &self.0.fields {
            Inner::Straight(s) => s.name.as_str(),
            Inner::Mapped(m) => unsafe { &*m.name },
            Inner::Blackhole => "blackhole(acto/0)",
        }
    }

    #[inline(always)]
    fn with_count(&self, f: impl FnOnce(&AtomicUsize) -> usize) -> usize {
        match &self.0.fields {
            Inner::Straight(s) => f(&s.count),
            Inner::Mapped(m) => f(unsafe { &*m.count }),
            Inner::Blackhole => 0,
        }
    }

    fn dead(&self) {
        match &self.0.fields {
            Inner::Straight(s) => s.dead.store(true, Ordering::Release),
            Inner::Mapped(m) => unsafe { &*m.dead }.store(true, Ordering::Release),
            Inner::Blackhole => {}
        }
    }

    fn waker(&self, waker: Waker) {
        match &self.0.fields {
            Inner::Straight(s) => *s.waker.lock() = Some(waker),
            Inner::Mapped(m) => *unsafe { &*m.waker }.lock() = Some(waker),
            Inner::Blackhole => {}
        }
    }

    fn wake(&self) {
        let waker = match &self.0.fields {
            Inner::Straight(s) => s.waker.lock().take(),
            Inner::Mapped(m) => unsafe { &*m.waker }.lock().take(),
            Inner::Blackhole => None,
        };
        if let Some(waker) = waker {
            waker.wake();
        }
    }

    /// Check whether the referenced actor is in principle still ready to receive messages.
    ///
    /// Note that this is not the same as [`ActoHandle::is_finished`], which checks whether
    /// the actor’s task is done. An actor could drop its [`ActoCell`] (yielding `true` here)
    /// or it could move it to another async task (yielding `true` from `ActoHandle`).
    ///
    /// Note that a “blackhole” reference is immortal.
    pub fn is_gone(&self) -> bool {
        match &self.0.fields {
            Inner::Straight(s) => s.dead.load(Ordering::Acquire),
            Inner::Mapped(m) => unsafe { &*m.dead }.load(Ordering::Acquire),
            Inner::Blackhole => false,
        }
    }
}

impl<M: Send + 'static> ActoRef<M> {
    /// Send a message to the referenced actor.
    ///
    /// The employed channel may be at its capacity bound and the target actor may
    /// already be terminated, in which cases the message is dropped and `false` is
    /// returned.
    ///
    /// This method does not return the unsent message in the above cases because the
    /// message may not be delivered even though `true` is returned. In other words,
    /// you cannot rely on the return value to conclude that the message was delivered.
    /// Instead, have the target actor send a confirmation message back to you.
    pub fn send(&self, msg: M) -> bool {
        tracing::trace!(target = ?self, "send");
        self.0.sender.send(msg)
    }

    /// Derive an ActoRef accepting a different type of message, typically embedded in an `enum`.
    ///
    /// ```rust
    /// # use acto::{AcTokio, ActoRuntime, ActoCell, ActoInput};
    /// use tokio::sync::oneshot;
    /// async fn actor(mut cell: ActoCell<Option<oneshot::Sender<i32>>, impl ActoRuntime>) {
    ///     while let ActoInput::Message(Some(channel)) = cell.recv().await {
    ///         channel.send(42).ok();
    ///     }
    /// }
    /// let rt = AcTokio::new("test", 1).unwrap();
    /// let ar = rt.spawn_actor("a", actor).me.contramap(|msg| Some(msg));
    /// let (tx, rx) = oneshot::channel();
    /// ar.send(tx);
    /// # let response = rt.with_rt(|rt| rt.block_on(rx)).unwrap().unwrap();
    /// # assert_eq!(response, 42);
    /// ```
    pub fn contramap<M2>(&self, f: impl Fn(M2) -> M + Send + Sync + 'static) -> ActoRef<M2> {
        let fields = match &self.0.fields {
            Inner::Straight(s) => Inner::Mapped(s.into()),
            Inner::Mapped(x) => Inner::Mapped(*x),
            Inner::Blackhole => Inner::Blackhole,
        };
        let count = self.with_count(|c| c.fetch_add(1, Ordering::SeqCst));
        tracing::trace!(count, ?self, "contramap");
        let orig = self.0.clone();
        let inner = ActoRefInner {
            fields,
            id: orig.id,
            sender: move |msg| orig.sender.send(f(msg)),
        };
        ActoRef(Arc::new(inner))
    }

    pub fn is_blackhole(&self) -> bool {
        matches!(self.0.fields, Inner::Blackhole)
    }
}

impl<M> Clone for ActoRef<M> {
    fn clone(&self) -> Self {
        let count = self.with_count(|c| c.fetch_add(1, Ordering::SeqCst));
        tracing::trace!(count, ?self, "clone");
        Self(self.0.clone())
    }
}

impl<M> Drop for ActoRef<M> {
    fn drop(&mut self) {
        let count = self.with_count(|c| c.fetch_sub(1, Ordering::SeqCst));
        tracing::trace!(count, ?self, "drop");
        if count == 1 {
            self.wake()
        }
    }
}

impl<M> Debug for ActoRef<M> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "ActoRef({})", self.name())
    }
}
pub enum PanicOrAbort {
    Panic(Box<dyn PanicInfo>),
    Abort(ActoAborted),
}

impl Display for PanicOrAbort {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            PanicOrAbort::Panic(p) => write!(f, "Actor panicked: {}", p.cause()),
            PanicOrAbort::Abort(_) => write!(f, "Actor aborted via Acto"),
        }
    }
}

impl Debug for PanicOrAbort {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Panic(arg0) => f.debug_tuple("Panic").field(arg0).finish(),
            Self::Abort(arg0) => Debug::fmt(arg0, f),
        }
    }
}

pub trait PanicInfo: Debug + Send + 'static {
    fn payload(&self) -> Option<&(dyn Any + Send + 'static)>;
    fn is_cancelled(&self) -> bool;
    fn cause(&self) -> String;
}

/// This error is returned when an actor has been aborted.
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct ActoAborted(SmolStr);

impl ActoAborted {
    pub fn new(name: impl AsRef<str>) -> Self {
        Self(name.into())
    }
}

impl std::error::Error for ActoAborted {}

impl fmt::Display for ActoAborted {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        write!(f, "Actor aborted: {}", self.0)
    }
}

/// The confines of an actor, and the engine that makes it work.
///
/// Every actor is provided with an `ActoCell` when it is started, which is its
/// means of interacting with other actors.
///
/// The type parameter `R` is present so that the Actor can formulate further
/// requirements in its type signature (e.g. `R: MailboxSize`).
pub struct ActoCell<M: Send + 'static, R: ActoRuntime, S: 'static = ()> {
    me: ActoRef<M>,
    runtime: R,
    recv: R::Receiver<M>,
    supervised: Vec<Pin<Box<dyn ActoHandle<Output = S>>>>,
    no_senders_signaled: bool,
}

impl<M: Send + 'static, R: ActoRuntime, S: 'static> Drop for ActoCell<M, R, S> {
    fn drop(&mut self) {
        for mut h in self.supervised.drain(..) {
            h.as_mut().abort_pinned();
        }
        self.me.dead();
    }
}

impl<M: Send + 'static, R: ActoRuntime, S: Send + 'static> ActoCell<M, R, S> {
    /// Get access to the [`ActoRuntime`] driving this actor, e.g. to customize mailbox size for spawned actors.
    ///
    /// See [`MailboxSize`].
    pub fn rt(&self) -> &R {
        &self.runtime
    }

    /// The actor’s own [`ActoRef`] handle, which it may send elsewhere to receive messages.
    pub fn me(&mut self) -> ActoRef<M> {
        self.no_senders_signaled = false;
        self.me.clone()
    }

    /// Asynchronously `.await` the reception of inputs.
    ///
    /// These may either be a message (sent via an [`ActoRef`]), the notification that all
    /// external `ActoRef`s have been dropped, or the termination notice of a supervised
    /// actor.
    pub fn recv(&mut self) -> impl Future<Output = ActoInput<M, S>> + '_ {
        poll_fn(|cx| {
            for idx in 0..self.supervised.len() {
                let p = self.supervised[idx].as_mut().poll(cx);
                if let Poll::Ready(result) = p {
                    let handle = self.supervised.remove(idx);
                    tracing::trace!(src = ?handle.name(), "supervision");
                    return Poll::Ready(ActoInput::Supervision {
                        id: handle.id(),
                        name: handle.name().to_owned(),
                        result,
                    });
                }
            }
            if let Poll::Ready(msg) = self.recv.poll(cx) {
                tracing::trace!("got message");
                return Poll::Ready(ActoInput::Message(msg));
            }
            if self.me.with_count(|c| c.load(Ordering::SeqCst)) == 0 {
                tracing::trace!("no more senders");
                if !self.no_senders_signaled {
                    self.no_senders_signaled = true;
                    return Poll::Ready(ActoInput::NoMoreSenders);
                }
            } else if !self.no_senders_signaled {
                // only install waker if we’re interested in emitting NoMoreSenders
                self.me.waker(cx.waker().clone());
                // re-check in case last ref was dropped between check and lock
                if self.me.with_count(|c| c.load(Ordering::SeqCst)) == 0 {
                    tracing::trace!(me = ?self.me.name(), "no sender");
                    self.no_senders_signaled = true;
                    return Poll::Ready(ActoInput::NoMoreSenders);
                }
            }
            tracing::trace!("Poll::Pending");
            Poll::Pending
        })
    }

    /// Create a new actor on the same [`ActoRuntime`] as the current one.
    ///
    /// ```rust
    /// use acto::{ActoCell, ActoInput, ActoRuntime};
    ///
    /// async fn actor<M: Send + 'static, R: ActoRuntime>(cell: ActoCell<M, R>) {
    ///     // spawn and forget
    ///     cell.spawn("name", |cell: ActoCell<i32, _>| async move { todo!() });
    ///     // spawn, retrieve handle, do not supervise
    ///     let a_ref = cell.spawn("super", |mut cell: ActoCell<_, _, ()>| async move {
    ///         if let ActoInput::Message(msg) = cell.recv().await {
    ///             cell.supervise(msg);
    ///         }
    ///     }).me;
    ///     // spawn and let some other actor supervise
    ///     let s_ref = cell.spawn("other", |cell: ActoCell<i32, _>| async move { todo!() });
    ///     a_ref.send(s_ref);
    /// }
    /// ```
    pub fn spawn<M2, F, Fut, S2>(
        &self,
        name: &str,
        actor: F,
    ) -> SupervisionRef<M2, R::ActoHandle<Fut::Output>>
    where
        F: FnOnce(ActoCell<M2, R, S2>) -> Fut,
        Fut: Future + Send + 'static,
        Fut::Output: Send + 'static,
        S2: Send + 'static,
        M2: Send + 'static,
    {
        self.runtime.spawn_actor(name, actor)
    }

    /// Create a new actor on the same [`ActoRuntime`] as the current one and [`ActoCell::supervise`] it.
    pub fn spawn_supervised<M2, F, Fut, S2, O>(&mut self, name: &str, actor: F) -> ActoRef<M2>
    where
        F: FnOnce(ActoCell<M2, R, S2>) -> Fut,
        Fut: Future<Output = O> + Send + 'static,
        O: Into<S> + Send + 'static,
        S2: Send + 'static,
        M2: Send + 'static,
    {
        self.supervise(self.spawn(name, actor))
    }

    /// Supervise another actor.
    ///
    /// When that actor terminates, this actor will receive [`ActoInput::Supervision`] for it.
    /// When this actor terminates, all supervised actors will be aborted.
    pub fn supervise<T, H, O>(&mut self, actor: SupervisionRef<T, H>) -> ActoRef<T>
    where
        T: Send + 'static,
        H: ActoHandle<Output = O> + Send + 'static,
        O: Into<S> + Send + 'static,
    {
        tracing::trace!(target = ?actor.me.name(), "supervise");
        self.supervised
            .push(Box::pin(actor.handle.map(|x: O| x.into())));
        actor.me
    }
}

/// A package of an actor’s [`ActoRef`] and [`ActoHandle`].
///
/// This is the result of [`ActoCell::spawn`] and can be passed to [`ActoCell::supervise`].
pub struct SupervisionRef<M, H> {
    pub me: ActoRef<M>,
    pub handle: H,
}

impl<M: Send + 'static, H: ActoHandle> SupervisionRef<M, H> {
    /// Derive a new reference by embedding the supervisor-required type `M2` into the message schema.
    ///
    /// ```rust
    /// # use acto::{ActoCell, ActoInput, ActoRef, ActoRuntime, SupervisionRef};
    /// struct Shutdown;
    ///
    /// enum ActorCommand {
    ///     Shutdown(Shutdown),
    ///     DoStuff(String),
    /// }
    ///
    /// async fn top_level(mut cell: ActoCell<(), impl ActoRuntime>) {
    ///     let supervisor = cell.spawn_supervised("super",
    ///         |mut cell: ActoCell<SupervisionRef<Shutdown, _>, _>| async move {
    ///             while let ActoInput::Message(actor) = cell.recv().await {
    ///                 let actor_ref = cell.supervise(actor);
    ///                 // use reference to shut it down at a later time
    ///             }
    ///             // if any of them fail, shut all of them down
    ///         }
    ///     );
    ///     let actor = cell.spawn("actor", |mut cell: ActoCell<_, _>| async move {
    ///         while let ActoInput::Message(msg) = cell.recv().await {
    ///             if let ActorCommand::Shutdown(_) = msg {
    ///                 break;
    ///             }
    ///         }
    ///     });
    ///     let actor_ref = actor.me.clone();
    ///     supervisor.send(actor.contramap(ActorCommand::Shutdown));
    ///     // do stuff with actor_ref
    /// }
    /// ```
    pub fn contramap<M2>(
        self,
        f: impl Fn(M2) -> M + Send + Sync + 'static,
    ) -> SupervisionRef<M2, H> {
        let Self { me, handle } = self;
        let fields = match &me.0.fields {
            Inner::Straight(s) => Inner::Mapped(s.into()),
            Inner::Mapped(x) => Inner::Mapped(*x),
            Inner::Blackhole => Inner::Blackhole,
        };
        let id = me.id();
        let inner = ActoRefInner {
            fields,
            id,
            sender: move |msg| me.0.sender.send(f(msg)),
        };
        let me = ActoRef(Arc::new(inner));
        SupervisionRef { me, handle }
    }

    /// Map the return type of the contained [`ActoHandle`] to match the intended supervisor.
    ///
    /// ```rust
    /// # use acto::{ActoCell, ActoInput, ActoRef, ActoRuntime, SupervisionRef, AcTokio, ActoHandle};
    /// async fn top_level(mut cell: ActoCell<(), impl ActoRuntime, String>) -> ActoInput<(), String> {
    ///     let actor = cell.spawn("actor", |mut cell: ActoCell<(), _>| async move {
    ///         // some async computation that leads to the result
    ///         42
    ///     });
    ///     // cannot supervise without transforming result to a String
    ///     let ar = cell.supervise(actor.map_handle(|number| number.to_string()));
    ///     // now do something with the actor reference
    /// #   cell.recv().await
    /// }
    /// # let sys = AcTokio::new("doc", 1).unwrap();
    /// # let ah = sys.spawn_actor("top", top_level);
    /// # let ah = ah.handle;
    /// # let ActoInput::Supervision { name, result, ..} = sys.with_rt(|rt| rt.block_on(ah.join())).unwrap().unwrap() else { panic!("wat") };
    /// # assert!(name.starts_with("actor(doc/"));
    /// # assert_eq!(result.unwrap(), "42");
    /// ```
    pub fn map_handle<S, F>(self, f: F) -> SupervisionRef<M, MappedActoHandle<H, F, S>>
    where
        F: FnOnce(H::Output) -> S + Send + Sync + Unpin + 'static,
        S: Send + 'static,
    {
        SupervisionRef {
            me: self.me,
            handle: MappedActoHandle::new(self.handle, f),
        }
    }
}

/// Actor input as received with [`ActoCell::recv`].
#[derive(Debug)]
pub enum ActoInput<M, S> {
    /// All previously generated [`ActoRef`] handles were dropped, leaving only
    /// the one within [`ActoCell`]; the actor may wish to terminate unless it has
    /// other sources of input.
    ///
    /// Obtaining a new handle with [`ActoCell::me`] and dropping it will again
    /// generate this input.
    NoMoreSenders,
    /// A supervised actor with the given [`ActoId`] has terminated.
    ///
    /// The result contains either the value returned by the actor or the value
    /// with which the actor’s task panicked (if the underlying runtime handles this).
    ///
    /// ## Important notice
    ///
    /// Supervision notifications are delivered as soon as possible after the
    /// supervised actor’s task has finished. Messages sent by that actor — possibly
    /// to the supervisor — may still be in flight at this point.
    Supervision {
        id: ActoId,
        name: String,
        result: Result<S, PanicOrAbort>,
    },
    /// A message has been received via our [`ActoRef`] handle.
    Message(M),
}

impl<M, S> ActoInput<M, S> {
    pub fn is_sender_gone(&self) -> bool {
        matches!(self, ActoInput::NoMoreSenders)
    }

    pub fn is_supervision(&self) -> bool {
        matches!(self, ActoInput::Supervision { .. })
    }

    pub fn is_message(&self) -> bool {
        matches!(self, ActoInput::Message(_))
    }

    /// Obtain input message or supervision unless this is [`ActoInput::NoMoreSenders`].
    ///
    /// ```rust
    /// use acto::{ActoCell, ActoRuntime};
    ///
    /// async fn actor(mut cell: ActoCell<String, impl ActoRuntime>) {
    ///     while let Some(input) = cell.recv().await.has_senders() {
    ///         // do something with it
    ///     }
    ///     // actor automatically stops when all senders are gone
    /// }
    /// ```
    pub fn has_senders(self) -> Option<ActoMsgSuper<M, S>> {
        match self {
            ActoInput::NoMoreSenders => None,
            ActoInput::Supervision { id, name, result } => {
                Some(ActoMsgSuper::Supervision { id, name, result })
            }
            ActoInput::Message(msg) => Some(ActoMsgSuper::Message(msg)),
        }
    }
}

impl<M: PartialEq, S> PartialEq for ActoInput<M, S> {
    fn eq(&self, other: &Self) -> bool {
        match (self, other) {
            (Self::Supervision { id: left, .. }, Self::Supervision { id: right, .. }) => {
                left == right
            }
            (Self::Message(l0), Self::Message(r0)) => l0 == r0,
            _ => discriminant(self) == discriminant(other),
        }
    }
}

/// A filtered variant of [`ActoInput`] that omits `NoMoreSenders`.
///
/// see [`ActoInput::has_senders`]
pub enum ActoMsgSuper<M, S> {
    Supervision {
        id: ActoId,
        name: String,
        result: Result<S, PanicOrAbort>,
    },
    /// A message has been received via our [`ActoRef`] handle.
    Message(M),
}

/// For implementors: the interface of a runtime for operating actors.
///
/// Cloning a runtime should be cheap, it SHOULD be using the `Arc<Inner>` pattern.
pub trait ActoRuntime: Clone + Send + Sync + 'static {
    /// The type of handle used for joining the actor’s task.
    type ActoHandle<O: Send + 'static>: ActoHandle<Output = O>;
    /// The type of sender for emitting messages towards the actor.
    type Sender<M: Send + 'static>: Sender<M>;
    /// The type of receiver for obtaining messages sent towards the actor.
    type Receiver<M: Send + 'static>: Receiver<M>;

    /// A name for this runtime, used mainly in logging.
    fn name(&self) -> &str;

    /// Create a new pair of sender and receiver for a fresh actor.
    fn mailbox<M: Send + 'static>(&self) -> (Self::Sender<M>, Self::Receiver<M>);

    /// Spawn an actor’s task to be driven independently and return an [`ActoHandle`]
    /// to abort or join it.
    fn spawn_task<T>(&self, id: ActoId, name: SmolStr, task: T) -> Self::ActoHandle<T::Output>
    where
        T: Future + Send + 'static,
        T::Output: Send + 'static;

    /// Provided function for spawning actors.
    ///
    /// Uses the above utilities and cannot be implemented by downstream crates.
    fn spawn_actor<M, F, Fut, S>(
        &self,
        name: &str,
        actor: F,
    ) -> SupervisionRef<M, Self::ActoHandle<Fut::Output>>
    where
        M: Send + 'static,
        F: FnOnce(ActoCell<M, Self, S>) -> Fut,
        Fut: Future + Send + 'static,
        Fut::Output: Send + 'static,
        S: Send + 'static,
    {
        let (sender, recv) = self.mailbox();
        let id = ActoId::next();
        let mut id_str = [0u8; 16];
        let id_str = write_id(&mut id_str, id);
        let name = [name, "(", self.name(), "/", id_str, ")"]
            .into_iter()
            .collect::<SmolStr>();
        let name2 = name.clone();
        let inner = ActoRefInner {
            id,
            fields: Inner::Straight(Straight {
                name,
                count: AtomicUsize::new(0),
                dead: AtomicBool::new(false),
                waker: Mutex::new(None),
            }),
            sender,
        };
        let me = ActoRef(Arc::new(inner));
        let ctx = ActoCell {
            me: me.clone(),
            runtime: self.clone(),
            recv,
            supervised: vec![],
            no_senders_signaled: false,
        };
        let _span = tracing::debug_span!("creating", actor = %name2).entered();
        tracing::trace!("create");
        let task = LoggingTask::new(name2.clone(), (actor)(ctx));
        let join = self.spawn_task(id, name2, task);
        SupervisionRef { me, handle: join }
    }
}

pin_project_lite::pin_project! {
    struct LoggingTask<F> {
        name: SmolStr,
        #[pin]
        future: F,
    }
}

impl<F> LoggingTask<F> {
    pub fn new(name: SmolStr, future: F) -> Self {
        Self { name, future }
    }
}

impl<F> Future for LoggingTask<F>
where
    F: Future + Send + 'static,
    F::Output: Send + 'static,
{
    type Output = F::Output;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let this = self.project();
        let _span = tracing::debug_span!("poll", actor = %this.name).entered();
        this.future.poll(cx)
    }
}

fn write_id(buf: &mut [u8; 16], id: ActoId) -> &str {
    let id = id.0;
    if id == 0 {
        return "0";
    }
    let mut written = 0;
    let mut shift = (16 - (id.leading_zeros()) / 4) * 4;
    while shift != 0 {
        shift -= 4;
        const HEX: [u8; 16] = *b"0123456789abcdef";
        buf[written] = HEX[(id >> shift) & 15];
        written += 1;
    }
    unsafe { std::str::from_utf8_unchecked(&buf[..written]) }
}

#[test]
fn test_write_id() {
    let mut buf = [0u8; 16];
    assert_eq!(write_id(&mut buf, ActoId(0)), "0");
    assert_eq!(write_id(&mut buf, ActoId(1)), "1");
    assert_eq!(write_id(&mut buf, ActoId(10)), "a");
    assert_eq!(write_id(&mut buf, ActoId(100)), "64");
}

/// This trait is implemented by [`ActoRuntime`]s that allow customization of the mailbox size.
///
/// ```rust
/// # use acto::{ActoCell, ActoRef, ActoRuntime, MailboxSize};
/// async fn actor(cell: ActoCell<String, impl MailboxSize>) {
///     let child: ActoRef<u8> = cell.rt()
///         .with_mailbox_size(10)
///         .spawn_actor("child", |_: ActoCell<_, _>| async {})
///         .me;
/// }
/// ```
pub trait MailboxSize: ActoRuntime {
    type Output: ActoRuntime;

    fn with_mailbox_size(&self, mailbox_size: usize) -> Self::Output;
}

/// A named closure for sending messages to a given actor.
///
/// This type is used between a runtime implementation and `acto`.
pub trait Sender<M>: Send + Sync + 'static {
    fn send(&self, msg: M) -> bool;
}

impl<M, F: Fn(M) -> bool + Send + Sync + 'static> Sender<M> for F {
    fn send(&self, msg: M) -> bool {
        (self)(msg)
    }
}

/// A named closure for receiving messages at a given actor.
///
/// This type is used between a runtime implementation and `acto`.
pub trait Receiver<M>: Send + 'static {
    fn poll(&mut self, cx: &mut Context<'_>) -> Poll<M>;
}

/// A handle for aborting or joining a running actor.
pub trait ActoHandle: Send + 'static {
    type Output;

    /// The ID of the underlying actor.
    fn id(&self) -> ActoId;

    /// The name of the underlying actor.
    fn name(&self) -> &str;

    /// Abort the actor’s task.
    ///
    /// Use this method if you don’t need the handle afterwards; otherwise use [`abort_pinned`].
    /// Behavior is undefined if the actor is not [cancellation safe].
    ///
    /// [cancellation safe]: https://docs.rs/tokio/latest/tokio/macro.select.html#cancellation-safety
    fn abort(mut self)
    where
        Self: Sized,
    {
        // safety:
        // - we drop `self` at the end of the scope without moving it
        // - we don’t use Deref or DerefMut; if the implementor does, it’s their responsibility
        let this = unsafe { Pin::new_unchecked(&mut self) };
        this.abort_pinned();
    }

    /// Abort the actor’s task.
    ///
    /// Use this method if you want to [`join`] the actor’s task later, otherwise
    /// prefer the [`abort`] method that can be called without pinning first.
    /// Behavior is undefined if the actor is not [cancellation safe].
    ///
    /// [cancellation safe]: https://docs.rs/tokio/latest/tokio/macro.select.html#cancellation-safety
    fn abort_pinned(self: Pin<&mut Self>);

    /// Check whether the actor’s task is no longer running
    ///
    /// This may be the case even after [`ActoHandle::abort`] has returned, since task
    /// termination may be asynchronous.
    ///
    /// Note that this is not the same as [`ActoRef::is_gone`], which checks whether
    /// the actor is still capable of receiving messages. An actor could drop its
    /// [`ActoCell`] (yielding `true` there) or it could move it to another async task
    /// (yielding `true` here).
    fn is_finished(&self) -> bool;

    /// Poll this handle for whether the actor is now terminated.
    ///
    /// This method has [`Future`] semantics.
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>)
        -> Poll<Result<Self::Output, PanicOrAbort>>;

    /// Transform the output value of this handle, e.g. before calling [`ActoCell::supervise`].
    fn map<O, F>(self, transform: F) -> MappedActoHandle<Self, F, O>
    where
        Self: Sized,
    {
        MappedActoHandle::new(self, transform)
    }

    /// A future for awaiting the termination of the actor underlying this handle.
    fn join(self) -> ActoHandleFuture<Self>
    where
        Self: Sized,
    {
        ActoHandleFuture { handle: self }
    }
}

pin_project_lite::pin_project! {
    /// The future returned by [`ActoHandle::join`].
    pub struct ActoHandleFuture<J> {
        #[pin] handle: J
    }
}

impl<J: ActoHandle> Future for ActoHandleFuture<J> {
    type Output = Result<J::Output, PanicOrAbort>;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let _span = tracing::debug_span!("poll", join = ?self.as_ref().handle.name());
        self.project().handle.poll(cx)
    }
}

pin_project_lite::pin_project! {
    /// An [`ActoHandle`] that results from [`ActoHandle::map`].
    pub struct MappedActoHandle<H, F, O> {
        #[pin] inner: H,
        transform: Option<F>,
        _ph: PhantomData<O>,
    }
}

impl<H, F, O> MappedActoHandle<H, F, O> {
    fn new(inner: H, transform: F) -> Self {
        Self {
            inner,
            transform: Some(transform),
            _ph: PhantomData,
        }
    }
}

impl<H, F, O> ActoHandle for MappedActoHandle<H, F, O>
where
    H: ActoHandle,
    F: FnOnce(H::Output) -> O + Send + Sync + Unpin + 'static,
    O: Send + 'static,
{
    type Output = O;

    fn id(&self) -> ActoId {
        self.inner.id()
    }

    fn name(&self) -> &str {
        self.inner.name()
    }

    fn abort_pinned(self: Pin<&mut Self>) {
        self.project().inner.abort_pinned()
    }

    fn is_finished(&self) -> bool {
        self.inner.is_finished()
    }

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<O, PanicOrAbort>> {
        let this = self.project();
        this.inner.poll(cx).map(|r| {
            let transform = this.transform.take().expect("polled after finish");
            r.map(transform)
        })
    }
}