portal/src/resource.rs

//! Generic, authorized reads of NATS KV data declared in content
//! (`content::ResourceSpec`, on a `Feature`). The one property that
//! matters: the bucket/key a resource reads is never a client-supplied
//! parameter, only something the server resolves from its own
//! trusted, YAML-loaded content - the client only ever names a
//! question id + feature name, so it can't probe an arbitrary bucket
//! by just naming it.

use leptos::prelude::*;

/// Fetches the live data for `question_id`'s `alternative`'s
/// `feature_name` feature. Fails closed: a resource with neither
/// `public: true` nor `requires_group` set is unreachable, not "open"
/// by omission.
#[server]
pub async fn get_resource(
    question_id: String,
    alternative: String,
    feature_name: String,
) -> Result<serde_json::Value, ServerFnError> {
    use crate::auth::{User, SESSION_USER_KEY};
    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"))?;
    // Scoped to the named alternative first, not flattened across all
    // of them - a feature name (often just "") is only unique within
    // its own alternative, not across a whole question. Flattening
    // silently resolved every same-named feature to whichever
    // alternative happened to be first, so "Subscribers" (and any
    // other later resource-listing alternative sharing an unnamed
    // feature with an earlier one on the same question) always read
    // the first alternative's bucket instead of its own.
    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"))?;

    if !resource.public {
        let group = resource
            .requires_group
            .as_deref()
            .ok_or_else(|| ServerFnError::new("resource is not accessible"))?;
        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 store = state
        .jetstream
        .get_key_value(&resource.bucket)
        .await
        .map_err(|e| ServerFnError::new(format!("resource bucket unavailable: {e}")))?;

    match &resource.key {
        Some(key) => {
            let bytes = store
                .get(key)
                .await
                .map_err(|e| ServerFnError::new(e.to_string()))?
                .ok_or_else(|| ServerFnError::new("resource key not found"))?;
            serde_json::from_slice(&bytes).map_err(|e| ServerFnError::new(e.to_string()))
        }
        None => {
            use futures::TryStreamExt;
            let keys: Vec<String> = store
                .keys()
                .await
                .map_err(|e| ServerFnError::new(e.to_string()))?
                .try_collect()
                .await
                .map_err(|e| ServerFnError::new(e.to_string()))?;
            let mut items = Vec::new();
            for key in keys {
                if let Ok(Some(bytes)) = store.get(&key).await {
                    if let Ok(value) = serde_json::from_slice::<serde_json::Value>(&bytes) {
                        items.push(value);
                    }
                }
            }
            Ok(serde_json::Value::Array(items))
        }
    }
}