dyniak 0.0.2

//! Phase types for a Riak-style MapReduce pipeline.
//!
//! The pipeline is an ordered list of phases. Each phase reads from
//! an inbound mpsc, runs its function, and pushes outputs to the
//! next phase's inbound. Phases are dispatched by name through the
//! [`crate::mapreduce::registry::PhaseRegistry`] except for
//! [`Phase::Link`] (whose semantics are baked into the executor) and
//! [`Phase::WasmModule`] (reserved for a future task; the executor
//! returns [`crate::mapreduce::MrError::WasmNotImplemented`] when it
//! encounters one).

use serde::{Deserialize, Serialize};
use serde_json::Value;

/// A single phase in a MapReduce pipeline.
///
/// The variants mirror Riak's `{map, ...} | {reduce, ...} | {link,
/// ...}` shapes from the HTTP `/mapred` schema, plus a fourth
/// `WasmModule` reservation slot for the future Wasm-fitting task.
///
/// `keep` matches Riak's `keep` flag: when set, the phase's outputs
/// are also captured into the final response. The last phase is
/// always considered to keep its outputs even when the flag is
/// `false`, mirroring Riak's behaviour.
#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum Phase {
    /// Map phase: invoke `fn_name` once per inbound item; emit zero
    /// or more outputs.
    Map {
        /// Registered map function name.
        #[serde(rename = "name")]
        fn_name: String,
        /// Optional JSON argument forwarded to the map function.
        #[serde(default)]
        arg: Option<Value>,
        /// Keep the phase's outputs in the final response.
        #[serde(default)]
        keep: bool,
    },
    /// Reduce phase: invoke `fn_name` once over every inbound item;
    /// emit zero or more outputs.
    Reduce {
        /// Registered reduce function name.
        #[serde(rename = "name")]
        fn_name: String,
        /// Optional JSON argument forwarded to the reduce function.
        #[serde(default)]
        arg: Option<Value>,
        /// Keep the phase's outputs in the final response.
        #[serde(default)]
        keep: bool,
    },
    /// Link phase: walk links from each inbound object and emit
    /// `(bucket, key)` pairs for matching links. The link semantics
    /// follow Riak's `Link-Walking` syntax: `bucket` and `tag` are
    /// optional patterns, with `None` meaning "match any".
    Link {
        /// Bucket pattern; `None` matches any bucket.
        #[serde(default)]
        bucket: Option<String>,
        /// Tag pattern; `None` matches any tag.
        #[serde(default)]
        tag: Option<String>,
        /// Keep the phase's outputs in the final response.
        #[serde(default)]
        keep: bool,
    },
    /// Reserved phase variant for a future Wasm fitting. Present
    /// in the enum so the JSON schema is forwards-compatible. The
    /// executor rejects this variant with
    /// [`crate::mapreduce::MrError::WasmNotImplemented`] until the
    /// Wasm fitting task lands.
    WasmModule {
        /// Wasm module identifier (registry-style name).
        module_id: String,
        /// Function name within the module.
        fn_name: String,
        /// Optional JSON argument forwarded to the function.
        #[serde(default)]
        arg: Option<Value>,
        /// Keep the phase's outputs in the final response.
        #[serde(default)]
        keep: bool,
    },
}

impl Phase {
    /// Whether the phase's outputs should be captured in the final
    /// response.
    #[must_use]
    pub fn keep(&self) -> bool {
        match self {
            Self::Map { keep, .. }
            | Self::Reduce { keep, .. }
            | Self::Link { keep, .. }
            | Self::WasmModule { keep, .. } => *keep,
        }
    }
}

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

    #[test]
    fn keep_is_per_variant() {
        let p = Phase::Map {
            fn_name: "f".into(),
            arg: None,
            keep: true,
        };
        assert!(p.keep());
        let p = Phase::Reduce {
            fn_name: "f".into(),
            arg: None,
            keep: false,
        };
        assert!(!p.keep());
    }

    #[test]
    fn map_phase_round_trips_through_json() {
        let p = Phase::Map {
            fn_name: "map_object_value".into(),
            arg: Some(serde_json::json!({"k": "v"})),
            keep: true,
        };
        let s = serde_json::to_string(&p).expect("encode");
        let back: Phase = serde_json::from_str(&s).expect("decode");
        assert_eq!(back, p);
    }

    #[test]
    fn reduce_phase_round_trips_through_json() {
        let p = Phase::Reduce {
            fn_name: "reduce_sum".into(),
            arg: None,
            keep: false,
        };
        let s = serde_json::to_string(&p).expect("encode");
        let back: Phase = serde_json::from_str(&s).expect("decode");
        assert_eq!(back, p);
    }

    #[test]
    fn link_phase_round_trips_through_json() {
        let p = Phase::Link {
            bucket: Some("friends".into()),
            tag: Some("knows".into()),
            keep: true,
        };
        let s = serde_json::to_string(&p).expect("encode");
        let back: Phase = serde_json::from_str(&s).expect("decode");
        assert_eq!(back, p);
    }

    #[test]
    fn wasm_phase_is_present_in_enum() {
        // The Wasm variant must round-trip even though execution is
        // not implemented yet. This guarantees the JSON schema is
        // forwards-compatible: clients can submit Wasm-bearing jobs
        // today and receive a typed error.
        let p = Phase::WasmModule {
            module_id: "m".into(),
            fn_name: "f".into(),
            arg: None,
            keep: false,
        };
        let s = serde_json::to_string(&p).expect("encode");
        let back: Phase = serde_json::from_str(&s).expect("decode");
        assert_eq!(back, p);
    }
}