feat: add asset-trigger dispatcher primitives and save-time signature check

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

backend/windmill-api-scripts/src/scripts.rs

@@ -999,6 +999,22 @@ async fn create_script_internal<'c>(
         )
     };
 
+    // Stage 5 validation: reject asset-triggered scripts whose signature has
+    // required args outside the reserved dispatcher set, so the script can
+    // actually be run by a reactive fire.
+    {
+        let asset_annotations = windmill_common::trigger_annotations::parse_trigger_annotations(
+            &ns.content,
+            ns.language,
+        )
+        .map_err(|e| Error::BadRequest(e.to_string()))?;
+        if !asset_annotations.is_empty() {
+            let schema_raw = ns.schema.as_ref().map(|s| &**s.0.as_ref());
+            windmill_common::implicit_triggers::validate_asset_triggered_script_schema(schema_raw)
+                .map_err(Error::BadRequest)?;
+        }
+    }
+
     sqlx::query!(
         "INSERT INTO script (workspace_id, hash, path, parent_hashes, summary, description, \
          content, created_by, schema, is_template, extra_perms, lock, language, kind, tag, \

backend/windmill-common/src/asset_dispatcher.rs

@@ -0,0 +1,148 @@
+//! Reactive dispatcher for asset-change triggers (stage 4).
+//!
+//! Polls `asset_event` for new rows, matches them against
+//! `asset_trigger.subscription_set`, and enqueues downstream runs.
+//!
+//! This module intentionally exposes **stateless** building blocks (match +
+//! policy application + reserved-arg construction); the runtime wiring that
+//! turns them into a long-running task with server-ownership and job push
+//! lives in `windmill-worker` so that `windmill-common` can stay free of a
+//! `windmill-queue` dependency.
+
+use serde::{Deserialize, Serialize};
+use sqlx::{Pool, Postgres};
+
+use crate::{assets::AssetKind, error::Result};
+
+/// One pending event picked up from `asset_event`. The dispatcher hands
+/// matched rows plus the triggering event down to the enqueue path.
+#[derive(Debug, Clone, sqlx::FromRow)]
+pub struct PendingAssetEvent {
+    pub id: i64,
+    pub workspace_id: String,
+    pub asset_kind: AssetKind,
+    pub asset_path: String,
+    pub partition_key: Option<String>,
+    pub job_id: uuid::Uuid,
+    pub at: chrono::DateTime<chrono::Utc>,
+}
+
+/// Row shape returned by the matching query — just what the enqueue path
+/// needs, not the full trigger row.
+#[derive(Debug, Clone, sqlx::FromRow)]
+pub struct MatchedAssetTrigger {
+    pub workspace_id: String,
+    pub path: String,
+    pub script_path: String,
+    pub is_flow: bool,
+    pub fires: String,
+    pub debounce_s: i32,
+    pub cancel_on_new: bool,
+    pub owner_script_hash: Option<i64>,
+}
+
+/// Fetch events with id greater than `last_id`, ordered by id ascending.
+/// Bounded to a single batch so the dispatcher makes forward progress even
+/// under burst.
+pub async fn fetch_pending_events(
+    db: &Pool<Postgres>,
+    last_id: i64,
+    batch_size: i32,
+) -> Result<Vec<PendingAssetEvent>> {
+    let events = sqlx::query_as::<_, PendingAssetEvent>(
+        "SELECT id, workspace_id, asset_kind, asset_path, partition_key, job_id, at \
+         FROM asset_event \
+         WHERE id > $1 \
+         ORDER BY id ASC \
+         LIMIT $2",
+    )
+    .bind(last_id)
+    .bind(batch_size)
+    .fetch_all(db)
+    .await?;
+    Ok(events)
+}
+
+/// Select `asset_trigger` rows whose subscription set contains the
+/// `(kind, path)` of the given event. Uses JSONB containment so the GIN
+/// index on `subscription_set` kicks in.
+pub async fn match_triggers_for_event(
+    db: &Pool<Postgres>,
+    event: &PendingAssetEvent,
+) -> Result<Vec<MatchedAssetTrigger>> {
+    let subscription_probe = serde_json::json!({
+        "paths": [{ "kind": event.asset_kind, "path": event.asset_path }]
+    });
+    let rows = sqlx::query_as::<_, MatchedAssetTrigger>(
+        "SELECT workspace_id, path, script_path, is_flow, fires, debounce_s, \
+                cancel_on_new, owner_script_hash \
+         FROM asset_trigger \
+         WHERE workspace_id = $1 \
+           AND mode = 'enabled' \
+           AND subscription_set @> $2::jsonb",
+    )
+    .bind(&event.workspace_id)
+    .bind(&subscription_probe)
+    .fetch_all(db)
+    .await?;
+    Ok(rows)
+}
+
+/// Reserved argument payload passed to asset-triggered scripts. The worker
+/// fills the corresponding `wm_asset_event` / `wm_partition` / `wm_backfill`
+/// parameters at job materialization (stage 5).
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct AssetEventArg {
+    pub id: i64,
+    pub workspace_id: String,
+    pub asset_kind: AssetKind,
+    pub asset_path: String,
+    pub partition_key: Option<String>,
+    pub job_id: uuid::Uuid,
+    pub at: chrono::DateTime<chrono::Utc>,
+}
+
+impl From<&PendingAssetEvent> for AssetEventArg {
+    fn from(e: &PendingAssetEvent) -> Self {
+        Self {
+            id: e.id,
+            workspace_id: e.workspace_id.clone(),
+            asset_kind: e.asset_kind,
+            asset_path: e.asset_path.clone(),
+            partition_key: e.partition_key.clone(),
+            job_id: e.job_id,
+            at: e.at,
+        }
+    }
+}
+
+/// For `fires = 'all'`, the trigger should only fire when *every* path in
+/// the subscription set has seen a newer event than the last fire. This
+/// helper checks that condition against `asset_event`. Returns `true` if
+/// the trigger is ready to fire.
+pub async fn fires_all_ready(
+    db: &Pool<Postgres>,
+    workspace_id: &str,
+    subscription_paths: &[(AssetKind, String)],
+    last_fired_at: Option<chrono::DateTime<chrono::Utc>>,
+) -> Result<bool> {
+    let last = last_fired_at.unwrap_or_else(|| {
+        chrono::DateTime::<chrono::Utc>::from_timestamp(0, 0).unwrap_or_default()
+    });
+    for (kind, path) in subscription_paths {
+        let max_at = sqlx::query_scalar::<_, Option<chrono::DateTime<chrono::Utc>>>(
+            "SELECT MAX(at) FROM asset_event \
+             WHERE workspace_id = $1 AND asset_kind = $2 AND asset_path = $3",
+        )
+        .bind(workspace_id)
+        .bind(kind)
+        .bind(path)
+        .fetch_one(db)
+        .await?;
+        match max_at {
+            Some(t) if t > last => continue,
+            _ => return Ok(false),
+        }
+    }
+    Ok(true)
+}

backend/windmill-common/src/implicit_triggers.rs

@@ -26,6 +26,49 @@ use crate::{
 /// work unchanged (`SPLIT_PART(path, '/', 1)` — 'u' / 'f' / 'g').
 pub const IMPLICIT_TRIGGER_PATH_SUFFIX: &str = "/__asset_trigger__";
 
+/// Reserved argument names that asset-triggered scripts may receive from
+/// the dispatcher. When a `#trigger: asset` annotation is present, any
+/// *other* required-without-default arg makes the script un-fillable and
+/// is rejected at save time (stage 5).
+pub const RESERVED_ARG_NAMES: &[&str] = &["wm_asset_event", "wm_partition", "wm_backfill"];
+
+/// Validate that an asset-triggered script's schema has no required args
+/// outside the reserved set. Schema follows the JSON-Schema convention
+/// used everywhere else in Windmill: `required: [name,...]` plus
+/// `properties: { name: { default?: ... } }`.
+///
+/// Returns `Err` with a user-facing message on the first offending arg.
+pub fn validate_asset_triggered_script_schema(
+    schema: Option<&serde_json::value::RawValue>,
+) -> std::result::Result<(), String> {
+    let Some(raw) = schema else { return Ok(()) };
+    let Ok(parsed) = serde_json::from_str::<serde_json::Value>(raw.get()) else {
+        return Ok(());
+    };
+    let Some(required) = parsed.get("required").and_then(|v| v.as_array()) else {
+        return Ok(());
+    };
+    let properties = parsed.get("properties").and_then(|v| v.as_object());
+    for arg in required {
+        let Some(name) = arg.as_str() else { continue };
+        if RESERVED_ARG_NAMES.contains(&name) {
+            continue;
+        }
+        let has_default = properties
+            .and_then(|p| p.get(name))
+            .and_then(|v| v.as_object())
+            .map(|obj| obj.contains_key("default"))
+            .unwrap_or(false);
+        if !has_default {
+            return Err(format!(
+                "Script has required parameter `{name}` but is asset-triggered. \
+                 Give it a default, make it optional, or remove the asset trigger annotation."
+            ));
+        }
+    }
+    Ok(())
+}
+
 pub fn implicit_trigger_path(script_path: &str) -> String {
     format!("{script_path}{IMPLICIT_TRIGGER_PATH_SUFFIX}")
 }

backend/windmill-common/src/lib.rs

@@ -34,6 +34,7 @@ pub mod ai_google;
 pub mod ai_providers;
 pub mod ai_types;
 pub mod apps;
+pub mod asset_dispatcher;
 pub mod assets;
 pub mod audit;
 pub mod auth;