windmill/backend/windmill-worker/src/php_executor.rs

use convert_case::{Case, Casing};
use itertools::Itertools;
use regex::Regex;
use serde_json::value::RawValue;
use std::{collections::HashMap, process::Stdio};
use tokio::{fs::File, io::AsyncReadExt, process::Command};
use uuid::Uuid;
use windmill_common::{
    error::{self, to_anyhow, Result},
    scripts::ScriptLang,
    utils::calculate_hash,
    worker::{write_file, Connection},
    workspace_dependencies::clean_lock_from_annotations,
};
use windmill_queue::MiniPulledJob;

use windmill_parser::Typ;
use windmill_queue::{append_logs, CanceledBy};

use crate::{
    common::{
        build_command_with_isolation, check_executor_binary_exists, create_args_and_out_file,
        get_reserved_variables, read_result, resolve_nsjail_timeout, resolve_nsjail_tmp_mount_block,
        start_child_process, MaybeLock, OccupancyMetrics,
    },
    handle_child::handle_child,
    is_sandboxing_enabled, COMPOSER_CACHE_DIR, COMPOSER_PATH, DISABLE_NUSER, NSJAIL_PATH, PHP_PATH,
};
use windmill_common::client::AuthedClient;

const NSJAIL_CONFIG_RUN_PHP_CONTENT: &str = include_str!("../nsjail/run.php.config.proto");

lazy_static::lazy_static! {
    static ref RE: Regex = Regex::new(r"^//\s?(\S+)\s*$").unwrap();
    // Set COMPOSER_VENDOR_CACHE_DISABLED=1 to always run `composer install`
    // instead of reusing a previously cached vendor directory.
    pub static ref COMPOSER_VENDOR_CACHE_DISABLED: bool =
        std::env::var("COMPOSER_VENDOR_CACHE_DISABLED").is_ok();
}

const COMPOSER_LOCK_SPLIT: &str = "\nLOCK\n";

pub fn parse_php_imports(code: &str) -> anyhow::Result<Option<String>> {
    let find_requirements = code
        .lines()
        .find_position(|x| x.starts_with("//require:") || x.starts_with("// require:"));

    if let Some((pos, _)) = find_requirements {
        let requirements = code
            .lines()
            .skip(pos + 1)
            .map_while(|x| {
                RE.captures(x).map(|x| {
                    match x.get(1).unwrap().as_str().split("@").collect_vec()[..] {
                        [path, version] => (path.to_string(), version.to_string()),
                        [path] | [path, ..] => (path.to_string(), "*".to_string()),
                        [] => unreachable!(),
                    }
                })
            })
            .collect::<HashMap<_, _>>();

        let composer_json =
            serde_json::to_string_pretty(&HashMap::from([("require", requirements)]))
                .map_err(to_anyhow)?;

        Ok(Some(composer_json))
    } else {
        Ok(None)
    }
}

pub async fn composer_install(
    mem_peak: &mut i32,
    canceled_by: &mut Option<CanceledBy>,
    job_id: &Uuid,
    w_id: &str,
    conn: &Connection,
    job_dir: &str,
    worker_name: &str,
    requirements: String,
    lock: Option<String>,
    occupancy_metrics: &mut OccupancyMetrics,
) -> Result<String> {
    check_executor_binary_exists("php", PHP_PATH.as_str(), "php")?;

    // When no lock is provided (previews), try to reuse a previously resolved
    // lockfile from the DB so we can hit the same vendor cache as deployed scripts.
    let lock = if lock.is_none() && !*COMPOSER_VENDOR_CACHE_DISABLED {
        let req_hash = format!("composer-{}", calculate_hash(&requirements));
        if let Some(db) = conn.as_sql() {
            sqlx::query_scalar!(
                "SELECT lockfile FROM pip_resolution_cache WHERE hash = $1",
                req_hash
            )
            .fetch_optional(db)
            .await
            .ok()
            .flatten()
        } else {
            None
        }
    } else {
        lock
    };

    // Cache the installed vendor/ directory keyed by requirements + lock content.
    // Set COMPOSER_VENDOR_CACHE_DISABLED=1 to opt out.
    let vendor_cache_hit = if !*COMPOSER_VENDOR_CACHE_DISABLED {
        if let Some(ref lock_content) = lock {
            let hash = calculate_hash(&format!("{requirements}{lock_content}"));
            let vendor_cache_path = format!("{}/vendor/{hash}", *COMPOSER_CACHE_DIR);
            let remote_path = format!("composer/vendor/{hash}");

            let (hit, cache_logs) =
                crate::global_cache::load_cache(&vendor_cache_path, &remote_path, true).await;

            if hit {
                append_logs(
                    job_id,
                    w_id,
                    format!("vendor cache hit, skipping composer install\n{cache_logs}"),
                    conn,
                )
                .await;

                #[cfg(unix)]
                std::os::unix::fs::symlink(&vendor_cache_path, format!("{job_dir}/vendor"))
                    .map_err(|e| {
                        error::Error::ExecutionErr(format!(
                            "could not symlink cached vendor dir: {e}"
                        ))
                    })?;

                #[cfg(not(unix))]
                windmill_common::worker::copy_dir_recursively(
                    &std::path::PathBuf::from(&vendor_cache_path),
                    &std::path::PathBuf::from(&format!("{job_dir}/vendor")),
                )?;

                true
            } else {
                false
            }
        } else {
            false
        }
    } else {
        false
    };

    if vendor_cache_hit {
        return Ok(match lock {
            Some(l) => format!("{requirements}{COMPOSER_LOCK_SPLIT}{l}"),
            None => unreachable!("cache hit requires a lock"),
        });
    }

    write_file(job_dir, "composer.json", &requirements)?;

    if let Some(lock) = lock.as_ref() {
        write_file(job_dir, "composer.lock", lock)?;
    }

    let mut child_cmd = Command::new(&*COMPOSER_PATH);
    let args = vec!["install", "--no-dev", "--no-progress"];
    child_cmd
        .current_dir(job_dir)
        .env("COMPOSER_HOME", &*COMPOSER_CACHE_DIR)
        .args(args)
        .stdout(Stdio::piped())
        .stderr(Stdio::piped());
    let child_process = start_child_process(child_cmd, &*COMPOSER_PATH, false).await?;

    handle_child(
        job_id,
        conn,
        mem_peak,
        canceled_by,
        child_process,
        false,
        worker_name,
        w_id,
        "composer install",
        None,
        false,
        &mut Some(occupancy_metrics),
        None,
        None,
    )
    .await?;

    // lock was `None` means composer resolved deps from scratch (no lock from
    // caller or DB). This is the only case where we should update the DB cache.
    let freshly_resolved = lock.is_none();
    let resolved_lock = match lock {
        Some(l) => l,
        None => {
            let mut lock_content = String::new();
            let mut lock_file = File::open(format!("{job_dir}/composer.lock")).await?;
            lock_file.read_to_string(&mut lock_content).await?;
            lock_content
        }
    };

    // Save the freshly installed vendor/ to the cache for future executions.
    if !*COMPOSER_VENDOR_CACHE_DISABLED {
        let hash = calculate_hash(&format!("{requirements}{resolved_lock}"));
        let vendor_cache_path = format!("{}/vendor/{hash}", *COMPOSER_CACHE_DIR);
        if let Err(e) = crate::global_cache::save_cache(
            &vendor_cache_path,
            &format!("composer/vendor/{hash}"),
            &format!("{job_dir}/vendor"),
            true,
        )
        .await
        {
            tracing::warn!("Could not save composer vendor dir to cache: {e:?}");
        }

        // Cache the resolved lockfile in the DB so future previews (which lack a
        // lock file) can look it up by requirements hash and hit the same vendor
        // cache. TTL of 7 days keeps previews reasonably fresh.
        // Only write when composer resolved from scratch (no lock from caller or
        // DB) to avoid endlessly refreshing the TTL on stale resolutions.
        if freshly_resolved {
            let req_hash = format!("composer-{}", calculate_hash(&requirements));
            if let Some(db) = conn.as_sql() {
                if let Err(e) = sqlx::query!(
                    "INSERT INTO pip_resolution_cache (hash, lockfile, expiration) VALUES ($1, $2, now() + ('7 days')::interval) ON CONFLICT (hash) DO UPDATE SET lockfile = EXCLUDED.lockfile, expiration = EXCLUDED.expiration",
                    req_hash,
                    &resolved_lock
                )
                .execute(db)
                .await
                {
                    tracing::warn!("Could not cache composer lockfile resolution: {e:?}");
                }
            }
        }
    }

    Ok(format!(
        "{requirements}{COMPOSER_LOCK_SPLIT}{resolved_lock}"
    ))
}

fn generate_resource_class(rt_name: &str, arg_name: &str) -> String {
    let rt_name = rt_name.to_case(Case::Pascal);
    format!(
        "#[AllowDynamicProperties]
class {rt_name} {{
  public function __construct($data) {{
    foreach ($data AS $key => $value) $this->{{$key}} = $value;
  }}
}}
$args->{arg_name} = new {rt_name}($args->{arg_name});"
    )
}

fn split_reqs_and_lock(content: &String) -> error::Result<(Option<String>, Option<String>)> {
    let splitted = content.split(COMPOSER_LOCK_SPLIT).collect_vec();
    if splitted.len() != 2 {
        return Err(error::Error::ExecutionErr(format!(
            "Invalid requirements, expected to find LOCK split pattern in reqs. Found: |{content}|"
        )));
    }

    Ok((
        Some(clean_lock_from_annotations(splitted[0], ScriptLang::Php)),
        Some(splitted[1].to_string()),
    ))
}

#[tracing::instrument(level = "trace", skip_all)]
pub async fn handle_php_job(
    maybe_lock: MaybeLock,
    mem_peak: &mut i32,
    canceled_by: &mut Option<CanceledBy>,
    job: &MiniPulledJob,
    conn: &Connection,
    client: &AuthedClient,
    parent_runnable_path: Option<String>,
    job_dir: &str,
    inner_content: &String,
    base_internal_url: &str,
    worker_name: &str,
    envs: HashMap<String, String>,
    shared_mount: &str,
    occupancy_metrics: &mut OccupancyMetrics,
) -> error::Result<Box<RawValue>> {
    check_executor_binary_exists("php", PHP_PATH.as_str(), "php")?;

    let (composer_json, composer_lock) = match &maybe_lock {
        MaybeLock::Resolved { lock } if !lock.is_empty() => split_reqs_and_lock(lock)?,
        MaybeLock::Unresolved { workspace_dependencies } => (
            workspace_dependencies
                .get_php()?
                .or(parse_php_imports(inner_content)?),
            None,
        ),
        _ => (parse_php_imports(inner_content)?, None),
    };

    let autoload_line = if let Some(composer_json) = composer_json {
        let logs1 = "\n\n--- COMPOSER INSTALL ---\n".to_string();
        append_logs(&job.id, &job.workspace_id, logs1, conn).await;

        composer_install(
            mem_peak,
            canceled_by,
            &job.id,
            &job.workspace_id,
            conn,
            job_dir,
            worker_name,
            composer_json,
            composer_lock,
            occupancy_metrics,
        )
        .await?;
        "require './vendor/autoload.php';"
    } else {
        ""
    };

    let init_logs = "\n\n--- PHP CODE EXECUTION ---\n".to_string();

    append_logs(&job.id, job.workspace_id.to_string(), init_logs, conn).await;

    let _ = write_file(job_dir, "main.php", inner_content)?;

    let main_override = job.script_entrypoint_override.as_deref();

    let write_wrapper_f = async {
        let args = windmill_parser_php::parse_php_signature(
            inner_content,
            main_override.map(ToString::to_string),
        )?
        .args;

        let args_to_include = args
            .iter()
            .filter(|x| {
                !x.has_default || job.args.as_ref().is_some_and(|a| a.contains_key(&x.name))
            })
            .collect::<Vec<_>>();

        let func_args_str = args_to_include
            .iter()
            .map(|x| format!("$args->{}", x.name))
            .collect::<Vec<String>>()
            .join(",");

        let resource_classes = args_to_include
            .iter()
            .filter_map(|x| match &x.typ {
                Typ::Resource(name) => Some((name, &x.name)),
                _ => None,
            })
            .unique()
            .map(|(rt_name, arg_name)| generate_resource_class(rt_name, arg_name))
            .collect::<Vec<_>>()
            .join("\n\n");

        let main_name = main_override.unwrap_or("main");

        let wrapper_content: String = format!(
            r#"
<?php
{autoload_line}
require './main.php';

$args_str = file_get_contents('./args.json');
$args = json_decode($args_str);

{resource_classes}

try {{
  $res = {main_name}({func_args_str});
  file_put_contents("result.json", json_encode($res));
}} catch (Exception $e) {{
    $err = [
        "message" => $e->getMessage(),
        "name" => get_class($e),
        "stack" => $e->getTraceAsString()
    ];
    $step_id = getenv('WM_FLOW_STEP_ID');
    if ($step_id) {{
        $err["step_id"] = $step_id;
    }}
    file_put_contents("result.json", json_encode($err));
    exit(1);
}}
    "#,
        );
        write_file(job_dir, "wrapper.php", &wrapper_content)?;
        Ok(()) as error::Result<()>
    };

    let reserved_variables_args_out_f = async {
        let args_and_out_f = async {
            create_args_and_out_file(&client, job, job_dir, conn).await?;
            Ok(()) as Result<()>
        };
        let reserved_variables_f = async {
            let vars =
                get_reserved_variables(job, &client.token, conn, parent_runnable_path.clone())
                    .await?;
            Ok(vars) as Result<HashMap<String, String>>
        };
        let (_, reserved_variables) = tokio::try_join!(args_and_out_f, reserved_variables_f)?;
        Ok(reserved_variables) as error::Result<HashMap<String, String>>
    };

    let (reserved_variables, _) = tokio::try_join!(reserved_variables_args_out_f, write_wrapper_f)?;

    let child = if is_sandboxing_enabled() {
        let nsjail_timeout =
            resolve_nsjail_timeout(conn, &job.workspace_id, job.id, job.timeout).await;
        let _ = write_file(
            job_dir,
            "run.config.proto",
            &NSJAIL_CONFIG_RUN_PHP_CONTENT
                .replace("{JOB_DIR}", job_dir)
                .replace("{CLONE_NEWUSER}", &(!*DISABLE_NUSER).to_string())
                .replace("{SHARED_MOUNT}", shared_mount)
                .replace(
                    "{TMP_MOUNT_BLOCK}",
                    &resolve_nsjail_tmp_mount_block(job_dir).await,
                )
                .replace("{TIMEOUT}", &nsjail_timeout),
        )?;

        let mut nsjail_cmd = Command::new(NSJAIL_PATH.as_str());
        let args = vec![
            "--config",
            "run.config.proto",
            "--",
            &PHP_PATH,
            "/tmp/wrapper.php",
        ];
        nsjail_cmd
            .current_dir(job_dir)
            .env_clear()
            .envs(envs)
            .envs(reserved_variables)
            .envs(crate::get_otel_context_envs(&job.id))
            .env("COMPOSER_HOME", &*COMPOSER_CACHE_DIR)
            .env("BASE_INTERNAL_URL", base_internal_url)
            .args(args)
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());
        start_child_process(nsjail_cmd, NSJAIL_PATH.as_str(), false).await?
    } else {
        let script_path = format!("{job_dir}/wrapper.php");
        let args = vec![script_path.as_str()];

        let mut php_cmd = build_command_with_isolation(&*PHP_PATH, &args);
        php_cmd
            .current_dir(job_dir)
            .env_clear()
            .envs(envs)
            .envs(reserved_variables)
            .envs(crate::get_otel_context_envs(&job.id))
            .env("COMPOSER_HOME", &*COMPOSER_CACHE_DIR)
            .env("BASE_INTERNAL_URL", base_internal_url)
            .stdin(Stdio::null())
            .stdout(Stdio::piped())
            .stderr(Stdio::piped());

        start_child_process(php_cmd, &*PHP_PATH, false).await?
    };

    handle_child(
        &job.id,
        conn,
        mem_peak,
        canceled_by,
        child,
        is_sandboxing_enabled(),
        worker_name,
        &job.workspace_id,
        "php run",
        job.timeout,
        false,
        &mut Some(occupancy_metrics),
        None,
        None,
    )
    .await?;
    read_result(job_dir, None).await
}