bot-forge 1.0.2

//! Durable checkpoints used to recover or abandon interrupted installation runs.

use std::fs;
use std::path::{Path, PathBuf};

use serde::{Deserialize, Serialize};

use crate::error::ForgeError;
use crate::fsutil::{NamedFileLease, acquire_path_lock, atomic_write_file, read_to_string};
use crate::paths::journal_dir;
use crate::state::append_json_line;
use crate::util::{now_secs, valid_sha256, valid_storage_id};

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
/// Transaction phase persisted for one component.
pub(crate) enum JournalPhase {
    Prepared,
    Stored,
    Activating,
    Activated,
    Recorded,
    RolledBack,
    Abandoned,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
/// Recoverable state for one component in an installation run.
pub(crate) struct JournalCheckpoint {
    pub(crate) run_id: String,
    pub(crate) component: String,
    /// Missing only in checkpoints written before profile recording was introduced.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub(crate) profile: Option<String>,
    pub(crate) plan_hash: String,
    pub(crate) config_hash: String,
    pub(crate) phase: JournalPhase,
    pub(crate) artifact_id: Option<String>,
    pub(crate) previous_artifact_id: Option<String>,
    pub(crate) activation_paths: Vec<PathBuf>,
    pub(crate) updated_at: u64,
}

impl JournalCheckpoint {
    /// Create an in-memory checkpoint in the [`JournalPhase::Prepared`] phase.
    pub(crate) fn new(
        run_id: impl Into<String>,
        component: impl Into<String>,
        plan_hash: impl Into<String>,
        config_hash: impl Into<String>,
        profile: impl Into<String>,
    ) -> Self {
        Self {
            run_id: run_id.into(),
            component: component.into(),
            profile: Some(profile.into()),
            plan_hash: plan_hash.into(),
            config_hash: config_hash.into(),
            phase: JournalPhase::Prepared,
            artifact_id: None,
            previous_artifact_id: None,
            activation_paths: Vec::new(),
            updated_at: now_secs(),
        }
    }

    /// Persist an allowed phase transition.
    ///
    /// # Errors
    ///
    /// Returns [`ForgeError`] for an invalid transition or persistence failure.
    pub(crate) fn transition(&mut self, phase: JournalPhase) -> Result<(), ForgeError> {
        self.set_phase(phase)?;
        self.save()
    }

    fn set_phase(&mut self, phase: JournalPhase) -> Result<(), ForgeError> {
        if !valid_transition(self.phase, phase) {
            return Err(ForgeError::Config(format!(
                "journal phase cannot transition from {:?} to {:?}",
                self.phase, phase
            )));
        }
        self.phase = phase;
        self.updated_at = now_secs();
        Ok(())
    }

    /// Attach stored artifact and activation metadata, then persist the checkpoint.
    ///
    /// # Errors
    ///
    /// Returns [`ForgeError`] when validation, serialization, or persistence fails.
    pub(crate) fn update_artifacts(
        &mut self,
        artifact_id: impl Into<String>,
        previous_artifact_id: Option<String>,
        activation_paths: Vec<PathBuf>,
    ) -> Result<(), ForgeError> {
        self.artifact_id = Some(artifact_id.into());
        self.previous_artifact_id = previous_artifact_id;
        self.activation_paths = activation_paths;
        self.updated_at = now_secs();
        self.save()
    }

    /// Validate and atomically save the checkpoint, then append its history event.
    ///
    /// # Errors
    ///
    /// Returns [`ForgeError`] when checkpoint validation, serialization, or either write fails.
    pub(crate) fn save(&self) -> Result<(), ForgeError> {
        let _lease = acquire_run_lock(&self.run_id)?;
        self.save_unlocked()
    }

    fn save_unlocked(&self) -> Result<(), ForgeError> {
        validate_checkpoint(self)?;
        let path = self.path();
        let bytes = serde_json::to_vec_pretty(self).map_err(|error| {
            ForgeError::Parse(format!("failed to serialize journal checkpoint: {error}"))
        })?;
        atomic_write_file(&path, &bytes)?;
        append_json_line(&self.history_path(), self)
    }

    /// Return the canonical JSON checkpoint path for this run and component.
    pub(crate) fn path(&self) -> PathBuf {
        journal_dir()
            .join(&self.run_id)
            .join(format!("{}.json", self.component))
    }

    fn history_path(&self) -> PathBuf {
        journal_dir().join(&self.run_id).join("events.jsonl")
    }
}

/// Load non-terminal checkpoints in stable run/component order.
///
/// # Errors
///
/// Returns [`ForgeError`] when journal directories or checkpoint files cannot be read or parsed.
pub(crate) fn load_pending() -> Result<Vec<JournalCheckpoint>, ForgeError> {
    let root = journal_dir();
    if !root.is_dir() {
        return Ok(Vec::new());
    }
    let mut pending = Vec::new();
    collect_checkpoints(&root, &mut pending)?;
    pending.retain(|checkpoint| {
        !matches!(
            checkpoint.phase,
            JournalPhase::Recorded | JournalPhase::RolledBack | JournalPhase::Abandoned
        )
    });
    pending.sort_by(|left, right| {
        (&left.run_id, &left.component).cmp(&(&right.run_id, &right.component))
    });
    Ok(pending)
}

/// Return run IDs that still contain non-terminal checkpoints.
///
/// # Errors
///
/// Returns [`ForgeError`] when pending journals cannot be loaded.
pub(crate) fn runs() -> Result<Vec<String>, ForgeError> {
    let mut ids = std::collections::BTreeSet::new();
    for checkpoint in load_pending()? {
        ids.insert(checkpoint.run_id);
    }
    Ok(ids.into_iter().collect())
}

/// Mark every pending checkpoint for `run_id` as abandoned.
///
/// # Errors
///
/// Returns [`ForgeError`] when journals cannot be loaded or an updated checkpoint cannot be saved.
pub(crate) fn abandon(run_id: &str) -> Result<usize, ForgeError> {
    if !valid_storage_id(run_id) {
        return Err(ForgeError::Config(
            "journal run_id must be a safe single-segment identifier".to_string(),
        ));
    }
    if !journal_dir().join(run_id).is_dir() {
        return Ok(0);
    }
    let _lease = acquire_run_lock(run_id)?;
    let mut checkpoints = load_pending()?
        .into_iter()
        .filter(|checkpoint| checkpoint.run_id == run_id)
        .collect::<Vec<_>>();
    for checkpoint in &mut checkpoints {
        checkpoint.set_phase(JournalPhase::Abandoned)?;
        checkpoint.save_unlocked()?;
    }
    Ok(checkpoints.len())
}

fn acquire_run_lock(run_id: &str) -> Result<NamedFileLease, ForgeError> {
    acquire_path_lock(
        &journal_dir().join(run_id).join("journal.lock"),
        "journal run lock",
    )
}

fn collect_checkpoints(root: &Path, output: &mut Vec<JournalCheckpoint>) -> Result<(), ForgeError> {
    for entry in fs::read_dir(root).map_err(|source| ForgeError::Io {
        path: root.to_path_buf(),
        source,
    })? {
        let entry = entry.map_err(|source| ForgeError::Io {
            path: root.to_path_buf(),
            source,
        })?;
        let path = entry.path();
        let metadata = fs::symlink_metadata(&path).map_err(|source| ForgeError::Io {
            path: path.clone(),
            source,
        })?;
        if metadata.file_type().is_symlink() {
            return Err(ForgeError::Config(format!(
                "journal does not allow symbolic links: {}",
                path.display()
            )));
        }
        if path.is_dir() {
            collect_checkpoints(&path, output)?;
        } else if path.extension().and_then(|value| value.to_str()) == Some("json") {
            let checkpoint: JournalCheckpoint = serde_json::from_str(&read_to_string(&path)?)
                .map_err(|error| {
                    ForgeError::Parse(format!("journal {} is corrupted: {error}", path.display()))
                })?;
            validate_checkpoint(&checkpoint).map_err(|error| {
                ForgeError::Parse(format!(
                    "journal {} is semantically invalid: {error}",
                    path.display()
                ))
            })?;
            output.push(checkpoint);
        }
    }
    Ok(())
}

fn validate_checkpoint(checkpoint: &JournalCheckpoint) -> Result<(), ForgeError> {
    if checkpoint
        .profile
        .as_deref()
        .is_some_and(|profile| crate::model::Profile::parse(profile).is_none())
    {
        return Err(ForgeError::Config(
            "journal contains an invalid profile".into(),
        ));
    }
    if !valid_storage_id(&checkpoint.run_id) || !valid_storage_id(&checkpoint.component) {
        return Err(ForgeError::Config(
            "journal run_id and component must be safe single-segment identifiers".to_string(),
        ));
    }
    if !valid_sha256(&checkpoint.plan_hash) || !valid_sha256(&checkpoint.config_hash) {
        return Err(ForgeError::Config(
            "journal plan_hash and config_hash must be 64-character lowercase hexadecimal digests"
                .to_string(),
        ));
    }
    let artifacts_present = checkpoint.artifact_id.is_some();
    if matches!(
        checkpoint.phase,
        JournalPhase::Stored
            | JournalPhase::Activating
            | JournalPhase::Activated
            | JournalPhase::Recorded
    ) && (!artifacts_present || checkpoint.activation_paths.is_empty())
    {
        return Err(ForgeError::Config(format!(
            "journal {:?} phase is missing an artifact or activation path",
            checkpoint.phase
        )));
    }
    Ok(())
}

fn valid_transition(from: JournalPhase, to: JournalPhase) -> bool {
    matches!(
        (from, to),
        (JournalPhase::Prepared, JournalPhase::Stored)
            | (JournalPhase::Stored, JournalPhase::Activating)
            | (JournalPhase::Activating, JournalPhase::Activated)
            | (JournalPhase::Activating, JournalPhase::RolledBack)
            | (JournalPhase::Activated, JournalPhase::Recorded)
            | (JournalPhase::Activated, JournalPhase::RolledBack)
            | (
                JournalPhase::Prepared
                    | JournalPhase::Stored
                    | JournalPhase::Activating
                    | JournalPhase::Activated,
                JournalPhase::Abandoned
            )
    )
}

#[cfg(test)]
mod tests {
    use std::path::PathBuf;

    use crate::state::journal::{
        JournalCheckpoint, JournalPhase, valid_transition, validate_checkpoint,
    };

    #[test]
    fn journal_enforces_commit_point_state_machine() {
        let mut checkpoint =
            JournalCheckpoint::new("run", "component", hash('a'), hash('b'), "minimal");
        assert!(!valid_transition(checkpoint.phase, JournalPhase::Activated));
        checkpoint.phase = JournalPhase::Stored;
        assert!(valid_transition(checkpoint.phase, JournalPhase::Activating));
        checkpoint.phase = JournalPhase::Activated;
        assert!(valid_transition(checkpoint.phase, JournalPhase::Recorded));
    }

    #[test]
    fn checkpoint_is_strict_and_has_no_development_version_field() {
        let checkpoint =
            JournalCheckpoint::new("run", "component", hash('a'), hash('b'), "minimal");
        let encoded = serde_json::to_string(&checkpoint).unwrap();
        assert!(!encoded.contains("schema_version"));
        let stale = encoded.replacen('{', "{\"schema_version\":1,", 1);
        assert!(serde_json::from_str::<JournalCheckpoint>(&stale).is_err());
    }

    #[test]
    fn checkpoint_rejects_path_traversal_and_invalid_commit_metadata() {
        let mut checkpoint =
            JournalCheckpoint::new("../escape", "component", hash('a'), hash('b'), "minimal");
        assert!(validate_checkpoint(&checkpoint).is_err());
        checkpoint.run_id = "safe-run".into();
        checkpoint.phase = JournalPhase::Stored;
        assert!(validate_checkpoint(&checkpoint).is_err());
        checkpoint.artifact_id = Some("artifact".into());
        checkpoint
            .activation_paths
            .push(PathBuf::from("/managed/bin/demo"));
        assert!(validate_checkpoint(&checkpoint).is_ok());
    }

    fn hash(byte: char) -> String {
        std::iter::repeat_n(byte, 64).collect()
    }

    #[test]
    fn crash_recovery_matrix_has_only_explicit_terminal_paths() {
        let allowed = [
            (JournalPhase::Prepared, JournalPhase::Stored),
            (JournalPhase::Stored, JournalPhase::Activating),
            (JournalPhase::Activating, JournalPhase::Activated),
            (JournalPhase::Activating, JournalPhase::RolledBack),
            (JournalPhase::Activated, JournalPhase::Recorded),
            (JournalPhase::Activated, JournalPhase::RolledBack),
        ];
        for (from, to) in allowed {
            assert!(valid_transition(from, to), "missing {from:?} -> {to:?}");
        }
        for phase in [
            JournalPhase::Prepared,
            JournalPhase::Stored,
            JournalPhase::Activating,
            JournalPhase::Activated,
        ] {
            assert!(valid_transition(phase, JournalPhase::Abandoned));
        }
        assert!(!valid_transition(
            JournalPhase::Recorded,
            JournalPhase::Abandoned
        ));
        assert!(!valid_transition(
            JournalPhase::RolledBack,
            JournalPhase::Abandoned
        ));
        assert!(!valid_transition(
            JournalPhase::Prepared,
            JournalPhase::Activated
        ));
        assert!(!valid_transition(
            JournalPhase::Stored,
            JournalPhase::Recorded
        ));
        assert!(!valid_transition(
            JournalPhase::RolledBack,
            JournalPhase::Recorded
        ));
        assert!(!valid_transition(
            JournalPhase::Abandoned,
            JournalPhase::Recorded
        ));
    }
}