mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-11 00:06:06 +00:00
[ee] feat: add asset-change implicit triggers foundation (stages 1-3)
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 4.7
Claude Opus 4.7
parent
eeacff24d9
commit
45d520738a
+2
-2
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"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) UPDATE gcp_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4",
|
||||
"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 asset_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4 AND is_implicit = false) 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": {
|
||||
@@ -13,5 +13,5 @@
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "38e782610bbabd5902e9b95ea6e026efb056f90e5cc6a505f7164d69ca046a3f"
|
||||
"hash": "03dceb989267d5bea286b83adedf43d53fd14f6f66206a805305e35abbc5a691"
|
||||
}
|
||||
+1
-1
@@ -15,7 +15,7 @@
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "5a219a2532517869578c4504ff3153c43903f929ae5d62fbba12610f89c36d55"
|
||||
|
||||
+23
@@ -0,0 +1,23 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "SELECT owner_script_path FROM asset_trigger WHERE workspace_id = $1 AND path = $2 AND is_implicit = true",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "owner_script_path",
|
||||
"type_info": "Varchar"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
true
|
||||
]
|
||||
},
|
||||
"hash": "941212fc8cc6db845d42320e337823424ba8a82b21efce174e1215094342c816"
|
||||
}
|
||||
+15
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "DELETE FROM asset_trigger WHERE workspace_id = $1 AND owner_script_path = $2 AND is_implicit = true",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "94e3820c19701da35d8f8c3441ffb0b696dee5eb77f74a9635852221528169aa"
|
||||
}
|
||||
+26
@@ -0,0 +1,26 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "INSERT INTO asset_trigger ( workspace_id, path, script_path, is_flow, owner_script_path, owner_script_hash, is_implicit, on_event, subscription_set, fires, debounce_s, partition_map, cancel_on_new, backlog, edited_by, email ) VALUES ( $1, $2, $3, false, $3, $4, true, $5, $6, $7, $8, $9, $10, $11, $12, $13 )",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Varchar",
|
||||
"Varchar",
|
||||
"Varchar",
|
||||
"Int8",
|
||||
"Varchar",
|
||||
"Jsonb",
|
||||
"Varchar",
|
||||
"Int4",
|
||||
"Jsonb",
|
||||
"Bool",
|
||||
"Varchar",
|
||||
"Varchar",
|
||||
"Varchar"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "e3d9b2bd73f1864255d27df33f2bdafbc126b6261c0046358022a345c4c8cb34"
|
||||
}
|
||||
+17
@@ -0,0 +1,17 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "UPDATE asset_trigger SET owner_script_path = $1, script_path = $1, path = $2, server_id = NULL WHERE workspace_id = $3 AND owner_script_path = $4 AND is_implicit = true",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Varchar",
|
||||
"Varchar",
|
||||
"Text",
|
||||
"Text"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "f5f544a5b75cc46ad1f418b24859c4a04ef4c74a6df1c5b49c6e0b9a3665060a"
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS asset_event;
|
||||
@@ -0,0 +1,25 @@
|
||||
-- Append-only event log for reactive asset-change triggers.
|
||||
-- Written whenever a job produces/updates an asset ('w' or 'rw' access).
|
||||
-- Stage 1 of asset-change implicit triggers plan.
|
||||
|
||||
CREATE TABLE asset_event (
|
||||
id BIGSERIAL PRIMARY KEY,
|
||||
workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
|
||||
asset_kind ASSET_KIND NOT NULL,
|
||||
asset_path VARCHAR(255) NOT NULL,
|
||||
partition_key VARCHAR(255),
|
||||
job_id UUID NOT NULL,
|
||||
script_hash BIGINT,
|
||||
access_type ASSET_ACCESS_TYPE NOT NULL,
|
||||
columns JSONB,
|
||||
metadata JSONB,
|
||||
at TIMESTAMPTZ NOT NULL DEFAULT now()
|
||||
);
|
||||
|
||||
CREATE INDEX asset_event_kind_path_at_idx ON asset_event (workspace_id, asset_kind, asset_path, at DESC);
|
||||
CREATE INDEX asset_event_job_id_idx ON asset_event (job_id);
|
||||
|
||||
GRANT ALL ON asset_event TO windmill_user;
|
||||
GRANT ALL ON asset_event TO windmill_admin;
|
||||
GRANT ALL ON asset_event_id_seq TO windmill_user;
|
||||
GRANT ALL ON asset_event_id_seq TO windmill_admin;
|
||||
@@ -0,0 +1 @@
|
||||
DROP TABLE IF EXISTS asset_trigger;
|
||||
@@ -0,0 +1,91 @@
|
||||
-- `asset_trigger` holds one row per reactive asset-change subscription.
|
||||
-- Implicit triggers (is_implicit=true) are projected from `#trigger: asset`
|
||||
-- annotations on every script save. Never editable via API/CLI.
|
||||
|
||||
CREATE TABLE asset_trigger (
|
||||
workspace_id VARCHAR(50) NOT NULL REFERENCES workspace(id) ON DELETE CASCADE ON UPDATE CASCADE,
|
||||
path VARCHAR(255) NOT NULL,
|
||||
script_path VARCHAR(255) NOT NULL,
|
||||
is_flow BOOLEAN NOT NULL DEFAULT false,
|
||||
|
||||
-- implicit-trigger pinning
|
||||
owner_script_path VARCHAR(255),
|
||||
owner_script_hash BIGINT,
|
||||
is_implicit BOOLEAN NOT NULL DEFAULT false,
|
||||
|
||||
-- trigger config
|
||||
on_event VARCHAR(32) NOT NULL,
|
||||
subscription_set JSONB NOT NULL,
|
||||
fires VARCHAR(16) NOT NULL DEFAULT 'all',
|
||||
debounce_s INT NOT NULL DEFAULT 30,
|
||||
partition_map JSONB,
|
||||
cancel_on_new BOOLEAN NOT NULL DEFAULT false,
|
||||
backlog VARCHAR(16) NOT NULL DEFAULT 'coalesce',
|
||||
|
||||
-- boilerplate shared with other trigger types
|
||||
error TEXT,
|
||||
server_id VARCHAR(64),
|
||||
last_server_ping TIMESTAMPTZ,
|
||||
error_handler_path VARCHAR(255),
|
||||
error_handler_args JSONB,
|
||||
retry JSONB,
|
||||
extra_perms JSONB NOT NULL DEFAULT '{}',
|
||||
edited_by VARCHAR(255) NOT NULL,
|
||||
edited_at TIMESTAMPTZ NOT NULL DEFAULT now(),
|
||||
email VARCHAR(255) NOT NULL,
|
||||
mode TRIGGER_MODE NOT NULL DEFAULT 'enabled',
|
||||
labels TEXT[],
|
||||
|
||||
PRIMARY KEY (workspace_id, path)
|
||||
);
|
||||
|
||||
CREATE INDEX asset_trigger_owner_idx ON asset_trigger (workspace_id, owner_script_path) WHERE is_implicit = true;
|
||||
CREATE INDEX asset_trigger_subscription_set_idx ON asset_trigger USING GIN (subscription_set);
|
||||
|
||||
GRANT ALL ON asset_trigger TO windmill_user;
|
||||
GRANT ALL ON asset_trigger TO windmill_admin;
|
||||
|
||||
ALTER TABLE asset_trigger ENABLE ROW LEVEL SECURITY;
|
||||
|
||||
CREATE POLICY admin_policy ON asset_trigger FOR ALL TO windmill_admin USING (true);
|
||||
|
||||
CREATE POLICY see_folder_extra_perms_user_select ON asset_trigger FOR SELECT TO windmill_user
|
||||
USING (SPLIT_PART(asset_trigger.path, '/', 1) = 'f' AND SPLIT_PART(asset_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_read'), ',')::text[]));
|
||||
CREATE POLICY see_folder_extra_perms_user_insert ON asset_trigger FOR INSERT TO windmill_user
|
||||
WITH CHECK (SPLIT_PART(asset_trigger.path, '/', 1) = 'f' AND SPLIT_PART(asset_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
|
||||
CREATE POLICY see_folder_extra_perms_user_update ON asset_trigger FOR UPDATE TO windmill_user
|
||||
USING (SPLIT_PART(asset_trigger.path, '/', 1) = 'f' AND SPLIT_PART(asset_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
|
||||
CREATE POLICY see_folder_extra_perms_user_delete ON asset_trigger FOR DELETE TO windmill_user
|
||||
USING (SPLIT_PART(asset_trigger.path, '/', 1) = 'f' AND SPLIT_PART(asset_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.folders_write'), ',')::text[]));
|
||||
|
||||
CREATE POLICY see_own ON asset_trigger FOR ALL TO windmill_user
|
||||
USING (SPLIT_PART(asset_trigger.path, '/', 1) = 'u' AND SPLIT_PART(asset_trigger.path, '/', 2) = current_setting('session.user'));
|
||||
CREATE POLICY see_member ON asset_trigger FOR ALL TO windmill_user
|
||||
USING (SPLIT_PART(asset_trigger.path, '/', 1) = 'g' AND SPLIT_PART(asset_trigger.path, '/', 2) = any(regexp_split_to_array(current_setting('session.groups'), ',')::text[]));
|
||||
|
||||
CREATE POLICY see_extra_perms_user_select ON asset_trigger FOR SELECT TO windmill_user
|
||||
USING (extra_perms ? CONCAT('u/', current_setting('session.user')));
|
||||
CREATE POLICY see_extra_perms_user_insert ON asset_trigger FOR INSERT TO windmill_user
|
||||
WITH CHECK ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
|
||||
CREATE POLICY see_extra_perms_user_update ON asset_trigger FOR UPDATE TO windmill_user
|
||||
USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
|
||||
CREATE POLICY see_extra_perms_user_delete ON asset_trigger FOR DELETE TO windmill_user
|
||||
USING ((extra_perms ->> CONCAT('u/', current_setting('session.user')))::boolean);
|
||||
|
||||
CREATE POLICY see_extra_perms_groups_select ON asset_trigger FOR SELECT TO windmill_user
|
||||
USING (extra_perms ?| regexp_split_to_array(current_setting('session.pgroups'), ',')::text[]);
|
||||
CREATE POLICY see_extra_perms_groups_insert ON asset_trigger FOR INSERT TO windmill_user
|
||||
WITH CHECK (exists(
|
||||
SELECT key, value FROM jsonb_each_text(extra_perms)
|
||||
WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
|
||||
AND value::boolean));
|
||||
CREATE POLICY see_extra_perms_groups_update ON asset_trigger FOR UPDATE TO windmill_user
|
||||
USING (exists(
|
||||
SELECT key, value FROM jsonb_each_text(extra_perms)
|
||||
WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
|
||||
AND value::boolean));
|
||||
CREATE POLICY see_extra_perms_groups_delete ON asset_trigger FOR DELETE TO windmill_user
|
||||
USING (exists(
|
||||
SELECT key, value FROM jsonb_each_text(extra_perms)
|
||||
WHERE SPLIT_PART(key, '/', 1) = 'g' AND key = ANY(regexp_split_to_array(current_setting('session.pgroups'), ',')::text[])
|
||||
AND value::boolean));
|
||||
@@ -1161,6 +1161,13 @@ async fn create_script_internal<'c>(
|
||||
"Error updating triggers due to runnable path change: {e:#}"
|
||||
))
|
||||
})?;
|
||||
windmill_common::implicit_triggers::rename_implicit_asset_triggers_for_script(
|
||||
&mut tx, &w_id, p_path, &ns.path,
|
||||
)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
error::Error::internal_err(format!("Error renaming implicit asset triggers: {e:#}"))
|
||||
})?;
|
||||
}
|
||||
|
||||
for schedule in schedulables {
|
||||
@@ -1277,6 +1284,19 @@ async fn create_script_internal<'c>(
|
||||
.await?;
|
||||
}
|
||||
|
||||
windmill_common::implicit_triggers::sync_implicit_asset_triggers_for_script(
|
||||
&mut tx,
|
||||
&w_id,
|
||||
&ns.path,
|
||||
hash.0,
|
||||
ns.language,
|
||||
&ns.content,
|
||||
ns.assets.as_deref().unwrap_or(&[]),
|
||||
&authed.email,
|
||||
&authed.username,
|
||||
)
|
||||
.await?;
|
||||
|
||||
let permissioned_as = username_to_permissioned_as(&authed.username);
|
||||
if let Some(parent_hash) = ns.parent_hash {
|
||||
tracing::info!(
|
||||
@@ -2179,6 +2199,10 @@ async fn archive_script_by_path(
|
||||
.map_err(|e| Error::internal_err(format!("archiving script in {w_id}: {e:#}")))?;
|
||||
|
||||
clear_static_asset_usage(&mut *tx, &w_id, path, AssetUsageKind::Script).await?;
|
||||
windmill_common::implicit_triggers::delete_implicit_asset_triggers_for_script(
|
||||
&mut tx, &w_id, path,
|
||||
)
|
||||
.await?;
|
||||
|
||||
audit_log(
|
||||
&mut *tx,
|
||||
@@ -2415,6 +2439,11 @@ async fn delete_script_by_path(
|
||||
.await
|
||||
.map_err(|e| Error::internal_err(format!("deleting script by path {w_id}: {e:#}")))?;
|
||||
|
||||
windmill_common::implicit_triggers::delete_implicit_asset_triggers_for_script(
|
||||
&mut tx, &w_id, path,
|
||||
)
|
||||
.await?;
|
||||
|
||||
if !trash_scripts.is_empty() {
|
||||
let mut trash_data = serde_json::json!({"scripts": trash_scripts});
|
||||
if !trash_drafts.is_empty() {
|
||||
|
||||
@@ -0,0 +1,149 @@
|
||||
//! Read-only HTTP surface for `asset_trigger` rows (stage 3).
|
||||
//!
|
||||
//! v1 of asset-change triggers exposes only implicit triggers (projected
|
||||
//! from `#trigger: asset` annotations in script source). Mutations go
|
||||
//! through the script save path, so every write endpoint here returns
|
||||
//! **409 Conflict** with a pointer back to the owning script.
|
||||
|
||||
use axum::{
|
||||
extract::{Extension, Path},
|
||||
http::StatusCode,
|
||||
routing::{delete, get, post},
|
||||
Json, Router,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sqlx::types::Json as SqlxJson;
|
||||
|
||||
use windmill_common::{
|
||||
db::UserDB,
|
||||
error::{Error, JsonResult, Result},
|
||||
utils::StripPath,
|
||||
};
|
||||
|
||||
use crate::db::ApiAuthed;
|
||||
|
||||
pub fn workspaced_service() -> Router {
|
||||
Router::new()
|
||||
.route("/list", get(list_asset_triggers))
|
||||
.route("/get/{*path}", get(get_asset_trigger))
|
||||
.route("/create", post(reject_mutate))
|
||||
.route("/update/{*path}", post(reject_mutate_path))
|
||||
.route("/delete/{*path}", delete(reject_mutate_path))
|
||||
}
|
||||
|
||||
#[derive(Serialize, Deserialize, Debug, sqlx::FromRow)]
|
||||
pub struct AssetTriggerRow {
|
||||
pub workspace_id: String,
|
||||
pub path: String,
|
||||
pub script_path: String,
|
||||
pub is_flow: bool,
|
||||
pub owner_script_path: Option<String>,
|
||||
pub owner_script_hash: Option<i64>,
|
||||
pub is_implicit: bool,
|
||||
pub on_event: String,
|
||||
pub subscription_set: SqlxJson<serde_json::Value>,
|
||||
pub fires: String,
|
||||
pub debounce_s: i32,
|
||||
pub partition_map: Option<SqlxJson<serde_json::Value>>,
|
||||
pub cancel_on_new: bool,
|
||||
pub backlog: String,
|
||||
pub error: Option<String>,
|
||||
pub edited_by: String,
|
||||
pub edited_at: chrono::DateTime<chrono::Utc>,
|
||||
pub email: String,
|
||||
}
|
||||
|
||||
async fn list_asset_triggers(
|
||||
authed: ApiAuthed,
|
||||
Extension(user_db): Extension<UserDB>,
|
||||
Path(w_id): Path<String>,
|
||||
) -> JsonResult<Vec<AssetTriggerRow>> {
|
||||
let mut tx = user_db.begin(&authed).await?;
|
||||
let rows = sqlx::query_as::<_, AssetTriggerRow>(
|
||||
"SELECT workspace_id, path, script_path, is_flow, owner_script_path, \
|
||||
owner_script_hash, is_implicit, on_event, subscription_set, fires, \
|
||||
debounce_s, partition_map, cancel_on_new, backlog, error, edited_by, \
|
||||
edited_at, email \
|
||||
FROM asset_trigger \
|
||||
WHERE workspace_id = $1 \
|
||||
ORDER BY edited_at DESC",
|
||||
)
|
||||
.bind(&w_id)
|
||||
.fetch_all(&mut *tx)
|
||||
.await?;
|
||||
tx.commit().await?;
|
||||
Ok(Json(rows))
|
||||
}
|
||||
|
||||
async fn get_asset_trigger(
|
||||
authed: ApiAuthed,
|
||||
Extension(user_db): Extension<UserDB>,
|
||||
Path((w_id, path)): Path<(String, StripPath)>,
|
||||
) -> JsonResult<AssetTriggerRow> {
|
||||
let path = path.to_path();
|
||||
let mut tx = user_db.begin(&authed).await?;
|
||||
let row = sqlx::query_as::<_, AssetTriggerRow>(
|
||||
"SELECT workspace_id, path, script_path, is_flow, owner_script_path, \
|
||||
owner_script_hash, is_implicit, on_event, subscription_set, fires, \
|
||||
debounce_s, partition_map, cancel_on_new, backlog, error, edited_by, \
|
||||
edited_at, email \
|
||||
FROM asset_trigger \
|
||||
WHERE workspace_id = $1 AND path = $2",
|
||||
)
|
||||
.bind(&w_id)
|
||||
.bind(path)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?
|
||||
.ok_or_else(|| Error::NotFound(format!("asset_trigger {path} not found")))?;
|
||||
tx.commit().await?;
|
||||
Ok(Json(row))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RejectBody {
|
||||
#[allow(dead_code)]
|
||||
path: Option<String>,
|
||||
}
|
||||
|
||||
async fn reject_mutate(
|
||||
_authed: ApiAuthed,
|
||||
Extension(_user_db): Extension<UserDB>,
|
||||
Path(_w_id): Path<String>,
|
||||
Json(_body): Json<RejectBody>,
|
||||
) -> Result<()> {
|
||||
Err(implicit_mutation_error(None))
|
||||
}
|
||||
|
||||
async fn reject_mutate_path(
|
||||
authed: ApiAuthed,
|
||||
Extension(user_db): Extension<UserDB>,
|
||||
Path((w_id, path)): Path<(String, StripPath)>,
|
||||
) -> Result<()> {
|
||||
let path = path.to_path();
|
||||
let mut tx = user_db.begin(&authed).await?;
|
||||
let owner = sqlx::query_scalar!(
|
||||
"SELECT owner_script_path FROM asset_trigger \
|
||||
WHERE workspace_id = $1 AND path = $2 AND is_implicit = true",
|
||||
&w_id,
|
||||
path,
|
||||
)
|
||||
.fetch_optional(&mut *tx)
|
||||
.await?
|
||||
.flatten();
|
||||
tx.commit().await?;
|
||||
Err(implicit_mutation_error(owner))
|
||||
}
|
||||
|
||||
fn implicit_mutation_error(owner: Option<String>) -> Error {
|
||||
let msg = match owner {
|
||||
Some(script_path) => format!(
|
||||
"Implicit asset trigger — edit `#trigger: asset` annotations in script {script_path}."
|
||||
),
|
||||
None => {
|
||||
"Asset triggers are implicit-only in v1. Edit `#trigger: asset` annotations in the \
|
||||
owning script to modify them."
|
||||
.to_string()
|
||||
}
|
||||
};
|
||||
Error::Generic(StatusCode::CONFLICT, msg)
|
||||
}
|
||||
@@ -5681,6 +5681,8 @@ fn register_potential_assets_on_inline_execution(
|
||||
job_id,
|
||||
workspace_id: w_id.to_string(),
|
||||
created_at: None,
|
||||
partition_key: None,
|
||||
script_hash: None,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -68,6 +68,7 @@ use windmill_common::error::AppError;
|
||||
mod ai;
|
||||
mod apps;
|
||||
pub mod args;
|
||||
mod asset_triggers;
|
||||
mod audit;
|
||||
pub mod auth;
|
||||
#[cfg(all(feature = "private", feature = "parquet"))]
|
||||
@@ -546,6 +547,7 @@ pub async fn run_server(
|
||||
.nest("/acls", granular_acls::workspaced_service())
|
||||
.nest("/apps", apps::workspaced_service(request_size_limit * 5))
|
||||
.nest("/assets", windmill_api_assets::workspaced_service())
|
||||
.nest("/asset_triggers", asset_triggers::workspaced_service())
|
||||
.nest("/audit", audit::workspaced_service())
|
||||
.nest("/capture", capture::workspaced_service())
|
||||
.nest(
|
||||
|
||||
@@ -0,0 +1,245 @@
|
||||
//! Implicit asset triggers: projected from `#trigger: asset` annotations
|
||||
//! in script source onto rows in `asset_trigger` with `is_implicit = true`.
|
||||
//!
|
||||
//! Every script save re-parses annotations and calls
|
||||
//! [`sync_implicit_asset_triggers_for_script`], which computes the
|
||||
//! subscription set from the script's inferred read assets and upserts one
|
||||
//! row per asset annotation (v1 allows at most one). Implicit rows are never
|
||||
//! edited via API or CLI — the trigger stays in perfect sync with the script.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sqlx::{types::Json, Postgres, Transaction};
|
||||
|
||||
use windmill_types::assets::{AssetKind, AssetUsageAccessType, AssetWithAltAccessType};
|
||||
|
||||
use crate::{
|
||||
error::Error,
|
||||
scripts::ScriptLang,
|
||||
trigger_annotations::{
|
||||
parse_trigger_annotations, AssetTriggerAnnotation, TriggerAnnotation,
|
||||
TriggerAnnotationError,
|
||||
},
|
||||
};
|
||||
|
||||
/// Deterministic path used by implicit asset triggers. Keeps the original
|
||||
/// script path prefix so row-level-security policies on `asset_trigger`
|
||||
/// work unchanged (`SPLIT_PART(path, '/', 1)` — 'u' / 'f' / 'g').
|
||||
pub const IMPLICIT_TRIGGER_PATH_SUFFIX: &str = "/__asset_trigger__";
|
||||
|
||||
pub fn implicit_trigger_path(script_path: &str) -> String {
|
||||
format!("{script_path}{IMPLICIT_TRIGGER_PATH_SUFFIX}")
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct SubscriptionEntry {
|
||||
pub kind: AssetKind,
|
||||
pub path: String,
|
||||
}
|
||||
|
||||
/// JSONB shape stored in `asset_trigger.subscription_set`. Uses an object
|
||||
/// wrapper with a `paths` array so stage 8 (partitioning) can add sibling
|
||||
/// fields without a migration. Indexed with GIN — containment queries like
|
||||
/// `subscription_set -> 'paths' @> ...` remain indexable.
|
||||
#[derive(Debug, Clone, Serialize, Deserialize)]
|
||||
pub struct SubscriptionSet {
|
||||
pub paths: Vec<SubscriptionEntry>,
|
||||
}
|
||||
|
||||
#[derive(thiserror::Error, Debug)]
|
||||
pub enum SyncError {
|
||||
#[error("trigger annotation: {0}")]
|
||||
Annotation(#[from] TriggerAnnotationError),
|
||||
#[error("sql: {0}")]
|
||||
Sql(#[from] sqlx::Error),
|
||||
}
|
||||
|
||||
impl From<SyncError> for Error {
|
||||
fn from(e: SyncError) -> Self {
|
||||
Error::BadRequest(e.to_string())
|
||||
}
|
||||
}
|
||||
|
||||
/// Re-sync implicit asset triggers for a script. Called inside the script-
|
||||
/// save transaction after the script row has been inserted.
|
||||
///
|
||||
/// `inferred_read_assets` is the static asset-usage list captured at save
|
||||
/// time (`ns.assets` on the save handler). Only entries with `r` or `rw`
|
||||
/// access contribute to the subscription set — reactive triggers subscribe
|
||||
/// to *upstreams*, not the script's own writes.
|
||||
pub async fn sync_implicit_asset_triggers_for_script(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
workspace_id: &str,
|
||||
script_path: &str,
|
||||
script_hash: i64,
|
||||
language: ScriptLang,
|
||||
content: &str,
|
||||
inferred_read_assets: &[AssetWithAltAccessType],
|
||||
author_email: &str,
|
||||
author_username: &str,
|
||||
) -> std::result::Result<(), SyncError> {
|
||||
let anns = parse_trigger_annotations(content, language)?;
|
||||
let asset_annotations: Vec<&AssetTriggerAnnotation> = anns
|
||||
.iter()
|
||||
.map(|a| match a {
|
||||
TriggerAnnotation::Asset(inner) => inner,
|
||||
})
|
||||
.collect();
|
||||
|
||||
// Delete all existing implicit rows for this owner — we'll re-insert the
|
||||
// current ones below. Safer than a diff since v1 is single-annotation.
|
||||
sqlx::query!(
|
||||
"DELETE FROM asset_trigger \
|
||||
WHERE workspace_id = $1 AND owner_script_path = $2 AND is_implicit = true",
|
||||
workspace_id,
|
||||
script_path,
|
||||
)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
|
||||
for ann in asset_annotations {
|
||||
let paths = resolve_subscription_set(ann, inferred_read_assets);
|
||||
let subscription_set = SubscriptionSet { paths };
|
||||
let trigger_path = implicit_trigger_path(script_path);
|
||||
let on_event = match ann.on {
|
||||
crate::trigger_annotations::AssetTriggerEvent::Change => "change",
|
||||
};
|
||||
let fires = match ann.fires {
|
||||
crate::trigger_annotations::FiresMode::All => "all",
|
||||
crate::trigger_annotations::FiresMode::Any => "any",
|
||||
};
|
||||
let backlog = match ann.backlog {
|
||||
crate::trigger_annotations::BacklogMode::Coalesce => "coalesce",
|
||||
crate::trigger_annotations::BacklogMode::Replay => "replay",
|
||||
crate::trigger_annotations::BacklogMode::Skip => "skip",
|
||||
};
|
||||
sqlx::query!(
|
||||
"INSERT INTO asset_trigger ( \
|
||||
workspace_id, path, script_path, is_flow, \
|
||||
owner_script_path, owner_script_hash, is_implicit, \
|
||||
on_event, subscription_set, fires, debounce_s, partition_map, \
|
||||
cancel_on_new, backlog, \
|
||||
edited_by, email \
|
||||
) VALUES ( \
|
||||
$1, $2, $3, false, \
|
||||
$3, $4, true, \
|
||||
$5, $6, $7, $8, $9, \
|
||||
$10, $11, \
|
||||
$12, $13 \
|
||||
)",
|
||||
workspace_id,
|
||||
trigger_path,
|
||||
script_path,
|
||||
script_hash,
|
||||
on_event,
|
||||
Json(&subscription_set) as _,
|
||||
fires,
|
||||
ann.debounce_s,
|
||||
Json(&ann.partition_map) as _,
|
||||
ann.cancel_on_new,
|
||||
backlog,
|
||||
author_username,
|
||||
author_email,
|
||||
)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
}
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Resolve the reactive subscription set from the annotation and the
|
||||
/// script's inferred reads. `only` overrides the inferred reads entirely;
|
||||
/// otherwise we take inferred reads minus `exclude` plus `extra`, filtered
|
||||
/// by `kinds` if specified. Writes are ignored.
|
||||
fn resolve_subscription_set(
|
||||
ann: &AssetTriggerAnnotation,
|
||||
inferred_reads: &[AssetWithAltAccessType],
|
||||
) -> Vec<SubscriptionEntry> {
|
||||
let read_paths: Vec<SubscriptionEntry> = if let Some(only) = &ann.only {
|
||||
// `only` specifies bare paths without kind — default to Resource.
|
||||
// Kinds filter (below) still applies, so `kinds=[s3object]` with
|
||||
// `only=[f/foo]` yields an s3object subscription.
|
||||
let kind = ann
|
||||
.kinds
|
||||
.as_ref()
|
||||
.and_then(|ks| ks.first().copied())
|
||||
.unwrap_or(AssetKind::Resource);
|
||||
only.iter()
|
||||
.map(|p| SubscriptionEntry { kind, path: p.clone() })
|
||||
.collect()
|
||||
} else {
|
||||
inferred_reads
|
||||
.iter()
|
||||
.filter(|a| {
|
||||
matches!(
|
||||
a.access_type.or(a.alt_access_type),
|
||||
Some(AssetUsageAccessType::R) | Some(AssetUsageAccessType::RW)
|
||||
)
|
||||
})
|
||||
.filter(|a| !ann.exclude.iter().any(|e| e == &a.path))
|
||||
.map(|a| SubscriptionEntry { kind: a.kind, path: a.path.clone() })
|
||||
.collect()
|
||||
};
|
||||
|
||||
let extras = ann.extra.iter().map(|p| SubscriptionEntry {
|
||||
kind: ann
|
||||
.kinds
|
||||
.as_ref()
|
||||
.and_then(|ks| ks.first().copied())
|
||||
.unwrap_or(AssetKind::Resource),
|
||||
path: p.clone(),
|
||||
});
|
||||
|
||||
let mut all: Vec<SubscriptionEntry> = read_paths.into_iter().chain(extras).collect();
|
||||
|
||||
if let Some(kinds) = &ann.kinds {
|
||||
all.retain(|e| kinds.contains(&e.kind));
|
||||
}
|
||||
|
||||
// Dedup (preserve first-seen order).
|
||||
let mut seen = std::collections::HashSet::new();
|
||||
all.retain(|e| seen.insert((e.kind, e.path.clone())));
|
||||
all
|
||||
}
|
||||
|
||||
/// Delete all implicit asset triggers owned by a script. Called when a
|
||||
/// script is archived or deleted.
|
||||
pub async fn delete_implicit_asset_triggers_for_script(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
workspace_id: &str,
|
||||
script_path: &str,
|
||||
) -> std::result::Result<(), sqlx::Error> {
|
||||
sqlx::query!(
|
||||
"DELETE FROM asset_trigger \
|
||||
WHERE workspace_id = $1 AND owner_script_path = $2 AND is_implicit = true",
|
||||
workspace_id,
|
||||
script_path,
|
||||
)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Rename implicit triggers when the owning script is renamed. Keeps the
|
||||
/// trigger path aligned with the new script path so the deterministic
|
||||
/// addressing remains stable.
|
||||
pub async fn rename_implicit_asset_triggers_for_script(
|
||||
tx: &mut Transaction<'_, Postgres>,
|
||||
workspace_id: &str,
|
||||
old_script_path: &str,
|
||||
new_script_path: &str,
|
||||
) -> std::result::Result<(), sqlx::Error> {
|
||||
let new_trigger_path = implicit_trigger_path(new_script_path);
|
||||
sqlx::query!(
|
||||
"UPDATE asset_trigger \
|
||||
SET owner_script_path = $1, script_path = $1, path = $2, server_id = NULL \
|
||||
WHERE workspace_id = $3 AND owner_script_path = $4 AND is_implicit = true",
|
||||
new_script_path,
|
||||
new_trigger_path,
|
||||
workspace_id,
|
||||
old_script_path,
|
||||
)
|
||||
.execute(&mut **tx)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
@@ -63,6 +63,7 @@ pub mod flow_status;
|
||||
pub mod flows;
|
||||
pub mod folders;
|
||||
pub mod global_settings;
|
||||
pub mod implicit_triggers;
|
||||
pub mod indexer;
|
||||
pub mod instance_config;
|
||||
pub mod job_metrics;
|
||||
@@ -70,6 +71,7 @@ pub mod log_context;
|
||||
pub mod min_version;
|
||||
pub mod notify_events;
|
||||
pub mod runtime_assets;
|
||||
pub mod trigger_annotations;
|
||||
pub mod workspace_dependencies;
|
||||
|
||||
#[cfg(feature = "private")]
|
||||
|
||||
@@ -69,6 +69,9 @@ pub struct InsertRuntimeAssetParams {
|
||||
pub access_type: Option<AssetUsageAccessType>,
|
||||
pub created_at: Option<chrono::DateTime<chrono::Utc>>,
|
||||
pub columns: Option<BTreeMap<String, AssetUsageAccessType>>,
|
||||
// Forward-compatibility fields for asset-change triggers (v1 unused: partitioning lands in stage 8).
|
||||
pub partition_key: Option<String>,
|
||||
pub script_hash: Option<i64>,
|
||||
}
|
||||
|
||||
async fn insert_runtime_assets(
|
||||
@@ -90,6 +93,57 @@ async fn insert_runtime_assets(
|
||||
query_builder.push(" ON CONFLICT DO NOTHING");
|
||||
query_builder.build().execute(executor).await?;
|
||||
}
|
||||
insert_asset_events(executor, assets).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Append rows to `asset_event` for every write/read-write asset access,
|
||||
/// and pg_notify on channel `asset_event` so the asset-trigger dispatcher
|
||||
/// can react in near-real-time. No-op for read-only accesses.
|
||||
async fn insert_asset_events(
|
||||
executor: &Pool<Postgres>,
|
||||
assets: &[InsertRuntimeAssetParams],
|
||||
) -> error::Result<()> {
|
||||
let write_assets: Vec<&InsertRuntimeAssetParams> = assets
|
||||
.iter()
|
||||
.filter(|a| {
|
||||
matches!(
|
||||
a.access_type,
|
||||
Some(AssetUsageAccessType::W) | Some(AssetUsageAccessType::RW)
|
||||
)
|
||||
})
|
||||
.collect();
|
||||
if write_assets.is_empty() {
|
||||
return Ok(());
|
||||
}
|
||||
for chunk in write_assets.chunks(1000) {
|
||||
let mut query_builder = QueryBuilder::new(
|
||||
"WITH inserted AS (INSERT INTO asset_event \
|
||||
(workspace_id, asset_kind, asset_path, partition_key, job_id, script_hash, access_type, columns, at) ",
|
||||
);
|
||||
query_builder.push_values(chunk, |mut b, asset| {
|
||||
b.push_bind(&asset.workspace_id)
|
||||
.push_bind(&asset.asset_kind)
|
||||
.push_bind(&asset.asset_path)
|
||||
.push_bind(&asset.partition_key)
|
||||
.push_bind(asset.job_id)
|
||||
.push_bind(&asset.script_hash)
|
||||
.push_bind(asset.access_type.expect("filtered above"))
|
||||
.push_bind(Json(&asset.columns))
|
||||
.push_bind(asset.created_at.unwrap_or_else(chrono::Utc::now));
|
||||
});
|
||||
query_builder.push(
|
||||
" RETURNING id, workspace_id, asset_kind::text AS asset_kind, asset_path, partition_key) \
|
||||
SELECT pg_notify('asset_event', json_build_object( \
|
||||
'id', id, \
|
||||
'workspace_id', workspace_id, \
|
||||
'asset_kind', asset_kind, \
|
||||
'asset_path', asset_path, \
|
||||
'partition_key', partition_key \
|
||||
)::text) FROM inserted",
|
||||
);
|
||||
query_builder.build().execute(executor).await?;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
@@ -0,0 +1,489 @@
|
||||
//! Parse `#trigger: asset ...` annotations out of script source.
|
||||
//!
|
||||
//! This is the source-of-truth for implicit asset triggers. A script's
|
||||
//! annotations are re-parsed on every save and projected into rows in the
|
||||
//! `asset_trigger` table.
|
||||
|
||||
use std::str::FromStr;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::{assets::AssetKind, scripts::ScriptLang};
|
||||
|
||||
#[derive(thiserror::Error, Debug, PartialEq, Eq)]
|
||||
pub enum TriggerAnnotationError {
|
||||
#[error("trigger annotation is not valid: {0}")]
|
||||
Invalid(String),
|
||||
#[error("trigger option `{key}` has unsupported value `{value}`: {reason}")]
|
||||
UnsupportedValue { key: String, value: String, reason: String },
|
||||
#[error("unknown trigger option `{0}`")]
|
||||
UnknownOption(String),
|
||||
#[error("script has more than one `trigger: asset` annotation — v1 supports a single asset trigger per script")]
|
||||
MultipleAssetAnnotations,
|
||||
}
|
||||
|
||||
pub type Result<T> = std::result::Result<T, TriggerAnnotationError>;
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum AssetTriggerEvent {
|
||||
Change,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum FiresMode {
|
||||
All,
|
||||
Any,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "lowercase")]
|
||||
pub enum BacklogMode {
|
||||
Coalesce,
|
||||
Replay,
|
||||
Skip,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(tag = "mode", rename_all = "lowercase")]
|
||||
pub enum PartitionMap {
|
||||
Identity,
|
||||
None,
|
||||
All,
|
||||
Window { back_days: i32, forward_days: i32 },
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct AssetTriggerAnnotation {
|
||||
pub on: AssetTriggerEvent,
|
||||
pub fires: FiresMode,
|
||||
pub debounce_s: i32,
|
||||
/// If `Some`, only these upstream paths (filtered by kinds) are subscribed
|
||||
/// to; overrides the script's inferred read set.
|
||||
pub only: Option<Vec<String>>,
|
||||
pub exclude: Vec<String>,
|
||||
pub extra: Vec<String>,
|
||||
/// If `Some`, the subscription set is narrowed to these asset kinds.
|
||||
pub kinds: Option<Vec<AssetKind>>,
|
||||
pub partition_map: PartitionMap,
|
||||
pub cancel_on_new: bool,
|
||||
pub backlog: BacklogMode,
|
||||
}
|
||||
|
||||
impl Default for AssetTriggerAnnotation {
|
||||
fn default() -> Self {
|
||||
Self {
|
||||
on: AssetTriggerEvent::Change,
|
||||
fires: FiresMode::All,
|
||||
debounce_s: 30,
|
||||
only: None,
|
||||
exclude: Vec::new(),
|
||||
extra: Vec::new(),
|
||||
kinds: None,
|
||||
partition_map: PartitionMap::Identity,
|
||||
cancel_on_new: false,
|
||||
backlog: BacklogMode::Coalesce,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub enum TriggerAnnotation {
|
||||
Asset(AssetTriggerAnnotation),
|
||||
}
|
||||
|
||||
/// Parse every `<comment> trigger: ...` annotation found anywhere in the file.
|
||||
///
|
||||
/// Returns `Ok(vec![])` when no trigger annotation is present — this is the
|
||||
/// common case. Errors are only returned for malformed annotations so a
|
||||
/// missing trigger never blocks a save.
|
||||
pub fn parse_trigger_annotations(code: &str, lang: ScriptLang) -> Result<Vec<TriggerAnnotation>> {
|
||||
let prefix = lang.as_comment_lit();
|
||||
let marker = format!("{prefix} trigger:");
|
||||
let mut out = Vec::new();
|
||||
let mut asset_seen = false;
|
||||
for raw in code.lines() {
|
||||
let line = raw.trim_end();
|
||||
let trimmed = line.trim_start();
|
||||
let Some(rest) = trimmed.strip_prefix(&marker) else {
|
||||
continue;
|
||||
};
|
||||
let rest = rest.trim();
|
||||
if let Some(body) = rest.strip_prefix("asset").map(|r| r.trim()) {
|
||||
if asset_seen {
|
||||
return Err(TriggerAnnotationError::MultipleAssetAnnotations);
|
||||
}
|
||||
asset_seen = true;
|
||||
out.push(TriggerAnnotation::Asset(parse_asset_body(body)?));
|
||||
} else {
|
||||
return Err(TriggerAnnotationError::Invalid(format!(
|
||||
"expected `trigger: asset ...`, got `trigger: {rest}`"
|
||||
)));
|
||||
}
|
||||
}
|
||||
Ok(out)
|
||||
}
|
||||
|
||||
fn parse_asset_body(body: &str) -> Result<AssetTriggerAnnotation> {
|
||||
let mut ann = AssetTriggerAnnotation::default();
|
||||
for token in split_top_level(body)? {
|
||||
let Some((key, value)) = token.split_once('=') else {
|
||||
return Err(TriggerAnnotationError::Invalid(format!(
|
||||
"expected `key=value`, got `{token}`"
|
||||
)));
|
||||
};
|
||||
let key = key.trim();
|
||||
let value = value.trim();
|
||||
match key {
|
||||
"on" => {
|
||||
if value != "change" {
|
||||
return Err(TriggerAnnotationError::UnsupportedValue {
|
||||
key: key.to_string(),
|
||||
value: value.to_string(),
|
||||
reason: "only `on=change` is supported in v1".to_string(),
|
||||
});
|
||||
}
|
||||
ann.on = AssetTriggerEvent::Change;
|
||||
}
|
||||
"fires" => {
|
||||
ann.fires = match value {
|
||||
"all" => FiresMode::All,
|
||||
"any" => FiresMode::Any,
|
||||
other => {
|
||||
return Err(TriggerAnnotationError::UnsupportedValue {
|
||||
key: key.to_string(),
|
||||
value: other.to_string(),
|
||||
reason: "expected `all` or `any`".to_string(),
|
||||
});
|
||||
}
|
||||
};
|
||||
}
|
||||
"debounce" => {
|
||||
ann.debounce_s = parse_duration_seconds(value).map_err(|e| {
|
||||
TriggerAnnotationError::UnsupportedValue {
|
||||
key: key.to_string(),
|
||||
value: value.to_string(),
|
||||
reason: e,
|
||||
}
|
||||
})?;
|
||||
}
|
||||
"only" => ann.only = Some(parse_list(value)),
|
||||
"exclude" => ann.exclude = parse_list(value),
|
||||
"extra" => ann.extra = parse_list(value),
|
||||
"kinds" => {
|
||||
let list = parse_list(value);
|
||||
let mut kinds = Vec::with_capacity(list.len());
|
||||
for k in list {
|
||||
kinds.push(parse_asset_kind(&k).ok_or_else(|| {
|
||||
TriggerAnnotationError::UnsupportedValue {
|
||||
key: "kinds".to_string(),
|
||||
value: k.clone(),
|
||||
reason: "expected one of datatable|ducklake|s3object|resource|volume"
|
||||
.to_string(),
|
||||
}
|
||||
})?);
|
||||
}
|
||||
ann.kinds = Some(kinds);
|
||||
}
|
||||
"partition_map" => {
|
||||
ann.partition_map = parse_partition_map(value).map_err(|reason| {
|
||||
TriggerAnnotationError::UnsupportedValue {
|
||||
key: key.to_string(),
|
||||
value: value.to_string(),
|
||||
reason,
|
||||
}
|
||||
})?;
|
||||
}
|
||||
"cancel_on_new" => {
|
||||
ann.cancel_on_new =
|
||||
parse_bool(value).ok_or_else(|| TriggerAnnotationError::UnsupportedValue {
|
||||
key: key.to_string(),
|
||||
value: value.to_string(),
|
||||
reason: "expected `true` or `false`".to_string(),
|
||||
})?;
|
||||
}
|
||||
"backlog" => {
|
||||
ann.backlog = match value {
|
||||
"coalesce" => BacklogMode::Coalesce,
|
||||
"replay" => BacklogMode::Replay,
|
||||
"skip" => BacklogMode::Skip,
|
||||
other => {
|
||||
return Err(TriggerAnnotationError::UnsupportedValue {
|
||||
key: key.to_string(),
|
||||
value: other.to_string(),
|
||||
reason: "expected `coalesce`, `replay`, or `skip`".to_string(),
|
||||
});
|
||||
}
|
||||
};
|
||||
}
|
||||
other => return Err(TriggerAnnotationError::UnknownOption(other.to_string())),
|
||||
}
|
||||
}
|
||||
Ok(ann)
|
||||
}
|
||||
|
||||
/// Split a body like `on=change fires=all only=[a,b] debounce=30s` into
|
||||
/// tokens, preserving bracketed list values as a single token.
|
||||
fn split_top_level(body: &str) -> Result<Vec<String>> {
|
||||
let mut tokens = Vec::new();
|
||||
let mut current = String::new();
|
||||
let mut depth: i32 = 0;
|
||||
for ch in body.chars() {
|
||||
match ch {
|
||||
'[' => {
|
||||
depth += 1;
|
||||
current.push(ch);
|
||||
}
|
||||
']' => {
|
||||
depth -= 1;
|
||||
if depth < 0 {
|
||||
return Err(TriggerAnnotationError::Invalid(
|
||||
"unbalanced `]` in trigger annotation".to_string(),
|
||||
));
|
||||
}
|
||||
current.push(ch);
|
||||
}
|
||||
c if c.is_whitespace() && depth == 0 => {
|
||||
if !current.is_empty() {
|
||||
tokens.push(std::mem::take(&mut current));
|
||||
}
|
||||
}
|
||||
_ => current.push(ch),
|
||||
}
|
||||
}
|
||||
if depth != 0 {
|
||||
return Err(TriggerAnnotationError::Invalid(
|
||||
"unbalanced `[` in trigger annotation".to_string(),
|
||||
));
|
||||
}
|
||||
if !current.is_empty() {
|
||||
tokens.push(current);
|
||||
}
|
||||
Ok(tokens)
|
||||
}
|
||||
|
||||
fn parse_list(value: &str) -> Vec<String> {
|
||||
let trimmed = value.trim();
|
||||
let inner = trimmed
|
||||
.strip_prefix('[')
|
||||
.and_then(|s| s.strip_suffix(']'))
|
||||
.unwrap_or(trimmed);
|
||||
inner
|
||||
.split(',')
|
||||
.map(str::trim)
|
||||
.filter(|s| !s.is_empty())
|
||||
.map(str::to_string)
|
||||
.collect()
|
||||
}
|
||||
|
||||
fn parse_bool(value: &str) -> Option<bool> {
|
||||
match value {
|
||||
"true" => Some(true),
|
||||
"false" => Some(false),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_asset_kind(s: &str) -> Option<AssetKind> {
|
||||
match s {
|
||||
"s3object" => Some(AssetKind::S3Object),
|
||||
"resource" => Some(AssetKind::Resource),
|
||||
"variable" => Some(AssetKind::Variable),
|
||||
"ducklake" => Some(AssetKind::Ducklake),
|
||||
"datatable" => Some(AssetKind::DataTable),
|
||||
"volume" => Some(AssetKind::Volume),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse durations like `30s`, `5m`, `2h`, `1d`. Returns whole seconds.
|
||||
/// Bare integers are accepted as seconds for backward-compatibility with
|
||||
/// simple `debounce=30`.
|
||||
fn parse_duration_seconds(s: &str) -> std::result::Result<i32, String> {
|
||||
let s = s.trim();
|
||||
if s.is_empty() {
|
||||
return Err("empty duration".to_string());
|
||||
}
|
||||
let (num, unit) = match s
|
||||
.char_indices()
|
||||
.find(|(_, c)| !c.is_ascii_digit() && *c != '-')
|
||||
{
|
||||
Some((i, _)) => (&s[..i], &s[i..]),
|
||||
None => (s, "s"),
|
||||
};
|
||||
let n: i32 = i32::from_str(num).map_err(|_| format!("not a number: `{num}`"))?;
|
||||
let mult = match unit {
|
||||
"s" | "" => 1,
|
||||
"m" => 60,
|
||||
"h" => 60 * 60,
|
||||
"d" => 24 * 60 * 60,
|
||||
other => return Err(format!("unknown duration unit `{other}`")),
|
||||
};
|
||||
Ok(n * mult)
|
||||
}
|
||||
|
||||
fn parse_partition_map(value: &str) -> std::result::Result<PartitionMap, String> {
|
||||
match value {
|
||||
"identity" => Ok(PartitionMap::Identity),
|
||||
"none" => Ok(PartitionMap::None),
|
||||
"all" => Ok(PartitionMap::All),
|
||||
other if other.starts_with("window(") && other.ends_with(')') => {
|
||||
let inside = &other["window(".len()..other.len() - 1];
|
||||
let mut parts = inside.split(',').map(str::trim);
|
||||
let back = parts
|
||||
.next()
|
||||
.ok_or_else(|| "window() needs two arguments".to_string())?;
|
||||
let forward = parts
|
||||
.next()
|
||||
.ok_or_else(|| "window() needs two arguments".to_string())?;
|
||||
if parts.next().is_some() {
|
||||
return Err("window() takes exactly two arguments".to_string());
|
||||
}
|
||||
let back_days = parse_signed_days(back)?;
|
||||
let forward_days = parse_signed_days(forward)?;
|
||||
Ok(PartitionMap::Window { back_days, forward_days })
|
||||
}
|
||||
other => Err(format!(
|
||||
"expected `identity`, `none`, `all`, or `window(-Nd,Md)`, got `{other}`"
|
||||
)),
|
||||
}
|
||||
}
|
||||
|
||||
fn parse_signed_days(s: &str) -> std::result::Result<i32, String> {
|
||||
let s = s.trim();
|
||||
let rest = s
|
||||
.strip_suffix('d')
|
||||
.ok_or_else(|| format!("expected a signed day literal like `-7d` or `3d`, got `{s}`"))?;
|
||||
i32::from_str(rest).map_err(|_| format!("not a number: `{rest}`"))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn parses_none_when_no_annotation() {
|
||||
let out =
|
||||
parse_trigger_annotations("let x = 1;\n// just a comment", ScriptLang::Bun).unwrap();
|
||||
assert!(out.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_minimal_python_annotation() {
|
||||
let src = "# trigger: asset on=change\nprint('hi')";
|
||||
let out = parse_trigger_annotations(src, ScriptLang::Python3).unwrap();
|
||||
let TriggerAnnotation::Asset(ann) = &out[0];
|
||||
assert_eq!(ann.on, AssetTriggerEvent::Change);
|
||||
assert_eq!(ann.fires, FiresMode::All);
|
||||
assert_eq!(ann.debounce_s, 30);
|
||||
assert_eq!(ann.cancel_on_new, false);
|
||||
assert_eq!(ann.backlog, BacklogMode::Coalesce);
|
||||
assert_eq!(ann.partition_map, PartitionMap::Identity);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_bun_annotation_anywhere_in_file() {
|
||||
let src = "import x from 'y';\n\n// trigger: asset on=change fires=any\nexport async function main() {}";
|
||||
let out = parse_trigger_annotations(src, ScriptLang::Bun).unwrap();
|
||||
assert_eq!(out.len(), 1);
|
||||
let TriggerAnnotation::Asset(ann) = &out[0];
|
||||
assert_eq!(ann.fires, FiresMode::Any);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_sql_annotation() {
|
||||
let src = "-- trigger: asset on=change debounce=5m\nSELECT 1;";
|
||||
let out = parse_trigger_annotations(src, ScriptLang::Postgresql).unwrap();
|
||||
let TriggerAnnotation::Asset(ann) = &out[0];
|
||||
assert_eq!(ann.debounce_s, 300);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_rust_annotation_with_its_bespoke_prefix() {
|
||||
let src = "//! trigger: asset on=change backlog=skip\nfn main() {}";
|
||||
let out = parse_trigger_annotations(src, ScriptLang::Rust).unwrap();
|
||||
let TriggerAnnotation::Asset(ann) = &out[0];
|
||||
assert_eq!(ann.backlog, BacklogMode::Skip);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn parses_full_option_set() {
|
||||
let src = "# trigger: asset on=change fires=any debounce=2h only=[f/a,f/b] exclude=[f/c] extra=[f/d] kinds=[datatable,s3object] partition_map=window(-7d,0d) cancel_on_new=true backlog=replay\n";
|
||||
let out = parse_trigger_annotations(src, ScriptLang::Python3).unwrap();
|
||||
let TriggerAnnotation::Asset(ann) = &out[0];
|
||||
assert_eq!(ann.fires, FiresMode::Any);
|
||||
assert_eq!(ann.debounce_s, 2 * 60 * 60);
|
||||
assert_eq!(ann.only, Some(vec!["f/a".to_string(), "f/b".to_string()]));
|
||||
assert_eq!(ann.exclude, vec!["f/c".to_string()]);
|
||||
assert_eq!(ann.extra, vec!["f/d".to_string()]);
|
||||
assert_eq!(
|
||||
ann.kinds,
|
||||
Some(vec![AssetKind::DataTable, AssetKind::S3Object])
|
||||
);
|
||||
assert_eq!(
|
||||
ann.partition_map,
|
||||
PartitionMap::Window { back_days: -7, forward_days: 0 }
|
||||
);
|
||||
assert_eq!(ann.cancel_on_new, true);
|
||||
assert_eq!(ann.backlog, BacklogMode::Replay);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_unsupported_on_value() {
|
||||
let src = "# trigger: asset on=freshness\n";
|
||||
let err = parse_trigger_annotations(src, ScriptLang::Python3).unwrap_err();
|
||||
assert!(matches!(
|
||||
err,
|
||||
TriggerAnnotationError::UnsupportedValue { .. }
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_duplicate_asset_annotation() {
|
||||
let src = "# trigger: asset on=change\n# trigger: asset on=change\n";
|
||||
let err = parse_trigger_annotations(src, ScriptLang::Python3).unwrap_err();
|
||||
assert_eq!(err, TriggerAnnotationError::MultipleAssetAnnotations);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_unknown_option() {
|
||||
let src = "# trigger: asset on=change zombie=true\n";
|
||||
let err = parse_trigger_annotations(src, ScriptLang::Python3).unwrap_err();
|
||||
assert!(matches!(err, TriggerAnnotationError::UnknownOption(o) if o == "zombie"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_malformed_kind() {
|
||||
let src = "# trigger: asset on=change kinds=[s3object,foobar]\n";
|
||||
let err = parse_trigger_annotations(src, ScriptLang::Python3).unwrap_err();
|
||||
assert!(
|
||||
matches!(err, TriggerAnnotationError::UnsupportedValue { key, .. } if key == "kinds")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rejects_bad_partition_window() {
|
||||
let src = "# trigger: asset on=change partition_map=window(abc,def)\n";
|
||||
let err = parse_trigger_annotations(src, ScriptLang::Python3).unwrap_err();
|
||||
assert!(
|
||||
matches!(err, TriggerAnnotationError::UnsupportedValue { key, .. } if key == "partition_map")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn accepts_bare_integer_debounce_as_seconds() {
|
||||
let src = "# trigger: asset on=change debounce=45\n";
|
||||
let out = parse_trigger_annotations(src, ScriptLang::Python3).unwrap();
|
||||
let TriggerAnnotation::Asset(ann) = &out[0];
|
||||
assert_eq!(ann.debounce_s, 45);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn ignores_lines_with_other_comment_prefix() {
|
||||
// `# trigger: ...` in a bun file uses `//` as prefix, so a `#` line is ignored.
|
||||
let src = "# trigger: asset on=change\nconsole.log('hi');";
|
||||
let out = parse_trigger_annotations(src, ScriptLang::Bun).unwrap();
|
||||
assert!(out.is_empty());
|
||||
}
|
||||
}
|
||||
@@ -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 asset_trigger SET script_path = $1, server_id = NULL WHERE script_path = $2 AND workspace_id = $3 AND is_flow = $4 AND is_implicit = false) \
|
||||
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,
|
||||
|
||||
@@ -3313,6 +3313,8 @@ async fn detect_and_store_runtime_assets_from_job_args(
|
||||
access_type: None,
|
||||
created_at: None,
|
||||
columns: None,
|
||||
partition_key: None,
|
||||
script_hash: None,
|
||||
};
|
||||
register_runtime_asset(asset);
|
||||
}
|
||||
|
||||
@@ -1457,6 +1457,11 @@ export async function elementsToMap(
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
// Asset triggers are implicit: source-of-truth is `#trigger: asset` in
|
||||
// script annotations, never editable via CLI. Skip unconditionally.
|
||||
if (path.endsWith(".asset_trigger" + ext)) {
|
||||
continue;
|
||||
}
|
||||
if (!skips.includeUsers && path.endsWith(".user" + ext)) continue;
|
||||
if (!skips.includeGroups && path.endsWith(".group" + ext)) continue;
|
||||
if (!skips.includeSettings && path === "settings" + ext) continue;
|
||||
|
||||
Reference in New Issue
Block a user