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>,
    /// The mail a record's arrival in a state sends (see `mail.rs`):
    /// declared on the state in aggregates.yaml, next to the event
    /// that enters it, so content decides when a person hears
    /// something and the mailer never has to know a site.
    pub mail_for_state: HashMap<String, MailSpec>,
    /// What a state grants the person whose record entered it: a group
    /// they become a member of. A firm approved to file is a member of
    /// the approved firms from that moment, and the decision that made
    /// it so is the same one the desk already pressed - nobody should
    /// have to remember to invite them afterwards.
    pub grants_for_state: HashMap<String, String>,
    /// The move a clock makes rather than a person. A hearing that gets
    /// no answer in six weeks has to go on without one, and nothing in
    /// a building is going to press that button.
    pub after_for_state: HashMap<String, AfterSpec>,
    /// How long a record stays in the list once it has entered a
    /// state, when that is not for ever: a card taken off the front
    /// page leaves the desk's list a little later, instead of standing
    /// there greyed out. The event log keeps it.
    pub leaves_for_state: HashMap<String, LeavesSpec>,
}

/// `leaves_after: { minutes: 10 }` on a state: how long a record that
/// has entered it stays in its bucket's list. Days, hours and minutes
/// add up; the sweep that takes it out runs every minute.
#[derive(Clone, Debug, Default, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct LeavesSpec {
    #[serde(default, skip_serializing_if = "is_zero")]
    pub days: u32,
    #[serde(default, skip_serializing_if = "is_zero")]
    pub hours: u32,
    #[serde(default, skip_serializing_if = "is_zero")]
    pub minutes: u32,
}

fn is_zero(n: &u32) -> bool {
    *n == 0
}

impl LeavesSpec {
    pub fn millis(&self) -> i64 {
        ((i64::from(self.days) * 24 + i64::from(self.hours)) * 60 + i64::from(self.minutes)) * 60 * 1000
    }
}

/// A mail to send when a record enters a state. `to` is `submitter`
/// (the record's `email` field, which the lint then requires the
/// forms to collect), one fixed address, or a field of the record
/// this one belongs to: `{notice.email}` when the form has a
/// `type: record` field called `notice`. That is how an answer to a
/// notice reaches whoever posted it. Subject and body are templates:
/// `{field}` for any field of the record, `{notice.field}` for one of
/// the record it belongs to, plus `{site}` (the instance's public
/// URL), `{chain}` (the record id), `{email}`, `{bucket}` and
/// `{state}`. The body is markdown; the mailer sends it as plain text
/// with an HTML alternative.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct MailSpec {
    pub to: String,
    pub subject: String,
    pub body: String,
}

impl MailSpec {
    pub fn to_submitter(&self) -> bool {
        self.to == "submitter"
    }
}

/// A move made by a clock: `after: { days: 42, to: no_answer }` on the
/// state a record is waiting in. The wait is measured from when the
/// record entered that state, and `to` must be an edge the same schema
/// already declares - a deadline may only do what a person could have
/// done. Days, because every deadline a public body keeps is in days
/// and a shorter unit would invite this to be used as a job queue.
#[derive(Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
#[serde(deny_unknown_fields)]
pub struct AfterSpec {
    pub days: u32,
    pub to: String,
}

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

    pub fn mail_for(&self, state: &str) -> Option<&MailSpec> {
        self.mail_for_state.get(state)
    }

    pub fn grants_for(&self, state: &str) -> Option<&str> {
        self.grants_for_state.get(state).map(String::as_str)
    }

    pub fn after_for(&self, state: &str) -> Option<&AfterSpec> {
        self.after_for_state.get(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,
    #[serde(default)]
    mail: Option<MailSpec>,
    #[serde(default)]
    grants: Option<String>,
    #[serde(default)]
    after: Option<AfterSpec>,
    #[serde(default)]
    leaves_after: Option<LeavesSpec>,
}

#[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"
                    );
                }
            }
        }

        let mut mail_for_state = HashMap::new();
        let mut grants_for_state = HashMap::new();
        let mut after_for_state = HashMap::new();
        let mut leaves_for_state = HashMap::new();
        for (state, raw_state) in raw_schema.states {
            if let Some(leaves) = raw_state.leaves_after {
                if leaves.millis() == 0 {
                    anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: leaves_after needs some days, hours or minutes");
                }
                leaves_for_state.insert(state.clone(), leaves);
            }
            if let Some(group) = raw_state.grants {
                if group.trim().is_empty() {
                    anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: grants needs a group");
                }
                grants_for_state.insert(state.clone(), group);
            }
            if let Some(after) = raw_state.after {
                if after.days == 0 {
                    anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: after.days must be at least one");
                }
                // A clock may only make a move a person could have
                // made, so it is held to the same graph.
                if !raw_schema.transitions.get(&state).is_some_and(|tos| tos.contains(&after.to)) {
                    anyhow::bail!(
                        "aggregates.yaml: bucket {bucket:?}: state {state:?}: after.to {:?} is not a declared transition from it",
                        after.to
                    );
                }
                after_for_state.insert(state.clone(), after);
            }
            if let Some(mail) = raw_state.mail {
                if mail.subject.trim().is_empty() || mail.body.trim().is_empty() {
                    anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: mail needs a subject and a body");
                }
                let through_a_record = crate::mail::parent_of(&mail.to).is_some();
                if !mail.to_submitter() && !through_a_record && !(mail.to.contains('@') && !mail.to.contains(char::is_whitespace)) {
                    anyhow::bail!(
                        "aggregates.yaml: bucket {bucket:?}: state {state:?}: mail.to must be `submitter`, one address, or a field of the record this one belongs to like `{{notice.email}}`, not {:?}",
                        mail.to
                    );
                }
                mail_for_state.insert(state, mail);
            }
        }

        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,
                mail_for_state,
                grants_for_state,
                after_for_state,
                leaves_for_state,
            },
        );
    }
    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()
    }

    #[test]
    fn a_state_may_carry_the_mail_its_arrival_sends() {
        let schemas = parse_aggregates_yaml(
            r#"
aggregates:
  - bucket: reports
    initial: open
    states:
      open: { event: received, mail: { to: submitter, subject: "Got it, {name}", body: "We read it. {site}/decide?chain={chain}" } }
      fixed: { event: fixed, mail: { to: board@example.no, subject: "Fixed: {where}", body: "Done." } }
      dropped: { event: dropped }
    transitions:
      open: [fixed, dropped]
"#,
        )
        .unwrap();
        let s = &schemas["reports"];
        assert!(s.mail_for("open").unwrap().to_submitter());
        assert_eq!(s.mail_for("fixed").unwrap().to, "board@example.no");
        assert!(s.mail_for("dropped").is_none());

        for bad in [
            "mail: { to: submitter, subject: \"\", body: b }",
            "mail: { to: \"the board\", subject: s, body: b }",
            "mail: { to: submitter, subject: s, body: b, cc: x }",
        ] {
            let yaml = format!("aggregates:\n  - bucket: r\n    initial: open\n    states:\n      open: {{ event: received, {bad} }}\n");
            assert!(parse_aggregates_yaml(&yaml).is_err(), "{bad}");
        }
    }

    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);
            }
        }
    }
}