portal/src/aggregates/mod.rs

//! Generic pure-replay + CAS-write machinery for event-sourced
//! aggregates - the state graph itself (what states exist, what
//! event-type string represents each, what transitions are legal) is
//! loaded at runtime from `questions/aggregates.yaml`, not compiled
//! Rust. Content declares the graph; this module only knows how to
//! replay/append against whatever graph it's handed
//! (`content::load_aggregates_from_gitea`, hot-swapped in
//! `AppState.aggregates` the same way `AppState.questions` is).
//!
//! `AggregateSchema.bucket` doubles as both the KV bucket name
//! (`answers.rs`'s read-model projection) and the JetStream
//! event-subject segment (`events::store`'s `events.<bucket>.<id>`) -
//! one name, not two, since content only ever had one name for this to
//! begin with.
#![cfg(feature = "ssr")]

use std::collections::HashMap;

use crate::events::store::{append_event, load_events, Event};

/// One aggregate type's full state graph, already validated (see
/// `parse_aggregates_yaml`) - `initial` is guaranteed to be a key of
/// `event_for_state`/`state_for_event`, every `transitions` key and
/// target is guaranteed to be a declared state, and every state's
/// event-type string is guaranteed unique within this schema.
#[derive(Clone, Debug)]
pub struct AggregateSchema {
    pub bucket: String,
    pub initial: String,
    pub event_for_state: HashMap<String, String>,
    pub state_for_event: HashMap<String, String>,
    pub transitions: HashMap<String, Vec<String>>,
    /// Names what consumes this bucket's answers when no page in the
    /// content repo reads it (e.g. "n8n newsletter compose").
    /// Free-text - it exists so `validate_questions`' attended-bucket
    /// rule has an explicit, auditable opt-out instead of a silent
    /// one, and so the next reader knows where the answers go.
    pub attended_by: Option<String>,
}

impl AggregateSchema {
    pub fn has_state(&self, state: &str) -> bool {
        self.event_for_state.contains_key(state)
    }

    pub fn allowed(&self, state: &str) -> &[String] {
        self.transitions
            .get(state)
            .map(|v| v.as_slice())
            .unwrap_or(&[])
    }
}

#[derive(Debug, serde::Deserialize)]
struct RawState {
    event: String,
}

#[derive(Debug, serde::Deserialize)]
struct RawAggregateSchema {
    bucket: String,
    initial: String,
    states: HashMap<String, RawState>,
    #[serde(default)]
    transitions: HashMap<String, Vec<String>>,
    #[serde(default)]
    attended_by: Option<String>,
}

#[derive(Debug, serde::Deserialize)]
struct AggregatesFile {
    aggregates: Vec<RawAggregateSchema>,
}

/// Parses and validates `aggregates.yaml`'s full contents. A malformed
/// entry (an `initial`/`transitions` key or target that isn't a
/// declared state, a duplicate `bucket`, or two states sharing one
/// event-type string) is rejected here with a clear message - the same
/// "bad content never silently corrupts a read/write" policy
/// `content::validate_questions` already applies to pages.
pub fn parse_aggregates_yaml(raw: &str) -> anyhow::Result<HashMap<String, AggregateSchema>> {
    let file: AggregatesFile =
        serde_yaml::from_str(raw).map_err(|e| anyhow::anyhow!("parsing aggregates.yaml: {e}"))?;

    let mut out = HashMap::new();
    for raw_schema in file.aggregates {
        let bucket = raw_schema.bucket.clone();
        if out.contains_key(&bucket) {
            anyhow::bail!("aggregates.yaml: bucket {bucket:?} declared more than once");
        }
        if !raw_schema.states.contains_key(&raw_schema.initial) {
            anyhow::bail!(
                "aggregates.yaml: bucket {bucket:?}: initial state {:?} is not declared in states",
                raw_schema.initial
            );
        }

        let mut event_for_state = HashMap::new();
        let mut state_for_event = HashMap::new();
        for (state, raw_state) in &raw_schema.states {
            if let Some(existing) = state_for_event.insert(raw_state.event.clone(), state.clone())
            {
                anyhow::bail!(
                    "aggregates.yaml: bucket {bucket:?}: event {:?} used by both state {:?} and {:?}",
                    raw_state.event, existing, state
                );
            }
            event_for_state.insert(state.clone(), raw_state.event.clone());
        }

        for (from, targets) in &raw_schema.transitions {
            if !raw_schema.states.contains_key(from) {
                anyhow::bail!(
                    "aggregates.yaml: bucket {bucket:?}: transitions declared from undeclared state {from:?}"
                );
            }
            for to in targets {
                if !raw_schema.states.contains_key(to) {
                    anyhow::bail!(
                        "aggregates.yaml: bucket {bucket:?}: transition {from:?} -> {to:?} targets an undeclared state"
                    );
                }
            }
        }

        out.insert(
            bucket.clone(),
            AggregateSchema {
                bucket,
                initial: raw_schema.initial,
                event_for_state,
                state_for_event,
                transitions: raw_schema.transitions,
                attended_by: raw_schema.attended_by,
            },
        );
    }
    Ok(out)
}

/// One aggregate's current view, replayed from its event history.
#[derive(Clone, Debug)]
pub struct Aggregate {
    pub id: String,
    pub state: String,
    pub payload: serde_json::Value,
    pub sequence: u64,
}

/// Pure, no IO, never panics. An event whose type isn't declared in
/// `schema`, or that doesn't fit the transition table from wherever
/// replay has gotten to so far, is silently skipped - not an error,
/// not even the *first* recognized event has anything to be "from" yet
/// so it's always accepted. This lenience (not enforcing the
/// transition table during replay, only at write time - see
/// `transition` below) is deliberate: a corrupted or out-of-order
/// event never crashes a read, it's just ignored.
pub fn replay(schema: &AggregateSchema, id: &str, events: &[Event]) -> Option<Aggregate> {
    let mut current: Option<Aggregate> = None;
    for event in events {
        let Some(next) = schema.state_for_event.get(&event.event_type) else {
            continue;
        };
        let accept = match &current {
            None => true,
            Some(agg) => *next == agg.state || schema.allowed(&agg.state).iter().any(|s| s == next),
        };
        if !accept {
            continue;
        }
        current = Some(Aggregate {
            id: id.to_string(),
            state: next.clone(),
            payload: event.payload.clone(),
            sequence: event.sequence,
        });
    }
    current
}

#[derive(Debug, Clone)]
pub enum TransitionError {
    /// No creation event found for this id - nothing to transition.
    UnknownAggregate,
    DisallowedTransition { from: String, to: String },
    /// A concurrent writer advanced this aggregate past the sequence
    /// this call read (see `events::store::append_event`'s CAS).
    Conflict,
    Store(String),
}

impl std::fmt::Display for TransitionError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::UnknownAggregate => write!(f, "unknown item"),
            Self::DisallowedTransition { from, to } => {
                write!(f, "cannot go from {from} to {to}")
            }
            Self::Conflict => write!(f, "item was just changed by someone else, try again"),
            Self::Store(e) => write!(f, "storage error: {e}"),
        }
    }
}

fn classify_publish_err(e: async_nats::jetstream::context::PublishError) -> TransitionError {
    use async_nats::jetstream::context::PublishErrorKind;
    match e.kind() {
        PublishErrorKind::WrongLastSequence | PublishErrorKind::WrongLastMessageId => {
            TransitionError::Conflict
        }
        _ => TransitionError::Store(e.to_string()),
    }
}

/// Appends the creation event for a brand-new aggregate. `expected_seq:
/// Some(0)` (no message yet on this id's subject) doubles as a
/// create-only-once guard - a double-submit race on the same id fails
/// the same way a disallowed transition does, rather than silently
/// overwriting.
pub async fn create(
    js: &async_nats::jetstream::Context,
    schema: &AggregateSchema,
    id: &str,
    payload: serde_json::Value,
    occurred_at_ms: i64,
) -> Result<Aggregate, TransitionError> {
    let event_type = schema
        .event_for_state
        .get(&schema.initial)
        .expect("schema.initial is validated to have an event mapping");
    let ack = append_event(
        js,
        &schema.bucket,
        id,
        event_type,
        payload.clone(),
        occurred_at_ms,
        Some(0),
    )
    .await
    .map_err(classify_publish_err)?;
    Ok(Aggregate {
        id: id.to_string(),
        state: schema.initial.clone(),
        payload,
        sequence: ack.sequence,
    })
}

/// Reconstructs a minimal one-event history for an aggregate whose
/// event log is empty but whose KV projection still knows its current
/// state - what makes wiping the EVENTS stream a recoverable
/// operation rather than one that strands every pre-wipe record in an
/// untransitionable limbo. `expected_seq: Some(0)` guards two racing
/// reseeds the same way `create` guards a double submit.
pub async fn reseed(
    js: &async_nats::jetstream::Context,
    schema: &AggregateSchema,
    id: &str,
    state: &str,
    occurred_at_ms: i64,
) -> Result<(), TransitionError> {
    let event_type = schema.event_for_state.get(state).ok_or_else(|| {
        TransitionError::Store(format!(
            "cannot reseed {id}: state {state:?} is not declared for bucket {:?}",
            schema.bucket
        ))
    })?;
    append_event(
        js,
        &schema.bucket,
        id,
        event_type,
        serde_json::json!({ "reseeded_from_projection": true }),
        occurred_at_ms,
        Some(0),
    )
    .await
    .map_err(classify_publish_err)?;
    Ok(())
}

/// Loads and replays `id`'s current state, then - if `target` is
/// actually reachable from it - appends the corresponding event with
/// the aggregate's just-replayed sequence as the CAS guard.
pub async fn transition(
    js: &async_nats::jetstream::Context,
    schema: &AggregateSchema,
    id: &str,
    target: &str,
    payload: serde_json::Value,
    occurred_at_ms: i64,
) -> Result<Aggregate, TransitionError> {
    let events = load_events(js, &schema.bucket, id)
        .await
        .map_err(|e| TransitionError::Store(e.to_string()))?;
    let current = replay(schema, id, &events).ok_or(TransitionError::UnknownAggregate)?;
    if !schema.allowed(&current.state).iter().any(|s| s == target) {
        return Err(TransitionError::DisallowedTransition {
            from: current.state.clone(),
            to: target.to_string(),
        });
    }
    // Validated by parse_aggregates_yaml: every transitions target is a
    // declared state, so this is always Some given the check above.
    let event_type = schema
        .event_for_state
        .get(target)
        .expect("transition target is validated to be a declared state");
    let ack = append_event(
        js,
        &schema.bucket,
        id,
        event_type,
        payload.clone(),
        occurred_at_ms,
        Some(current.sequence),
    )
    .await
    .map_err(classify_publish_err)?;
    Ok(Aggregate {
        id: id.to_string(),
        state: target.to_string(),
        payload,
        sequence: ack.sequence,
    })
}

#[cfg(test)]
mod tests {
    use super::*;

    fn test_schema() -> AggregateSchema {
        parse_aggregates_yaml(
            r#"
aggregates:
  - bucket: test
    initial: open
    states:
      open: { event: opened }
      a: { event: a }
      b: { event: b }
    transitions:
      open: [a, b]
"#,
        )
        .unwrap()
        .remove("test")
        .unwrap()
    }

    fn event(seq: u64, event_type: &str) -> Event {
        Event {
            id: uuid::Uuid::new_v4(),
            aggregate_type: "test".to_string(),
            aggregate_id: "x".to_string(),
            event_type: event_type.to_string(),
            payload: serde_json::json!({}),
            occurred_at_ms: 0,
            sequence: seq,
        }
    }

    #[test]
    fn empty_history_is_no_aggregate() {
        let schema = test_schema();
        assert!(replay(&schema, "x", &[]).is_none());
    }

    #[test]
    fn valid_path_replays() {
        let schema = test_schema();
        let events = vec![event(1, "opened"), event(2, "a")];
        let agg = replay(&schema, "x", &events).unwrap();
        assert_eq!(agg.state, "a");
        assert_eq!(agg.sequence, 2);
    }

    #[test]
    fn disallowed_edge_is_skipped_not_fatal() {
        // "a" then "b" - b isn't reachable from a, so the second event
        // is silently ignored, not an error.
        let schema = test_schema();
        let events = vec![event(1, "opened"), event(2, "a"), event(3, "b")];
        let agg = replay(&schema, "x", &events).unwrap();
        assert_eq!(agg.state, "a");
    }

    #[test]
    fn unrecognized_event_type_is_skipped() {
        let schema = test_schema();
        let events = vec![event(1, "opened"), event(2, "bogus"), event(3, "a")];
        let agg = replay(&schema, "x", &events).unwrap();
        assert_eq!(agg.state, "a");
    }

    #[test]
    fn initial_state_not_declared_is_rejected() {
        let err = parse_aggregates_yaml(
            r#"
aggregates:
  - bucket: bad
    initial: missing
    states:
      open: { event: opened }
"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("initial state"));
    }

    #[test]
    fn transition_target_not_declared_is_rejected() {
        let err = parse_aggregates_yaml(
            r#"
aggregates:
  - bucket: bad
    initial: open
    states:
      open: { event: opened }
    transitions:
      open: [nonexistent]
"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("undeclared state"));
    }

    #[test]
    fn duplicate_event_type_is_rejected() {
        let err = parse_aggregates_yaml(
            r#"
aggregates:
  - bucket: bad
    initial: open
    states:
      open: { event: same }
      other: { event: same }
"#,
        )
        .unwrap_err();
        assert!(err.to_string().contains("used by both"));
    }

    proptest::proptest! {
        #[test]
        fn replay_never_panics_and_state_always_valid(
            types in proptest::collection::vec(
                proptest::sample::select(vec!["opened", "a", "b", "bogus"]),
                0..30,
            )
        ) {
            let schema = test_schema();
            let events: Vec<Event> = types
                .iter()
                .enumerate()
                .map(|(i, t)| event(i as u64, t))
                .collect();
            let result = replay(&schema, "x", &events);
            if let Some(agg) = result {
                assert!(schema.has_state(&agg.state));
                assert!(agg.sequence <= events.len() as u64);
            }
        }
    }
}