phonograph/interim-server/src/routes/workspaces_single/service_credentials_handler.rs

use anyhow::anyhow;
use askama::Template;
use axum::{
    debug_handler,
    extract::{Path, State},
    response::{Html, IntoResponse},
};
use futures::{lock::Mutex, prelude::*, stream};
use interim_models::{service_cred::ServiceCred, workspace::Workspace};
use interim_pgtypes::{pg_class::PgClass, pg_role::RoleTree};
use redact::Secret;
use serde::Deserialize;
use url::Url;
use uuid::Uuid;

use crate::{
    app::{App, AppDbConn},
    errors::AppError,
    navigator::Navigator,
    roles::RoleDisplay,
    settings::Settings,
    user::CurrentUser,
    workspace_nav::WorkspaceNav,
    workspace_pooler::{RoleAssignment, WorkspacePooler},
    workspace_utils::PHONO_TABLE_NAMESPACE,
};

#[derive(Debug, Deserialize)]
pub(super) struct PathParams {
    workspace_id: Uuid,
}

/// HTTP GET handler for the page at which a user manages their service
/// credentials for a workspace.
#[debug_handler(state = App)]
pub(super) async fn get(
    State(settings): State<Settings>,
    CurrentUser(user): CurrentUser,
    AppDbConn(mut app_db): AppDbConn,
    Path(PathParams { workspace_id }): Path<PathParams>,
    navigator: Navigator,
    State(mut pooler): State<WorkspacePooler>,
) -> Result<impl IntoResponse, AppError> {
    // FIXME: auth

    let workspace = Workspace::with_id(workspace_id)
        .fetch_one(&mut app_db)
        .await?;
    let cluster = workspace.fetch_cluster(&mut app_db).await?;

    // Mutex is required to use client in async closures.
    let workspace_client = Mutex::new(
        pooler
            .acquire_for(workspace_id, RoleAssignment::User(user.id))
            .await?,
    );

    struct ServiceCredInfo {
        service_cred: ServiceCred,
        member_of: Vec<RoleDisplay>,
        conn_string: Secret<Url>,
        conn_string_redacted: String,
    }

    impl ServiceCredInfo {
        fn is_reader_of(&self, relname: &str) -> bool {
            self.member_of.iter().any(|role_display| {
                if let RoleDisplay::TableReader {
                    relname: role_relname,
                    ..
                } = role_display
                {
                    role_relname == relname
                } else {
                    false
                }
            })
        }

        fn is_writer_of(&self, relname: &str) -> bool {
            self.member_of.iter().any(|role_display| {
                if let RoleDisplay::TableWriter {
                    relname: role_relname,
                    ..
                } = role_display
                {
                    role_relname == relname
                } else {
                    false
                }
            })
        }

        fn is_owner_of(&self, relname: &str) -> bool {
            self.member_of.iter().any(|role_display| {
                if let RoleDisplay::TableOwner {
                    relname: role_relname,
                    ..
                } = role_display
                {
                    role_relname == relname
                } else {
                    false
                }
            })
        }
    }

    let service_cred_info = stream::iter(
        ServiceCred::belonging_to_user(user.id)
            .fetch_all(&mut app_db)
            .await?,
    )
    .then(async |cred| {
        let member_of: Vec<RoleDisplay> = stream::iter({
            // Guard must be assigned to a local variable,
            // lest the mutex become deadlocked.
            let mut locked_client = workspace_client.lock().await;
            RoleTree::granted_to_rolname(&cred.rolname)
                .fetch_tree(&mut locked_client)
                .await?
                .ok_or(anyhow!("listing roles for service cred: role tree is None"))?
                .flatten_inherited()
                .into_iter()
                .filter(|role| role.rolname != cred.rolname)
        })
        .then(async |role| {
            let mut locked_client = workspace_client.lock().await;
            RoleDisplay::from_rolname(&role.rolname, &mut locked_client).await
        })
        .collect::<Vec<Result<_, _>>>()
        .await
        .into_iter()
        // [`futures::stream::StreamExt::collect`]
        // can only collect to types that implement [`Default`],
        // so we must do result handling with the sync version of `collect()`.
        .collect::<Result<Vec<_>, _>>()?
        .into_iter()
        .flatten()
        .collect();
        let conn_string = cluster.conn_str_for_db(
            &workspace.db_name,
            Some((
                cred.rolname.as_str(),
                Secret::new(cred.password.expose_secret().as_str()),
            )),
        )?;
        Ok(ServiceCredInfo {
            conn_string,
            conn_string_redacted: "postgresql://********".to_owned(),
            member_of,
            service_cred: cred,
        })
    })
    .collect::<Vec<Result<ServiceCredInfo, AppError>>>()
    .await
    .into_iter()
    .collect::<Result<Vec<ServiceCredInfo>, AppError>>()?;

    #[derive(Template)]
    #[template(path = "workspaces_single/service_credentials.html")]
    struct ResponseTemplate {
        all_rels: Vec<PgClass>,
        service_cred_info: Vec<ServiceCredInfo>,
        settings: Settings,
        workspace_nav: WorkspaceNav,
    }

    let mut locked_client = workspace_client
        .try_lock()
        .ok_or(anyhow!("mutex should be unlocked"))?;
    Ok(Html(
        ResponseTemplate {
            all_rels: PgClass::belonging_to_namespace(PHONO_TABLE_NAMESPACE)
                .fetch_all(&mut locked_client)
                .await?
                .into_iter()
                .filter(|rel| rel.relkind == 'r' as i8)
                .collect(),
            workspace_nav: WorkspaceNav::builder()
                .navigator(navigator)
                .workspace(workspace.clone())
                .populate_rels(&mut app_db, &mut locked_client)
                .await?
                .build()?,
            service_cred_info,
            settings,
        }
        .render()?,
    ))
}