diff --git a/Cargo.toml b/Cargo.toml index dfb6775..6e0090f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -108,11 +108,6 @@ name = "question_lint" path = "src/bin/question_lint.rs" required-features = ["ssr"] -[[bin]] -name = "backfill_events" -path = "src/bin/backfill_events.rs" -required-features = ["ssr"] - [profile.wasm-release] inherits = "release" opt-level = 'z' @@ -121,9 +116,9 @@ codegen-units = 1 [package.metadata.leptos] output-name = "portal" -# Disambiguates the app binary from the two utility [[bin]] targets -# (question_lint, backfill_events) added alongside it - cargo-leptos -# otherwise can't tell which of the three is "the" server. +# Disambiguates the app binary from the utility [[bin]] target +# (question_lint) added alongside it - cargo-leptos otherwise can't +# tell which of the two is "the" server. bin-target = "portal" site-root = "target/site" site-pkg-dir = "pkg" diff --git a/src/aggregates/applicant.rs b/src/aggregates/applicant.rs deleted file mode 100644 index cadc4cf..0000000 --- a/src/aggregates/applicant.rs +++ /dev/null @@ -1,40 +0,0 @@ -//! Collaborator applicants (`/review`'s "Applicants" alternative, -//! `record_as: applicants`). `Open -> Invited | Declined`, straight -//! from the alternatives already declared in `questions/review.yaml`. -use super::AggregateKind; - -#[derive(Copy, Clone, PartialEq, Eq, Debug)] -pub enum State { - Open, - Invited, - Declined, -} - -impl AggregateKind for State { - const AGGREGATE_TYPE: &'static str = "applicant"; - const INITIAL_STATE: Self = State::Open; - - fn from_event_type(event_type: &str) -> Option { - match event_type { - "applied" => Some(State::Open), - "invited" => Some(State::Invited), - "declined" => Some(State::Declined), - _ => None, - } - } - - fn event_type(self) -> &'static str { - match self { - State::Open => "applied", - State::Invited => "invited", - State::Declined => "declined", - } - } - - fn allowed(self) -> &'static [Self] { - match self { - State::Open => &[State::Invited, State::Declined], - State::Invited | State::Declined => &[], - } - } -} diff --git a/src/aggregates/mod.rs b/src/aggregates/mod.rs index 0c424fc..12934b7 100644 --- a/src/aggregates/mod.rs +++ b/src/aggregates/mod.rs @@ -1,79 +1,167 @@ -//! 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` 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`). +//! 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..`) - +//! one name, not two, since content only ever had one name for this to +//! begin with. #![cfg(feature = "ssr")] -pub mod applicant; -pub mod organization; -pub mod project; -pub mod subscriber; +use std::collections::HashMap; 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; +/// 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, + pub state_for_event: HashMap, + pub transitions: HashMap>, +} - /// 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; - /// 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]; +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, + #[serde(default)] + transitions: HashMap>, +} + +#[derive(Debug, serde::Deserialize)] +struct AggregatesFile { + aggregates: Vec, +} + +/// 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> { + 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, + }, + ); + } + Ok(out) } /// One aggregate's current view, replayed from its event history. #[derive(Clone, Debug)] -pub struct Aggregate { +pub struct Aggregate { pub id: String, - pub state: K, + pub state: String, 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(id: &str, events: &[Event]) -> Option> { - let mut current: Option> = None; +/// 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 { + let mut current: Option = None; for event in events { - let Some(next) = K::from_event_type(&event.event_type) else { + let Some(next) = schema.state_for_event.get(&event.event_type) else { continue; }; let accept = match ¤t { None => true, - Some(agg) => next == agg.state || agg.state.allowed().contains(&next), + 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, + state: next.clone(), payload: event.payload.clone(), sequence: event.sequence, }); @@ -122,17 +210,22 @@ fn classify_publish_err(e: async_nats::jetstream::context::PublishError) -> Tran /// 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( +pub async fn create( js: &async_nats::jetstream::Context, + schema: &AggregateSchema, id: &str, payload: serde_json::Value, occurred_at_ms: i64, -) -> Result, TransitionError> { +) -> Result { + 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, - K::AGGREGATE_TYPE, + &schema.bucket, id, - K::INITIAL_STATE.event_type(), + event_type, payload.clone(), occurred_at_ms, Some(0), @@ -141,7 +234,7 @@ pub async fn create( .map_err(classify_publish_err)?; Ok(Aggregate { id: id.to_string(), - state: K::INITIAL_STATE, + state: schema.initial.clone(), payload, sequence: ack.sequence, }) @@ -150,28 +243,35 @@ pub async fn create( /// 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( +pub async fn transition( js: &async_nats::jetstream::Context, + schema: &AggregateSchema, id: &str, - target: K, + target: &str, payload: serde_json::Value, occurred_at_ms: i64, -) -> Result, TransitionError> { - let events = load_events(js, K::AGGREGATE_TYPE, id) +) -> Result { + let events = load_events(js, &schema.bucket, id) .await .map_err(|e| TransitionError::Store(e.to_string()))?; - let current = replay::(id, &events).ok_or(TransitionError::UnknownAggregate)?; - if !current.state.allowed().contains(&target) { + let current = replay(schema, id, &events).ok_or(TransitionError::UnknownAggregate)?; + if !schema.allowed(¤t.state).iter().any(|s| s == target) { return Err(TransitionError::DisallowedTransition { - from: format!("{:?}", current.state), - to: format!("{target:?}"), + 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, - K::AGGREGATE_TYPE, + &schema.bucket, id, - target.event_type(), + event_type, payload.clone(), occurred_at_ms, Some(current.sequence), @@ -180,7 +280,7 @@ pub async fn transition( .map_err(classify_publish_err)?; Ok(Aggregate { id: id.to_string(), - state: target, + state: target.to_string(), payload, sequence: ack.sequence, }) @@ -189,40 +289,24 @@ pub async fn transition( #[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 { - 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 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 { @@ -239,31 +323,83 @@ mod tests { #[test] fn empty_history_is_no_aggregate() { - assert!(replay::("x", &[]).is_none()); + 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::("x", &events).unwrap(); - assert_eq!(agg.state, TestState::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 + // "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::("x", &events).unwrap(); - assert_eq!(agg.state, TestState::A); + 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::("x", &events).unwrap(); - assert_eq!(agg.state, TestState::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! { @@ -274,20 +410,15 @@ mod tests { 0..30, ) ) { + let schema = test_schema(); let events: Vec = types .iter() .enumerate() .map(|(i, t)| event(i as u64, t)) .collect(); - let result = replay::("x", &events); + let result = replay(&schema, "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!(schema.has_state(&agg.state)); assert!(agg.sequence <= events.len() as u64); } } diff --git a/src/aggregates/organization.rs b/src/aggregates/organization.rs deleted file mode 100644 index 065b811..0000000 --- a/src/aggregates/organization.rs +++ /dev/null @@ -1,45 +0,0 @@ -//! Organizations (companies). "Client" is a *status* this aggregate can -//! be in, not a separate aggregate type - `Prospect -> Client -> -//! PastClient`, with `Prospect -> PastClient` also allowed directly -//! (a prospect that never converted, archived without ever having been -//! a client). Greenfield: no content or KV bucket references this yet -//! (see the event-sourcing plan's Deferred section) - this lands -//! compiled and ready, not wired into any live content. -use super::AggregateKind; - -#[derive(Copy, Clone, PartialEq, Eq, Debug)] -pub enum State { - Prospect, - Client, - PastClient, -} - -impl AggregateKind for State { - const AGGREGATE_TYPE: &'static str = "organization"; - const INITIAL_STATE: Self = State::Prospect; - - fn from_event_type(event_type: &str) -> Option { - match event_type { - "identified" => Some(State::Prospect), - "became_client" => Some(State::Client), - "churned" => Some(State::PastClient), - _ => None, - } - } - - fn event_type(self) -> &'static str { - match self { - State::Prospect => "identified", - State::Client => "became_client", - State::PastClient => "churned", - } - } - - fn allowed(self) -> &'static [Self] { - match self { - State::Prospect => &[State::Client, State::PastClient], - State::Client => &[State::PastClient], - State::PastClient => &[], - } - } -} diff --git a/src/aggregates/project.rs b/src/aggregates/project.rs deleted file mode 100644 index 6262882..0000000 --- a/src/aggregates/project.rs +++ /dev/null @@ -1,43 +0,0 @@ -//! Project submissions (`record_as: projects` - absorbs what used to be -//! called "inquiries"). A visitor submits a project idea plus who it's -//! for (org/contact details, captured as plain payload fields - no -//! separate Organization aggregate yet, see the event-sourcing plan). -//! `Open -> Accepted | Declined` - a real b2b-pipeline pair of terminal -//! states rather than the old generic `Handled`. -use super::AggregateKind; - -#[derive(Copy, Clone, PartialEq, Eq, Debug)] -pub enum State { - Open, - Accepted, - Declined, -} - -impl AggregateKind for State { - const AGGREGATE_TYPE: &'static str = "project"; - const INITIAL_STATE: Self = State::Open; - - fn from_event_type(event_type: &str) -> Option { - match event_type { - "submitted" => Some(State::Open), - "accepted" => Some(State::Accepted), - "declined" => Some(State::Declined), - _ => None, - } - } - - fn event_type(self) -> &'static str { - match self { - State::Open => "submitted", - State::Accepted => "accepted", - State::Declined => "declined", - } - } - - fn allowed(self) -> &'static [Self] { - match self { - State::Open => &[State::Accepted, State::Declined], - State::Accepted | State::Declined => &[], - } - } -} diff --git a/src/aggregates/subscriber.rs b/src/aggregates/subscriber.rs deleted file mode 100644 index 83a689a..0000000 --- a/src/aggregates/subscriber.rs +++ /dev/null @@ -1,38 +0,0 @@ -//! Newsletter subscribers (`record_as: subscribers`). `Open -> -//! Unsubscribed`, fired via `self_transition` (see `answers.rs`) rather -//! than an owner-gated `transition`, so a subscriber never needs a -//! signed-in session to leave. -use super::AggregateKind; - -#[derive(Copy, Clone, PartialEq, Eq, Debug)] -pub enum State { - Open, - Unsubscribed, -} - -impl AggregateKind for State { - const AGGREGATE_TYPE: &'static str = "subscriber"; - const INITIAL_STATE: Self = State::Open; - - fn from_event_type(event_type: &str) -> Option { - match event_type { - "subscribed" => Some(State::Open), - "unsubscribed" => Some(State::Unsubscribed), - _ => None, - } - } - - fn event_type(self) -> &'static str { - match self { - State::Open => "subscribed", - State::Unsubscribed => "unsubscribed", - } - } - - fn allowed(self) -> &'static [Self] { - match self { - State::Open => &[State::Unsubscribed], - State::Unsubscribed => &[], - } - } -} diff --git a/src/answers.rs b/src/answers.rs index 05babcd..8e00197 100644 --- a/src/answers.rs +++ b/src/answers.rs @@ -1,15 +1,16 @@ //! Read-model storage for submitted answers - the KV-backed projection //! of the durable event log now underneath it (`events::store`, -//! `aggregates`) for the three buckets migrated onto that machinery -//! this pass (`content::aggregate_type_for_bucket`). An -//! `Alternative.record_as`/`ResourceSpec`'s bucket is still just a -//! bucket name as far as `resource::get_resource` is concerned - this -//! file is where that name additionally gets checked against a real, -//! Rust-defined transition table, for the buckets that have one. -//! A bucket outside that three-entry map still works exactly as -//! before (direct KV mutate-in-place, no event log, no CAS) - this is -//! deliberately not a hard cutover, so content isn't forced to migrate -//! before it needs to. +//! `aggregates`) for every bucket that has a declared state graph in +//! `AppState.aggregates` (loaded from `aggregates.yaml` - see +//! `content::load_aggregates_from_gitea`). An `Alternative.record_as`/ +//! `ResourceSpec`'s bucket is still just a bucket name as far as +//! `resource::get_resource` is concerned - this file is where that name +//! additionally gets checked against the real, content-declared +//! transition table, for the buckets that have one. A bucket with no +//! entry in `aggregates` still works exactly as before (direct KV +//! mutate-in-place, no event log, no CAS) - this is deliberately not a +//! hard cutover, so content isn't forced to declare a state graph +//! before it needs one. use leptos::prelude::*; use serde::{Deserialize, Serialize}; @@ -41,8 +42,10 @@ pub struct Answer { /// there's no fixed, pre-declared list of buckets, since content can /// name any. #[cfg(feature = "ssr")] +#[allow(clippy::too_many_arguments)] pub async fn store_answer( js: &async_nats::jetstream::Context, + aggregates: &std::collections::HashMap, bucket: &str, id: String, question_id: &str, @@ -73,28 +76,12 @@ pub async fn store_answer( let data = serde_json::to_vec(&answer)?; store.put(id.clone(), data.into()).await?; - // Also seed the event log for the three aggregate types migrated - // onto it this pass - best-effort, same "a hiccup here shouldn't - // fail a submission the NATS notification event has already - // recorded" policy as the KV write above. - if let Some(agg_type) = crate::content::aggregate_type_for_bucket(bucket) { - use crate::aggregates::{applicant, create, organization, project, subscriber}; - let seed = match agg_type { - "applicant" => create::(js, &id, responses.clone(), submitted_ms) - .await - .map(|_| ()), - "subscriber" => create::(js, &id, responses.clone(), submitted_ms) - .await - .map(|_| ()), - "project" => create::(js, &id, responses.clone(), submitted_ms) - .await - .map(|_| ()), - "organization" => create::(js, &id, responses.clone(), submitted_ms) - .await - .map(|_| ()), - _ => Ok(()), - }; - if let Err(e) = seed { + // Also seed the event log for any bucket that has a declared state + // graph - best-effort, same "a hiccup here shouldn't fail a + // submission the NATS notification event has already recorded" + // policy as the KV write above. + if let Some(schema) = aggregates.get(bucket) { + if let Err(e) = crate::aggregates::create(js, schema, &id, responses.clone(), submitted_ms).await { tracing::error!(bucket, id, error = %e, "failed to seed event log for submission"); } } @@ -187,15 +174,15 @@ pub async fn transition_answer( let decision_payload_for_event = serde_json::json!({ "to": to, "item": item_id, "by": user.username }); - // For the three buckets migrated onto the event log this pass, the - // real fix for the lost-update race two concurrent decisions on the - // same item used to hit: append with CAS on the aggregate's - // just-replayed sequence, so a second racing caller's write is - // rejected instead of silently overwriting the first. A bucket - // outside that map falls back to the direct KV mutate this always - // did - not every resource has to be migrated to keep working. - if let Some(agg_type) = crate::content::aggregate_type_for_bucket(bucket) { - transition_by_aggregate(&state.jetstream, agg_type, &item_id, &to, decision_payload_for_event.clone(), decided_ms) + // For any bucket with a declared state graph, the real fix for the + // lost-update race two concurrent decisions on the same item used to + // hit: append with CAS on the aggregate's just-replayed sequence, so + // a second racing caller's write is rejected instead of silently + // overwriting the first. A bucket outside `aggregates` falls back to + // the direct KV mutate this always did - not every resource has to + // be event-sourced to keep working. + if let Some(schema) = state.aggregates.load().get(bucket) { + crate::aggregates::transition(&state.jetstream, schema, &item_id, &to, decision_payload_for_event.clone(), decided_ms) .await .map_err(|e| ServerFnError::new(e.to_string()))?; } @@ -231,44 +218,6 @@ pub async fn transition_answer( Ok(()) } -/// Dispatches to the right aggregate type's `transition()` by name - -/// the one place `answers.rs` has to know the fixed, three-entry set of -/// migrated aggregate types (`content::aggregate_type_for_bucket`'s -/// counterpart), since `crate::aggregates` itself stays generic over -/// any `AggregateKind` and has no list of its own to dispatch through. -#[cfg(feature = "ssr")] -async fn transition_by_aggregate( - js: &async_nats::jetstream::Context, - aggregate_type: &str, - item_id: &str, - to: &str, - payload: serde_json::Value, - occurred_at_ms: i64, -) -> Result<(), crate::aggregates::TransitionError> { - use crate::aggregates::{applicant, organization, project, subscriber, transition, AggregateKind, TransitionError}; - - macro_rules! dispatch { - ($state:ty) => {{ - let target = <$state>::from_event_type(to).ok_or_else(|| { - TransitionError::DisallowedTransition { - from: "?".to_string(), - to: to.to_string(), - } - })?; - transition::<$state>(js, item_id, target, payload, occurred_at_ms).await?; - }}; - } - - match aggregate_type { - "applicant" => dispatch!(applicant::State), - "subscriber" => dispatch!(subscriber::State), - "project" => dispatch!(project::State), - "organization" => dispatch!(organization::State), - _ => return Err(TransitionError::UnknownAggregate), - } - Ok(()) -} - /// The self-service counterpart to `transition_answer`: no signed-in /// session, no group check - authorized instead by already holding /// `item_id` (a chain hash, opaque and unguessable) plus a matching @@ -329,10 +278,10 @@ pub async fn self_transition_answer( let decision_payload_for_event = serde_json::json!({ "to": self_transition.to, "item": item_id, "by": email }); - if let Some(agg_type) = crate::content::aggregate_type_for_bucket(&self_transition.bucket) { - transition_by_aggregate( + if let Some(schema) = state.aggregates.load().get(&self_transition.bucket) { + crate::aggregates::transition( &state.jetstream, - agg_type, + schema, &item_id, &self_transition.to, decision_payload_for_event.clone(), diff --git a/src/app.rs b/src/app.rs index ce23ae2..44669cd 100644 --- a/src/app.rs +++ b/src/app.rs @@ -1359,6 +1359,7 @@ pub async fn submit_answer( if let Some(bucket) = &record_as { if let Err(e) = store_answer( &state.jetstream, + &state.aggregates.load(), bucket, chain_hash.clone(), &question_id, diff --git a/src/bin/backfill_events.rs b/src/bin/backfill_events.rs deleted file mode 100644 index f64e0a2..0000000 --- a/src/bin/backfill_events.rs +++ /dev/null @@ -1,148 +0,0 @@ -//! One-time migration: for each of the three renamed buckets -//! (`portal_applicants` -> `applicants`, `portal_subscribers` -> -//! `subscribers`, `portal_project_inquiries` -> `projects`), seeds the -//! new event log with a creation event (plus a decision event for any -//! record already decided) and re-puts the record into the new-named -//! KV bucket - then immediately replays what it just wrote and checks -//! it reproduces the original record's state exactly, per the -//! event-sourcing plan's verification step. -//! -//! Defaults to a dry run (lists what it *would* do, writes nothing) - -//! pass `--apply` to actually write. The old-named buckets are never -//! touched or deleted by this tool; remove them by hand once every -//! record's replay has been confirmed to match. -#![cfg(feature = "ssr")] - -use portal::aggregates::{self, applicant, project, subscriber, AggregateKind}; -use portal::answers::{Answer, OPEN_STATE}; -use portal::events::store as event_store; - -const MIGRATIONS: &[(&str, &str, &str)] = &[ - ("portal_applicants", "applicants", "applicant"), - ("portal_subscribers", "subscribers", "subscriber"), - ("portal_project_inquiries", "projects", "project"), -]; - -#[tokio::main] -async fn main() -> anyhow::Result<()> { - let apply = std::env::args().any(|a| a == "--apply"); - - let nats_url = - std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string()); - let parsed = url::Url::parse(&nats_url)?; - let mut opts = async_nats::ConnectOptions::new(); - if !parsed.username().is_empty() { - opts = opts.user_and_password( - parsed.username().to_string(), - parsed.password().unwrap_or_default().to_string(), - ); - } - let nats = opts.connect(&nats_url).await?; - let js = async_nats::jetstream::new(nats); - event_store::ensure_stream(&js).await?; - - println!("mode: {}\n", if apply { "APPLY (writing)" } else { "DRY RUN (no writes)" }); - - for (old_bucket, new_bucket, agg_type) in MIGRATIONS { - println!("=== {old_bucket} -> {new_bucket} ({agg_type}) ==="); - let store = match js.get_key_value(*old_bucket).await { - Ok(s) => s, - Err(e) => { - println!(" skip: bucket unavailable: {e}"); - continue; - } - }; - use futures::TryStreamExt; - let keys: Vec = store.keys().await?.try_collect().await?; - if keys.is_empty() { - println!(" (empty)"); - } - for key in keys { - let Some(bytes) = store.get(&key).await? else { continue }; - let answer: Answer = serde_json::from_slice(&bytes)?; - println!(" {key}: state={:?} submitted_ms={}", answer.state, answer.submitted_ms); - if apply { - if let Err(e) = seed_and_verify(&js, agg_type, &key, &answer, new_bucket).await { - println!(" ! {e}"); - } - } - } - } - - if !apply { - println!("\nDry run only - no writes made. Re-run with --apply to actually backfill."); - } - Ok(()) -} - -async fn seed_and_verify( - js: &async_nats::jetstream::Context, - agg_type: &str, - id: &str, - answer: &Answer, - new_bucket: &str, -) -> anyhow::Result<()> { - match agg_type { - "applicant" => seed::(js, id, answer).await?, - "subscriber" => seed::(js, id, answer).await?, - "project" => seed::(js, id, answer).await?, - other => anyhow::bail!("unknown aggregate type {other}"), - } - - let new_store = match js.get_key_value(new_bucket).await { - Ok(s) => s, - Err(_) => { - js.create_key_value(async_nats::jetstream::kv::Config { - bucket: new_bucket.to_string(), - ..Default::default() - }) - .await? - } - }; - new_store.put(id, serde_json::to_vec(answer)?.into()).await?; - Ok(()) -} - -/// Appends the initial event, then - if the original record was -/// already decided - a second event carrying the real decision, before -/// replaying and checking the result matches. Two events for a decided -/// record, not one: a single "imported" event would show a decided -/// record as forever open on replay. -async fn seed( - js: &async_nats::jetstream::Context, - id: &str, - answer: &Answer, -) -> anyhow::Result<()> { - aggregates::create::(js, id, answer.responses.clone(), answer.submitted_ms) - .await - .map_err(|e| anyhow::anyhow!("seeding initial event: {e}"))?; - - if answer.state != OPEN_STATE { - match K::from_event_type(&answer.state) { - Some(target) => { - let payload = serde_json::json!({ - "to": answer.state, - "item": id, - "by": answer.decided_by, - }); - let occurred = answer.decided_ms.unwrap_or(answer.submitted_ms); - aggregates::transition::(js, id, target, payload, occurred) - .await - .map_err(|e| anyhow::anyhow!("seeding decision event: {e}"))?; - } - None => println!(" ! unrecognized decided state {:?}, only seeded initial event", answer.state), - } - } - - let events = event_store::load_events(js, K::AGGREGATE_TYPE, id).await?; - match aggregates::replay::(id, &events) { - Some(agg) if agg.state.event_type() == answer.state => { - println!(" OK: replay matches original state {:?}", answer.state); - } - Some(agg) => { - println!(" MISMATCH: replayed {:?}, original was {:?}", agg.state, answer.state); - } - None => println!(" MISMATCH: no aggregate replayed at all"), - } - Ok(()) -} diff --git a/src/bin/question_lint.rs b/src/bin/question_lint.rs index 5980232..528193d 100644 --- a/src/bin/question_lint.rs +++ b/src/bin/question_lint.rs @@ -1,13 +1,15 @@ -//! Headless schema-check binary - loads content (from a Gitea repo URL -//! or a local directory) and validates it exactly the way -//! `content::watch_for_reload`/`main.rs`'s boot path do, with no NATS, -//! OIDC, web server, or JetStream connection involved. Built once by -//! portal's own deploy workflow and downloaded as a release artifact by -//! `questions`' own CI, rather than compiled there - keeps that repo's -//! CI coupling to "run a static binary," not "build a Rust workspace." +//! Headless schema-check binary - loads content and `aggregates.yaml` +//! (from a Gitea repo URL or a local directory) and validates them +//! exactly the way `content::watch_for_reload`/`main.rs`'s boot path +//! do, with no NATS, OIDC, web server, or JetStream connection +//! involved. Built once by portal's own deploy workflow and run +//! directly by `questions`' own CI (same bare-metal runner/host, +//! published to a stable path - no artifact download needed), rather +//! than compiled there - keeps that repo's CI coupling to "run a +//! static binary," not "build a Rust workspace." #![cfg(feature = "ssr")] -use portal::content; +use portal::{aggregates, content}; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -30,10 +32,12 @@ async fn main() -> anyhow::Result<()> { } } - let questions = if let Some(dir) = path { - load_from_dir(&dir)? + let (questions, aggregates_map) = if let Some(dir) = path { + (load_from_dir(&dir)?, load_aggregates_from_dir(&dir)?) } else if let Some(repo_url) = repo { - content::load_questions_from_gitea(&repo_url, &branch, &subdir).await? + let questions = content::load_questions_from_gitea(&repo_url, &branch, &subdir).await?; + let aggregates_map = content::load_aggregates_from_gitea(&repo_url, &branch).await?; + (questions, aggregates_map) } else { eprintln!( "usage: question-lint --repo [--branch main] [--subdir questions] | --path " @@ -41,9 +45,13 @@ async fn main() -> anyhow::Result<()> { std::process::exit(2); }; - match content::validate_questions(&questions) { + match content::validate_questions(&questions, &aggregates_map) { Ok(()) => { - println!("OK: {} question(s) valid", questions.len()); + println!( + "OK: {} question(s), {} aggregate(s) valid", + questions.len(), + aggregates_map.len() + ); Ok(()) } Err(e) => { @@ -53,6 +61,29 @@ async fn main() -> anyhow::Result<()> { } } +/// The offline counterpart to `content::load_aggregates_from_gitea` - +/// `aggregates.yaml` lives at the repo root, one level up from the +/// pages directory `--path` names, so `dir`'s parent is where it's +/// looked for. Missing entirely is not an error here (unlike +/// `--repo` mode, where `aggregates.yaml` is required) - a local +/// checkout being linted may not have one, and every declared +/// transition still gets checked against whatever *is* found; an +/// empty map just means nothing is checked. +fn load_aggregates_from_dir( + dir: &str, +) -> anyhow::Result> { + let aggregates_path = std::path::Path::new(dir) + .parent() + .unwrap_or_else(|| std::path::Path::new(".")) + .join("aggregates.yaml"); + if !aggregates_path.exists() { + return Ok(std::collections::HashMap::new()); + } + let raw = std::fs::read_to_string(&aggregates_path) + .map_err(|e| anyhow::anyhow!("reading {}: {e}", aggregates_path.display()))?; + aggregates::parse_aggregates_yaml(&raw) +} + /// The offline counterpart to `content::load_questions_from_gitea` - /// same "every `*.yaml` file becomes a `Question` keyed by its own /// `id`" shape, just reading a local checkout instead of Gitea's API, diff --git a/src/content.rs b/src/content.rs index ffe9e1a..3daa4df 100644 --- a/src/content.rs +++ b/src/content.rs @@ -262,6 +262,91 @@ pub fn gitea_api_base(repo_url: &str) -> anyhow::Result { )) } +/// 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 { + 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> { + 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> { - 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(to: &str) -> bool { - K::INITIAL_STATE.allowed().iter().any(|s| s.event_type() == to) - } - match aggregate_type { - "applicant" => check::(to), - "subscriber" => check::(to), - "project" => check::(to), - "organization" => check::(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, + aggregates: &std::collections::HashMap, ) -> 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>>, + aggregates: std::sync::Arc< + arc_swap::ArcSwap>, + >, ) { 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) => { diff --git a/src/main.rs b/src/main.rs index 47b4bcb..a60b1c8 100644 --- a/src/main.rs +++ b/src/main.rs @@ -29,10 +29,12 @@ async fn main() -> anyhow::Result<()> { .unwrap_or_else(|_| "https://project.uhhm.no/uhhm/questions".to_string()); let content_branch = std::env::var("CONTENT_BRANCH").unwrap_or_else(|_| "main".to_string()); let gitea_base = content::gitea_api_base(&content_repo)?; + let aggregates = content::load_aggregates_from_gitea(&content_repo, &content_branch).await?; let questions = content::load_questions_from_gitea(&content_repo, &content_branch, "questions").await?; - content::validate_questions(&questions)?; + content::validate_questions(&questions, &aggregates)?; tracing::info!(count = questions.len(), repo = %content_repo, branch = %content_branch, "loaded content"); let questions = Arc::new(arc_swap::ArcSwap::from_pointee(questions)); + let aggregates = Arc::new(arc_swap::ArcSwap::from_pointee(aggregates)); let nats_url = std::env::var("NATS_URL").unwrap_or_else(|_| "nats://127.0.0.1:4222".to_string()); @@ -66,6 +68,7 @@ async fn main() -> anyhow::Result<()> { content_branch, "questions".to_string(), questions.clone(), + aggregates.clone(), )); let state = AppState { @@ -73,6 +76,7 @@ async fn main() -> anyhow::Result<()> { nats, jetstream, questions, + aggregates, gitea_base, oidc: oidc_state, garage, diff --git a/src/server/mod.rs b/src/server/mod.rs index 3c0105c..7b357f9 100644 --- a/src/server/mod.rs +++ b/src/server/mod.rs @@ -22,6 +22,12 @@ pub struct AppState { /// a lock, just an atomic pointer load, so a reload never blocks or /// is blocked by an in-flight request. pub questions: Arc>>, + /// Every bucket's event-sourced state graph, loaded from + /// `aggregates.yaml` (see `content::load_aggregates_from_gitea`) and + /// hot-swapped alongside `questions` on the same reload. A bucket + /// with no entry here isn't event-sourced - plain KV mutate-in-place + /// still works (see `answers.rs`). + pub aggregates: Arc>>, /// `scheme://host` of the Gitea instance content is loaded from /// (see `content::gitea_api_base`) - kept alongside `questions` /// rather than re-derived per call, since `resolve_gitea_repo` needs