datum-core 0.10.1

use std::{
    collections::VecDeque,
    fmt,
    sync::{
        Arc, Condvar, Mutex, MutexGuard, Weak,
        atomic::{AtomicBool, Ordering},
    },
    time::{Duration, Instant},
};

use futures::channel::oneshot;

use crate::stream::{
    BoxStream, Cancellable, NotUsed, Sink, Source, StreamCancellation, StreamCompletion,
    StreamError, StreamResult,
};

use super::{Actor, ActorProcessingErr, ActorRef, ActorResult, Message, block_on_ractor_runtime};

const DEFAULT_STREAM_REF_BUFFER_CAPACITY: usize = 32;
const DEFAULT_STREAM_REF_SUBSCRIPTION_TIMEOUT: Duration = Duration::from_secs(30);
const DEFAULT_STREAM_REF_DEMAND_REDELIVERY: Duration = Duration::from_secs(1);
const STREAM_REF_WAIT_POLL: Duration = Duration::from_millis(1);

mod stream_ref_actor_bound {
    pub trait Bound {}

    impl<T> Bound for T {}
}

/// Settings used by [`StreamRefs`] endpoints.
///
/// The defaults mirror Akka StreamRefs: a 32 element receive buffer, a 30 second
/// subscription timeout, and 1 second demand redelivery. Demand is cumulative,
/// so redelivery is idempotent.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct StreamRefSettings {
    buffer_capacity: usize,
    subscription_timeout: Duration,
    demand_redelivery_interval: Duration,
}

impl Default for StreamRefSettings {
    fn default() -> Self {
        Self {
            buffer_capacity: DEFAULT_STREAM_REF_BUFFER_CAPACITY,
            subscription_timeout: DEFAULT_STREAM_REF_SUBSCRIPTION_TIMEOUT,
            demand_redelivery_interval: DEFAULT_STREAM_REF_DEMAND_REDELIVERY,
        }
    }
}

impl StreamRefSettings {
    #[must_use]
    pub fn buffer_capacity(&self) -> usize {
        self.buffer_capacity
    }

    #[must_use]
    pub fn subscription_timeout(&self) -> Duration {
        self.subscription_timeout
    }

    #[must_use]
    pub fn demand_redelivery_interval(&self) -> Duration {
        self.demand_redelivery_interval
    }

    #[must_use]
    pub fn with_buffer_capacity(mut self, capacity: usize) -> Self {
        assert!(
            capacity > 0,
            "StreamRef buffer capacity must be greater than zero"
        );
        self.buffer_capacity = capacity;
        self
    }

    #[must_use]
    pub fn with_subscription_timeout(mut self, timeout: Duration) -> Self {
        self.subscription_timeout = timeout;
        self
    }

    #[must_use]
    pub fn with_demand_redelivery_interval(mut self, interval: Duration) -> Self {
        self.demand_redelivery_interval = interval;
        self
    }
}

/// Factories for Akka-style stream references.
///
/// `source_ref` is a local sink that materializes a [`SourceRef`]. The returned
/// handle can be used once to create a remote source. `sink_ref` is the dual: a
/// local source that materializes a [`SinkRef`], which can be used once as a
/// remote sink.
pub struct StreamRefs;

impl StreamRefs {
    #[must_use]
    pub fn source_ref<T>() -> Sink<T, SourceRef<T>>
    where
        T: Send + 'static,
    {
        Self::source_ref_with_settings(StreamRefSettings::default())
    }

    #[must_use]
    pub fn source_ref_with_settings<T>(settings: StreamRefSettings) -> Sink<T, SourceRef<T>>
    where
        T: Send + 'static,
    {
        stream_ref_source_sink(settings)
    }

    #[must_use]
    pub fn sink_ref<T>() -> Source<T, SinkRef<T>>
    where
        T: Send + 'static,
    {
        Self::sink_ref_with_settings(StreamRefSettings::default())
    }

    #[must_use]
    pub fn sink_ref_with_settings<T>(settings: StreamRefSettings) -> Source<T, SinkRef<T>>
    where
        T: Send + 'static,
    {
        stream_ref_sink_source(settings)
    }
}

/// A handle that exposes a remote stream as a local [`Source`].
///
/// Stream refs are one-shot pairings. Calling [`SourceRef::source`] more than
/// once returns a source that fails when materialized.
pub struct SourceRef<T> {
    inner: Arc<SourceRefInner<T>>,
}

struct SourceRefInner<T> {
    #[allow(dead_code)]
    producer: LazyProducerStatus<T>,
    direct: Arc<SourceRefDirectState<T>>,
    settings: StreamRefSettings,
    subscribed: AtomicBool,
    timeout: Option<Cancellable>,
}

impl<T> Clone for SourceRef<T> {
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
        }
    }
}

impl<T> fmt::Debug for SourceRef<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SourceRef").finish_non_exhaustive()
    }
}

impl<T> SourceRef<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    #[must_use]
    pub fn source(&self) -> Source<T, NotUsed> {
        let inner = Arc::clone(&self.inner);
        Source::from_materialized_factory(move |_materializer| {
            if inner.subscribed.swap(true, Ordering::SeqCst) {
                return Ok((
                    failed_once("source ref has already been materialized"),
                    NotUsed,
                ));
            }

            if let Some(timeout) = &inner.timeout {
                timeout.cancel();
            }
            match inner.direct.claim_input() {
                Ok(input) => Ok((
                    Box::new(SourceRefDirectStream::new(
                        input,
                        inner.settings,
                        Arc::clone(&inner),
                    )) as BoxStream<T>,
                    NotUsed,
                )),
                Err(error) => Ok((failed_stream(error), NotUsed)),
            }
        })
    }
}

pub(super) fn proto_source<T>(source_ref: &SourceRef<T>) -> Source<T, NotUsed>
where
    T: Send + 'static,
{
    let inner = Arc::clone(&source_ref.inner);
    Source::from_materialized_factory(move |_materializer| {
        if inner.subscribed.swap(true, Ordering::SeqCst) {
            return Ok((
                failed_stream(StreamError::Failed(
                    "source ref has already been materialized".to_owned(),
                )),
                NotUsed,
            ));
        }

        if let Some(timeout) = &inner.timeout {
            timeout.cancel();
        }
        match inner.direct.claim_input() {
            Ok(input) => Ok((
                Box::new(SourceRefDirectStream::new(
                    input,
                    inner.settings,
                    Arc::clone(&inner),
                )) as BoxStream<T>,
                NotUsed,
            )),
            Err(error) => Ok((failed_stream(error), NotUsed)),
        }
    })
}

/// A handle that exposes a remote stream as a local [`Sink`].
///
/// Stream refs are one-shot pairings. Calling [`SinkRef::sink`] more than once
/// returns a sink whose materialized completion fails.
pub struct SinkRef<T> {
    inner: Arc<SinkRefInner<T>>,
}

struct SinkRefInner<T> {
    direct: Arc<SinkRefDirectState<T>>,
    subscribed: AtomicBool,
}

impl<T> Clone for SinkRef<T> {
    fn clone(&self) -> Self {
        Self {
            inner: Arc::clone(&self.inner),
        }
    }
}

impl<T> fmt::Debug for SinkRef<T> {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("SinkRef").finish_non_exhaustive()
    }
}

impl<T> SinkRef<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    #[must_use]
    pub fn sink(&self) -> Sink<T, StreamCompletion<NotUsed>> {
        let inner = Arc::clone(&self.inner);
        Sink::from_runner(move |input, _materializer| {
            if inner.subscribed.swap(true, Ordering::SeqCst) {
                return Ok(StreamCompletion::ready(Err(StreamError::Failed(
                    "sink ref has already been materialized".to_owned(),
                ))));
            }

            Ok(inner.direct.attach_input(input))
        })
    }
}

#[allow(dead_code)]
enum ProducerCommand<T> {
    Subscribe {
        consumer: ActorRef<ConsumerCommand<T>>,
    },
    Demand {
        consumer: ActorRef<ConsumerCommand<T>>,
        cumulative: u64,
    },
    Cancel {
        consumer: ActorRef<ConsumerCommand<T>>,
    },
    RemoteFailure {
        consumer: ActorRef<ConsumerCommand<T>>,
        error: StreamError,
    },
    Ack,
}

#[allow(dead_code)]
enum ConsumerCommand<T> {
    OnSubscribe {
        producer: ActorRef<ProducerCommand<T>>,
    },
    Element {
        producer: ActorRef<ProducerCommand<T>>,
        seq: u64,
        item: T,
    },
    Complete {
        producer: ActorRef<ProducerCommand<T>>,
        seq: u64,
    },
    Failure {
        producer: ActorRef<ProducerCommand<T>>,
        error: StreamError,
    },
}

#[allow(dead_code)]
enum ConsumerEndpoint<T> {
    Actor(ActorRef<ConsumerCommand<T>>),
    Direct(Weak<ConsumerShared<T>>),
}

impl<T> Clone for ConsumerEndpoint<T> {
    fn clone(&self) -> Self {
        match self {
            Self::Actor(actor) => Self::Actor(actor.clone()),
            Self::Direct(shared) => Self::Direct(Weak::clone(shared)),
        }
    }
}

#[allow(dead_code)]
enum ProducerEndpoint<T> {
    Actor(ActorRef<ProducerCommand<T>>),
    Direct(Weak<ProducerShared<T>>),
}

impl<T> Clone for ProducerEndpoint<T> {
    fn clone(&self) -> Self {
        match self {
            Self::Actor(actor) => Self::Actor(actor.clone()),
            Self::Direct(shared) => Self::Direct(Weak::clone(shared)),
        }
    }
}

#[allow(dead_code)]
enum ProducerEndpointHandle<T> {
    Actor {
        actor: ActorRef<ProducerCommand<T>>,
        shared: Arc<ProducerShared<T>>,
    },
    Direct(Arc<ProducerShared<T>>),
}

impl<T> ProducerEndpointHandle<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    fn stop_actor(&self) {
        if let Self::Actor { actor, .. } = self {
            actor.stop(None);
        }
    }

    fn actor_ref(&self) -> Option<ActorRef<ProducerCommand<T>>> {
        match self {
            Self::Actor { actor, .. } => Some(actor.clone()),
            Self::Direct(_) => None,
        }
    }

    fn direct_shared(&self) -> Option<Arc<ProducerShared<T>>> {
        match self {
            Self::Actor { shared, .. } | Self::Direct(shared) => Some(Arc::clone(shared)),
        }
    }
}

#[allow(dead_code)]
struct LazyProducerStatus<T> {
    direct: Arc<SourceRefDirectState<T>>,
}

#[allow(dead_code)]
impl<T> LazyProducerStatus<T>
where
    T: Send + 'static,
{
    fn new(direct: Arc<SourceRefDirectState<T>>) -> Self {
        Self { direct }
    }

    fn get_status(&self) -> ractor::ActorStatus {
        self.direct.status()
    }
}

struct SourceRefDirectState<T> {
    inner: Mutex<SourceRefDirectInner<T>>,
}

struct SourceRefDirectInner<T> {
    input: Option<BoxStream<T>>,
    terminal: Option<StreamError>,
    subscribed: bool,
}

impl<T> SourceRefDirectState<T>
where
    T: Send + 'static,
{
    fn new(input: BoxStream<T>) -> Self {
        Self {
            inner: Mutex::new(SourceRefDirectInner {
                input: Some(input),
                terminal: None,
                subscribed: false,
            }),
        }
    }

    fn lock(&self) -> MutexGuard<'_, SourceRefDirectInner<T>> {
        self.inner
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    fn claim_input(&self) -> StreamResult<BoxStream<T>> {
        let mut inner = self.lock();
        inner.subscribed = true;
        if let Some(error) = inner.terminal.clone() {
            return Err(error);
        }
        inner.input.take().ok_or(StreamError::ActorTerminated)
    }

    fn timeout_if_unsubscribed(&self) {
        let mut inner = self.lock();
        if !inner.subscribed && inner.terminal.is_none() {
            let input = inner.input.take();
            inner.terminal = Some(subscription_timeout_error("stream ref sink"));
            drop(inner);
            drop(input);
        }
    }

    #[allow(dead_code)]
    fn status(&self) -> ractor::ActorStatus {
        let inner = self.lock();
        if inner.terminal.is_some() || inner.input.is_none() {
            ractor::ActorStatus::Stopped
        } else {
            ractor::ActorStatus::Running
        }
    }
}

struct SourceRefDirectStream<T>
where
    T: Send + 'static,
{
    input: Option<BoxStream<T>>,
    queue: VecDeque<T>,
    terminal: Option<ConsumerTerminal>,
    settings: StreamRefSettings,
    terminated: bool,
    _keep_alive: Arc<SourceRefInner<T>>,
}

impl<T> SourceRefDirectStream<T>
where
    T: Send + 'static,
{
    fn new(
        input: BoxStream<T>,
        settings: StreamRefSettings,
        keep_alive: Arc<SourceRefInner<T>>,
    ) -> Self {
        Self {
            input: Some(input),
            queue: VecDeque::new(),
            terminal: None,
            settings,
            terminated: false,
            _keep_alive: keep_alive,
        }
    }

    fn fill_prefetch(&mut self) {
        while self.terminal.is_none() && self.queue.len() < self.settings.buffer_capacity {
            let Some(input) = self.input.as_mut() else {
                self.terminal = Some(ConsumerTerminal::Error(StreamError::Cancelled));
                break;
            };
            match input.next() {
                Some(Ok(item)) => self.queue.push_back(item),
                Some(Err(error)) => {
                    self.input.take();
                    self.terminal = Some(ConsumerTerminal::Error(error));
                }
                None => {
                    self.input.take();
                    self.terminal = Some(ConsumerTerminal::Complete);
                }
            }
        }
    }
}

impl<T> Iterator for SourceRefDirectStream<T>
where
    T: Send + 'static,
{
    type Item = StreamResult<T>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.terminated {
            return None;
        }

        self.fill_prefetch();
        if let Some(item) = self.queue.pop_front() {
            self.fill_prefetch();
            return Some(Ok(item));
        }

        match self.terminal.clone() {
            Some(ConsumerTerminal::Complete) => {
                self.terminated = true;
                None
            }
            Some(ConsumerTerminal::Error(error)) => {
                self.terminated = true;
                Some(Err(error))
            }
            None => None,
        }
    }
}

impl<T> Drop for SourceRefDirectStream<T>
where
    T: Send + 'static,
{
    fn drop(&mut self) {
        self.input.take();
        self.queue.clear();
    }
}

struct SinkRefDirectState<T> {
    inner: Mutex<SinkRefDirectInner<T>>,
    changed: Condvar,
}

struct SinkRefDirectInner<T> {
    input: Option<BoxStream<T>>,
    completion: Option<oneshot::Sender<StreamResult<NotUsed>>>,
    completion_cancelled: Option<Arc<AtomicBool>>,
    terminal: Option<StreamError>,
    subscribed: bool,
}

impl<T> SinkRefDirectState<T>
where
    T: Send + 'static,
{
    fn new() -> Self {
        Self {
            inner: Mutex::new(SinkRefDirectInner {
                input: None,
                completion: None,
                completion_cancelled: None,
                terminal: None,
                subscribed: false,
            }),
            changed: Condvar::new(),
        }
    }

    fn lock(&self) -> MutexGuard<'_, SinkRefDirectInner<T>> {
        self.inner
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    fn attach_input(&self, input: BoxStream<T>) -> StreamCompletion<NotUsed> {
        let mut inner = self.lock();
        if let Some(error) = inner.terminal.clone() {
            return StreamCompletion::ready(Err(error));
        }
        if inner.subscribed {
            return StreamCompletion::ready(Err(StreamError::Failed(
                "stream ref was already subscribed by another endpoint".to_owned(),
            )));
        }

        let (sender, receiver) = oneshot::channel();
        let cancellation = StreamCancellation::for_external_completion();
        inner.input = Some(input);
        inner.completion = Some(sender);
        inner.completion_cancelled = Some(cancellation.cancelled());
        inner.subscribed = true;
        drop(inner);
        self.changed.notify_all();
        StreamCompletion::from_receiver(receiver, Some(cancellation))
    }

    fn wait_for_input(&self, settings: StreamRefSettings) -> StreamResult<BoxStream<T>> {
        let deadline = Instant::now()
            .checked_add(settings.subscription_timeout)
            .unwrap_or_else(far_future);
        let mut inner = self.lock();
        loop {
            if let Some(cancelled) = &inner.completion_cancelled
                && cancelled.load(Ordering::SeqCst)
            {
                let input = inner.input.take();
                inner.terminal = Some(StreamError::Cancelled);
                inner.completion.take();
                inner.completion_cancelled = None;
                drop(inner);
                drop(input);
                self.changed.notify_all();
                return Err(StreamError::Cancelled);
            }
            if let Some(input) = inner.input.take() {
                return Ok(input);
            }
            if let Some(error) = inner.terminal.clone() {
                return Err(error);
            }
            let now = Instant::now();
            if now >= deadline {
                let error = subscription_timeout_error("stream ref source");
                inner.terminal = Some(error.clone());
                if let Some(sender) = inner.completion.take() {
                    let _ = sender.send(Err(error.clone()));
                }
                return Err(error);
            }
            let remaining = deadline.saturating_duration_since(now);
            let (next, _) =
                wait_timeout_unpoison(&self.changed, inner, remaining.min(STREAM_REF_WAIT_POLL));
            inner = next;
        }
    }

    fn settle(&self, result: StreamResult<NotUsed>) {
        let mut inner = self.lock();
        if let Some(sender) = inner.completion.take() {
            let _ = sender.send(result);
        }
        inner.completion_cancelled = None;
        drop(inner);
        self.changed.notify_all();
    }

    fn fail_unattached(&self, error: StreamError) {
        let mut inner = self.lock();
        if inner.terminal.is_none() {
            let input = inner.input.take();
            inner.terminal = Some(error.clone());
            if let Some(sender) = inner.completion.take() {
                let _ = sender.send(Err(error));
            }
            inner.completion_cancelled = None;
            drop(inner);
            drop(input);
            self.changed.notify_all();
        }
    }

    fn is_completion_cancelled(&self) -> bool {
        self.lock()
            .completion_cancelled
            .as_ref()
            .is_some_and(|cancelled| cancelled.load(Ordering::SeqCst))
    }
}

struct SinkRefDirectStream<T>
where
    T: Send + 'static,
{
    state: Arc<SinkRefDirectState<T>>,
    input: Option<BoxStream<T>>,
    queue: VecDeque<T>,
    terminal: Option<ConsumerTerminal>,
    settings: StreamRefSettings,
    terminated: bool,
}

impl<T> SinkRefDirectStream<T>
where
    T: Send + 'static,
{
    fn new(state: Arc<SinkRefDirectState<T>>, settings: StreamRefSettings) -> Self {
        Self {
            state,
            input: None,
            queue: VecDeque::new(),
            terminal: None,
            settings,
            terminated: false,
        }
    }

    fn ensure_input(&mut self) -> StreamResult<()> {
        if self.input.is_none() && self.terminal.is_none() {
            self.input = Some(self.state.wait_for_input(self.settings)?);
        }
        Ok(())
    }

    fn fill_prefetch(&mut self) -> StreamResult<()> {
        self.ensure_input()?;
        while self.terminal.is_none() && self.queue.len() < self.settings.buffer_capacity {
            if self.state.is_completion_cancelled() {
                self.input.take();
                self.state.settle(Err(StreamError::Cancelled));
                self.terminal = Some(ConsumerTerminal::Error(StreamError::Cancelled));
                break;
            }
            let Some(input) = self.input.as_mut() else {
                self.terminal = Some(ConsumerTerminal::Error(StreamError::Cancelled));
                break;
            };
            match input.next() {
                Some(Ok(item)) => self.queue.push_back(item),
                Some(Err(error)) => {
                    self.input.take();
                    self.state.settle(Err(error.clone()));
                    self.terminal = Some(ConsumerTerminal::Error(error));
                }
                None => {
                    self.input.take();
                    self.state.settle(Ok(NotUsed));
                    self.terminal = Some(ConsumerTerminal::Complete);
                }
            }
        }
        Ok(())
    }
}

impl<T> Iterator for SinkRefDirectStream<T>
where
    T: Send + 'static,
{
    type Item = StreamResult<T>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.terminated {
            return None;
        }

        if let Err(error) = self.fill_prefetch() {
            self.terminated = true;
            return Some(Err(error));
        }
        if let Some(item) = self.queue.pop_front() {
            if let Err(error) = self.fill_prefetch() {
                self.terminal = Some(ConsumerTerminal::Error(error));
            }
            return Some(Ok(item));
        }

        match self.terminal.clone() {
            Some(ConsumerTerminal::Complete) => {
                self.terminated = true;
                None
            }
            Some(ConsumerTerminal::Error(error)) => {
                self.terminated = true;
                Some(Err(error))
            }
            None => None,
        }
    }
}

impl<T> Drop for SinkRefDirectStream<T>
where
    T: Send + 'static,
{
    fn drop(&mut self) {
        if !self.terminated {
            self.input.take();
            self.queue.clear();
            self.state.fail_unattached(StreamError::Cancelled);
            self.state.settle(Err(StreamError::Cancelled));
        }
    }
}

#[allow(dead_code)]
struct ProducerShared<T> {
    inner: Mutex<ProducerInner<T>>,
    changed: Condvar,
}

#[allow(dead_code)]
struct ProducerInner<T> {
    consumer: Option<ConsumerEndpoint<T>>,
    cumulative_demand: u64,
    sent: u64,
    stopped: Option<StreamError>,
}

#[allow(dead_code)]
impl<T> ProducerShared<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    fn new() -> Self {
        Self {
            inner: Mutex::new(ProducerInner {
                consumer: None,
                cumulative_demand: 0,
                sent: 0,
                stopped: None,
            }),
            changed: Condvar::new(),
        }
    }

    fn lock(&self) -> MutexGuard<'_, ProducerInner<T>> {
        self.inner
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    fn set_consumer(
        &self,
        consumer: ActorRef<ConsumerCommand<T>>,
    ) -> Result<bool, ActorRef<ConsumerCommand<T>>> {
        let mut inner = self.lock();
        match &inner.consumer {
            Some(ConsumerEndpoint::Actor(existing)) if same_actor(existing, &consumer) => Ok(false),
            Some(_) => Err(consumer),
            None if inner.stopped.is_some() => Err(consumer),
            None => {
                inner.consumer = Some(ConsumerEndpoint::Actor(consumer));
                drop(inner);
                self.changed.notify_all();
                Ok(true)
            }
        }
    }

    fn set_direct_consumer(&self, consumer: &Arc<ConsumerShared<T>>) -> StreamResult<bool> {
        let mut inner = self.lock();
        match &inner.consumer {
            Some(ConsumerEndpoint::Direct(existing))
                if existing.ptr_eq(&Arc::downgrade(consumer)) =>
            {
                Ok(false)
            }
            Some(_) => Err(StreamError::Failed(
                "stream ref was already subscribed by another endpoint".to_owned(),
            )),
            None => {
                if let Some(error) = inner.stopped.clone() {
                    return Err(error);
                }
                inner.consumer = Some(ConsumerEndpoint::Direct(Arc::downgrade(consumer)));
                drop(inner);
                self.changed.notify_all();
                Ok(true)
            }
        }
    }

    fn update_demand(&self, consumer: &ActorRef<ConsumerCommand<T>>, cumulative: u64) {
        self.update_demand_if(cumulative, |existing| {
            matches!(existing, ConsumerEndpoint::Actor(actor) if same_actor(actor, consumer))
        });
    }

    fn update_direct_demand(&self, consumer: &Arc<ConsumerShared<T>>, cumulative: u64) {
        let consumer = Arc::downgrade(consumer);
        self.update_demand_if(cumulative, |existing| {
            matches!(existing, ConsumerEndpoint::Direct(shared) if shared.ptr_eq(&consumer))
        });
    }

    fn update_demand_if(
        &self,
        cumulative: u64,
        matches_consumer: impl FnOnce(&ConsumerEndpoint<T>) -> bool,
    ) {
        let mut inner = self.lock();
        if inner.consumer.as_ref().is_some_and(matches_consumer)
            && cumulative > inner.cumulative_demand
        {
            inner.cumulative_demand = cumulative;
            drop(inner);
            self.changed.notify_all();
        }
    }

    fn stop_from_consumer(&self, consumer: &ActorRef<ConsumerCommand<T>>, error: StreamError) {
        self.stop_from_consumer_if(error, |existing| {
            matches!(existing, ConsumerEndpoint::Actor(actor) if same_actor(actor, consumer))
        });
    }

    fn stop_from_direct_consumer(&self, consumer: &Arc<ConsumerShared<T>>, error: StreamError) {
        let consumer = Arc::downgrade(consumer);
        self.stop_from_consumer_if(error, |existing| {
            matches!(existing, ConsumerEndpoint::Direct(shared) if shared.ptr_eq(&consumer))
        });
    }

    fn stop_from_consumer_if(
        &self,
        error: StreamError,
        matches_consumer: impl FnOnce(&ConsumerEndpoint<T>) -> bool,
    ) {
        let mut inner = self.lock();
        if inner.consumer.as_ref().is_none_or(matches_consumer) && inner.stopped.is_none() {
            inner.stopped = Some(error);
            drop(inner);
            self.changed.notify_all();
        }
    }

    fn stop_unless_finished(&self, error: StreamError) {
        let mut inner = self.lock();
        if inner.stopped.is_none() {
            inner.stopped = Some(error);
            drop(inner);
            self.changed.notify_all();
        }
    }
}

#[allow(dead_code)]
struct ConsumerShared<T> {
    inner: Mutex<ConsumerInner<T>>,
    changed: Condvar,
}

#[allow(dead_code)]
struct ConsumerInner<T> {
    producer: Option<ProducerEndpoint<T>>,
    queue: VecDeque<T>,
    terminal: Option<ConsumerTerminal>,
    expected_seq: u64,
    delivered: u64,
    cumulative_demand: u64,
}

#[derive(Clone)]
enum ConsumerTerminal {
    Complete,
    Error(StreamError),
}

#[allow(dead_code)]
impl<T> ConsumerShared<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    fn new(_settings: StreamRefSettings) -> Self {
        Self {
            inner: Mutex::new(ConsumerInner {
                producer: None,
                queue: VecDeque::new(),
                terminal: None,
                expected_seq: 0,
                delivered: 0,
                cumulative_demand: 0,
            }),
            changed: Condvar::new(),
        }
    }

    fn lock(&self) -> MutexGuard<'_, ConsumerInner<T>> {
        self.inner
            .lock()
            .unwrap_or_else(|poison| poison.into_inner())
    }

    fn set_producer(&self, producer: ActorRef<ProducerCommand<T>>) -> bool {
        let mut inner = self.lock();
        match &inner.producer {
            Some(ProducerEndpoint::Actor(existing)) => same_actor(existing, &producer),
            Some(ProducerEndpoint::Direct(_)) => false,
            None => {
                inner.producer = Some(ProducerEndpoint::Actor(producer));
                drop(inner);
                self.changed.notify_all();
                true
            }
        }
    }

    fn set_direct_producer(&self, producer: &Arc<ProducerShared<T>>) -> StreamResult<bool> {
        let mut inner = self.lock();
        match &inner.producer {
            Some(ProducerEndpoint::Direct(existing))
                if existing.ptr_eq(&Arc::downgrade(producer)) =>
            {
                Ok(false)
            }
            Some(_) => Err(StreamError::Failed(
                "stream ref was already subscribed by another endpoint".to_owned(),
            )),
            None => {
                if let Some(terminal) = inner.terminal.clone() {
                    return match terminal {
                        ConsumerTerminal::Complete => Err(StreamError::ActorTerminated),
                        ConsumerTerminal::Error(error) => Err(error),
                    };
                }
                inner.producer = Some(ProducerEndpoint::Direct(Arc::downgrade(producer)));
                drop(inner);
                self.changed.notify_all();
                Ok(true)
            }
        }
    }

    fn push(&self, producer: &ActorRef<ProducerCommand<T>>, seq: u64, item: T) {
        self.push_if(seq, item, |inner| producer_matches_actor(inner, producer));
    }

    fn push_direct(&self, producer: &Arc<ProducerShared<T>>, seq: u64, item: T) {
        let producer = Arc::downgrade(producer);
        self.push_if(seq, item, |inner| producer_matches_direct(inner, &producer));
    }

    fn push_if(&self, seq: u64, item: T, matches_producer: impl FnOnce(&ConsumerInner<T>) -> bool) {
        let mut inner = self.lock();
        if inner.terminal.is_some() || !matches_producer(&inner) {
            return;
        }
        let should_notify;
        if seq != inner.expected_seq {
            inner.queue.clear();
            inner.terminal = Some(ConsumerTerminal::Error(invalid_sequence_error(
                inner.expected_seq,
                seq,
                "stream ref element sequence gap",
            )));
            should_notify = true;
        } else {
            should_notify = inner.queue.is_empty();
            inner.expected_seq += 1;
            inner.queue.push_back(item);
        }
        drop(inner);
        if should_notify {
            self.changed.notify_all();
        }
    }

    fn complete(&self, producer: &ActorRef<ProducerCommand<T>>, seq: u64) {
        self.complete_if(seq, |inner| producer_matches_actor(inner, producer));
    }

    fn complete_direct(&self, producer: &Arc<ProducerShared<T>>, seq: u64) {
        let producer = Arc::downgrade(producer);
        self.complete_if(seq, |inner| producer_matches_direct(inner, &producer));
    }

    fn complete_if(&self, seq: u64, matches_producer: impl FnOnce(&ConsumerInner<T>) -> bool) {
        let mut inner = self.lock();
        if inner.terminal.is_some() || !matches_producer(&inner) {
            return;
        }
        if seq != inner.expected_seq {
            inner.queue.clear();
            inner.terminal = Some(ConsumerTerminal::Error(invalid_sequence_error(
                inner.expected_seq,
                seq,
                "stream ref completion sequence gap",
            )));
        } else {
            inner.terminal = Some(ConsumerTerminal::Complete);
        }
        drop(inner);
        self.changed.notify_all();
    }

    fn fail(&self, producer: &ActorRef<ProducerCommand<T>>, error: StreamError) {
        self.fail_if(error, |inner| producer_matches_actor(inner, producer));
    }

    fn fail_direct(&self, producer: &Arc<ProducerShared<T>>, error: StreamError) {
        let producer = Arc::downgrade(producer);
        self.fail_if(error, |inner| producer_matches_direct(inner, &producer));
    }

    fn fail_if(
        &self,
        error: StreamError,
        matches_producer: impl FnOnce(&ConsumerInner<T>) -> bool,
    ) {
        let mut inner = self.lock();
        if inner.terminal.is_some() || !matches_producer(&inner) {
            return;
        }
        inner.queue.clear();
        inner.terminal = Some(ConsumerTerminal::Error(error));
        drop(inner);
        self.changed.notify_all();
    }

    fn fail_local(&self, error: StreamError) {
        let mut inner = self.lock();
        if inner.terminal.is_none() {
            inner.queue.clear();
            inner.terminal = Some(ConsumerTerminal::Error(error));
            drop(inner);
            self.changed.notify_all();
        }
    }
}

#[allow(dead_code)]
fn producer_matches_actor<T>(
    inner: &ConsumerInner<T>,
    producer: &ActorRef<ProducerCommand<T>>,
) -> bool
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    matches!(
        inner.producer.as_ref(),
        Some(ProducerEndpoint::Actor(existing)) if same_actor(existing, producer)
    )
}

#[allow(dead_code)]
fn producer_matches_direct<T>(inner: &ConsumerInner<T>, producer: &Weak<ProducerShared<T>>) -> bool
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    matches!(
        inner.producer.as_ref(),
        Some(ProducerEndpoint::Direct(existing)) if existing.ptr_eq(producer)
    )
}

#[allow(dead_code)]
struct ProducerActor<T> {
    shared: Arc<ProducerShared<T>>,
    initial_consumer: Option<ActorRef<ConsumerCommand<T>>>,
}

impl<T> Actor for ProducerActor<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    type Msg = ProducerCommand<T>;
    type State = ();
    type Arguments = ();

    async fn pre_start(
        &self,
        myself: ActorRef<Self::Msg>,
        _args: Self::Arguments,
    ) -> Result<Self::State, ActorProcessingErr> {
        if let Some(consumer) = &self.initial_consumer {
            register_producer_consumer(&self.shared, myself, consumer.clone());
        }
        Ok(())
    }

    async fn handle(
        &self,
        myself: ActorRef<Self::Msg>,
        message: Self::Msg,
        _state: &mut Self::State,
    ) -> ActorResult {
        match message {
            ProducerCommand::Subscribe { consumer } => {
                register_producer_consumer(&self.shared, myself, consumer);
            }
            ProducerCommand::Demand {
                consumer,
                cumulative,
            } => self.shared.update_demand(&consumer, cumulative),
            ProducerCommand::Cancel { consumer } => {
                self.shared
                    .stop_from_consumer(&consumer, StreamError::Cancelled);
            }
            ProducerCommand::RemoteFailure { consumer, error } => {
                self.shared.stop_from_consumer(&consumer, error);
            }
            ProducerCommand::Ack => myself.stop(None),
        }
        Ok(())
    }

    async fn post_stop(
        &self,
        _myself: ActorRef<Self::Msg>,
        _state: &mut Self::State,
    ) -> ActorResult {
        self.shared
            .stop_unless_finished(StreamError::ActorTerminated);
        Ok(())
    }
}

#[allow(dead_code)]
fn register_producer_consumer<T>(
    shared: &Arc<ProducerShared<T>>,
    producer: ActorRef<ProducerCommand<T>>,
    consumer: ActorRef<ConsumerCommand<T>>,
) where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    match shared.set_consumer(consumer.clone()) {
        Ok(true) | Ok(false) => {
            let _ = cast_consumer_on_subscribe(&consumer, producer.clone());
        }
        Err(duplicate) => {
            let _ = cast_consumer_failure(
                &duplicate,
                producer,
                StreamError::Failed(
                    "stream ref was already subscribed by another endpoint".to_owned(),
                ),
            );
        }
    }
}

#[allow(dead_code)]
fn register_direct_producer_consumer<T>(
    producer: &Arc<ProducerShared<T>>,
    consumer: &Arc<ConsumerShared<T>>,
) -> StreamResult<()>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    consumer.set_direct_producer(producer)?;
    if let Err(error) = producer.set_direct_consumer(consumer) {
        consumer.fail_local(error.clone());
        return Err(error);
    }
    Ok(())
}

#[allow(dead_code)]
struct ConsumerActor<T> {
    shared: Arc<ConsumerShared<T>>,
    initial_producer: Option<ActorRef<ProducerCommand<T>>>,
}

impl<T> Actor for ConsumerActor<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    type Msg = ConsumerCommand<T>;
    type State = ();
    type Arguments = ();

    async fn pre_start(
        &self,
        myself: ActorRef<Self::Msg>,
        _args: Self::Arguments,
    ) -> Result<Self::State, ActorProcessingErr> {
        if let Some(producer) = &self.initial_producer {
            self.shared.set_producer(producer.clone());
            if let Err(error) = cast_producer_subscribe(producer, myself.clone()) {
                self.shared.fail_local(error);
                myself.stop(None);
            }
        }
        Ok(())
    }

    async fn handle(
        &self,
        _myself: ActorRef<Self::Msg>,
        message: Self::Msg,
        _state: &mut Self::State,
    ) -> ActorResult {
        match message {
            ConsumerCommand::OnSubscribe { producer } => {
                if !self.shared.set_producer(producer.clone()) {
                    let _ = cast_producer_remote_failure(
                        &producer,
                        _myself.clone(),
                        StreamError::Failed(
                            "stream ref was already subscribed by another endpoint".to_owned(),
                        ),
                    );
                }
            }
            ConsumerCommand::Element {
                producer,
                seq,
                item,
            } => self.shared.push(&producer, seq, item),
            ConsumerCommand::Complete { producer, seq } => {
                self.shared.complete(&producer, seq);
                let _ = cast_actor(&producer, ProducerCommand::Ack);
            }
            ConsumerCommand::Failure { producer, error } => {
                self.shared.fail(&producer, error);
                let _ = cast_actor(&producer, ProducerCommand::Ack);
            }
        }
        Ok(())
    }

    async fn post_stop(
        &self,
        _myself: ActorRef<Self::Msg>,
        _state: &mut Self::State,
    ) -> ActorResult {
        self.shared.fail_local(StreamError::ActorTerminated);
        Ok(())
    }
}

#[allow(dead_code)]
struct ConsumerStream<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    shared: Arc<ConsumerShared<T>>,
    actor_ref: Option<ActorRef<ConsumerCommand<T>>>,
    settings: StreamRefSettings,
    terminated: bool,
    source_ref_keep_alive: Option<Arc<SourceRefInner<T>>>,
}

impl<T> Iterator for ConsumerStream<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    type Item = StreamResult<T>;

    fn next(&mut self) -> Option<Self::Item> {
        if self.terminated {
            return None;
        }

        if let Err(error) = self.wait_for_subscription() {
            self.terminated = true;
            return Some(Err(error));
        }

        if let Err(error) = self.redeliver_or_extend_demand() {
            self.terminated = true;
            return Some(Err(error));
        }

        let mut next_redelivery = next_redelivery_deadline(self.settings);
        loop {
            let demand_after_pop = {
                let mut inner = self.shared.lock();
                if let Some(item) = inner.queue.pop_front() {
                    inner.delivered = inner.delivered.saturating_add(1);
                    let demand = next_demand(&mut inner, self.settings);
                    drop(inner);
                    if let Some((producer, cumulative)) = demand
                        && let Err(error) =
                            send_demand(&self.shared, &self.actor_ref, &producer, cumulative)
                    {
                        self.terminated = true;
                        return Some(Err(error));
                    }
                    return Some(Ok(item));
                }

                if let Some(terminal) = inner.terminal.clone() {
                    drop(inner);
                    match terminal {
                        ConsumerTerminal::Complete => {
                            self.terminated = true;
                            self.stop_actor();
                            return None;
                        }
                        ConsumerTerminal::Error(error) => {
                            self.terminated = true;
                            self.stop_actor();
                            return Some(Err(error));
                        }
                    }
                }
                None::<(ProducerEndpoint<T>, u64)>
            };
            debug_assert!(demand_after_pop.is_none());

            let now = Instant::now();
            let timeout = next_redelivery.saturating_duration_since(now);
            let mut inner = self.shared.lock();
            if !inner.queue.is_empty() || inner.terminal.is_some() {
                continue;
            }
            let (next_inner, result) = wait_timeout_unpoison(
                &self.shared.changed,
                inner,
                timeout.min(STREAM_REF_WAIT_POLL),
            );
            inner = next_inner;
            drop(inner);
            if result.timed_out() && Instant::now() >= next_redelivery {
                if let Err(error) = self.redeliver_demand() {
                    self.terminated = true;
                    return Some(Err(error));
                }
                next_redelivery = next_redelivery_deadline(self.settings);
            }
        }
    }
}

#[allow(dead_code)]
impl<T> ConsumerStream<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    fn wait_for_subscription(&self) -> StreamResult<()> {
        let deadline = Instant::now()
            .checked_add(self.settings.subscription_timeout)
            .unwrap_or_else(far_future);
        let mut inner = self.shared.lock();
        loop {
            if inner.producer.is_some() {
                return Ok(());
            }
            if let Some(terminal) = inner.terminal.clone() {
                return match terminal {
                    ConsumerTerminal::Complete => Ok(()),
                    ConsumerTerminal::Error(error) => Err(error),
                };
            }
            let now = Instant::now();
            if now >= deadline {
                drop(inner);
                let error = subscription_timeout_error("stream ref source");
                self.shared.fail_local(error.clone());
                self.stop_actor_ref();
                return Err(error);
            }
            let remaining = deadline.saturating_duration_since(now);
            let (next, _) = wait_timeout_unpoison(
                &self.shared.changed,
                inner,
                remaining.min(STREAM_REF_WAIT_POLL),
            );
            inner = next;
        }
    }

    fn redeliver_or_extend_demand(&self) -> StreamResult<()> {
        let demand = {
            let mut inner = self.shared.lock();
            next_demand(&mut inner, self.settings)
        };
        if let Some((producer, cumulative)) = demand {
            send_demand(&self.shared, &self.actor_ref, &producer, cumulative)?;
        }
        Ok(())
    }

    fn redeliver_demand(&self) -> StreamResult<()> {
        let (producer, cumulative) = {
            let inner = self.shared.lock();
            let Some(producer) = inner.producer.clone() else {
                return Ok(());
            };
            if inner.cumulative_demand == 0 {
                return Ok(());
            }
            (producer, inner.cumulative_demand)
        };
        send_demand(&self.shared, &self.actor_ref, &producer, cumulative)
    }

    fn stop_actor(&mut self) {
        if let Some(actor_ref) = self.actor_ref.take() {
            actor_ref.stop(None);
        }
    }

    fn stop_actor_ref(&self) {
        if let Some(actor_ref) = &self.actor_ref {
            actor_ref.stop(None);
        }
    }
}

impl<T> Drop for ConsumerStream<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    fn drop(&mut self) {
        if !self.terminated {
            let producer = self.shared.lock().producer.clone();
            match producer {
                Some(ProducerEndpoint::Actor(producer)) => {
                    if let Some(consumer) = &self.actor_ref {
                        let _ = cast_producer_cancel(&producer, consumer.clone());
                    }
                }
                Some(ProducerEndpoint::Direct(producer)) => {
                    if let Some(producer) = producer.upgrade() {
                        producer.stop_from_direct_consumer(&self.shared, StreamError::Cancelled);
                    }
                }
                None => {}
            }
        }
        self.stop_actor();
        drop(self.source_ref_keep_alive.take());
    }
}

#[allow(dead_code)]
fn next_demand<T>(
    inner: &mut ConsumerInner<T>,
    settings: StreamRefSettings,
) -> Option<(ProducerEndpoint<T>, u64)> {
    if inner.terminal.is_some() {
        return None;
    }
    let remaining_credit = inner.cumulative_demand.saturating_sub(inner.delivered);
    if inner.cumulative_demand != 0 && remaining_credit > demand_replenish_threshold(settings) {
        return None;
    }
    let target = inner
        .delivered
        .saturating_add(settings.buffer_capacity as u64);
    if inner.cumulative_demand >= target {
        return None;
    }
    inner.cumulative_demand = target;
    inner
        .producer
        .as_ref()
        .map(|producer| (producer.clone(), inner.cumulative_demand))
}

#[allow(dead_code)]
fn send_demand<T>(
    consumer_shared: &Arc<ConsumerShared<T>>,
    consumer: &Option<ActorRef<ConsumerCommand<T>>>,
    producer: &ProducerEndpoint<T>,
    cumulative: u64,
) -> StreamResult<()>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    match producer {
        ProducerEndpoint::Actor(producer) => {
            let Some(consumer) = consumer else {
                return Err(StreamError::ActorTerminated);
            };
            cast_producer_demand(producer, consumer.clone(), cumulative)
        }
        ProducerEndpoint::Direct(producer) => {
            let Some(producer) = producer.upgrade() else {
                return Err(StreamError::ActorTerminated);
            };
            producer.update_direct_demand(consumer_shared, cumulative);
            Ok(())
        }
    }
}

#[allow(dead_code)]
fn demand_replenish_threshold(settings: StreamRefSettings) -> u64 {
    (settings.buffer_capacity as u64) / 2
}

fn stream_ref_source_sink<T>(settings: StreamRefSettings) -> Sink<T, SourceRef<T>>
where
    T: Send + 'static,
{
    Sink::from_runner(move |input, materializer| {
        let direct = Arc::new(SourceRefDirectState::new(input));
        let direct_for_timeout = Arc::clone(&direct);
        let timeout = materializer.schedule_once(settings.subscription_timeout, move || {
            direct_for_timeout.timeout_if_unsubscribed();
        });
        Ok(SourceRef {
            inner: Arc::new(SourceRefInner {
                producer: LazyProducerStatus::new(Arc::clone(&direct)),
                direct,
                settings,
                subscribed: AtomicBool::new(false),
                timeout: Some(timeout),
            }),
        })
    })
}

fn stream_ref_sink_source<T>(settings: StreamRefSettings) -> Source<T, SinkRef<T>>
where
    T: Send + 'static,
{
    Source::from_materialized_factory(move |_materializer| {
        let direct = Arc::new(SinkRefDirectState::new());
        let sink_ref = SinkRef {
            inner: Arc::new(SinkRefInner {
                direct: Arc::clone(&direct),
                subscribed: AtomicBool::new(false),
            }),
        };
        Ok((
            Box::new(SinkRefDirectStream::new(direct, settings)) as BoxStream<T>,
            sink_ref,
        ))
    })
}

#[allow(dead_code)]
fn send_element_to_consumer<T>(
    consumer: &ConsumerEndpoint<T>,
    producer: &ProducerEndpointHandle<T>,
    seq: u64,
    item: T,
) -> StreamResult<()>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    match consumer {
        ConsumerEndpoint::Actor(consumer) => {
            let Some(producer_actor) = producer.actor_ref() else {
                return Err(StreamError::ActorTerminated);
            };
            cast_actor(
                consumer,
                ConsumerCommand::Element {
                    producer: producer_actor,
                    seq,
                    item,
                },
            )
        }
        ConsumerEndpoint::Direct(consumer) => {
            let Some(consumer) = consumer.upgrade() else {
                return Err(StreamError::Cancelled);
            };
            let Some(producer) = producer.direct_shared() else {
                return Err(StreamError::ActorTerminated);
            };
            consumer.push_direct(&producer, seq, item);
            Ok(())
        }
    }
}

#[allow(dead_code)]
fn fail_consumer<T>(
    consumer: &ConsumerEndpoint<T>,
    producer: &ProducerEndpointHandle<T>,
    error: StreamError,
) -> StreamResult<bool>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    match consumer {
        ConsumerEndpoint::Actor(consumer) => {
            let Some(producer_actor) = producer.actor_ref() else {
                return Err(StreamError::ActorTerminated);
            };
            cast_actor(
                consumer,
                ConsumerCommand::Failure {
                    producer: producer_actor,
                    error,
                },
            )?;
            Ok(false)
        }
        ConsumerEndpoint::Direct(consumer) => {
            let Some(consumer) = consumer.upgrade() else {
                return Err(StreamError::Cancelled);
            };
            let Some(producer) = producer.direct_shared() else {
                return Err(StreamError::ActorTerminated);
            };
            consumer.fail_direct(&producer, error);
            Ok(true)
        }
    }
}

#[allow(dead_code)]
fn complete_consumer<T>(
    consumer: &ConsumerEndpoint<T>,
    producer: &ProducerEndpointHandle<T>,
    seq: u64,
) -> StreamResult<bool>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    match consumer {
        ConsumerEndpoint::Actor(consumer) => {
            let Some(producer_actor) = producer.actor_ref() else {
                return Err(StreamError::ActorTerminated);
            };
            cast_actor(
                consumer,
                ConsumerCommand::Complete {
                    producer: producer_actor,
                    seq,
                },
            )?;
            Ok(false)
        }
        ConsumerEndpoint::Direct(consumer) => {
            let Some(consumer) = consumer.upgrade() else {
                return Err(StreamError::Cancelled);
            };
            let Some(producer) = producer.direct_shared() else {
                return Err(StreamError::ActorTerminated);
            };
            consumer.complete_direct(&producer, seq);
            Ok(true)
        }
    }
}

#[allow(dead_code)]
fn run_producer_endpoint<T>(
    mut input: BoxStream<T>,
    shared: Arc<ProducerShared<T>>,
    producer_endpoint: ProducerEndpointHandle<T>,
    settings: StreamRefSettings,
    cancelled: Arc<AtomicBool>,
) -> StreamResult<NotUsed>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    let deadline = Instant::now()
        .checked_add(settings.subscription_timeout)
        .unwrap_or_else(far_future);

    loop {
        let consumer = match wait_for_remote_demand(&shared, deadline, &cancelled) {
            Ok(consumer) => consumer,
            Err(error) => {
                producer_endpoint.stop_actor();
                return Err(error);
            }
        };
        if cancelled.load(Ordering::SeqCst) {
            return Err(StreamError::Cancelled);
        }

        match input.next() {
            Some(Ok(item)) => {
                let seq = {
                    let mut inner = shared.lock();
                    let seq = inner.sent;
                    inner.sent = inner.sent.saturating_add(1);
                    seq
                };
                if let Err(error) =
                    send_element_to_consumer(&consumer, &producer_endpoint, seq, item)
                {
                    producer_endpoint.stop_actor();
                    return Err(match error {
                        StreamError::ActorTerminated => StreamError::Cancelled,
                        other => other,
                    });
                }
            }
            Some(Err(error)) => {
                if fail_consumer(&consumer, &producer_endpoint, error.clone()).unwrap_or(false) {
                    producer_endpoint.stop_actor();
                }
                return Err(error);
            }
            None => {
                let seq = shared.lock().sent;
                if complete_consumer(&consumer, &producer_endpoint, seq).unwrap_or(false) {
                    producer_endpoint.stop_actor();
                }
                return Ok(NotUsed);
            }
        }
    }
}

#[allow(dead_code)]
fn wait_for_remote_demand<T>(
    shared: &Arc<ProducerShared<T>>,
    deadline: Instant,
    cancelled: &Arc<AtomicBool>,
) -> StreamResult<ConsumerEndpoint<T>>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    let mut inner = shared.lock();
    loop {
        if cancelled.load(Ordering::SeqCst) {
            return Err(StreamError::Cancelled);
        }
        if let Some(error) = inner.stopped.clone() {
            return Err(error);
        }
        if let Some(consumer) = &inner.consumer
            && inner.sent < inner.cumulative_demand
        {
            return Ok(consumer.clone());
        }
        let now = Instant::now();
        if inner.consumer.is_none() && now >= deadline {
            return Err(subscription_timeout_error("stream ref sink"));
        }
        let remaining = deadline.saturating_duration_since(now);
        let timeout = if inner.consumer.is_none() {
            remaining.min(STREAM_REF_WAIT_POLL)
        } else {
            STREAM_REF_WAIT_POLL
        };
        let (next, _) = wait_timeout_unpoison(&shared.changed, inner, timeout);
        inner = next;
    }
}

#[allow(dead_code)]
fn spawn_producer_actor<T>(
    initial_consumer: Option<ActorRef<ConsumerCommand<T>>>,
    shared: Arc<ProducerShared<T>>,
) -> StreamResult<(
    ActorRef<ProducerCommand<T>>,
    ractor::concurrency::JoinHandle<()>,
)>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    block_on_ractor_runtime(Actor::spawn(
        None,
        ProducerActor {
            shared,
            initial_consumer,
        },
        (),
    ))?
    .map_err(|error| {
        StreamError::Failed(format!(
            "stream ref producer actor failed to spawn: {error}"
        ))
    })
}

#[allow(dead_code)]
fn spawn_consumer_actor<T>(
    initial_producer: Option<ActorRef<ProducerCommand<T>>>,
    shared: Arc<ConsumerShared<T>>,
) -> StreamResult<(
    ActorRef<ConsumerCommand<T>>,
    ractor::concurrency::JoinHandle<()>,
)>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    block_on_ractor_runtime(Actor::spawn(
        None,
        ConsumerActor {
            shared,
            initial_producer,
        },
        (),
    ))?
    .map_err(|error| {
        StreamError::Failed(format!(
            "stream ref consumer actor failed to spawn: {error}"
        ))
    })
}

#[allow(dead_code)]
fn cast_producer_subscribe<T>(
    producer: &ActorRef<ProducerCommand<T>>,
    consumer: ActorRef<ConsumerCommand<T>>,
) -> StreamResult<()>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    cast_actor(producer, ProducerCommand::Subscribe { consumer })
}

#[allow(dead_code)]
fn cast_producer_demand<T>(
    producer: &ActorRef<ProducerCommand<T>>,
    consumer: ActorRef<ConsumerCommand<T>>,
    cumulative: u64,
) -> StreamResult<()>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    cast_actor(
        producer,
        ProducerCommand::Demand {
            consumer,
            cumulative,
        },
    )
}

#[allow(dead_code)]
fn cast_producer_cancel<T>(
    producer: &ActorRef<ProducerCommand<T>>,
    consumer: ActorRef<ConsumerCommand<T>>,
) -> StreamResult<()>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    cast_actor(producer, ProducerCommand::Cancel { consumer })
}

#[allow(dead_code)]
fn cast_producer_remote_failure<T>(
    producer: &ActorRef<ProducerCommand<T>>,
    consumer: ActorRef<ConsumerCommand<T>>,
    error: StreamError,
) -> StreamResult<()>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    cast_actor(producer, ProducerCommand::RemoteFailure { consumer, error })
}

#[allow(dead_code)]
fn cast_consumer_on_subscribe<T>(
    consumer: &ActorRef<ConsumerCommand<T>>,
    producer: ActorRef<ProducerCommand<T>>,
) -> StreamResult<()>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    cast_actor(consumer, ConsumerCommand::OnSubscribe { producer })
}

#[allow(dead_code)]
fn cast_consumer_failure<T>(
    consumer: &ActorRef<ConsumerCommand<T>>,
    producer: ActorRef<ProducerCommand<T>>,
    error: StreamError,
) -> StreamResult<()>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    cast_actor(consumer, ConsumerCommand::Failure { producer, error })
}

#[allow(dead_code)]
fn cast_actor<Msg>(actor_ref: &ActorRef<Msg>, message: Msg) -> StreamResult<()>
where
    Msg: Message,
{
    match actor_ref.cast(message) {
        Ok(()) => Ok(()),
        Err(ractor::MessagingErr::SendErr(_)) | Err(ractor::MessagingErr::ChannelClosed) => {
            Err(StreamError::ActorTerminated)
        }
        Err(error) => Err(StreamError::ActorAskSendFailed {
            reason: error.to_string(),
        }),
    }
}

#[allow(dead_code)]
fn same_actor<MsgA, MsgB>(left: &ActorRef<MsgA>, right: &ActorRef<MsgB>) -> bool
where
    MsgA: Message,
    MsgB: Message,
{
    left.get_cell().get_id() == right.get_cell().get_id()
}

fn failed_once<T>(reason: &str) -> BoxStream<T>
where
    T: stream_ref_actor_bound::Bound + Send + 'static,
{
    failed_stream(StreamError::Failed(reason.to_owned()))
}

fn failed_stream<T>(error: StreamError) -> BoxStream<T>
where
    T: Send + 'static,
{
    Box::new(std::iter::once(Err(error)))
}

fn subscription_timeout_error(side: &str) -> StreamError {
    StreamError::Failed(format!(
        "{side} remote side did not subscribe within subscription timeout"
    ))
}

#[allow(dead_code)]
fn invalid_sequence_error(expected: u64, got: u64, context: &str) -> StreamError {
    StreamError::Failed(format!(
        "{context}: expected sequence {expected}, got {got}"
    ))
}

#[allow(dead_code)]
fn next_redelivery_deadline(settings: StreamRefSettings) -> Instant {
    Instant::now()
        .checked_add(settings.demand_redelivery_interval)
        .unwrap_or_else(far_future)
}

fn far_future() -> Instant {
    Instant::now() + Duration::from_secs(60 * 60 * 24 * 365)
}

fn wait_timeout_unpoison<'a, T>(
    condvar: &Condvar,
    guard: MutexGuard<'a, T>,
    timeout: Duration,
) -> (MutexGuard<'a, T>, std::sync::WaitTimeoutResult) {
    condvar
        .wait_timeout(guard, timeout)
        .unwrap_or_else(|poison| poison.into_inner())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{
        stream::{Keep, Source},
        testkit::TestSink,
    };
    use std::sync::{
        Arc as StdArc,
        atomic::{AtomicBool, AtomicUsize, Ordering},
    };

    fn wait_until(timeout: Duration, mut condition: impl FnMut() -> bool) -> bool {
        let deadline = Instant::now() + timeout;
        while Instant::now() < deadline {
            if condition() {
                return true;
            }
            std::thread::park_timeout(Duration::from_millis(1));
        }
        condition()
    }

    fn assert_condition_holds(timeout: Duration, mut condition: impl FnMut() -> bool) {
        let deadline = Instant::now() + timeout;
        while Instant::now() < deadline {
            assert!(condition());
            std::thread::park_timeout(Duration::from_millis(1));
        }
        assert!(condition());
    }

    fn short_settings() -> StreamRefSettings {
        StreamRefSettings::default()
            .with_buffer_capacity(1)
            .with_subscription_timeout(Duration::from_millis(50))
            .with_demand_redelivery_interval(Duration::from_millis(10))
    }

    #[test]
    fn source_ref_streams_elements_and_completion() {
        let source_ref = Source::from_iter(1_u64..=3)
            .run_with(StreamRefs::source_ref())
            .unwrap();

        assert_eq!(source_ref.source().run_collect().unwrap(), vec![1, 2, 3]);
    }

    #[test]
    fn same_process_source_ref_terminal_fold_keeps_direct_splice() {
        let closed = StdArc::new(AtomicBool::new(false));
        let close_flag = StdArc::clone(&closed);
        let source = Source::unfold_resource(
            || Ok(0_u64),
            |next| {
                if *next >= 128 {
                    Ok(None)
                } else {
                    let value = *next;
                    *next += 1;
                    Ok(Some(value))
                }
            },
            move |_state| {
                close_flag.store(true, Ordering::SeqCst);
                Ok(())
            },
        );
        let source_ref = source.run_with(StreamRefs::source_ref()).unwrap();

        let sum = source_ref
            .source()
            .run_with(Sink::fold(0_u64, |acc, item| acc + item))
            .unwrap()
            .wait()
            .unwrap();

        assert_eq!(sum, (0_u64..128).sum::<u64>());
        assert!(closed.load(Ordering::SeqCst));
    }

    #[test]
    fn sink_ref_streams_elements_and_completion() {
        let (sink_ref, completion) = StreamRefs::sink_ref::<u64>()
            .to_mat(Sink::collect(), Keep::both)
            .run()
            .unwrap();

        Source::from_iter(1_u64..=3)
            .run_with(sink_ref.sink())
            .unwrap()
            .wait()
            .unwrap();

        assert_eq!(completion.wait().unwrap(), vec![1, 2, 3]);
    }

    #[test]
    fn source_ref_propagates_upstream_failure() {
        let source_ref = Source::<u64>::failed(StreamError::Failed("boom".to_owned()))
            .run_with(StreamRefs::source_ref())
            .unwrap();

        assert_eq!(
            source_ref.source().run_collect(),
            Err(StreamError::Failed("boom".to_owned()))
        );
    }

    #[test]
    fn sink_ref_propagates_upstream_failure() {
        let (sink_ref, completion) = StreamRefs::sink_ref::<u64>()
            .to_mat(Sink::collect(), Keep::both)
            .run()
            .unwrap();

        let failure = Source::<u64>::failed(StreamError::Failed("remote boom".to_owned()))
            .run_with(sink_ref.sink())
            .unwrap()
            .wait();

        assert_eq!(failure, Err(StreamError::Failed("remote boom".to_owned())));
        assert_eq!(
            completion.wait(),
            Err(StreamError::Failed("remote boom".to_owned()))
        );
    }

    #[test]
    fn source_ref_cancellation_reaches_origin() {
        let closed = StdArc::new(AtomicBool::new(false));
        let close_flag = StdArc::clone(&closed);
        let source = Source::unfold_resource(
            || Ok(()),
            |_state| Ok(Some(1_u64)),
            move |_state| {
                close_flag.store(true, Ordering::SeqCst);
                Ok(())
            },
        );
        let source_ref = source.run_with(StreamRefs::source_ref()).unwrap();

        assert_eq!(source_ref.source().take(1).run_collect().unwrap(), vec![1]);
        assert!(wait_until(Duration::from_secs(1), || {
            closed.load(Ordering::SeqCst)
        }));
    }

    #[test]
    fn sink_ref_cancellation_stops_remote_producer() {
        let (sink_ref, completion) = StreamRefs::sink_ref::<u64>()
            .take(1)
            .to_mat(Sink::collect(), Keep::both)
            .run()
            .unwrap();

        let producer = Source::repeat(1_u64)
            .run_with(sink_ref.sink())
            .unwrap()
            .wait();

        assert_eq!(completion.wait().unwrap(), vec![1]);
        assert_eq!(producer, Err(StreamError::Cancelled));
    }

    #[test]
    fn source_ref_backpressures_across_ref() {
        let pulled = StdArc::new(AtomicUsize::new(0));
        let pulled_for_source = StdArc::clone(&pulled);
        let source = Source::unfold(0_u64, move |next| {
            pulled_for_source.fetch_add(1, Ordering::SeqCst);
            Some((next + 1, next))
        });
        let source_ref = source
            .run_with(StreamRefs::source_ref_with_settings(short_settings()))
            .unwrap();
        let mut probe = source_ref.source().run_with(TestSink::probe()).unwrap();

        probe.request(1);
        probe.assert_next(0);
        assert!(wait_until(Duration::from_secs(1), || {
            pulled.load(Ordering::SeqCst) >= 2
        }));
        assert_condition_holds(Duration::from_millis(50), || {
            pulled.load(Ordering::SeqCst) <= 2
        });

        probe.request(1);
        probe.assert_next(1);
        assert!(wait_until(Duration::from_secs(1), || {
            pulled.load(Ordering::SeqCst) >= 3
        }));
        assert_condition_holds(Duration::from_millis(50), || {
            pulled.load(Ordering::SeqCst) <= 3
        });

        probe.cancel();
    }

    #[test]
    fn source_ref_late_subscription_observes_timeout() {
        let source_ref = Source::repeat(1_u64)
            .run_with(StreamRefs::source_ref_with_settings(short_settings()))
            .unwrap();

        assert!(wait_until(Duration::from_secs(1), || {
            source_ref.inner.producer.get_status() == ractor::ActorStatus::Stopped
        }));

        let result = source_ref.source().run_collect();
        assert!(matches!(
            result,
            Err(StreamError::ActorTerminated) | Err(StreamError::Failed(_))
        ));
    }

    #[test]
    fn sink_ref_subscription_timeout_fails_local_source() {
        let (_sink_ref, probe) = StreamRefs::sink_ref_with_settings::<u64>(short_settings())
            .to_mat(TestSink::probe(), Keep::both)
            .run()
            .unwrap();

        probe.request(1);
        let error = probe.expect_error();
        assert!(
            matches!(error, StreamError::Failed(message) if message.contains("did not subscribe"))
        );
    }

    #[test]
    fn stream_refs_are_one_shot() {
        let source_ref = Source::from_iter([1_u64])
            .run_with(StreamRefs::source_ref())
            .unwrap();
        assert_eq!(source_ref.source().run_collect().unwrap(), vec![1]);
        assert!(matches!(
            source_ref.source().run_collect(),
            Err(StreamError::Failed(message)) if message.contains("already")
        ));

        let (sink_ref, completion) = StreamRefs::sink_ref::<u64>()
            .to_mat(Sink::collect(), Keep::both)
            .run()
            .unwrap();
        Source::single(1_u64)
            .run_with(sink_ref.sink())
            .unwrap()
            .wait()
            .unwrap();
        assert!(matches!(
            Source::single(2_u64).run_with(sink_ref.sink()).unwrap().wait(),
            Err(StreamError::Failed(message)) if message.contains("already")
        ));
        assert_eq!(completion.wait().unwrap(), vec![1]);
    }
}