a3s-flow 0.4.0

use chrono::{DateTime, Utc};
use serde::Serialize;
use std::collections::BTreeSet;

use crate::error::{FlowError, Result};
use crate::model::{
    HookSnapshot, RetryPolicy, StepCommand, StepSnapshot, WaitSnapshot, WorkflowRunSnapshot,
    WorkflowSpec,
};

pub(super) fn validate_run_id(run_id: &str) -> Result<()> {
    if run_id.is_empty()
        || !run_id
            .chars()
            .all(|ch| ch.is_ascii_alphanumeric() || ch == '-' || ch == '_')
    {
        return Err(FlowError::InvalidRunId(run_id.to_string()));
    }
    Ok(())
}

pub(super) fn ensure_same_start(
    run_id: &str,
    snapshot: &WorkflowRunSnapshot,
    spec: &WorkflowSpec,
    input: &serde_json::Value,
) -> Result<()> {
    if snapshot.spec != *spec {
        return Err(FlowError::RunConflict {
            run_id: run_id.to_string(),
            reason: "workflow spec differs".to_string(),
        });
    }
    if snapshot.input != *input {
        return Err(FlowError::RunConflict {
            run_id: run_id.to_string(),
            reason: "workflow input differs".to_string(),
        });
    }
    Ok(())
}

pub(super) fn ensure_step_batch_valid(steps: &[StepCommand]) -> Result<()> {
    if steps.is_empty() {
        return Err(FlowError::InvalidTransition(
            "schedule_steps requires at least one step".to_string(),
        ));
    }

    let mut ids = BTreeSet::new();
    for step in steps {
        if step.step_id.trim().is_empty() {
            return Err(FlowError::InvalidTransition(
                "scheduled step id must not be empty".to_string(),
            ));
        }
        if !ids.insert(step.step_id.as_str()) {
            return Err(FlowError::InvalidTransition(format!(
                "schedule_steps contains duplicate step id {}",
                step.step_id
            )));
        }
    }

    Ok(())
}

pub(super) fn ensure_step_command_matches(
    run_id: &str,
    step: &StepSnapshot,
    step_name: &str,
    input: &serde_json::Value,
    retry: RetryPolicy,
) -> Result<()> {
    if step.step_name != step_name {
        return Err(FlowError::NonDeterministic {
            run_id: run_id.to_string(),
            reason: format!(
                "step {} name differs: {}",
                step.step_id,
                replay_diff(&step.step_name, &step_name.to_string())
            ),
        });
    }
    if step.input != *input {
        return Err(FlowError::NonDeterministic {
            run_id: run_id.to_string(),
            reason: format!(
                "step {} input differs: {}",
                step.step_id,
                replay_diff(&step.input, input)
            ),
        });
    }
    if step.retry != retry {
        return Err(FlowError::NonDeterministic {
            run_id: run_id.to_string(),
            reason: format!(
                "step {} retry policy differs: {}",
                step.step_id,
                replay_diff(&step.retry, &retry)
            ),
        });
    }
    Ok(())
}

pub(super) fn ensure_wait_command_matches(
    run_id: &str,
    wait: &WaitSnapshot,
    resume_at: DateTime<Utc>,
) -> Result<()> {
    if wait.resume_at != resume_at {
        return Err(FlowError::NonDeterministic {
            run_id: run_id.to_string(),
            reason: format!(
                "wait {} resume_at differs: {}",
                wait.wait_id,
                replay_diff(&wait.resume_at, &resume_at)
            ),
        });
    }
    Ok(())
}

pub(super) fn ensure_hook_command_matches(
    run_id: &str,
    hook: &HookSnapshot,
    token: &str,
    metadata: &serde_json::Value,
) -> Result<()> {
    if hook.token != token {
        return Err(FlowError::NonDeterministic {
            run_id: run_id.to_string(),
            reason: format!(
                "hook {} token differs: history token and replay token are different (values redacted)",
                hook.hook_id
            ),
        });
    }
    if hook.metadata != *metadata {
        return Err(FlowError::NonDeterministic {
            run_id: run_id.to_string(),
            reason: format!(
                "hook {} metadata differs: {}",
                hook.hook_id,
                replay_diff(&hook.metadata, metadata)
            ),
        });
    }
    Ok(())
}

fn replay_diff<T>(history: &T, replay: &T) -> String
where
    T: Serialize,
{
    format!(
        "history={}; replay={}",
        compact_json(history),
        compact_json(replay)
    )
}

fn compact_json<T>(value: &T) -> String
where
    T: Serialize,
{
    serde_json::to_string(value).unwrap_or_else(|err| format!("<failed to encode: {err}>"))
}

pub(super) fn is_event_conflict(err: &FlowError) -> bool {
    matches!(err, FlowError::EventConflict { .. })
}