feat(worker): support workers to run natively on windows

backend/parsers/windmill-parser-yaml/src/lib.rs

@@ -399,15 +399,11 @@ fn parse_ansible_options(opts: &Vec<Yaml>) -> AnsiblePlaybookOptions {
                             if c > 0 && c <= 6 {
                                 ret.verbosity = Some("v".repeat(c.min(6)));
                             }
-
                         }
                     }
-                    _ => ()
-
+                    _ => (),
                 }
             }
-
-
         }
     }
 
@@ -422,10 +418,10 @@ fn count_consecutive_vs(s: &str) -> usize {
         if c == 'v' {
             current_count += 1;
             if current_count == 6 {
-                return 6;  // Stop early if we reach 6
+                return 6; // Stop early if we reach 6
             }
         } else {
-            current_count = 0;  // Reset count if the character is not 'v'
+            current_count = 0; // Reset count if the character is not 'v'
         }
         max_count = max_count.max(current_count);
     }

backend/windmill-api/src/jobs.rs

@@ -11,14 +11,14 @@ use axum::http::HeaderValue;
 use quick_cache::sync::Cache;
 use serde_json::value::RawValue;
 use sqlx::Pool;
-use windmill_common::error::JsonResult;
 use std::collections::HashMap;
 #[cfg(feature = "prometheus")]
 use std::sync::atomic::Ordering;
 use tokio::io::AsyncReadExt;
 #[cfg(feature = "prometheus")]
 use tokio::time::Instant;
 use tower::ServiceBuilder;
+use windmill_common::error::JsonResult;
 use windmill_common::flow_status::{JobResult, RestartedFrom};
 use windmill_common::jobs::{
     format_completed_job_result, format_result, CompletedJobWithFormattedResult, FormattedResult,
@@ -78,6 +78,7 @@ use windmill_common::s3_helpers::OBJECT_STORE_CACHE_SETTINGS;
 #[cfg(feature = "prometheus")]
 use windmill_common::{METRICS_DEBUG_ENABLED, METRICS_ENABLED};
 
+use windmill_common::utils::paginate_without_limits;
 use windmill_common::{get_latest_deployed_hash_for_path, BASE_URL};
 use windmill_queue::{
     cancel_job, get_queued_job, get_result_by_id_from_running_flow, job_is_complete, push,
@@ -293,11 +294,8 @@ pub fn workspace_unauthed_service() -> Router {
 
 pub fn global_root_service() -> Router {
     Router::new()
-    .route("/db_clock", get(get_db_clock))
-    .route(
-        "/completed/count_by_tag",
-        get(count_by_tag),
-    )
+        .route("/db_clock", get(get_db_clock))
+        .route("/completed/count_by_tag", get(count_by_tag))
 }
 
 #[derive(Deserialize)]
@@ -4683,8 +4681,8 @@ async fn get_job_update(
     .fetch_optional(&db)
     .await?;
 
-    let progress: Option<i32> = if get_progress == Some(true){
-         sqlx::query_scalar!(
+    let progress: Option<i32> = if get_progress == Some(true) {
+        sqlx::query_scalar!(
                 "SELECT scalar_int FROM job_stats WHERE workspace_id = $1 AND job_id = $2 AND metric_id = $3",
                 &w_id,
                  job_id,
@@ -5115,8 +5113,6 @@ async fn get_completed_job_result(
     Ok(Json(result).into_response())
 }
 
-
-
 #[derive(Deserialize)]
 struct CountByTagQuery {
     horizon_secs: Option<i64>,
@@ -5130,7 +5126,7 @@ struct TagCount {
 }
 
 async fn count_by_tag(
-    ApiAuthed { email, ..}: ApiAuthed,
+    ApiAuthed { email, .. }: ApiAuthed,
     Extension(db): Extension<DB>,
     Query(query): Query<CountByTagQuery>,
 ) -> JsonResult<Vec<TagCount>> {

backend/windmill-api/src/resources.rs

@@ -58,7 +58,10 @@ pub fn workspaced_service() -> Router {
         .route("/type/exists/:name", get(exists_resource_type))
         .route("/type/update/:name", post(update_resource_type))
         .route("/type/delete/:name", delete(delete_resource_type))
-        .route("/file_resource_type_to_file_ext_map", get(file_resource_ext_to_resource_type))
+        .route(
+            "/file_resource_type_to_file_ext_map",
+            get(file_resource_ext_to_resource_type),
+        )
         .route("/type/create", post(create_resource_type))
 }
 

backend/windmill-api/src/settings.rs

@@ -24,9 +24,9 @@ use axum::{
 #[cfg(feature = "enterprise")]
 use axum::extract::Query;
 
+use serde::Deserialize;
 #[cfg(feature = "enterprise")]
 use windmill_common::ee::{send_critical_alert, CriticalAlertKind, CriticalErrorChannel};
-use serde::Deserialize;
 use windmill_common::{
     error::{self, JsonResult, Result},
     global_settings::{

backend/windmill-indexer/src/indexer_ee.rs

@@ -1,6 +1,6 @@
+use anyhow::anyhow;
 use sqlx::{Pool, Postgres};
 use windmill_common::error::Error;
-use anyhow::anyhow;
 
 #[derive(Clone)]
 pub struct IndexReader;

backend/windmill-worker/src/ansible_executor.rs

@@ -1,10 +1,18 @@
+#[cfg(unix)]
 use std::{
     collections::HashMap,
     os::unix::fs::PermissionsExt,
     path::{Path, PathBuf},
     process::Stdio,
 };
 
+#[cfg(windows)]
+use std::{
+    collections::HashMap,
+    path::{Path, PathBuf},
+    process::Stdio,
+};
+
 use anyhow::anyhow;
 use itertools::Itertools;
 use serde_json::value::RawValue;
@@ -378,6 +386,7 @@ fi
 
         let file = write_file(job_dir, "wrapper.sh", &wrapper)?;
 
+        #[cfg(unix)]
         file.metadata()?.permissions().set_mode(0o777);
         // let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str());
         let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str());

backend/windmill-worker/src/bash_executor.rs

@@ -32,6 +32,9 @@ use crate::{
     POWERSHELL_CACHE_DIR, POWERSHELL_PATH, TZ_ENV,
 };
 
+#[cfg(windows)]
+use crate::SYSTEM_ROOT;
+
 lazy_static::lazy_static! {
 
     pub static ref ANSI_ESCAPE_RE: Regex = Regex::new(r"\x1b\[[0-9;]*m").unwrap();
@@ -226,13 +229,19 @@ pub async fn handle_powershell_job(
             .collect::<Vec<_>>()
     };
 
+    #[cfg(windows)]
+    let split_char = '\\';
+
+    #[cfg(unix)]
+    let split_char = '/';
+
     let installed_modules = fs::read_dir(POWERSHELL_CACHE_DIR)?
         .filter_map(|x| {
             x.ok().map(|x| {
                 x.path()
                     .display()
                     .to_string()
-                    .split('/')
+                    .split(split_char)
                     .last()
                     .unwrap_or_default()
                     .to_lowercase()
@@ -289,6 +298,7 @@ pub async fn handle_powershell_job(
     append_logs(&job.id, &job.workspace_id, logs2, db).await;
 
     // make sure default (only allhostsallusers) modules are loaded, disable autoload (cache can be large to explore especially on cloud) and add /tmp/windmill/cache to PSModulePath
+    #[cfg(unix)]
     let profile = format!(
         "$PSModuleAutoloadingPreference = 'None'
 $PSModulePathBackup = $env:PSModulePath
@@ -297,6 +307,17 @@ Get-Module -ListAvailable | Import-Module
 $env:PSModulePath = \"{}:$PSModulePathBackup\"",
         POWERSHELL_CACHE_DIR
     );
+
+    #[cfg(windows)]
+    let profile = format!(
+        "$PSModuleAutoloadingPreference = 'None'
+$PSModulePathBackup = $env:PSModulePath
+$env:PSModulePath = \"C:\\WINDOWS\\System32\\WindowsPowerShell\\v1.0\\Modules;C:\\Program Files\\PowerShell\\7\\Modules\"
+Get-Module -ListAvailable | Import-Module
+$env:PSModulePath = \"{};$PSModulePathBackup\"",
+        POWERSHELL_CACHE_DIR
+    );
+
     // make sure param() is first
     let param_match = windmill_parser_bash::RE_POWERSHELL_PARAM.find(&content);
     let content: String = if let Some(param_match) = param_match {
@@ -312,11 +333,29 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"",
     };
 
     write_file(job_dir, "main.ps1", content.as_str())?;
+
+    #[cfg(unix)]
     write_file(
         job_dir,
         "wrapper.sh",
         &format!("set -o pipefail\nset -e\nmkfifo bp\ncat bp | tail -1 > ./result2.out &\n{} -F ./main.ps1 \"$@\" 2>&1 | tee bp\nwait $!", POWERSHELL_PATH.as_str()),
     )?;
+
+    #[cfg(windows)]
+    write_file(
+        job_dir,
+        "wrapper.ps1",
+        &format!(
+            " param([string[]]$args)\n\
+    $ErrorActionPreference = 'Stop'\n\
+    $pipe = New-TemporaryFile\n\
+    & \"{}\" -File ./main.ps1 @args 2>&1 | Tee-Object -FilePath $pipe\n\
+    Get-Content -Path $pipe | Select-Object -Last 1 | Set-Content -Path './result2.out'\n\
+    Remove-Item $pipe\n",
+            POWERSHELL_PATH.as_str()
+        ),
+    )?;
+
     let token = client.get_token().await;
     let mut reserved_variables = get_reserved_variables(job, &token, db).await?;
     reserved_variables.insert("RUST_LOG".to_string(), "info".to_string());
@@ -355,22 +394,53 @@ $env:PSModulePath = \"{}:$PSModulePathBackup\"",
             .stderr(Stdio::piped())
             .spawn()?
     } else {
-        let mut cmd_args = vec!["wrapper.sh"];
-        cmd_args.extend(pwsh_args.iter().map(|x| x.as_str()));
-        Command::new(BIN_BASH.as_str())
-            .current_dir(job_dir)
+        let mut cmd;
+        let mut cmd_args;
+
+        #[cfg(unix)]
+        {
+            cmd_args = vec!["wrapper.sh"];
+            cmd_args.extend(pwsh_args.iter().map(|x| x.as_str()));
+            cmd = Command::new(BIN_BASH.as_str());
+        }
+
+        #[cfg(windows)]
+        {
+            cmd_args = vec![r".\wrapper.ps1".to_string()];
+            cmd_args.extend(pwsh_args.iter().map(|x| x.replace("--", "-")));
+            cmd = Command::new(POWERSHELL_PATH.as_str());
+        }
+
+        cmd.current_dir(job_dir)
             .env_clear()
             .envs(envs)
             .envs(reserved_variables)
             .env("TZ", TZ_ENV.as_str())
             .env("PATH", PATH_ENV.as_str())
             .env("BASE_INTERNAL_URL", base_internal_url)
             .env("HOME", HOME_ENV.as_str())
-            .args(cmd_args)
+            .args(&cmd_args)
             .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
-            .spawn()?
+            .stderr(Stdio::piped());
+
+        #[cfg(windows)]
+        {
+            cmd.env("SystemRoot", SYSTEM_ROOT.as_str())
+                .env(
+                    "TMP",
+                    std::env::var("TMP").unwrap_or_else(|_| String::from("/tmp")),
+                )
+                .env(
+                    "PATHEXT",
+                    std::env::var("PATHEXT").unwrap_or_else(|_| {
+                        String::from(".COM;.EXE;.BAT;.CMD;.VBS;.VBE;.JS;.JSE;.WSF;.WSH;.MSC;.CPL")
+                    }),
+                );
+        }
+
+        cmd.spawn()?
     };
+
     handle_child(
         &job.id,
         db,