feat: SQS triggers

backend/Cargo.toml

@@ -74,6 +74,7 @@ mssql = ["windmill-worker/mssql"]
 bigquery = ["windmill-worker/bigquery"]
 websocket = ["windmill-api/websocket"]
 postgres_trigger = ["windmill-api/postgres_trigger"]
+sqs_trigger = ["windmill-api/sqs_trigger"]
 python = ["windmill-worker/python"]
 smtp = ["windmill-api/smtp", "windmill-common/smtp"]
 csharp = ["windmill-worker/csharp"]
@@ -306,6 +307,7 @@ datafusion = "39.0.0"
 object_store = { version = "0.10.0", features = ["aws", "azure"] }
 openidconnect = { version = "4.0.0-rc.1" }
 aws-config = "^1"
+aws-sdk-sqs = "1.57.0"
 aws-sdk-sts = "^1"
 
 crc = "^3"

backend/migrations/20250130184358_sqs_trigger.down.sql

@@ -0,0 +1,2 @@
+-- Add down migration script here
+DROP TABLE IF EXISTS sqs_trigger;

backend/migrations/20250130184358_sqs_trigger.up.sql

@@ -0,0 +1,69 @@
+-- Add up migration script here
+CREATE TABLE sqs_trigger(
+    path VARCHAR(255) NOT NULL,
+    queue_url VARCHAR(255) NOT NULL,
+    aws_resource_path VARCHAR(255) NOT NULL,
+    message_attributes TEXT[],
+    script_path VARCHAR(255) NOT NULL,
+    is_flow BOOLEAN NOT NULL,
+    workspace_id VARCHAR(50) NOT NULL,
+    edited_by VARCHAR(50) NOT NULL,
+    email VARCHAR(255) NOT NULL,
+    edited_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    extra_perms JSONB NULL,
+    error TEXT NULL,
+    server_id VARCHAR(50) NULL,
+    last_server_ping TIMESTAMPTZ NULL,
+    enabled BOOLEAN NOT NULL,
+    CONSTRAINT PK_sqs_trigger PRIMARY KEY (path,workspace_id),
+    CONSTRAINT fk_sqs_trigger_workspace FOREIGN KEY (workspace_id)
+        REFERENCES workspace(id) ON DELETE CASCADE
+);
+
+GRANT ALL ON sqs_trigger TO windmill_user;
+GRANT ALL ON sqs_trigger TO windmill_admin;
+
+ALTER TABLE sqs_trigger ENABLE ROW LEVEL SECURITY;
+
+CREATE POLICY admin_policy ON sqs_trigger FOR ALL TO windmill_admin USING (true);
+
+CREATE POLICY see_folder_extra_perms_user_select ON sqs_trigger FOR SELECT TO windmill_user
+USING (SPLIT_PART(sqs_trigger.path, '/', 1) = 'f' AND SPLIT_PART(sqs_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_read'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_insert ON sqs_trigger FOR INSERT TO windmill_user
+WITH CHECK (SPLIT_PART(sqs_trigger.path, '/', 1) = 'f' AND SPLIT_PART(sqs_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_update ON sqs_trigger FOR UPDATE TO windmill_user
+USING (SPLIT_PART(sqs_trigger.path, '/', 1) = 'f' AND SPLIT_PART(sqs_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+CREATE POLICY see_folder_extra_perms_user_delete ON sqs_trigger FOR DELETE TO windmill_user
+USING (SPLIT_PART(sqs_trigger.path, '/', 1) = 'f' AND SPLIT_PART(sqs_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
+
+CREATE POLICY see_own ON sqs_trigger FOR ALL TO windmill_user
+USING (SPLIT_PART(sqs_trigger.path, '/', 1) = 'u' AND SPLIT_PART(sqs_trigger.path, '/', 2) = current_setting('session.user'));
+CREATE POLICY see_member ON sqs_trigger FOR ALL TO windmill_user
+USING (SPLIT_PART(sqs_trigger.path, '/', 1) = 'g' AND SPLIT_PART(sqs_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.groups'), ',')::text[]));
+
+CREATE POLICY see_extra_perms_user_select ON sqs_trigger FOR SELECT TO windmill_user
+USING (extra_perms ? CONCAT('u/', current_setting('session.user')));
+CREATE POLICY see_extra_perms_user_insert ON sqs_trigger FOR INSERT TO windmill_user
+WITH CHECK ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+CREATE POLICY see_extra_perms_user_update ON sqs_trigger FOR UPDATE TO windmill_user
+USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+CREATE POLICY see_extra_perms_user_delete ON sqs_trigger FOR DELETE TO windmill_user
+USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
+
+CREATE POLICY see_extra_perms_groups_select ON sqs_trigger FOR SELECT TO windmill_user
+USING (extra_perms ?| regexp_split_to_array(current_setting('session.pgroups'), ',')::text[]);
+CREATE POLICY see_extra_perms_groups_insert ON sqs_trigger FOR INSERT TO windmill_user
+WITH CHECK (exists(
+    SELECT key, value FROM jsonb_each_text(extra_perms) 
+    WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
+    AND value::boolean));
+CREATE POLICY see_extra_perms_groups_update ON sqs_trigger FOR UPDATE TO windmill_user
+USING (exists(
+    SELECT key, value FROM jsonb_each_text(extra_perms) 
+    WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
+    AND value::boolean));
+CREATE POLICY see_extra_perms_groups_delete ON sqs_trigger FOR DELETE  TO windmill_user
+USING (exists(
+    SELECT key, value FROM jsonb_each_text(extra_perms) 
+    WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
+    AND value::boolean));

backend/migrations/20250205003539_add_sqs_type_value_to_trigger_kind_type.down.sql

@@ -0,0 +1,3 @@
+-- Add down migration script here
+DROP TYPE TRIGGER_KIND;
+CREATE TYPE TRIGGER_KIND AS ENUM ('webhook', 'http', 'websocket', 'kafka', 'email');

backend/migrations/20250205003539_add_sqs_type_value_to_trigger_kind_type.up.sql

@@ -0,0 +1,2 @@
+-- Add up migration script here
+ALTER TYPE TRIGGER_KIND ADD VALUE 'sqs';

backend/windmill-api/Cargo.toml

@@ -29,6 +29,7 @@ oauth2 = ["dep:async-oauth2"]
 http_trigger = ["dep:matchit"]
 static_frontend = ["dep:rust-embed"]
 postgres_trigger = ["dep:rust-postgres", "dep:pg_escape", "dep:byteorder", "dep:thiserror", "dep:rust_decimal", "dep:rust-postgres-native-tls"]
+sqs_trigger = ["dep:aws-sdk-sqs", "dep:thiserror", "dep:aws-config"]
 
 [dependencies]
 windmill-queue.workspace = true
@@ -117,4 +118,6 @@ pg_escape = { workspace = true, optional = true }
 byteorder = { workspace = true, optional = true }
 thiserror = { workspace = true, optional = true }
 rust_decimal = { workspace = true, optional = true }
-rust-postgres-native-tls = { workspace = true, optional = true}
+rust-postgres-native-tls = { workspace = true, optional = true}
+aws-sdk-sqs = { workspace = true, optional = true } 
+aws-config = { workspace = true, optional = true}