acts 0.25.0

use crate::{
    Event, Message, Result, Vars, event::ActWorkflowMessageHandle, scheduler::Runtime, utils,
};
use std::pin::Pin;
use std::sync::Arc;
use tracing::{debug, error, info, warn};

/// channel match filters: (type, state, uses, options) globs
type GlobSet = (
    globset::GlobMatcher,
    globset::GlobMatcher,
    globset::GlobMatcher,
    Vec<(String, globset::GlobMatcher)>,
);

#[derive(Debug, Clone)]
pub struct ChannelOptions {
    /// The channel key this subscription registers under. Under an `[acl]` the
    /// transports compose it with the caller's subject
    /// ([`ChannelOptions::subscription_id`]) so two callers naming the same
    /// client id cannot take over each other's channel; the filters below stay
    /// self-declared either way, and a message is delivered to every channel
    /// whose filters match it.
    pub id: String,

    /// need ack the message
    pub ack: bool,

    /// use the glob pattern to match the message type
    /// eg. {workflow,step,branch,req,msg}
    pub r#type: String,
    /// use the glob pattern to match the message state
    /// eg. {created,completed}
    pub state: String,

    /// use the glob pattern to match the message uses
    pub uses: String,

    /// use the custom glob pattern
    pub options: Vars,
}

impl Default for ChannelOptions {
    fn default() -> Self {
        Self {
            id: utils::shortid(),
            ack: false,
            r#type: "*".to_string(),
            state: "*".to_string(),
            uses: "*".to_string(),
            options: Vars::new(),
        }
    }
}

impl ChannelOptions {
    /// The channel key of one client's subscription: `{subject}/{client_id}`.
    ///
    /// The key is what the emitter registry holds a channel's handler under,
    /// and a registration under an existing key *replaces* that handler
    /// (`Emitter::on_message`) — so it has to carry the subject. Under a bare
    /// client id a second caller subscribing with the same id would take over
    /// the first caller's channel and receive its messages, including the
    /// redeliveries it never acked. A role name carrying the separator is
    /// refused at config load, so the prefix is never ambiguous.
    pub fn subscription_id(subject: &str, client_id: &str) -> String {
        format!("{subject}/{client_id}")
    }
    pub fn pattern(&self) -> String {
        let mut options = Vars::new()
            .with("ack", self.ack)
            .with("type", self.r#type.clone())
            .with("state", self.state.clone())
            .with("uses", self.uses.clone());

        for (key, value) in self.options.iter() {
            options.set(key, value);
        }
        options.to_string()
    }
}

/// Just a export struct for the event::Emitter
///
pub struct Channel {
    runtime: Arc<Runtime>,
    ack: bool,
    chan_id: String,
    pattern: String,
    glob: GlobSet,
}

impl Channel {
    pub fn new(rt: &Arc<Runtime>) -> Self {
        Self::channel(rt, &ChannelOptions::default())
    }

    /// create a emit channel to receive message
    /// if the message is not received by client, the engine will re-send at the next time interval
    #[allow(clippy::self_named_constructors)]
    pub fn channel(rt: &Arc<Runtime>, options: &ChannelOptions) -> Self {
        debug!("channel created");
        let pat_type = compile_glob(&options.r#type, "type");
        let pat_state = compile_glob(&options.state, "state");
        let pat_uses = compile_glob(&options.uses, "uses");
        let opt_globs: Vec<(String, globset::GlobMatcher)> = options
            .options
            .iter()
            .filter_map(|(k, v)| {
                v.as_str()
                    .and_then(|pattern| globset::Glob::new(pattern).ok())
                    .map(|g| (k.clone(), g.compile_matcher()))
            })
            .collect();
        Self {
            runtime: rt.clone(),
            ack: options.ack,
            chan_id: options.id.clone(),
            pattern: options.pattern(),
            glob: (pat_type, pat_state, pat_uses, opt_globs),
        }
    }

    ///  Receive act message
    ///
    /// Example
    /// ```rust,no_run
    /// use acts::{Engine, Act, Principal, Workflow, Vars, Message};
    ///
    /// #[tokio::main]
    /// async fn main() {
    ///     let engine = Engine::builder().start().await.unwrap();
    ///     let workflow = Workflow::new().with_id("m1").with_step(|step| {
    ///             step.with_id("step1").with_uses("acts.core.irq", Vars::new().with("var1", 10))
    ///     });
    ///
    ///     engine.channel().on_message(move |e| async move {
    ///         if let Some(uses) = &e.uses && e.r#type == "act" && uses == "acts.core.irq" {
    ///             println!("act message: state={} inputs={:?} outputs={:?}", e.state, e.inputs, e.outputs);
    ///         }
    ///     });
    ///     let exec = engine.executor(&Principal::unrestricted());
    ///     exec.model().deploy(&workflow, None).await.expect("fail to deploy workflow");
    ///     let mut vars = Vars::new();
    ///     vars.set("pid", "w1");
    ///     exec.proc().start(&workflow.id, vars).await.unwrap();
    /// }
    /// ```
    pub fn on_message<F, Fut>(self: &Arc<Self>, f: F)
    where
        F: Fn(Event<Message>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = ()> + Send + 'static,
    {
        let glob = self.glob.clone();
        let runtime = self.runtime.clone();
        let ack = self.ack;
        let chan_id = self.chan_id.clone();
        let pattern = self.pattern.clone();
        let handle: ActWorkflowMessageHandle =
            Arc::new(move |e| -> Pin<Box<dyn Future<Output = ()> + Send>> { Box::pin(f(e)) });
        let chan = chan_id.clone();
        self.runtime.emitter().on_message(&self.chan_id, move |e| {
            debug!(chan = %chan, "on message");
            let glob = glob.clone();
            let runtime = runtime.clone();
            let handle = handle.clone();
            let chan_id = chan_id.clone();
            let pattern = pattern.clone();
            async move {
                deliver(&glob, &runtime, ack, &chan_id, &pattern, &handle, e).await;
            }
        });
    }

    pub fn on_start<F, Fut>(self: &Arc<Self>, f: F)
    where
        F: Fn(Event<Message>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = ()> + Send + 'static,
    {
        let glob = self.glob.clone();
        let runtime = self.runtime.clone();
        let ack = self.ack;
        let chan_id = self.chan_id.clone();
        let pattern = self.pattern.clone();
        let handle: ActWorkflowMessageHandle =
            Arc::new(move |e| -> Pin<Box<dyn Future<Output = ()> + Send>> { Box::pin(f(e)) });
        self.runtime.emitter().on_start(&self.chan_id, move |e| {
            let glob = glob.clone();
            let runtime = runtime.clone();
            let handle = handle.clone();
            let chan_id = chan_id.clone();
            let pattern = pattern.clone();
            async move {
                deliver(&glob, &runtime, ack, &chan_id, &pattern, &handle, e).await;
            }
        });
    }

    pub fn on_complete<F, Fut>(self: &Arc<Self>, f: F)
    where
        F: Fn(Event<Message>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = ()> + Send + 'static,
    {
        let glob = self.glob.clone();
        let runtime = self.runtime.clone();
        let ack = self.ack;
        let chan_id = self.chan_id.clone();
        let pattern = self.pattern.clone();
        let handle: ActWorkflowMessageHandle =
            Arc::new(move |e| -> Pin<Box<dyn Future<Output = ()> + Send>> { Box::pin(f(e)) });
        let chan = chan_id.clone();
        self.runtime.emitter().on_complete(&self.chan_id, move |e| {
            debug!(chan = %chan, "on complete");
            let glob = glob.clone();
            let runtime = runtime.clone();
            let handle = handle.clone();
            let chan_id = chan_id.clone();
            let pattern = pattern.clone();
            async move {
                deliver(&glob, &runtime, ack, &chan_id, &pattern, &handle, e).await;
            }
        });
    }

    pub fn on_error<F, Fut>(self: &Arc<Self>, f: F)
    where
        F: Fn(Event<Message>) -> Fut + Send + Sync + 'static,
        Fut: Future<Output = ()> + Send + 'static,
    {
        let glob = self.glob.clone();
        let runtime = self.runtime.clone();
        let ack = self.ack;
        let chan_id = self.chan_id.clone();
        let pattern = self.pattern.clone();
        let handle: ActWorkflowMessageHandle =
            Arc::new(move |e| -> Pin<Box<dyn Future<Output = ()> + Send>> { Box::pin(f(e)) });
        self.runtime.emitter().on_error(&self.chan_id, move |e| {
            let glob = glob.clone();
            let runtime = runtime.clone();
            let handle = handle.clone();
            let chan_id = chan_id.clone();
            let pattern = pattern.clone();
            async move {
                deliver(&glob, &runtime, ack, &chan_id, &pattern, &handle, e).await;
            }
        });
    }

    /// Deregister this channel's handler from the engine emitter.
    ///
    /// Transport layers (SSE, gRPC, ...) must call this when a client
    /// disconnects, otherwise dead handlers accumulate in the emitter map:
    /// every future message pays glob matching plus ack-delivery store
    /// writes for channels nobody listens on anymore.
    pub fn close(&self) {
        self.runtime.emitter().remove(&self.chan_id);
    }
}

/// Deliver a message event to a channel handler. When the channel requires
/// acks and the event is a fresh emission (not a redelivery), it is first
/// stored as a delivery row of this channel — the handler event is then
/// tagged with the new delivery id so the client can ack this exact delivery.
/// Redeliveries already carry their delivery id and pass through untouched.
/// If a required store fails the message is not delivered.
async fn deliver(
    glob: &GlobSet,
    runtime: &Arc<Runtime>,
    ack: bool,
    chan_id: &str,
    pattern: &str,
    f: &ActWorkflowMessageHandle,
    e: Event<Message>,
) {
    if !is_match(glob, &e) {
        return;
    }

    match store_if(runtime, ack, chan_id, pattern, &e).await {
        Ok(Some(delivery_id)) => {
            let mut msg = e.inner().clone();
            msg.delivery_id = Some(delivery_id.clone());
            let event = Event::from_inner(msg);
            f(event).await;
            // delivery succeeded: the channel handler ran to completion —
            // record it explicitly (a handler that acked/closed the row while
            // running is never downgraded). A row left `Created` means it was
            // never successfully handed over (no handler / crash
            // mid-dispatch) and still needs a (re-)dispatch; `Delivered` ones
            // only wait for an ack or the task close.
            if let Err(err) = runtime.cache().store().mark_delivered(&delivery_id).await {
                error!(error = %err, delivery_id = %delivery_id, "mark delivery succeeded failed");
            }
        }
        Ok(None) => {
            f(e).await;
        }
        Err(err) => error!(error = %err, chan = %chan_id, "delivery store failed, message dropped"),
    }
}

/// Store the message as a delivery row of one channel when the channel must
/// ack it. The canonical message row is stored once per message id and every
/// channel delivery of the same event gets its own delivery row. Returns
/// `Ok(Some(delivery_id))` when a fresh delivery row was stored, `Ok(None)`
/// when nothing needs storing (non-ack channel, a redelivery that already has
/// its row, or a process whose rows are already gone), `Err` when the store
/// failed.
///
/// The row is written through the process's writer shard
/// ([`Cache::store_delivery`]), not straight into the store: it has to be
/// ordered with the process's task writes, and the close of a finished task is
/// what settles the deliveries of that task. A row created off that order can
/// land after the close — open on a task that is already over — and keep the
/// process's rows alive forever.
async fn store_if(
    runtime: &Arc<Runtime>,
    ack: bool,
    chan_id: &str,
    pattern: &str,
    message: &Message,
) -> Result<Option<String>> {
    if ack && !chan_id.is_empty() && message.delivery_id.is_none() {
        info!(r#type = message.r#type, pid = %message.pid, tid = %message.tid, mid = %message.mid,  state = %message.state, "delivery stored");

        // each channel delivery gets its own delivery row
        let delivery = message.into_delivery(chan_id, pattern);
        match runtime
            .cache()
            .store_delivery(&message.into_message(), &delivery)
            .await
        {
            // stored: the handler is handed this delivery id to ack
            Ok(true) => Ok(Some(delivery.id)),
            // the process is gone — its rows were swept — so the message is
            // handed over without a delivery row: there is nothing left to
            // ack, and re-creating rows behind the removal would strand them
            Ok(false) => Ok(None),
            Err(err) => {
                error!(error = %err, "channel store failure");
                Err(err)
            }
        }
    } else {
        Ok(None)
    }
}

fn is_match(glob: &GlobSet, e: &Event<Message>) -> bool {
    let (pat_type, pat_state, pat_uses, pat_options) = glob;
    if !pat_type.is_match(&e.r#type)
        || !pat_state.is_match(e.state.as_ref())
        || !pat_uses.is_match(e.uses.as_deref().unwrap_or_default())
    {
        return false;
    }

    let msg_options = e.options();
    for (key, pat) in pat_options {
        let value = msg_options
            .as_ref()
            .and_then(|o| o.get::<String>(key))
            .unwrap_or_default();
        if !pat.is_match(&value) {
            return false;
        }
    }
    true
}

fn compile_glob(pattern: &str, field: &str) -> globset::GlobMatcher {
    globset::Glob::new(pattern)
        .map(|glob| glob.compile_matcher())
        .unwrap_or_else(|err| {
            warn!(field = field, pattern, error = %err, "invalid channel glob pattern; falling back to wildcard");
            globset::Glob::new("*")
                .expect("fallback glob is valid")
                .compile_matcher()
        })
}

#[cfg(test)]
mod tests {
    use super::compile_glob;

    #[test]
    fn invalid_glob_falls_back_to_match_all() {
        let matcher = compile_glob("[", "type");
        assert!(matcher.is_match("act"));
        assert!(matcher.is_match(""));
    }
}