//! Durable, queryable storage for submitted answers - the KV-backed //! counterpart to the ephemeral `events::AnswerSubmitted` every //! submission already publishes. The runtime here has no notion of //! "applicant" or any other specific workflow: an `Alternative.record_as` //! names a NATS KV bucket (nothing more); on submit, the response gets //! wrapped in the same envelope (`Answer`) regardless of what content //! produced it. `resource::get_resource` reads it back generically, and //! `transition_answer` below moves an answer between states named in //! content (`ResourceSpec.transitions`), never states hardcoded here. //! Whatever workflow this ends up powering - applicant review, order //! triage, comment moderation - is entirely a property of some //! `portal-content` YAML, not of this file. use leptos::prelude::*; use serde::{Deserialize, Serialize}; /// An answer's starting state - content decides what it may transition /// to (`ResourceSpec.transitions`), never this file. pub const OPEN_STATE: &str = "open"; #[derive(Clone, Debug, Serialize, Deserialize)] pub struct Answer { /// The submission's chain hash - stable, unique, and already what /// ties this answer back into the DAG a transition later extends. pub id: String, pub question_id: String, pub alternative: String, /// The submitted form fields, verbatim - the runtime never inspects /// these keys, only content (via its own copy/labels) gives them /// meaning. pub responses: serde_json::Value, pub state: String, pub submitted_ms: i64, #[serde(default)] pub decided_ms: Option, #[serde(default)] pub decided_by: Option, } /// Writes a submitted answer into `bucket`, creating it on first use - /// there's no fixed, pre-declared list of buckets, since content can /// name any. #[cfg(feature = "ssr")] pub async fn store_answer( js: &async_nats::jetstream::Context, bucket: &str, id: String, question_id: &str, alternative: &str, responses: &serde_json::Value, submitted_ms: i64, ) -> anyhow::Result<()> { let store = match js.get_key_value(bucket).await { Ok(store) => store, Err(_) => { js.create_key_value(async_nats::jetstream::kv::Config { bucket: bucket.to_string(), ..Default::default() }) .await? } }; let answer = Answer { id: id.clone(), question_id: question_id.to_string(), alternative: alternative.to_string(), responses: responses.clone(), state: OPEN_STATE.to_string(), submitted_ms, decided_ms: None, decided_by: None, }; let data = serde_json::to_vec(&answer)?; store.put(id, data.into()).await?; Ok(()) } /// Moves a stored answer from `open` to one of the target states its /// own resource spec declares (`ResourceSpec.transitions`) - `to` is /// checked against that content-declared allow-list, never accepted as /// an arbitrary string. Requires a signed-in member of the resource's /// `requires_group`, regardless of whether the resource is otherwise /// `public` - reads can be public, mutations always need identity. #[server] pub async fn transition_answer( question_id: String, feature_name: String, item_id: String, to: String, ) -> Result<(), ServerFnError> { use crate::auth::{User, SESSION_USER_KEY}; use crate::chain::hash_node; use crate::events::{emit_answer_submitted, AnswerSubmitted}; use crate::server::AppState; let state = expect_context::(); let question = state .questions .load() .get(&question_id) .cloned() .ok_or_else(|| ServerFnError::new("unknown question"))?; let feature = question .alternatives .iter() .flat_map(|a| &a.features) .find(|f| f.name == feature_name) .ok_or_else(|| ServerFnError::new("unknown feature"))?; let resource = feature .resource .as_ref() .ok_or_else(|| ServerFnError::new("feature has no resource"))?; let transition = resource .transitions .iter() .find(|t| t.to == to) .ok_or_else(|| ServerFnError::new("no such transition"))? .clone(); let group = resource .requires_group .as_deref() .ok_or_else(|| ServerFnError::new("resource has no transitions available"))?; let session: tower_sessions::Session = leptos_axum::extract().await?; let user = session .get::(SESSION_USER_KEY) .await .map_err(|e| ServerFnError::new(e.to_string()))? .ok_or_else(|| ServerFnError::new("not signed in"))?; if !user.groups.iter().any(|g| g == group) { return Err(ServerFnError::new("not authorized")); } let store = state .jetstream .get_key_value(&resource.bucket) .await .map_err(|e| ServerFnError::new(format!("resource bucket unavailable: {e}")))?; let bytes = store .get(&item_id) .await .map_err(|e| ServerFnError::new(e.to_string()))? .ok_or_else(|| ServerFnError::new("unknown answer"))?; let mut answer: Answer = serde_json::from_slice(&bytes).map_err(|e| ServerFnError::new(e.to_string()))?; if answer.state != OPEN_STATE { return Err(ServerFnError::new("already decided")); } let decided_ms = chrono::Utc::now().timestamp_millis(); answer.state = to.clone(); answer.decided_ms = Some(decided_ms); answer.decided_by = Some(user.username.clone()); let data = serde_json::to_vec(&answer).map_err(|e| ServerFnError::new(e.to_string()))?; store .put(&item_id, data.into()) .await .map_err(|e| ServerFnError::new(e.to_string()))?; // Extend the DAG: the decision is a child node of the answer's own // submission hash, published the same way any other answer is - // question_id/alternative come from this call's own arguments and // the content-declared transition label, never a hardcoded value. let decision_payload = serde_json::json!({ "to": to, "item": item_id, "by": user.username }); let parent_hashes = vec![item_id.clone()]; let chain_hash = hash_node(&question_id, &parent_hashes, &decision_payload, decided_ms); let event = AnswerSubmitted { chain_hash, parent_hashes, question_id, alternative: transition.label, responses: decision_payload, timestamp_ms: decided_ms, }; emit_answer_submitted(&state.nats, &event) .await .map_err(|e| ServerFnError::new(format!("nats publish failed: {e}")))?; Ok(()) }