Merge remote-tracking branch 'origin/main' into dieri/triggers-queue-mode

backend/windmill-api/src/apps.rs

@@ -10,6 +10,7 @@ use std::{collections::HashMap, sync::Arc};
 use crate::{
     auth::OptTokened,
     db::{ApiAuthed, DB},
+    jobs::RunJobQuery,
     resources::get_resource_value_interpolated_internal,
     users::{require_owner_of_path, OptAuthed},
     utils::{check_scopes, WithStarredInfoQuery},
@@ -1654,6 +1655,8 @@ pub struct ExecuteApp {
     pub force_viewer_static_fields: Option<StaticFields>,
     pub force_viewer_one_of_fields: Option<OneOfFields>,
     pub force_viewer_allow_user_resources: Option<AllowUserResources>,
+    /// Runnable query parameters (e.g., memory_id for chat-enabled flows)
+    pub run_query_params: Option<RunJobQuery>,
 }
 
 fn digest(code: &str) -> String {
@@ -1894,6 +1897,12 @@ async fn execute_component(
     )
     .await?;
 
+    let is_flow = payload
+        .path
+        .as_ref()
+        .map(|p| p.starts_with("flow/"))
+        .unwrap_or(false);
+
     let (job_payload, tag, on_behalf_of) = match (payload.path, payload.raw_code, payload.id) {
         // flow or script:
         (Some(path), None, None) => get_payload_tag_from_prefixed_path(&path, &db, &w_id).await?,
@@ -1920,7 +1929,7 @@ async fn execute_component(
 
     let end_user_email = opt_authed.as_ref().map(|a| a.email.clone());
 
-    let (uuid, tx) = push(
+    let (uuid, mut tx) = push(
         &db,
         tx,
         &w_id,
@@ -1954,6 +1963,14 @@ async fn execute_component(
         None
     )
     .await?;
+
+    // Apply runnable query parameters if provided
+    if let Some(ref run_query) = payload.run_query_params {
+        if is_flow {
+            crate::jobs::process_flow_run_query_params(&mut tx, uuid, run_query).await?;
+        }
+    }
+
     tx.commit().await?;
 
     Ok(uuid.to_string())

backend/windmill-api/src/flow_conversations.rs

@@ -134,6 +134,7 @@ pub async fn get_or_create_conversation_with_id(
     } else {
         title.to_string()
     };
+
     // Create new conversation with provided ID
     let conversation = sqlx::query_as!(
         FlowConversation,
@@ -148,6 +149,7 @@ pub async fn get_or_create_conversation_with_id(
     )
     .fetch_one(&mut **tx)
     .await?;
+
     Ok(conversation)
 }
 

backend/windmill-api/src/flows.rs

@@ -1583,6 +1583,7 @@ mod tests {
                         skip_failures: true,
                         parallel: false,
                         parallelism: None,
+                        squash: None,
                     }),
                     stop_after_if: Some(StopAfterIf {
                         expr: "previous.isEmpty()".to_string(),

backend/windmill-api/src/jobs.rs

@@ -1741,7 +1741,7 @@ pub struct ListableCompletedJob {
     pub labels: Option<serde_json::Value>,
 }
 
-#[derive(Deserialize, Clone, Default)]
+#[derive(Debug, Deserialize, Clone, Default)]
 pub struct RunJobQuery {
     pub scheduled_for: Option<chrono::DateTime<chrono::Utc>>,
     pub scheduled_in_secs: Option<i64>,
@@ -3955,13 +3955,13 @@ async fn set_flow_memory_id(
     memory_id: Uuid,
 ) -> error::Result<()> {
     sqlx::query!(
-        "UPDATE v2_job_status 
-            SET flow_status = jsonb_set(
-                flow_status,
-                '{memory_id}',
-                to_jsonb($2::uuid)
-            )
-            WHERE id = $1",
+        "UPDATE v2_job_status
+         SET flow_status = jsonb_set(
+             flow_status,
+             '{memory_id}',
+             to_jsonb($2::uuid)
+         )
+         WHERE id = $1",
         job_id,
         memory_id
     )
@@ -3970,6 +3970,19 @@ async fn set_flow_memory_id(
     Ok(())
 }
 
+/// Apply flow-specific query parameters after job creation
+pub async fn process_flow_run_query_params(
+    tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
+    job_id: Uuid,
+    run_query: &RunJobQuery,
+) -> error::Result<()> {
+    // Set memory_id if provided (for agent memory)
+    if let Some(memory_id) = run_query.memory_id {
+        set_flow_memory_id(tx, job_id, memory_id).await?;
+    }
+    Ok(())
+}
+
 async fn handle_chat_conversation_messages(
     tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
     authed: &ApiAuthed,

backend/windmill-api/src/scripts.rs

@@ -41,11 +41,11 @@ use windmill_audit::ActionKind;
 use windmill_worker::{process_relative_imports, scoped_dependency_map::ScopedDependencyMap};
 
 use windmill_common::{
-    assets::{clear_asset_usage, insert_asset_usage, AssetUsageKind, AssetWithAltAccessType},
+    assets::{AssetUsageKind, AssetWithAltAccessType, clear_asset_usage, insert_asset_usage},
     error::to_anyhow,
     s3_helpers::upload_artifact_to_store,
     scripts::hash_script,
-    utils::WarnAfterExt,
+    utils::{WarnAfterExt, paginate_without_limits},
     worker::{CLOUD_HOSTED, MIN_VERSION_SUPPORTS_DEBOUNCING},
 };
 
@@ -61,7 +61,7 @@ use windmill_common::{
     },
     users::username_to_permissioned_as,
     utils::{
-        not_found_if_none, paginate, query_elems_from_hub, require_admin, Pagination, StripPath,
+        not_found_if_none, query_elems_from_hub, require_admin, Pagination, StripPath,
     },
     worker::to_raw_value,
     HUB_BASE_URL,
@@ -193,7 +193,7 @@ async fn list_search_scripts(
 ) -> JsonResult<Vec<SearchScript>> {
     let mut tx = user_db.begin(&authed).await?;
     #[cfg(feature = "enterprise")]
-    let n = 1000;
+    let n = 10000;
 
     #[cfg(not(feature = "enterprise"))]
     let n = 10;
@@ -219,7 +219,7 @@ async fn list_scripts(
     Query(pagination): Query<Pagination>,
     Query(lq): Query<ListScriptQuery>,
 ) -> JsonResult<Vec<ListableScript>> {
-    let (per_page, offset) = paginate(pagination);
+    let (per_page, offset) = paginate_without_limits(pagination);
     let mut sqlb = SqlBuilder::select_from("script as o")
         .fields(&[
             "hash",

backend/windmill-api/src/workspaces.rs

@@ -50,7 +50,7 @@ use windmill_common::{
     oauth2::WORKSPACE_SLACK_BOT_TOKEN_PATH,
     utils::{paginate, rd_string, require_admin, Pagination},
 };
-use windmill_git_sync::{handle_fork_branch_creation, handle_deployment_metadata, DeployedObject};
+use windmill_git_sync::{handle_deployment_metadata, handle_fork_branch_creation, DeployedObject};
 use windmill_worker::scoped_dependency_map::{DependencyMap, ScopedDependencyMap};
 
 #[cfg(feature = "enterprise")]
@@ -73,6 +73,7 @@ lazy_static::lazy_static! {
 
 pub fn workspaced_service() -> Router {
     let router = Router::new()
+        .route("/get_as_superadmin", get(get_workspace_as_superadmin))
         .route("/list_pending_invites", get(list_pending_invites))
         .route("/update", post(edit_workspace))
         .route("/archive", post(archive_workspace))
@@ -163,7 +164,10 @@ pub fn workspaced_service() -> Router {
             post(acknowledge_all_critical_alerts),
         )
         .route("/critical_alerts/mute", post(mute_critical_alerts))
-        .route("/create_workspace_fork_branch", post(create_workspace_fork_branch))
+        .route(
+            "/create_workspace_fork_branch",
+            post(create_workspace_fork_branch),
+        )
         .route("/operator_settings", post(update_operator_settings));
 
     #[cfg(all(feature = "stripe", feature = "enterprise"))]
@@ -706,7 +710,9 @@ async fn get_slack_oauth_config(
     .await?;
 
     // Mask the secret if it exists
-    let masked_secret = settings.slack_oauth_client_secret.map(|_| "***".to_string());
+    let masked_secret = settings
+        .slack_oauth_client_secret
+        .map(|_| "***".to_string());
 
     Ok(Json(GetSlackOAuthConfigResponse {
         slack_oauth_client_id: settings.slack_oauth_client_id,
@@ -2227,6 +2233,35 @@ async fn get_used_triggers(
     Ok(Json(websocket_used))
 }
 
+async fn get_workspace_as_superadmin(
+    authed: ApiAuthed,
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+) -> JsonResult<Workspace> {
+    require_super_admin(&db, &authed.email).await?;
+    let workspace = sqlx::query_as!(
+        Workspace,
+        "SELECT
+            workspace.id AS \"id!\",
+            workspace.name AS \"name!\",
+            workspace.owner AS \"owner!\",
+            workspace.deleted AS \"deleted!\",
+            workspace.premium AS \"premium!\",
+            workspace_settings.color AS \"color\",
+            workspace.parent_workspace_id AS \"parent_workspace_id\"
+        FROM workspace
+        LEFT JOIN workspace_settings ON workspace.id = workspace_settings.workspace_id
+        WHERE workspace.id = $1",
+        w_id
+    )
+    .fetch_optional(&db)
+    .await?;
+
+    let workspace = not_found_if_none(workspace, "workspace", w_id)?;
+
+    Ok(Json(workspace))
+}
+
 async fn list_workspaces_as_super_admin(
     authed: ApiAuthed,
     Extension(db): Extension<DB>,
@@ -2310,6 +2345,13 @@ lazy_static::lazy_static! {
         }
     };
 
+    pub static ref DISABLE_WORKSPACE_FORK: bool = {
+        match std::env::var("DISABLE_WORKSPACE_FORK") {
+            Ok(val) => val == "true",
+            Err(_) => false,
+        }
+    };
+
 }
 
 async fn create_workspace_require_superadmin() -> String {
@@ -2566,9 +2608,14 @@ async fn update_workspace_settings(
         WorkspaceGitSyncSettings::default()
     };
 
-    // We only keep the first git sync repo, since it is considered the main one
+    // We only keep the first git sync repo that is sync mode (use_individual_branch = false), since it is considered the main one
     // Context: see WIN-1559
-    git_sync_settings.repositories.truncate(1);
+    git_sync_settings.repositories = git_sync_settings
+        .repositories
+        .into_iter()
+        .filter(|r| !r.use_individual_branch.unwrap_or(false))
+        .take(1)
+        .collect();
 
     let serialized_config = serde_json::to_value::<WorkspaceGitSyncSettings>(git_sync_settings)
         .map_err(|err| Error::internal_err(err.to_string()))?;
@@ -2638,6 +2685,17 @@ async fn clone_groups(
     .execute(&mut **tx)
     .await?;
 
+    sqlx::query!(
+        "INSERT INTO usr_to_group (workspace_id, group_, usr)
+         SELECT $2, group_, usr
+         FROM usr_to_group
+         WHERE workspace_id = $1",
+        source_workspace_id,
+        target_workspace_id,
+    )
+    .execute(&mut **tx)
+    .await?;
+
     Ok(())
 }
 
@@ -3008,6 +3066,10 @@ async fn create_workspace_fork_branch(
         )));
     }
 
+    if *DISABLE_WORKSPACE_FORK {
+        require_super_admin(&db, &authed.email).await?;
+    }
+
     Ok(Json(
         handle_fork_branch_creation(&authed.email, &authed.username, &db, &w_id, &nw.id).await?,
     ))
@@ -3025,6 +3087,10 @@ async fn create_workspace_fork(
         )));
     }
 
+    if *DISABLE_WORKSPACE_FORK {
+        require_super_admin(&db, &authed.email).await?;
+    }
+
     let mut tx: Transaction<'_, Postgres> = db.begin().await?;
 
     // Generate unique forked workspace ID with wm-fork prefix