use chrono::{DateTime, Utc};
use meerkat_core::lifecycle::{InputId, RunId};
use crate::identifiers::IdempotencyKey;
use crate::input_state::{
InputAbandonReason, InputLifecycleState, InputTerminalOutcome, StoredInputState,
};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum InteractionSelector {
InputId(InputId),
IdempotencyKey(String),
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TerminalWitnessSource {
LiveRuntime,
DurableStore,
}
#[derive(Debug, Clone, PartialEq)]
pub struct InteractionTerminalReport {
pub input_id: InputId,
pub phase: InputLifecycleState,
pub terminal: Option<InputTerminalOutcome>,
pub resolving_run_id: Option<RunId>,
pub attempt_count: u32,
pub idempotency_key: Option<IdempotencyKey>,
pub updated_at: DateTime<Utc>,
}
#[derive(Debug, Clone, PartialEq)]
pub enum RunResolutionClass {
Consumed,
Abandoned { reason: InputAbandonReason },
Displaced,
}
#[derive(Debug, Clone, PartialEq)]
pub enum RunTerminalStatus {
InFlight,
Resolved { outcome: RunResolutionClass },
NoDurableWitness,
}
#[derive(Debug, Clone, PartialEq)]
pub struct RunTerminalReport {
pub run_id: RunId,
pub status: RunTerminalStatus,
pub witnesses: Vec<InteractionTerminalReport>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct Sourced<T> {
pub source: TerminalWitnessSource,
pub report: T,
}
#[must_use]
pub fn interaction_report(bundle: &StoredInputState) -> InteractionTerminalReport {
InteractionTerminalReport {
input_id: bundle.state.input_id.clone(),
phase: bundle.seed.phase,
terminal: bundle.seed.terminal_outcome.clone(),
resolving_run_id: bundle.seed.last_run_id.clone(),
attempt_count: bundle.seed.attempt_count,
idempotency_key: bundle.state.idempotency_key.clone(),
updated_at: bundle.state.updated_at,
}
}
#[must_use]
pub fn find_by_idempotency_key<'a>(
inputs: &'a [StoredInputState],
key: &str,
) -> Option<&'a StoredInputState> {
inputs.iter().find(|stored| {
stored
.state
.idempotency_key
.as_ref()
.is_some_and(|stored_key| stored_key.0 == key)
})
}
#[must_use]
pub fn evaluate_run(run_id: &RunId, inputs: &[StoredInputState]) -> RunTerminalReport {
let mut witnesses: Vec<&StoredInputState> = inputs
.iter()
.filter(|stored| stored.seed.last_run_id.as_ref() == Some(run_id))
.collect();
witnesses.sort_by(|a, b| {
let a_key = (
a.seed.admission_sequence.is_none(),
a.seed.admission_sequence,
);
let b_key = (
b.seed.admission_sequence.is_none(),
b.seed.admission_sequence,
);
a_key
.cmp(&b_key)
.then_with(|| a.state.input_id.0.cmp(&b.state.input_id.0))
});
let status = if witnesses.is_empty() {
RunTerminalStatus::NoDurableWitness
} else if witnesses
.iter()
.any(|stored| stored.seed.terminal_outcome.is_none())
{
RunTerminalStatus::InFlight
} else if witnesses.iter().any(|stored| {
matches!(
stored.seed.terminal_outcome,
Some(InputTerminalOutcome::Consumed)
)
}) {
RunTerminalStatus::Resolved {
outcome: RunResolutionClass::Consumed,
}
} else if let Some(reason) =
witnesses
.iter()
.find_map(|stored| match &stored.seed.terminal_outcome {
Some(InputTerminalOutcome::Abandoned { reason }) => Some(reason.clone()),
_ => None,
})
{
RunTerminalStatus::Resolved {
outcome: RunResolutionClass::Abandoned { reason },
}
} else {
RunTerminalStatus::Resolved {
outcome: RunResolutionClass::Displaced,
}
};
RunTerminalReport {
run_id: run_id.clone(),
status,
witnesses: witnesses.into_iter().map(interaction_report).collect(),
}
}
#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::expect_used)]
mod tests {
use super::*;
use crate::input_state::{InputState, InputStateSeed};
fn witness(
run_id: Option<&RunId>,
terminal: Option<InputTerminalOutcome>,
admission_sequence: Option<u64>,
idempotency_key: Option<&str>,
) -> StoredInputState {
let input_id = InputId::new();
let mut state = InputState::new_accepted(input_id.clone());
state.idempotency_key = idempotency_key.map(IdempotencyKey::new);
let phase = match &terminal {
None => InputLifecycleState::Staged,
Some(InputTerminalOutcome::Consumed) => InputLifecycleState::Consumed,
Some(InputTerminalOutcome::Superseded { .. }) => InputLifecycleState::Superseded,
Some(InputTerminalOutcome::Coalesced { .. }) => InputLifecycleState::Coalesced,
Some(InputTerminalOutcome::Abandoned { .. }) => InputLifecycleState::Abandoned,
};
StoredInputState {
state,
seed: InputStateSeed {
phase,
last_run_id: run_id.cloned(),
last_boundary_sequence: None,
admission_sequence,
terminal_outcome: terminal,
attempt_count: 1,
recovery_lane: None,
},
}
}
fn abandoned(reason: InputAbandonReason) -> Option<InputTerminalOutcome> {
Some(InputTerminalOutcome::Abandoned { reason })
}
#[test]
fn consumed_takes_precedence_over_abandoned_and_displaced() {
let run_id = RunId::new();
let inputs = vec![
witness(
Some(&run_id),
abandoned(InputAbandonReason::Cancelled),
Some(1),
None,
),
witness(
Some(&run_id),
Some(InputTerminalOutcome::Consumed),
Some(2),
None,
),
witness(
Some(&run_id),
Some(InputTerminalOutcome::Superseded {
superseded_by: InputId::new(),
}),
Some(3),
None,
),
];
let report = evaluate_run(&run_id, &inputs);
assert_eq!(
report.status,
RunTerminalStatus::Resolved {
outcome: RunResolutionClass::Consumed
}
);
assert_eq!(report.witnesses.len(), 3);
}
#[test]
fn abandoned_cause_is_first_abandoned_witness_in_admission_order() {
let run_id = RunId::new();
let inputs = vec![
witness(
Some(&run_id),
abandoned(InputAbandonReason::Stopped),
Some(9),
None,
),
witness(
Some(&run_id),
abandoned(InputAbandonReason::Cancelled),
Some(2),
None,
),
];
let report = evaluate_run(&run_id, &inputs);
assert_eq!(
report.status,
RunTerminalStatus::Resolved {
outcome: RunResolutionClass::Abandoned {
reason: InputAbandonReason::Cancelled
}
},
"the FIRST abandoned witness in admission order supplies the cause"
);
}
#[test]
fn all_superseded_or_coalesced_is_displaced() {
let run_id = RunId::new();
let inputs = vec![
witness(
Some(&run_id),
Some(InputTerminalOutcome::Superseded {
superseded_by: InputId::new(),
}),
Some(1),
None,
),
witness(
Some(&run_id),
Some(InputTerminalOutcome::Coalesced {
aggregate_id: InputId::new(),
}),
Some(2),
None,
),
];
let report = evaluate_run(&run_id, &inputs);
assert_eq!(
report.status,
RunTerminalStatus::Resolved {
outcome: RunResolutionClass::Displaced
}
);
}
#[test]
fn one_non_terminal_witness_is_in_flight() {
let run_id = RunId::new();
let inputs = vec![
witness(
Some(&run_id),
Some(InputTerminalOutcome::Consumed),
Some(1),
None,
),
witness(Some(&run_id), None, Some(2), None),
];
let report = evaluate_run(&run_id, &inputs);
assert_eq!(report.status, RunTerminalStatus::InFlight);
}
#[test]
fn empty_witness_set_is_no_durable_witness() {
let run_id = RunId::new();
let other_run = RunId::new();
let inputs = vec![witness(
Some(&other_run),
Some(InputTerminalOutcome::Consumed),
Some(1),
None,
)];
let report = evaluate_run(&run_id, &inputs);
assert_eq!(report.status, RunTerminalStatus::NoDurableWitness);
assert!(report.witnesses.is_empty());
}
#[test]
fn find_by_idempotency_key_is_exact_match_only() {
let run_id = RunId::new();
let inputs = vec![
witness(
Some(&run_id),
Some(InputTerminalOutcome::Consumed),
Some(1),
Some("interaction-1"),
),
witness(Some(&run_id), None, Some(2), None),
];
assert!(find_by_idempotency_key(&inputs, "interaction-1").is_some());
assert!(
find_by_idempotency_key(&inputs, "interaction").is_none(),
"prefix must not match"
);
assert!(
find_by_idempotency_key(&inputs, "interaction-12").is_none(),
"superstring must not match"
);
}
#[test]
fn interaction_report_projects_seed_and_shell_facts() {
let run_id = RunId::new();
let stored = witness(
Some(&run_id),
Some(InputTerminalOutcome::Consumed),
Some(7),
Some("key-7"),
);
let report = interaction_report(&stored);
assert_eq!(report.input_id, stored.state.input_id);
assert_eq!(report.phase, InputLifecycleState::Consumed);
assert_eq!(report.terminal, Some(InputTerminalOutcome::Consumed));
assert_eq!(report.resolving_run_id, Some(run_id));
assert_eq!(report.attempt_count, 1);
assert_eq!(report.idempotency_key, Some(IdempotencyKey::new("key-7")));
}
}