phonograph/interim-server/src/routes/relations_single/insert_handler.rs

use std::collections::HashMap;

use axum::{
    debug_handler,
    extract::{Path, State},
    response::Response,
};
// [`axum_extra`]'s form extractor is required to support repeated keys:
// https://docs.rs/axum-extra/0.10.1/axum_extra/extract/struct.Form.html#differences-from-axumextractform
use axum_extra::extract::Form;
use interim_models::{
    datum::Datum,
    portal::Portal,
    workspace::Workspace,
    workspace_user_perm::{self, WorkspaceUserPerm},
};
use interim_pgtypes::{escape_identifier, pg_class::PgClass};
use serde::Deserialize;
use sqlx::{postgres::types::Oid, query};
use uuid::Uuid;

use crate::{
    app::{App, AppDbConn},
    errors::{AppError, forbidden},
    navigator::{Navigator, NavigatorPage as _},
    user::CurrentUser,
    workspace_pooler::{RoleAssignment, WorkspacePooler},
};

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

/// HTTP POST handler for inserting one or more rows into a table. This handler
/// takes a form where the keys are column names, with keys optionally repeated
/// to insert multiple rows at once. If any key is repeated, the others should
/// be repeated the same number of times. Form values are expected to be JSON-
/// serialized representations of the `[Datum]` type.
#[debug_handler(state = App)]
pub(super) async fn post(
    State(mut workspace_pooler): State<WorkspacePooler>,
    AppDbConn(mut app_db): AppDbConn,
    CurrentUser(user): CurrentUser,
    navigator: Navigator,
    Path(PathParams {
        portal_id,
        rel_oid,
        workspace_id,
    }): Path<PathParams>,
    Form(form): Form<HashMap<String, Vec<String>>>,
) -> Result<Response, AppError> {
    // Check workspace authorization.
    let workspace_perms = WorkspaceUserPerm::belonging_to_user(user.id)
        .fetch_all(&mut app_db)
        .await?;
    if workspace_perms.iter().all(|p| {
        p.workspace_id != workspace_id || p.perm != workspace_user_perm::PermissionValue::Connect
    }) {
        return Err(forbidden!("access denied to workspace"));
    }
    // FIXME ensure workspace corresponds to rel/portal, and that user has
    // permission to access/alter both as needed.
    // FIXME CSRF

    let portal = Portal::with_id(portal_id).fetch_one(&mut app_db).await?;
    let workspace = Workspace::with_id(portal.workspace_id)
        .fetch_one(&mut app_db)
        .await?;

    let mut workspace_client = workspace_pooler
        .acquire_for(workspace.id, RoleAssignment::User(user.id))
        .await?;

    let rel = PgClass::with_oid(Oid(rel_oid))
        .fetch_one(&mut workspace_client)
        .await?;

    let col_names: Vec<String> = form.keys().cloned().collect();

    // Prevent users from modifying Phonograph metadata columns.
    if col_names.iter().any(|col| col.starts_with('_')) {
        return Err(forbidden!("access denied to update system metadata column"));
    }

    let col_list_sql = col_names
        .iter()
        .map(|value| escape_identifier(value))
        .collect::<Vec<String>>()
        .join(", ");

    let n_rows = form.values().map(|value| value.len()).max().unwrap_or(0);
    if n_rows > 0 {
        let mut param_index = 1;
        let mut params: Vec<Datum> = vec![];
        let mut row_list: Vec<String> = vec![];
        for i in 0..n_rows {
            let mut param_slots: Vec<String> = vec![];
            for col in col_names.iter() {
                let maybe_value: Option<Datum> = form
                    .get(col)
                    .and_then(|col_values| col_values.get(i))
                    .map(|value_raw| serde_json::from_str(value_raw))
                    .transpose()?;
                if let Some(value) = maybe_value.filter(|value| !value.is_none()) {
                    params.push(value);
                    param_slots.push(format!("${param_index}"));
                    param_index += 1;
                } else {
                    param_slots.push("default".to_owned());
                }
            }
            row_list.push(format!("({0})", param_slots.join(", ")));
        }
        let row_list_sql = row_list.join(",\n");

        let query_sql = &format!(
            "insert into {ident} ({col_list_sql}) values {row_list_sql}",
            ident = rel.get_identifier(),
        );
        let mut q = query(query_sql);
        for param in params {
            q = param.bind_onto(q);
        }
        q.execute(workspace_client.get_conn()).await?;
    }

    Ok(navigator
        .portal_page()
        .workspace_id(workspace_id)
        .rel_oid(Oid(rel_oid))
        .portal_id(portal_id)
        .build()?
        .redirect_to())
}