starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-08-25 00:01:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(triggers): add AMQP (RabbitMQ) trigger via lapin (#10230)

Browse Source 
* feat(triggers): add AMQP (RabbitMQ) trigger using the lapin library

Fixes WIN-2214

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* chore(triggers): defer AMQP cross-workspace deploy pending utils-internal publish

Revert the amqp_trigger additions to the shared windmill-utils-internal
TriggerDeployKind and the frontend cross-workspace deploy adapter: the
frontend installs the published npm package, which lacks the new kind
until a release is cut. AMQP create/edit/delete/list/sync/capture are
unaffected (they use local types); only cross-workspace deploy/merge of
AMQP triggers waits on the package bump. Also document the at-most-once
ack in the consumer loop.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(triggers): address AMQP review — at-least-once ack, workspace cascade, contracts

- ack AMQP deliveries only after successful dispatch; nack+requeue on failure
- add ON DELETE CASCADE workspace FK so amqp_trigger rows are cleaned on
  workspace deletion (and the listener stops)
- fix the /amqp_triggers/test OpenAPI body and add amqp_trigger to
  WorkspaceDiffRow.kind
- register AMQP in the generated workspace trigger tool (create_trigger)
- drop banned $bindable defaults on optional props in the config section
- add build_uri unit tests (encoding, ports, vhost)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(triggers): stop AMQP poison-message loop and reconnect on transient drops

Chaos testing against a live RabbitMQ broker showed the previous
nack(requeue) + immediate re-poll spun a tight redelivery loop (~1000
critical-error reports/sec) on a poison message, and any connection blip
permanently disabled the trigger (lapin has no built-in reconnect).

- on dispatch failure: nack+requeue then stop consuming; the listener
  framework re-lists the trigger after its ping goes stale (~15s), backing
  redelivery off to that cadence instead of a tight loop (verified: rate
  dropped from ~1000/s to ~1 per ~26s, message preserved)
- on connection/stream error: stop and let the framework reconnect instead
  of disabling; persistent failures are still disabled via get_consumer
  (verified: a forced connection close now auto-reconnects and resumes)
- finish the AI create-trigger action wiring for AMQP: add amqp to
  CreatedResourceTriggerKind, the action-card registry, and the drawer
  registry so the result card renders and its "Open" action works

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(triggers): complete AMQP frontend registries and defer merge rows

- add amqp to capturableTriggerTypes (so AmqpCapture mounts), the Runs
  jobTriggerKinds filter, and CLOUD_DISABLED_TRIGGER_TYPES
- wire AMQP into global AI chat mode: TRIGGER_KINDS, the request union,
  writeTriggerSchema, triggerServices, and the draft adapter
- stop emitting actionable AMQP fork-comparison rows (revert amqp_trigger
  from TRIGGER_OR_SCHEDULE_TABLES) since cross-workspace deploy is deferred
  until windmill-utils-internal is published — avoids a deploy that fails
  with "Unknown kind: amqp_trigger"
- use design-system TextInput instead of raw <input> in the config section

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(triggers): complete AMQP session/draft registries and constrain prefetch

- add amqp to the session-deploy, draft-compare, preview-router, and
  copilot workspace-item registries so AMQP drafts/deploys/nav/path
  resolution work
- include amqp_count in the MoveDrawer attached-trigger rename warning
- replace the raw prefetch <input> with a design-system TextInput bounded
  to an integer 1-65535 (backend u16) and block save on invalid values

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(triggers): make AMQP disconnect/reconnect consistent with the Kafka trigger

lapin, like rdkafka, has no transparent reconnect, so the AMQP listener now
mirrors the Kafka trigger's explicit reconnect loop instead of relying on the
framework re-list (which disabled the trigger once get_consumer failed on a
sustained outage):

- get_consumer returns cheaply; consume owns a (re)connect loop that retries
  with a 30s backoff, reports a critical error every 10 failed attempts, and
  reports a recovered critical error once it reconnects — never disabling the
  trigger on a connectivity failure
- a consumer/stream error breaks out to reconnect rather than disabling
- dispatch failure still nacks+requeues (at-least-once) with a short backoff
  to avoid a tight poison-message loop, keeping the connection alive

Verified against a live RabbitMQ broker: killing the broker keeps the trigger
enabled and retrying (attempt N), and restarting it auto-reconnects (logs
"reconnected after N attempts") and resumes dispatch.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(triggers): complete AMQP capture registries and constrain prefetch contract

- add the 'amqp' case to triggerKindToTriggerType so opening the AMQP editor
  from a capture button no longer throws "Unknown TriggerKind: amqp"
- register AmqpIcon in CaptureTable's icon map and add an AMQP entry to the
  script/flow CaptureButton menu
- bound the OpenAPI prefetch_count to an integer 1-65535 (matches the Rust
  u16) and regenerate clients/prompts
- require a non-empty exchange name when the exchange binding is enabled
- build_uri: fall back to "/" on a blank vhost and bracket IPv6 hosts (+ tests)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(triggers): wire AMQP into pipeline graph, git-sync, and preprocessor types

- asset_graph: discover attached amqp_trigger rows and emit an AMQP TriggerEdge
  so AMQP triggers render (and can be opened/deleted) on the data-pipeline canvas
- frontend pipeline graph: add amqp to NativeTriggerKind, the add-trigger menu,
  node presentation, event-trigger set, annotation keywords, and the
  editor/service registrations
- git-sync: add the amqp_trigger include pattern (+ test) so an AMQP git-sync
  deployment stages only its .amqp_trigger.* file, not an unrelated same-path object
- preprocessor starters: add the AMQP event to the generated TS/Python/PHP
  trigger event types (kind/payload/exchange/routing_key/queue_name/redelivered/
  delivery_tag)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(triggers): finish AMQP pipeline/parser wiring, prefetch validation, source lists

- fix a stray edit that corrupted the pre-existing MqttTriggerEditor import
  ($lib/... path) in PipelineTriggerEditors.svelte
- reject prefetch_count = 0 server-side in validate_config (RabbitMQ treats 0
  as unlimited) and defensively skip basic_qos(0) in build_consumer (covers
  the capture path that bypasses CRUD validation)
- recognize `// on amqp` in the canonical parser (TriggerSpec::Amqp) and add
  amqp to the CLI non-autorun/event-trigger sets so a pipeline cascade never
  runs an AMQP-only node as a manual root without an event
- add amqp to the preprocessor intro lists and both pipeline AI instructions

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(triggers): reject zero AMQP prefetch in all paths and finish guidance lists

- extract a shared validate_amqp_options used by both CRUD validate_config
  and build_consumer, so capture configs (which bypass CRUD validation) also
  reject prefetch 0 instead of silently connecting with an unlimited buffer
  (+ unit tests for 0/1/65535/None)
- add AMQP to the main script-writing preprocessor-sources prompt and the CLI
  triggers-skill guidance list

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* docs(triggers): de-duplicate AMQP prefetch comment and fix GET response text

- keep the zero-prefetch rationale only on the shared validate_amqp_options
  doc; drop the redundant call-site comments
- correct the getAmqpTrigger OpenAPI 200 description ("deleted" -> "retrieved")

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* chore: update ee-repo-ref to aaa6cb89b05b76139252c64f057e53b94d12ac60

This commit updates the EE repository reference after PR #680 was merged in windmill-ee-private.

Previous ee-repo-ref: 5da5fd65aca9594b2611837a52e4677b544b0380

New ee-repo-ref: aaa6cb89b05b76139252c64f057e53b94d12ac60

Automated by sync-ee-ref workflow.

* chore(migrations): consolidate the four AMQP migrations into one

The table and the three enum ADD VALUE statements (trigger_kind, job_trigger_kind,
draft_kind) are one atomic feature. ALTER TYPE ... ADD VALUE runs inside the
migration transaction on PG >= 14 (Windmill's minimum) since the amqp_trigger
table doesn't reference those enum types, so they can share a single migration
instead of four. Verified applying cleanly in a single transaction on a fresh DB.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>
This commit is contained in:
Ruben Fiszel
2026-07-21 17:10:29 +02:00
committed by GitHub
parent c7ee192680
commit 68debab877
165 changed files with 4753 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-02e39bb9957d3ccfe3d573994cd281a22467010d2c4a0611c8ae02742336f433.json
Generated
+38
View File
@@ -0,0 +1,38 @@
{
"db_name": "PostgreSQL",
"query": "\n INSERT INTO amqp_trigger (\n amqp_resource_path,\n queue_name,\n exchange,\n options,\n workspace_id,\n path,\n script_path,\n is_flow,\n permissioned_as,\n mode,\n edited_by,\n error_handler_path,\n error_handler_args,\n retry\n )\n VALUES (\n $1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14\n )",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Jsonb",
"Jsonb",
"Varchar",
"Varchar",
"Varchar",
"Bool",
"Varchar",
{
"Custom": {
"name": "trigger_mode",
"kind": {
"Enum": [
"enabled",
"disabled",
"suspended"
]
}
}
},
"Varchar",
"Varchar",
"Jsonb",
"Jsonb"
]
},
"nullable": []
},
"hash": "02e39bb9957d3ccfe3d573994cd281a22467010d2c4a0611c8ae02742336f433"
}
backend/.sqlx/query-05843e117361d6b7cd9da652596d3be7a970d602d3809396d9253d7e5dc88152.json
Generated
+41
View File
@@ -0,0 +1,41 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT kind, path, script_path, is_flow FROM (\n SELECT 'schedule' AS kind, path, script_path, is_flow FROM schedule\n WHERE workspace_id = $1\n AND script_path IS NOT NULL\n UNION ALL\n SELECT 'email', path, script_path, is_flow FROM email_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'kafka', path, script_path, is_flow FROM kafka_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'mqtt', path, script_path, is_flow FROM mqtt_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'amqp', path, script_path, is_flow FROM amqp_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'nats', path, script_path, is_flow FROM nats_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'postgres', path, script_path, is_flow FROM postgres_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'sqs', path, script_path, is_flow FROM sqs_trigger\n WHERE workspace_id = $1\n UNION ALL\n SELECT 'gcp', path, script_path, is_flow FROM gcp_trigger\n WHERE workspace_id = $1\n ) t\n WHERE ($2::text IS NULL OR script_path LIKE $2)\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "kind",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "path",
"type_info": "Varchar"
},
{
"ordinal": 2,
"name": "script_path",
"type_info": "Varchar"
},
{
"ordinal": 3,
"name": "is_flow",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text",
"Text"
]
},
"nullable": [
null,
null,
null,
null
]
},
"hash": "05843e117361d6b7cd9da652596d3be7a970d602d3809396d9253d7e5dc88152"
}
backend/.sqlx/query-089d7bc7acdbb97cf477159e111bc7e9ee85289ff5c52af43166928337c257e7.json
Generated
+2 -1
View File
@@ -36,7 +36,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-0f7e01b613a94b29784aae6d7b17b23d6dcf2e5364852a5e85b3c41c417bace2.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-16d438374b03a9c515f4c2d638366f38ffe2f3a0958adea53e67757c6ac463ec.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-1cd16f93bdc90b24ceb1e86495069c3592218d34e58d617253e8b563bc661a7a.json
Generated
+17
View File
@@ -0,0 +1,17 @@
{
"db_name": "PostgreSQL",
"query": "WITH t1 AS (UPDATE websocket_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t2 AS (UPDATE kafka_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t3 AS (UPDATE postgres_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t4 AS (UPDATE mqtt_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t5 AS (UPDATE nats_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t6 AS (UPDATE sqs_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), t7 AS (UPDATE amqp_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4) UPDATE gcp_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text",
"Text",
"Bool"
]
},
"nullable": []
},
"hash": "1cd16f93bdc90b24ceb1e86495069c3592218d34e58d617253e8b563bc661a7a"
}
backend/.sqlx/query-212553c83e4dcdc6d045eb2fe2dadbb2860ce52d37a56b2861de1215260ecff8.json
Generated
+4 -2
View File
@@ -40,7 +40,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
@@ -79,7 +80,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-22e0e8a1aa48f8b21763452bd36fbe7db4887c4ac5295052c796bd78a7edc50b.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-23419adcd74c326d716527293eff518b42f4cdb33e034441015494bd26c172d2.json
Generated
+2 -1
View File
@@ -43,7 +43,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-2bed492ef32edf36e60e8a03268fa25bfb67dd641153d1ca23f7d0d2ae73624e.json
Generated
+2 -1
View File
@@ -43,7 +43,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-3c84781704b84b8a927ecce5a3fcb3adcf0175d0a71013f2497397c1c8ccc619.json
Generated
+2 -1
View File
@@ -129,7 +129,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-42b4b73e9d60348e2d90fcade9dcad6d8995242dc20a4e14c1a8fae4fc6a9fd2.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-45024b932383199974616bba1fc2f7175cc6f2e02d9c565bb5159cae3e0b6835.json
Generated
+2 -1
View File
@@ -33,7 +33,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-46f00a75b2e7e4ac70758a9687070f68bc0421f1aa228f80157adda63191d33b.json
Generated
+2 -1
View File
@@ -42,7 +42,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-4a43d4df6c5b2e8dda4308dcb88c23caf312ec377dd91e5307f00d3fb8ec325d.json
Generated
+2 -1
View File
@@ -81,7 +81,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-4d272cf4a77aab7007a5b35589e08532a1020cabaf5e22325a1e05f0491d785c.json
Generated
+2 -1
View File
@@ -40,7 +40,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-4efaa4c3c14987fe95a508cc0cbac91ee793ad6f5ceed5b3eda9b48617c5434d.json
Generated
+2 -1
View File
@@ -47,7 +47,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-4f547c0fd54f3bc57212ce87810e35adf640d44d607e62a1fb296e38ac3fdd36.json
Generated
+4 -2
View File
@@ -35,7 +35,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
@@ -76,7 +77,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-54ad5cc89563fdbbacdd6bfde9be6ecfcdec3d505d28cc65b5920dcf87c0014f.json
Generated
+2 -1
View File
@@ -36,7 +36,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-560539adbad0ecfa57fa477c3b82d82c350857166fc27fe9eecc88bcc4b229bc.json
Generated
+2 -1
View File
@@ -43,7 +43,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-5884ce1906015f6b96231f311226e490d5dfcdd7a94dcbe5d05c9e5af37ac4a4.json
Generated
+2 -1
View File
@@ -36,7 +36,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-5f2e66e53166840c68efbc521fe9393023e79f948976e6ffc0e9bfc94623393c.json
Generated
+27
View File
@@ -0,0 +1,27 @@
{
"db_name": "PostgreSQL",
"query": "\n UPDATE\n amqp_trigger\n SET\n amqp_resource_path = $1,\n queue_name = $2,\n exchange = $3,\n options = $4,\n is_flow = $5,\n edited_by = $6,\n permissioned_as = $7,\n script_path = $8,\n path = $9,\n edited_at = now(),\n error = NULL,\n server_id = NULL,\n error_handler_path = $12,\n error_handler_args = $13,\n retry = $14\n WHERE\n workspace_id = $10 AND\n path = $11\n ",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Varchar",
"Jsonb",
"Jsonb",
"Bool",
"Varchar",
"Varchar",
"Varchar",
"Varchar",
"Text",
"Text",
"Varchar",
"Jsonb",
"Jsonb"
]
},
"nullable": []
},
"hash": "5f2e66e53166840c68efbc521fe9393023e79f948976e6ffc0e9bfc94623393c"
}
backend/.sqlx/query-66a0e51cf149ba532463e29dd361a803e1bced2f8e1a12f8933b7598ee85a147.json
Generated
+2 -1
View File
@@ -38,7 +38,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-67e25a7c19ea0ffaf7ea5303fcd04af5a7eb488c76f783e690af0c2153b1d6a8.json
Generated
+2 -1
View File
@@ -162,7 +162,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-6b9348e60cc1ce158314a93fc7aa55a9f8fa854b29edcea83710a9170124edf0.json
Generated
+2 -1
View File
@@ -47,7 +47,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-7065f23d04e26831664048f2cfc4f412c57af931f80621aee5012e9cb3535626.json
Generated
+2 -1
View File
@@ -32,7 +32,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-73fdd01bad58b8be1a52f89faef8d92a983470adcd3cc850734960c905e61e83.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-7fbf72d9059fcd77e4c1112fa4fa22e4276c1da653475628889ce17dc904fbaa.json
Generated
+2 -1
View File
@@ -30,7 +30,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-836bac47d89113d90bd03a471446eb9016207975af1e37042d81df8cb6ae2c53.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-87564a196a1662f524407d853db506bf08c28efe82b68b3d44bafbd3d0e91c29.json
Generated
+2 -1
View File
@@ -38,7 +38,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-8f163ee5adf4caaaa12a5698e68c749524f1db10a51f3de6cadd4826e6c1d422.json
Generated
+2 -1
View File
@@ -36,7 +36,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-90da952a912de746abee641cc4268644853987e62cbcedc8d3a46626a60c9fb0.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "UPDATE amqp_trigger SET workspace_id = $1 WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "90da952a912de746abee641cc4268644853987e62cbcedc8d3a46626a60c9fb0"
}
backend/.sqlx/query-940b6d78bab940a37a42492f030d2393e297043e4e58555d872b5c4dd89c196a.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-9c50e3a136a8ee3ec56e083f26d3a960b89e02ec40b292f3b5198baf2a1d3dbf.json
Generated
+2 -1
View File
@@ -35,7 +35,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-9ecb404e46a4eac55f977f05a3afbafe5dc3cdecc17a3d5a7476b160c1b6e7e1.json
Generated
+2 -1
View File
@@ -36,7 +36,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-a47f7d625a96ad47277609ae173c408eda4323ec12ade20ec0da565051128f26.json
Generated
+100
View File
@@ -0,0 +1,100 @@
{
"db_name": "PostgreSQL",
"query": "\n SELECT\n EXISTS(SELECT 1 FROM websocket_trigger WHERE workspace_id = $1) AS \"websocket_used!\",\n EXISTS(SELECT 1 FROM http_trigger WHERE workspace_id = $1) AS \"http_routes_used!\",\n EXISTS(SELECT 1 FROM kafka_trigger WHERE workspace_id = $1) as \"kafka_used!\",\n EXISTS(SELECT 1 FROM nats_trigger WHERE workspace_id = $1) as \"nats_used!\",\n EXISTS(SELECT 1 FROM postgres_trigger WHERE workspace_id = $1) AS \"postgres_used!\",\n EXISTS(SELECT 1 FROM mqtt_trigger WHERE workspace_id = $1) AS \"mqtt_used!\",\n EXISTS(SELECT 1 FROM amqp_trigger WHERE workspace_id = $1) AS \"amqp_used!\",\n EXISTS(SELECT 1 FROM sqs_trigger WHERE workspace_id = $1) AS \"sqs_used!\",\n EXISTS(SELECT 1 FROM gcp_trigger WHERE workspace_id = $1) AS \"gcp_used!\",\n EXISTS(SELECT 1 FROM azure_trigger WHERE workspace_id = $1) AS \"azure_used!\",\n EXISTS(SELECT 1 FROM email_trigger WHERE workspace_id = $1) AS \"email_used!\",\n EXISTS(SELECT 1 FROM native_trigger WHERE workspace_id = $1 AND service_name = 'nextcloud'::native_trigger_service) AS \"nextcloud_used!\",\n EXISTS(SELECT 1 FROM native_trigger WHERE workspace_id = $1 AND service_name = 'google'::native_trigger_service) AS \"google_used!\",\n EXISTS(SELECT 1 FROM native_trigger WHERE workspace_id = $1 AND service_name = 'github'::native_trigger_service) AS \"github_used!\"\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "websocket_used!",
"type_info": "Bool"
},
{
"ordinal": 1,
"name": "http_routes_used!",
"type_info": "Bool"
},
{
"ordinal": 2,
"name": "kafka_used!",
"type_info": "Bool"
},
{
"ordinal": 3,
"name": "nats_used!",
"type_info": "Bool"
},
{
"ordinal": 4,
"name": "postgres_used!",
"type_info": "Bool"
},
{
"ordinal": 5,
"name": "mqtt_used!",
"type_info": "Bool"
},
{
"ordinal": 6,
"name": "amqp_used!",
"type_info": "Bool"
},
{
"ordinal": 7,
"name": "sqs_used!",
"type_info": "Bool"
},
{
"ordinal": 8,
"name": "gcp_used!",
"type_info": "Bool"
},
{
"ordinal": 9,
"name": "azure_used!",
"type_info": "Bool"
},
{
"ordinal": 10,
"name": "email_used!",
"type_info": "Bool"
},
{
"ordinal": 11,
"name": "nextcloud_used!",
"type_info": "Bool"
},
{
"ordinal": 12,
"name": "google_used!",
"type_info": "Bool"
},
{
"ordinal": 13,
"name": "github_used!",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Text"
]
},
"nullable": [
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null,
null
]
},
"hash": "a47f7d625a96ad47277609ae173c408eda4323ec12ade20ec0da565051128f26"
}
backend/.sqlx/query-a4b6371d33206010b2f3ffd2b09e33244fe8ab9a803248fc23f334034d24aad4.json
Generated
+2 -1
View File
@@ -192,7 +192,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-b3771b690c5966272b1f42c9965bb6a8f961c119516e4c33dc928cd3b4f4edbc.json
Generated
+2 -1
View File
@@ -167,7 +167,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-b3f0595cacba194e08b9a3e244d9e637e9e156cd85b69126c87dfff89a47711d.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-b6cfb752675a3f36975e6cc6c454267d67f58f9cbff0d164b4df45de885b0e9d.json
Generated
+2 -1
View File
@@ -47,7 +47,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-bcfa34cf80abea05f0c24883b9e77429c51e6166c414bcc5ce2e97fac25bcd77.json
Generated
+2 -1
View File
@@ -81,7 +81,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-be6d2c92a62b7b284651c45af809746147aa9b8d0a81642a7b7cb4738a0cad66.json
Generated
+2 -1
View File
@@ -112,7 +112,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-c3b1152b554812d65eb27f95b1fd434f860922fbc021185beffb9827647feb8e.json
Generated
+2 -1
View File
@@ -34,7 +34,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-c8fb2a1491f90951a6d2e4e9c56d288eb60f6c6644a73cdf949de0159215a7f7.json
Generated
+2 -1
View File
@@ -43,7 +43,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-ccef7a1bde5cac6c362c5fedb6c13f1f882b695f896f94e5cf91d205633355a1.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-d41ea93fd58381b89e151c965eae1ea2fe96a1b94f5a92953fb1c1642d15c016.json
Generated
+2 -1
View File
@@ -112,7 +112,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-d4211392e174a0e8f89c7fcebdf120e5b0f629f9f04e08a2982df33ff23ac7a9.json
Generated
+2 -1
View File
@@ -252,7 +252,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-d495c94b580fd34d5ae90615ef21a8a9cc35f362197c0766a5787436af141106.json
Generated
+2 -1
View File
@@ -28,7 +28,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-d4e0da9f9653d532770066310f85e59e5edda1faea72f39603814afb6a3cd596.json
Generated
+2 -1
View File
@@ -48,7 +48,8 @@
"trigger_nextcloud",
"trigger_google",
"trigger_github",
"data_pipeline"
"data_pipeline",
"trigger_amqp"
]
}
}
backend/.sqlx/query-d9d997591a163f25a9be5134bffacd0068c51ca21018d6751cc1044e38c06730.json
Generated
+15
View File
@@ -0,0 +1,15 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO amqp_trigger (\n amqp_resource_path, queue_name, exchange, options, path, script_path, is_flow,\n workspace_id, edited_by, edited_at, extra_perms, server_id, last_server_ping,\n error, error_handler_path, error_handler_args, retry, mode, permissioned_as, labels\n )\n SELECT\n amqp_resource_path, queue_name, exchange, options, path, script_path, is_flow,\n $1, edited_by, edited_at, extra_perms, NULL, NULL,\n NULL, error_handler_path, error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels\n FROM amqp_trigger WHERE workspace_id = $2",
"describe": {
"columns": [],
"parameters": {
"Left": [
"Varchar",
"Text"
]
},
"nullable": []
},
"hash": "d9d997591a163f25a9be5134bffacd0068c51ca21018d6751cc1044e38c06730"
}
backend/.sqlx/query-e4d71278fb80126a7a9da73f1889352d4d1e3cb3a8a08f1c9c03055a1cab1235.json
Generated
+2 -1
View File
@@ -192,7 +192,8 @@
"github",
"azure",
"asset",
"freshness"
"freshness",
"amqp"
]
}
}
backend/.sqlx/query-e80177f3ffd4c1f52cdb4757483f03f72ef81db302d727e18e63a307ac902022.json
Generated
+2 -1
View File
@@ -34,7 +34,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-eac595e19e5c8e70f1514ef29dec35c7342ac9a814c73f6290e1d6ebd3a55423.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/.sqlx/query-ed8facbf29ebb670d05fe8aa34b50d6a6935420fbedc83aa3ad1e9be7465c8dd.json
Generated
+2 -1
View File
@@ -27,7 +27,8 @@
"nextcloud",
"google",
"github",
"azure"
"azure",
"amqp"
]
}
}
backend/Cargo.lock
Generated
+688 -30
View File
File diff suppressed because it is too large Load Diff
backend/Cargo.toml
+8 -2
View File
@@ -22,6 +22,7 @@ members = [
"./windmill-trigger-kafka",
"./windmill-trigger-postgres",
"./windmill-trigger-mqtt",
"./windmill-trigger-amqp",
"./windmill-trigger-websocket",
"./windmill-trigger-email",
"./windmill-trigger-nats",
@@ -141,6 +142,7 @@ postgres_trigger = ["windmill-api/postgres_trigger"]
mcp = ["windmill-ai/mcp", "windmill-api/mcp", "windmill-worker/mcp"]
bedrock = ["windmill-ai/bedrock", "windmill-api/bedrock", "windmill-worker/bedrock"]
mqtt_trigger = ["windmill-api/mqtt_trigger"]
amqp_trigger = ["windmill-api/amqp_trigger"]
native_trigger = ["windmill-api/native_trigger"]
sqs_trigger = ["windmill-api/sqs_trigger", "windmill-common/aws_auth", "windmill-api/openidconnect"]
gcp_trigger = ["windmill-api/gcp_trigger"]
@@ -180,7 +182,7 @@ run_inline = ["windmill-api/run_inline"]
oss_core = [
"embedding", "parquet", "openidconnect", "license",
"http_trigger", "zip", "oauth2", "postgres_trigger",
"mqtt_trigger", "websocket", "smtp", "native_trigger",
"mqtt_trigger", "amqp_trigger", "websocket", "smtp", "native_trigger",
"static_frontend", "mcp", "bedrock", "run_inline",
"quickjs"
]
@@ -201,7 +203,7 @@ ee_rhel = ["ce_core", "ee_core", "kafka-gssapi", "all_languages"]
ee_windows = ["ce_core", "ee_core", "all_languages_windows"]
all_sqlx_features = ["all_languages", "enterprise", "enterprise_saml", "embedding", "parquet", "prometheus", "flow_testing",
"openidconnect", "cloud", "jemalloc", "tantivy", "sqlx", "kafka", "kafka-gssapi", "nats", "otel", "dind", "websocket", "http_trigger",
"postgres_trigger", "mcp", "mqtt_trigger", "sqs_trigger", "gcp_trigger", "azure_trigger", "smtp", "stripe",
"postgres_trigger", "mcp", "mqtt_trigger", "amqp_trigger", "sqs_trigger", "gcp_trigger", "azure_trigger", "smtp", "stripe",
"license", "oauth2", "zip", "static_frontend", "scoped_cache", "agent_worker_server", "bedrock", "native_trigger", "quickjs",
"windmill-git-sync/all_sqlx_features"]
@@ -361,6 +363,7 @@ windmill-trigger = { path = "./windmill-trigger" }
windmill-trigger-kafka = { path = "./windmill-trigger-kafka" }
windmill-trigger-postgres = { path = "./windmill-trigger-postgres" }
windmill-trigger-mqtt = { path = "./windmill-trigger-mqtt" }
windmill-trigger-amqp = { path = "./windmill-trigger-amqp" }
windmill-trigger-websocket = { path = "./windmill-trigger-websocket" }
windmill-trigger-email = { path = "./windmill-trigger-email" }
windmill-trigger-nats = { path = "./windmill-trigger-nats" }
@@ -678,6 +681,9 @@ tree-sitter-ruby = "=0.23.1"
tree-sitter-r = "=1.2.0"
oracle = { version = "0.6.3", features = ["chrono"] }
rumqttc = { version = "0.24.0", features = ["use-native-tls"]}
lapin = "2.5"
tokio-executor-trait = "2.1"
tokio-reactor-trait = "1.1"
strum = { version = "0.27", features = ["derive"] }
strum_macros = "0.27"
hudsucker = { version = "0.22", features = ["rcgen-ca", "native-tls-client"] }
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
eb3690a34b41134c91b9cb6f90b11daa541ad171
aaa6cb89b05b76139252c64f057e53b94d12ac60
backend/migrations/20260721082249_add_amqp_trigger.down.sql
+2
View File
@@ -0,0 +1,2 @@
-- Add down migration script here
DROP TABLE amqp_trigger;
backend/migrations/20260721082249_add_amqp_trigger.up.sql
+83
View File
@@ -0,0 +1,83 @@
-- Add up migration script here
CREATE TABLE amqp_trigger (
amqp_resource_path VARCHAR(255) NOT NULL,
queue_name VARCHAR(255) NOT NULL,
exchange JSONB NULL,
options JSONB NULL,
path VARCHAR(255) NOT NULL,
script_path VARCHAR(255) NOT NULL,
is_flow BOOLEAN NOT NULL,
workspace_id VARCHAR(50) NOT NULL,
edited_by VARCHAR(50) NOT NULL,
edited_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
extra_perms JSONB NOT NULL DEFAULT '{}',
server_id VARCHAR(50) NULL,
last_server_ping TIMESTAMPTZ NULL,
error TEXT NULL,
error_handler_path VARCHAR(255) NULL,
error_handler_args JSONB NULL,
retry JSONB NULL,
mode TRIGGER_MODE NOT NULL DEFAULT 'enabled'::TRIGGER_MODE,
permissioned_as VARCHAR(255) NOT NULL,
labels TEXT[] NULL,
PRIMARY KEY (path, workspace_id),
FOREIGN KEY (workspace_id) REFERENCES workspace(id) ON DELETE CASCADE
);
CREATE INDEX idx_amqp_trigger_labels ON amqp_trigger USING gin (labels) WHERE labels IS NOT NULL;
GRANT ALL ON amqp_trigger TO windmill_user;
GRANT ALL ON amqp_trigger TO windmill_admin;
ALTER TABLE amqp_trigger ENABLE ROW LEVEL SECURITY;
CREATE POLICY admin_policy ON amqp_trigger FOR ALL TO windmill_admin USING (true);
CREATE POLICY see_folder_extra_perms_user_select ON amqp_trigger FOR SELECT TO windmill_user
USING (SPLIT_PART(amqp_trigger.path, '/', 1) = 'f' AND SPLIT_PART(amqp_trigger.path, '/', 2) = any((select regexp_split_to_array(current_setting('session.folders_read'), ','))::text[]));
CREATE POLICY see_folder_extra_perms_user_insert ON amqp_trigger FOR INSERT TO windmill_user
WITH CHECK (SPLIT_PART(amqp_trigger.path, '/', 1) = 'f' AND SPLIT_PART(amqp_trigger.path, '/', 2) = any((select regexp_split_to_array(current_setting('session.folders_write'), ','))::text[]));
CREATE POLICY see_folder_extra_perms_user_update ON amqp_trigger FOR UPDATE TO windmill_user
USING (SPLIT_PART(amqp_trigger.path, '/', 1) = 'f' AND SPLIT_PART(amqp_trigger.path, '/', 2) = any((select regexp_split_to_array(current_setting('session.folders_write'), ','))::text[]));
CREATE POLICY see_folder_extra_perms_user_delete ON amqp_trigger FOR DELETE TO windmill_user
USING (SPLIT_PART(amqp_trigger.path, '/', 1) = 'f' AND SPLIT_PART(amqp_trigger.path, '/', 2) = any((select regexp_split_to_array(current_setting('session.folders_write'), ','))::text[]));
CREATE POLICY see_own ON amqp_trigger FOR ALL TO windmill_user
USING (SPLIT_PART(amqp_trigger.path, '/', 1) = 'u' AND SPLIT_PART(amqp_trigger.path, '/', 2) = (select current_setting('session.user')));
CREATE POLICY see_member ON amqp_trigger FOR ALL TO windmill_user
USING (SPLIT_PART(amqp_trigger.path, '/', 1) = 'g' AND SPLIT_PART(amqp_trigger.path, '/', 2) = any((select regexp_split_to_array(current_setting('session.groups'), ','))::text[]));
CREATE POLICY see_extra_perms_user_select ON amqp_trigger FOR SELECT TO windmill_user
USING (extra_perms ? (select concat('u/', current_setting('session.user'))));
CREATE POLICY see_extra_perms_user_insert ON amqp_trigger FOR INSERT TO windmill_user
WITH CHECK ((extra_perms ->> (select concat('u/', current_setting('session.user'))))::boolean);
CREATE POLICY see_extra_perms_user_update ON amqp_trigger FOR UPDATE TO windmill_user
USING ((extra_perms ->> (select concat('u/', current_setting('session.user'))))::boolean);
CREATE POLICY see_extra_perms_user_delete ON amqp_trigger FOR DELETE TO windmill_user
USING ((extra_perms ->> (select concat('u/', current_setting('session.user'))))::boolean);
CREATE POLICY see_extra_perms_groups_select ON amqp_trigger FOR SELECT TO windmill_user
USING (extra_perms ?| (select regexp_split_to_array(current_setting('session.pgroups'), ','))::text[]);
CREATE POLICY see_extra_perms_groups_insert ON amqp_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((select regexp_split_to_array(current_setting('session.pgroups'), ','))::text[])
AND value::boolean));
CREATE POLICY see_extra_perms_groups_update ON amqp_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((select regexp_split_to_array(current_setting('session.pgroups'), ','))::text[])
AND value::boolean));
CREATE POLICY see_extra_perms_groups_delete ON amqp_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((select regexp_split_to_array(current_setting('session.pgroups'), ','))::text[])
AND value::boolean));
-- Enum values for the new trigger kind. ALTER TYPE ... ADD VALUE runs inside the
-- migration transaction on PG >= 14 (Windmill's minimum) as long as the value
-- isn't used in the same transaction — the amqp_trigger table above does not
-- reference these enum types.
ALTER TYPE TRIGGER_KIND ADD VALUE IF NOT EXISTS 'amqp';
ALTER TYPE job_trigger_kind ADD VALUE IF NOT EXISTS 'amqp';
ALTER TYPE draft_kind ADD VALUE IF NOT EXISTS 'trigger_amqp';
backend/parsers/windmill-parser/src/asset_parser.rs
+6 -1
View File
@@ -172,6 +172,7 @@ pub enum TriggerSpec {
Email,
Kafka,
Mqtt,
Amqp,
Nats,
Postgres,
Sqs,
@@ -1527,6 +1528,7 @@ fn parse_trigger_spec(s: &str) -> Option<TriggerSpec> {
("email", TriggerSpec::Email),
("kafka", TriggerSpec::Kafka),
("mqtt", TriggerSpec::Mqtt),
("amqp", TriggerSpec::Amqp),
("nats", TriggerSpec::Nats),
("postgres", TriggerSpec::Postgres),
("sqs", TriggerSpec::Sqs),
@@ -1722,7 +1724,10 @@ mod pipeline_annotation_tests {
);
// A non-ASCII body must not panic on a byte offset that is not a char
// boundary.
assert_eq!(split_measure_filter("sum(amount) + π"), ("sum(amount) + π", None));
assert_eq!(
split_measure_filter("sum(amount) + π"),
("sum(amount) + π", None)
);
assert_eq!(
split_measure_filter("sum(π) where region = 'π'"),
("sum(π)", Some("region = 'π'".to_string()))
backend/parsers/windmill-parser/tests/pipeline_annotations_parity.rs
+1
View File
@@ -121,6 +121,7 @@ fn native_str(t: &TriggerSpec) -> Option<&'static str> {
TriggerSpec::Email => "email",
TriggerSpec::Kafka => "kafka",
TriggerSpec::Mqtt => "mqtt",
TriggerSpec::Amqp => "amqp",
TriggerSpec::Nats => "nats",
TriggerSpec::Postgres => "postgres",
TriggerSpec::Sqs => "sqs",
backend/windmill-api-assets/src/lib.rs
+9
View File
@@ -784,6 +784,11 @@ enum TriggerEdge {
runnable_kind: AssetUsageKind,
runnable_path: String,
},
Amqp {
path: String,
runnable_kind: AssetUsageKind,
runnable_path: String,
},
Nats {
path: String,
runnable_kind: AssetUsageKind,
@@ -928,6 +933,9 @@ async fn asset_graph(
SELECT 'mqtt', path, script_path, is_flow FROM mqtt_trigger
WHERE workspace_id = $1
UNION ALL
SELECT 'amqp', path, script_path, is_flow FROM amqp_trigger
WHERE workspace_id = $1
UNION ALL
SELECT 'nats', path, script_path, is_flow FROM nats_trigger
WHERE workspace_id = $1
UNION ALL
@@ -1210,6 +1218,7 @@ async fn asset_graph(
"email" => TriggerEdge::Email { path, runnable_kind, runnable_path: script_path },
"kafka" => TriggerEdge::Kafka { path, runnable_kind, runnable_path: script_path },
"mqtt" => TriggerEdge::Mqtt { path, runnable_kind, runnable_path: script_path },
"amqp" => TriggerEdge::Amqp { path, runnable_kind, runnable_path: script_path },
"nats" => TriggerEdge::Nats { path, runnable_kind, runnable_path: script_path },
"postgres" => TriggerEdge::Postgres { path, runnable_kind, runnable_path: script_path },
"sqs" => TriggerEdge::Sqs { path, runnable_kind, runnable_path: script_path },
backend/windmill-api-auth/src/scopes.rs
+3
View File
@@ -261,6 +261,7 @@ pub enum ScopeDomain {
KafkaTriggers,
NatsTriggers,
MqttTriggers,
AmqpTriggers,
SqsTriggers,
GcpTriggers,
AzureTriggers,
@@ -323,6 +324,7 @@ impl ScopeDomain {
Self::KafkaTriggers => "kafka_triggers",
Self::NatsTriggers => "nats_triggers",
Self::MqttTriggers => "mqtt_triggers",
Self::AmqpTriggers => "amqp_triggers",
Self::SqsTriggers => "sqs_triggers",
Self::GcpTriggers => "gcp_triggers",
Self::AzureTriggers => "azure_triggers",
@@ -378,6 +380,7 @@ impl ScopeDomain {
"kafka_triggers" => Some(Self::KafkaTriggers),
"nats_triggers" => Some(Self::NatsTriggers),
"mqtt_triggers" => Some(Self::MqttTriggers),
"amqp_triggers" => Some(Self::AmqpTriggers),
"sqs_triggers" => Some(Self::SqsTriggers),
"gcp_triggers" => Some(Self::GcpTriggers),
"azure_triggers" => Some(Self::AzureTriggers),
backend/windmill-api-groups/src/granular_acls.rs
+4 -3
View File
@@ -46,13 +46,13 @@ fn audit_action_prefix_for_acl_kind(kind: &str) -> Option<&'static str> {
"variable" => Some("variables"),
"schedule" => Some("schedules"),
"http_trigger" | "websocket_trigger" | "kafka_trigger" | "nats_trigger"
| "postgres_trigger" | "mqtt_trigger" | "gcp_trigger" | "azure_trigger" | "sqs_trigger"
| "email_trigger" => Some("triggers"),
| "postgres_trigger" | "mqtt_trigger" | "amqp_trigger" | "gcp_trigger"
| "azure_trigger" | "sqs_trigger" | "email_trigger" => Some("triggers"),
_ => None,
}
}
const KINDS: [&str; 20] = [
const KINDS: [&str; 21] = [
"script",
"group_",
"resource",
@@ -68,6 +68,7 @@ const KINDS: [&str; 20] = [
"nats_trigger",
"postgres_trigger",
"mqtt_trigger",
"amqp_trigger",
"gcp_trigger",
"azure_trigger",
"sqs_trigger",
backend/windmill-api-users/src/users.rs
+1
View File
@@ -1751,6 +1751,7 @@ pub async fn delete_workspace_user_internal(
"kafka_trigger",
"postgres_trigger",
"mqtt_trigger",
"amqp_trigger",
"nats_trigger",
"sqs_trigger",
"gcp_trigger",
backend/windmill-api-workspaces/src/workspaces.rs
+19
View File
@@ -4444,6 +4444,7 @@ struct UsedTriggers {
pub nats_used: bool,
pub postgres_used: bool,
pub mqtt_used: bool,
pub amqp_used: bool,
pub sqs_used: bool,
pub gcp_used: bool,
pub azure_used: bool,
@@ -4469,6 +4470,7 @@ async fn get_used_triggers(
EXISTS(SELECT 1 FROM nats_trigger WHERE workspace_id = $1) as "nats_used!",
EXISTS(SELECT 1 FROM postgres_trigger WHERE workspace_id = $1) AS "postgres_used!",
EXISTS(SELECT 1 FROM mqtt_trigger WHERE workspace_id = $1) AS "mqtt_used!",
EXISTS(SELECT 1 FROM amqp_trigger WHERE workspace_id = $1) AS "amqp_used!",
EXISTS(SELECT 1 FROM sqs_trigger WHERE workspace_id = $1) AS "sqs_used!",
EXISTS(SELECT 1 FROM gcp_trigger WHERE workspace_id = $1) AS "gcp_used!",
EXISTS(SELECT 1 FROM azure_trigger WHERE workspace_id = $1) AS "azure_used!",
@@ -5146,6 +5148,23 @@ async fn clone_triggers_and_schedules(
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"INSERT INTO amqp_trigger (
amqp_resource_path, queue_name, exchange, options, path, script_path, is_flow,
workspace_id, edited_by, edited_at, extra_perms, server_id, last_server_ping,
error, error_handler_path, error_handler_args, retry, mode, permissioned_as, labels
)
SELECT
amqp_resource_path, queue_name, exchange, options, path, script_path, is_flow,
$1, edited_by, edited_at, extra_perms, NULL, NULL,
NULL, error_handler_path, error_handler_args, retry, 'disabled'::TRIGGER_MODE, permissioned_as, labels
FROM amqp_trigger WHERE workspace_id = $2"#,
target_workspace_id,
source_workspace_id,
)
.execute(&mut **tx)
.await?;
sqlx::query!(
r#"INSERT INTO sqs_trigger (
path, queue_url, aws_resource_path, message_attributes, script_path,
backend/windmill-api-workspaces/src/workspaces_extra.rs
+9
View File
@@ -279,6 +279,15 @@ pub(crate) async fn change_workspace_id(
.execute(&mut *tx)
.await?;
info!("Updating amqp_trigger table");
sqlx::query!(
"UPDATE amqp_trigger SET workspace_id = $1 WHERE workspace_id = $2",
&rw.new_id,
&old_id
)
.execute(&mut *tx)
.await?;
info!("Updating gcp_trigger table");
sqlx::query!(
"UPDATE gcp_trigger SET workspace_id = $1 WHERE workspace_id = $2",
backend/windmill-api/Cargo.toml
+4 -2
View File
@@ -10,8 +10,8 @@ path = "src/lib.rs"
[features]
default = []
private = ["windmill-audit/private", "windmill-common/private", "windmill-api-auth/private", "windmill-store/private", "windmill-api-users/private", "windmill-api-workspaces/private", "windmill-api-groups/private", "windmill-api-configs/private", "windmill-api-settings/private", "windmill-api-assets/private", "windmill-api-agent-workers?/private", "windmill-trigger-kafka?/private", "windmill-trigger-postgres?/private", "windmill-trigger-mqtt?/private", "windmill-trigger-websocket?/private", "windmill-trigger-nats?/private", "windmill-trigger-sqs?/private", "windmill-trigger-gcp?/private", "windmill-trigger-azure?/private", "windmill-trigger-email?/private", "windmill-git-sync/private", "windmill-autoscaling?/private", "windmill-object-store/private"]
enterprise = ["windmill-queue/enterprise", "windmill-audit/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker?/enterprise", "windmill-api-auth/enterprise", "windmill-store/enterprise", "windmill-api-jobs/enterprise", "windmill-api-scripts/enterprise", "windmill-api-flows/enterprise", "windmill-api-users/enterprise", "windmill-api-workspaces/enterprise", "windmill-api-groups/enterprise", "windmill-api-configs/enterprise", "windmill-api-settings/enterprise", "windmill-api-schedule/enterprise", "windmill-api-agent-workers?/enterprise", "windmill-trigger/enterprise", "windmill-trigger-kafka?/enterprise", "windmill-trigger-postgres?/enterprise", "windmill-trigger-mqtt?/enterprise", "windmill-trigger-websocket?/enterprise", "windmill-trigger-email?/enterprise", "windmill-trigger-nats?/enterprise", "windmill-trigger-sqs?/enterprise", "windmill-trigger-gcp?/enterprise", "windmill-trigger-azure?/enterprise", "windmill-trigger-http?/enterprise", "windmill-native-triggers?/enterprise", "dep:windmill-autoscaling", "windmill-autoscaling/enterprise", "license"]
private = ["windmill-audit/private", "windmill-common/private", "windmill-api-auth/private", "windmill-store/private", "windmill-api-users/private", "windmill-api-workspaces/private", "windmill-api-groups/private", "windmill-api-configs/private", "windmill-api-settings/private", "windmill-api-assets/private", "windmill-api-agent-workers?/private", "windmill-trigger-kafka?/private", "windmill-trigger-postgres?/private", "windmill-trigger-mqtt?/private", "windmill-trigger-amqp?/private", "windmill-trigger-websocket?/private", "windmill-trigger-nats?/private", "windmill-trigger-sqs?/private", "windmill-trigger-gcp?/private", "windmill-trigger-azure?/private", "windmill-trigger-email?/private", "windmill-git-sync/private", "windmill-autoscaling?/private", "windmill-object-store/private"]
enterprise = ["windmill-queue/enterprise", "windmill-audit/enterprise", "windmill-git-sync/enterprise", "windmill-common/enterprise", "windmill-worker?/enterprise", "windmill-api-auth/enterprise", "windmill-store/enterprise", "windmill-api-jobs/enterprise", "windmill-api-scripts/enterprise", "windmill-api-flows/enterprise", "windmill-api-users/enterprise", "windmill-api-workspaces/enterprise", "windmill-api-groups/enterprise", "windmill-api-configs/enterprise", "windmill-api-settings/enterprise", "windmill-api-schedule/enterprise", "windmill-api-agent-workers?/enterprise", "windmill-trigger/enterprise", "windmill-trigger-kafka?/enterprise", "windmill-trigger-postgres?/enterprise", "windmill-trigger-mqtt?/enterprise", "windmill-trigger-amqp?/enterprise", "windmill-trigger-websocket?/enterprise", "windmill-trigger-email?/enterprise", "windmill-trigger-nats?/enterprise", "windmill-trigger-sqs?/enterprise", "windmill-trigger-gcp?/enterprise", "windmill-trigger-azure?/enterprise", "windmill-trigger-http?/enterprise", "windmill-native-triggers?/enterprise", "dep:windmill-autoscaling", "windmill-autoscaling/enterprise", "license"]
stripe = []
run_inline = ["dep:windmill-worker", "windmill-api-configs/run_inline"]
agent_worker_server = ["dep:windmill-worker", "dep:windmill-api-agent-workers"]
@@ -34,6 +34,7 @@ http_trigger = ["dep:matchit", "dep:windmill-trigger-http", "windmill-store/http
static_frontend = ["dep:rust-embed"]
postgres_trigger = ["dep:windmill-trigger-postgres", "windmill-store/postgres_trigger"]
mqtt_trigger = ["dep:windmill-trigger-mqtt", "windmill-store/mqtt_trigger"]
amqp_trigger = ["dep:windmill-trigger-amqp", "windmill-store/amqp_trigger"]
native_trigger = ["dep:windmill-native-triggers", "windmill-native-triggers/native_trigger", "dep:strum", "oauth2"]
sqs_trigger = ["dep:windmill-trigger-sqs", "windmill-store/sqs_trigger"]
gcp_trigger = ["dep:windmill-trigger-gcp", "windmill-store/gcp_trigger"]
@@ -141,6 +142,7 @@ matchit = { workspace = true, optional = true }
windmill-trigger-kafka = { workspace = true, optional = true }
windmill-trigger-postgres = { workspace = true, optional = true }
windmill-trigger-mqtt = { workspace = true, optional = true }
windmill-trigger-amqp = { workspace = true, optional = true }
windmill-trigger-websocket = { workspace = true, optional = true }
windmill-trigger-email = { workspace = true, optional = true }
windmill-trigger-nats = { workspace = true, optional = true }
backend/windmill-api/openapi.yaml
+394
View File
@@ -5629,6 +5629,8 @@ paths:
type: boolean
mqtt_used:
type: boolean
amqp_used:
type: boolean
gcp_used:
type: boolean
azure_used:
@@ -5650,6 +5652,7 @@ paths:
- nats_used
- postgres_used
- mqtt_used
- amqp_used
- gcp_used
- azure_used
- sqs_used
@@ -17457,6 +17460,211 @@ paths:
schema:
type: string
/w/{workspace}/amqp_triggers/create:
post:
summary: create amqp trigger
operationId: createAmqpTrigger
tags:
- amqp_trigger
parameters:
- $ref: "#/components/parameters/WorkspaceId"
requestBody:
description: new amqp trigger
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/NewAmqpTrigger"
responses:
"201":
description: amqp trigger created
content:
text/plain:
schema:
type: string
/w/{workspace}/amqp_triggers/update/{path}:
post:
summary: update amqp trigger
operationId: updateAmqpTrigger
tags:
- amqp_trigger
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
requestBody:
description: updated trigger
required: true
content:
application/json:
schema:
$ref: "#/components/schemas/EditAmqpTrigger"
responses:
"200":
description: amqp trigger updated
content:
text/plain:
schema:
type: string
/w/{workspace}/amqp_triggers/delete/{path}:
delete:
summary: delete amqp trigger
operationId: deleteAmqpTrigger
tags:
- amqp_trigger
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
responses:
"200":
description: amqp trigger deleted
content:
text/plain:
schema:
type: string
/w/{workspace}/amqp_triggers/get/{path}:
get:
summary: get amqp trigger
operationId: getAmqpTrigger
tags:
- amqp_trigger
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
- $ref: "#/components/parameters/GetDraft"
responses:
"200":
description: amqp trigger retrieved
content:
application/json:
schema:
allOf:
- $ref: "#/components/schemas/AmqpTrigger"
- $ref: "#/components/schemas/UserDraftOverlay"
/w/{workspace}/amqp_triggers/list:
get:
summary: list amqp triggers
operationId: listAmqpTriggers
tags:
- amqp_trigger
parameters:
- $ref: "#/components/parameters/WorkspaceId"
required: true
- $ref: "#/components/parameters/Page"
- $ref: "#/components/parameters/PerPage"
- name: path
description: filter by path
in: query
schema:
type: string
- name: is_flow
in: query
schema:
type: boolean
- name: path_start
in: query
schema:
type: string
- name: label
in: query
required: false
schema:
type: string
description: Filter by label
- $ref: "#/components/parameters/IncludeDraftOnly"
responses:
"200":
description: amqp trigger list
content:
application/json:
schema:
type: array
items:
$ref: "#/components/schemas/AmqpTrigger"
/w/{workspace}/amqp_triggers/exists/{path}:
get:
summary: does amqp trigger exists
operationId: existsAmqpTrigger
tags:
- amqp_trigger
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
responses:
"200":
description: amqp trigger exists
content:
application/json:
schema:
type: boolean
/w/{workspace}/amqp_triggers/setmode/{path}:
post:
summary: set enabled amqp trigger
operationId: setAmqpTriggerMode
tags:
- amqp_trigger
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
requestBody:
description: updated amqp trigger enable
required: true
content:
application/json:
schema:
type: object
properties:
mode:
$ref: "#/components/schemas/TriggerMode"
force:
type: boolean
description: >
Bypass the parent-state conflict warning when enabling a
trigger in a fork whose parent has the same path enabled.
required:
- mode
responses:
"200":
description: amqp trigger enabled set
content:
text/plain:
schema:
type: string
/w/{workspace}/amqp_triggers/test:
post:
summary: test amqp connection
operationId: testAmqpConnection
tags:
- amqp_trigger
parameters:
- $ref: "#/components/parameters/WorkspaceId"
requestBody:
description: test amqp connection
required: true
content:
application/json:
schema:
type: object
properties:
amqp_resource_path:
type: string
description: Path to the AMQP resource containing broker connection configuration
required:
- amqp_resource_path
responses:
"200":
description: successfully connected to amqp
content:
text/plain:
schema:
type: string
/w/{workspace}/gcp_triggers/create:
post:
summary: create gcp trigger
@@ -19966,6 +20174,7 @@ paths:
nats_trigger,
postgres_trigger,
mqtt_trigger,
amqp_trigger,
gcp_trigger,
azure_trigger,
sqs_trigger,
@@ -20013,6 +20222,7 @@ paths:
nats_trigger,
postgres_trigger,
mqtt_trigger,
amqp_trigger,
gcp_trigger,
azure_trigger,
sqs_trigger,
@@ -20071,6 +20281,7 @@ paths:
nats_trigger,
postgres_trigger,
mqtt_trigger,
amqp_trigger,
gcp_trigger,
azure_trigger,
sqs_trigger,
@@ -23229,6 +23440,7 @@ components:
- trigger_kafka
- trigger_nats
- trigger_mqtt
- trigger_amqp
- trigger_sqs
- trigger_gcp
- trigger_azure
@@ -26415,6 +26627,7 @@ components:
- kafka
- nats
- mqtt
- amqp
- sqs
- gcp
- azure
@@ -26919,6 +27132,8 @@ components:
type: number
mqtt_count:
type: number
amqp_count:
type: number
gcp_count:
type: number
azure_count:
@@ -27435,6 +27650,183 @@ components:
- subscribe_topics
- mqtt_resource_path
AmqpExchange:
type: object
properties:
exchange_name:
type: string
description: Name of the exchange to bind the consumed queue to
routing_keys:
type: array
items:
type: string
description: Routing keys used to bind the queue to the exchange
required:
- exchange_name
AmqpOptions:
type: object
properties:
declare_queue:
type: boolean
description: Declare the queue (durable) before consuming; when false the queue is declared passively and must already exist
prefetch_count:
type: integer
format: int32
minimum: 1
maximum: 65535
description: Maximum number of unacknowledged messages the broker delivers at once (1-65535)
AmqpTrigger:
allOf:
- $ref: "#/components/schemas/TriggerExtraProperty"
type: object
properties:
amqp_resource_path:
type: string
description: Path to the AMQP resource containing broker connection configuration
queue_name:
type: string
description: Name of the queue to consume messages from
exchange:
$ref: "#/components/schemas/AmqpExchange"
nullable: true
description: Optional exchange binding for the consumed queue
options:
$ref: "#/components/schemas/AmqpOptions"
nullable: true
description: Optional consumer options (queue declaration, prefetch)
server_id:
type: string
description: ID of the server currently handling this trigger (internal)
last_server_ping:
type: string
format: date-time
description: Timestamp of last server heartbeat (internal)
error:
type: string
description: Last error message if the trigger failed
error_handler_path:
type: string
description: Path to a script or flow to run when the triggered job fails
error_handler_args:
$ref: "#/components/schemas/ScriptArgs"
description: Arguments to pass to the error handler
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
required:
- amqp_resource_path
- queue_name
NewAmqpTrigger:
type: object
properties:
amqp_resource_path:
type: string
description: Path to the AMQP resource containing broker connection configuration
queue_name:
type: string
description: Name of the queue to consume messages from
exchange:
nullable: true
$ref: "#/components/schemas/AmqpExchange"
description: Optional exchange binding for the consumed queue
options:
nullable: true
$ref: "#/components/schemas/AmqpOptions"
description: Optional consumer options (queue declaration, prefetch)
path:
type: string
description: The unique Windmill path for this trigger. Must be of the form `u/<user>/<path>` or `f/<folder>/<path>`.
script_path:
type: string
description: Path to the script or flow to execute when a message is received
is_flow:
type: boolean
description: True if script_path points to a flow, false if it points to a script
mode:
$ref: "#/components/schemas/TriggerMode"
error_handler_path:
type: string
description: Path to a script or flow to run when the triggered job fails
error_handler_args:
$ref: "#/components/schemas/ScriptArgs"
description: Arguments to pass to the error handler
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
permissioned_as:
type: string
description: The user or group this trigger runs as. Used during deployment to preserve the original trigger owner.
preserve_permissioned_as:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original permissioned_as value instead of overwriting it."
labels:
type: array
items:
type: string
required:
- path
- script_path
- is_flow
- amqp_resource_path
- queue_name
EditAmqpTrigger:
type: object
properties:
amqp_resource_path:
type: string
description: Path to the AMQP resource containing broker connection configuration
queue_name:
type: string
description: Name of the queue to consume messages from
exchange:
nullable: true
$ref: "#/components/schemas/AmqpExchange"
description: Optional exchange binding for the consumed queue
options:
nullable: true
$ref: "#/components/schemas/AmqpOptions"
description: Optional consumer options (queue declaration, prefetch)
path:
type: string
description: The unique Windmill path for this trigger. Must be of the form `u/<user>/<path>` or `f/<folder>/<path>`.
script_path:
type: string
description: Path to the script or flow to execute when a message is received
is_flow:
type: boolean
description: True if script_path points to a flow, false if it points to a script
mode:
$ref: "#/components/schemas/TriggerMode"
error_handler_path:
type: string
description: Path to a script or flow to run when the triggered job fails
error_handler_args:
$ref: "#/components/schemas/ScriptArgs"
description: Arguments to pass to the error handler
retry:
$ref: "../../openflow.openapi.yaml#/components/schemas/Retry"
description: Retry configuration for failed executions
permissioned_as:
type: string
description: The user or group this trigger runs as. Used during deployment to preserve the original trigger owner.
preserve_permissioned_as:
type: boolean
description: "When true and the caller is a member of the 'wm_deployers' group, preserves the original permissioned_as value instead of overwriting it."
labels:
type: array
items:
type: string
required:
- path
- script_path
- is_flow
- amqp_resource_path
- queue_name
DeliveryType:
type: string
enum:
@@ -30393,6 +30785,7 @@ components:
postgres,
sqs,
mqtt,
amqp,
gcp,
azure,
email,
@@ -30575,6 +30968,7 @@ components:
"nats_trigger",
"postgres_trigger",
"mqtt_trigger",
"amqp_trigger",
"sqs_trigger",
"gcp_trigger",
"azure_trigger",
backend/windmill-api/src/capture.rs
+12
View File
@@ -48,6 +48,8 @@ use windmill_common::error::Error;
#[cfg(all(feature = "enterprise", feature = "kafka", feature = "private"))]
use crate::triggers::kafka::KafkaTriggerConfigConnection;
#[cfg(feature = "amqp_trigger")]
use crate::triggers::amqp::{AmqpOptions, ExchangeConfig};
#[cfg(feature = "mqtt_trigger")]
use crate::triggers::mqtt::{MqttClientVersion, MqttV3Config, MqttV5Config, SubscribeTopic};
@@ -234,6 +236,14 @@ pub struct MqttTriggerConfig {
pub client_version: Option<MqttClientVersion>,
pub client_id: Option<String>,
}
#[cfg(feature = "amqp_trigger")]
#[derive(Debug, Serialize, Deserialize)]
pub struct AmqpTriggerConfig {
pub amqp_resource_path: String,
pub queue_name: String,
pub exchange: Option<ExchangeConfig>,
pub options: Option<AmqpOptions>,
}
#[cfg(feature = "postgres_trigger")]
#[derive(Serialize, Deserialize, Debug)]
pub struct PostgresTriggerConfig {
@@ -271,6 +281,8 @@ enum TriggerConfig {
Nats(NatsTriggerConfig),
#[cfg(feature = "mqtt_trigger")]
Mqtt(MqttTriggerConfig),
#[cfg(feature = "amqp_trigger")]
Amqp(AmqpTriggerConfig),
#[cfg(all(feature = "enterprise", feature = "gcp_trigger", feature = "private"))]
Gcp(GcpTriggerConfig),
#[cfg(all(feature = "enterprise", feature = "azure_trigger", feature = "private"))]
backend/windmill-api/src/offboarding.rs
+4 -1
View File
@@ -186,6 +186,7 @@ async fn get_offboard_preview(
"kafka_trigger",
"postgres_trigger",
"mqtt_trigger",
"amqp_trigger",
"nats_trigger",
"sqs_trigger",
"gcp_trigger",
@@ -750,6 +751,7 @@ async fn check_path_conflicts(
"kafka_trigger",
"postgres_trigger",
"mqtt_trigger",
"amqp_trigger",
"nats_trigger",
"sqs_trigger",
"gcp_trigger",
@@ -1017,13 +1019,14 @@ async fn offboard_user_from_workspace<'c>(
&new_permissioned_as, username, w_id
).execute(&mut **tx).await?;
// ---- triggers (all 9 types with path/permissioned_as) ----
// ---- triggers (all 10 types with path/permissioned_as) ----
let trigger_tables = [
"http_trigger",
"websocket_trigger",
"kafka_trigger",
"postgres_trigger",
"mqtt_trigger",
"amqp_trigger",
"nats_trigger",
"sqs_trigger",
"gcp_trigger",
backend/windmill-api/src/token.rs
+1
View File
@@ -24,6 +24,7 @@ fn build_trigger_scope_domains() -> Vec<ScopeDomain> {
("kafka_triggers", "Kafka"),
("nats_triggers", "NATS"),
("mqtt_triggers", "MQTT"),
("amqp_triggers", "AMQP"),
("sqs_triggers", "AWS SQS"),
("gcp_triggers", "GCP Pub/Sub"),
("azure_triggers", "Azure Event Grid"),
backend/windmill-api/src/trash.rs
+1
View File
@@ -467,6 +467,7 @@ async fn restore_trigger(tx: &mut sqlx::PgConnection, item: &TrashItemWithData)
"nats_trigger",
"postgres_trigger",
"mqtt_trigger",
"amqp_trigger",
"sqs_trigger",
"gcp_trigger",
"azure_trigger",
backend/windmill-api/src/triggers/amqp/mod.rs
+1
View File
@@ -0,0 +1 @@
pub use windmill_trigger_amqp::*;
backend/windmill-api/src/triggers/handler.rs
+23
View File
@@ -61,6 +61,16 @@ pub fn generate_trigger_routers() -> Router {
);
}
#[cfg(feature = "amqp_trigger")]
{
use crate::triggers::amqp::AmqpTrigger;
router = router.nest(
AmqpTrigger::ROUTE_PREFIX,
complete_trigger_routes(AmqpTrigger),
);
}
#[cfg(all(feature = "enterprise", feature = "sqs_trigger", feature = "private"))]
{
use crate::triggers::sqs::SqsTrigger;
@@ -143,6 +153,7 @@ pub struct TriggersCount {
nats_count: i64,
postgres_count: i64,
mqtt_count: i64,
amqp_count: i64,
sqs_count: i64,
gcp_count: i64,
azure_count: i64,
@@ -245,6 +256,17 @@ pub async fn get_triggers_count_internal(
#[cfg(not(feature = "mqtt_trigger"))]
let mqtt_count = 0;
#[cfg(feature = "amqp_trigger")]
let amqp_count = {
use crate::triggers::amqp::AmqpTrigger;
let count = AmqpTrigger
.trigger_count(&mut tx, w_id, is_flow, path)
.await;
count
};
#[cfg(not(feature = "amqp_trigger"))]
let amqp_count = 0;
#[cfg(all(feature = "sqs_trigger", feature = "enterprise", feature = "private"))]
let sqs_count = {
use crate::triggers::sqs::SqsTrigger;
@@ -362,6 +384,7 @@ pub async fn get_triggers_count_internal(
nats_count,
postgres_count,
mqtt_count,
amqp_count,
gcp_count,
azure_count,
sqs_count,
backend/windmill-api/src/triggers/listener.rs
+8
View File
@@ -51,6 +51,14 @@ pub fn start_all_listeners(db: DB, killpill_rx: &tokio::sync::broadcast::Receive
listen_to(MqttTrigger, db.clone(), mqtt_killpill_rx)
}
#[cfg(feature = "amqp_trigger")]
{
let amqp_killpill_rx = killpill_rx.resubscribe();
use crate::triggers::amqp::AmqpTrigger;
listen_to(AmqpTrigger, db.clone(), amqp_killpill_rx)
}
#[cfg(feature = "websocket")]
{
let mqtt_killpill_rx = killpill_rx.resubscribe();
backend/windmill-api/src/triggers/mod.rs
+2
View File
@@ -1,4 +1,6 @@
// Concrete trigger submodules (feature-gated)
#[cfg(feature = "amqp_trigger")]
pub mod amqp;
#[cfg(all(feature = "azure_trigger", feature = "enterprise", feature = "private"))]
pub mod azure;
#[cfg(all(feature = "smtp", feature = "private"))]
backend/windmill-api/src/workspaces_export.rs
+33
View File
@@ -19,6 +19,7 @@ use windmill_api_auth::check_scopes;
feature = "websocket",
feature = "postgres_trigger",
feature = "mqtt_trigger",
feature = "amqp_trigger",
all(
feature = "enterprise",
any(
@@ -144,6 +145,7 @@ pub fn is_none_or_false(val: &Option<bool>) -> bool {
feature = "websocket",
feature = "postgres_trigger",
feature = "mqtt_trigger",
feature = "amqp_trigger",
feature = "native_trigger",
all(
feature = "enterprise",
@@ -189,6 +191,7 @@ async fn fork_parent_trigger_modes(
feature = "websocket",
feature = "postgres_trigger",
feature = "mqtt_trigger",
feature = "amqp_trigger",
feature = "native_trigger",
all(
feature = "enterprise",
@@ -1226,6 +1229,36 @@ pub(crate) async fn tarball_workspace(
}
}
#[cfg(feature = "amqp_trigger")]
{
use crate::triggers::amqp::AmqpTrigger;
let handler = AmqpTrigger;
let amqp_triggers = handler.list_triggers(&mut *tx, &w_id, None, None).await?;
let parent_modes = fork_parent_trigger_modes(
&db,
<AmqpTrigger as TriggerCrud>::TABLE_NAME,
parent_workspace_id.as_deref(),
)
.await?;
for trigger in amqp_triggers {
let mode_override = trigger_mode_override(&parent_modes, &trigger.base.path);
let trigger_str = &to_string_without_metadata_inner(
&trigger,
ExtraPermsBehavior::Drop,
None,
mode_override.as_ref(),
)
.unwrap();
archive
.write_to_archive(
&trigger_str,
&format!("{}.amqp_trigger.json", trigger.base.path),
)
.await?;
}
}
#[cfg(all(feature = "enterprise", feature = "smtp", feature = "private"))]
{
use crate::triggers::email::EmailTrigger;
backend/windmill-common/src/assets.rs
+1
View File
@@ -632,6 +632,7 @@ pub fn trigger_spec_to_row(spec: &TriggerSpec) -> Option<(ScriptTriggerKind, Str
| TriggerSpec::Email
| TriggerSpec::Kafka
| TriggerSpec::Mqtt
| TriggerSpec::Amqp
| TriggerSpec::Nats
| TriggerSpec::Postgres
| TriggerSpec::Sqs
backend/windmill-common/src/triggers.rs
+2 -1
View File
@@ -39,7 +39,8 @@ pub async fn update_triggers_script_path(
t3 AS (UPDATE postgres_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), \
t4 AS (UPDATE mqtt_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), \
t5 AS (UPDATE nats_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), \
t6 AS (UPDATE sqs_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4) \
t6 AS (UPDATE sqs_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4), \
t7 AS (UPDATE amqp_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4) \
UPDATE gcp_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4",
new_path,
old_path,
backend/windmill-common/src/user_drafts.rs
+5 -1
View File
@@ -45,6 +45,7 @@ pub enum UserDraftItemKind {
TriggerKafka,
TriggerNats,
TriggerMqtt,
TriggerAmqp,
TriggerSqs,
TriggerGcp,
TriggerAzure,
@@ -80,6 +81,7 @@ impl UserDraftItemKind {
UserDraftItemKind::TriggerKafka => "trigger_kafka",
UserDraftItemKind::TriggerNats => "trigger_nats",
UserDraftItemKind::TriggerMqtt => "trigger_mqtt",
UserDraftItemKind::TriggerAmqp => "trigger_amqp",
UserDraftItemKind::TriggerSqs => "trigger_sqs",
UserDraftItemKind::TriggerGcp => "trigger_gcp",
UserDraftItemKind::TriggerAzure => "trigger_azure",
@@ -94,7 +96,7 @@ impl UserDraftItemKind {
/// Every variant, for code that must enumerate kinds (e.g. generating
/// the `draft_only` existence SQL).
pub const ALL: [UserDraftItemKind; 25] = [
pub const ALL: [UserDraftItemKind; 26] = [
UserDraftItemKind::Script,
UserDraftItemKind::Flow,
UserDraftItemKind::App,
@@ -111,6 +113,7 @@ impl UserDraftItemKind {
UserDraftItemKind::TriggerKafka,
UserDraftItemKind::TriggerNats,
UserDraftItemKind::TriggerMqtt,
UserDraftItemKind::TriggerAmqp,
UserDraftItemKind::TriggerSqs,
UserDraftItemKind::TriggerGcp,
UserDraftItemKind::TriggerAzure,
@@ -144,6 +147,7 @@ impl UserDraftItemKind {
TriggerKafka => Some("kafka_trigger"),
TriggerNats => Some("nats_trigger"),
TriggerMqtt => Some("mqtt_trigger"),
TriggerAmqp => Some("amqp_trigger"),
TriggerSqs => Some("sqs_trigger"),
TriggerGcp => Some("gcp_trigger"),
TriggerAzure => Some("azure_trigger"),
backend/windmill-git-sync/src/lib.rs
+11
View File
@@ -93,6 +93,10 @@ pub enum DeployedObject {
path: String,
parent_path: Option<String>,
},
AmqpTrigger {
path: String,
parent_path: Option<String>,
},
SqsTrigger {
path: String,
parent_path: Option<String>,
@@ -144,6 +148,7 @@ impl DeployedObject {
DeployedObject::NatsTrigger { path, .. } => path.to_owned(),
DeployedObject::PostgresTrigger { path, .. } => path.to_owned(),
DeployedObject::MqttTrigger { path, .. } => path.to_owned(),
DeployedObject::AmqpTrigger { path, .. } => path.to_owned(),
DeployedObject::SqsTrigger { path, .. } => path.to_owned(),
DeployedObject::GcpTrigger { path, .. } => path.to_owned(),
DeployedObject::AzureTrigger { path, .. } => path.to_owned(),
@@ -186,6 +191,7 @@ impl DeployedObject {
DeployedObject::NatsTrigger { parent_path, .. } => parent_path.to_owned(),
DeployedObject::PostgresTrigger { parent_path, .. } => parent_path.to_owned(),
DeployedObject::MqttTrigger { parent_path, .. } => parent_path.to_owned(),
DeployedObject::AmqpTrigger { parent_path, .. } => parent_path.to_owned(),
DeployedObject::SqsTrigger { parent_path, .. } => parent_path.to_owned(),
DeployedObject::GcpTrigger { parent_path, .. } => parent_path.to_owned(),
DeployedObject::AzureTrigger { parent_path, .. } => parent_path.to_owned(),
@@ -216,6 +222,7 @@ impl DeployedObject {
DeployedObject::NatsTrigger { .. } => "nats_trigger",
DeployedObject::PostgresTrigger { .. } => "postgres_trigger",
DeployedObject::MqttTrigger { .. } => "mqtt_trigger",
DeployedObject::AmqpTrigger { .. } => "amqp_trigger",
DeployedObject::SqsTrigger { .. } => "sqs_trigger",
DeployedObject::GcpTrigger { .. } => "gcp_trigger",
DeployedObject::AzureTrigger { .. } => "azure_trigger",
@@ -446,6 +453,10 @@ mod tests {
DeployedObject::MqttTrigger { path: "t".to_string(), parent_path: None }.get_kind(),
"mqtt_trigger"
);
assert_eq!(
DeployedObject::AmqpTrigger { path: "t".to_string(), parent_path: None }.get_kind(),
"amqp_trigger"
);
assert_eq!(
DeployedObject::SqsTrigger { path: "t".to_string(), parent_path: None }.get_kind(),
"sqs_trigger"
backend/windmill-store/Cargo.toml
+1
View File
@@ -19,6 +19,7 @@ mcp = ["dep:windmill-mcp", "windmill-mcp/server", "windmill-mcp/auth"]
http_trigger = []
postgres_trigger = []
mqtt_trigger = []
amqp_trigger = []
sqs_trigger = []
gcp_trigger = []
azure_trigger = []
backend/windmill-store/src/resources.rs
+1
View File
@@ -2386,6 +2386,7 @@ async fn update_resource_type(
feature = "http_trigger",
feature = "postgres_trigger",
feature = "mqtt_trigger",
feature = "amqp_trigger",
all(
feature = "enterprise",
any(
backend/windmill-trigger-amqp/Cargo.toml
+37
View File
@@ -0,0 +1,37 @@
[package]
name = "windmill-trigger-amqp"
version.workspace = true
authors.workspace = true
edition.workspace = true
[lib]
name = "windmill_trigger_amqp"
path = "src/lib.rs"
[features]
default = []
enterprise = ["windmill-common/enterprise", "windmill-store/enterprise", "windmill-trigger/enterprise"]
private = ["windmill-common/private", "windmill-store/private"]
[dependencies]
windmill-common = { workspace = true, default-features = false }
windmill-api-auth.workspace = true
windmill-store = { workspace = true, features = ["amqp_trigger"] }
windmill-trigger.workspace = true
windmill-git-sync.workspace = true
lapin.workspace = true
tokio-executor-trait.workspace = true
tokio-reactor-trait.workspace = true
futures.workspace = true
urlencoding.workspace = true
axum.workspace = true
serde.workspace = true
serde_json.workspace = true
sqlx.workspace = true
tokio.workspace = true
tracing.workspace = true
async-trait.workspace = true
itertools.workspace = true
base64.workspace = true
anyhow.workspace = true
thiserror.workspace = true
backend/windmill-trigger-amqp/src/handler.rs
+203
View File
@@ -0,0 +1,203 @@
use async_trait::async_trait;
use sqlx::{types::Json as SqlxJson, PgConnection};
use windmill_api_auth::ApiAuthed;
use windmill_common::DB;
use windmill_common::{
db::UserDB,
error::{Error, Result},
};
use windmill_git_sync::DeployedObject;
use windmill_store::resources::try_get_resource_from_db_as;
use windmill_trigger::{Trigger, TriggerCrud, TriggerData};
use super::{
AmqpClientBuilder, AmqpConfig, AmqpConfigRequest, AmqpOptions, AmqpResource, AmqpTrigger,
ExchangeConfig, TestAmqpConfig,
};
#[async_trait]
impl TriggerCrud for AmqpTrigger {
type TriggerConfig = AmqpConfig;
type Trigger = Trigger<Self::TriggerConfig>;
type TriggerConfigRequest = AmqpConfigRequest;
type TestConnectionConfig = TestAmqpConfig;
const TABLE_NAME: &'static str = "amqp_trigger";
const TRIGGER_TYPE: &'static str = "amqp";
const DRAFT_KIND: windmill_common::user_drafts::UserDraftItemKind =
windmill_common::user_drafts::UserDraftItemKind::TriggerAmqp;
const SUPPORTS_SERVER_STATE: bool = true;
const SUPPORTS_TEST_CONNECTION: bool = true;
const ROUTE_PREFIX: &'static str = "/amqp_triggers";
const DEPLOYMENT_NAME: &'static str = "AMQP trigger";
const ADDITIONAL_SELECT_FIELDS: &[&'static str] =
&["amqp_resource_path", "queue_name", "exchange", "options"];
const IS_ALLOWED_ON_CLOUD: bool = false;
fn get_deployed_object(path: String, parent_path: Option<String>) -> DeployedObject {
DeployedObject::AmqpTrigger { path, parent_path }
}
async fn validate_config(
&self,
_db: &DB,
config: &Self::TriggerConfigRequest,
_workspace_id: &str,
) -> Result<()> {
if config.amqp_resource_path.trim().is_empty() {
return Err(Error::BadRequest(
"AMQP resource path cannot be empty".to_string(),
));
}
if config.queue_name.trim().is_empty() {
return Err(Error::BadRequest("Queue name cannot be empty".to_string()));
}
super::validate_amqp_options(config.options.as_ref()).map_err(Error::BadRequest)?;
Ok(())
}
async fn create_trigger(
&self,
_db: &DB,
tx: &mut PgConnection,
authed: &ApiAuthed,
w_id: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
) -> Result<()> {
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_permissioned_as = trigger.base.resolve_permissioned_as(authed);
let exchange = trigger.config.exchange.map(SqlxJson);
let options = trigger.config.options.map(SqlxJson);
sqlx::query!(
r#"
INSERT INTO amqp_trigger (
amqp_resource_path,
queue_name,
exchange,
options,
workspace_id,
path,
script_path,
is_flow,
permissioned_as,
mode,
edited_by,
error_handler_path,
error_handler_args,
retry
)
VALUES (
$1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14
)"#,
trigger.config.amqp_resource_path,
trigger.config.queue_name,
exchange as Option<SqlxJson<ExchangeConfig>>,
options as Option<SqlxJson<AmqpOptions>>,
w_id,
trigger.base.path,
trigger.base.script_path,
trigger.base.is_flow,
resolved_permissioned_as,
trigger.base.mode() as _,
&resolved_edited_by,
trigger.error_handling.error_handler_path,
trigger.error_handling.error_handler_args as _,
trigger.error_handling.retry as _
)
.execute(tx)
.await?;
Ok(())
}
async fn update_trigger(
&self,
_db: &DB,
tx: &mut PgConnection,
authed: &ApiAuthed,
workspace_id: &str,
path: &str,
trigger: TriggerData<Self::TriggerConfigRequest>,
) -> Result<()> {
let resolved_edited_by = trigger.base.resolve_edited_by(authed);
let resolved_permissioned_as = trigger.base.resolve_permissioned_as(authed);
let exchange = trigger.config.exchange.map(SqlxJson);
let options = trigger.config.options.map(SqlxJson);
// Important to set server_id to NULL to stop the current amqp listener
sqlx::query!(
r#"
UPDATE
amqp_trigger
SET
amqp_resource_path = $1,
queue_name = $2,
exchange = $3,
options = $4,
is_flow = $5,
edited_by = $6,
permissioned_as = $7,
script_path = $8,
path = $9,
edited_at = now(),
error = NULL,
server_id = NULL,
error_handler_path = $12,
error_handler_args = $13,
retry = $14
WHERE
workspace_id = $10 AND
path = $11
"#,
trigger.config.amqp_resource_path,
trigger.config.queue_name,
exchange as Option<SqlxJson<ExchangeConfig>>,
options as Option<SqlxJson<AmqpOptions>>,
trigger.base.is_flow,
&resolved_edited_by,
resolved_permissioned_as,
trigger.base.script_path,
trigger.base.path,
workspace_id,
path,
trigger.error_handling.error_handler_path,
trigger.error_handling.error_handler_args as _,
trigger.error_handling.retry as _
)
.execute(tx)
.await?;
Ok(())
}
async fn test_connection(
&self,
db: &DB,
authed: &ApiAuthed,
user_db: &UserDB,
workspace_id: &str,
config: Self::TestConnectionConfig,
) -> Result<()> {
let amqp_resource = try_get_resource_from_db_as::<AmqpResource>(
authed,
Some(user_db.clone()),
db,
&config.amqp_resource_path,
workspace_id,
)
.await?;
let client_builder = AmqpClientBuilder::new(amqp_resource, "", None, None);
client_builder
.test_connection()
.await
.map_err(|err| Error::BadConfig(format!("Error connecting to AMQP broker: {}", err)))?;
Ok(())
}
}
backend/windmill-trigger-amqp/src/lib.rs
+341
View File
@@ -0,0 +1,341 @@
use base64::engine;
use base64::prelude::*;
use lapin::{
options::{BasicConsumeOptions, BasicQosOptions, QueueBindOptions, QueueDeclareOptions},
types::FieldTable,
Channel, Connection, ConnectionProperties, Consumer,
};
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;
use sqlx::{types::Json as SqlxJson, FromRow};
use std::collections::HashMap;
use windmill_common::{error::Error, triggers::TriggerKind, worker::to_raw_value};
use windmill_trigger::trigger_helpers::TriggerJobArgs;
pub mod handler;
pub mod listener;
#[derive(Clone, Copy)]
pub struct AmqpTrigger;
impl TriggerJobArgs for AmqpTrigger {
type Payload = Vec<u8>;
const TRIGGER_KIND: TriggerKind = TriggerKind::Amqp;
fn v1_payload_fn(payload: &Self::Payload) -> HashMap<String, Box<RawValue>> {
HashMap::from([("payload".to_string(), to_raw_value(&payload))])
}
fn v2_payload_fn(payload: &Self::Payload) -> HashMap<String, Box<RawValue>> {
let base64_payload = engine::general_purpose::STANDARD.encode(payload);
HashMap::from([("payload".to_string(), to_raw_value(&base64_payload))])
}
}
#[derive(Debug, Deserialize)]
pub struct AmqpResource {
pub host: String,
pub port: Option<u16>,
pub username: Option<String>,
pub password: Option<String>,
pub vhost: Option<String>,
pub tls: Option<bool>,
}
/// Binding of the consumed queue to an exchange. When present, the queue is bound
/// to `exchange_name` for each routing key so messages published to the exchange
/// are routed to it.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ExchangeConfig {
pub exchange_name: String,
#[serde(default)]
pub routing_keys: Vec<String>,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct AmqpOptions {
/// Declare the queue (durable) before consuming. When false the queue is
/// declared passively, i.e. it must already exist on the broker.
pub declare_queue: Option<bool>,
/// Maximum number of unacknowledged messages the broker delivers at once.
pub prefetch_count: Option<u16>,
}
/// Shared validation for AMQP options used by both the CRUD handler and the
/// consumer builder (which also covers capture configs, that bypass CRUD
/// validation). RabbitMQ treats prefetch 0 as unlimited (unbounded consumer
/// buffer), so a set prefetch must be at least 1.
pub fn validate_amqp_options(options: Option<&AmqpOptions>) -> Result<(), String> {
if options.and_then(|o| o.prefetch_count) == Some(0) {
return Err("Prefetch count must be at least 1".to_string());
}
Ok(())
}
#[derive(Debug, Clone, FromRow, Serialize, Deserialize)]
pub struct AmqpConfig {
pub amqp_resource_path: String,
pub queue_name: String,
pub exchange: Option<SqlxJson<ExchangeConfig>>,
pub options: Option<SqlxJson<AmqpOptions>>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AmqpConfigRequest {
pub amqp_resource_path: String,
pub queue_name: String,
pub exchange: Option<ExchangeConfig>,
pub options: Option<AmqpOptions>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TestAmqpConfig {
pub amqp_resource_path: String,
}
#[derive(Debug, thiserror::Error)]
pub enum AmqpError {
#[error("{0}")]
Common(#[from] Error),
#[error("{0}")]
Lapin(#[from] lapin::Error),
}
/// Consumer bundle. The connection and channel are kept alive for as long as the
/// consumer stream is polled: dropping them would tear down the AMQP consumer.
pub struct AmqpConsumer {
pub connection: Connection,
pub channel: Channel,
pub consumer: Consumer,
pub queue_name: String,
}
pub const CONSUMER_TAG: &str = "windmill";
fn build_uri(resource: &AmqpResource) -> String {
let tls = resource.tls.unwrap_or(false);
let scheme = if tls { "amqps" } else { "amqp" };
let port = resource.port.unwrap_or(if tls { 5671 } else { 5672 });
let credentials = match (resource.username.as_deref(), resource.password.as_deref()) {
(Some(user), password) if !user.is_empty() => format!(
"{}:{}@",
urlencoding::encode(user),
urlencoding::encode(password.unwrap_or("")),
),
_ => String::new(),
};
// The URI path is the virtual host and must be percent-encoded; the default
// vhost "/" therefore becomes "%2F". An empty vhost also falls back to "/".
let vhost = match resource.vhost.as_deref() {
Some(v) if !v.is_empty() => v,
_ => "/",
};
let vhost_encoded = urlencoding::encode(vhost);
// Bracket an IPv6 literal host so `host:port` parses correctly.
let host = if resource.host.contains(':') && !resource.host.starts_with('[') {
format!("[{}]", resource.host)
} else {
resource.host.clone()
};
format!(
"{}://{}{}:{}/{}",
scheme, credentials, host, port, vhost_encoded
)
}
fn connection_properties() -> ConnectionProperties {
ConnectionProperties::default()
.with_executor(tokio_executor_trait::Tokio::current())
.with_reactor(tokio_reactor_trait::Tokio)
}
pub struct AmqpClientBuilder<'client> {
resource: AmqpResource,
queue_name: &'client str,
exchange: Option<&'client ExchangeConfig>,
options: Option<&'client AmqpOptions>,
}
impl<'client> AmqpClientBuilder<'client> {
pub fn new(
resource: AmqpResource,
queue_name: &'client str,
exchange: Option<&'client ExchangeConfig>,
options: Option<&'client AmqpOptions>,
) -> Self {
Self { resource, queue_name, exchange, options }
}
async fn connect(&self) -> Result<Connection, AmqpError> {
let uri = build_uri(&self.resource);
let connection = Connection::connect(&uri, connection_properties()).await?;
Ok(connection)
}
/// Establish a connection and open a channel to verify the broker is reachable
/// with the provided credentials.
pub async fn test_connection(&self) -> Result<(), AmqpError> {
let connection = self.connect().await?;
connection.create_channel().await?;
Ok(())
}
pub async fn build_consumer(&self) -> Result<AmqpConsumer, AmqpError> {
let connection = self.connect().await?;
let channel = connection.create_channel().await?;
validate_amqp_options(self.options).map_err(|e| AmqpError::Common(Error::BadConfig(e)))?;
if let Some(prefetch_count) = self.options.and_then(|o| o.prefetch_count) {
channel
.basic_qos(prefetch_count, BasicQosOptions::default())
.await?;
}
let declare_queue = self.options.and_then(|o| o.declare_queue).unwrap_or(true);
let queue_declare_options = QueueDeclareOptions {
passive: !declare_queue,
durable: declare_queue,
exclusive: false,
auto_delete: false,
nowait: false,
};
channel
.queue_declare(
self.queue_name,
queue_declare_options,
FieldTable::default(),
)
.await?;
if let Some(exchange) = self.exchange {
if !exchange.exchange_name.trim().is_empty() {
// Bind the queue for every routing key; an empty list binds once
// with an empty routing key (fanout exchanges ignore it anyway).
let routing_keys = if exchange.routing_keys.is_empty() {
vec![String::new()]
} else {
exchange.routing_keys.clone()
};
for routing_key in routing_keys {
channel
.queue_bind(
self.queue_name,
&exchange.exchange_name,
&routing_key,
QueueBindOptions::default(),
FieldTable::default(),
)
.await?;
}
}
}
let consumer = channel
.basic_consume(
self.queue_name,
CONSUMER_TAG,
BasicConsumeOptions::default(),
FieldTable::default(),
)
.await?;
Ok(AmqpConsumer { connection, channel, consumer, queue_name: self.queue_name.to_string() })
}
}
#[cfg(test)]
mod tests {
use super::*;
fn resource(
host: &str,
port: Option<u16>,
username: Option<&str>,
password: Option<&str>,
vhost: Option<&str>,
tls: Option<bool>,
) -> AmqpResource {
AmqpResource {
host: host.to_string(),
port,
username: username.map(str::to_string),
password: password.map(str::to_string),
vhost: vhost.map(str::to_string),
tls,
}
}
#[test]
fn build_uri_defaults_encode_root_vhost_and_pick_plaintext_port() {
let uri = build_uri(&resource("broker", None, None, None, None, None));
assert_eq!(uri, "amqp://broker:5672/%2F");
}
#[test]
fn build_uri_tls_picks_amqps_and_5671() {
let uri = build_uri(&resource("broker", None, None, None, None, Some(true)));
assert_eq!(uri, "amqps://broker:5671/%2F");
}
#[test]
fn build_uri_encodes_credentials_and_custom_vhost() {
let uri = build_uri(&resource(
"broker",
Some(5673),
Some("us er"),
Some("p@ss/word"),
Some("my/vhost"),
None,
));
assert_eq!(uri, "amqp://us%20er:p%40ss%2Fword@broker:5673/my%2Fvhost");
}
#[test]
fn build_uri_omits_credentials_when_username_empty() {
let uri = build_uri(&resource(
"broker",
None,
Some(""),
Some("secret"),
None,
None,
));
assert_eq!(uri, "amqp://broker:5672/%2F");
}
#[test]
fn build_uri_blank_vhost_falls_back_to_root() {
let uri = build_uri(&resource("broker", None, None, None, Some(""), None));
assert_eq!(uri, "amqp://broker:5672/%2F");
}
#[test]
fn build_uri_brackets_ipv6_host() {
let uri = build_uri(&resource("::1", Some(5672), None, None, None, None));
assert_eq!(uri, "amqp://[::1]:5672/%2F");
}
fn options(prefetch: Option<u16>) -> AmqpOptions {
AmqpOptions { declare_queue: None, prefetch_count: prefetch }
}
#[test]
fn validate_amqp_options_rejects_zero_prefetch() {
assert!(validate_amqp_options(Some(&options(Some(0)))).is_err());
}
#[test]
fn validate_amqp_options_accepts_valid_prefetch_and_none() {
assert!(validate_amqp_options(Some(&options(Some(1)))).is_ok());
assert!(validate_amqp_options(Some(&options(Some(65535)))).is_ok());
assert!(validate_amqp_options(Some(&options(None))).is_ok());
assert!(validate_amqp_options(None).is_ok());
}
}
backend/windmill-trigger-amqp/src/listener.rs
+255
View File
@@ -0,0 +1,255 @@
use std::{collections::HashMap, sync::Arc, time::Duration};
use async_trait::async_trait;
use futures::StreamExt;
use lapin::options::{BasicAckOptions, BasicNackOptions};
use tokio::sync::RwLock;
use windmill_common::{
db::UserDB,
error::{Error, Result},
jobs::JobTriggerKind,
utils::{report_critical_error, report_recovered_critical_error},
worker::to_raw_value,
DB,
};
use windmill_store::resources::try_get_resource_from_db_as;
use windmill_trigger::listener::ListeningTrigger;
use windmill_trigger::trigger_helpers::TriggerJobArgs;
use windmill_trigger::Listener;
use super::{AmqpClientBuilder, AmqpConfig, AmqpConsumer, AmqpResource, AmqpTrigger};
// lapin (like rdkafka) has no transparent reconnect, so — mirroring the Kafka
// trigger — the connection is (re)established in `consume` with a backoff retry
// loop rather than disabling the trigger on a transient broker outage.
const RECONNECT_BACKOFF_SECS: u64 = 30;
// Back off after a failed dispatch so a poison message that always fails can't
// spin a tight redelivery loop; the connection stays up for other messages.
const DISPATCH_FAILURE_BACKOFF_SECS: u64 = 5;
impl AmqpTrigger {
async fn build_amqp_consumer(
&self,
db: &DB,
listening_trigger: &ListeningTrigger<AmqpConfig>,
) -> Result<AmqpConsumer> {
let AmqpConfig { amqp_resource_path, queue_name, exchange, options } =
&listening_trigger.trigger_config;
let authed = listening_trigger
.authed(db, &Self::TRIGGER_KIND.to_string())
.await?;
let amqp_resource = try_get_resource_from_db_as::<AmqpResource>(
&authed,
Some(UserDB::new(db.clone())),
db,
amqp_resource_path,
&listening_trigger.workspace_id,
)
.await?;
let client_builder = AmqpClientBuilder::new(
amqp_resource,
queue_name,
exchange.as_ref().map(|e| &e.0),
options.as_ref().map(|o| &o.0),
);
client_builder
.build_consumer()
.await
.map_err(|e| Error::BadConfig(format!("Failed to build AMQP consumer: {}", e)))
}
}
#[async_trait]
impl Listener for AmqpTrigger {
type Consumer = ();
type Extra = ();
type ExtraState = ();
const JOB_TRIGGER_KIND: JobTriggerKind = JobTriggerKind::Amqp;
async fn get_consumer(
&self,
_db: &DB,
_listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
_err_message: Arc<RwLock<Option<String>>>,
_killpill_rx: tokio::sync::broadcast::Receiver<()>,
) -> Result<Option<Self::Consumer>> {
// The connection is established (and re-established) in `consume` so that
// a transient broker outage retries with backoff instead of disabling the
// trigger — mirroring the Kafka trigger, whose client also lacks a
// transparent reconnect.
Ok(Some(()))
}
async fn consume(
&self,
db: &DB,
_consumer: Self::Consumer,
listening_trigger: &ListeningTrigger<Self::TriggerConfig>,
err_message: Arc<RwLock<Option<String>>>,
_killpill_rx: tokio::sync::broadcast::Receiver<()>,
_extra_state: Option<&Self::ExtraState>,
) {
let path = &listening_trigger.path;
let workspace_id = &listening_trigger.workspace_id;
let alert_id = format!("amqp_trigger:{}", path);
let mut tries = 0_usize;
// (Re)connect loop: retries forever with backoff; the framework's
// `select!` around `consume` cancels it on killpill, and
// `update_ping_and_loop_ping_status` returning None (trigger removed /
// disabled / capture stopped) breaks us out.
loop {
let mut consumer = match self.build_amqp_consumer(db, listening_trigger).await {
Ok(consumer) => consumer,
Err(e) => {
let status = format!(
"Failed to connect (attempt {}), retrying in {}s: {}",
tries + 1,
RECONNECT_BACKOFF_SECS,
e
);
if self
.update_ping_and_loop_ping_status(
db,
listening_trigger,
err_message.clone(),
Some(status),
)
.await
.is_none()
{
return;
}
tracing::error!(
"AMQP trigger {} failed to connect (attempt {}): {}",
path,
tries + 1,
e
);
if tries % 10 == 0 && listening_trigger.trigger_mode {
report_critical_error(
format!(
"Failed to connect AMQP trigger {} (attempt {}), retrying every {}s. This alert repeats every 10 failed attempts. Error: {}",
path, tries + 1, RECONNECT_BACKOFF_SECS, e
),
db.clone(),
Some(workspace_id),
Some(&alert_id),
)
.await;
}
tries += 1;
tokio::time::sleep(Duration::from_secs(RECONNECT_BACKOFF_SECS)).await;
continue;
}
};
// Connected: clear any "reconnecting" status.
if self
.update_ping_and_loop_ping_status(db, listening_trigger, err_message.clone(), None)
.await
.is_none()
{
return;
}
if tries > 0 {
tracing::info!("AMQP trigger {} reconnected after {} attempts", path, tries);
if listening_trigger.trigger_mode {
report_recovered_critical_error(
format!("AMQP trigger {} reconnected", path),
db.clone(),
Some(workspace_id),
Some(&alert_id),
)
.await;
}
tries = 0;
}
// Consume until the stream errors, then break out to reconnect.
loop {
match consumer.consumer.next().await {
Some(Ok(delivery)) => {
let trigger_info = HashMap::from([
(
"exchange".to_string(),
to_raw_value(&delivery.exchange.as_str()),
),
(
"routing_key".to_string(),
to_raw_value(&delivery.routing_key.as_str()),
),
("queue_name".to_string(), to_raw_value(&consumer.queue_name)),
(
"redelivered".to_string(),
to_raw_value(&delivery.redelivered),
),
(
"delivery_tag".to_string(),
to_raw_value(&delivery.delivery_tag),
),
]);
let dispatched = self
.handle_event(
db,
listening_trigger,
delivery.data.clone(),
trigger_info,
None,
)
.await;
// Only ack once the job/capture was dispatched. On failure
// nack with requeue so the broker redelivers rather than
// dropping the message (at-least-once).
let dispatch_failed = dispatched.is_err();
let ack_result = if dispatch_failed {
delivery
.acker
.nack(BasicNackOptions { requeue: true, multiple: false })
.await
} else {
delivery.acker.ack(BasicAckOptions::default()).await
};
if let Err(err) = ack_result {
// Channel is gone; break out to reconnect.
tracing::warn!(
"AMQP trigger {} ack/nack failed, reconnecting: {}",
path,
err
);
break;
}
if dispatch_failed {
// The message was requeued: back off before consuming
// again so a poison message can't spin a tight
// redelivery loop, while keeping the connection alive.
tokio::time::sleep(Duration::from_secs(DISPATCH_FAILURE_BACKOFF_SECS))
.await;
}
}
Some(Err(err)) => {
tracing::warn!(
"AMQP trigger {} consumer error, reconnecting: {}",
path,
err
);
break;
}
None => {
tracing::warn!("AMQP trigger {} consumer stream ended, reconnecting", path);
break;
}
}
}
}
}
}
backend/windmill-trigger/src/trigger_helpers.rs
+1 -1
View File
@@ -140,7 +140,7 @@ fn runnable_format_from_schema_without_preprocessor(
schema: Option<sqlx::types::Json<PartialSchema>>,
) -> RunnableFormat {
match trigger_kind {
TriggerKind::Mqtt
TriggerKind::Mqtt | TriggerKind::Amqp
if schema.as_ref().is_some_and(|schema| {
schema.properties.as_ref().is_some_and(|properties| {
properties.iter().any(|(key, def)| {
backend/windmill-types/src/jobs.rs
+2
View File
@@ -32,6 +32,7 @@ pub enum JobTriggerKind {
Email,
Nats,
Mqtt,
Amqp,
Sqs,
Postgres,
Schedule,
@@ -66,6 +67,7 @@ impl std::fmt::Display for JobTriggerKind {
JobTriggerKind::Email => "email",
JobTriggerKind::Nats => "nats",
JobTriggerKind::Mqtt => "mqtt",
JobTriggerKind::Amqp => "amqp",
JobTriggerKind::Sqs => "sqs",
JobTriggerKind::Postgres => "postgres",
JobTriggerKind::Schedule => "schedule",
backend/windmill-types/src/triggers.rs
+3
View File
@@ -16,6 +16,7 @@ pub enum TriggerKind {
Email,
Nats,
Mqtt,
Amqp,
Sqs,
Postgres,
Gcp,
@@ -36,6 +37,7 @@ impl TriggerKind {
TriggerKind::DefaultEmail => "email".to_string(),
TriggerKind::Nats => "nats".to_string(),
TriggerKind::Mqtt => "mqtt".to_string(),
TriggerKind::Amqp => "amqp".to_string(),
TriggerKind::Sqs => "sqs".to_string(),
TriggerKind::Postgres => "postgres".to_string(),
TriggerKind::Gcp => "gcp".to_string(),
@@ -58,6 +60,7 @@ impl fmt::Display for TriggerKind {
TriggerKind::DefaultEmail => "default_email",
TriggerKind::Nats => "nats",
TriggerKind::Mqtt => "mqtt",
TriggerKind::Amqp => "amqp",
TriggerKind::Sqs => "sqs",
TriggerKind::Postgres => "postgres",
TriggerKind::Gcp => "gcp",
cli/src/commands/pipeline/boundedCascade.ts
+1
View File
@@ -71,6 +71,7 @@ const assetNodeId = (kind: string, path: string): string => `${kind}:${path}`;
const NON_AUTORUN_TRIGGER_KINDS = new Set([
"kafka",
"mqtt",
"amqp",
"nats",
"postgres",
"sqs",
cli/src/commands/pipeline/localGraph.ts
+1
View File
@@ -259,6 +259,7 @@ const NATIVE_KINDS = new Set([
"email",
"kafka",
"mqtt",
"amqp",
"nats",
"postgres",
"sqs",
cli/src/commands/sync/sync.ts
+9 -1
View File
@@ -1717,6 +1717,7 @@ export async function elementsToMap(
path.endsWith(".nats_trigger" + ext) ||
path.endsWith(".postgres_trigger" + ext) ||
path.endsWith(".mqtt_trigger" + ext) ||
path.endsWith(".amqp_trigger" + ext) ||
path.endsWith(".sqs_trigger" + ext) ||
path.endsWith(".gcp_trigger" + ext) ||
path.endsWith(".azure_trigger" + ext) ||
@@ -2470,6 +2471,7 @@ function getOrderFromPath(p: string) {
typ == "nats_trigger" ||
typ == "postgres_trigger" ||
typ == "mqtt_trigger" ||
typ == "amqp_trigger" ||
typ == "sqs_trigger" ||
typ == "gcp_trigger" ||
typ == "azure_trigger" ||
@@ -4237,7 +4239,7 @@ export async function push(
}
}
const rules = folderRulesCache.get(folderName)!;
const remotePath = change.path.replace(/\.(script|schedule|http_trigger|websocket_trigger|kafka_trigger|nats_trigger|postgres_trigger|mqtt_trigger|sqs_trigger|gcp_trigger|azure_trigger|email_trigger)\.(yaml|json)$/, "").replace(/(\.flow|__flow)\/flow\.(yaml|json)$/, "").replace(/\.(app|raw_app)(\/app\.(yaml|json))?$/, "");
const remotePath = change.path.replace(/\.(script|schedule|http_trigger|websocket_trigger|kafka_trigger|nats_trigger|postgres_trigger|mqtt_trigger|amqp_trigger|sqs_trigger|gcp_trigger|azure_trigger|email_trigger)\.(yaml|json)$/, "").replace(/(\.flow|__flow)\/flow\.(yaml|json)$/, "").replace(/\.(app|raw_app)(\/app\.(yaml|json))?$/, "");
const relative = remotePath.slice(`f/${folderName}/`.length);
if (!relative) continue;
for (const rule of rules) {
@@ -4931,6 +4933,12 @@ export async function push(
path: removeSuffix(target, ".mqtt_trigger.json"),
});
break;
case "amqp_trigger":
await wmill.deleteAmqpTrigger({
workspace: workspaceId,
path: removeSuffix(target, ".amqp_trigger.json"),
});
break;
case "sqs_trigger":
await wmill.deleteSqsTrigger({
workspace: workspaceId,
cli/src/commands/trigger/trigger.ts
+18 -3
View File
@@ -9,6 +9,7 @@ import {
HttpTrigger,
KafkaTrigger,
MqttTrigger,
AmqpTrigger,
NatsTrigger,
PostgresTrigger,
SqsTrigger,
@@ -48,6 +49,7 @@ type Trigger = {
nats: NatsTrigger;
postgres: PostgresTrigger;
mqtt: MqttTrigger;
amqp: AmqpTrigger;
sqs: SqsTrigger;
gcp: GcpTrigger;
azure: AzureTrigger;
@@ -84,6 +86,7 @@ async function getTrigger<K extends TriggerType>(
nats: wmill.getNatsTrigger,
postgres: wmill.getPostgresTrigger,
mqtt: wmill.getMqttTrigger,
amqp: wmill.getAmqpTrigger,
sqs: wmill.getSqsTrigger,
gcp: wmill.getGcpTrigger,
azure: wmill.getAzureTrigger,
@@ -114,6 +117,7 @@ async function updateTrigger<K extends TriggerType>(
nats: wmill.updateNatsTrigger,
postgres: wmill.updatePostgresTrigger,
mqtt: wmill.updateMqttTrigger,
amqp: wmill.updateAmqpTrigger,
sqs: wmill.updateSqsTrigger,
gcp: wmill.updateGcpTrigger,
azure: wmill.updateAzureTrigger,
@@ -142,6 +146,7 @@ async function createTrigger<K extends TriggerType>(
nats: wmill.createNatsTrigger,
postgres: wmill.createPostgresTrigger,
mqtt: wmill.createMqttTrigger,
amqp: wmill.createAmqpTrigger,
sqs: wmill.createSqsTrigger,
gcp: wmill.createGcpTrigger,
azure: wmill.createAzureTrigger,
@@ -381,6 +386,13 @@ const triggerTemplates: Record<TriggerType, Record<string, any>> = {
subscribe_topics: [],
enabled: false,
},
amqp: {
script_path: "",
is_flow: false,
amqp_resource_path: "",
queue_name: "",
enabled: false,
},
sqs: {
script_path: "",
is_flow: false,
@@ -536,6 +548,7 @@ async function list(opts: GlobalOptions & { json?: boolean }) {
natsTriggers,
postgresTriggers,
mqttTriggers,
amqpTriggers,
sqsTriggers,
gcpTriggers,
azureTriggers,
@@ -547,6 +560,7 @@ async function list(opts: GlobalOptions & { json?: boolean }) {
listOrEmpty(() => wmill.listNatsTriggers({ workspace: ws })),
listOrEmpty(() => wmill.listPostgresTriggers({ workspace: ws })),
listOrEmpty(() => wmill.listMqttTriggers({ workspace: ws })),
listOrEmpty(() => wmill.listAmqpTriggers({ workspace: ws })),
listOrEmpty(() => wmill.listSqsTriggers({ workspace: ws })),
listOrEmpty(() => wmill.listGcpTriggers({ workspace: ws })),
listOrEmpty(() => wmill.listAzureTriggers({ workspace: ws })),
@@ -559,6 +573,7 @@ async function list(opts: GlobalOptions & { json?: boolean }) {
...natsTriggers.map((x) => ({ path: x.path, kind: "nats" })),
...postgresTriggers.map((x) => ({ path: x.path, kind: "postgres" })),
...mqttTriggers.map((x) => ({ path: x.path, kind: "mqtt" })),
...amqpTriggers.map((x) => ({ path: x.path, kind: "amqp" })),
...sqsTriggers.map((x) => ({ path: x.path, kind: "sqs" })),
...gcpTriggers.map((x) => ({ path: x.path, kind: "gcp" })),
...azureTriggers.map((x) => ({ path: x.path, kind: "azure" })),
@@ -643,11 +658,11 @@ const command = new Command()
.command("get", "get a trigger's details")
.arguments("<path:string>")
.option("--json", "Output as JSON (for piping to jq)")
.option("--kind <kind:string>", "Trigger kind (http, websocket, kafka, nats, postgres, mqtt, sqs, gcp, azure, email). Recommended for faster lookup")
.option("--kind <kind:string>", "Trigger kind (http, websocket, kafka, nats, postgres, mqtt, amqp, sqs, gcp, azure, email). Recommended for faster lookup")
.action(get as any)
.command("new", "create a new trigger locally")
.arguments("<path:string>")
.option("--kind <kind:string>", "Trigger kind (required: http, websocket, kafka, nats, postgres, mqtt, sqs, gcp, azure, email)")
.option("--kind <kind:string>", "Trigger kind (required: http, websocket, kafka, nats, postgres, mqtt, amqp, sqs, gcp, azure, email)")
.action(newTrigger as any)
.command(
"push",
@@ -662,7 +677,7 @@ const command = new Command()
.arguments("<path:string> <email:string>")
.option(
"--kind <kind:string>",
"Trigger kind (required: http, websocket, kafka, nats, postgres, mqtt, sqs, gcp, azure, email)"
"Trigger kind (required: http, websocket, kafka, nats, postgres, mqtt, amqp, sqs, gcp, azure, email)"
)
.action((async (opts: any, triggerPath: string, email: string) => {
const workspace = await resolveWorkspace(opts);
cli/src/guidance/core.ts
+1 -1
View File
@@ -89,7 +89,7 @@ When a new app needs to be created, YOU run \`wmill app new\` yourself with \`--
## Triggers
You MUST use the \`triggers\` skill to configure HTTP routes, WebSocket, Kafka, NATS, SQS, MQTT, GCP, Azure, Email, or Postgres CDC triggers.
You MUST use the \`triggers\` skill to configure HTTP routes, WebSocket, Kafka, NATS, SQS, MQTT, AMQP, GCP, Azure, Email, or Postgres CDC triggers.
## Schedules
cli/src/guidance/skills.gen.ts
+118 -3
View File
@@ -7232,12 +7232,12 @@ trigger related commands
- \`--json\` - Output as JSON (for piping to jq)
- \`trigger get <path:string>\` - get a trigger's details
- \`--json\` - Output as JSON (for piping to jq)
- \`--kind <kind:string>\` - Trigger kind (http, websocket, kafka, nats, postgres, mqtt, sqs, gcp, azure, email). Recommended for faster lookup
- \`--kind <kind:string>\` - Trigger kind (http, websocket, kafka, nats, postgres, mqtt, amqp, sqs, gcp, azure, email). Recommended for faster lookup
- \`trigger new <path:string>\` - create a new trigger locally
- \`--kind <kind:string>\` - Trigger kind (required: http, websocket, kafka, nats, postgres, mqtt, sqs, gcp, azure, email)
- \`--kind <kind:string>\` - Trigger kind (required: http, websocket, kafka, nats, postgres, mqtt, amqp, sqs, gcp, azure, email)
- \`trigger push <file_path:string> <remote_path:string>\` - push a local trigger spec. This overrides any remote versions.
- \`trigger set-permissioned-as <path:string> <email:string>\` - Set the email (run-as user) for a trigger (requires admin or wm_deployers group)
- \`--kind <kind:string>\` - Trigger kind (required: http, websocket, kafka, nats, postgres, mqtt, sqs, gcp, azure, email)
- \`--kind <kind:string>\` - Trigger kind (required: http, websocket, kafka, nats, postgres, mqtt, amqp, sqs, gcp, azure, email)
### user
@@ -7509,6 +7509,120 @@ Both print the job result, are safe to run yourself, and don't deploy.
// YAML schema content for triggers and schedules
export const SCHEMAS: Record<string, string> = {
"amqp_trigger": `type: object
properties:
script_path:
type: string
description: Path to the script or flow to execute when triggered
permissioned_as:
type: string
description: The user or group this trigger runs as (permissioned_as)
is_flow:
type: boolean
description: True if script_path points to a flow, false if it points to a script
labels:
type: array
items:
type: string
draft_only:
type: boolean
description: 'True when this row is a per-user draft with no deployed
trigger at the same path. Set by list endpoints when
\`include_draft_only=true\` synthesizes the row from the
draft. Frontend renders a "Draft" badge.
'
is_draft:
type: boolean
description: 'True when the authed user has a per-user draft at this path
(over a deployed row or a synthesized draft-only row).
Frontend appends a \`*\` to the displayed name.
'
amqp_resource_path:
type: string
description: Path to the AMQP resource containing broker connection configuration
queue_name:
type: string
description: Name of the queue to consume messages from
exchange:
type: object
properties:
exchange_name:
type: string
description: Name of the exchange to bind the consumed queue to
routing_keys:
type: array
items:
type: string
description: Routing keys used to bind the queue to the exchange
options:
type: object
properties:
declare_queue:
type: boolean
description: Declare the queue (durable) before consuming; when false the
queue is declared passively and must already exist
prefetch_count:
type: integer
format: int32
minimum: 1
maximum: 65535
description: Maximum number of unacknowledged messages the broker delivers
at once (1-65535)
error_handler_path:
type: string
description: Path to a script or flow to run when the triggered job fails
error_handler_args:
type: object
description: The arguments to pass to the script or flow
retry:
type: object
properties:
constant:
type: object
description: Retry with constant delay between attempts
properties:
attempts:
type: integer
description: Number of retry attempts
seconds:
type: integer
description: Seconds to wait between retries
exponential:
type: object
description: Retry with exponential backoff (delay doubles each time)
properties:
attempts:
type: integer
description: Number of retry attempts
multiplier:
type: integer
description: Multiplier for exponential backoff
seconds:
type: integer
minimum: 1
description: Initial delay in seconds
random_factor:
type: integer
minimum: 0
maximum: 100
description: Random jitter percentage (0-100) to avoid thundering herd
retry_if:
$ref: '#/components/schemas/RetryIf'
description: Retry configuration for failed module executions
required:
- script_path
- permissioned_as
- is_flow
- amqp_resource_path
- queue_name
`,
"azure_trigger": `type: object
properties:
script_path:
@@ -8815,6 +8929,7 @@ export const SCHEMA_MAPPINGS: Record<string, SchemaMapping[]> = {
{ name: "NatsTrigger", schemaKey: "nats_trigger", filePattern: "*.nats_trigger.yaml" },
{ name: "PostgresTrigger", schemaKey: "postgres_trigger", filePattern: "*.postgres_trigger.yaml" },
{ name: "MqttTrigger", schemaKey: "mqtt_trigger", filePattern: "*.mqtt_trigger.yaml" },
{ name: "AmqpTrigger", schemaKey: "amqp_trigger", filePattern: "*.amqp_trigger.yaml" },
{ name: "SqsTrigger", schemaKey: "sqs_trigger", filePattern: "*.sqs_trigger.yaml" },
{ name: "GcpTrigger", schemaKey: "gcp_trigger", filePattern: "*.gcp_trigger.yaml" },
{ name: "AzureTrigger", schemaKey: "azure_trigger", filePattern: "*.azure_trigger.yaml" },
cli/src/types.ts
+5
View File
@@ -63,6 +63,7 @@ export const TRIGGER_TYPES = [
"nats",
"postgres",
"mqtt",
"amqp",
"sqs",
"gcp",
"azure",
@@ -243,6 +244,8 @@ export async function pushObj(
await pushTrigger("postgres", workspace, p, befObj, newObj, permissionedAsContext);
} else if (typeEnding === "mqtt_trigger") {
await pushTrigger("mqtt", workspace, p, befObj, newObj, permissionedAsContext);
} else if (typeEnding === "amqp_trigger") {
await pushTrigger("amqp", workspace, p, befObj, newObj, permissionedAsContext);
} else if (typeEnding === "sqs_trigger") {
await pushTrigger("sqs", workspace, p, befObj, newObj, permissionedAsContext);
} else if (typeEnding === "gcp_trigger") {
@@ -335,6 +338,7 @@ export function getTypeStrFromPath(
| "nats_trigger"
| "postgres_trigger"
| "mqtt_trigger"
| "amqp_trigger"
| "sqs_trigger"
| "gcp_trigger"
| "azure_trigger"
@@ -419,6 +423,7 @@ export function getTypeStrFromPath(
typeEnding === "nats_trigger" ||
typeEnding === "postgres_trigger" ||
typeEnding === "mqtt_trigger" ||
typeEnding === "amqp_trigger" ||
typeEnding === "sqs_trigger" ||
typeEnding === "gcp_trigger" ||
typeEnding === "azure_trigger" ||
cli/src/utils/git.ts
+2
View File
@@ -285,6 +285,8 @@ export function gitSyncIncludePattern(
return `${path}.postgres_trigger.*`;
case "mqtttrigger":
return `${path}.mqtt_trigger.*`;
case "amqptrigger":
return `${path}.amqp_trigger.*`;
case "sqstrigger":
return `${path}.sqs_trigger.*`;
case "gcptrigger":
cli/test/git_unit.test.ts
+3
View File
@@ -274,6 +274,9 @@ describe("gitSyncIncludePattern", () => {
expect(gitSyncIncludePattern("gcptrigger", "f/t")).toBe(
"f/t.gcp_trigger.*"
);
expect(gitSyncIncludePattern("amqptrigger", "f/t")).toBe(
"f/t.amqp_trigger.*"
);
});
});

Some files were not shown because too many files have changed in this diff Show More