Make aggregate state graphs content-driven, not compiled Rust

Six hand-maintained copies of the bucket->aggregate-type table
(4 per-type State enums, content::aggregate_type_for_bucket +
is_valid_transition_target, answers.rs's two dispatch matches,
backfill_events.rs's own migration table) collapse into one:
questions/aggregates.yaml, loaded and hot-swapped in AppState.aggregates
the same way AppState.questions already is. aggregates/mod.rs's
replay/create/transition now run off a runtime-loaded AggregateSchema
instead of a compile-time AggregateKind trait impl per type - the
CAS/JetStream mechanics underneath are unchanged. A new aggregate type
is now a content-only change, no portal deploy required.

backfill_events.rs is deleted outright rather than ported - its
migration job was already done and production data here is minimal
and expendable (the EVENTS stream gets purged by hand post-deploy).

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

src/content.rs

@@ -262,6 +262,91 @@ pub fn gitea_api_base(repo_url: &str) -> anyhow::Result<String> {
     ))
 }
 
+/// Splits a repo's normal browser URL into `(owner, repo)` - shared by
+/// every loader in this module that needs to build a Gitea contents API
+/// URL (`load_questions_from_gitea`, `load_aggregates_from_gitea`).
+#[cfg(feature = "ssr")]
+fn parse_owner_repo(repo_url: &str) -> anyhow::Result<(String, String)> {
+    let parsed = url::Url::parse(repo_url)
+        .map_err(|e| anyhow::anyhow!("parsing content repo url {repo_url}: {e}"))?;
+    let mut segments = parsed
+        .path_segments()
+        .ok_or_else(|| anyhow::anyhow!("no path in content repo url {repo_url}"))?;
+    let owner = segments
+        .next()
+        .filter(|s| !s.is_empty())
+        .ok_or_else(|| anyhow::anyhow!("missing owner in content repo url {repo_url}"))?
+        .to_string();
+    let repo = segments
+        .next()
+        .filter(|s| !s.is_empty())
+        .ok_or_else(|| anyhow::anyhow!("missing repo name in content repo url {repo_url}"))?
+        .to_string();
+    Ok((owner, repo))
+}
+
+/// Fetches one file from a Gitea repo's contents API and returns its raw
+/// text - the single-file counterpart to `load_questions_from_gitea`'s
+/// directory-listing loop, used by `load_aggregates_from_gitea` for the
+/// one `aggregates.yaml` file at the repo root.
+#[cfg(feature = "ssr")]
+async fn fetch_gitea_file(
+    client: &openidconnect::reqwest::Client,
+    api_base: &str,
+    owner: &str,
+    repo: &str,
+    branch: &str,
+    path: &str,
+) -> anyhow::Result<String> {
+    let meta_url = format!("{api_base}/api/v1/repos/{owner}/{repo}/contents/{path}?ref={branch}");
+    let meta_text = client
+        .get(&meta_url)
+        .send()
+        .await
+        .map_err(|e| anyhow::anyhow!("fetching {meta_url}: {e}"))?
+        .error_for_status()
+        .map_err(|e| anyhow::anyhow!("fetching {meta_url}: {e}"))?
+        .text()
+        .await
+        .map_err(|e| anyhow::anyhow!("reading contents response from {meta_url}: {e}"))?;
+    let meta: serde_json::Value = serde_json::from_str(&meta_text)
+        .map_err(|e| anyhow::anyhow!("parsing contents response from {meta_url}: {e}"))?;
+    let download_url = meta
+        .get("download_url")
+        .and_then(|v| v.as_str())
+        .ok_or_else(|| anyhow::anyhow!("no download_url for {path}"))?;
+    client
+        .get(download_url)
+        .send()
+        .await
+        .map_err(|e| anyhow::anyhow!("fetching {path}: {e}"))?
+        .error_for_status()
+        .map_err(|e| anyhow::anyhow!("fetching {path}: {e}"))?
+        .text()
+        .await
+        .map_err(|e| anyhow::anyhow!("reading {path}: {e}"))
+}
+
+/// Fetches and parses `aggregates.yaml` from a Gitea repo's root -
+/// sibling to the pages `subdir`, not inside it, so
+/// `load_questions_from_gitea`'s "every `*.yaml` under `subdir` is a
+/// page" directory-listing convention needs no special-case exclusion.
+/// Validation (well-formed states/transitions, no duplicate buckets or
+/// event-type strings) happens inside `aggregates::parse_aggregates_yaml`
+/// itself - a malformed file fails here, before ever reaching
+/// `AppState`.
+#[cfg(feature = "ssr")]
+pub async fn load_aggregates_from_gitea(
+    repo_url: &str,
+    branch: &str,
+) -> anyhow::Result<std::collections::HashMap<String, crate::aggregates::AggregateSchema>> {
+    let (owner, repo) = parse_owner_repo(repo_url)?;
+    let api_base = gitea_api_base(repo_url)?;
+    let client = openidconnect::reqwest::Client::new();
+    let raw = fetch_gitea_file(&client, &api_base, &owner, &repo, branch, "aggregates.yaml").await?;
+    crate::aggregates::parse_aggregates_yaml(&raw)
+}
+
 /// Fetches every `*.yaml` file under `subdir` in a Gitea repo as a
 /// `Question`, keyed by its own `id`. `repo_url` is the repo's normal
 /// browser URL (e.g. `https://project.uhhm.no/uhhm/questions`) - the
@@ -275,20 +360,8 @@ pub async fn load_questions_from_gitea(
     branch: &str,
     subdir: &str,
 ) -> anyhow::Result<std::collections::HashMap<String, Question>> {
-    let parsed = url::Url::parse(repo_url)
-        .map_err(|e| anyhow::anyhow!("parsing content repo url {repo_url}: {e}"))?;
+    let (owner, repo) = parse_owner_repo(repo_url)?;
     let api_base = gitea_api_base(repo_url)?;
-    let mut segments = parsed
-        .path_segments()
-        .ok_or_else(|| anyhow::anyhow!("no path in content repo url {repo_url}"))?;
-    let owner = segments
-        .next()
-        .filter(|s| !s.is_empty())
-        .ok_or_else(|| anyhow::anyhow!("missing owner in content repo url {repo_url}"))?;
-    let repo = segments
-        .next()
-        .filter(|s| !s.is_empty())
-        .ok_or_else(|| anyhow::anyhow!("missing repo name in content repo url {repo_url}"))?;
 
     let client = openidconnect::reqwest::Client::new();
     let list_url = format!("{api_base}/api/v1/repos/{owner}/{repo}/contents/{subdir}?ref={branch}");
@@ -332,58 +405,28 @@ pub async fn load_questions_from_gitea(
     Ok(out)
 }
 
-/// Maps a content-declared KV bucket name to the aggregate type
-/// (`crate::aggregates::*::State::AGGREGATE_TYPE`) it's event-sourced
-/// under - the fixed, three-entry table for this pass (see the
-/// event-sourcing plan). A bucket name not listed here is left alone
-/// entirely (no validation applied) - deliberate: only buckets actually
-/// migrated onto the generic aggregate machinery get their transitions
-/// checked.
-#[cfg(feature = "ssr")]
-pub fn aggregate_type_for_bucket(bucket: &str) -> Option<&'static str> {
-    match bucket {
-        "applicants" => Some("applicant"),
-        "subscribers" => Some("subscriber"),
-        "projects" => Some("project"),
-        "organizations" => Some("organization"),
-        _ => None,
-    }
-}
-
-#[cfg(feature = "ssr")]
-fn is_valid_transition_target(aggregate_type: &str, to: &str) -> bool {
-    use crate::aggregates::AggregateKind;
-    fn check<K: AggregateKind>(to: &str) -> bool {
-        K::INITIAL_STATE.allowed().iter().any(|s| s.event_type() == to)
-    }
-    match aggregate_type {
-        "applicant" => check::<crate::aggregates::applicant::State>(to),
-        "subscriber" => check::<crate::aggregates::subscriber::State>(to),
-        "project" => check::<crate::aggregates::project::State>(to),
-        "organization" => check::<crate::aggregates::organization::State>(to),
-        _ => false,
-    }
-}
-
 /// Validates every declared transition target (`SelfTransition.to`,
-/// `ResourceSpec.transitions[].to`) against the real compiled
-/// transition table for its bucket's aggregate type, for every bucket
-/// this pass actually migrated (see `aggregate_type_for_bucket`).
-/// Called on every content load/reload (`watch_for_reload`, `main.rs`'s
-/// boot path) and by the standalone `question-lint` binary, so a YAML
-/// typo becomes a caught, logged rejection instead of a silently
-/// -accepted, later-broken string.
+/// `ResourceSpec.transitions[].to`) against `aggregates` - the real
+/// state graph loaded from `aggregates.yaml` - for every bucket that
+/// has one declared. A bucket with no entry in `aggregates` is left
+/// alone entirely (no validation applied): not every bucket needs to
+/// be event-sourced to keep working. Called on every content
+/// load/reload (`watch_for_reload`, `main.rs`'s boot path) and by the
+/// standalone `question-lint` binary, so a YAML typo becomes a caught,
+/// logged rejection instead of a silently-accepted, later-broken
+/// string.
 #[cfg(feature = "ssr")]
 pub fn validate_questions(
     questions: &std::collections::HashMap<String, Question>,
+    aggregates: &std::collections::HashMap<String, crate::aggregates::AggregateSchema>,
 ) -> anyhow::Result<()> {
     for question in questions.values() {
         for alternative in &question.alternatives {
             if let Some(st) = &alternative.self_transition {
-                if let Some(agg) = aggregate_type_for_bucket(&st.bucket) {
-                    if !is_valid_transition_target(agg, &st.to) {
+                if let Some(schema) = aggregates.get(&st.bucket) {
+                    if !schema.has_state(&st.to) {
                         anyhow::bail!(
-                            "question {:?} alternative {:?}: self_transition.to {:?} is not a valid transition for bucket {:?} ({agg})",
+                            "question {:?} alternative {:?}: self_transition.to {:?} is not a declared state for bucket {:?}",
                             question.id, alternative.name, st.to, st.bucket
                         );
                     }
@@ -404,13 +447,13 @@ pub fn validate_questions(
                 let Some(bucket) = resource.bucket() else {
                     continue;
                 };
-                let Some(agg) = aggregate_type_for_bucket(bucket) else {
+                let Some(schema) = aggregates.get(bucket) else {
                     continue;
                 };
                 for transition in &resource.transitions {
-                    if !is_valid_transition_target(agg, &transition.to) {
+                    if !schema.has_state(&transition.to) {
                         anyhow::bail!(
-                            "question {:?} alternative {:?} feature {:?}: transition.to {:?} is not a valid transition for bucket {:?} ({agg})",
+                            "question {:?} alternative {:?} feature {:?}: transition.to {:?} is not a declared state for bucket {:?}",
                             question.id, alternative.name, feature.name, transition.to, bucket
                         );
                     }
@@ -429,11 +472,13 @@ pub fn validate_questions(
 pub const CONTENT_RELOAD_SUBJECT: &str = "portal.content.reload";
 
 /// Runs for the life of the process: re-fetches `repo_url`/`branch` and
-/// atomically swaps it into `questions` on every `CONTENT_RELOAD_SUBJECT`
-/// message. A fetch/parse failure logs and keeps serving the last-good
-/// content rather than clearing it - a bad push to the content repo
-/// (which should already have been caught by its own lint step) doesn't
-/// take the site down.
+/// atomically swaps both `questions` and `aggregates` on every
+/// `CONTENT_RELOAD_SUBJECT` message - the two only ever swap together,
+/// after both have loaded and validated successfully, so a reader never
+/// sees pages that reference a half-updated state graph. A fetch/parse
+/// failure logs and keeps serving the last-good content rather than
+/// clearing it - a bad push to the content repo (which should already
+/// have been caught by its own lint step) doesn't take the site down.
 #[cfg(feature = "ssr")]
 pub async fn watch_for_reload(
     nats: async_nats::Client,
@@ -441,6 +486,9 @@ pub async fn watch_for_reload(
     branch: String,
     subdir: String,
     questions: std::sync::Arc<arc_swap::ArcSwap<std::collections::HashMap<String, Question>>>,
+    aggregates: std::sync::Arc<
+        arc_swap::ArcSwap<std::collections::HashMap<String, crate::aggregates::AggregateSchema>>,
+    >,
 ) {
     let mut sub = match nats.subscribe(CONTENT_RELOAD_SUBJECT).await {
         Ok(sub) => sub,
@@ -451,14 +499,22 @@ pub async fn watch_for_reload(
     };
     use futures::StreamExt;
     while sub.next().await.is_some() {
+        let loaded_aggregates = match load_aggregates_from_gitea(&repo_url, &branch).await {
+            Ok(loaded) => loaded,
+            Err(e) => {
+                tracing::error!(error = %e, "aggregates.yaml reload failed, keeping last-good content");
+                continue;
+            }
+        };
         match load_questions_from_gitea(&repo_url, &branch, &subdir).await {
             Ok(loaded) => {
-                if let Err(e) = validate_questions(&loaded) {
+                if let Err(e) = validate_questions(&loaded, &loaded_aggregates) {
                     tracing::error!(error = %e, "content reload failed validation, keeping last-good content");
                     continue;
                 }
                 let count = loaded.len();
                 questions.store(std::sync::Arc::new(loaded));
+                aggregates.store(std::sync::Arc::new(loaded_aggregates));
                 tracing::info!(count, "reloaded content");
             }
             Err(e) => {