use axum::{ extract::{Extension, Json, Path, Query}, routing::{delete, get, post}, Router, }; use serde::Deserialize; use windmill_audit::{audit_oss::audit_log, ActionKind}; use windmill_common::{ db::UserDB, error::{Error, Result}, trashbin::{self, TrashItem, TrashItemWithData}, trigger_history::{ self, TriggerHistoryEvent, TriggerOperation, TriggerSource, SCHEDULE_TRIGGER_KIND, }, utils::require_admin, }; use crate::db::{ApiAuthed, DB}; pub fn workspaced_service() -> Router { Router::new() .route("/list", get(list_trash)) .route("/get/{id}", get(get_trash_item)) .route("/restore/{id}", post(restore_trash_item)) .route("/delete/{id}", delete(permanently_delete_item)) .route("/empty", post(empty_trash)) } #[derive(Deserialize)] struct ListTrashQuery { item_kind: Option, page: Option, per_page: Option, } async fn list_trash( authed: ApiAuthed, Extension(db): Extension, Path(w_id): Path, Query(query): Query, ) -> Result>> { require_admin(authed.is_admin, &authed.username)?; let items = trashbin::list_trash( &db, &w_id, query.item_kind.as_deref(), query.page, query.per_page, ) .await?; Ok(Json(items)) } async fn get_trash_item( authed: ApiAuthed, Extension(db): Extension, Path((w_id, id)): Path<(String, i64)>, ) -> Result> { require_admin(authed.is_admin, &authed.username)?; let item = trashbin::get_trash_item(&db, &w_id, id).await?; Ok(Json(item)) } async fn restore_trash_item( authed: ApiAuthed, Extension(db): Extension, Extension(user_db): Extension, Path((w_id, id)): Path<(String, i64)>, ) -> Result { require_admin(authed.is_admin, &authed.username)?; let item = trashbin::get_trash_item(&db, &w_id, id).await?; let mut tx = user_db.begin(&authed).await?; match item.item_kind.as_str() { "script" => restore_script(&mut tx, &item).await?, "flow" => restore_flow(&mut tx, &item).await?, "app" => restore_app(&mut tx, &item).await?, "schedule" => restore_schedule(&mut tx, &item).await?, "variable" => restore_variable(&mut tx, &item).await?, "resource" => restore_resource(&mut tx, &item).await?, kind if kind.ends_with("_trigger") => restore_trigger(&mut tx, &item).await?, _ => { return Err(Error::BadRequest(format!( "Unknown item kind: {}", item.item_kind ))) } } sqlx::query!("DELETE FROM trashbin WHERE id = $1", item.id) .execute(&mut *tx) .await?; // A restore puts the trigger back, so the history has to say so: otherwise // the last thing it records for a live trigger is its own deletion. The // trashed row is the snapshot, so this needs no extra read. let restored_trigger_kind = match item.item_kind.as_str() { SCHEDULE_TRIGGER_KIND => Some(SCHEDULE_TRIGGER_KIND), // `_trigger` is what `delete_trigger` trashes it under, and the // stem is the `TRIGGER_TYPE` the history records against. kind => kind.strip_suffix("_trigger"), }; if let Some(trigger_kind) = restored_trigger_kind { trigger_history::record( &mut *tx, TriggerHistoryEvent { workspace_id: &w_id, trigger_kind, path: &item.item_path, operation: TriggerOperation::Create, source: TriggerSource::of_request(authed.is_session_token), username: Some(&authed.username), changes: trigger_history::summarize_changes(None, item.item_data.get("row")), }, ) .await?; } audit_log( &mut *tx, &authed, "trash.restore", ActionKind::Create, &w_id, Some(&item.item_path), None, ) .await?; tx.commit().await?; Ok(format!("{} '{}' restored", item.item_kind, item.item_path)) } async fn permanently_delete_item( authed: ApiAuthed, Extension(db): Extension, Extension(user_db): Extension, Path((w_id, id)): Path<(String, i64)>, ) -> Result { require_admin(authed.is_admin, &authed.username)?; let item = trashbin::get_trash_item(&db, &w_id, id).await?; let mut tx = user_db.begin(&authed).await?; trashbin::permanently_delete_item(&mut *tx, &w_id, id).await?; audit_log( &mut *tx, &authed, "trashbin.permanently_delete", ActionKind::Delete, &w_id, Some(&item.item_path), None, ) .await?; tx.commit().await?; Ok("permanently deleted".to_string()) } async fn empty_trash( authed: ApiAuthed, Extension(_db): Extension, Extension(user_db): Extension, Path(w_id): Path, ) -> Result { require_admin(authed.is_admin, &authed.username)?; let mut tx = user_db.begin(&authed).await?; let count = trashbin::empty_trash(&mut *tx, &w_id).await?; audit_log( &mut *tx, &authed, "trashbin.empty", ActionKind::Delete, &w_id, None, None, ) .await?; tx.commit().await?; Ok(format!("{} items permanently deleted", count)) } // --- Restore functions per item kind --- async fn restore_script(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> { let data = &item.item_data; // Check for path conflict let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM script WHERE path = $1 AND workspace_id = $2)", &item.item_path, &item.workspace_id, ) .fetch_one(&mut *tx) .await? .unwrap_or(false); if exists { return Err(Error::BadRequest(format!( "A script already exists at path '{}'", item.item_path ))); } // Scripts are stored as an array (all versions for the path) let scripts = data .get("scripts") .and_then(|v| v.as_array()) .ok_or_else(|| Error::internal_err("Invalid trash data for script"))?; for script in scripts { sqlx::query("INSERT INTO script SELECT * FROM jsonb_populate_record(null::script, $1)") .bind(script) .execute(&mut *tx) .await .map_err(|e| Error::internal_err(format!("restoring script: {e:#}")))?; } // Restore drafts if present if let Some(drafts) = data.get("drafts").and_then(|v| v.as_array()) { for draft in drafts { sqlx::query( "INSERT INTO draft SELECT * FROM jsonb_populate_record(null::draft, $1) ON CONFLICT DO NOTHING", ) .bind(draft) .execute(&mut *tx) .await?; } } Ok(()) } async fn restore_flow(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> { let data = &item.item_data; let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM flow WHERE path = $1 AND workspace_id = $2)", &item.item_path, &item.workspace_id, ) .fetch_one(&mut *tx) .await? .unwrap_or(false); if exists { return Err(Error::BadRequest(format!( "A flow already exists at path '{}'", item.item_path ))); } let row = data .get("row") .ok_or_else(|| Error::internal_err("Invalid trash data for flow"))?; sqlx::query("INSERT INTO flow SELECT * FROM jsonb_populate_record(null::flow, $1)") .bind(row) .execute(&mut *tx) .await?; // Restore flow_versions if let Some(versions) = data.get("flow_versions").and_then(|v| v.as_array()) { for version in versions { sqlx::query( "INSERT INTO flow_version SELECT * FROM jsonb_populate_record(null::flow_version, $1) ON CONFLICT DO NOTHING", ) .bind(version) .execute(&mut *tx) .await?; } } // Restore flow_nodes if let Some(nodes) = data.get("flow_nodes").and_then(|v| v.as_array()) { for node in nodes { sqlx::query( "INSERT INTO flow_node SELECT * FROM jsonb_populate_record(null::flow_node, $1) ON CONFLICT DO NOTHING", ) .bind(node) .execute(&mut *tx) .await?; } } // Restore drafts if let Some(drafts) = data.get("drafts").and_then(|v| v.as_array()) { for draft in drafts { sqlx::query( "INSERT INTO draft SELECT * FROM jsonb_populate_record(null::draft, $1) ON CONFLICT DO NOTHING", ) .bind(draft) .execute(&mut *tx) .await?; } } Ok(()) } async fn restore_app(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> { let data = &item.item_data; let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM app WHERE path = $1 AND workspace_id = $2)", &item.item_path, &item.workspace_id, ) .fetch_one(&mut *tx) .await? .unwrap_or(false); if exists { return Err(Error::BadRequest(format!( "An app already exists at path '{}'", item.item_path ))); } let row = data .get("row") .ok_or_else(|| Error::internal_err("Invalid trash data for app"))?; // Insert app first (app_version has FK to app.id) sqlx::query("INSERT INTO app SELECT * FROM jsonb_populate_record(null::app, $1)") .bind(row) .execute(&mut *tx) .await .map_err(|e| Error::internal_err(format!("restoring app row: {e:#}")))?; // Then restore app_versions if let Some(versions) = data.get("app_versions").and_then(|v| v.as_array()) { for version in versions { sqlx::query( "INSERT INTO app_version SELECT * FROM jsonb_populate_record(null::app_version, $1) ON CONFLICT DO NOTHING", ) .bind(version) .execute(&mut *tx) .await .map_err(|e| Error::internal_err(format!("restoring app_version: {e:#}")))?; } } // Restore drafts if let Some(drafts) = data.get("drafts").and_then(|v| v.as_array()) { for draft in drafts { sqlx::query( "INSERT INTO draft SELECT * FROM jsonb_populate_record(null::draft, $1) ON CONFLICT DO NOTHING", ) .bind(draft) .execute(&mut *tx) .await?; } } Ok(()) } async fn restore_schedule(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> { let data = &item.item_data; let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM schedule WHERE path = $1 AND workspace_id = $2)", &item.item_path, &item.workspace_id, ) .fetch_one(&mut *tx) .await? .unwrap_or(false); if exists { return Err(Error::BadRequest(format!( "A schedule already exists at path '{}'", item.item_path ))); } let row = data .get("row") .ok_or_else(|| Error::internal_err("Invalid trash data for schedule"))?; sqlx::query("INSERT INTO schedule SELECT * FROM jsonb_populate_record(null::schedule, $1)") .bind(row) .execute(&mut *tx) .await?; Ok(()) } async fn restore_variable(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> { let data = &item.item_data; let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM variable WHERE path = $1 AND workspace_id = $2)", &item.item_path, &item.workspace_id, ) .fetch_one(&mut *tx) .await? .unwrap_or(false); if exists { return Err(Error::BadRequest(format!( "A variable already exists at path '{}'", item.item_path ))); } let row = data .get("row") .ok_or_else(|| Error::internal_err("Invalid trash data for variable"))?; sqlx::query("INSERT INTO variable SELECT * FROM jsonb_populate_record(null::variable, $1)") .bind(row) .execute(&mut *tx) .await?; // Restore linked resource if present if let Some(linked_resource) = data.get("linked_resource") { if !linked_resource.is_null() { sqlx::query( "INSERT INTO resource SELECT * FROM jsonb_populate_record(null::resource, $1) ON CONFLICT DO NOTHING", ) .bind(linked_resource) .execute(&mut *tx) .await?; } } Ok(()) } async fn restore_resource(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> { let data = &item.item_data; let exists = sqlx::query_scalar!( "SELECT EXISTS(SELECT 1 FROM resource WHERE path = $1 AND workspace_id = $2)", &item.item_path, &item.workspace_id, ) .fetch_one(&mut *tx) .await? .unwrap_or(false); if exists { return Err(Error::BadRequest(format!( "A resource already exists at path '{}'", item.item_path ))); } let row = data .get("row") .ok_or_else(|| Error::internal_err("Invalid trash data for resource"))?; sqlx::query("INSERT INTO resource SELECT * FROM jsonb_populate_record(null::resource, $1)") .bind(row) .execute(&mut *tx) .await?; // Restore linked variables if present if let Some(linked_vars) = data.get("linked_variables").and_then(|v| v.as_array()) { for var in linked_vars { sqlx::query( "INSERT INTO variable SELECT * FROM jsonb_populate_record(null::variable, $1) ON CONFLICT DO NOTHING", ) .bind(var) .execute(&mut *tx) .await?; } } Ok(()) } async fn restore_trigger(tx: &mut sqlx::PgConnection, item: &TrashItemWithData) -> Result<()> { let data = &item.item_data; let table_name = data .get("table_name") .and_then(|v| v.as_str()) .ok_or_else(|| Error::internal_err("Invalid trash data for trigger: missing table_name"))?; // Validate table name to prevent SQL injection let valid_tables = [ "http_trigger", "websocket_trigger", "kafka_trigger", "nats_trigger", "postgres_trigger", "mqtt_trigger", "amqp_trigger", "sqs_trigger", "gcp_trigger", "azure_trigger", "email_trigger", ]; if !valid_tables.contains(&table_name) { return Err(Error::BadRequest(format!( "Invalid trigger table: {}", table_name ))); } // SAFETY: `table_name` has been validated against the `valid_tables` allowlist above. let exists: bool = sqlx::query_scalar(&format!( "SELECT EXISTS(SELECT 1 FROM {} WHERE path = $1 AND workspace_id = $2)", table_name )) .bind(&item.item_path) .bind(&item.workspace_id) .fetch_one(&mut *tx) .await?; if exists { return Err(Error::BadRequest(format!( "A trigger already exists at path '{}'", item.item_path ))); } let row = data .get("row") .ok_or_else(|| Error::internal_err("Invalid trash data for trigger"))?; // SAFETY: `table_name` has been validated against the `valid_tables` allowlist above. sqlx::query(&format!( "INSERT INTO {} SELECT * FROM jsonb_populate_record(null::{}, $1)", table_name, table_name )) .bind(row) .execute(&mut *tx) .await?; Ok(()) }