Database triggers

backend/Cargo.toml

@@ -71,6 +71,7 @@ mysql = ["windmill-worker/mysql"]
 mssql = ["windmill-worker/mssql"]
 bigquery = ["windmill-worker/bigquery"]
 websocket = ["windmill-api/websocket"]
+database = ["windmill-api/database"]
 python = ["windmill-worker/python"]
 smtp = ["windmill-api/smtp", "windmill-common/smtp"]
 csharp = ["windmill-worker/csharp"]

backend/migrations/20241123152203_database_triggers.up.sql

@@ -1,5 +1,4 @@
 -- Add up migration script here
-CREATE TYPE transaction AS ENUM ('Insert', 'Update', 'Delete');
 CREATE TABLE database_trigger(
     path VARCHAR(255) NOT NULL,
     script_path VARCHAR(255) NOT NULL,
@@ -16,5 +15,7 @@ CREATE TABLE database_trigger(
     replication_slot_name VARCHAR(255) NOT NULL,
     publication_name VARCHAR(255) NOT NULL,
     enabled BOOLEAN NOT NULL,
-    CONSTRAINT PK_database_trigger PRIMARY KEY (path,workspace_id)
+    CONSTRAINT PK_database_trigger PRIMARY KEY (path,workspace_id),
+    CONSTRAINT fk_database_trigger_workspace FOREIGN KEY (workspace_id)
+        REFERENCES workspace(id) ON DELETE CASCADE
 );

backend/windmill-api/Cargo.toml

@@ -27,6 +27,7 @@ zip = ["dep:async_zip"]
 oauth2 = ["dep:async-oauth2"]
 http_trigger = ["dep:matchit"]
 static_frontend = ["dep:rust-embed"]
+database = ["dep:rust-postgres", "dep:pg_escape", "dep:byteorder", "dep:memchr", "dep:thiserror", "dep:rust_decimal"]
 
 [dependencies]
 windmill-queue.workspace = true
@@ -104,13 +105,13 @@ matchit = { workspace = true, optional = true }
 tokio-tungstenite = { workspace = true, optional = true}
 rdkafka = { workspace = true, optional = true }
 const_format.workspace = true
-rust-postgres.workspace = true
 pin-project.workspace = true
 http.workspace = true
 async-stream.workspace = true
 ulid.workspace = true
-pg_escape.workspace = true
-byteorder.workspace = true
-memchr.workspace = true
-thiserror.workspace = true
-rust_decimal.workspace = true
+rust-postgres = { workspace = true, optional = true }
+pg_escape = { workspace = true, optional = true }
+byteorder = { workspace = true, optional = true }
+memchr = { workspace = true, optional = true }
+thiserror = { workspace = true, optional = true }
+rust_decimal = { workspace = true, optional = true }

backend/windmill-api/openapi.yaml

@@ -8005,7 +8005,7 @@ paths:
               schema:
                 type: string
 
-  /w/{workspace}/database_triggers/get-template-script:
+  /w/{workspace}/database_triggers/get_template_script:
     post:
       summary: get template script
       operationId: getTemplateScript

backend/windmill-api/src/database_triggers/handler.rs

@@ -25,15 +25,14 @@ use windmill_common::error::Error;
 use windmill_common::{
     db::UserDB,
     error::{self, JsonResult},
-    resource::get_resource,
     utils::{not_found_if_none, paginate, Pagination, StripPath},
-    variables::get_variable_or_self,
     worker::CLOUD_HOSTED,
 };
 
 use crate::{
     database_triggers::mapper::{Mapper, MappingInfo},
     db::{ApiAuthed, DB},
+    resources::get_resource_value_interpolated_internal,
 };
 
 #[derive(FromRow, Serialize, Deserialize, Debug)]
@@ -228,19 +227,36 @@ pub struct DatabaseTrigger {
 }
 
 pub async fn get_database_resource(
+    authed: ApiAuthed,
+    user_db: UserDB,
     db: &DB,
     database_resource_path: &str,
     w_id: &str,
 ) -> Result<Database, Error> {
-    let mut resource = get_resource::<Database>(db, database_resource_path, w_id)
-        .await
-        .map_err(Error::SqlErr)?;
-
-    if !resource.value.password.is_empty() {
-        resource.value.password = get_variable_or_self(resource.value.password, db, w_id).await?;
-    }
+    let resource = get_resource_value_interpolated_internal(
+        &authed,
+        Some(user_db),
+        &db,
+        &w_id,
+        &database_resource_path,
+        None,
+        "",
+    )
+    .await
+    .map_err(|_| Error::NotFound("Database resource do not exist".to_string()))?;
+
+    let resource = match resource {
+        Some(resource) => serde_json::from_value::<Database>(resource).map_err(Error::SerdeJson)?,
+        None => {
+            return {
+                Err(Error::NotFound(
+                    "Database resource do not exist".to_string(),
+                ))
+            }
+        }
+    };
 
-    Ok(resource.value)
+    Ok(resource)
 }
 
 #[derive(Deserialize, Serialize)]
@@ -453,10 +469,13 @@ pub struct SlotList {
 }
 
 pub async fn list_slot_name(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
     Path((w_id, database_resource_path)): Path<(String, String)>,
 ) -> error::Result<Json<Vec<SlotList>>> {
-    let database = get_database_resource(&db, &database_resource_path, &w_id).await?;
+    let database =
+        get_database_resource(authed, user_db, &db, &database_resource_path, &w_id).await?;
 
     let mut connection = get_raw_postgres_connection(&database).await?;
 
@@ -485,11 +504,14 @@ pub struct Slot {
 }
 
 pub async fn create_slot(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
     Path((w_id, database_resource_path)): Path<(String, String)>,
     Json(Slot { name }): Json<Slot>,
 ) -> error::Result<String> {
-    let database = get_database_resource(&db, &database_resource_path, &w_id).await?;
+    let database =
+        get_database_resource(authed, user_db, &db, &database_resource_path, &w_id).await?;
 
     let mut connection = get_raw_postgres_connection(&database).await?;
 
@@ -508,11 +530,14 @@ pub async fn create_slot(
 }
 
 pub async fn drop_slot_name(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
     Path((w_id, database_resource_path)): Path<(String, String)>,
     Json(Slot { name }): Json<Slot>,
 ) -> error::Result<String> {
-    let database = get_database_resource(&db, &database_resource_path, &w_id).await?;
+    let database =
+        get_database_resource(authed, user_db, &db, &database_resource_path, &w_id).await?;
 
     let mut connection = get_raw_postgres_connection(&database).await?;
 
@@ -527,10 +552,13 @@ struct PublicationName {
 }
 
 pub async fn list_database_publication(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
     Path((w_id, database_resource_path)): Path<(String, String)>,
 ) -> error::Result<Json<Vec<String>>> {
-    let database = get_database_resource(&db, &database_resource_path, &w_id).await?;
+    let database =
+        get_database_resource(authed, user_db, &db, &database_resource_path, &w_id).await?;
 
     let mut connection = get_raw_postgres_connection(&database).await?;
 
@@ -550,10 +578,13 @@ pub async fn list_database_publication(
 }
 
 pub async fn get_publication_info(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
     Path((w_id, publication_name, database_resource_path)): Path<(String, String, String)>,
 ) -> error::Result<Json<PublicationData>> {
-    let database = get_database_resource(&db, &database_resource_path, &w_id).await?;
+    let database =
+        get_database_resource(authed, user_db, &db, &database_resource_path, &w_id).await?;
 
     let mut connection = get_raw_postgres_connection(&database).await?;
 
@@ -649,13 +680,16 @@ async fn new_publication(
 }
 
 pub async fn create_publication(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
     Path((w_id, publication_name, database_resource_path)): Path<(String, String, String)>,
     Json(publication_data): Json<PublicationData>,
 ) -> error::Result<String> {
     let PublicationData { table_to_track, transaction_to_track } = publication_data;
 
-    let database = get_database_resource(&db, &database_resource_path, &w_id).await?;
+    let database =
+        get_database_resource(authed, user_db, &db, &database_resource_path, &w_id).await?;
 
     let mut connection = get_raw_postgres_connection(&database).await?;
 
@@ -686,10 +720,14 @@ async fn drop_publication(
 }
 
 pub async fn delete_publication(
-    Path((w_id, publication_name, database_resource_path)): Path<(String, String, String)>,
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
+    Path((w_id, publication_name, database_resource_path)): Path<(String, String, String)>,
 ) -> error::Result<String> {
-    let database = get_database_resource(&db, &database_resource_path, &w_id).await?;
+    let database =
+        get_database_resource(authed, user_db, &db, &database_resource_path, &w_id).await?;
+
     let mut connection = get_raw_postgres_connection(&database).await?;
 
     drop_publication(&publication_name, &mut connection).await?;
@@ -701,12 +739,16 @@ pub async fn delete_publication(
 }
 
 pub async fn alter_publication(
-    Path((w_id, publication_name, database_resource_path)): Path<(String, String, String)>,
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
+    Path((w_id, publication_name, database_resource_path)): Path<(String, String, String)>,
     Json(publication_data): Json<PublicationData>,
 ) -> error::Result<String> {
     let PublicationData { table_to_track, transaction_to_track } = publication_data;
-    let database = get_database_resource(&db, &database_resource_path, &w_id).await?;
+    let database =
+        get_database_resource(authed, user_db, &db, &database_resource_path, &w_id).await?;
+
     let mut connection = get_raw_postgres_connection(&database).await?;
 
     let (all_table, _) =
@@ -1124,6 +1166,8 @@ pub async fn set_enabled(
 }
 
 pub async fn get_template_script(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
     Extension(db): Extension<DB>,
     Path(w_id): Path<String>,
     Json(template_script): Json<TemplateScript>,
@@ -1132,16 +1176,15 @@ pub async fn get_template_script(
 
     let TemplateScript { database_resource_path, relations, language } = template_script;
     if relations.is_none() {
-        return Err(error::Error::BadRequest(
+        return Err(Error::BadRequest(
             "You must at least choose schema to fetch table from".to_string(),
         ));
     }
 
-    let resource = get_resource::<Database>(&db, &database_resource_path, &w_id)
-        .await
-        .map_err(|_| Error::NotFound("Database resource do not exist".to_string()))?;
+    let database =
+        get_database_resource(authed, user_db, &db, &database_resource_path, &w_id).await?;
 
-    let mut pg_connection = get_raw_postgres_connection(&resource.value).await?;
+    let mut pg_connection = get_raw_postgres_connection(&database).await?;
 
     #[derive(Debug, FromRow, Deserialize)]
     struct ColumnInfo {

backend/windmill-api/src/database_triggers/trigger.rs

@@ -2,7 +2,6 @@ use std::{collections::HashMap, pin::Pin};
 
 use crate::{
     database_triggers::{
-        handler::get_database_resource,
         relation::RelationConverter,
         replication_message::{
             LogicalReplicationMessage::{Begin, Commit, Delete, Insert, Relation, Type, Update},
@@ -19,7 +18,7 @@ use pg_escape::{quote_identifier, quote_literal};
 use rand::seq::SliceRandom;
 use rust_postgres::{Client, Config, CopyBothDuplex, NoTls, SimpleQueryMessage};
 use serde_json::to_value;
-use windmill_common::{worker::to_raw_value, INSTANCE_NAME};
+use windmill_common::{variables::get_variable_or_self, worker::to_raw_value, INSTANCE_NAME};
 
 use super::{
     handler::{Database, DatabaseTrigger},
@@ -210,13 +209,32 @@ async fn listen_to_transactions(
     db: DB,
     mut killpill_rx: tokio::sync::broadcast::Receiver<()>,
 ) -> Result<(), Error> {
-    let resource = get_database_resource(
-        &db,
+    let resource = sqlx::query_scalar!(
+        "SELECT value from resource WHERE path = $1 AND workspace_id = $2",
         &database_trigger.database_resource_path,
-        &database_trigger.workspace_id,
+        &database_trigger.workspace_id
     )
+    .fetch_optional(&db)
     .await
-    .map_err(Error::Common)?;
+    .map_err(|e| Error::Common(windmill_common::error::Error::SqlErr(e)))?
+    .flatten();
+
+    let mut resource = match resource {
+        Some(resource) => serde_json::from_value::<Database>(resource)
+            .map_err(|e| Error::Common(windmill_common::error::Error::SerdeJson(e)))?,
+        None => {
+            return {
+                Err(Error::Common(windmill_common::error::Error::NotFound(
+                    "Database resource do not exist".to_string(),
+                )))
+            }
+        }
+    };
+
+    resource.password =
+        get_variable_or_self(resource.password, &db, &database_trigger.workspace_id)
+            .await
+            .map_err(Error::Common)?;
 
     let client = PostgresSimpleClient::new(&resource).await?;
 

backend/windmill-api/src/lib.rs

@@ -59,6 +59,7 @@ mod auth;
 mod capture;
 mod concurrency_groups;
 mod configs;
+#[cfg(feature = "database")]
 mod database_triggers;
 mod db;
 mod drafts;
@@ -287,8 +288,11 @@ pub async fn run_server(
             let kafka_killpill_rx = rx.resubscribe();
             kafka_triggers_ee::start_kafka_consumers(db.clone(), kafka_killpill_rx).await;
         }
-        let db_killpill_rx = rx.resubscribe();
-        database_triggers::start_database(db.clone(), db_killpill_rx).await;
+        #[cfg(feature = "database")]
+        {
+            let db_killpill_rx = rx.resubscribe();
+            database_triggers::start_database(db.clone(), db_killpill_rx).await;
+        }
     }
 
     // build our application with a route
@@ -358,10 +362,15 @@ pub async fn run_server(
                             Router::new()
                         })
                         .nest("/kafka_triggers", kafka_triggers_service)
-                        .nest(
-                            "/database_triggers",
-                            database_triggers::workspaced_service(),
-                        ),
+                        .nest("/database_triggers", {
+                            #[cfg(feature = "database")]
+                            {
+                                database_triggers::workspaced_service()
+                            }
+
+                            #[cfg(not(feature = "database"))]
+                            Router::new()
+                        }),
                 )
                 .nest("/workspaces", workspaces::global_service())
                 .nest(

backend/windmill-common/src/lib.rs

@@ -47,7 +47,6 @@ pub mod tracing_init;
 pub mod users;
 pub mod utils;
 pub mod variables;
-pub mod resource;
 pub mod worker;
 pub mod workspaces;