portal/src/aggregates/mod.rs

//! Generic pure-replay + CAS-write machinery shared by every aggregate
//! type (`applicant`, `subscriber`, `project`) - one small `AggregateKind`
//! impl per type supplies its own states/transition table/event-type
//! names, this module supplies the mechanics once. Mirrors dodrenett's
//! `lib/machines/*.ts` shape (a `Record<State, State[]>` table + a pure
//! `replay()`), but generalized across types instead of copy-pasted per
//! type, and with real optimistic concurrency on write (dodrenett had
//! none - see `crate::events::store`).
#![cfg(feature = "ssr")]

pub mod applicant;
pub mod organization;
pub mod project;
pub mod subscriber;

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

/// A state machine's shape: its states, its transition table, and how
/// states map to/from the event-log's `event_type` strings. Kept as a
/// trait (not a generic enum param alone) so each aggregate module can
/// own its own `State` enum and transition table as plain, readable
/// Rust, not a shared data structure everyone has to thread through.
pub trait AggregateKind: Copy + Clone + PartialEq + Eq + std::fmt::Debug + 'static {
    /// The bucket-name -> aggregate-type key used both for `events.rs`'s
    /// subject scheme and for `content.rs`'s load-time transition
    /// validation (see `content::aggregate_type_for_bucket`).
    const AGGREGATE_TYPE: &'static str;
    /// The state a brand-new aggregate starts in - also the state a
    /// creation event (see `create`) asserts.
    const INITIAL_STATE: Self;

    /// Maps an event-log `event_type` string to the state it represents
    /// - `None` for anything this aggregate type doesn't recognize
    /// (replay skips it, never errors).
    fn from_event_type(event_type: &str) -> Option<Self>;
    /// The event-log `event_type` string that represents transitioning
    /// *into* this state.
    fn event_type(self) -> &'static str;
    /// States reachable directly from this one. Checked at write time
    /// (`transition`); replay is deliberately more lenient (see below).
    fn allowed(self) -> &'static [Self];
}

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

/// Pure, no IO, never panics. An event 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 and matches dodrenett's proven
/// "self-healing against malformed history" policy: a corrupted or
/// out-of-order event never crashes a read, it's just ignored.
pub fn replay<K: AggregateKind>(id: &str, events: &[Event]) -> Option<Aggregate<K>> {
    let mut current: Option<Aggregate<K>> = None;
    for event in events {
        let Some(next) = K::from_event_type(&event.event_type) else {
            continue;
        };
        let accept = match &current {
            None => true,
            Some(agg) => next == agg.state || agg.state.allowed().contains(&next),
        };
        if !accept {
            continue;
        }
        current = Some(Aggregate {
            id: id.to_string(),
            state: next,
            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 already advanced this aggregate past the
    /// sequence this call read - the fix for the lost-update race
    /// `answers::transition_answer` used to have (see
    /// `crate::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<K: AggregateKind>(
    js: &async_nats::jetstream::Context,
    id: &str,
    payload: serde_json::Value,
    occurred_at_ms: i64,
) -> Result<Aggregate<K>, TransitionError> {
    let ack = append_event(
        js,
        K::AGGREGATE_TYPE,
        id,
        K::INITIAL_STATE.event_type(),
        payload.clone(),
        occurred_at_ms,
        Some(0),
    )
    .await
    .map_err(classify_publish_err)?;
    Ok(Aggregate {
        id: id.to_string(),
        state: K::INITIAL_STATE,
        payload,
        sequence: ack.sequence,
    })
}

/// 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<K: AggregateKind>(
    js: &async_nats::jetstream::Context,
    id: &str,
    target: K,
    payload: serde_json::Value,
    occurred_at_ms: i64,
) -> Result<Aggregate<K>, TransitionError> {
    let events = load_events(js, K::AGGREGATE_TYPE, id)
        .await
        .map_err(|e| TransitionError::Store(e.to_string()))?;
    let current = replay::<K>(id, &events).ok_or(TransitionError::UnknownAggregate)?;
    if !current.state.allowed().contains(&target) {
        return Err(TransitionError::DisallowedTransition {
            from: format!("{:?}", current.state),
            to: format!("{target:?}"),
        });
    }
    let ack = append_event(
        js,
        K::AGGREGATE_TYPE,
        id,
        target.event_type(),
        payload.clone(),
        occurred_at_ms,
        Some(current.sequence),
    )
    .await
    .map_err(classify_publish_err)?;
    Ok(Aggregate {
        id: id.to_string(),
        state: target,
        payload,
        sequence: ack.sequence,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::events::store::Event;

    #[derive(Copy, Clone, PartialEq, Eq, Debug)]
    enum TestState {
        Open,
        A,
        B,
    }

    impl AggregateKind for TestState {
        const AGGREGATE_TYPE: &'static str = "test";
        const INITIAL_STATE: Self = TestState::Open;

        fn from_event_type(event_type: &str) -> Option<Self> {
            match event_type {
                "opened" => Some(TestState::Open),
                "a" => Some(TestState::A),
                "b" => Some(TestState::B),
                _ => None,
            }
        }
        fn event_type(self) -> &'static str {
            match self {
                TestState::Open => "opened",
                TestState::A => "a",
                TestState::B => "b",
            }
        }
        fn allowed(self) -> &'static [Self] {
            match self {
                TestState::Open => &[TestState::A, TestState::B],
                TestState::A | TestState::B => &[],
            }
        }
    }

    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() {
        assert!(replay::<TestState>("x", &[]).is_none());
    }

    #[test]
    fn valid_path_replays() {
        let events = vec![event(1, "opened"), event(2, "a")];
        let agg = replay::<TestState>("x", &events).unwrap();
        assert_eq!(agg.state, TestState::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 events = vec![event(1, "opened"), event(2, "a"), event(3, "b")];
        let agg = replay::<TestState>("x", &events).unwrap();
        assert_eq!(agg.state, TestState::A);
    }

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

    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 events: Vec<Event> = types
                .iter()
                .enumerate()
                .map(|(i, t)| event(i as u64, t))
                .collect();
            let result = replay::<TestState>("x", &events);
            if let Some(agg) = result {
                // Whatever state we landed in must itself be a real
                // state this machine declares - trivially true here
                // since TestState::allowed() only ever returns TestState
                // values, but this is the shape the real aggregates'
                // proptests repeat: replaying arbitrary event-type noise
                // must never produce anything but a legitimate state.
                let _: TestState = agg.state;
                assert!(agg.sequence <= events.len() as u64);
            }
        }
    }
}