cross-stream 0.13.4

use std::str::FromStr;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use serde::{Deserialize, Serialize};

use tokio::io::AsyncReadExt;

use nu_protocol::Value;

use scru128::Scru128Id;

use crate::error::Error;
use crate::nu;
use crate::nu::value_to_json;
use crate::nu::{NuScriptConfig, ReturnOptions};
use crate::store::{FollowOption, Frame, ReadOptions, Store};

#[derive(Clone)]
pub struct Actor {
    pub id: Scru128Id,
    pub topic: String,
    config: ActorConfig,
    engine: nu::Engine,
    closure: nu_protocol::engine::Closure,
    initial_state: Value,
    output: Arc<Mutex<Vec<Frame>>>,
}

#[derive(Clone, Debug)]
struct ActorConfig {
    start: Start,
    pulse: Option<u64>,
    return_options: Option<ReturnOptions>,
    /// Optional topic filter, normalized to one pattern per element. Applied
    /// at the read level (see `configure_read_options`): frames whose topic
    /// matches none of these patterns never reach the actor loop. Each
    /// pattern is an exact topic, a `prefix.*` wildcard, or `*`, the same
    /// pattern language as `ReadOptions::topic`. Absent means all frames.
    topics: Option<Vec<String>>,
}

#[derive(Clone, Debug, Default, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
enum Start {
    First,
    #[default]
    New,
    After(Scru128Id),
}

/// Options that can be deserialized directly from a script.
#[derive(Deserialize, Debug, Default)]
#[serde(default)] // Use default values when fields are missing
struct ActorScriptOptions {
    /// Actor can specify where to start: "first", "new", or a specific ID
    start: Option<String>,
    /// Optional heartbeat interval in milliseconds
    pulse: Option<u64>,
    /// Optional customizations for return frames
    return_options: Option<ReturnOptions>,
    /// Initial state for the actor closure's second parameter
    initial: Option<serde_json::Value>,
    /// Optional list of topic patterns the actor cares about. Accepts a
    /// nushell list of strings or a single comma-separated string.
    topics: Option<TopicsSpec>,
}

/// A topics spec from the actor script: a single string (commas allowed) or
/// a list of strings.
#[derive(Deserialize, Debug)]
#[serde(untagged)]
enum TopicsSpec {
    One(String),
    Many(Vec<String>),
}

impl TopicsSpec {
    /// Normalize to one pattern per element, splitting comma-separated
    /// strings and dropping empty elements.
    fn into_patterns(self) -> Vec<String> {
        let parts: Vec<String> = match self {
            TopicsSpec::One(s) => vec![s],
            TopicsSpec::Many(v) => v,
        };
        parts
            .iter()
            .flat_map(|s| s.split(','))
            .map(|p| p.trim().to_string())
            .filter(|p| !p.is_empty())
            .collect()
    }
}

pub(super) enum ClosureResult {
    Continue {
        output: Option<Value>,
        next_state: Value,
    },
    Stop {
        output: Option<Value>,
    },
}

impl Actor {
    pub async fn new(
        id: Scru128Id,
        topic: String,
        mut engine: nu::Engine,
        expression: String,
        store: Store,
    ) -> Result<Self, Error> {
        let output = Arc::new(Mutex::new(Vec::new()));
        // Reads come from the prepared base; only the per-instance buffered
        // `.append` (for atomic batch commit) is added here.
        nu::add_write_commands(
            &mut engine,
            &store,
            nu::AppendMode::Buffered(output.clone()),
        )?;

        // Parse configuration using the new generic API
        let nu_script_config = nu::parse_config(&mut engine, &expression)?;

        // Deserialize actor-specific options from the full_config_value
        let (actor_config, initial_json) = extract_actor_config(&nu_script_config)?;

        // Validate closure signature and resolve initial state
        let block = engine
            .state
            .get_block(nu_script_config.run_closure.block_id);
        let num_required = block.signature.required_positional.len();
        let num_optional = block.signature.optional_positional.len();

        let total_positional = num_required + num_optional;
        if total_positional != 2 {
            return Err(format!(
                "Closure must accept exactly 2 params (frame, state), got {total_positional}"
            )
            .into());
        }

        // Resolve initial state: config `initial` > param default > null
        let span = nu_protocol::Span::unknown();
        let initial_state = if let Some(json) = initial_json {
            crate::nu::util::json_to_value(&json, span)
        } else if num_optional > 0 {
            let state_param = &block.signature.optional_positional[0];
            state_param
                .default_value
                .clone()
                .unwrap_or_else(|| Value::nothing(span))
        } else {
            Value::nothing(span)
        };

        Ok(Self {
            id,
            topic,
            config: actor_config,
            engine,
            closure: nu_script_config.run_closure,
            initial_state,
            output,
        })
    }

    /// Evaluate the actor closure for one frame, inline on the calling
    /// thread. The whole actor loop runs on a dedicated OS thread (see
    /// `run_blocking`), so there is no async->worker->oneshot handoff per
    /// frame -- that round trip (two context switches) was the dominant
    /// per-frame cost in backlog replay. `state` is threaded by the caller.
    fn eval_frame(&mut self, frame: &Frame, state: &Value) -> Result<ClosureResult, Error> {
        let frame_val = crate::nu::frame_to_value(frame, nu_protocol::Span::unknown(), false);
        self.engine
            .eval_closure_no_job(&self.closure, vec![frame_val, state.clone()], None)
            .map_err(|e| {
                let working_set = nu_protocol::engine::StateWorkingSet::new(&self.engine.state);
                Error::from(nu_protocol::format_cli_error(None, &working_set, &*e, None))
            })
            .and_then(|pd| {
                pd.into_value(nu_protocol::Span::unknown())
                    .map_err(Error::from)
            })
            .and_then(interpret_closure_result)
    }

    fn process_frame(
        &mut self,
        frame: &Frame,
        store: &Store,
        state: &mut Value,
    ) -> Result<bool, Error> {
        let result = self.eval_frame(frame, state)?;

        let (output, should_continue) = match result {
            ClosureResult::Continue {
                output,
                ref next_state,
            } => {
                *state = next_state.clone();
                (output, true)
            }
            ClosureResult::Stop { output } => (output, false),
        };

        // Check if the evaluated value is an append frame
        let additional_frame = match output {
            Some(ref value)
                if !is_value_an_append_frame_from_actor(value, &self.id)
                    && !matches!(value, Value::Nothing { .. }) =>
            {
                let return_options = self.config.return_options.as_ref();
                let suffix = return_options
                    .and_then(|ro| ro.suffix.as_deref())
                    .unwrap_or(".out");
                let use_cas = return_options
                    .and_then(|ro| ro.target.as_deref())
                    .is_some_and(|t| t == "cas");

                let topic = format!("{topic}{suffix}", topic = self.topic, suffix = suffix);
                let ttl = return_options.and_then(|ro| ro.ttl.clone());

                if use_cas {
                    let hash = match value {
                        Value::Binary { val, .. } => store.cas_insert_sync(val)?,
                        _ => store.cas_insert_sync(value_to_json(value).to_string())?,
                    };
                    Some(
                        Frame::builder(topic)
                            .maybe_ttl(ttl)
                            .maybe_hash(Some(hash))
                            .build(),
                    )
                } else {
                    // Default: records go to meta, non-records are an error
                    match value {
                        Value::Record { .. } => {
                            let json = value_to_json(value);
                            Some(Frame::builder(topic).maybe_ttl(ttl).meta(json).build())
                        }
                        _ => {
                            return Err(format!(
                                "Actor output must be a record when target is not \"cas\"; got {}. \
                                 Set return_options.target to \"cas\" for non-record output.",
                                value.get_type()
                            )
                            .into());
                        }
                    }
                }
            }
            _ => None,
        };

        // Process buffered appends and the additional frame
        let output_to_process: Vec<_> = {
            let mut output = self.output.lock().unwrap();
            output.drain(..).chain(additional_frame).collect()
        };

        for mut output_frame in output_to_process {
            let meta_obj = output_frame
                .meta
                .get_or_insert_with(|| serde_json::Value::Object(Default::default()))
                .as_object_mut()
                .expect("meta should be an object");

            meta_obj.insert(
                "actor_id".to_string(),
                serde_json::Value::String(self.id.to_string()),
            );
            meta_obj.insert(
                "frame_id".to_string(),
                serde_json::Value::String(frame.id.to_string()),
            );

            let _ = store.append(output_frame);
        }

        Ok(should_continue)
    }

    /// The actor's hot loop, run on a dedicated OS thread. Pulls frames with
    /// blocking_recv (no async runtime hop) and evaluates each inline -- the
    /// per-frame async->worker->oneshot round trip is gone. State is threaded
    /// here and passed by `&mut` into `process_frame`. All store ops on this
    /// path (`append`, `cas_insert_sync`) are synchronous.
    fn run_blocking(mut self, mut recver: mpsc::Receiver<Frame>, store: Store) {
        // One long-lived background job, so per-frame eval can skip job churn.
        self.engine.attach_background_job("actor");
        let mut state = self.initial_state.clone();

        let create_topic = format!("xs.actor.{}.create", self.topic);
        let term_topic = format!("xs.actor.{}.term", self.topic);
        let store = &store;

        while let Some(frame) = recver.blocking_recv() {
            // Skip lifecycle activity that occurred before this actor was registered
            if (frame.topic == create_topic || frame.topic == term_topic) && frame.id <= self.id {
                continue;
            }

            // A newer .create wins: this actor steps aside. Emit .replaced
            // (the successor's .active will be next).
            if frame.topic == create_topic {
                let _ = store.append(
                    Frame::builder(format!("xs.actor.{}.replaced", &self.topic))
                        .meta(serde_json::json!({
                            "actor_id": self.id.to_string(),
                            "frame_id": frame.id.to_string(),
                        }))
                        .build(),
                );
                break;
            }

            // User-requested stop.
            if frame.topic == term_topic {
                let _ = store.append(
                    Frame::builder(format!("xs.actor.{}.fin.term", &self.topic))
                        .meta(serde_json::json!({
                            "actor_id": self.id.to_string(),
                            "frame_id": frame.id.to_string(),
                        }))
                        .build(),
                );
                break;
            }

            // Skip frames that were generated by this actor
            if frame
                .meta
                .as_ref()
                .and_then(|meta| meta.get("actor_id"))
                .and_then(|actor_id| actor_id.as_str())
                .filter(|actor_id| *actor_id == self.id.to_string())
                .is_some()
            {
                continue;
            }

            match self.process_frame(&frame, store, &mut state) {
                Ok(true) => {}
                Ok(false) => {
                    // Actor self-terminated (natural completion).
                    let _ = store.append(
                        Frame::builder(format!("xs.actor.{}.fin.ok", &self.topic))
                            .meta(serde_json::json!({
                                "actor_id": self.id.to_string(),
                                "frame_id": frame.id.to_string(),
                            }))
                            .build(),
                    );
                    break;
                }
                Err(err) => {
                    // Runtime crash.
                    let _ = store.append(
                        Frame::builder(format!("xs.actor.{}.fin.error", &self.topic))
                            .meta(serde_json::json!({
                                "actor_id": self.id.to_string(),
                                "frame_id": frame.id.to_string(),
                                "error": err.to_string(),
                            }))
                            .build(),
                    );
                    break;
                }
            }
        }
    }

    pub async fn spawn(self, store: Store) -> Result<(), Error> {
        let options = self.configure_read_options().await;
        // Set up the read stream on the async runtime, then run the actor's
        // loop on a dedicated OS thread that drains it with blocking_recv.
        // This keeps the synchronous nushell eval off the tokio runtime
        // without paying a per-frame thread handoff.
        let recver = store.read(options).await;

        let _ = store.append(
            Frame::builder(format!("xs.actor.{}.active", &self.topic))
                .meta(serde_json::json!({
                    "actor_id": self.id.to_string(),
                    "start": self.config.start,
                }))
                .build(),
        );

        let store_for_loop = store.clone();
        std::thread::Builder::new()
            .name(format!("actor-{}", self.topic))
            .spawn(move || {
                self.run_blocking(recver, store_for_loop);
            })
            .map_err(|e| Error::from(format!("Failed to spawn actor thread: {e}")))?;

        Ok(())
    }

    pub async fn from_frame(frame: &Frame, store: &Store) -> Result<Self, Error> {
        let topic = frame
            .topic
            .strip_prefix("xs.actor.")
            .and_then(|rest| rest.strip_suffix(".create"))
            .ok_or("Frame topic must be xs.actor.<name>.create")?;

        // Get hash and read expression
        let hash = frame.hash.as_ref().ok_or("Missing hash field")?;
        let mut reader = store
            .cas_reader(hash.clone())
            .await
            .map_err(|e| format!("Failed to get cas reader: {e}"))?;

        let mut expression = String::new();
        reader
            .read_to_string(&mut expression)
            .await
            .map_err(|e| format!("Failed to read expression: {e}"))?;

        // Prepared base (Plain reads); the actor's per-instance buffered
        // `.append` is added in Actor::new. Modules as of this frame.
        let mut engine = nu::prepared_base(store, nu::ReadMode::Plain, false)?;
        let modules = store.nu_modules_at(&frame.id);
        nu::load_modules(&mut engine.state, store, &modules)?;

        let actor = Actor::new(
            frame.id,
            topic.to_string(),
            engine,
            expression,
            store.clone(),
        )
        .await?;

        Ok(actor)
    }

    async fn configure_read_options(&self) -> ReadOptions {
        // Determine after and new flag based on Start
        let (after, is_new) = match &self.config.start {
            Start::First => (None, false),
            Start::New => (None, true),
            Start::After(id) => (Some(*id), false),
        };

        // Configure follow option based on pulse setting
        let follow_option = self
            .config
            .pulse
            .map(|pulse| FollowOption::WithHeartbeat(Duration::from_millis(pulse)))
            .unwrap_or(FollowOption::On);

        // Apply the actor's topic filter at the read level: non-matching
        // frames are skipped inside the store and never reach the actor
        // loop. The actor's own lifecycle topics are always included so the
        // loop still sees .create/.term frames; synthetic frames (the
        // heartbeat xs.pulse when `pulse` is set, and xs.threshold) bypass
        // read-level topic filtering entirely, so they are unaffected.
        let topic = self.config.topics.as_ref().map(|patterns| {
            let mut patterns = patterns.clone();
            patterns.push(format!("xs.actor.{}.create", self.topic));
            patterns.push(format!("xs.actor.{}.term", self.topic));
            patterns.join(",")
        });

        ReadOptions::builder()
            .follow(follow_option)
            .new(is_new)
            .maybe_after(after)
            .maybe_topic(topic)
            .build()
    }
}

use tokio::sync::mpsc;

fn interpret_closure_result(value: Value) -> Result<ClosureResult, Error> {
    match value {
        Value::Nothing { .. } => Ok(ClosureResult::Stop { output: None }),
        Value::Record { ref val, .. } => {
            for key in val.columns() {
                if key != "out" && key != "next" {
                    return Err(format!(
                        "Unexpected key '{key}' in closure return record; only 'out' and 'next' are allowed"
                    )
                    .into());
                }
            }
            let output = val.get("out").cloned();
            match val.get("next").cloned() {
                Some(next_state) => Ok(ClosureResult::Continue { output, next_state }),
                None => Ok(ClosureResult::Stop { output }),
            }
        }
        _ => Err(format!(
            "Closure must return a record with 'out' and/or 'next' keys, or nothing; got {}",
            value.get_type()
        )
        .into()),
    }
}

fn is_value_an_append_frame_from_actor(value: &Value, actor_id: &Scru128Id) -> bool {
    value
        .as_record()
        .ok()
        .filter(|record| record.get("id").is_some() && record.get("topic").is_some())
        .and_then(|record| record.get("meta"))
        .and_then(|meta| meta.as_record().ok())
        .and_then(|meta_record| meta_record.get("actor_id"))
        .and_then(|id| id.as_str().ok())
        .filter(|id| *id == actor_id.to_string())
        .is_some()
}

/// Extract actor-specific configuration from the generic NuScriptConfig
fn extract_actor_config(
    script_config: &NuScriptConfig,
) -> Result<(ActorConfig, Option<serde_json::Value>), Error> {
    // Deserialize the actor script options using the deserialize_options method
    let script_options: ActorScriptOptions = script_config.deserialize_options()?;

    // Process start into the proper enum
    let start =
        match script_options.start.as_deref() {
            Some("first") => Start::First,
            Some("new") => Start::New,
            Some(id_str) => Start::After(Scru128Id::from_str(id_str).map_err(|_| -> Error {
                format!("Invalid scru128 ID for start: {id_str}").into()
            })?),
            None => Start::default(), // Default if not specified in script
        };

    // Build and return the ActorConfig and initial state
    Ok((
        ActorConfig {
            start,
            pulse: script_options.pulse,
            return_options: script_options.return_options,
            topics: script_options.topics.map(TopicsSpec::into_patterns),
        },
        script_options.initial,
    ))
}