foundation_core 0.0.4

#![allow(clippy::type_complexity)]

mod pool_guard;

pub use pool_guard::PoolGuard;

use std::cell::OnceCell;
use std::sync::Arc;

use crate::valtron::{
    executors::constants::{
        BACK_OFF_JITER, BACK_OFF_MAX_DURATION, BACK_OFF_MIN_DURATION, BACK_OFF_THREAD_FACTOR,
        DEFAULT_YIELD_WAIT_TIME, MAX_ROUNDS_IDLE_COUNT, MAX_ROUNDS_WHEN_SLEEPING_ENDS,
    },
    ExecutionAction, LocalThreadExecutor, PriorityOrder, ProcessController, ProgressIndicator,
    SharedTaskQueue, TaskIterator, TaskReadyResolver, TaskSpawnConfig,
};

use crate::{
    retries::ExponentialBackoffDecider,
    synca::{IdleMan, SleepyMan},
};
use concurrent_queue::ConcurrentQueue;
use foundation_nostd::primitives::SpinWaiter;

/// NoThreadController provides a single-threaded ProcessController
/// that uses iteration-based spinning for environments without OS sleep support (WASM).
///
/// Uses SpinWaiter from foundation_nostd which provides:
/// - Iteration-based timing (not Duration::ZERO which would spin forever)
/// - Interruptible via AtomicBool flag
/// - Bounded spinning (never spins forever)
#[derive(Clone)]
pub struct NoThreadController {
    waiter: SpinWaiter,
}

impl NoThreadController {
    /// Creates a new NoThreadController with platform-appropriate settings.
    #[must_use]
    pub fn new() -> Self {
        Self {
            #[cfg(target_family = "wasm")]
            waiter: SpinWaiter::wasm(),
            #[cfg(not(target_family = "wasm"))]
            waiter: SpinWaiter::embedded(),
        }
    }
}

impl Default for NoThreadController {
    fn default() -> Self {
        Self::new()
    }
}

impl ProcessController for NoThreadController {
    type YieldSignal = ();

    fn yield_for(&self, dur: std::time::Duration) {
        // Use SpinWaiter instead of doing nothing
        // This provides actual delay via iteration counting
        // Works in WASM, no_std, and single-threaded environments
        self.waiter.wait(dur);
    }
}

// Default controller: JSThreadYielder on wasm32 (JS event loop aware),
// NoThreadController on native (iteration-based spinning). The JS yielder is
// wasm-only, so the selection is gated on the target too — a native build with a
// `js-*` feature still uses NoThreadController.
#[cfg(all(
    target_family = "wasm",
    any(feature = "js-wasmbindgen", feature = "js-foundation-wasm")
))]
use crate::valtron::wasm::JSThreadYielder as DefaultController;

// NoThreadController is defined in this same file below — no import needed for native.
#[cfg(not(all(
    target_family = "wasm",
    any(feature = "js-wasmbindgen", feature = "js-foundation-wasm")
)))]
type DefaultController = NoThreadController;

thread_local! {
    #[allow(clippy::missing_const_for_thread_local)]
    static GLOBAL_LOCAL_EXECUTOR_ENGINE: OnceCell<LocalThreadExecutor<DefaultController>> = OnceCell::new();
}

/// Optional singleton guard. Module-level so we can reset in tests.
static VALTRON_SINGLETON: std::sync::Mutex<Option<PoolGuard>> = std::sync::Mutex::new(None);

/// Optional convenience wrapper for initializing the single-threaded valtron executor.
///
/// Provides "init once, get guard" semantics similar to `CfHttpAppSingleton`.
/// Only available on wasm32/wasm64 targets (where single-threaded mode is the norm).
///
/// The existing free functions (`initialize_pool`, `spawn`, `run_until_complete`, etc.)
/// continue to work independently — use this when you want explicit guard ownership.
pub struct SingleExecutorSingleton;

impl SingleExecutorSingleton {
    /// Initialize the executor and return a PoolGuard.
    ///
    /// On first call: calls `initialize_pool(seed)`, stores a PoolGuard,
    /// runs the setup closure. On subsequent calls: returns a cloned guard
    /// (PoolGuard is zero-sized in single mode).
    pub fn get_or_init<F: FnOnce(&PoolGuard)>(seed: u64, setup: F) -> PoolGuard {
        initialize_pool(seed);
        let mut guard = VALTRON_SINGLETON.lock().unwrap();
        if guard.is_none() {
            *guard = Some(PoolGuard);
        }
        setup(guard.as_ref().unwrap());
        guard.clone().unwrap()
    }

    /// Get a PoolGuard. Panics if `get_or_init` has not been called.
    pub fn guard() -> PoolGuard {
        VALTRON_SINGLETON
            .lock()
            .unwrap()
            .clone()
            .expect("Valtron not initialized — call SingleExecutorSingleton::get_or_init() first")
    }

    /// Check if the singleton has been initialized.
    pub fn is_initialized() -> bool {
        VALTRON_SINGLETON.lock().unwrap().is_some()
    }

    /// Force-reset for testing. cfg-gated to `#[cfg(test)]`.
    #[cfg(test)]
    pub fn reset() {
        *VALTRON_SINGLETON.lock().unwrap() = None;
    }
}

/// `initialize` initializes the local single-threaded
/// execution engine, and is required to call this as your
/// first call when using this in WebAssembly or `SingleThreaded`
/// environment.
pub fn initialize_pool(seed_for_rng: u64) {
    GLOBAL_LOCAL_EXECUTOR_ENGINE.with(|pool| {
        let _ = pool.get_or_init(|| {
            // `SharedTaskQueue` is `Arc<…>` in BOTH cfgs so one alias serves the
            // multi-threaded (Send) and single-threaded (non-Send) executors; in
            // this thread-local engine the Arc never crosses a thread, so the
            // non-Send contents are fine.
            #[allow(clippy::arc_with_non_send_sync)]
            let tasks: SharedTaskQueue = Arc::new(ConcurrentQueue::unbounded());
            LocalThreadExecutor::from_seed(
                seed_for_rng,
                "single-thread-pool".to_string(),
                tasks,
                IdleMan::new(
                    MAX_ROUNDS_IDLE_COUNT,
                    None,
                    SleepyMan::new(
                        MAX_ROUNDS_WHEN_SLEEPING_ENDS,
                        ExponentialBackoffDecider::new(
                            BACK_OFF_THREAD_FACTOR,
                            BACK_OFF_JITER,
                            BACK_OFF_MIN_DURATION,
                            Some(BACK_OFF_MAX_DURATION),
                        ),
                    ),
                ),
                PriorityOrder::Top,
                DefaultController::new(),
                DEFAULT_YIELD_WAIT_TIME,
                None,
                None,
            )
        });
    });
}

/// `run_until` calls the `LocalExecution` queue and processes
/// the next pending message by moving it forward just until
/// the provided function returns true indicating it should stop.
///
/// This lets you review the state or some internal conditions
/// upon which you wish to stop processing.
///
/// It works like [`run_once`] but not just executed once but till
/// the condition is met.
///
/// # Panics
///
/// 1. Will panic if no pool exists.
///
/// # Returns
///
/// The progress indicator from the execution
pub fn run_until<S>(checker: S)
where
    S: Fn(ProgressIndicator) -> bool,
{
    GLOBAL_LOCAL_EXECUTOR_ENGINE.with(|pool| match pool.get() {
        Some(pool) => pool.run_until(checker),
        None => panic!("Thread pool not initialized, ensure to call initialize() first"),
    });
}

/// `run_once` calls the `LocalExecution` queue and processes
/// the next pending message by moving it forward just once.
///
/// I rearly see you using this, but there might be situations
/// where more fine-grained control on how much work you wish
/// to do matters and you do not want a [`run_until_complete`]
/// where the underlying point of stopping is not controlled
/// by you.
///
/// # Panics
///
/// 1. Will panic if no pool exists.
///
/// # Returns
///
/// The progress indicator from the execution
#[must_use]
pub fn run_once() -> ProgressIndicator {
    GLOBAL_LOCAL_EXECUTOR_ENGINE.with(|pool| match pool.get() {
        Some(pool) => pool.run_once(),
        None => panic!("Thread pool not initialized, ensure to call initialize() first"),
    })
}

/// `run_until_complete` calls the `LocalExecution` queue and processes
/// all pending messages till they are completed.
///
/// # Panics
///
/// 1. Will panic if no pool exists.
///
pub fn run_until_complete() {
    GLOBAL_LOCAL_EXECUTOR_ENGINE.with(|pool| match pool.get() {
        Some(pool) => pool.block_until_finished(),
        None => panic!("Thread pool not initialized, ensure to call initialize() first"),
    });
}

/// `spawn` provides a builder which specifically allows you to build out
/// the underlying tasks to be scheduled into the global queue.
///
/// It expects you infer the type of `Task` and `Action` from the
/// type implementing `TaskIterator`.
///
/// # Panics
///
/// 1. Will panic if no pool exists.
///
/// # Returns
///
/// A builder for creating a task iterator
#[must_use]
pub fn spawn<Task, Action>() -> TaskSpawnConfig<
    Task::Ready,
    Task::Pending,
    Task::Spawner,
    Box<dyn TaskReadyResolver<Task::Spawner, Task::Ready, Task::Pending> + 'static>,
    Task,
>
where
    Task::Ready: 'static,
    Task::Pending: 'static,
    Task: TaskIterator<Spawner = Action> + 'static,
    Action: ExecutionAction + 'static,
{
    GLOBAL_LOCAL_EXECUTOR_ENGINE.with(|pool| match pool.get() {
        Some(pool) => TaskSpawnConfig::new(pool.boxed_engine()),
        None => panic!("Thread pool not initialized, ensure to call initialize() first"),
    })
}

/// `spawn2` provides a builder which specifically allows you to build out
/// the underlying tasks to be scheduled into the global queue.
///
/// It expects you to provide the `Resolver` type explicitly.
///
/// # Panics
///
/// 1. Will panic if no pool exists.
///
/// # Returns
///
/// A builder for creating a task iterator
#[must_use]
pub fn spawn2<Task, Action, Resolver>(
) -> TaskSpawnConfig<Task::Ready, Task::Pending, Action, Resolver, Task>
where
    Task::Ready: 'static,
    Task::Pending: 'static,
    Task: TaskIterator<Spawner = Action> + 'static,
    Action: ExecutionAction + 'static,
    Resolver: TaskReadyResolver<Action, Task::Ready, Task::Pending> + 'static,
{
    GLOBAL_LOCAL_EXECUTOR_ENGINE.with(|pool| match pool.get() {
        Some(pool) => TaskSpawnConfig::new(pool.boxed_engine()),
        None => panic!("Thread pool not initialized, ensure to call initialize() first"),
    })
}

/// Execute a blocking closure inline on the current thread.
///
/// WHY: Single-threaded environments have no background threads — run immediately
/// WHAT: Executes the closure with `catch_unwind` for consistency with multi behavior
/// HOW: Wraps in `AssertUnwindSafe` and calls `catch_unwind`
///
/// # Errors
///
/// Returns an error if the closure panics.
pub fn run_background_job(job: impl FnOnce() + 'static) -> crate::valtron::GenericResult<()> {
    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(job));
    match result {
        Ok(()) => Ok(()),
        Err(panic_info) => {
            tracing::error!("Background job panicked (single-threaded): {panic_info:?}");
            Err("Background job panicked".into())
        }
    }
}

#[cfg(all(test, not(target_family = "wasm")))]
mod single_threaded_tests {
    use std::{cell::RefCell, rc::Rc};

    use foundation_compact::rng::Rng;
    use tracing_test::traced_test;

    use crate::valtron::{
        single::{initialize_pool, run_until, run_until_complete, spawn},
        FnReady, NoSpawner, Stream, TaskIterator, TaskStatus,
    };

    struct Counter(usize, Rc<RefCell<Vec<usize>>>);

    impl Counter {
        pub fn new(val: usize, list: Rc<RefCell<Vec<usize>>>) -> Self {
            Self(val, list)
        }
    }

    impl TaskIterator for Counter {
        type Pending = ();

        type Ready = usize;

        type Spawner = NoSpawner;

        fn next_status(
            &mut self,
        ) -> Option<crate::valtron::TaskStatus<Self::Ready, Self::Pending, Self::Spawner>> {
            let item_size = self.1.borrow().len();

            if item_size == self.0 {
                return None;
            }

            self.1.borrow_mut().push(item_size);

            Some(crate::valtron::TaskStatus::Ready(self.1.borrow().len()))
        }
    }

    #[test]
    #[traced_test]
    fn can_queue_task_only() {
        let seed = foundation_compact::rng::rng().next_u64();

        let shared_list = Rc::new(RefCell::new(Vec::new()));
        let counter = Counter::new(5, shared_list.clone());

        initialize_pool(seed);

        spawn()
            .with_task(counter)
            .with_resolver(Box::new(FnReady::new(|item, _| {
                tracing::info!("Received next: {:?}", item)
            })))
            .as_scheduled()
            .spawn()
            .expect("should deliver task");

        assert_eq!(shared_list.borrow().len(), 0);
    }

    #[test]
    #[traced_test]
    fn can_queue_and_complete_task_with_run_until() {
        let seed = foundation_compact::rng::rng().next_u64();

        let shared_list = Rc::new(RefCell::new(Vec::new()));
        let counter = Counter::new(5, shared_list.clone());

        initialize_pool(seed);

        spawn()
            .with_task(counter)
            .with_resolver(Box::new(FnReady::new(|item, _| {
                tracing::info!("Received next: {:?}", item);
            })))
            .as_scheduled()
            .spawn()
            .expect("should deliver task");

        let handle = shared_list.clone();
        run_until(|_| {
            if handle.borrow_mut().len() >= 5 {
                return true;
            }
            false
        });

        assert_eq!(shared_list.borrow().clone(), vec![0, 1, 2, 3, 4]);
    }

    #[test]
    #[traced_test]
    fn can_queue_and_complete_task() {
        let seed = foundation_compact::rng::rng().next_u64();

        let shared_list = Rc::new(RefCell::new(Vec::new()));
        let counter = Counter::new(5, shared_list.clone());

        initialize_pool(seed);

        spawn()
            .with_task(counter)
            .with_resolver(Box::new(FnReady::new(|item, _| {
                tracing::info!("Received next: {:?}", item)
            })))
            .as_scheduled()
            .spawn()
            .expect("should deliver task");

        run_until_complete();

        assert_eq!(shared_list.borrow().clone(), vec![0, 1, 2, 3, 4]);
    }

    #[test]
    #[traced_test]
    fn can_queue_and_complete_task_with_iterator() {
        let seed = foundation_compact::rng::rng().next_u64();

        let shared_list = Rc::new(RefCell::new(Vec::new()));
        let counter = Counter::new(5, shared_list.clone());

        initialize_pool(seed);

        let iter = spawn()
            .with_task(counter)
            .as_scheduled()
            .recv(std::time::Duration::from_nanos(50))
            .expect("should deliver task");

        run_until_complete();

        let complete: Vec<usize> = iter
            .filter_map(|item| match item {
                crate::valtron::executors::local::NotificationItem::Ready(TaskStatus::Ready(
                    value,
                )) => Some(value),
                _ => None,
            })
            .collect();

        assert_eq!(complete, vec![1, 2, 3, 4, 5]);
        assert_eq!(shared_list.borrow().clone(), vec![0, 1, 2, 3, 4]);
    }

    #[test]
    #[traced_test]
    fn can_queue_and_complete_stream_with_iterator() {
        let seed = foundation_compact::rng::rng().next_u64();

        let shared_list = Rc::new(RefCell::new(Vec::new()));
        let counter = Counter::new(5, shared_list.clone());

        initialize_pool(seed);

        let iter = spawn()
            .with_task(counter)
            .as_scheduled()
            .stream(std::time::Duration::from_nanos(50))
            .expect("should deliver task");

        run_until_complete();

        let complete: Vec<usize> = iter
            .map(|item| match item {
                Stream::Next(value) => Some(value),
                _ => None,
            })
            .take_while(|t| t.is_some())
            .map(|t| t.unwrap())
            .collect();

        assert_eq!(complete, vec![1, 2, 3, 4, 5]);
        assert_eq!(shared_list.borrow().clone(), vec![0, 1, 2, 3, 4]);
    }
}