liminal-server 0.25.4

//! The bound builder's durable plan and retained change stream.

use super::site::{SiteError, SiteLimits};
use super::{Lifecycle, ServiceRecord};
use serde::{Deserialize, Serialize};

/// A step's closed status vocabulary.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum StepStatus {
    Todo,
    Doing,
    Done,
    Blocked,
}

/// One ordinal step; its note and title never enter diagnostics.
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Step {
    /// Stable one-based ordinal, preserved for finished steps.
    pub id: usize,
    /// The builder's words for the work.
    pub title: String,
    /// Its current state.
    pub status: StepStatus,
    /// Optional explanation carried as data only.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub note: Option<String>,
    /// Epoch milliseconds of this step's last change.
    pub at_ms: u64,
}

/// Every write retains the complete resulting step list with its actor.
#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PlanTransition {
    /// The admitted builder.
    pub holder: String,
    /// Epoch milliseconds of this write.
    pub at_ms: u64,
    /// The new revision.
    pub revision: u64,
    /// Complete steps at that revision.
    pub steps: Vec<Step>,
}

/// Current state and complete history, restored with the service.
#[derive(Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct Plan {
    /// Steps in ordinal order.
    pub steps: Vec<Step>,
    /// The sole doing step's id, or none.
    pub current: Option<usize>,
    /// Increases once for each accepted write.
    pub revision: u64,
    /// Every accepted write, oldest first.
    #[serde(default)]
    pub transitions: Vec<PlanTransition>,
}

/// Watchers receive the current plan; full history stays in the owning record.
#[must_use]
pub fn current_value(plan: &Plan) -> serde_json::Value {
    serde_json::json!({"steps":plan.steps,"current":plan.current,"revision":plan.revision})
}

/// A site completes with its plan finished: the plan is kept on the record after completion and
/// can no longer be written, so a step left todo, doing or blocked would stand unfinished for good.
///
/// # Errors
/// Names every step that is not done.
pub(crate) fn refuse_unfinished(plan: &Plan) -> Result<(), SiteError> {
    let open: Vec<String> = plan
        .steps
        .iter()
        .filter(|step| step.status != StepStatus::Done)
        .map(|step| step.id.to_string())
        .collect();
    if open.is_empty() {
        Ok(())
    } else {
        Err(SiteError::Refused(format!(
            "plan steps not done: {}; mark them done before completing",
            open.join(", ")
        )))
    }
}

fn text(value: &str, limits: &SiteLimits) -> Result<(), SiteError> {
    if value.is_empty() || value.len() > limits.text_bytes.get() || value.contains('\0') {
        Err(SiteError::Invalid("plan text"))
    } else {
        Ok(())
    }
}

impl Plan {
    /// Validates the complete stored plan before any retained value is served.
    ///
    /// # Errors
    /// A configured bound, invalid ordinal, conflicting current or broken history.
    pub fn validate(&self, limits: &SiteLimits) -> Result<(), SiteError> {
        if self.transitions.len() > limits.transitions.get() {
            return Err(SiteError::Bound("plan history"));
        }
        check_steps(&self.steps, limits)?;
        let current = self
            .steps
            .iter()
            .find(|step| step.status == StepStatus::Doing)
            .map(|step| step.id);
        if current != self.current
            || self.revision
                != u64::try_from(self.transitions.len()).map_err(|_| SiteError::Store)?
        {
            return Err(SiteError::Invalid("stored plan revision or current"));
        }
        let mut previous: &[Step] = &[];
        for (index, transition) in self.transitions.iter().enumerate() {
            if transition.revision != u64::try_from(index + 1).map_err(|_| SiteError::Store)?
                || transition.holder.is_empty()
                || transition.holder.len() > limits.text_bytes.get()
                || transition.holder.chars().any(char::is_control)
            {
                return Err(SiteError::Invalid("stored plan transition"));
            }
            check_steps(&transition.steps, limits)?;
            for done in previous
                .iter()
                .filter(|step| step.status == StepStatus::Done)
            {
                if transition
                    .steps
                    .get(done.id - 1)
                    .is_none_or(|next| next.title != done.title || next.status != StepStatus::Done)
                {
                    return Err(SiteError::Invalid("stored finished step changed"));
                }
            }
            previous = &transition.steps;
        }
        if previous != self.steps.as_slice() {
            return Err(SiteError::Invalid(
                "stored plan history differs from current",
            ));
        }
        Ok(())
    }
    /// Replaces the list without removing or renaming any finished ordinal.
    ///
    /// # Errors
    /// A configured bound, invalid title or finished step change.
    pub fn set(
        &mut self,
        titles: &[String],
        holder: &str,
        at_ms: u64,
        limits: &SiteLimits,
    ) -> Result<(), SiteError> {
        if titles.is_empty() || titles.len() > limits.steps.get() {
            return Err(SiteError::Bound("steps"));
        }
        for title in titles {
            text(title, limits)?;
        }
        for old in &self.steps {
            if old.status == StepStatus::Done && titles.get(old.id - 1) != Some(&old.title) {
                return Err(SiteError::Refused(format!(
                    "done step {} cannot be removed or renamed",
                    old.id
                )));
            }
        }
        let steps = titles
            .iter()
            .enumerate()
            .map(|(ordinal, title)| {
                let id = ordinal + 1;
                self.steps
                    .get(ordinal)
                    .filter(|old| old.title == *title)
                    .cloned()
                    .unwrap_or_else(|| Step {
                        id,
                        title: title.clone(),
                        status: StepStatus::Todo,
                        note: None,
                        at_ms,
                    })
            })
            .collect();
        self.write(steps, holder, at_ms, limits)
    }

    /// Changes one step, preserving the single doing step and terminal completion.
    ///
    /// # Errors
    /// Unknown ordinal, invalid note, another doing step or a configured bound.
    pub fn step(
        &mut self,
        id: usize,
        status: StepStatus,
        note: Option<String>,
        holder: &str,
        at_ms: u64,
        limits: &SiteLimits,
    ) -> Result<(), SiteError> {
        if let Some(note) = &note {
            text(note, limits)?;
        }
        if status == StepStatus::Doing && self.current.is_some_and(|current| current != id) {
            return Err(SiteError::Refused(format!(
                "step {} is already doing",
                self.current.ok_or(SiteError::Store)?
            )));
        }
        let mut steps = self.steps.clone();
        let step = steps
            .iter_mut()
            .find(|step| step.id == id)
            .ok_or_else(|| SiteError::Refused(format!("plan has no step {id}")))?;
        if step.status == StepStatus::Done && status != StepStatus::Done {
            return Err(SiteError::Refused(format!(
                "done step {id} cannot change status"
            )));
        }
        step.status = status;
        step.note = note;
        step.at_ms = at_ms;
        self.write(steps, holder, at_ms, limits)
    }

    fn write(
        &mut self,
        steps: Vec<Step>,
        holder: &str,
        at_ms: u64,
        limits: &SiteLimits,
    ) -> Result<(), SiteError> {
        if self.transitions.len() >= limits.transitions.get() {
            return Err(SiteError::Bound("transitions"));
        }
        let revision = self
            .revision
            .checked_add(1)
            .ok_or(SiteError::Bound("revision"))?;
        self.current = steps
            .iter()
            .find(|step| step.status == StepStatus::Doing)
            .map(|step| step.id);
        self.transitions.push(PlanTransition {
            holder: holder.into(),
            at_ms,
            revision,
            steps: steps.clone(),
        });
        self.steps = steps;
        self.revision = revision;
        Ok(())
    }
}

fn check_steps(steps: &[Step], limits: &SiteLimits) -> Result<(), SiteError> {
    if steps.len() > limits.steps.get() {
        return Err(SiteError::Bound("steps"));
    }
    let mut doing = false;
    for (ordinal, step) in steps.iter().enumerate() {
        if step.id != ordinal + 1 {
            return Err(SiteError::Invalid("stored plan ordinal"));
        }
        text(&step.title, limits)?;
        if let Some(note) = &step.note {
            text(note, limits)?;
        }
        if step.status == StepStatus::Doing {
            if doing {
                return Err(SiteError::Invalid("stored plan has two doing steps"));
            }
            doing = true;
        }
    }
    Ok(())
}

impl Lifecycle {
    /// Only the signed builder with an outright site plan right can reach writes.
    ///
    /// # Errors
    /// The named holder is not the currently bound builder.
    pub fn check_plan_holder(&self, record: &ServiceRecord, holder: &str) -> Result<(), SiteError> {
        if record.builder_bound && record.builder.as_deref() == Some(holder) {
            return Ok(());
        }
        Err(SiteError::Refused(format!(
            "pass holder {holder} is not the builder of {}",
            record.name
        )))
    }

    /// Writes one whole replacement after the caller's plan grant is decided.
    ///
    /// # Errors
    /// Wrong builder, non-building service, invalid list or durable write failure.
    pub(crate) fn plan_set(
        &self,
        service: &str,
        holder: &str,
        titles: &[String],
        at_ms: u64,
    ) -> Result<Plan, SiteError> {
        self.write_plan(service, holder, at_ms, |plan, limits| {
            plan.set(titles, holder, at_ms, limits)
        })
    }

    /// Writes one step after the caller's plan grant is decided.
    ///
    /// # Errors
    /// Wrong builder, non-building service, invalid transition or durable failure.
    pub(crate) fn plan_step(
        &self,
        service: &str,
        holder: &str,
        id: usize,
        status: StepStatus,
        note: Option<String>,
        at_ms: u64,
    ) -> Result<Plan, SiteError> {
        self.write_plan(service, holder, at_ms, |plan, limits| {
            plan.step(id, status, note, holder, at_ms, limits)
        })
    }

    fn write_plan(
        &self,
        service: &str,
        holder: &str,
        at_ms: u64,
        apply: impl FnOnce(&mut Plan, &SiteLimits) -> Result<(), SiteError>,
    ) -> Result<Plan, SiteError> {
        let limits = &self.site_bindings()?.limits;
        let mut registry = self.lock()?;
        let mut record = registry
            .records
            .get(service)
            .cloned()
            .ok_or_else(|| SiteError::Refused(format!("unknown_service: {service}")))?;
        self.check_plan_holder(&record, holder)?;
        if record.state != super::ServiceState::Building {
            return Err(SiteError::Invalid("plan requires building"));
        }
        if record.transitions.len() >= limits.transitions.get() {
            return Err(SiteError::Bound("transitions"));
        }
        apply(&mut record.plan, limits)?;
        if serde_json::to_vec(&current_value(&record.plan))
            .map_err(|_| SiteError::Store)?
            .len()
            > limits.request_bytes.get()
        {
            return Err(SiteError::Bound("plan retained bytes"));
        }
        Self::site_transition(&mut record, holder, at_ms, "plan written");
        self.keep(&mut registry, &record)?;
        self.publish_plan(&record)?;
        drop(registry);
        Ok(record.plan)
    }
}

#[cfg(test)]
#[path = "plan_tests.rs"]
mod tests;