use crate::compiler::util::{
    checksum_buffer, checksum_file, cleanup_specific_directories, cleanup_specific_files,
    copy_file, copy_file_if_checksum_differs, crate_name_pipeline_globals,
    crate_name_pipeline_main, create_dir_if_not_exists, create_new_file, decode_string_as_dir,
    pipeline_binary_filename, program_info_filename, read_file_content, recreate_dir,
    recreate_file_with_content, truncate_sha256_checksum, write_file, CleanupDecision,
    DirectoryContent, ProcessGroupTerminator, UtilError,
};
use crate::config::{CommonConfig, CompilerConfig};
use crate::db::error::DBError;
use crate::db::storage::Storage;
use crate::db::storage_postgres::StoragePostgres;
use crate::db::types::pipeline::PipelineId;
use crate::db::types::program::{CompilationProfile, RuntimeSelector, RustCompilationInfo};
use crate::db::types::tenant::TenantId;
use crate::db::types::utils::{validate_program_config, validate_program_info};
use crate::db::types::version::Version;
use crate::error::source_error;
use crate::has_unstable_feature;
use chrono::{DateTime, Utc};
use feldera_types::config::PipelineConfigProgramInfo;
use futures_util::stream;
use indoc::formatdoc;
use openssl::sha;
use openssl::sha::sha256;
use std::collections::{BTreeMap, HashSet};
use std::path::{Path, PathBuf};
use std::time::{Instant, SystemTime};
use std::{process::Stdio, sync::Arc};
use tokio::{
    fs,
    io::AsyncReadExt,
    process::Command,
    sync::Mutex,
    time::{sleep, Duration},
};
use tracing::{debug, error, info, trace, warn};

/// The frequency at which the compiler polls the database for new Rust compilation requests.
/// It balances resource consumption due to polling and a fast Rust compilation response.
const POLL_INTERVAL: Duration = Duration::from_millis(500);

/// Minimum frequency at which the compiler polls the database.
/// This minimum is a preventative measure to avoid the Rust compiler
/// from ever polling the database in an uninterrupted loop.
const POLL_INTERVAL_MINIMUM: Duration = Duration::from_millis(25);

/// The poll frequency when an unexpected database error occurred.
/// This is set relatively long to not flood the logs when
/// for instance the database becomes temporarily unreachable.
const POLL_ERROR_INTERVAL: Duration = Duration::from_secs(30);

/// The frequency at which during Rust compilation it is checked whether
/// the process has finished and as well whether it needs to be canceled
/// if the program is outdated.
const COMPILATION_CHECK_INTERVAL: Duration = Duration::from_millis(250);

/// The frequency at which Rust cleanup is performed.
const CLEANUP_INTERVAL: Duration = Duration::from_secs(120);

/// Minimum time between when the Rust cleanup has detected a pipeline to be deleted until
/// its compilation artifacts are actually cleaned up. It is a minimum, as cleanup both
/// happens at an interval, and as well is interleaved with compilation which can take
/// a significant amount of time.
const CLEANUP_RETENTION: Duration = Duration::from_secs(3600);

/// Maximum chunk size for uploading binary to HTTP endpoint.
/// This balances memory usage and upload performance.
const MAX_CHUNK_SIZE_FOR_UPLOAD_BINARY: usize = 10 * 1024 * 1024; // 10MB

/// Rust compilation task that wakes up periodically.
/// Sleeps inbetween ticks which affects the response time of Rust compilation.
/// This task cannot fail, and any internal errors are caught and written to log if need-be.
pub async fn rust_compiler_task(
    worker_id: usize,
    total_workers: usize,
    common_config: CommonConfig,
    config: CompilerConfig,
    db: Arc<Mutex<StoragePostgres>>,
) {
    let mut last_cleanup: Option<Instant> = None;
    loop {
        let mut unexpected_error = false;

        // Clean up
        if last_cleanup.is_none_or(|ts| ts.elapsed() >= CLEANUP_INTERVAL) {
            if let Err(e) = cleanup_rust_compilation(&config, db.clone()).await {
                match e {
                    RustCompilationCleanupError::Database(e) => {
                        error!("Rust worker {worker_id}: compilation cleanup failed: database error occurred: {e}");
                    }
                    RustCompilationCleanupError::Utility(e) => {
                        error!("Rust worker {worker_id}: compilation cleanup failed: utility error occurred: {e}");
                    }
                }
                unexpected_error = true;
            }
            last_cleanup = Some(Instant::now());
        }

        // Compile
        let result = attempt_end_to_end_rust_compilation(
            worker_id,
            total_workers,
            &common_config,
            &config,
            db.clone(),
        )
        .await;
        if let Err(e) = &result {
            match e {
                DBError::UnknownPipeline { pipeline_id } => {
                    debug!(
                        pipeline_id = %pipeline_id,
                        pipeline = "N/A",
                        "Rust worker {worker_id}: compilation canceled: pipeline no longer exists"
                    );
                }
                DBError::OutdatedProgramVersion {
                    outdated_version,
                    latest_version,
                } => {
                    debug!("Rust worker {worker_id}: compilation canceled: pipeline program version ({outdated_version}) is outdated by latest ({latest_version})");
                }
                e => {
                    unexpected_error = true;
                    error!("Rust worker {worker_id}: compilation canceled: unexpected database error occurred: {e}");
                }
            }
        }

        // Wait
        if unexpected_error {
            // Unexpected error occurred
            sleep(POLL_ERROR_INTERVAL).await;
        } else if result.is_ok_and(|found| !found) {
            // No pipeline was found to attempt to be compiled
            sleep(POLL_INTERVAL).await;
        } else {
            // A pipeline was attempted to be compiled or an expected
            // database error occurred (e.g., no longer exists, outdated)
            sleep(POLL_INTERVAL_MINIMUM).await;
        }
    }
}

/// Performs Rust compilation:
/// 1. Reset in the database any pipeline with `program_status` of `CompilingRust` back to
///    `SqlCompiled` if they are of the current `platform_version`. Any pipeline with
///    `program_status` of `SqlCompiled` or `CompilingRust` with a non-current `platform_version`
///    will have it updated to current and its `program_status` set back to `Pending`.
/// 2. Queries the database for a pipeline which has `program_status` of `SqlCompiled` the longest
/// 3. Updates pipeline database `program_status` to `CompilingRust`
/// 4. Performs Rust compilation on `program_info.main`, `udf_rust` and `udf_toml`, configured
///    with `program_config`
/// 5. Upon completion, the compilation status is set to `Success`
///
/// Note that this function assumes it runs in isolation, and as such at the beginning resets
/// any lingering pipelines that have `CompilingRust` status to `SqlCompiled`. This recovers from
/// if the compiler was interrupted (e.g., it was unexpectedly terminated) or a database
/// operation failed.
///
/// Returns with `Ok(true)` if there was an attempt to compile the Rust of a pipeline.
/// It does not necessarily mean the compilation was a success.
/// Returns with `Ok(false)` if no pipeline is found for which to compile Rust.
/// Returns with `Err(...)` if a database operation fails, e.g., due to:
/// - The pipeline no longer exists
/// - The pipeline program is detected to be updated (it became outdated)
/// - The database cannot be reached
async fn attempt_end_to_end_rust_compilation(
    worker_id: usize,
    total_workers: usize,
    common_config: &CommonConfig,
    config: &CompilerConfig,
    db: Arc<Mutex<StoragePostgres>>,
) -> Result<bool, DBError> {
    trace!("Rust worker {worker_id}: Performing Rust compilation...");

    // (1) Reset any pipeline which is `CompilingRust` back to `SqlCompiled`
    // Only the owner worker (by modulo) resets stuck CompilingRust pipelines to SqlCompiled
    // This ensures that only the assigned worker for a pipeline can reset its status
    db.lock()
        .await
        .clear_ongoing_rust_compilation_for_worker(
            &common_config.platform_version,
            worker_id,
            total_workers,
        )
        .await?;

    // (2) Find pipeline which needs Rust compilation
    // Atomically claim the next pipeline for Rust compilation using modulo sharding
    let Some((tenant_id, pipeline)) = db
        .lock()
        .await
        .get_next_rust_compilation(&common_config.platform_version, worker_id, total_workers)
        .await?
    else {
        trace!("No pipeline found which needs Rust compilation");
        return Ok(false);
    };

    // (3) Update database that Rust compilation is ongoing
    db.lock()
        .await
        .transit_program_status_to_compiling_rust(tenant_id, pipeline.id, pipeline.program_version)
        .await?;

    // (4) Perform Rust compilation
    let compilation_result = perform_rust_compilation(
        common_config,
        config,
        Some(db.clone()),
        tenant_id,
        pipeline.id,
        Some(pipeline.name.clone()),
        &pipeline.platform_version,
        pipeline.program_version,
        &pipeline.program_config,
        &pipeline.program_info,
        &pipeline.udf_rust,
        &pipeline.udf_toml,
    )
    .await;

    // (5) Update database that Rust compilation is finished
    match compilation_result {
        Ok(RustCompilationResult {
            source_checksum,
            binary_integrity_checksum,
            binary_size,
            program_info_integrity_checksum,
            profile,
            duration,
            rustc_result,
        }) => {
            info!(
                pipeline_id = %pipeline.id,
                pipeline = %pipeline.name,
                "Rust compilation success (program version: {}) (took {:.2}s; source checksum: {}; binary integrity checksum: {}, size: {} bytes, program info integrity checksum: {}, profile: {})",
                pipeline.program_version,
                duration.as_secs_f64(),
                truncate_sha256_checksum(&source_checksum),
                truncate_sha256_checksum(&binary_integrity_checksum),
                binary_size,
                truncate_sha256_checksum(&program_info_integrity_checksum),
                profile
            );
            db.lock()
                .await
                .transit_program_status_to_success(
                    tenant_id,
                    pipeline.id,
                    pipeline.program_version,
                    &rustc_result,
                    &source_checksum,
                    &binary_integrity_checksum,
                    &program_info_integrity_checksum,
                )
                .await?;
        }
        Err(e) => match e {
            RustCompilationError::NoLongerExists => {
                info!(
                    pipeline_id = %pipeline.id,
                    pipeline = %pipeline.name,
                    "Rust compilation canceled: pipeline no longer exists"
                );
            }
            RustCompilationError::Outdated => {
                info!(
                    pipeline_id = %pipeline.id,
                    pipeline = %pipeline.name,
                    "Rust compilation canceled: program version {} is outdated",
                    pipeline.program_version
                );
            }
            RustCompilationError::TerminatedBySignal => {
                error!(
                    pipeline_id = %pipeline.id,
                    pipeline = %pipeline.name,
                    "Rust compilation interrupted: compilation process was terminated by a signal (program version: {})",
                    pipeline.program_version
                );
            }
            RustCompilationError::RustError(compilation_info) => {
                db.lock()
                    .await
                    .transit_program_status_to_rust_error(
                        tenant_id,
                        pipeline.id,
                        pipeline.program_version,
                        &compilation_info,
                    )
                    .await?;
                info!(
                    pipeline_id = %pipeline.id,
                    pipeline = %pipeline.name,
                    "Rust compilation failed due to Rust error (program version: {})",
                    pipeline.program_version
                );
            }
            RustCompilationError::SystemError(internal_system_error) => {
                db.lock()
                    .await
                    .transit_program_status_to_system_error(
                        tenant_id,
                        pipeline.id,
                        pipeline.program_version,
                        &internal_system_error,
                    )
                    .await?;
                error!(
                    pipeline_id = %pipeline.id,
                    pipeline = %pipeline.name,
                    "Rust compilation failed due to system error (program version: {}): {internal_system_error}",
                    pipeline.program_version
                );
            }
            RustCompilationError::FileUploadError(upload_error) => {
                db.lock()
                    .await
                    .transit_program_status_to_system_error(
                        tenant_id,
                        pipeline.id,
                        pipeline.program_version,
                        &upload_error,
                    )
                    .await?;
                error!(
                    pipeline_id = %pipeline.id,
                    pipeline = %pipeline.name,
                    "Rust compilation failed due to binary upload error (program version: {}): {upload_error}",
                    pipeline.program_version
                );
            }
        },
    }
    Ok(true)
}

/// Calculates sha256 checksum across the fields.
/// It is calculated using: `H(H(field1) || H(field2) || ...)`.
/// `H(x)` is the hashing function. `||` represents concatenation.
/// The order of the fields is always the same.
#[allow(clippy::too_many_arguments)]
fn calculate_source_checksum(
    platform_version: &str,
    runtime_version: &RuntimeSelector,
    profile: &CompilationProfile,
    main_rust: &str,
    udf_stubs: &str,
    udf_rust: &str,
    udf_toml: &str,
) -> String {
    let mut hasher = sha::Sha256::new();
    let profile = profile.to_string();
    let to_hash = vec![
        ("platform_version", platform_version.as_bytes()),
        ("runtime_version", runtime_version.as_bytes()),
        ("profile", profile.as_bytes()),
        ("program_info.main_rust", main_rust.as_bytes()),
        ("program_info.udf_stubs", udf_stubs.as_bytes()),
        ("udf_rust", udf_rust.as_bytes()),
        ("udf_toml", udf_toml.as_bytes()),
    ];

    for (name, data) in to_hash {
        let checksum = sha256(data);
        hasher.update(&checksum);
        trace!(
            "Rust compilation: checksum of {name}: {}",
            hex::encode(checksum)
        );
    }
    hex::encode(hasher.finish())
}

/// Metadata for a compiler output file (binary or program info)
#[derive(Debug, Clone)]
struct FileUploadMetadata {
    pipeline_id: PipelineId,
    program_version: Version,
    source_checksum: String,
    integrity_checksum: String,
}

impl FileUploadMetadata {
    /// Generate filename for the binary
    fn binary_filename(&self) -> String {
        pipeline_binary_filename(
            &self.pipeline_id,
            self.program_version,
            &self.source_checksum,
            &self.integrity_checksum,
        )
    }

    /// Generate filename for the program info file
    fn program_info_filename(&self) -> String {
        program_info_filename(
            &self.pipeline_id,
            self.program_version,
            &self.source_checksum,
            &self.integrity_checksum,
        )
    }
}

/// Binary delivery mode for compiled pipeline binaries.
#[derive(Debug, Clone)]
pub enum FileDeliveryMode {
    /// Copy binary to local filesystem destination
    LocalCopy,
    /// Upload binary to HTTP endpoint
    HttpUpload {
        endpoint: String,
        timeout_secs: u64,
        max_retries: u32,
        retry_delay_ms: u64,
    },
}

impl FileDeliveryMode {
    /// Create binary delivery mode from configuration
    pub fn from_config(config: &CompilerConfig) -> Self {
        match &config.binary_upload_endpoint {
            Some(endpoint) => Self::HttpUpload {
                endpoint: endpoint.clone(),
                timeout_secs: config.binary_upload_timeout_secs,
                max_retries: config.binary_upload_max_retries,
                retry_delay_ms: config.binary_upload_retry_delay_ms,
            },
            None => Self::LocalCopy,
        }
    }
}

/// Rust compilation possible error outcomes.
#[derive(Debug)]
pub enum RustCompilationError {
    /// In the meanwhile the pipeline was deleted, as such the Rust
    /// compilation is no longer useful.
    NoLongerExists,
    /// In the meanwhile the pipeline was already updated, as such the
    /// Rust compilation is outdated and no longer useful.
    Outdated,
    /// The Rust compilation process was terminated by a signal.
    /// This can happen for instance when the compiler server is terminated by signal,
    /// and processes started by it are first terminated before itself. The signal is likely not
    /// related to the program itself inherently being unable to compile, nor the compiler
    /// server reaching an inconsistent state. As such, retrying is the desired
    /// behavior rather than declaring failure to compile the specific program.
    TerminatedBySignal,
    /// Rust compilation was unable to be performed.
    /// This is either due to:
    /// - A check failed when setting up the working directory in which
    ///   the Rust compiler will be called
    /// - The Rust compiler call failed and returned an error
    ///   (e.g., a syntax error in the SQL-generated Rust or
    ///   the user-provided Rust implementation of UDFs).
    RustError(RustCompilationInfo),
    /// General system problem occurred (e.g., I/O error)
    SystemError(String),
    /// File upload (binary or program info) to HTTP endpoint failed
    FileUploadError(String),
}

/// Utility errors are system errors during Rust compilation.
impl From<UtilError> for RustCompilationError {
    fn from(value: UtilError) -> Self {
        RustCompilationError::SystemError(value.to_string())
    }
}

/// Delivers the compiled binary according to the configured delivery mode.
///
/// For local copy mode, the binary is copied to the pipeline-binaries directory.
/// For HTTP upload mode, the binary is uploaded to the configured endpoint as multipart/form-data.
///
/// Returns the final binary location/identifier.
async fn deliver_binary(
    common_config: &CommonConfig,
    delivery_mode: &FileDeliveryMode,
    source_file_path: &Path,
    metadata: &FileUploadMetadata,
    pipeline_binaries_dir: &Path,
) -> Result<String, RustCompilationError> {
    match delivery_mode {
        FileDeliveryMode::LocalCopy => {
            // Original behavior: copy to local directory
            let target_file_path = pipeline_binaries_dir.join(metadata.binary_filename());

            // Copy binary from Cargo target profile directory to the pipeline-binaries directory
            copy_file(source_file_path, &target_file_path).await?;

            Ok(target_file_path.to_string_lossy().to_string())
        }
        FileDeliveryMode::HttpUpload { .. } => {
            // New behavior: upload to HTTP endpoint with retries
            upload_binary_to_endpoint_with_retries(
                common_config,
                delivery_mode,
                source_file_path,
                metadata,
            )
            .await
        }
    }
}

async fn deliver_program_info(
    common_config: &CommonConfig,
    delivery_mode: &FileDeliveryMode,
    program_info: &str,
    metadata: &FileUploadMetadata,
    pipeline_binaries_dir: &Path,
) -> Result<(), RustCompilationError> {
    match delivery_mode {
        FileDeliveryMode::LocalCopy => {
            // Original behavior: copy to local directory
            let target_file_path = pipeline_binaries_dir.join(metadata.program_info_filename());

            write_file(program_info.as_bytes(), &target_file_path).await?;

            Ok(())
        }
        FileDeliveryMode::HttpUpload { .. } => {
            // New behavior: upload to HTTP endpoint with retries
            upload_program_info_to_endpoint_with_retries(
                common_config,
                delivery_mode,
                program_info,
                metadata,
            )
            .await?;
            Ok(())
        }
    }
}

/// Uploads the compiled binary to an HTTP endpoint with retry logic.
async fn upload_binary_to_endpoint_with_retries(
    common_config: &CommonConfig,
    delivery_mode: &FileDeliveryMode,
    binary_path: &Path,
    metadata: &FileUploadMetadata,
) -> Result<String, RustCompilationError> {
    let mut attempts = 0;

    let (endpoint, timeout_secs, max_retries, mut retry_delay_ms) = match delivery_mode {
        FileDeliveryMode::HttpUpload {
            endpoint,
            timeout_secs,
            max_retries,
            retry_delay_ms,
        } => (endpoint, *timeout_secs, *max_retries, *retry_delay_ms),
        _ => {
            return Err(RustCompilationError::SystemError(
                "Invalid delivery mode for HTTP upload".to_string(),
            ))
        }
    };

    loop {
        attempts += 1;

        match upload_binary_to_endpoint(
            common_config,
            endpoint,
            timeout_secs,
            binary_path,
            metadata,
        )
        .await
        {
            Ok(result) => {
                if attempts >= 1 {
                    info!(
                        pipeline_id = %metadata.pipeline_id,
                        pipeline = "N/A",
                        "Binary upload succeeded on attempt {} (program version: {})",
                        attempts,
                        metadata.program_version
                    );
                }
                return Ok(result);
            }
            Err(e) => {
                if attempts > max_retries {
                    error!(
                        pipeline_id = %metadata.pipeline_id,
                        pipeline = "N/A",
                        "Binary upload failed after {} attempts (program version: {}): {:?}",
                        attempts,
                        metadata.program_version,
                        e
                    );
                    return Err(e);
                }

                warn!(
                    pipeline_id = %metadata.pipeline_id,
                    pipeline = "N/A",
                    "Binary upload attempt {} failed (program version: {}): {:?}. Retrying in {}ms...",
                    attempts,
                    metadata.program_version,
                    e,
                    retry_delay_ms
                );

                // Wait before retrying with exponential backoff
                sleep(Duration::from_millis(retry_delay_ms)).await;
                retry_delay_ms *= 2; // Double the delay for next attempt
            }
        }
    }
}

/// Uploads program info to an HTTP endpoint with retry logic.
async fn upload_program_info_to_endpoint_with_retries(
    common_config: &CommonConfig,
    delivery_mode: &FileDeliveryMode,
    program_info: &str,
    metadata: &FileUploadMetadata,
) -> Result<String, RustCompilationError> {
    let mut attempts = 0;

    let (endpoint, timeout_secs, max_retries, mut retry_delay_ms) = match delivery_mode {
        FileDeliveryMode::HttpUpload {
            endpoint,
            timeout_secs,
            max_retries,
            retry_delay_ms,
        } => (endpoint, *timeout_secs, *max_retries, *retry_delay_ms),
        _ => {
            return Err(RustCompilationError::SystemError(
                "Invalid delivery mode for HTTP upload".to_string(),
            ))
        }
    };

    loop {
        attempts += 1;

        match upload_program_info_to_endpoint(
            common_config,
            endpoint,
            timeout_secs,
            program_info,
            metadata,
        )
        .await
        {
            Ok(result) => {
                if attempts >= 1 {
                    info!(
                        pipeline_id = %metadata.pipeline_id,
                        pipeline = "N/A",
                        "Program info upload succeeded on attempt {} (program version: {})",
                        attempts,
                        metadata.program_version
                    );
                }
                return Ok(result);
            }
            Err(e) => {
                if attempts > max_retries {
                    error!(
                        pipeline_id = %metadata.pipeline_id,
                        pipeline = "N/A",
                        "Program info upload failed after {} attempts (program version: {}): {:?}",
                        attempts,
                        metadata.program_version,
                        e
                    );
                    return Err(e);
                }

                warn!(
                    pipeline_id = %metadata.pipeline_id,
                    pipeline = "N/A",
                    "Program info upload attempt {} failed (program version: {}): {:?}. Retrying in {}ms...",
                    attempts,
                    metadata.program_version,
                    e,
                    retry_delay_ms
                );

                // Wait before retrying with exponential backoff
                sleep(Duration::from_millis(retry_delay_ms)).await;
                retry_delay_ms *= 2; // Double the delay for next attempt
            }
        }
    }
}

/// Uploads the compiled binary to an HTTP endpoint (single attempt) using streaming.
async fn upload_binary_to_endpoint(
    common_config: &CommonConfig,
    endpoint: &str,
    timeout_secs: u64,
    binary_path: &Path,
    metadata: &FileUploadMetadata,
) -> Result<String, RustCompilationError> {
    // Open binary file for streaming
    let file = fs::File::open(binary_path).await.map_err(|e| {
        RustCompilationError::FileUploadError(format!(
            "Failed to open binary file '{}': {}",
            binary_path.display(),
            e
        ))
    })?;

    // Get file size for Content-Length header
    let file_metadata = file.metadata().await.map_err(|e| {
        RustCompilationError::FileUploadError(format!(
            "Failed to get file metadata '{}': {}",
            binary_path.display(),
            e
        ))
    })?;
    let file_size = file_metadata.len();

    let stream = stream::try_unfold(file, |mut file| async move {
        let mut buffer = vec![0u8; MAX_CHUNK_SIZE_FOR_UPLOAD_BINARY];
        match file.read(&mut buffer).await {
            Ok(0) => Ok(None), // EOF
            Ok(n) => {
                buffer.truncate(n);
                Ok(Some((buffer, file)))
            }
            Err(e) => Err(e),
        }
    });

    let body = reqwest::Body::wrap_stream(stream);

    // Build URL with path parameters
    let url = format!(
        "{}/binary/{}/{}/{}/{}",
        endpoint.trim_end_matches('/'),
        metadata.pipeline_id,
        metadata.program_version,
        metadata.source_checksum,
        metadata.integrity_checksum
    );

    // Use the properly configured HTTP client from common_config
    let client = common_config.reqwest_client().await;

    // Send request with timeout
    let response = client
        .post(&url)
        .header("Content-Type", "application/octet-stream")
        .header("Content-Length", file_size.to_string())
        .body(body)
        .timeout(Duration::from_secs(timeout_secs))
        .send()
        .await
        .map_err(|e| {
            RustCompilationError::FileUploadError(format!(
                "Failed to upload binary to '{}': {} ({})",
                url,
                &e,
                source_error(&e)
            ))
        })?;

    // Check response status
    if !response.status().is_success() {
        let status = response.status();
        let error_text = response
            .text()
            .await
            .unwrap_or_else(|_| "Unknown error".to_string());
        return Err(RustCompilationError::FileUploadError(format!(
            "Binary upload failed with status {}: {}",
            status, error_text
        )));
    }

    // Get response body for any location information
    let response_text = response
        .text()
        .await
        .unwrap_or_else(|_| endpoint.to_string());

    debug!(
        pipeline_id = %metadata.pipeline_id,
        pipeline = "N/A",
        "Successfully uploaded binary (program version: {}) to {}",
        metadata.program_version,
        endpoint
    );

    Ok(response_text)
}

/// Uploads program info JSON file to an HTTP endpoint (single attempt) using streaming.
async fn upload_program_info_to_endpoint(
    common_config: &CommonConfig,
    endpoint: &str,
    timeout_secs: u64,
    program_info: &str,
    metadata: &FileUploadMetadata,
) -> Result<String, RustCompilationError> {
    let url = format!(
        "{}/program_info/{}/{}/{}/{}",
        endpoint.trim_end_matches('/'),
        metadata.pipeline_id,
        metadata.program_version,
        metadata.source_checksum,
        metadata.integrity_checksum
    );

    // Use the properly configured HTTP client from common_config
    let client = common_config.reqwest_client().await;

    // Send request with timeout
    let response = client
        .post(&url)
        .header("Content-Type", "application/json")
        .header("Content-Length", program_info.len().to_string())
        .body(program_info.to_string())
        .timeout(Duration::from_secs(timeout_secs))
        .send()
        .await
        .map_err(|e| {
            RustCompilationError::FileUploadError(format!(
                "Failed to upload program info to '{}': {} ({})",
                url,
                &e,
                source_error(&e)
            ))
        })?;

    // Check response status
    if !response.status().is_success() {
        let status = response.status();
        let error_text = response
            .text()
            .await
            .unwrap_or_else(|_| "Unknown error".to_string());
        return Err(RustCompilationError::FileUploadError(format!(
            "Program info upload failed with status {}: {}",
            status, error_text
        )));
    }

    // Get response body for any location information
    let response_text = response
        .text()
        .await
        .unwrap_or_else(|_| endpoint.to_string());

    debug!(
        pipeline_id = %metadata.pipeline_id,
        pipeline = "N/A",
        "Successfully uploaded program info (program version: {}) to {}",
        metadata.program_version,
        endpoint
    );

    Ok(response_text)
}

pub(super) struct RustCompilationResult {
    pub(super) source_checksum: String,
    pub(super) binary_integrity_checksum: String,
    pub(super) binary_size: usize,
    pub(super) program_info_integrity_checksum: String,
    pub(super) profile: CompilationProfile,
    pub(super) duration: Duration,
    pub(super) rustc_result: RustCompilationInfo,
}

/// Performs the Rust compilation.
///
/// Returns the program binary URL, source checksum, integrity checksum,
/// duration, and compilation information.
#[allow(clippy::too_many_arguments)]
pub async fn perform_rust_compilation(
    common_config: &CommonConfig,
    config: &CompilerConfig,
    db: Option<Arc<Mutex<StoragePostgres>>>,
    tenant_id: TenantId,
    pipeline_id: PipelineId,
    pipeline_name: Option<String>,
    platform_version: &str,
    program_version: Version,
    program_config: &serde_json::Value,
    program_info_value: &Option<serde_json::Value>,
    udf_rust: &str,
    udf_toml: &str,
) -> Result<RustCompilationResult, RustCompilationError> {
    let start = Instant::now();

    // These must always be the same, the Rust compiler should never pick up
    // a pipeline program which is not of its current platform version.
    if common_config.platform_version != platform_version {
        return Err(RustCompilationError::SystemError(format!(
            "Platform version {platform_version} is not equal to current {}",
            common_config.platform_version
        )));
    }

    // Program configuration
    let program_config = validate_program_config(program_config, true).map_err(|error| {
        RustCompilationError::SystemError(formatdoc! {"
                The program configuration:
                {program_config:#}

                ... is not valid due to: {error}.

                This indicates a backward-incompatible platform upgrade occurred.
                Update the 'program_config' field of the pipeline to resolve this.
            "})
    })?;
    let runtime_selector = program_config.runtime_version();
    assert!(has_unstable_feature("runtime_version") || runtime_selector.is_platform());
    let pipeline_name_for_log = pipeline_name.clone().unwrap_or_else(|| "N/A".to_string());
    info!(
        pipeline_id = %pipeline_id,
        pipeline = pipeline_name_for_log.as_str(),
        "Rust compilation started (program version: {}{})",
        program_version,
        if !runtime_selector.is_platform() {
            format!(", runtime version: {runtime_selector}")
        } else {
            "".to_string()
        }
    );

    // Program info
    let program_info_value = match program_info_value {
        None => {
            return Err(RustCompilationError::SystemError(
                "Unable to Rust compile pipeline program because its program information is missing".to_string()
            ));
        }
        Some(program_info_value) => program_info_value.clone(),
    };
    let program_info = validate_program_info(&program_info_value).map_err(|error| {
        RustCompilationError::SystemError(formatdoc! {"
                The program information:
                {program_info_value:#}

                ... is not valid due to: {error}.

                This indicates a backward-incompatible platform upgrade occurred.
                Recompile the pipeline to resolve this.
            "})
    })?;
    let main_rust = program_info.main_rust.clone();
    let udf_stubs = program_info.udf_stubs.clone();

    // Compilation profile is the one specified by the program configuration,
    // and if it is not, defaults to the one provided in the compiler arguments.
    let profile = program_config
        .profile
        .clone()
        .unwrap_or(config.compilation_profile.clone());

    // Calculate source checksum across fields
    let source_checksum = calculate_source_checksum(
        platform_version,
        &runtime_selector,
        &profile,
        &main_rust,
        &udf_stubs,
        udf_rust,
        udf_toml,
    );
    trace!("Rust compilation: calculated source checksum: {source_checksum}");

    // Prepare the workspace for the compilation of this specific pipeline
    prepare_workspace(
        config,
        &runtime_selector,
        pipeline_id,
        &main_rust,
        udf_rust,
        udf_toml,
    )
    .await?;

    // Perform the compilation in the workspace
    let (compilation_info, binary_integrity_checksum, binary_size, program_info_integrity_checksum) =
        call_compiler(
            common_config,
            db,
            tenant_id,
            pipeline_id,
            pipeline_name.clone(),
            program_version,
            config,
            &source_checksum,
            &profile,
            &runtime_selector,
            &program_info.to_pipeline_config_program_info(),
        )
        .await?;

    Ok(RustCompilationResult {
        source_checksum,
        binary_integrity_checksum: binary_integrity_checksum.clone(),
        binary_size,
        program_info_integrity_checksum,
        profile,
        duration: start.elapsed(),
        rustc_result: compilation_info,
    })
}

/// Checkout the requested runtime version.
async fn checkout_runtime_version(
    repo_location: &PathBuf,
    requested_runtime_version: &RuntimeSelector,
    platform_sources: &Path,
) -> Result<(), RustCompilationError> {
    assert!(
        has_unstable_feature("runtime_version"),
        "This code-path is only enabled with the `runtime_version` feature."
    );

    async fn checkout(
        repo_location: &PathBuf,
        requested_runtime_version: &RuntimeSelector,
    ) -> Result<std::process::Output, std::io::Error> {
        Command::new("git")
            .current_dir(repo_location)
            .args([
                "checkout",
                "--progress",
                "--force",
                requested_runtime_version.as_commitish(),
            ])
            .output()
            .await
    }

    async fn fetch(
        repo_location: &PathBuf,
        requested_runtime_version: &RuntimeSelector,
    ) -> Result<std::process::Output, std::io::Error> {
        Command::new("git")
            .current_dir(repo_location)
            .args([
                "-c",
                "protocol.version=2",
                "fetch",
                "--prune",
                "--tags",
                "https://github.com/feldera/feldera.git",
                requested_runtime_version.as_commitish(),
            ])
            .output()
            .await
    }

    async fn clone(repo_location: &Path) -> Result<std::process::Output, std::io::Error> {
        Command::new("git")
            .args([
                "-c",
                "protocol.version=2",
                "clone",
                "https://github.com/feldera/feldera.git",
                &repo_location.to_string_lossy(),
            ])
            .output()
            .await
    }

    async fn is_valid_git_repo(repo_location: &PathBuf) -> bool {
        match Command::new("git")
            .current_dir(repo_location)
            .args(["rev-list", "--max-parents=0", "HEAD"])
            .output()
            .await
        {
            Ok(output) => {
                output.status.success()
                    && String::from_utf8_lossy(&output.stdout).trim()
                        == "de8879fbda0c9e9392e3b94064c683a1b4bae216"
            }
            Err(_e) => false,
        }
    }

    if repo_location.exists() && !is_valid_git_repo(repo_location).await {
        debug!(
            "Removing invalid git repository at {}",
            repo_location.display()
        );
        let _r = tokio::fs::remove_dir_all(&repo_location).await;
    }

    if !repo_location.exists() {
        match clone(repo_location).await {
            Ok(output) => {
                if !output.status.success() {
                    return Err(RustCompilationError::SystemError(format!("Unable to clone latest runtime version for '{requested_runtime_version}' for compilation.\n`git clone` failed with exit code: {}\nstderr:\n{}\nstdout:\n{}",
                        output.status.code().unwrap_or(-1),
                        String::from_utf8_lossy(&output.stderr),
                        String::from_utf8_lossy(&output.stdout))));
                }
            }
            Err(e) => return Err(RustCompilationError::SystemError(format!(
                "Unable to clone repo for runtime version override to '{requested_runtime_version}' for compilation, `git clone` failed: {e}",
            ))),
        }
    }

    // Link enterprise platform sources into the feldera-checkout repository.
    //
    // This is necessary because the crates depend on some crates within the runtime source tree,
    // and so they can't use the platform crates for that otherwise cargo complains
    // https://github.com/rust-lang/cargo/issues/8639
    if cfg!(feature = "feldera-enterprise") {
        // Helper function to remove directory if it exists, ignore NotFound errors
        async fn remove_if_exists(path: &std::path::Path) -> Result<(), RustCompilationError> {
            let _r = tokio::fs::remove_dir_all(path).await;
            let _r = tokio::fs::remove_file(path).await;
            Ok(())
        }

        pub async fn symlink_dir(
            original: impl AsRef<Path>,
            link: impl AsRef<Path>,
        ) -> std::io::Result<()> {
            let original = original.as_ref();
            let link = link.as_ref();
            #[cfg(windows)]
            {
                fs::symlink_dir(original, link).await
            }
            #[cfg(unix)]
            {
                fs::symlink(original, link).await
            }
            #[cfg(not(any(windows, unix)))]
            {
                compile_error!("Unsupported platform for symlinks")
            }
        }

        for cr in ["dbsp-enterprise", "sync-checkpoint"] {
            remove_if_exists(&repo_location.join("crates").join(cr)).await?;
            symlink_dir(
                platform_sources.join("crates").join(cr),
                repo_location.join("crates").join(cr),
            ).await.map_err(|e| {
                RustCompilationError::SystemError(format!(
                    "Unable to symlink {cr} crate in runtime version override to '{requested_runtime_version}' for compilation: {e}",
                ))
            })?;
        }
    }

    // Try to checkout the requested runtime version, if it fails, try to sync with git repository and checkout again.
    match checkout(repo_location, requested_runtime_version).await {
        Ok(output) => {
            if output.status.success() {
                Ok(())
            } else {
                debug!("Failed to checkout requested runtime, trying to sync with latest git repository.");
                let fetch_result = fetch(repo_location, requested_runtime_version).await.map_err(|e| {
                    RustCompilationError::SystemError(format!("Unable to switch runtime version to '{requested_runtime_version}' for compilation, `git fetch` failed: {e}"))})?;
                if !fetch_result.status.success() {
                    return Err(RustCompilationError::SystemError(format!("Unable to fetch latest runtime version for '{requested_runtime_version}' for compilation.\nGit command failed with exit code: {}\nstderr:\n{}\nstdout:\n{}",
                        fetch_result.status.code().unwrap_or(-1),
                        String::from_utf8_lossy(&fetch_result.stderr),
                        String::from_utf8_lossy(&fetch_result.stdout))));
                }

                let output = checkout(repo_location, requested_runtime_version)
                        .await
                        .map_err(|e| {
                            RustCompilationError::SystemError(format!("Unable to switch runtime version to '{requested_runtime_version}' for compilation, `git checkout` failed after fetch: {e}"))})?;
                if !output.status.success() {
                    return Err(RustCompilationError::SystemError(format!(
                        "Failed to checkout requested runtime version (check that the given SHA or version tag exists in the feldera/feldera repository): '{requested_runtime_version}'.\nGit command failed with exit code: {}\nstderr:\n{}\nstdout:\n{}",
                        output.status.code().unwrap_or(-1),
                        String::from_utf8_lossy(&output.stderr),
                        String::from_utf8_lossy(&output.stdout)
                    )));
                }
                Ok(())
            }
        }
        Err(e) => Err(RustCompilationError::SystemError(format!(
            "Unable to switch runtime version to '{requested_runtime_version}' for compilation, `git checkout` failed: {e}",
        ))),
    }
}

/// We need to pass the actual revision SHA to the binary for verification,
/// so in case we have a version tag, we need to convert it to the sha.
pub async fn resolve_runtime_sha(
    runtime_version: &RuntimeSelector,
    config: &CompilerConfig,
) -> Result<String, RustCompilationError> {
    match runtime_version {
        RuntimeSelector::Sha(sha) => Ok(sha.clone()),
        RuntimeSelector::Platform(platform_sha) => Ok(platform_sha.clone()),
        RuntimeSelector::Version(version) => {
            let repo_location = runtime_version.runtime_sources(config);
            match Command::new("git")
                .current_dir(&repo_location)
                .args([
                    "-c",
                    "protocol.version=2",
                    "rev-parse",
                    version,
                ])
                .output()
                .await
            {
                Ok(output) => {
                    if !output.status.success() {
                        return Err(RustCompilationError::SystemError(format!(
                            "Failed to determine commit SHA for '{runtime_version}'.\nGit command failed with exit code: {}\nstderr:\n{}\nstdout:\n{}",
                            output.status.code().unwrap_or(-1),
                            String::from_utf8_lossy(&output.stderr),
                            String::from_utf8_lossy(&output.stdout)
                        )));
                    }
                    Ok(String::from_utf8_lossy(&output.stdout).trim().to_string())
                }
                Err(e) => Err(RustCompilationError::SystemError(format!(
                    "Failed to determine commit SHA for '{runtime_version}', `git rev-parse` failed: {e}",
                ))),
            }
        }
    }
}

/// Prepare the workspace for compilation of the specific pipeline.
/// This involves extracting the results of the prior SQL compilation and
/// configuring the workspace itself (e.g., its Cargo.toml and Cargo.lock).
async fn prepare_workspace(
    config: &CompilerConfig,
    requested_runtime_version: &RuntimeSelector,
    pipeline_id: PipelineId,
    main_rust: &str,
    udf_rust: &str,
    udf_toml: &str,
) -> Result<(), RustCompilationError> {
    // Workspace directory
    let workspace_dir = config.working_dir().join("rust-compilation");
    create_dir_if_not_exists(&workspace_dir).await?;

    ////////////////////////////////
    // PHASE 1: EXTRACT AND MODIFY
    //
    // Extract the result from the SQL compilation, and make modifications:
    // (1) Save list of crates generated for later cleanup
    // (2) Set the user-provided UDF code and integrate the UDF dependencies in Cargo.toml
    //
    // The modifications are made such that they are accounted for in the copy logic
    // in the second phase (in particular, checksums which determine whether
    // a copy needs to occur).

    // Extract the Rust generated by the SQL compilation
    let extract_dir = workspace_dir.join(".temp-extract");
    recreate_dir(&extract_dir).await?;
    decode_string_as_dir(main_rust, &extract_dir)?;

    let crates_dir = extract_dir.join("crates");

    // List all crates in extracted
    let mut list_crates = vec![];
    for (path, name, is_file) in DirectoryContent::new(&crates_dir).await?.content {
        if !is_file {
            list_crates.push((path, name));
        } else if name != "Cargo.toml" {
            return Err(RustCompilationError::SystemError(format!(
                "Unexpected directory entry: '{}'",
                path.to_string_lossy()
            )));
        }
    }
    let mut list_crate_names: Vec<String> =
        list_crates.iter().map(|(_, name)| name.clone()).collect();
    list_crate_names.sort();

    // Name of pipeline crates
    let pipeline_main_crate = crate_name_pipeline_main(pipeline_id);
    let pipeline_globals_crate = crate_name_pipeline_globals(pipeline_id);

    // Validate the crates
    if !list_crate_names.contains(&pipeline_main_crate) {
        return Err(RustCompilationError::SystemError(format!(
            "Missing required main crate: '{pipeline_main_crate}'"
        )));
    }
    if !list_crate_names.contains(&pipeline_globals_crate) {
        return Err(RustCompilationError::SystemError(format!(
            "Missing required globals crate: '{pipeline_globals_crate}'"
        )));
    }
    for crate_name in &list_crate_names {
        if !crate_name.starts_with("feldera_pipe_") {
            return Err(RustCompilationError::SystemError(format!(
                "Unknown crate '{}' not starting with 'feldera_pipe_'",
                crate_name
            )));
        }
    }

    // Pipeline main crate: crates.json
    // --------------------------------
    // List all crates used for cleanup later.
    let crates_json_path = crates_dir.join(&pipeline_main_crate).join("crates.json");
    recreate_file_with_content(
        &crates_json_path,
        &serde_json::to_string_pretty(&list_crate_names).map_err(|e| {
            RustCompilationError::SystemError(format!("Unable to serialize crates list: {e}"))
        })?,
    )
    .await?;

    // Pipeline globals crate: src/udf.rs
    // ----------------------------------
    // Copy over the user-defined UDF code, overriding the pre-existing content.
    let udf_rs_path = crates_dir
        .join(&pipeline_globals_crate)
        .join("src")
        .join("udf.rs");
    recreate_file_with_content(&udf_rs_path, udf_rust).await?;

    // Pipeline globals crate: Cargo.toml
    // ----------------------------------
    // Copy over the user-defined UDF TOML dependencies.
    let udf_cargo_toml_path = crates_dir.join(&pipeline_globals_crate).join("Cargo.toml");
    let udf_cargo_toml_content = read_file_content(&udf_cargo_toml_path).await?.replace(
        "[dependencies]",
        &formatdoc! {"
            [dependencies]
            # START: UDF dependencies
            {udf_toml}
            # END: UDF dependencies
        "},
    );
    recreate_file_with_content(&udf_cargo_toml_path, &udf_cargo_toml_content).await?;

    ///////////////////////
    // PHASE 2: WORKSPACE
    //
    // (1) Copy over the modified extracted crates
    // (2) Generate the workspace Cargo.toml which targets the specific pipeline
    // (3) Copy over the Cargo.lock

    // Create the crates directory if it does not exist
    let crates_dir = workspace_dir.join("crates");
    create_dir_if_not_exists(&crates_dir).await?;

    // Copy over the content of the crates from the extracted archive
    for (source_crate_path, crate_name) in list_crates {
        let target_crate_path = crates_dir.join(&crate_name);
        create_dir_if_not_exists(&target_crate_path).await?;
        create_dir_if_not_exists(&target_crate_path.join("src")).await?;

        // Crate base: src/, Cargo.toml, crates.json
        let base_content = if crate_name.ends_with("_main") {
            vec![("src", false), ("Cargo.toml", true), ("crates.json", true)]
        } else {
            vec![("src", false), ("Cargo.toml", true)]
        };
        DirectoryContent::new(&source_crate_path)
            .await?
            .validate(&base_content)?;
        DirectoryContent::new(&target_crate_path)
            .await?
            .keep(
                &base_content,
                &vec![("stdout.log", true), ("stderr.log", true)],
            )
            .await?;

        // Crate base: copy files (src/ is the only directory)
        for (filename, is_file) in base_content {
            assert!(is_file || filename == "src");
            if is_file {
                copy_file_if_checksum_differs(
                    &source_crate_path.join(filename),
                    &target_crate_path.join(filename),
                )
                .await?;
            }
        }

        // Crate src/: main.rs, lib.rs, udf.rs, stubs.rs
        let src_content = if crate_name.ends_with("_main") {
            vec![("main.rs", true)]
        } else if crate_name.ends_with("_globals") {
            vec![("lib.rs", true), ("udf.rs", true), ("stubs.rs", true)]
        } else {
            vec![("lib.rs", true)]
        };
        DirectoryContent::new(&source_crate_path.join("src"))
            .await?
            .validate(&src_content)?;
        DirectoryContent::new(&target_crate_path.join("src"))
            .await?
            .keep(&src_content, &vec![])
            .await?;

        // Crate src/: copy content (should only be files)
        for (filename, is_file) in src_content {
            assert!(is_file);
            copy_file_if_checksum_differs(
                &source_crate_path.join("src").join(filename),
                &target_crate_path.join("src").join(filename),
            )
            .await?;
        }
    }

    // Sources: config.compiler_working_directory
    // ---------------------
    // Make sure the runtime version is checked out.
    let runtime_sources = requested_runtime_version.runtime_sources(config);
    if !requested_runtime_version.is_platform() {
        let repo_location = PathBuf::from(&runtime_sources);
        checkout_runtime_version(
            &repo_location,
            requested_runtime_version,
            &PathBuf::from(&config.dbsp_override_path),
        )
        .await?;
    };

    // Workspace: Cargo.toml
    let cargo_toml_source_path = extract_dir.join("Cargo.toml");
    let cargo_toml_target_path = workspace_dir.join("Cargo.toml");
    copy_file(&cargo_toml_source_path, &cargo_toml_target_path).await?;

    // Workspace: Cargo.lock
    // ---------------------
    // Contains all the (indirect and direct) dependencies of the crates besides UDF.
    // The original is copied over each time such that the starting point is the same.
    if requested_runtime_version.is_platform() {
        let cargo_lock_source_path = Path::new(&config.compilation_cargo_lock_path);
        let cargo_lock_target_path = workspace_dir.join("Cargo.lock");
        copy_file(cargo_lock_source_path, &cargo_lock_target_path).await?;
    } else if has_unstable_feature("runtime_version") && !requested_runtime_version.is_platform() {
        let cargo_lock_source_path = Path::new(&runtime_sources).join("Cargo.lock");
        let cargo_lock_target_path = workspace_dir.join("Cargo.lock");
        copy_file(&cargo_lock_source_path, &cargo_lock_target_path).await?;
    }

    // Remove temporary extraction directory
    fs::remove_dir_all(&extract_dir).await.map_err(|e| {
        UtilError::IoError(format!("removing directory '{}'", extract_dir.display()), e)
    })?;

    Ok(())
}

/// Calls the compiler on the project with the provided source checksum and using the compilation profile.
#[allow(clippy::too_many_arguments)]
async fn call_compiler(
    common_config: &CommonConfig,
    db: Option<Arc<Mutex<StoragePostgres>>>,
    tenant_id: TenantId,
    pipeline_id: PipelineId,
    pipeline_name: Option<String>,
    program_version: Version,
    config: &CompilerConfig,
    source_checksum: &str,
    profile: &CompilationProfile,
    runtime_selector: &RuntimeSelector,
    program_info: &PipelineConfigProgramInfo,
) -> Result<(RustCompilationInfo, String, usize, String), RustCompilationError> {
    // Workspace directory
    let workspace_dir = config.working_dir().join("rust-compilation");
    if !workspace_dir.is_dir() {
        return Err(RustCompilationError::SystemError(format!(
            "Expected {} to be a directory and exist, but it is or does not",
            workspace_dir.display()
        )));
    }

    // Pipeline main crate directory
    let pipeline_main_crate_dir = workspace_dir
        .join("crates")
        .join(crate_name_pipeline_main(pipeline_id));
    if !pipeline_main_crate_dir.is_dir() {
        return Err(RustCompilationError::SystemError(format!(
            "Expected {} to be a directory and exist, but it is or does not",
            pipeline_main_crate_dir.display()
        )));
    }

    // Create pipeline-binaries directory if it does not yet exist
    let pipeline_binaries_dir = workspace_dir.join("pipeline-binaries");
    create_dir_if_not_exists(&pipeline_binaries_dir).await?;
    let pipeline_name_for_log = pipeline_name.unwrap_or_else(|| "N/A".to_string());

    // Create file where stdout will be written to
    let stdout_file_path = pipeline_main_crate_dir.join("stdout.log");
    let stdout_file = create_new_file(&stdout_file_path).await?;

    // Create file where stderr will be written to
    let stderr_file_path = pipeline_main_crate_dir.join("stderr.log");
    let stderr_file = create_new_file(&stderr_file_path).await?;

    // By default, the command inherits all environment variables
    // from the originating process. This is fine when the compiler
    // server is run directly using its executable, but in the case
    // of being run from source using `cargo run`, this will result
    // in all kinds of cargo-related environment variables being set
    // such as `CARGO_PKG_VERSION`. This can cause unnecessary
    // recompilation for some dependency crates if they use
    // for example `cargo::rerun-if-env-changed=CARGO_PKG_VERSION`
    // in their build.rs, even if the actual `CARGO_PKG_VERSION`
    // they read does not change (as it is overwritten by cargo).
    //
    // Only a select number of environment variables are inherited
    // explicitly. All other inherited environment variables are cleared.
    let env_path = std::env::var_os("PATH").ok_or(RustCompilationError::SystemError(
        "The PATH environment variable is not set, which is needed to locate `cargo`".to_string(),
    ))?;
    let optional_env_rustflags = std::env::var_os("RUSTFLAGS");

    // Formulate command
    let mut command = Command::new("cargo");

    // get env vars that start with SCCACHE
    let preserved_env_vars: Vec<(String, String)> = std::env::vars()
        .filter(|(key, _)| key.starts_with("SCCACHE"))
        .collect();

    command.env_clear();
    command.env("PATH", env_path);
    if !runtime_selector.is_platform() {
        command.env(
            "FELDERA_RUNTIME_OVERRIDE",
            resolve_runtime_sha(runtime_selector, config).await?,
        );
    }
    if let Some(env_rustflags) = optional_env_rustflags {
        command.env("RUSTFLAGS", env_rustflags);
    }

    if let Some(rustc_wrapper) = std::env::var_os("RUSTC_WRAPPER") {
        command.env("RUSTC_WRAPPER", rustc_wrapper);
    }

    // Preserve CARGO_INCREMENTAL if set, to allow sccache to work properly.
    if let Some(cargo_incremental) = std::env::var_os("CARGO_INCREMENTAL") {
        command.env("CARGO_INCREMENTAL", cargo_incremental);
    }

    // Preserve AWS_PROFILE if set, to allow sccache to use
    // credentials from there.
    // we avoid passing all AWS_* env vars to prevent leaking
    // credentials from the malicious build scripts.
    if let Some(aws_profile) = std::env::var_os("AWS_PROFILE") {
        command.env("AWS_PROFILE", aws_profile);
    }

    for (key, value) in preserved_env_vars {
        command.env(key, value);
    }

    command
        // Set compiler stack size to 20MB (10x the default) to prevent
        // SIGSEGV when the compiler runs out of stack on large programs.
        .env("RUST_MIN_STACK", "20971520")
        .current_dir(&workspace_dir)
        .arg("build")
        .arg("--workspace")
        .arg("--profile")
        .arg(profile.to_string())
        .stdin(Stdio::null())
        .stdout(Stdio::from(stdout_file.into_std().await))
        .stderr(Stdio::from(stderr_file.into_std().await))
        // Setting it to zero sets the process group ID to the PID.
        // This is done to be able to kill any subprocesses that are spawned.
        .process_group(0);

    // Start process
    let mut process = command.spawn().map_err(|e| {
        RustCompilationError::SystemError(
            UtilError::IoError("running 'cargo build'".to_string(), e).to_string(),
        )
    })?;

    // Retrieve process group ID and create a terminator
    // which ends the group when going out of scope.
    let Some(process_group) = process.id() else {
        return Err(RustCompilationError::SystemError(
            "unable to retrieve pid".to_string(),
        ));
    };
    let mut terminator = ProcessGroupTerminator::new("Rust compilation", process_group);

    // Wait for process to exit while regularly checking if the pipeline still exists
    // and has not had its program get updated
    let exit_status = loop {
        match process.try_wait() {
            Ok(exit_status) => match exit_status {
                None => {
                    if let Some(db) = db.clone() {
                        match db
                            .lock()
                            .await
                            .get_pipeline_by_id_for_monitoring(tenant_id, pipeline_id)
                            .await
                        {
                            Ok(pipeline) => {
                                if pipeline.program_version != program_version {
                                    return Err(RustCompilationError::Outdated);
                                }
                            }
                            Err(DBError::UnknownPipeline { .. }) => {
                                return Err(RustCompilationError::NoLongerExists);
                            }
                            Err(e) => {
                                error!(
                                    pipeline_id = %pipeline_id,
                                    pipeline = pipeline_name_for_log.as_str(),
                                    "Rust compilation outdated check failed due to database error: {e}"
                                )
                                // As preemption check failing is not fatal, compilation will continue
                            }
                        }
                    }
                }
                Some(exit_status) => break exit_status,
            },
            Err(e) => {
                return Err(RustCompilationError::SystemError(
                    UtilError::IoError("waiting for 'cargo build'".to_string(), e).to_string(),
                ));
            }
        }
        sleep(COMPILATION_CHECK_INTERVAL).await;
    };

    // Once the process has exited, it is no longer needed to terminate its process group
    terminator.cancel();

    // Check presence of exit status code
    let Some(exit_code) = exit_status.code() else {
        // No exit status code present because the process was terminated by a signal
        return Err(RustCompilationError::TerminatedBySignal);
    };

    // Read stdout and stderr
    let stdout = read_file_content(&stdout_file_path).await?;
    let stderr = read_file_content(&stderr_file_path).await?;
    let compilation_info = RustCompilationInfo {
        exit_code,
        stdout,
        stderr,
    };

    // Compilation is successful if the return exit code is present and zero
    if exit_status.success() {
        // Source file
        let source_file_path = workspace_dir
            .join("target")
            .join(profile.to_target_folder())
            .join(crate_name_pipeline_main(pipeline_id));
        if !source_file_path.is_file() {
            return Err(RustCompilationError::SystemError(format!(
                "Rust compilation was successful but no binary was generated at '{}'",
                source_file_path.display()
            )));
        }

        // Integrity checksum of the source file
        let (file_size, binary_integrity_checksum) = checksum_file(&source_file_path).await?;

        // Create binary metadata
        let binary_metadata = FileUploadMetadata {
            pipeline_id,
            program_version,
            source_checksum: source_checksum.to_string(),
            integrity_checksum: binary_integrity_checksum.clone(),
        };

        // Determine binary delivery mode from configuration
        let delivery_mode = FileDeliveryMode::from_config(config);

        // Deliver binary according to the configured mode
        let _binary_location = deliver_binary(
            common_config,
            &delivery_mode,
            &source_file_path,
            &binary_metadata,
            &pipeline_binaries_dir,
        )
        .await?;

        let program_info_str = serde_json::to_string(&program_info).map_err(|e| {
            RustCompilationError::SystemError(format!("failed to serialize program info: {e}"))
        })?;
        // Integrity checksum of the source file
        let program_info_integrity_checksum = checksum_buffer(program_info_str.as_bytes()).await?;

        // Create binary metadata
        let program_info_metadata = FileUploadMetadata {
            pipeline_id,
            program_version,
            source_checksum: source_checksum.to_string(),
            integrity_checksum: program_info_integrity_checksum.clone(),
        };

        deliver_program_info(
            common_config,
            &delivery_mode,
            &program_info_str,
            &program_info_metadata,
            &pipeline_binaries_dir,
        )
        .await?;

        // Success
        Ok((
            compilation_info,
            binary_integrity_checksum,
            file_size,
            program_info_integrity_checksum,
        ))
    } else {
        Err(RustCompilationError::RustError(compilation_info))
    }
}

/// Rust compilation cleanup possible error outcomes.
enum RustCompilationCleanupError {
    /// Database error occurred (e.g., lost connectivity).
    Database(DBError),
    /// Utility function problem occurred (e.g., I/O error)
    Utility(UtilError),
}

impl From<DBError> for RustCompilationCleanupError {
    fn from(value: DBError) -> Self {
        RustCompilationCleanupError::Database(value)
    }
}

impl From<UtilError> for RustCompilationCleanupError {
    fn from(value: UtilError) -> Self {
        RustCompilationCleanupError::Utility(value)
    }
}

/// Makes the cleanup decision based on the provided name of the file or directory.
fn decide_cleanup(
    name: &str,
    remainder_separator: Option<(bool, char)>,
    deletion: &[String],
) -> CleanupDecision {
    // Ignore anything which does not start with "feldera_pipe_" or "libfeldera_pipe_"
    if !name.starts_with("feldera_pipe_") && !name.starts_with("libfeldera_pipe_") {
        return CleanupDecision::Ignore;
    }

    // The checksum needs to be retrieved from the remainder
    let core_name = match remainder_separator {
        None => name,
        // The remainder split can either be required or optional.
        // If it is required, it will not return the entire remainder as default
        // if it cannot split the remainder by the separator.
        Some((required, separator)) => {
            let remainder_spl: Vec<&str> = name.splitn(2, separator).collect();
            if remainder_spl.len() == 2 {
                remainder_spl[0]
            } else if required {
                // If remainder is required for it to be considered,
                // it will be ignored as it doesn't match the criteria
                return CleanupDecision::Ignore;
            } else {
                name
            }
        }
    };

    // Final decision whether to keep or not is determined whether it is marked for deletion
    if deletion.contains(&core_name.to_string()) {
        CleanupDecision::Remove
    } else {
        CleanupDecision::Keep {
            motivation: core_name.to_string(),
        }
    }
}

/// Cleans up the Rust compilation working directory by removing binaries
/// and compilation artifacts of pipeline programs that no longer exist.
async fn cleanup_rust_compilation(
    config: &CompilerConfig,
    db: Arc<Mutex<StoragePostgres>>,
) -> Result<(), RustCompilationCleanupError> {
    trace!("Performing Rust cleanup...");

    // Rust compilation directory
    let rust_compilation_dir = config.working_dir().join("rust-compilation");
    if !rust_compilation_dir.exists() {
        return Ok(());
    }

    ///////////////////////////////
    // PHASE 1: PIPELINE BINARIES
    // Only the latest version binaries of successfully compiled pipeline
    // programs are retained. Older version binaries are deleted.

    // Retrieve existing pipeline programs
    // (pipeline_id, program_version, program_binary_source_checksum, program_binary_integrity_checksum, program_info_integrity_checksum)
    let existing_pipeline_programs = db
        .lock()
        .await
        .list_pipeline_programs_across_all_tenants()
        .await?;

    // Clean up pipeline binaries
    // These are not subject to the retention period.
    let pipeline_binaries_dir = rust_compilation_dir.join("pipeline-binaries");
    let valid_pipeline_binary_filenames: Vec<String> = existing_pipeline_programs
        .iter()
        .map(
            |(
                pipeline_id,
                program_version,
                source_checksum,
                binary_integrity_checksum,
                _program_info_integrity_checksum,
            )| {
                if let (Some(source_checksum), Some(binary_integrity_checksum)) =
                    (source_checksum, binary_integrity_checksum)
                {
                    pipeline_binary_filename(
                        pipeline_id,
                        *program_version,
                        source_checksum,
                        binary_integrity_checksum,
                    )
                } else {
                    // this is when program status is 'CompilingRust'
                    // when compiler server is uploading the binary over http, the status is still 'CompilingRust'
                    // we don't want to delete such binaries, hence we keep a more relaxed matching pattern
                    // if either of the checksums is missing
                    format!("pipeline_{pipeline_id}_v{program_version}_")
                }
            },
        )
        .collect();

    let valid_program_info_filenames: Vec<String> = existing_pipeline_programs
        .iter()
        .map(
            |(
                pipeline_id,
                program_version,
                source_checksum,
                _binary_integrity_checksum,
                program_info_integrity_checksum,
            )| {
                if let (Some(source_checksum), Some(program_info_integrity_checksum)) =
                    (source_checksum, program_info_integrity_checksum)
                {
                    program_info_filename(
                        pipeline_id,
                        *program_version,
                        source_checksum,
                        program_info_integrity_checksum,
                    )
                } else {
                    format!("program_info_{pipeline_id}_v{program_version}_")
                }
            },
        )
        .collect();

    if pipeline_binaries_dir.is_dir() {
        cleanup_specific_files(
            "Rust compilation pipeline binaries",
            &pipeline_binaries_dir,
            Arc::new(
                move |filename: &str, _metadata: Option<std::fs::Metadata>| {
                    if filename.starts_with("pipeline_") {
                        if valid_pipeline_binary_filenames
                            .iter()
                            .any(|f| filename.starts_with(f))
                        {
                            CleanupDecision::Keep {
                                motivation: filename.to_string(),
                            }
                        } else {
                            CleanupDecision::Remove
                        }
                    } else if filename.starts_with("program_info_") {
                        if valid_program_info_filenames
                            .iter()
                            .any(|f| filename.starts_with(f))
                        {
                            CleanupDecision::Keep {
                                motivation: filename.to_string(),
                            }
                        } else {
                            CleanupDecision::Remove
                        }
                    } else {
                        CleanupDecision::Ignore
                    }
                },
            ),
            true,
            false,
        )
        .await?;
    }

    ///////////////////////////////////
    // PHASE 2: COMPILATION ARTIFACTS
    // Remove the artifacts used during compilation itself, notably
    // the crates and the corresponding files in the target directories.

    // Retrieve all existing pipeline identifiers
    let existing_pipeline_ids: Vec<PipelineId> = db
        .lock()
        .await
        .list_pipeline_ids_across_all_tenants()
        .await?
        .into_iter()
        .map(|(_, pipeline_id)| pipeline_id)
        .collect();

    // Location of the cleanup state file
    let cleanup_state_file_path = rust_compilation_dir.join("target_cleanup.json");

    // Attempt to read cleanup state from file.
    // If it fails, an error is written to log and the cleanup state is wiped.
    // The wipe means that all retention expirations will be reset.
    // The cleanup state includes the artifact names that are planned
    // to be deleted: "<crate>" and "lib<crate>".
    let mut cleanup_state: BTreeMap<String, SystemTime> =
        match read_file_content(&cleanup_state_file_path).await {
            Ok(content) => serde_json::from_str(&content).unwrap_or_else(|e| {
                error!("Unable to deserialize cleanup state due to: {e}");
                BTreeMap::new()
            }),
            Err(e) => {
                if !cleanup_state_file_path.exists() {
                    debug!(
                        "Cleanup state file does not yet exist -- \
                        it will be created at the end of the first cleanup cycle"
                    );
                } else {
                    error!("Unable to read cleanup state file due to: {e}");
                }
                BTreeMap::new()
            }
        };

    // Determine all the crates and their artifact names that are currently in use
    let mut crates_in_use: HashSet<String> = HashSet::new();
    let mut artifacts_in_use: HashSet<String> = HashSet::new();
    let existing_main_crates: Vec<String> = existing_pipeline_ids
        .iter()
        .map(|pipeline_id| crate_name_pipeline_main(*pipeline_id))
        .collect();
    for main_crate_name in existing_main_crates {
        let path_crates_json = rust_compilation_dir
            .join("crates")
            .join(main_crate_name)
            .join("crates.json");
        // Inability to read the `crates.json` occurs if:
        // - The pipeline has not yet been Rust compiled
        // - The Rust compilation preparation was interrupted
        //   before it could be written fully
        //
        // In either case, the crates uniquely belonging to it will not
        // be marked as in-use and thus any of its artifacts will be marked
        // for deletion.
        let content_crates_json = read_file_content(&path_crates_json)
            .await
            .unwrap_or_else(|e| {
                trace!(
                    "Could not read '{}' due to: {e}",
                    path_crates_json.display()
                );
                "[]".to_string()
            });
        let crates_json: Vec<String> =
            serde_json::from_str(&content_crates_json).unwrap_or_else(|e| {
                warn!(
                    "Could not convert '{}' to JSON due to: {e}",
                    content_crates_json
                );
                vec![]
            });
        for crate_name in crates_json {
            crates_in_use.insert(crate_name.clone());
            artifacts_in_use.insert(crate_name.clone());
            artifacts_in_use.insert(format!("lib{crate_name}"));
        }
    }
    debug!(
        "Rust compilation cleanup: there are {} crates listed as in use (thus {} artifacts in use)",
        crates_in_use.len(),
        artifacts_in_use.len()
    );

    // Current timestamp at which the cleanup takes place
    let current_timestamp = SystemTime::now();

    // Remove any artifact which is in use again
    for artifact_name in &artifacts_in_use {
        if cleanup_state.remove(artifact_name).is_some() {
            trace!("Rust compilation cleanup: artifact '{artifact_name}' is in use again")
        }
    }

    // The ones that have been marked longer ago than the retention period will be deleted
    let mut deletion = HashSet::<String>::new();
    for (artifact_name, expiration) in cleanup_state.iter() {
        if current_timestamp
            .duration_since(*expiration)
            .is_ok_and(|duration| duration >= CLEANUP_RETENTION)
        {
            deletion.insert(artifact_name.clone());
            trace!("Rust compilation cleanup: retention for artifact '{}' has expired -- marked for deletion", artifact_name);
        }
    }
    let deletion: Vec<String> = deletion.into_iter().collect();

    // All found artifacts during the decisions, which is used later to update the cleanup state
    let mut found = Vec::<String>::new();

    // (1) Clean up crates directories
    let crates_dir = rust_compilation_dir.join("crates");
    if crates_dir.is_dir() {
        let deletion_clone = deletion.clone();
        found.append(
            &mut cleanup_specific_directories(
                "Rust compilation crates",
                &crates_dir,
                Arc::new(move |name: &str| decide_cleanup(name, None, &deletion_clone)),
                true,
            )
            .await?,
        );
    }

    // (2) For each possible compilation profile, clean up their target artifacts
    for profile in [
        CompilationProfile::Dev,
        CompilationProfile::Unoptimized,
        CompilationProfile::Optimized,
    ] {
        let target_profile_folder = profile.to_target_folder();
        let target_subdir = rust_compilation_dir
            .join("target")
            .join(target_profile_folder);
        if target_subdir.is_dir() {
            // target/<folder>: "<artifact name>.<other>" or else "<artifact name>"
            let deletion_clone = deletion.clone();
            found.append(
                &mut cleanup_specific_files(
                    &format!("Rust compilation target/{target_profile_folder}/"),
                    &target_subdir,
                    Arc::new(move |name: &str, _metadata: Option<std::fs::Metadata>| {
                        decide_cleanup(name, Some((false, '.')), &deletion_clone)
                    }),
                    false,
                    false,
                )
                .await?,
            );

            // target/<folder>/deps: "<artifact name>-<other>"
            let deps_dir = target_subdir.join("deps");
            if deps_dir.is_dir() {
                let deletion_clone = deletion.clone();
                found.append(
                    &mut cleanup_specific_files(
                        &format!("Rust compilation target/{target_profile_folder}/deps"),
                        &deps_dir,
                        Arc::new(move |name: &str, _metadata: Option<std::fs::Metadata>| {
                            decide_cleanup(name, Some((true, '-')), &deletion_clone)
                        }),
                        false,
                        false,
                    )
                    .await?,
                );
            }

            // target/<folder>/.fingerprint: "<artifact name>-<other>"
            let fingerprint_dir = target_subdir.join(".fingerprint");
            if fingerprint_dir.is_dir() {
                let deletion_clone = deletion.clone();
                found.append(
                    &mut cleanup_specific_directories(
                        &format!("Rust compilation target/{target_profile_folder}/.fingerprint"),
                        &fingerprint_dir,
                        Arc::new(move |name: &str| {
                            decide_cleanup(name, Some((true, '-')), &deletion_clone)
                        }),
                        false,
                    )
                    .await?,
                );
            }

            // target/<folder>/incremental: "<artifact name>-<other>"
            let incremental_dir = target_subdir.join("incremental");
            if incremental_dir.is_dir() {
                let deletion_clone = deletion.clone();
                found.append(
                    &mut cleanup_specific_directories(
                        &format!("Rust compilation target/{target_profile_folder}/incremental"),
                        &incremental_dir,
                        Arc::new(move |name: &str| {
                            decide_cleanup(name, Some((true, '-')), &deletion_clone)
                        }),
                        false,
                    )
                    .await?,
                );
            }
        }
    }

    // Any artifact name that was found which:
    // (1) is not an artifact in use
    // (2) AND is not already in the cleanup state
    // ... will be added to the cleanup state.
    let found = HashSet::<String>::from_iter(found.into_iter());
    for artifact_name in found.iter() {
        if !artifacts_in_use.contains(artifact_name) && !cleanup_state.contains_key(artifact_name) {
            let expiration_datetime: DateTime<Utc> = (current_timestamp + CLEANUP_RETENTION).into();
            trace!(
                "Rust compilation cleanup: artifact '{}' will be deleted some time after retention expires (currently: {})",
                artifact_name,
                expiration_datetime.format("%Y-%m-%d %H:%M:%S")
            );
            cleanup_state.insert(artifact_name.clone(), current_timestamp);
        }
    }

    // Remove any artifact in the cleanup state which is no longer found
    let mut artifacts_removed: u64 = 0;
    for artifact_name in cleanup_state.clone().keys() {
        if !found.contains(artifact_name) {
            artifacts_removed += 1;
            cleanup_state.remove(artifact_name);
        }
    }
    if artifacts_removed > 0 {
        info!(
            "Rust compilation cleanup: removed {artifacts_removed} artifacts; {} artifacts remain",
            found.len()
        );
    }

    // Attempt to write cleanup state to file.
    // If it fails, an error is written to log and the cleanup state changes will be lost.
    // The changes being lost means that new retention expirations are not set, and old ones remain.
    match serde_json::to_string_pretty(&cleanup_state) {
        Ok(s) => match recreate_file_with_content(&cleanup_state_file_path, &s).await {
            Ok(()) => {}
            Err(e) => error!("Unable to write cleanup state to file due to: {e}"),
        },
        Err(e) => error!("Unable to serialize cleanup state due to: {e}"),
    }

    Ok(())
}

#[cfg(test)]
mod test {
    use crate::auth::TenantRecord;
    use crate::compiler::rust_compiler::prepare_workspace;
    use crate::compiler::rust_compiler::{calculate_source_checksum, decide_cleanup};
    use crate::compiler::test::{list_content_as_sorted_names, CompilerTest};
    use crate::compiler::util::{
        crate_name_pipeline_globals, crate_name_pipeline_main, read_file_content, CleanupDecision,
    };
    use crate::db::types::program::{CompilationProfile, ProgramStatus, RuntimeSelector};
    use crate::db::types::utils::validate_program_info;
    use std::collections::HashSet;

    /// Tests the calculation of the source checksum based on the input.
    #[tokio::test]
    async fn source_checksum_calculation() {
        // Different input, different source checksum
        let mut seen = HashSet::<String>::new();
        let platform_versions = ["", "v1", "v2"];
        let profiles = [
            CompilationProfile::Dev,
            CompilationProfile::Optimized,
            CompilationProfile::Unoptimized,
        ];
        let content = ["", "a", "aa", "b", "c", "d", "e"];
        for (platform_version, profile, main_rust, udf_stubs, udf_rust, udf_toml) in itertools::iproduct!(
            platform_versions,
            profiles,
            content,
            content,
            content,
            content
        ) {
            let source_checksum = calculate_source_checksum(
                platform_version,
                &RuntimeSelector::default(),
                &profile,
                main_rust,
                udf_stubs,
                udf_rust,
                udf_toml,
            );
            assert!(
                !seen.contains(&source_checksum),
                "checksum {source_checksum} has a duplicate"
            );
            seen.insert(source_checksum);
        }

        // Same input, same source checksum
        let checksum1 = calculate_source_checksum(
            "v0",
            &RuntimeSelector::default(),
            &CompilationProfile::Optimized,
            "main_rust",
            "udf_stubs",
            "udf_rust",
            "udf_toml",
        );
        let checksum2 = calculate_source_checksum(
            "v0",
            &RuntimeSelector::default(),
            &CompilationProfile::Optimized,
            "main_rust",
            "udf_stubs",
            "udf_rust",
            "udf_toml",
        );
        assert_eq!(checksum1, checksum2);
    }

    /// Tests the workspace preparation helper function.
    #[tokio::test]
    async fn workspace_preparation() {
        let test = CompilerTest::new().await;
        let tenant_id = TenantRecord::default().id;

        // Create a pipeline and SQL compile it
        let sql = "CREATE TABLE t1 (v1 INT);";
        let udf_rust = "abc";
        let udf_toml = "def";
        let pipeline_id = test
            .create_pipeline_with_udf(tenant_id, "p1", "v0", sql, udf_rust, udf_toml)
            .await;
        test.sql_compiler_tick().await;
        let pipeline_descr = test.get_pipeline(tenant_id, pipeline_id).await;
        assert_eq!(pipeline_descr.program_status, ProgramStatus::SqlCompiled);
        assert_eq!(pipeline_descr.program_code, sql);
        assert_eq!(pipeline_descr.udf_rust, udf_rust);
        assert_eq!(pipeline_descr.udf_toml, udf_toml);
        let program_info = validate_program_info(&pipeline_descr.program_info.unwrap()).unwrap();

        // Before, the rust-compilation directory will not exist
        assert!(!test.rust_workdir.exists());

        // Prepare the workspace directory using it
        prepare_workspace(
            &test.compiler_config,
            &RuntimeSelector::default(),
            pipeline_id,
            &program_info.main_rust,
            &pipeline_descr.udf_rust,
            &pipeline_descr.udf_toml,
        )
        .await
        .unwrap();

        // Afterward, check for the presence of files/directories and their content

        // Main crate
        let main_crate_name = crate_name_pipeline_main(pipeline_id);
        let main_crate_path = test.rust_workdir.join("crates").join(&main_crate_name);
        assert!(main_crate_path.is_dir());
        assert_eq!(
            list_content_as_sorted_names(&main_crate_path).await,
            vec!["Cargo.toml", "crates.json", "src"]
        );
        assert_eq!(
            list_content_as_sorted_names(&main_crate_path.join("src")).await,
            vec!["main.rs"]
        );
        assert!(
            read_file_content(&main_crate_path.join("src").join("main.rs"))
                .await
                .unwrap()
                .contains("fn main")
        );

        // Globals crate
        let globals_crate_path = test
            .rust_workdir
            .join("crates")
            .join(crate_name_pipeline_globals(pipeline_id));
        assert!(globals_crate_path.is_dir());
        assert_eq!(
            list_content_as_sorted_names(&globals_crate_path).await,
            vec!["Cargo.toml", "src"]
        );
        assert_eq!(
            list_content_as_sorted_names(&globals_crate_path.join("src")).await,
            vec!["lib.rs", "stubs.rs", "udf.rs"]
        );
        assert_eq!(
            read_file_content(&globals_crate_path.join("src").join("stubs.rs"))
                .await
                .unwrap(),
            program_info.udf_stubs
        );
        assert_eq!(
            read_file_content(&globals_crate_path.join("src").join("udf.rs"))
                .await
                .unwrap(),
            pipeline_descr.udf_rust
        );
        assert!(read_file_content(&globals_crate_path.join("Cargo.toml"))
            .await
            .unwrap()
            .contains(&pipeline_descr.udf_toml));

        // Workspace-wide Cargo.toml
        let workspace_toml_file = test.rust_workdir.join("Cargo.toml");
        assert!(workspace_toml_file.is_file());
        assert!(read_file_content(&workspace_toml_file)
            .await
            .unwrap()
            .contains(&format!("members = [\n  \"crates/{main_crate_name}\"\n]")));
    }

    /// Tests the binary delivery mode configuration.
    #[tokio::test]
    async fn binary_delivery_mode_from_config() {
        use crate::compiler::rust_compiler::FileDeliveryMode;
        use crate::config::CompilerConfig;
        use crate::db::types::program::CompilationProfile;

        // Test local copy mode (default)
        let config = CompilerConfig {
            compiler_working_directory: "/tmp".to_string(),
            compilation_profile: CompilationProfile::Dev,
            sql_compiler_path: "test.jar".to_string(),
            sql_compiler_cache_url: "http://test.com".to_string(),
            compilation_cargo_lock_path: "Cargo.lock".to_string(),
            dbsp_override_path: ".".to_string(),
            binary_upload_endpoint: None,
            binary_upload_timeout_secs: 300,
            binary_upload_max_retries: 3,
            binary_upload_retry_delay_ms: 1000,
            precompile: false,
        };

        let delivery_mode = FileDeliveryMode::from_config(&config);
        assert!(matches!(delivery_mode, FileDeliveryMode::LocalCopy));

        // Test HTTP upload mode
        let config = CompilerConfig {
            binary_upload_endpoint: Some("https://example.com".to_string()),
            binary_upload_timeout_secs: 600,
            binary_upload_max_retries: 5,
            binary_upload_retry_delay_ms: 2000,
            ..config
        };

        let delivery_mode = FileDeliveryMode::from_config(&config);
        match delivery_mode {
            FileDeliveryMode::HttpUpload {
                endpoint,
                timeout_secs,
                max_retries,
                retry_delay_ms,
            } => {
                assert_eq!(endpoint, "https://example.com");
                assert_eq!(timeout_secs, 600);
                assert_eq!(max_retries, 5);
                assert_eq!(retry_delay_ms, 2000);
            }
            _ => panic!("Expected HttpUpload mode"),
        }
    }

    /// Tests the cleanup decision helper function.
    #[tokio::test]
    #[rustfmt::skip]
    async fn cleanup_decision_helper() {
        for (i, (name, remainder_separator, deletion, expected)) in [
            // No remainder separator
            ("example", None, vec![], CleanupDecision::Ignore),
            ("feldera_pipe_a", None, vec![], CleanupDecision::Keep { motivation: "feldera_pipe_a".to_string() }),
            ("feldera_pipe_a-b", None, vec![], CleanupDecision::Keep { motivation: "feldera_pipe_a-b".to_string() }),
            ("feldera_pipe_a", None, vec!["feldera_pipe_a".to_string()], CleanupDecision::Remove),
            ("feldera_pipe_a-b", None, vec!["feldera_pipe_a".to_string()], CleanupDecision::Keep { motivation: "feldera_pipe_a-b".to_string() }),
            // Optional remainder separator
            ("example", Some((false, '-')), vec![], CleanupDecision::Ignore),
            ("feldera_pipe_a", Some((false, '-')), vec![], CleanupDecision::Keep { motivation: "feldera_pipe_a".to_string() }),
            ("feldera_pipe_a-b", Some((false, '-')), vec![], CleanupDecision::Keep { motivation: "feldera_pipe_a".to_string() }),
            ("feldera_pipe_a", Some((false, '-')), vec!["feldera_pipe_a".to_string()], CleanupDecision::Remove),
            ("feldera_pipe_a-b", Some((false, '-')), vec!["feldera_pipe_a".to_string()], CleanupDecision::Remove),
            // Mandatory remainder separator
            ("example", Some((true, '-')), vec![], CleanupDecision::Ignore),
            ("feldera_pipe_a", Some((true, '-')), vec![], CleanupDecision::Ignore),
            ("feldera_pipe_a-b", Some((true, '-')), vec![], CleanupDecision::Keep { motivation: "feldera_pipe_a".to_string() }),
            ("feldera_pipe_a", Some((true, '-')), vec!["feldera_pipe_a".to_string()], CleanupDecision::Ignore),
            ("feldera_pipe_a-b", Some((true, '-')), vec!["feldera_pipe_a".to_string()], CleanupDecision::Remove),
        ].iter().enumerate() {
            assert_eq!(
                decide_cleanup(name, *remainder_separator, deletion),
                *expected,
                "test case {i} (zero-based index) fails"
            );
        }
    }
}
