continuo 0.12.0

use std::sync::Mutex;
use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering};

use async_trait::async_trait;

use super::*;
use crate::{Component, Finalizable, Runnable};

// Deliberately no HasRegistry: low-level runtime management and DI remain
// independent, even when a later generation needs a newly published value.
#[derive(Clone, Default)]
struct State {
    shutdown: CancellationToken,
    #[cfg(feature = "fleet")]
    scope: crate::fleet::Scope,
}

impl SharedState for State {
    fn shutdown_token(&self) -> CancellationToken {
        self.shutdown.clone()
    }

    #[cfg(feature = "fleet")]
    fn process_scope(&self) -> crate::fleet::Scope {
        self.scope
    }
}

struct Dependency;

struct Consumer {
    policy: AtomicU8,
    seen: Mutex<Vec<Arc<Dependency>>>,
    finalized: AtomicUsize,
}

impl Consumer {
    fn new(policy: u8) -> Self {
        Self {
            policy: AtomicU8::new(policy),
            seen: Mutex::new(Vec::new()),
            finalized: AtomicUsize::new(0),
        }
    }
}

impl Component<State> for Consumer {
    type Deps = (Arc<Dependency>,);

    fn name(&self) -> &str {
        "consumer"
    }

    #[cfg(feature = "fleet")]
    fn as_scoped(&self) -> Option<&dyn crate::fleet::Scoped> {
        Some(self)
    }

    fn as_finalizable(&self) -> Option<&dyn Finalizable<State>> {
        Some(self)
    }

    fn as_runnable(
        self: Arc<Self>
    ) -> Option<Arc<dyn Runnable<State, Self::Deps>>> {
        Some(self)
    }
}

#[cfg(feature = "fleet")]
impl crate::fleet::Scoped for Consumer {
    fn scope(&self) -> crate::fleet::Scope {
        crate::fleet::Scope::Worker
    }
}

#[async_trait]
impl Runnable<State, (Arc<Dependency>,)> for Consumer {
    fn start_policy(
        &self,
        _: &State,
    ) -> ServiceStartPolicy {
        match self.policy.load(Ordering::SeqCst) {
            0 => ServiceStartPolicy::Unavailable,
            1 => ServiceStartPolicy::Manual,
            3 => {
                // A concurrent policy change after initial admission must not
                // change which initial generations were already admitted.
                self.policy.store(0, Ordering::SeqCst);
                ServiceStartPolicy::Automatic
            }
            _ => ServiceStartPolicy::Automatic,
        }
    }

    async fn run(
        self: Arc<Self>,
        (dependency,): (Arc<Dependency>,),
        _: State,
        _: RunContext,
    ) -> Result<()> {
        self.seen.lock().unwrap().push(dependency);
        Ok(())
    }
}

#[async_trait]
impl Finalizable<State> for Consumer {
    async fn finalize(
        &self,
        _: &State,
    ) -> Result<()> {
        self.finalized.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
}

#[tokio::test]
async fn dormant_dependency_is_required_only_at_activation() {
    let registry = Registry::new();
    let consumer = Arc::new(Consumer::new(0));
    registry.register(consumer.clone()).unwrap();
    let mut runtime = Runtime::default();
    assert_eq!(runtime.spawn_all(&registry, State::default()).unwrap(), 1);
    assert!(runtime.start_generation("consumer", false).is_err());

    consumer.policy.store(1, Ordering::SeqCst);
    let error = runtime.start_generation("consumer", false).unwrap_err();
    assert!(
        matches!(error, Error::Dependency { name, .. } if name.as_ref() == "consumer")
    );
    let snapshot = runtime.snapshots().pop().unwrap();
    assert_eq!(snapshot.generation(), 0);
    assert_eq!(snapshot.status(), ServiceStatus::Stopped);

    let dependency = Arc::new(Dependency);
    registry.insert(dependency.clone()).unwrap();
    for generation in 1..=2 {
        let snapshot = runtime.start_generation("consumer", false).unwrap();
        assert_eq!(snapshot.generation(), generation);
        let joined = runtime.join_set.join_next_with_id().await.unwrap();
        runtime.complete_join(joined).unwrap();
    }
    let seen = consumer.seen.lock().unwrap();
    assert_eq!(seen.len(), 2);
    assert!(seen.iter().all(|value| Arc::ptr_eq(value, &dependency)));
}

#[cfg(feature = "fleet")]
#[tokio::test]
async fn excluded_worker_does_not_require_worker_dependencies() {
    let registry = Registry::new();
    registry.register(Arc::new(Consumer::new(2))).unwrap();
    let mut runtime = Runtime::default();
    let state =
        State { scope: crate::fleet::Scope::Master, ..State::default() };
    assert_eq!(runtime.spawn_all(&registry, state).unwrap(), 0);
    assert!(runtime.join_set.is_empty());
}

struct First(AtomicUsize);

impl Component<State> for First {
    type Deps = ();
    fn name(&self) -> &str {
        "first"
    }
    fn run_priority(&self) -> Option<u8> {
        Some(crate::registry::priority::EARLY)
    }
    fn as_runnable(self: Arc<Self>) -> Option<Arc<dyn Runnable<State>>> {
        Some(self)
    }
}

#[async_trait]
impl Runnable<State> for First {
    async fn run(
        self: Arc<Self>,
        _: (),
        _: State,
        _: RunContext,
    ) -> Result<()> {
        self.0.fetch_add(1, Ordering::SeqCst);
        Ok(())
    }
}

#[tokio::test]
async fn initial_dependency_failure_starts_no_partial_generation() {
    let registry = Registry::new();
    let first = Arc::new(First(AtomicUsize::new(0)));
    registry.register(first.clone()).unwrap();
    registry.register(Arc::new(Consumer::new(2))).unwrap();
    let mut runtime = Runtime::default();
    assert!(matches!(
        runtime.spawn_all(&registry, State::default()),
        Err(Error::Dependency { .. })
    ));
    assert!(runtime.join_set.is_empty());
    assert!(runtime.services.is_empty());
    assert_eq!(first.0.load(Ordering::SeqCst), 0);
    registry.insert(Arc::new(Dependency)).unwrap();
    assert_eq!(runtime.spawn_all(&registry, State::default()).unwrap(), 2);
    runtime.drain().await.unwrap();
}

#[tokio::test]
async fn initial_activation_uses_the_admission_policy_snapshot() {
    let registry = Registry::new();
    let consumer = Arc::new(Consumer::new(3));
    registry.register(consumer.clone()).unwrap();
    registry.insert(Arc::new(Dependency)).unwrap();
    let mut runtime = Runtime::default();
    assert_eq!(runtime.spawn_all(&registry, State::default()).unwrap(), 1);
    runtime.drain().await.unwrap();
    assert_eq!(consumer.seen.lock().unwrap().len(), 1);
    // Observation and subsequent explicit activation still see live policy.
    assert!(runtime.start_generation("consumer", false).is_err());
}

#[cfg(feature = "fleet")]
#[tokio::test]
async fn initial_fleet_admission_validates_once_and_restart_revalidates() {
    use crate::fleet::{
        Dispatch, DispatchRequest, Dispatched, Dispatcher, Scope, Scoped,
        ServiceExecution,
    };

    struct FleetWork(AtomicUsize);
    impl Component<State> for FleetWork {
        type Deps = ();
        fn name(&self) -> &str {
            "fleet-work"
        }
        fn as_runnable(self: Arc<Self>) -> Option<Arc<dyn Runnable<State>>> {
            Some(self)
        }
        fn as_dispatched(&self) -> Option<&dyn Dispatched> {
            Some(self)
        }
    }
    impl Scoped for FleetWork {
        fn scope(&self) -> Scope {
            Scope::Worker
        }
    }
    impl Dispatched for FleetWork {
        fn dispatch(&self) -> Dispatch {
            Dispatch::One
        }
    }
    #[async_trait]
    impl Runnable<State> for FleetWork {
        async fn run(
            self: Arc<Self>,
            _: (),
            _: State,
            _: RunContext,
        ) -> Result<()> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
    }
    struct Adapter(AtomicUsize);
    #[async_trait]
    impl Dispatcher<State> for Adapter {
        fn validate(
            &self,
            request: &DispatchRequest,
            _: &State,
        ) -> Result<()> {
            // This adapter owns policy itself; the optional hook is not a
            // new requirement on every existing Dispatched implementation.
            assert!(request.policy.is_none());
            if self.0.fetch_add(1, Ordering::SeqCst) == 0 {
                Ok(())
            } else {
                Err(Error::msg("adapter policy changed"))
            }
        }
        async fn run(
            &self,
            _: DispatchRequest,
            execution: ServiceExecution<State>,
            context: RunContext,
        ) -> Result<()> {
            execution.run(context).await
        }
    }
    let registry = Registry::new();
    let work = Arc::new(FleetWork(AtomicUsize::new(0)));
    registry.register(work.clone()).unwrap();
    let adapter = Arc::new(Adapter(AtomicUsize::new(0)));
    let mut runtime =
        Runtime::default().with_dispatcher(adapter.clone()).unwrap();
    runtime.spawn_all(&registry, State::default()).unwrap();
    let joined = runtime.join_set.join_next_with_id().await.unwrap();
    runtime.complete_join(joined).unwrap();
    assert_eq!(adapter.0.load(Ordering::SeqCst), 1);
    assert_eq!(work.0.load(Ordering::SeqCst), 1);
    assert!(runtime.start_generation("fleet-work", false).is_err());
    assert_eq!(adapter.0.load(Ordering::SeqCst), 2);
    assert_eq!(runtime.snapshots()[0].generation(), 1);
}

/// One component's instances run under their own placements, and one that
/// declares none takes its component's.
#[cfg(feature = "fleet")]
#[tokio::test]
async fn an_instance_runs_under_its_own_placement() {
    use crate::RunnableInstance;
    use crate::fleet::{
        Dispatch, DispatchPolicy, DispatchRequest, Dispatcher,
        InstancePlacement, Scope, ServiceExecution,
    };
    use std::sync::Mutex;

    struct Once(AtomicUsize);
    #[async_trait]
    impl Runnable<State> for Once {
        async fn run(
            self: Arc<Self>,
            _: (),
            _: State,
            _: RunContext,
        ) -> Result<()> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
    }
    struct Factory(Arc<Once>);
    impl Component<State> for Factory {
        type Deps = ();
        fn name(&self) -> &str {
            "factory"
        }
        fn runnables(self: Arc<Self>) -> Vec<RunnableInstance<State>> {
            let policy = DispatchPolicy {
                capacity: std::num::NonZeroU16::MIN,
                lease_ttl: std::time::Duration::from_secs(20),
                renewal_interval: std::time::Duration::from_secs(5),
            };
            let once: Arc<dyn Runnable<State>> = self.0.clone();
            vec![
                RunnableInstance::new("plain", once.clone()),
                RunnableInstance::new("leased", once.clone()).placed(
                    InstancePlacement {
                        scope: Scope::Worker,
                        dispatch: Dispatch::One,
                        policy: Some(policy),
                    },
                ),
                RunnableInstance::new("elsewhere", once).placed(
                    InstancePlacement {
                        scope: Scope::Master,
                        dispatch: Dispatch::Each,
                        policy: None,
                    },
                ),
            ]
        }
    }
    #[derive(Default)]
    struct Adapter(Mutex<Vec<DispatchRequest>>);
    #[async_trait]
    impl Dispatcher<State> for Adapter {
        fn validate(
            &self,
            _: &DispatchRequest,
            _: &State,
        ) -> Result<()> {
            Ok(())
        }
        async fn run(
            &self,
            request: DispatchRequest,
            execution: ServiceExecution<State>,
            context: RunContext,
        ) -> Result<()> {
            self.0.lock().unwrap().push(request);
            execution.run(context).await
        }
    }

    let registry = Registry::new();
    let once = Arc::new(Once(AtomicUsize::new(0)));
    registry.register(Arc::new(Factory(once.clone()))).unwrap();
    let adapter = Arc::new(Adapter::default());
    let mut runtime =
        Runtime::default().with_dispatcher(adapter.clone()).unwrap();
    let state = State { scope: Scope::Worker, ..State::default() };
    runtime.spawn_all(&registry, state).unwrap();
    for _ in 0..2 {
        let joined = runtime.join_set.join_next_with_id().await.unwrap();
        runtime.complete_join(joined).unwrap();
    }

    let mut services: Vec<_> = runtime
        .snapshots()
        .iter()
        .map(|service| service.name().to_owned())
        .collect();
    services.sort();
    assert_eq!(services, ["leased", "plain"]);
    assert_eq!(once.0.load(Ordering::SeqCst), 2);
    let requests = adapter.0.lock().unwrap();
    assert_eq!(requests.len(), 1, "only the leased instance is dispatched");
    assert_eq!(&*requests[0].name, "leased");
    assert_eq!(requests[0].dispatch, Dispatch::One);
    assert!(requests[0].policy.is_some());
}

#[cfg(feature = "registry")]
#[tokio::test]
async fn prepared_resources_finalize_after_admission_failure() {
    use crate::HasRegistry;
    #[derive(Clone)]
    struct RegisteredState(Arc<Registry<RegisteredState>>, State);
    impl SharedState for RegisteredState {
        fn shutdown_token(&self) -> CancellationToken {
            self.1.shutdown_token()
        }
    }
    impl HasRegistry for RegisteredState {
        fn registry_ref(&self) -> &Registry<Self> {
            &self.0
        }
    }
    let state = RegisteredState(Arc::new(Registry::new()), State::default());
    // This entry uses the high-level state only to exercise run_prepared's
    // finalization boundary. No process, listener or filesystem is started.
    struct Prepared(AtomicUsize);
    impl Component<RegisteredState> for Prepared {
        type Deps = (Arc<Dependency>,);
        fn name(&self) -> &str {
            "prepared"
        }
        fn as_runnable(
            self: Arc<Self>
        ) -> Option<Arc<dyn Runnable<RegisteredState, Self::Deps>>> {
            Some(self)
        }
        fn as_finalizable(&self) -> Option<&dyn Finalizable<RegisteredState>> {
            Some(self)
        }
    }
    #[async_trait]
    impl Runnable<RegisteredState, (Arc<Dependency>,)> for Prepared {
        async fn run(
            self: Arc<Self>,
            _: (Arc<Dependency>,),
            _: RegisteredState,
            _: RunContext,
        ) -> Result<()> {
            unreachable!("missing dependencies reject admission")
        }
    }
    #[async_trait]
    impl Finalizable<RegisteredState> for Prepared {
        async fn finalize(
            &self,
            _: &RegisteredState,
        ) -> Result<()> {
            self.0.fetch_add(1, Ordering::SeqCst);
            Ok(())
        }
    }
    let prepared = Arc::new(Prepared(AtomicUsize::new(0)));
    state.registry_ref().register(prepared.clone()).unwrap();
    let error = Runtime::default().run_prepared(state).await.unwrap_err();
    assert!(matches!(error, Error::Dependency { .. }));
    assert_eq!(prepared.0.load(Ordering::SeqCst), 1);
}