cordis-core 0.0.4

use std::{
    future::Future,
    panic::AssertUnwindSafe,
    pin::Pin,
    sync::{Arc, Mutex},
};

use futures::FutureExt;
use tokio::sync::watch;

use crate::{Error, Result};

pub(crate) type EffectFuture = Pin<Box<dyn Future<Output = Result<()>> + Send + 'static>>;
pub(crate) type Effect = Box<dyn FnOnce() -> EffectFuture + Send + 'static>;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Lifecycle {
    Staging,
    Active,
    Disposing,
    Disposed,
}

struct ScopeState {
    lifecycle: Lifecycle,
    starts: Vec<Effect>,
    cleanups: Vec<Effect>,
    outcome: Option<Arc<str>>,
}

pub(crate) struct ScopeInner {
    pub id: u64,
    pub name: String,
    state: Mutex<ScopeState>,
    lifecycle: watch::Sender<Lifecycle>,
}

impl ScopeInner {
    pub fn new(id: u64, name: impl Into<String>) -> Arc<Self> {
        let (lifecycle, _) = watch::channel(Lifecycle::Staging);
        Arc::new(Self {
            id,
            name: name.into(),
            state: Mutex::new(ScopeState {
                lifecycle: Lifecycle::Staging,
                starts: Vec::new(),
                cleanups: Vec::new(),
                outcome: None,
            }),
            lifecycle,
        })
    }

    pub fn push(&self, cleanup: Effect) -> Result<()> {
        let mut state = self.state.lock().expect("scope lock poisoned");
        if matches!(state.lifecycle, Lifecycle::Disposing | Lifecycle::Disposed) {
            return Err(Error::ScopeInactive {
                name: self.name.clone(),
            });
        }
        state.cleanups.push(cleanup);
        Ok(())
    }

    /// Registers work that starts only after plugin apply succeeds.
    pub fn on_commit(&self, start: Effect) -> Result<()> {
        let mut state = self.state.lock().expect("scope lock poisoned");
        if state.lifecycle != Lifecycle::Staging {
            return Err(Error::ScopeInactive {
                name: self.name.clone(),
            });
        }
        state.starts.push(start);
        Ok(())
    }

    pub async fn commit(&self) -> Result<()> {
        let starts = {
            let mut state = self.state.lock().expect("scope lock poisoned");
            if state.lifecycle != Lifecycle::Staging {
                return Err(Error::ScopeInactive {
                    name: self.name.clone(),
                });
            }
            std::mem::take(&mut state.starts)
        };

        for start in starts {
            match AssertUnwindSafe(start()).catch_unwind().await {
                Ok(result) => result?,
                Err(payload) => return Err(Error::panic(payload)),
            }
        }

        let mut state = self.state.lock().expect("scope lock poisoned");
        if state.lifecycle != Lifecycle::Staging {
            return Err(Error::ScopeInactive {
                name: self.name.clone(),
            });
        }
        state.lifecycle = Lifecycle::Active;
        self.lifecycle.send_replace(Lifecycle::Active);
        Ok(())
    }

    /// Idempotent and safe for concurrent callers. Exactly one caller executes
    /// cleanup; other callers wait for the same completion.
    pub async fn dispose(&self) -> Result<()> {
        let mut lifecycle = self.lifecycle.subscribe();
        let cleanups = loop {
            let decision = {
                let mut state = self.state.lock().expect("scope lock poisoned");
                match state.lifecycle {
                    Lifecycle::Disposed => {
                        return state
                            .outcome
                            .as_ref()
                            .map_or(Ok(()), |message| Err(Error::cleanup(message.as_ref())));
                    }
                    Lifecycle::Disposing => None,
                    Lifecycle::Staging | Lifecycle::Active => {
                        state.lifecycle = Lifecycle::Disposing;
                        self.lifecycle.send_replace(Lifecycle::Disposing);
                        Some(std::mem::take(&mut state.cleanups))
                    }
                }
            };
            if let Some(cleanups) = decision {
                break cleanups;
            }
            lifecycle
                .changed()
                .await
                .map_err(|_| Error::ScopeInactive {
                    name: self.name.clone(),
                })?;
        };

        let mut first_error = None;
        for cleanup in cleanups.into_iter().rev() {
            let result = AssertUnwindSafe(cleanup()).catch_unwind().await;
            let error = match result {
                Ok(Ok(())) => None,
                Ok(Err(error)) => Some(error),
                Err(payload) => Some(Error::panic(payload)),
            };
            if let Some(error) = error {
                first_error.get_or_insert(error);
            }
        }

        let outcome = first_error
            .as_ref()
            .map(|error| Arc::<str>::from(error.to_string()));
        {
            let mut state = self.state.lock().expect("scope lock poisoned");
            state.outcome = outcome;
            state.lifecycle = Lifecycle::Disposed;
        }
        self.lifecycle.send_replace(Lifecycle::Disposed);
        first_error.map_or(Ok(()), Err)
    }
}