portal/src/answers.rs

//! Read-model storage for submitted answers: the KV projection of the
//! event log (`events::store`, `aggregates`) for every bucket with a
//! state graph in `aggregates.yaml`. A bucket with no graph still
//! works - direct KV mutate, no event log, no CAS - so content isn't
//! forced to declare one before it needs it.

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")]
#[allow(clippy::too_many_arguments)]
pub async fn store_answer(
    js: &async_nats::jetstream::Context,
    aggregates: &std::collections::HashMap<String, crate::aggregates::AggregateSchema>,
    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?
        }
    };
    // A schema-backed bucket's projection must agree with its event
    // log about where a record starts - organizations begin at
    // "prospect", not "open". Only schemaless buckets default to open.
    let initial_state = aggregates
        .get(bucket)
        .map(|s| s.initial.clone())
        .unwrap_or_else(|| OPEN_STATE.to_string());
    let answer = Answer {
        id: id.clone(),
        question_id: question_id.to_string(),
        alternative: alternative.to_string(),
        responses: responses.clone(),
        state: initial_state,
        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 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");
        }
    }
    Ok(())
}

/// One selected row's pending transition. `feature_name` is per-item
/// so a batch can span more than one resource feature on the same
/// alternative.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct TransitionItem {
    pub feature_name: String,
    pub item_id: String,
    pub to: String,
}

/// Moves every selected row to its chosen target state in one call
/// (the shared per-alternative Confirm button). Items are independent
/// - per-item CAS, no cross-item transaction - so one failure doesn't
/// roll back the others; failures are collected and reported together.
#[server(endpoint = "transition_answers")]
pub async fn transition_answers(
    question_id: String,
    alternative: String,
    items: Vec<TransitionItem>,
) -> Result<(), ServerFnError> {
    use crate::auth::{User, SESSION_USER_KEY};
    use crate::server::AppState;

    let state = expect_context::<AppState>();
    let alt = state
        .questions
        .load()
        .get(&question_id)
        .and_then(|q| q.alternatives.iter().find(|a| a.name == alternative).cloned())
        .ok_or_else(|| ServerFnError::new("unknown question or alternative"))?;

    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"))?;

    let mut errors = Vec::new();
    for item in items {
        let item_id = item.item_id.clone();
        if let Err(e) = apply_transition(&state, &question_id, &alt, &user, item).await {
            errors.push(format!("{item_id}: {e}"));
        }
    }
    if errors.is_empty() {
        Ok(())
    } else {
        Err(ServerFnError::new(errors.join("; ")))
    }
}

/// `aggregates::transition`, with one recovery path: an aggregate the
/// event log has never heard of but whose KV projection exists (the
/// log was purged, or predates event-sourcing) gets a minimal history
/// reseeded from the projection's own current state, then the
/// transition retries. Keeps "zap the whole event stream" a safe
/// operation instead of one that strands every pre-wipe record.
#[cfg(feature = "ssr")]
async fn transition_or_reseed(
    js: &async_nats::jetstream::Context,
    schema: &crate::aggregates::AggregateSchema,
    item_id: &str,
    to: &str,
    current_state: &str,
    payload: serde_json::Value,
    occurred_at_ms: i64,
) -> Result<(), crate::aggregates::TransitionError> {
    match crate::aggregates::transition(js, schema, item_id, to, payload.clone(), occurred_at_ms)
        .await
    {
        Err(crate::aggregates::TransitionError::UnknownAggregate) => {
            tracing::warn!(
                bucket = %schema.bucket, item_id, current_state,
                "no event history for a projected record - reseeding from the projection"
            );
            crate::aggregates::reseed(js, schema, item_id, current_state, occurred_at_ms).await?;
            crate::aggregates::transition(js, schema, item_id, to, payload, occurred_at_ms)
                .await
                .map(|_| ())
        }
        r => r.map(|_| ()),
    }
}

/// Same alternative-scoped feature lookup as `get_resource`, and for
/// the same reason - a feature name isn't unique across a whole
/// question, only within its own alternative.
#[cfg(feature = "ssr")]
async fn apply_transition(
    state: &crate::server::AppState,
    question_id: &str,
    alt: &crate::content::Alternative,
    user: &crate::auth::User,
    item: TransitionItem,
) -> Result<(), ServerFnError> {
    use crate::chain::hash_node;
    use crate::events::{emit_answer_submitted, AnswerSubmitted};

    let feature = alt
        .features
        .iter()
        .find(|f| f.name == item.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 group = resource
        .requires_group
        .as_deref()
        .ok_or_else(|| ServerFnError::new("resource has no transitions available"))?;
    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.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()))?;
    // Matched on (from, to), not to alone: two declared transitions may
    // share a target (open -> declined, in_dialogue -> declined), and
    // only the one starting from the row's actual state may fire.
    let transition = resource
        .transitions
        .iter()
        .find(|t| t.to == item.to && t.from == answer.state)
        .ok_or_else(|| {
            ServerFnError::new(format!(
                "no transition to {:?} from this item's current state {:?}",
                item.to, answer.state
            ))
        })?
        .clone();

    let decided_ms = chrono::Utc::now().timestamp_millis();
    let decision_payload_for_event =
        serde_json::json!({ "to": item.to, "item": item.item_id, "by": user.username });

    // CAS on the aggregate's replayed sequence guards two concurrent
    // decisions on the same item; a schemaless bucket falls back to the
    // plain KV mutate below.
    if let Some(schema) = state.aggregates.load().get(bucket) {
        transition_or_reseed(
            &state.jetstream,
            schema,
            &item.item_id,
            &item.to,
            &answer.state,
            decision_payload_for_event.clone(),
            decided_ms,
        )
        .await
        .map_err(|e| ServerFnError::new(e.to_string()))?;
    }

    answer.state = item.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.item_id, data.into())
        .await
        .map_err(|e| ServerFnError::new(e.to_string()))?;

    // The decision extends the DAG as a child of the answer's own
    // submission hash, published like any other answer.
    let parent_hashes = vec![item.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: question_id.to_string(),
        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(())
}

/// The self-service counterpart to `transition_answers`: no session,
/// no group check - authorized by holding `item_id` (an unguessable
/// chain hash) plus a matching `email`, both carried in the link
/// itself. Deliberately returns the same generic error for "no such
/// item" and "email doesn't match", so an unsubscribe link can't be
/// used to probe which hashes or emails exist.
#[server(endpoint = "self_transition_answer")]
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(schema) = state.aggregates.load().get(&self_transition.bucket) {
        transition_or_reseed(
            &state.jetstream,
            schema,
            &item_id,
            &self_transition.to,
            &answer.state,
            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"));
    }
}