starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-09 00:04:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into fork-datatable-schema-export

Browse Source
This commit is contained in:
Diego Imbert
2026-03-24 13:16:46 +01:00
parent 5a541c1e20 37886edda1
commit 01c3cfe7fa
717 changed files with 37037 additions and 5391 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api-workspaces/src/workspaces.rs
+272 -13
View File
@@ -6,7 +6,7 @@
* LICENSE-AGPL for a copy of the license.
*/
use windmill_api_auth::{require_super_admin, ApiAuthed};
use windmill_api_auth::{require_devops_role, require_super_admin, ApiAuthed};
use windmill_api_users::users::WorkspaceInvite;
use windmill_common::email_oss::send_email_if_possible;
use windmill_common::usernames::{get_instance_username_or_create_pending, VALID_USERNAME};
@@ -159,6 +159,9 @@ pub fn workspaced_service() -> Router {
"/protection_rules/:rule_name",
post(update_protection_rule).delete(delete_protection_rule),
)
.route("/log_chat", post(log_ai_chat))
.route("/cloud_quotas", get(get_cloud_quotas))
.route("/prune_versions", post(prune_versions))
}
pub fn global_service() -> Router {
Router::new()
@@ -2832,7 +2835,7 @@ async fn list_workspaces_as_super_admin(
Query(pagination): Query<Pagination>,
ApiAuthed { email, .. }: ApiAuthed,
) -> JsonResult<Vec<Workspace>> {
require_super_admin(&db, &email).await?;
require_devops_role(&db, &email).await?;
let (per_page, offset) = paginate(pagination);
let mut tx = user_db.begin(&authed).await?;
@@ -3345,8 +3348,8 @@ async fn clone_scripts(
envs, concurrent_limit, concurrency_time_window_s, cache_ttl,
dedicated_worker, ws_error_handler_muted, priority, timeout,
delete_after_use, restart_unless_cancelled, concurrency_key,
visible_to_runner_only, no_main_func, codebase, has_preprocessor,
on_behalf_of_email, assets
visible_to_runner_only, auto_kind, codebase, has_preprocessor,
on_behalf_of_email, assets, modules
)
SELECT
$1, hash, path, parent_hashes, summary, description, content,
@@ -3355,8 +3358,8 @@ async fn clone_scripts(
envs, concurrent_limit, concurrency_time_window_s, cache_ttl,
dedicated_worker, ws_error_handler_muted, priority, timeout,
delete_after_use, restart_unless_cancelled, concurrency_key,
visible_to_runner_only, no_main_func, codebase, has_preprocessor,
on_behalf_of_email, assets
visible_to_runner_only, auto_kind, codebase, has_preprocessor,
on_behalf_of_email, assets, modules
FROM script
WHERE workspace_id = $2"#,
target_workspace_id,
@@ -3494,10 +3497,12 @@ async fn clone_apps(
app_id_mapping.insert(app.id, new_app_id);
}
let mut version_id_mapping: HashMap<i64, i64> = HashMap::new();
{
// Clone app versions
let app_versions = sqlx::query!(
"SELECT app_id, value, created_by, created_at, raw_app
"SELECT id, app_id, value, created_by, created_at, raw_app
FROM app_version
WHERE app_id = ANY(SELECT id FROM app WHERE workspace_id = $1)
ORDER BY app_id, created_at",
@@ -3508,15 +3513,45 @@ async fn clone_apps(
for version in app_versions {
if let Some(&new_app_id) = app_id_mapping.get(&version.app_id) {
sqlx::query!(
let new_version_id = sqlx::query_scalar!(
"INSERT INTO app_version (app_id, value, created_by, created_at, raw_app)
VALUES ($1, $2, $3, $4, $5)",
VALUES ($1, $2, $3, $4, $5) RETURNING id",
new_app_id,
version.value,
version.created_by,
version.created_at,
version.raw_app,
)
.fetch_one(&mut **tx)
.await?;
version_id_mapping.insert(version.id, new_version_id);
}
}
}
// Clone app bundles for raw apps
if !version_id_mapping.is_empty() {
let old_ids: Vec<i64> = version_id_mapping.keys().copied().collect();
let bundles = sqlx::query!(
"SELECT app_version_id, file_type, data FROM app_bundles
WHERE app_version_id = ANY($1) AND w_id = $2",
&old_ids,
source_workspace_id
)
.fetch_all(&mut **tx)
.await?;
for bundle in bundles {
if let Some(&new_version_id) = version_id_mapping.get(&bundle.app_version_id) {
sqlx::query!(
"INSERT INTO app_bundles (app_version_id, w_id, file_type, data)
VALUES ($1, $2, $3, $4)",
new_version_id,
target_workspace_id,
bundle.file_type,
bundle.data,
)
.execute(&mut **tx)
.await?;
}
@@ -3841,7 +3876,7 @@ pub(crate) async fn archive_workspace_impl(
// Delete non-session tokens scoped to this workspace
let deleted_tokens = sqlx::query_scalar!(
"DELETE FROM token WHERE workspace_id = $1 AND label IS DISTINCT FROM 'session' RETURNING token",
"DELETE FROM token WHERE workspace_id = $1 AND label IS DISTINCT FROM 'session' RETURNING token_prefix",
w_id
)
.fetch_all(&mut *tx)
@@ -4985,7 +5020,7 @@ async fn compare_workspaces(
compare_two_flows(&db, &source_workspace_id, &fork_workspace_id, &item.path)
.await?,
),
"app" => Some(
"app" | "raw_app" => Some(
compare_two_apps(&db, &source_workspace_id, &fork_workspace_id, &item.path).await?,
),
"resource" => Some(
@@ -5080,7 +5115,10 @@ async fn compare_workspaces(
.fold(0, |acc, s| acc + s.try_into().unwrap_or(0)),
scripts_changed: visible_diffs.iter().filter(|s| s.kind == "script").count(),
flows_changed: visible_diffs.iter().filter(|s| s.kind == "flow").count(),
apps_changed: visible_diffs.iter().filter(|s| s.kind == "app").count(),
apps_changed: visible_diffs
.iter()
.filter(|s| s.kind == "app" || s.kind == "raw_app")
.count(),
resources_changed: visible_diffs
.iter()
.filter(|s| s.kind == "resource")
@@ -5189,7 +5227,7 @@ async fn query_visible_items<'c>(
.fetch_all(&mut **tx)
.await?
}
"app" => {
"app" | "raw_app" => {
sqlx::query_scalar!(
"SELECT path FROM app
WHERE workspace_id = $1 AND path = ANY($2)",
@@ -5630,3 +5668,224 @@ async fn compare_two_folders(
exists_in_fork: target_folder.is_some(),
});
}
#[derive(Deserialize)]
struct LogAiChatPayload {
session_id: String,
provider: String,
model: String,
mode: String,
}
async fn log_ai_chat(
Extension(db): Extension<DB>,
Json(payload): Json<LogAiChatPayload>,
) -> Result<StatusCode> {
sqlx::query!(
"INSERT INTO ai_chat_usage (session_id, provider, model, mode) VALUES ($1, $2, $3, $4)
ON CONFLICT (session_id) DO UPDATE SET message_count = ai_chat_usage.message_count + 1",
&payload.session_id,
&payload.provider,
&payload.model,
&payload.mode
)
.execute(&db)
.await?;
Ok(StatusCode::NO_CONTENT)
}
#[derive(Serialize)]
struct QuotaInfo {
used: i64,
limit: i64,
prunable: i64,
}
#[derive(Serialize)]
struct CloudQuotas {
scripts: QuotaInfo,
flows: QuotaInfo,
apps: QuotaInfo,
variables: QuotaInfo,
resources: QuotaInfo,
}
async fn get_cloud_quotas(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
) -> JsonResult<CloudQuotas> {
require_admin(authed.is_admin, &authed.username)?;
if !*CLOUD_HOSTED {
return Err(Error::BadRequest(
"Cloud quotas are only available on cloud-hosted instances".to_string(),
));
}
let scripts_used =
sqlx::query_scalar!("SELECT COUNT(*) FROM script WHERE workspace_id = $1", &w_id)
.fetch_one(&db)
.await?
.unwrap_or(0);
let scripts_prunable = sqlx::query_scalar!(
"SELECT COUNT(*) FROM script s WHERE s.workspace_id = $1 AND s.hash NOT IN (
SELECT DISTINCT ON (path) hash FROM script
WHERE workspace_id = $1 AND deleted = false AND draft_only IS NOT TRUE
ORDER BY path, created_at DESC
)",
&w_id
)
.fetch_one(&db)
.await?
.unwrap_or(0);
let flows_used =
sqlx::query_scalar!("SELECT COUNT(*) FROM flow WHERE workspace_id = $1", &w_id)
.fetch_one(&db)
.await?
.unwrap_or(0);
let flows_prunable = sqlx::query_scalar!(
"SELECT COUNT(*) FROM flow_version fv
JOIN flow f ON f.workspace_id = fv.workspace_id AND f.path = fv.path
WHERE fv.workspace_id = $1 AND fv.id != f.versions[array_upper(f.versions, 1)]",
&w_id
)
.fetch_one(&db)
.await?
.unwrap_or(0);
let apps_used = sqlx::query_scalar!("SELECT COUNT(*) FROM app WHERE workspace_id = $1", &w_id)
.fetch_one(&db)
.await?
.unwrap_or(0);
let apps_prunable = sqlx::query_scalar!(
"SELECT COUNT(*) FROM app_version av
JOIN app a ON a.id = av.app_id
WHERE a.workspace_id = $1 AND av.id != a.versions[array_upper(a.versions, 1)]",
&w_id
)
.fetch_one(&db)
.await?
.unwrap_or(0);
let variables_used = sqlx::query_scalar!(
"SELECT COUNT(*) FROM variable WHERE workspace_id = $1",
&w_id
)
.fetch_one(&db)
.await?
.unwrap_or(0);
let resources_used = sqlx::query_scalar!(
"SELECT COUNT(*) FROM resource WHERE workspace_id = $1",
&w_id
)
.fetch_one(&db)
.await?
.unwrap_or(0);
Ok(Json(CloudQuotas {
scripts: QuotaInfo { used: scripts_used, limit: 5000, prunable: scripts_prunable },
flows: QuotaInfo { used: flows_used, limit: 1000, prunable: flows_prunable },
apps: QuotaInfo { used: apps_used, limit: 1000, prunable: apps_prunable },
variables: QuotaInfo { used: variables_used, limit: 10000, prunable: 0 },
resources: QuotaInfo { used: resources_used, limit: 10000, prunable: 0 },
}))
}
#[derive(Deserialize)]
struct PruneVersionsRequest {
resource_type: String,
}
#[derive(Serialize)]
struct PruneVersionsResponse {
pruned: u64,
}
async fn prune_versions(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Path(w_id): Path<String>,
Json(req): Json<PruneVersionsRequest>,
) -> JsonResult<PruneVersionsResponse> {
require_admin(authed.is_admin, &authed.username)?;
if !*CLOUD_HOSTED {
return Err(Error::BadRequest(
"Version pruning is only available on cloud-hosted instances".to_string(),
));
}
let pruned = match req.resource_type.as_str() {
"scripts" => {
let result = sqlx::query(
"DELETE FROM script
WHERE workspace_id = $1 AND hash NOT IN (
SELECT DISTINCT ON (path) hash FROM script
WHERE workspace_id = $1 AND deleted = false AND draft_only IS NOT TRUE
ORDER BY path, created_at DESC
)",
)
.bind(&w_id)
.execute(&db)
.await?;
result.rows_affected()
}
"flows" => {
let deleted = sqlx::query(
"DELETE FROM flow_version fv
USING flow f
WHERE fv.workspace_id = f.workspace_id AND fv.path = f.path
AND fv.workspace_id = $1
AND fv.id != f.versions[array_upper(f.versions, 1)]",
)
.bind(&w_id)
.execute(&db)
.await?;
sqlx::query(
"UPDATE flow SET versions = ARRAY[versions[array_upper(versions, 1)]]
WHERE workspace_id = $1 AND array_length(versions, 1) > 1",
)
.bind(&w_id)
.execute(&db)
.await?;
deleted.rows_affected()
}
"apps" => {
let deleted = sqlx::query(
"DELETE FROM app_version av
USING app a
WHERE av.app_id = a.id AND a.workspace_id = $1
AND av.id != a.versions[array_upper(a.versions, 1)]",
)
.bind(&w_id)
.execute(&db)
.await?;
sqlx::query(
"UPDATE app SET versions = ARRAY[versions[array_upper(versions, 1)]]
WHERE workspace_id = $1 AND array_length(versions, 1) > 1",
)
.bind(&w_id)
.execute(&db)
.await?;
deleted.rows_affected()
}
_ => {
return Err(Error::BadRequest(format!(
"Invalid resource type '{}'. Must be 'scripts', 'flows', or 'apps'",
req.resource_type
)));
}
};
Ok(Json(PruneVersionsResponse { pruned }))
}