acts 0.25.0

use super::{ActTask, Runtime};
use crate::snapshot::{SnapshotPolicy, SnapshotStore, join_scope, resolve_scope_params};
use crate::{
    Act, ActError, Executor, Message, MessageState, MissingParamAction, NodeKind, Result,
    TaskState, Vars,
    event::Action,
    scheduler::{
        Node, Process, Task,
        tree::{NodeContent, dyn_build_act},
    },
    utils::{self, consts, shortid},
};
use parking_lot::RwLock;
use serde::{Deserialize, Serialize, de::DeserializeOwned};
use std::sync::Arc;
use tokio_util::sync::CancellationToken;
use tracing::{debug, instrument};

tokio::task_local! {
    static CONTEXT: Context;
}

pub struct Context {
    pub runtime: Arc<Runtime>,
    pub executor: Arc<Executor>,
    pub proc: Arc<Process>,
    task: RwLock<Arc<Task>>,
    action: RwLock<Option<Action>>,
    vars: RwLock<Vars>,
}

impl Clone for Context {
    fn clone(&self) -> Self {
        Context {
            runtime: self.runtime.clone(),
            executor: self.executor.clone(),
            proc: self.proc.clone(),
            task: RwLock::new(self.task.read().clone()),
            action: RwLock::new(self.action.read().clone()),
            vars: RwLock::new(self.vars.read().clone()),
        }
    }
}

impl std::fmt::Debug for Context {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("Context")
            .field("pid", &self.proc.id())
            .field("tid", &self.task().id)
            .field("action", &self.action())
            .finish()
    }
}

impl Context {
    fn init_vars(&self, task: &Arc<Task>) {
        let inputs = task.inputs();
        debug!(inputs = %inputs, "init vars");

        // set the inputs to task's data
        self.task().set_data_with(|data| {
            for (k, v) in inputs.iter() {
                data.set(k, v.clone());
            }
        });
    }

    pub fn new(proc: &Arc<Process>, task: &Arc<Task>) -> Self {
        Context {
            runtime: task.runtime().clone(),
            executor: Arc::new(Executor::engine(task.runtime())),
            proc: proc.clone(),
            action: RwLock::new(None),
            task: RwLock::new(task.clone()),
            vars: RwLock::new(Vars::new()),
        }
    }

    pub fn scope<T, F: Fn() -> T>(ctx: &Context, f: F) -> T {
        // This is only an "is a scheduler context active?" probe; cloning the
        // context here would copy all scoped vars on every nested evaluation.
        if CONTEXT.try_with(|_| ()).is_ok() {
            f()
        } else {
            CONTEXT.sync_scope(ctx.clone(), f)
        }
    }

    /// Access the active scheduler context without cloning it.
    pub fn try_with_current<T, F: FnOnce(&Context) -> T>(f: F) -> Result<T> {
        CONTEXT
            .try_with(f)
            .map_err(|e| ActError::Runtime(e.to_string()))
    }

    pub fn with<T, F: Fn(&Context) -> T>(f: F) -> T {
        CONTEXT.with(|ctx| f(ctx))
    }

    pub fn current() -> Result<Context> {
        CONTEXT
            .try_with(Clone::clone)
            .map_err(|e| ActError::Runtime(e.to_string()))
    }

    pub fn set_task(&self, task: &Arc<Task>) {
        if self.task.read().id != task.id {
            *self.task.write() = task.clone();
        }
    }

    pub fn task(&self) -> Arc<Task> {
        self.task.read().clone()
    }

    /// The directory this process's filesystem access is confined to
    /// (`<acl workdir root>/<pid>`), or `None` when the engine's ACL config
    /// declares no workdir root. Packages that touch the filesystem (shell,
    /// and any custom one) read it here and refuse to leave it; a workflow
    /// script reads the same directory as `$env.WORK_DIR` (see
    /// [`crate::utils::consts::ENV_WORK_DIR`]).
    pub fn workdir(&self) -> Option<std::path::PathBuf> {
        self.proc.workdir()
    }

    /// A token that fires once this act should stop the work it started: the
    /// engine is shutting down, or the task was overridden while it ran (an
    /// `abort`/`cancel`/`skip`/`remove`/`next` action, or an error — see
    /// [`Task::set_state`](crate::Task::set_state)).
    ///
    /// A package that starts something outside the engine — a child process,
    /// an HTTP request, a subscription — selects on it and gives that work up
    /// when it fires, so a cancelled act does not hold its scheduler lane
    /// until its own deadline. It is a "stop now" signal, not a bound on how
    /// long an act may run: that is the act's own timeout.
    ///
    /// ```no_run
    /// # use acts::{ActError, Context, Result};
    /// # async fn wait(ctx: &Context) -> Result<()> {
    /// let cancel = ctx.cancellation_token();
    /// tokio::select! {
    ///     response = some_request() => Ok(response),
    ///     _ = cancel.cancelled() => Err(ActError::Runtime("cancelled".to_string())),
    /// }
    /// # }
    /// # async fn some_request() -> () {}
    /// ```
    pub fn cancellation_token(&self) -> CancellationToken {
        self.task().cancellation_token()
    }

    pub async fn prepare(&self) -> Result<()> {
        self.init_vars(&self.task());
        self.resolve_sealed().await?;
        Ok(())
    }

    pub async fn resolve_sealed(&self) -> Result<()> {
        // Snapshot-backed sealed data — a local cache read only. Feeds
        // (gRPC/NATS/Kafka adapters or embedders calling
        // `engine.snapshot().upsert`) write out-of-band, so this never does
        // network I/O. Missing params / absent data follow the target's
        // `on_missing` action.
        //
        // The process's owner credential decides which scopes it may read: a
        // process started by one caller cannot seal another subject's value,
        // however its vars were set. Enforced here rather than at the start
        // operation because the scope key is only resolvable on the task
        // chain, at prepare time.
        let owner = self.proc.owner_scope();
        let snapshots: Vec<(String, Arc<SnapshotStore>)> = self.runtime.snapshot_registry().list();
        for (name, store) in snapshots {
            let task = self.task();
            let options = &store.options;
            // per-proc: freeze on the first lineage seal; every descendant
            // inherits the pinned value (sealed() walks the parent chain,
            // so a resumed/retried task also keeps its first value)
            if options.policy == SnapshotPolicy::PerProc && task.sealed(&name).is_some() {
                continue;
            }
            // retry determinism: this task row already carries a value
            if task.has_sealed_local(&name) {
                continue;
            }
            let values = match resolve_scope_params(&task, options) {
                Ok(values) => values,
                Err(missing) => match options.on_missing {
                    MissingParamAction::Skip => continue,
                    MissingParamAction::Error => {
                        return Err(ActError::Runtime(format!(
                            "snapshot '{name}' missing required params: {missing:?}"
                        )));
                    }
                },
            };
            let scope = join_scope(&values);
            if !owner.allows(&name, &scope) {
                return Err(ActError::Denied(format!(
                    "snapshot '{name}' scope '{scope}' is not owned by subject '{}'",
                    owner.subject
                )));
            }
            match store.get(&scope) {
                Some(entry) => task.set_sealed(&name, entry.data.clone()),
                None => match options.on_missing {
                    MissingParamAction::Skip => continue,
                    MissingParamAction::Error => {
                        return Err(ActError::Runtime(format!(
                            "snapshot '{name}' has no data for scope '{scope}'"
                        )));
                    }
                },
            }
        }
        Ok(())
    }

    pub fn set_action(&self, action: &Action) -> Result<()> {
        *self.action.write() = Some(action.clone());

        // set the action options to the context
        let mut vars = self.vars.write();
        for (name, v) in action.options.iter() {
            vars.entry(name.to_string())
                .and_modify(|i| *i = v.clone())
                .or_insert(v.clone());
        }

        Ok(())
    }
    pub fn vars(&self) -> Vars {
        self.vars.read().clone()
    }

    pub fn set_env<T>(&self, name: &str, value: T)
    where
        T: Serialize + Clone,
    {
        // The workdir is engine-owned (it is what `$env.WORK_DIR` reads):
        // answering it from a stored value would let a script redefine where
        // its run is. A write is dropped, like the private keys in `$env`.
        if name == consts::ENV_WORK_DIR {
            return;
        }
        // in context, the global env is not writable
        // just set the value to local env of the process
        self.proc.with_env_mut(|data| {
            data.set(name, value);
        });
    }

    pub fn get_env<T>(&self, name: &str) -> Option<T>
    where
        T: for<'de> Deserialize<'de> + Clone,
    {
        // The directory the process runs in has a readable name of its own
        // (`$env.WORK_DIR`), answered from the process itself — never from a
        // stored var or the OS environment of the same name, either of which
        // would name a directory the run is not in.
        if name == consts::ENV_WORK_DIR {
            let dir = self.workdir()?;
            return T::deserialize(serde_json::json!(dir.display().to_string())).ok();
        }
        // find the env from proc
        if let Some(v) = self.proc.with_env(|vars| vars.get(name)) {
            return Some(v);
        }

        // get from system env
        if let Ok(v) = std::env::var(name) {
            match T::deserialize(serde_json::json!(v)) {
                Ok(value) => return Some(value),
                Err(err) => {
                    tracing::warn!(
                        env = name,
                        target_type = std::any::type_name::<T>(),
                        error = %err,
                        "cannot convert system environment variable; treating it as absent"
                    );
                }
            }
        }

        None
    }

    pub fn set_var<T>(&self, name: &str, value: T)
    where
        T: Serialize + Clone,
    {
        self.vars.write().set(name, value);
    }

    pub fn get_var<T>(&self, name: &str) -> Option<T>
    where
        T: for<'de> Deserialize<'de> + Clone,
    {
        self.vars.read().get::<T>(name)
    }

    pub fn eval<T: DeserializeOwned + Serialize>(&self, expr: &str) -> Result<T> {
        Context::scope(self, || self.runtime.env().eval::<T>(expr))
    }

    #[allow(unused)]
    pub(in crate::scheduler) fn action(&self) -> Option<Action> {
        self.action.read().clone()
    }

    #[instrument(skip(self, node, prev))]
    pub fn sched_task(&self, node: &Arc<Node>, prev: Arc<Task>) -> Result<()> {
        debug!(nid = %node.id(), kind = %node.kind(), name = %node.name(), "task scheduled");
        let task = self.proc.create_task(node, Some(prev))?;
        self.runtime.push(&task)?;
        Ok(())
    }

    #[instrument(skip(self, node, vars, parent))]
    pub fn sched_task_with_vars(
        &self,
        node: &Arc<Node>,
        vars: Vars,
        parent: Arc<Task>,
    ) -> Result<()> {
        debug!(nid = %node.id(), kind = %node.kind(), name = %node.name(), "task scheduled");
        let task = self.proc.create_task(node, Some(parent))?;
        task.set_data(&vars);
        self.runtime.push(&task)?;
        Ok(())
    }

    /// Schedule `node` with `prev` as its predecessor, reusing the task
    /// instance that was already created for the same `(node, prev)` slot when
    /// it is still in flight. A crash mid-`next` (or its recovery replay) can
    /// re-run the propagation after the node was already scheduled; re-creating
    /// the task would duplicate it. Instances that were persisted but never
    /// started (state `None`) are re-enqueued; terminal instances (completed,
    /// skipped, …) are left alone so legitimate re-execution (redo, timeout
    /// re-evaluation) still creates fresh tasks.
    pub fn schedule_once(&self, node: &Arc<Node>, prev: Arc<Task>) -> Result<()> {
        if let Some(existing) = self.proc.task_for_node_prev(node.id(), &prev.id)
            && !existing.state().is_completed()
        {
            if existing.state().is_none() {
                self.runtime.push(&existing)?;
            }
            return Ok(());
        }
        self.sched_task(node, prev)
    }

    #[instrument(skip(self, act, vars), fields(uses = %act.uses, name = %act.name))]
    pub fn dispatch_act(&self, act: &Act, vars: Vars) -> Result<()> {
        debug!(nid = %act.id, "act dispatched");
        let task = self.task();

        if !task.state().is_none() {
            let mut id = act.id.to_string();
            if id.is_empty() {
                id = shortid();
            }
            let tree = self.proc.tree();
            let node = tree.append_node(
                self.task().node(),
                &id,
                NodeContent::Act(act.clone()),
                task.node().level + 1,
            )?;
            let task = self.proc.create_task(&node, Some(task))?;
            task.set_data(&vars);
            self.runtime.push(&task)?;
        }

        Ok(())
    }

    pub fn build_acts(&self, acts: &[Act], is_sequence: bool) -> Result<()> {
        let task = self.task();
        let tree = self.proc.tree();

        let mut prev = task.node().clone();
        let mut acts = acts.to_owned();
        for (index, act) in acts.iter_mut().enumerate() {
            dyn_build_act(
                act,
                &tree,
                task.node(),
                &mut prev,
                task.node().level + 1,
                index,
                is_sequence,
            )?;
        }

        Ok(())
    }

    /// redo the task and dispatch directly
    pub fn redo_task(&self, task: &Arc<Task>) -> Result<()> {
        if let Some(prev) = task.prev_id()
            && let Some(prev_task) = self.proc.task(&prev)
        {
            let task = self.proc.create_task(task.node(), Some(prev_task))?;
            self.runtime.push(&task)?;
        }

        Ok(())
    }

    pub async fn back_task(&self, task: &Arc<Task>, paths: &Vec<Arc<Task>>) -> Result<()> {
        for task in task.siblings().iter() {
            if task.state().is_completed() {
                continue;
            }
            task.set_state(TaskState::Skipped);
            self.emit_task(task).await?;
        }

        task.set_state(TaskState::Backed);
        self.emit_task(task).await?;

        // marks the state in the paths
        for p in paths {
            if p.state().is_running() {
                p.set_state(TaskState::Completed);
                self.emit_task(p).await?;
            } else if p.state().is_pending() {
                p.set_state(TaskState::Skipped);
                self.emit_task(p).await?;
            }
        }

        Ok(())
    }

    pub async fn abort_task(&self, task: &Arc<Task>) -> Result<()> {
        // abort all task's acts
        for task in task.siblings().iter() {
            if task.state().is_completed() {
                continue;
            }
            task.set_state(TaskState::Skipped);
            self.emit_task(task).await?;
        }

        task.set_state(TaskState::Aborted);
        task.set_data(&self.vars());
        self.emit_task(task).await?;

        // abort all running task
        let ctx = self;
        let mut parent = task.parent();
        let mut prev = task.clone();
        while let Some(task) = parent {
            task.set_state(TaskState::Aborted);
            ctx.set_task(&task);
            if prev.is_kind(NodeKind::Act) {
                // act task's data will update to parent
                task.update_data(&prev.outputs());
            }
            ctx.emit_task(&ctx.task()).await?;

            for t in task.children() {
                if t.state().is_pending() {
                    t.set_state(TaskState::Skipped);
                    ctx.emit_task(&t).await?;
                } else if t.state().is_running() {
                    t.set_state(TaskState::Aborted);
                    ctx.emit_task(&t).await?;
                }
            }

            prev = task.clone();
            parent = task.parent();
        }
        Ok(())
    }

    /// undo task
    /// the undo task is a step task, set the task as completed and set the children acts as cancelled
    pub async fn undo_task(&self, task: &Arc<Task>) -> Result<()> {
        if task.state().is_completed() {
            return Err(ActError::Action(format!(
                "task('{}') is not allowed to cancel",
                task.id
            )));
        }

        // cancel all of the task's children
        let mut children = task.children();
        while !children.is_empty() {
            let mut nexts = Vec::new();
            for t in &children {
                if t.state().is_completed() {
                    continue;
                }
                t.set_state(TaskState::Cancelled);
                self.emit_task(t).await?;
                nexts.extend_from_slice(&t.children());
            }

            children = nexts;
        }
        task.set_state(TaskState::Completed);
        self.emit_task(task).await?;

        Ok(())
    }

    pub async fn emit_error(&self) -> Result<()> {
        let task = self.task();
        debug!(pid = %task.pid, tid = %task.id, "emit error");
        if task.state().is_error() {
            self.emit_task(&task).await?;

            // after emitting, re-check the task state
            if task.state().is_error()
                && let Some(err) = task.err()
                && let Some(parent) = task.parent()
            {
                parent.set_err(&err);
                if task.is_kind(NodeKind::Act) {
                    // act task's data will update to parent
                    parent.update_data(&task.outputs());
                }
                // boxed: the parent chain can recurse back into `emit_error`
                // through `Task::on_error` (async recursion requires boxing)
                return Box::pin(parent.on_error(self)).await;
            }
        }

        Ok(())
    }

    pub async fn emit_task(&self, task: &Arc<Task>) -> Result<()> {
        debug!(pid = %task.pid, tid = %task.id, state = %task.state(), "emit task");

        // on workflow start
        if let NodeContent::Workflow(_) = &task.node().content
            && task.state().is_created()
        {
            if self.proc.state().is_none() {
                self.proc.set_state(TaskState::Running);
            }
            self.runtime.emitter().emit_proc_event(&self.proc).await;
        }

        self.runtime.emitter().emit_task_event(task).await?;

        // on workflow complete
        if let NodeContent::Workflow(_) = &task.node().content
            && task.state().is_completed()
        {
            self.proc.set_state(task.state());
            if let Some(err) = task.err() {
                self.proc.set_err(&err);
            }

            self.runtime.emitter().emit_proc_event(&self.proc).await;
        }

        Ok(())
    }

    pub async fn emit_message(&self, msg: &Act) -> Result<()> {
        debug!(uses = %msg.uses, name = %msg.name, "emit message");
        let mut inputs = utils::fill_inputs(&msg.vars(), self);

        // append workflow model to inputs
        self.proc.with_model(|workflow| {
            inputs.set(
                consts::WORKFLOW_MODEL_KEY,
                Vars::new()
                    .with("id", &workflow.id)
                    .with("name", &workflow.name)
                    .with("options", &workflow.options),
            );
        });

        // append act.optins to inputs
        inputs.set(consts::ACT_OPTIONS_KEY, msg.options.clone());

        // append act.params to inputs
        let params = utils::fill_params(&msg.params, self);
        inputs.set(consts::ACT_PARAMS_KEY, params);

        let task = self.task();
        if let Some(err) = task.err() {
            inputs.set(consts::ACT_ERR_MESSAGE, err.message);
            inputs.set(consts::ACT_ERR_CODE, err.ecode);
        }

        let state: MessageState = MessageState::Completed;
        let msg = Message {
            id: utils::longid(),
            r#type: "act".to_string(),
            state,
            pid: task.pid.clone(),
            tid: task.id.clone(),
            name: task.node().name(),
            uses: Some(msg.uses.clone()),
            inputs,
            ..Default::default()
        };

        self.runtime.emitter().emit_message(&msg);
        Ok(())
    }

    /// Enqueue the task's `next` propagation through the durable outbox:
    /// the task state is flushed and a `Pending` outbox record is written
    /// before the in-memory queue dispatch.
    pub async fn push_next(&self) -> Result<()> {
        self.runtime.enqueue_next(&self.task()).await
    }
}