portal/src/answers.rs

//! 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.

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<i64>,
    #[serde(default)]
    pub decided_by: Option<String>,
}

/// 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.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, project, subscriber};
        let seed = match agg_type {
            "applicant" => create::<applicant::State>(js, &id, responses.clone(), submitted_ms)
                .await
                .map(|_| ()),
            "subscriber" => create::<subscriber::State>(js, &id, responses.clone(), submitted_ms)
                .await
                .map(|_| ()),
            "project" => create::<project::State>(js, &id, responses.clone(), submitted_ms)
                .await
                .map(|_| ()),
            _ => Ok(()),
        };
        if let Err(e) = seed {
            tracing::error!(bucket, id, error = %e, "failed to seed event log for submission");
        }
    }
    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,
    alternative: 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::<AppState>();
    let question = state
        .questions
        .load()
        .get(&question_id)
        .cloned()
        .ok_or_else(|| ServerFnError::new("unknown question"))?;
    // Same alternative-scoped lookup as get_resource, and for the same
    // reason - a feature name isn't unique across a whole question,
    // only within its own alternative.
    let feature = question
        .alternatives
        .iter()
        .find(|a| a.name == alternative)
        .ok_or_else(|| ServerFnError::new("unknown alternative"))?
        .features
        .iter()
        .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::<User>(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 bucket = resource
        .bucket()
        .ok_or_else(|| ServerFnError::new("resource is not kv-backed, has no transitions"))?;
    let store = state
        .jetstream
        .get_key_value(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();
    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)
            .await
            .map_err(|e| ServerFnError::new(e.to_string()))?;
    }

    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 parent_hashes = vec![item_id.clone()];
    let chain_hash = hash_node(&question_id, &parent_hashes, &decision_payload_for_event, decided_ms);
    let event = AnswerSubmitted {
        chain_hash,
        parent_hashes,
        question_id,
        alternative: transition.label,
        responses: decision_payload_for_event,
        timestamp_ms: decided_ms,
    };
    emit_answer_submitted(&state.nats, &event)
        .await
        .map_err(|e| ServerFnError::new(format!("nats publish failed: {e}")))?;

    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, 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),
        _ => 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
/// `email`, both carried in the link itself (see
/// `content::SelfTransition`). Deliberately returns the same generic
/// error for "no such item" and "email doesn't match" - a real
/// unsubscribe link should never let someone probe which chain hashes
/// or emails exist.
#[server]
pub async fn self_transition_answer(
    question_id: String,
    alternative: String,
    item_id: String,
    email: String,
) -> Result<(), ServerFnError> {
    use crate::chain::hash_node;
    use crate::events::{emit_answer_submitted, AnswerSubmitted};
    use crate::server::AppState;

    let state = expect_context::<AppState>();
    let question = state
        .questions
        .load()
        .get(&question_id)
        .cloned()
        .ok_or_else(|| ServerFnError::new("unknown question"))?;
    let self_transition = question
        .alternatives
        .iter()
        .find(|a| a.name == alternative)
        .ok_or_else(|| ServerFnError::new("unknown alternative"))?
        .self_transition
        .clone()
        .ok_or_else(|| ServerFnError::new("alternative has no self transition"))?;

    let store = state
        .jetstream
        .get_key_value(&self_transition.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("not found"))?;
    let mut answer: Answer =
        serde_json::from_slice(&bytes).map_err(|e| ServerFnError::new(e.to_string()))?;

    let stored_email = answer.responses.get("email").and_then(|v| v.as_str());
    if stored_email != Some(email.as_str()) {
        return Err(ServerFnError::new("not found"));
    }
    if answer.state != OPEN_STATE {
        return Err(ServerFnError::new("already processed"));
    }

    let decided_ms = chrono::Utc::now().timestamp_millis();
    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(
            &state.jetstream,
            agg_type,
            &item_id,
            &self_transition.to,
            decision_payload_for_event.clone(),
            decided_ms,
        )
        .await
        .map_err(|e| ServerFnError::new(e.to_string()))?;
    }

    answer.state = self_transition.to.clone();
    answer.decided_ms = Some(decided_ms);
    answer.decided_by = Some(email.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()))?;

    let parent_hashes = vec![item_id.clone()];
    let chain_hash = hash_node(&question_id, &parent_hashes, &decision_payload_for_event, decided_ms);
    let event = AnswerSubmitted {
        chain_hash,
        parent_hashes,
        question_id,
        alternative: self_transition.label,
        responses: decision_payload_for_event,
        timestamp_ms: decided_ms,
    };
    emit_answer_submitted(&state.nats, &event)
        .await
        .map_err(|e| ServerFnError::new(format!("nats publish failed: {e}")))?;

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    /// `app.rs`'s list-rendering path only shows the nice
    /// button-per-transition UI if a resource's raw JSON value
    /// deserializes as `Vec<Answer>` - otherwise it silently falls
    /// back to a plain read-only dump with no error. Guards that shape
    /// contract directly, since nothing else would catch a drift here.
    #[test]
    fn answer_list_round_trips_through_the_shape_app_rs_expects() {
        let answers = vec![
            Answer {
                id: "abc123".to_string(),
                question_id: "/review".to_string(),
                alternative: "Applicants".to_string(),
                responses: serde_json::json!({"name": "Ada"}),
                state: OPEN_STATE.to_string(),
                submitted_ms: 1_700_000_000_000,
                decided_ms: None,
                decided_by: None,
            },
            Answer {
                id: "def456".to_string(),
                question_id: "/review".to_string(),
                alternative: "Applicants".to_string(),
                responses: serde_json::json!({"name": "Grace"}),
                state: "invited".to_string(),
                submitted_ms: 1_700_000_001_000,
                decided_ms: Some(1_700_000_002_000),
                decided_by: Some("bl".to_string()),
            },
        ];
        let value = serde_json::Value::Array(
            answers.iter().map(|a| serde_json::to_value(a).unwrap()).collect(),
        );
        let round_tripped: Vec<Answer> =
            serde_json::from_value(value).expect("Answer list round-trips");
        assert_eq!(round_tripped.len(), 2);
        assert_eq!(round_tripped[1].state, "invited");
        assert_eq!(round_tripped[1].decided_by.as_deref(), Some("bl"));
    }
}