af_workflow/

drafts.rs

//! Durable, immutable workflow proposals with an explicit owner decision.
use af_context::{InstanceId, WorkflowDraftId};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use serde_json::Value;

/// Optional activation committed with publication when a proposal is confirmed.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkflowActivation {
    /// Stable instance identity displayed in the proposal.
    pub instance_id: InstanceId,
    /// Lifecycle pinned by the created instance.
    pub lifecycle: crate::LifecyclePolicy,
    /// Instance configuration validated against the published Spec schema.
    pub config: Value,
    /// Trigger bindings to publish atomically; at most 32.
    #[serde(default)]
    pub bindings: Vec<crate::TriggerBinding>,
}
/// A proposal never changes after it is stored. Editing requires a new draft ID.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkflowProposal {
    /// Initial publication name.
    pub name: String,
    /// Exact revision to publish.
    pub revision: crate::WorkflowRevision,
    /// Exact execution profile to publish.
    pub profile: crate::ExecutionProfileRevision,
    /// Omit to publish without starting work.
    pub activation: Option<WorkflowActivation>,
}
impl WorkflowProposal {
    /// Validate structure and bound the atomic confirmation transaction.
    pub fn validate(&self) -> Result<(), String> {
        if self.name.trim().is_empty() || self.name.chars().count() > 256 {
            return Err("workflow name requires 1..256 characters".into());
        }
        self.revision
            .validate()
            .map_err(|error| error.to_string())?;
        self.profile.validate().map_err(|error| error.to_string())?;
        if let Some(activation) = &self.activation {
            activation
                .lifecycle
                .validate()
                .map_err(|error| error.to_string())?;
            if !activation.config.is_object() {
                return Err("workflow instance config must be an object".into());
            }
            // ponytail: cap one confirmation at 32 bindings; raise only after measuring lock time.
            if activation.bindings.len() > 32 {
                return Err("workflow draft exceeds 32 trigger bindings".into());
            }
            let mut ids = std::collections::BTreeSet::new();
            for binding in &activation.bindings {
                binding.validate().map_err(|error| error.to_string())?;
                if binding.instance_id != activation.instance_id || !ids.insert(&binding.id) {
                    return Err(
                        "draft bindings must be unique and target its proposed instance".into(),
                    );
                }
                if let Some(branch) = &binding.branch_id {
                    if !self
                        .revision
                        .spec
                        .branches
                        .iter()
                        .any(|candidate| candidate.branch_id == branch.as_str())
                    {
                        return Err("draft binding names an unknown branch".into());
                    }
                } else if self.revision.spec.branches.len() != 1 {
                    return Err("multi-branch draft bindings require branch_id".into());
                }
            }
        }
        if serde_json::to_vec(self)
            .map_err(|error| error.to_string())?
            .len()
            > 1_048_576
        {
            return Err("workflow draft exceeds 1 MiB".into());
        }
        Ok(())
    }
}
/// One irreversible decision on an immutable proposal.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WorkflowDraftStatus {
    /// Stored preview, with no published revision or active instance.
    Proposed,
    /// Publication and optional activation committed with the decision.
    Applied,
    /// Owner rejected the proposal.
    Rejected,
}
impl WorkflowDraftStatus {
    /// Stable storage and wire spelling.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Proposed => "proposed",
            Self::Applied => "applied",
            Self::Rejected => "rejected",
        }
    }
    /// Repeat the same decision or reject an attempted reversal.
    pub fn decide(self, accept: bool) -> Result<Self, String> {
        let target = if accept {
            Self::Applied
        } else {
            Self::Rejected
        };
        if self != Self::Proposed && self != target {
            return Err("workflow draft already decided".into());
        }
        Ok(target)
    }
}
/// Owner-scoped durable preview and its decision.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct WorkflowDraft {
    /// Stable UUID for proposal idempotency.
    pub draft_id: WorkflowDraftId,
    /// Immutable preview.
    pub proposal: WorkflowProposal,
    /// Current decision.
    pub status: WorkflowDraftStatus,
    /// Time the preview was created.
    pub created_at: DateTime<Utc>,
    /// Time the irreversible decision committed.
    pub decided_at: Option<DateTime<Utc>>,
}

#[cfg(test)]
mod tests {
    use super::*;
    fn proposal() -> WorkflowProposal {
        serde_json::from_value(serde_json::json!({
            "name":"Preview", "revision": {
                "definition_id":"definition", "revision":1, "content_digest":"revision",
                "kernel_abi_version":"1", "dependency_set_digest":"deps",
                "spec":{"spec_id":"spec", "version":"1", "branches":[{
                    "branch_id":"root", "nodes":[{"id":"in", "type":"ingress.event", "config":{}}], "edges":[]
                }]}
            },
            "profile":{"id":"profile", "revision":1,"content_digest":"profile", "mode":"paper", "durability_grade":"standard",
                "trigger_provider":"events", "data_provider":"reference", "clock_model":"database", "action_provider":"actions"},
            "activation":{"instance_id":"instance", "lifecycle":crate::LifecyclePolicy::run_once(), "config":{},"bindings":[{
                "id":"binding","revision":1,"source":"source","event_type":"event","instance_id":"instance",
                "branch_id":"root","predicate":{},"ordering":"commutative","gap_wait_ms":30000,"gap_limit":10
            }]}
        })).unwrap()
    }
    #[test]
    fn proposals_validate_bounds_and_exact_binding_targets() {
        let valid = proposal();
        assert!(valid.validate().is_ok());
        let mut changed = valid.clone();
        changed.activation = None;
        assert!(changed.validate().is_ok());
        for name in [" ".into(), "字".repeat(257)] {
            changed.name = name;
            assert!(changed.validate().is_err());
        }
        let mut changed = valid.clone();
        changed.revision.content_digest.clear();
        assert!(changed.validate().is_err());
        let mut changed = valid.clone();
        changed.profile.action_provider.clear();
        assert!(changed.validate().is_err());
        let mut changed = valid.clone();
        changed.activation.as_mut().unwrap().config = Value::Null;
        assert!(changed.validate().is_err());
        let mut changed = valid.clone();
        let binding = &mut changed.activation.as_mut().unwrap().bindings[0];
        binding.branch_id = None;
        assert!(changed.validate().is_ok());
        changed
            .revision
            .spec
            .branches
            .push(changed.revision.spec.branches[0].clone());
        changed.revision.spec.branches[1].branch_id = "second".into();
        assert!(changed.validate().is_err());
        let mut changed = valid.clone();
        changed.activation.as_mut().unwrap().bindings[0].branch_id =
            Some("missing".parse().unwrap());
        assert!(changed.validate().is_err());
        let mut changed = valid.clone();
        changed.activation.as_mut().unwrap().bindings[0].instance_id = "other".parse().unwrap();
        assert!(changed.validate().is_err());
        let mut changed = valid.clone();
        changed.activation.as_mut().unwrap().bindings =
            vec![valid.activation.as_ref().unwrap().bindings[0].clone(); 2];
        assert!(changed.validate().is_err());
        changed.activation.as_mut().unwrap().bindings =
            vec![valid.activation.as_ref().unwrap().bindings[0].clone(); 33];
        assert!(changed.validate().is_err());
        let mut changed = valid.clone();
        changed.activation.as_mut().unwrap().bindings[0]
            .source
            .clear();
        assert!(changed.validate().is_err());
        let mut changed = valid.clone();
        changed.activation.as_mut().unwrap().config =
            serde_json::json!({"oversized":"x".repeat(1_048_576)});
        assert!(changed.validate().is_err());
        let mut untrusted = serde_json::to_value(valid).unwrap();
        untrusted["subject_id"] = "other".into();
        assert!(serde_json::from_value::<WorkflowProposal>(untrusted).is_err());
    }
    #[test]
    fn decision_is_irreversible_and_wire_status_is_stable() {
        use WorkflowDraftStatus::*;
        for (state, spelling) in [
            (Proposed, "proposed"),
            (Applied, "applied"),
            (Rejected, "rejected"),
        ] {
            assert_eq!(state.as_str(), spelling);
            assert_eq!(serde_json::to_value(state).unwrap(), spelling);
        }
        assert_eq!(Proposed.decide(true).unwrap(), Applied);
        assert_eq!(Proposed.decide(false).unwrap(), Rejected);
        assert_eq!(Applied.decide(true).unwrap(), Applied);
        assert_eq!(Rejected.decide(false).unwrap(), Rejected);
        assert!(Applied.decide(false).is_err());
        assert!(Rejected.decide(true).is_err());
    }
}