//! Minimal scheduler abstraction (Phase 10.6b prep)
//!
//! Provides a pluggable interface to run tasks and yield cooperatively.

use crate::runtime::get_global_ring0;

pub trait Scheduler: Send + Sync {
    /// Spawn a task/closure. Default impl may run inline.
    fn spawn(&self, _name: &str, f: Box<dyn FnOnce() + Send + 'static>);
    /// Spawn a task after given delay milliseconds.
    fn spawn_after(&self, _delay_ms: u64, _name: &str, _f: Box<dyn FnOnce() + Send + 'static>) {}
    /// Poll scheduler: run due tasks and a limited number of queued tasks.
    fn poll(&self) {}
    /// Cooperative yield point (no-op for single-thread).
    fn yield_now(&self) {}

    /// Optional: spawn with a cancellation token. Default delegates to spawn.
    fn spawn_with_token(
        &self,
        name: &str,
        _token: CancellationToken,
        f: Box<dyn FnOnce() + Send + 'static>,
    ) {
        self.spawn(name, f)
    }
}

use std::collections::VecDeque;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};

/// Single-thread scheduler with a simple queue and delayed tasks.
pub struct SingleThreadScheduler {
    queue: Arc<Mutex<VecDeque<Box<dyn FnOnce() + Send + 'static>>>>,
    delayed: Arc<Mutex<Vec<(Instant, Box<dyn FnOnce() + Send + 'static>)>>>,
}

impl SingleThreadScheduler {
    pub fn new() -> Self {
        Self {
            queue: Arc::new(Mutex::new(VecDeque::new())),
            delayed: Arc::new(Mutex::new(Vec::new())),
        }
    }
}

impl Scheduler for SingleThreadScheduler {
    fn spawn(&self, _name: &str, f: Box<dyn FnOnce() + Send + 'static>) {
        if let Ok(mut q) = self.queue.lock() {
            q.push_back(f);
        }
    }
    fn spawn_after(&self, delay_ms: u64, _name: &str, f: Box<dyn FnOnce() + Send + 'static>) {
        let when = Instant::now() + Duration::from_millis(delay_ms);
        if let Ok(mut d) = self.delayed.lock() {
            d.push((when, f));
        }
    }
    fn poll(&self) {
        // Move due delayed tasks to queue
        let trace = std::env::var("NYASH_SCHED_TRACE").ok().as_deref() == Some("1");
        let now = Instant::now();
        let mut moved = 0usize;
        if let Ok(mut d) = self.delayed.lock() {
            let mut i = 0;
            while i < d.len() {
                if d[i].0 <= now {
                    let (_when, task) = d.remove(i);
                    if let Ok(mut q) = self.queue.lock() {
                        q.push_back(task);
                    }
                    moved += 1;
                } else {
                    i += 1;
                }
            }
        }
        // Run up to budget queued tasks
        let budget: usize = std::env::var("NYASH_SCHED_POLL_BUDGET")
            .ok()
            .and_then(|s| s.parse().ok())
            .filter(|&n: &usize| n > 0)
            .unwrap_or(1);
        let mut ran = 0usize;
        while ran < budget {
            let task_opt = {
                if let Ok(mut q) = self.queue.lock() {
                    q.pop_front()
                } else {
                    None
                }
            };
            if let Some(task) = task_opt {
                task();
                ran += 1;
            } else {
                break;
            }
        }
        if trace {
            get_global_ring0().log.debug(&format!(
                "[SCHED] poll moved={} ran={} budget={}",
                moved, ran, budget
            ));
        }
    }
}

use std::sync::atomic::{AtomicBool, Ordering};

/// Simple idempotent cancellation token for structured concurrency (skeleton)
#[derive(Clone, Debug)]
pub struct CancellationToken(Arc<AtomicBool>);

impl CancellationToken {
    pub fn new() -> Self {
        Self(Arc::new(AtomicBool::new(false)))
    }
    pub fn cancel(&self) {
        self.0.store(true, Ordering::SeqCst);
    }
    pub fn is_cancelled(&self) -> bool {
        self.0.load(Ordering::SeqCst)
    }
}
