Merge branch 'main' into di/data-tables

This commit is contained in:
Diego Imbert
2025-11-27 18:42:14 +01:00
204 changed files with 5551 additions and 3115 deletions
+18 -1
View File
@@ -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())
@@ -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)
}
+6 -1
View File
@@ -133,7 +133,11 @@ async fn list_flows(
"o.workspace_id",
"o.path",
"summary",
"description",
if !lq.without_description.unwrap_or(false) {
"description"
} else {
"NULL as description"
},
"fv.created_by as edited_by",
"fv.created_at as edited_at",
"archived",
@@ -1579,6 +1583,7 @@ mod tests {
skip_failures: true,
parallel: false,
parallelism: None,
squash: None,
}),
stop_after_if: Some(StopAfterIf {
expr: "previous.isEmpty()".to_string(),
+2 -2
View File
@@ -96,7 +96,7 @@ async fn list_folders(
let rows = sqlx::query_as!(
Folder,
"SELECT workspace_id, name, display_name, owners, extra_perms, summary, created_by, edited_at FROM folder WHERE workspace_id = $1 ORDER BY name desc LIMIT $2 OFFSET $3",
"SELECT workspace_id, name, display_name, owners, extra_perms, summary, created_by, edited_at FROM folder WHERE workspace_id = $1 ORDER BY name asc LIMIT $2 OFFSET $3",
w_id,
per_page as i64,
offset as i64
@@ -117,7 +117,7 @@ async fn list_foldernames(
let mut tx = user_db.begin(&authed).await?;
let rows = sqlx::query_scalar!(
"SELECT name FROM folder WHERE workspace_id = $1 ORDER BY name desc LIMIT $2 OFFSET $3",
"SELECT name FROM folder WHERE workspace_id = $1 ORDER BY name asc LIMIT $2 OFFSET $3",
w_id,
per_page as i64,
offset as i64
+2 -2
View File
@@ -103,7 +103,7 @@ async fn list_groups(
let rows = sqlx::query_as!(
Group,
"SELECT * FROM group_ WHERE workspace_id = $1 ORDER BY name desc LIMIT $2 OFFSET $3",
"SELECT * FROM group_ WHERE workspace_id = $1 ORDER BY name asc LIMIT $2 OFFSET $3",
w_id,
per_page as i64,
offset as i64
@@ -126,7 +126,7 @@ async fn list_group_names(
) -> JsonResult<Vec<String>> {
let rows = if !only_member_of.unwrap_or(false) {
sqlx::query_scalar!(
"SELECT name FROM group_ WHERE workspace_id = $1 UNION SELECT name FROM instance_group ORDER BY name desc",
"SELECT name FROM group_ WHERE workspace_id = $1 UNION SELECT name FROM instance_group ORDER BY name asc",
w_id
)
.fetch_all(&db)
+15 -2
View File
@@ -1743,7 +1743,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>,
@@ -3957,7 +3957,7 @@ async fn set_flow_memory_id(
memory_id: Uuid,
) -> error::Result<()> {
sqlx::query!(
"UPDATE v2_job_status
"UPDATE v2_job_status
SET flow_status = jsonb_set(
flow_status,
'{memory_id}',
@@ -3972,6 +3972,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,
+10 -5
View File
@@ -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",
@@ -228,6 +228,11 @@ async fn list_scripts(
"COALESCE(draft.created_at, o.created_at) as created_at",
"archived",
"extra_perms",
if !lq.without_description.unwrap_or(false) {
"description"
} else {
"NULL as description"
},
"CASE WHEN lock_error_logs IS NOT NULL THEN true ELSE false END as has_deploy_errors",
"language",
"favorite.path IS NOT NULL as starred",
+63 -2
View File
@@ -75,6 +75,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))
@@ -2342,6 +2343,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>,
@@ -2425,6 +2455,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 {
@@ -2681,9 +2718,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()))?;
@@ -2753,6 +2795,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(())
}
@@ -3123,6 +3176,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?,
))
@@ -3140,6 +3197,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