PiCloud/crates/manager-core/src/topics_api.rs

//! `/api/v1/admin/apps/{id}/topics*` — topic registration admin
//! endpoints (v1.1.6).
//!
//! These manage the `topics` table: the explicit registry of which
//! pub/sub topics are externally subscribable over SSE (design notes
//! §5). Internal-only topics never appear here.
//!
//!   * `POST   /apps/{id}/topics`            — register a topic.
//!   * `GET    /apps/{id}/topics`            — list registered topics.
//!   * `PATCH  /apps/{id}/topics/{name}`     — flip external/auth_mode.
//!   * `DELETE /apps/{id}/topics/{name}`     — unregister + disconnect.
//!
//! The PATCH endpoint is deliberately its OWN surface (not folded into a
//! generic topic update) so every change to externally-subscribable
//! status is a discrete, watchable/auditable API call (§5 commitment).
//!
//! Create / update / delete are gated by `AppTopicManage` (→ `app:admin`
//! scope); list is gated by the existing `AppRead`. DELETE also drops
//! the topic's in-process broadcast channel so live SSE subscribers
//! disconnect cleanly.

use std::sync::Arc;

use axum::extract::{Path, State};
use axum::http::StatusCode;
use axum::response::{IntoResponse, Json, Response};
use axum::routing::{get, patch};
use axum::{Extension, Router};
use picloud_shared::{AppId, Principal, RealtimeBroadcaster};
use serde::Deserialize;
use serde_json::json;

use crate::app_repo::AppRepository;
use crate::authz::{require, AuthzDenied, AuthzError, AuthzRepo, Capability};
use crate::topic_repo::{Topic, TopicAuthMode, TopicRepo, TopicRepoError};

#[derive(Clone)]
pub struct TopicsState {
    pub topics: Arc<dyn TopicRepo>,
    pub apps: Arc<dyn AppRepository>,
    pub authz: Arc<dyn AuthzRepo>,
    pub broadcaster: Arc<dyn RealtimeBroadcaster>,
}

pub fn topics_router(state: TopicsState) -> Router {
    Router::new()
        .route("/apps/{app_id}/topics", get(list_topics).post(create_topic))
        .route(
            "/apps/{app_id}/topics/{name}",
            patch(update_topic).delete(delete_topic),
        )
        .with_state(state)
}

#[derive(Debug, Deserialize)]
pub struct CreateTopicRequest {
    pub name: String,
    #[serde(default)]
    pub external_subscribable: bool,
    #[serde(default = "default_auth_mode")]
    pub auth_mode: TopicAuthMode,
}

const fn default_auth_mode() -> TopicAuthMode {
    TopicAuthMode::Public
}

#[derive(Debug, Deserialize)]
pub struct UpdateTopicRequest {
    #[serde(default)]
    pub external_subscribable: Option<bool>,
    #[serde(default)]
    pub auth_mode: Option<TopicAuthMode>,
}

/// Topic names are concrete (external pattern subscription is v1.2), so
/// reject empties and `*` wildcards at registration.
fn validate_topic_name(name: &str) -> Result<(), TopicsApiError> {
    if name.trim().is_empty() {
        return Err(TopicsApiError::Invalid(
            "topic name must not be empty".into(),
        ));
    }
    if name.contains('*') {
        return Err(TopicsApiError::Invalid(
            "topic name must be a concrete topic, not a pattern (no '*')".into(),
        ));
    }
    Ok(())
}

async fn create_topic(
    State(s): State<TopicsState>,
    Extension(principal): Extension<Principal>,
    Path(app_id): Path<AppId>,
    Json(input): Json<CreateTopicRequest>,
) -> Result<(StatusCode, Json<Topic>), TopicsApiError> {
    ensure_app_exists(&*s.apps, app_id).await?;
    require(
        s.authz.as_ref(),
        &principal,
        Capability::AppTopicManage(app_id),
    )
    .await?;
    validate_topic_name(&input.name)?;
    let topic = s
        .topics
        .create(
            app_id,
            input.name.trim(),
            input.external_subscribable,
            input.auth_mode,
        )
        .await?;
    Ok((StatusCode::CREATED, Json(topic)))
}

#[derive(Debug, serde::Serialize)]
struct ListTopicsResponse {
    topics: Vec<Topic>,
}

async fn list_topics(
    State(s): State<TopicsState>,
    Extension(principal): Extension<Principal>,
    Path(app_id): Path<AppId>,
) -> Result<Json<ListTopicsResponse>, TopicsApiError> {
    ensure_app_exists(&*s.apps, app_id).await?;
    require(s.authz.as_ref(), &principal, Capability::AppRead(app_id)).await?;
    let topics = s.topics.list(app_id).await?;
    Ok(Json(ListTopicsResponse { topics }))
}

async fn update_topic(
    State(s): State<TopicsState>,
    Extension(principal): Extension<Principal>,
    Path((app_id, name)): Path<(AppId, String)>,
    Json(input): Json<UpdateTopicRequest>,
) -> Result<Json<Topic>, TopicsApiError> {
    ensure_app_exists(&*s.apps, app_id).await?;
    require(
        s.authz.as_ref(),
        &principal,
        Capability::AppTopicManage(app_id),
    )
    .await?;
    let topic = s
        .topics
        .update(app_id, &name, input.external_subscribable, input.auth_mode)
        .await?
        .ok_or(TopicsApiError::NotFound)?;
    Ok(Json(topic))
}

async fn delete_topic(
    State(s): State<TopicsState>,
    Extension(principal): Extension<Principal>,
    Path((app_id, name)): Path<(AppId, String)>,
) -> Result<StatusCode, TopicsApiError> {
    ensure_app_exists(&*s.apps, app_id).await?;
    require(
        s.authz.as_ref(),
        &principal,
        Capability::AppTopicManage(app_id),
    )
    .await?;
    if !s.topics.delete(app_id, &name).await? {
        return Err(TopicsApiError::NotFound);
    }
    // Disconnect any live SSE subscribers for the now-unregistered topic.
    s.broadcaster.drop_topic(app_id, &name).await;
    Ok(StatusCode::NO_CONTENT)
}

async fn ensure_app_exists(apps: &dyn AppRepository, app_id: AppId) -> Result<(), TopicsApiError> {
    apps.get_by_id(app_id)
        .await
        .map_err(|e| TopicsApiError::Backend(e.to_string()))?
        .ok_or(TopicsApiError::AppNotFound)?;
    Ok(())
}

#[derive(Debug, thiserror::Error)]
pub enum TopicsApiError {
    #[error("app not found")]
    AppNotFound,
    #[error("topic not found")]
    NotFound,
    #[error("{0}")]
    AlreadyExists(String),
    #[error("invalid request: {0}")]
    Invalid(String),
    #[error("forbidden")]
    Forbidden,
    #[error("authorization repo error: {0}")]
    AuthzRepo(String),
    #[error("topics backend: {0}")]
    Backend(String),
}

impl From<AuthzDenied> for TopicsApiError {
    fn from(d: AuthzDenied) -> Self {
        match d {
            AuthzDenied::Denied => Self::Forbidden,
            AuthzDenied::Repo(e) => Self::AuthzRepo(e.to_string()),
        }
    }
}

impl From<AuthzError> for TopicsApiError {
    fn from(e: AuthzError) -> Self {
        Self::AuthzRepo(e.to_string())
    }
}

impl From<TopicRepoError> for TopicsApiError {
    fn from(e: TopicRepoError) -> Self {
        match e {
            TopicRepoError::AlreadyExists(name) => {
                Self::AlreadyExists(format!("a topic named {name:?} already exists in this app"))
            }
            other => Self::Backend(other.to_string()),
        }
    }
}

impl IntoResponse for TopicsApiError {
    fn into_response(self) -> Response {
        let (status, body) = match &self {
            Self::AppNotFound | Self::NotFound => {
                (StatusCode::NOT_FOUND, json!({ "error": self.to_string() }))
            }
            Self::AlreadyExists(_) => (StatusCode::CONFLICT, json!({ "error": self.to_string() })),
            Self::Invalid(_) => (
                StatusCode::UNPROCESSABLE_ENTITY,
                json!({ "error": self.to_string() }),
            ),
            Self::Forbidden => (StatusCode::FORBIDDEN, json!({ "error": self.to_string() })),
            Self::AuthzRepo(e) => {
                tracing::error!(error = %e, "topics admin authz repo error");
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    json!({ "error": "internal error" }),
                )
            }
            Self::Backend(e) => {
                tracing::error!(error = %e, "topics admin backend error");
                (
                    StatusCode::INTERNAL_SERVER_ERROR,
                    json!({ "error": "internal error" }),
                )
            }
        };
        (status, Json(body)).into_response()
    }
}

#[cfg(test)]
mod tests {
    //! In-memory handler tests: capability enforcement, the
    //! `external_subscribable` default, the flip being its own endpoint,
    //! cross-app isolation, and DELETE disconnecting subscribers. The
    //! Postgres repo is exercised by the schema + integration suites.

    use super::*;
    use crate::repo::ScriptRepositoryError;
    use async_trait::async_trait;
    use chrono::Utc;
    use picloud_shared::{
        AdminUserId, App, AppRole, BroadcasterError, InstanceRole, RealtimeEvent, UserId,
    };
    use std::collections::HashMap;
    use std::sync::Mutex as StdMutex;
    use tokio::sync::Mutex;

    #[derive(Default)]
    struct InMemoryTopicRepo {
        inner: Mutex<HashMap<(AppId, String), Topic>>,
    }

    #[async_trait]
    impl TopicRepo for InMemoryTopicRepo {
        async fn create(
            &self,
            app_id: AppId,
            name: &str,
            external_subscribable: bool,
            auth_mode: TopicAuthMode,
        ) -> Result<Topic, TopicRepoError> {
            let mut g = self.inner.lock().await;
            if g.contains_key(&(app_id, name.to_string())) {
                return Err(TopicRepoError::AlreadyExists(name.to_string()));
            }
            let now = Utc::now();
            let t = Topic {
                name: name.to_string(),
                external_subscribable,
                auth_mode,
                created_at: now,
                updated_at: now,
            };
            g.insert((app_id, name.to_string()), t.clone());
            Ok(t)
        }
        async fn list(&self, app_id: AppId) -> Result<Vec<Topic>, TopicRepoError> {
            let g = self.inner.lock().await;
            let mut v: Vec<Topic> = g
                .iter()
                .filter(|((a, _), _)| *a == app_id)
                .map(|(_, t)| t.clone())
                .collect();
            v.sort_by(|a, b| a.name.cmp(&b.name));
            Ok(v)
        }
        async fn get(&self, app_id: AppId, name: &str) -> Result<Option<Topic>, TopicRepoError> {
            Ok(self
                .inner
                .lock()
                .await
                .get(&(app_id, name.to_string()))
                .cloned())
        }
        async fn update(
            &self,
            app_id: AppId,
            name: &str,
            external_subscribable: Option<bool>,
            auth_mode: Option<TopicAuthMode>,
        ) -> Result<Option<Topic>, TopicRepoError> {
            let mut g = self.inner.lock().await;
            let Some(t) = g.get_mut(&(app_id, name.to_string())) else {
                return Ok(None);
            };
            if let Some(e) = external_subscribable {
                t.external_subscribable = e;
            }
            if let Some(m) = auth_mode {
                t.auth_mode = m;
            }
            t.updated_at = Utc::now();
            Ok(Some(t.clone()))
        }
        async fn delete(&self, app_id: AppId, name: &str) -> Result<bool, TopicRepoError> {
            Ok(self
                .inner
                .lock()
                .await
                .remove(&(app_id, name.to_string()))
                .is_some())
        }
    }

    struct InMemoryAppRepo(AppId);
    #[async_trait]
    impl AppRepository for InMemoryAppRepo {
        async fn list(&self) -> Result<Vec<App>, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn list_for_user(&self, _: AdminUserId) -> Result<Vec<App>, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn get_by_id(&self, id: AppId) -> Result<Option<App>, ScriptRepositoryError> {
            if id != self.0 {
                return Ok(None);
            }
            let now = Utc::now();
            Ok(Some(App {
                id,
                slug: "test".into(),
                name: "test".into(),
                description: None,
                created_at: now,
                updated_at: now,
            }))
        }
        async fn get_by_slug(&self, _: &str) -> Result<Option<App>, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn get_by_slug_or_history(
            &self,
            _: &str,
        ) -> Result<Option<crate::app_repo::AppLookup>, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn slug_in_history(&self, _: &str) -> Result<Option<App>, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn create(
            &self,
            _: &str,
            _: &str,
            _: Option<&str>,
        ) -> Result<App, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn create_with_takeover(
            &self,
            _: &str,
            _: &str,
            _: Option<&str>,
        ) -> Result<App, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn update(
            &self,
            _: AppId,
            _: Option<&str>,
            _: Option<Option<&str>>,
        ) -> Result<App, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn rename_slug(
            &self,
            _: AppId,
            _: &str,
            _: bool,
        ) -> Result<App, ScriptRepositoryError> {
            unimplemented!()
        }
        async fn delete(&self, _: AppId) -> Result<(), ScriptRepositoryError> {
            unimplemented!()
        }
        async fn delete_cascade(&self, _: AppId) -> Result<(), ScriptRepositoryError> {
            unimplemented!()
        }
        async fn count_scripts_in_app(&self, _: AppId) -> Result<i64, ScriptRepositoryError> {
            unimplemented!()
        }
    }

    /// Grants `AppAdmin` only for `granted_app`; denies elsewhere — used
    /// for the cross-app isolation test.
    struct PerAppAuthzRepo {
        granted_app: AppId,
    }
    #[async_trait]
    impl AuthzRepo for PerAppAuthzRepo {
        async fn membership(
            &self,
            _: UserId,
            app_id: AppId,
        ) -> Result<Option<AppRole>, AuthzError> {
            Ok((app_id == self.granted_app).then_some(AppRole::AppAdmin))
        }
    }

    struct DenyAuthzRepo;
    #[async_trait]
    impl AuthzRepo for DenyAuthzRepo {
        async fn membership(&self, _: UserId, _: AppId) -> Result<Option<AppRole>, AuthzError> {
            Ok(None)
        }
    }

    #[derive(Default)]
    struct RecordingBroadcaster {
        dropped: StdMutex<Vec<(AppId, String)>>,
    }
    #[async_trait]
    impl RealtimeBroadcaster for RecordingBroadcaster {
        async fn subscribe(
            &self,
            _: AppId,
            _: &str,
        ) -> Result<tokio::sync::broadcast::Receiver<RealtimeEvent>, BroadcasterError> {
            unimplemented!()
        }
        async fn publish(&self, _: AppId, _: &str, _: RealtimeEvent) {}
        async fn drop_topic(&self, app_id: AppId, topic: &str) {
            self.dropped
                .lock()
                .unwrap()
                .push((app_id, topic.to_string()));
        }
    }

    fn member() -> Principal {
        Principal {
            user_id: AdminUserId::new(),
            instance_role: InstanceRole::Member,
            scopes: None,
            app_binding: None,
        }
    }

    fn state(app_id: AppId, authz: Arc<dyn AuthzRepo>) -> (TopicsState, Arc<RecordingBroadcaster>) {
        let bc = Arc::new(RecordingBroadcaster::default());
        let state = TopicsState {
            topics: Arc::new(InMemoryTopicRepo::default()),
            apps: Arc::new(InMemoryAppRepo(app_id)),
            authz,
            broadcaster: bc.clone(),
        };
        (state, bc)
    }

    #[tokio::test]
    async fn register_defaults_external_subscribable_false() {
        let app = AppId::new();
        let (s, _) = state(app, Arc::new(PerAppAuthzRepo { granted_app: app }));
        let (status, Json(topic)) = create_topic(
            State(s),
            Extension(member()),
            Path(app),
            Json(CreateTopicRequest {
                name: "chat".into(),
                external_subscribable: false,
                auth_mode: TopicAuthMode::Public,
            }),
        )
        .await
        .unwrap();
        assert_eq!(status, StatusCode::CREATED);
        assert!(!topic.external_subscribable);
        assert_eq!(topic.name, "chat");
    }

    #[tokio::test]
    async fn flip_requires_app_admin_role() {
        let app = AppId::new();
        // Topic exists; the caller has no role → PATCH is forbidden.
        let (s, _) = state(app, Arc::new(DenyAuthzRepo));
        s.topics
            .create(app, "chat", false, TopicAuthMode::Public)
            .await
            .unwrap();
        let err = update_topic(
            State(s),
            Extension(member()),
            Path((app, "chat".to_string())),
            Json(UpdateTopicRequest {
                external_subscribable: Some(true),
                auth_mode: None,
            }),
        )
        .await
        .unwrap_err();
        assert!(matches!(err, TopicsApiError::Forbidden));
    }

    #[tokio::test]
    async fn flip_is_its_own_endpoint_and_toggles_external() {
        // The PATCH handler is a distinct surface from create; flipping
        // external_subscribable false→true is a single discrete call.
        let app = AppId::new();
        let (s, _) = state(app, Arc::new(PerAppAuthzRepo { granted_app: app }));
        s.topics
            .create(app, "chat", false, TopicAuthMode::Public)
            .await
            .unwrap();
        let Json(updated) = update_topic(
            State(s),
            Extension(member()),
            Path((app, "chat".to_string())),
            Json(UpdateTopicRequest {
                external_subscribable: Some(true),
                auth_mode: Some(TopicAuthMode::Token),
            }),
        )
        .await
        .unwrap();
        assert!(updated.external_subscribable);
        assert_eq!(updated.auth_mode, TopicAuthMode::Token);
    }

    #[tokio::test]
    async fn delete_disconnects_subscribers() {
        let app = AppId::new();
        let (s, bc) = state(app, Arc::new(PerAppAuthzRepo { granted_app: app }));
        s.topics
            .create(app, "chat", true, TopicAuthMode::Public)
            .await
            .unwrap();
        let status = delete_topic(
            State(s),
            Extension(member()),
            Path((app, "chat".to_string())),
        )
        .await
        .unwrap();
        assert_eq!(status, StatusCode::NO_CONTENT);
        assert_eq!(
            bc.dropped.lock().unwrap().as_slice(),
            &[(app, "chat".to_string())]
        );
    }

    #[tokio::test]
    async fn cross_app_admin_cannot_manage_other_app() {
        let app_a = AppId::new();
        let app_b = AppId::new();
        // Caller is admin of app A only; both apps exist via separate state.
        let authz = Arc::new(PerAppAuthzRepo { granted_app: app_a });
        // App-B-scoped state, but the caller only has A's grant.
        let (s, _) = state(app_b, authz);
        let err = create_topic(
            State(s),
            Extension(member()),
            Path(app_b),
            Json(CreateTopicRequest {
                name: "chat".into(),
                external_subscribable: true,
                auth_mode: TopicAuthMode::Public,
            }),
        )
        .await
        .unwrap_err();
        assert!(matches!(err, TopicsApiError::Forbidden));
    }

    #[tokio::test]
    async fn pattern_name_rejected() {
        let app = AppId::new();
        let (s, _) = state(app, Arc::new(PerAppAuthzRepo { granted_app: app }));
        let err = create_topic(
            State(s),
            Extension(member()),
            Path(app),
            Json(CreateTopicRequest {
                name: "user.*".into(),
                external_subscribable: true,
                auth_mode: TopicAuthMode::Public,
            }),
        )
        .await
        .unwrap_err();
        assert!(matches!(err, TopicsApiError::Invalid(_)));
    }
}