platform_module_management/

lib.rs

//! Console HTTP adapter for the shared Module management kernel.
//! The adapter accepts business changes and returns kernel-owned snapshots or
//! immutable plans; it never reimplements resolution, Cargo, or workspace rules.

mod system_plane;

pub use system_plane::*;

use axum::{
    Json,
    extract::{Path as AxumPath, Query, State},
};
use lenso_module_management::{
    ApproveModuleOperation, MANAGEMENT_HOLDER, ManagementActor, MigrationExecutionMode,
    ModuleEffectAdapter, ModuleEffectAdapterError, ModuleEffectExecution, ModuleEffectOutcome,
    ModulePlanEffect, ModuleRootChange, ServiceDeploymentAction, ServiceInstallationPlan,
    StartReviewedModulePlan, WorkspaceModuleManagement, WorkspaceModuleOperator,
    WorkspaceModuleOperatorError, WorkspaceServiceInstallationManager,
    application_module_lock_digest,
};
use platform_core::{AppContext, AppError, DbPool, ErrorCode, Shutdown, apply_module_migration};
use platform_http::{
    AdminActor, ApiErrorResponse, ApiOpenApiRouter, ErrorResponse, HttpRequestContext,
    OpenApiRouter, routes,
};
use serde::Deserialize;
use serde_json::{Value, json};
use sha2::{Digest as _, Sha256};
use std::{
    collections::BTreeSet,
    fs,
    path::{Path, PathBuf},
    process::Command,
};

pub fn router() -> ApiOpenApiRouter {
    OpenApiRouter::new()
        .routes(routes!(management_snapshot))
        .routes(routes!(preview_change_plan))
        .routes(routes!(start_operation))
        .routes(routes!(get_operation))
        .routes(routes!(get_operation_journal))
        .routes(routes!(apply_operation))
        .routes(routes!(approve_operation))
        .routes(routes!(cancel_operation))
        .routes(routes!(retry_operation))
        .routes(routes!(resume_operation))
        .routes(routes!(service_installation_snapshot))
        .routes(routes!(preview_service_installation))
        .routes(routes!(apply_service_installation_plan))
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct StartOperationBody {
    idempotency_key: String,
    plan: lenso_module_management::ModuleChangePlan,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct RevisionBody {
    expected_revision: u64,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct ApprovalBody {
    expected_revision: u64,
    boundary_id: String,
    reason: String,
    nonce: String,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct ServiceInstallationScope {
    system_id: String,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct PreviewServiceInstallationBody {
    system_id: String,
    environment_id: String,
    change: lenso_module_management::ServiceInstallationChange,
}

#[derive(Debug, Deserialize)]
#[serde(deny_unknown_fields)]
struct ApplyServiceInstallationBody {
    operation_id: String,
    plan: ServiceInstallationPlan,
}

#[utoipa::path(
    get,
    path = "/admin/modules/management",
    operation_id = "admin_modules_management_snapshot",
    tag = "module-management",
    params(("authorization" = String, Header, description = "Service or system bearer token")),
    responses(
        (status = 200, description = "Target-owned Module composition and planning readiness", body = Value, content_type = "application/json"),
        (status = 401, description = "Authentication is required", body = ErrorResponse, content_type = "application/problem+json"),
        (status = 403, description = "Service or system authentication is required", body = ErrorResponse, content_type = "application/problem+json"),
        (status = 500, description = "Module management state cannot be read", body = ErrorResponse, content_type = "application/problem+json"),
    )
)]
#[allow(clippy::result_large_err)]
async fn management_snapshot(
    _admin: AdminActor,
    HttpRequestContext(request_context): HttpRequestContext,
) -> Result<Json<Value>, ApiErrorResponse> {
    let root = std::env::current_dir().map_err(|error| {
        api_error(
            ErrorCode::Internal,
            format!("Module management root is unavailable: {error}"),
            &request_context,
        )
    })?;
    let snapshot = WorkspaceModuleManagement::new(root)
        .snapshot()
        .map_err(|error| {
            api_error(
                ErrorCode::Internal,
                format!("Module management snapshot failed: {error}"),
                &request_context,
            )
        })?;
    serde_json::to_value(snapshot).map(Json).map_err(|error| {
        api_error(
            ErrorCode::Internal,
            format!("Module management snapshot serialization failed: {error}"),
            &request_context,
        )
    })
}

#[utoipa::path(
    post,
    path = "/admin/modules/plans/preview",
    operation_id = "admin_modules_preview_change_plan",
    tag = "module-management",
    params(("authorization" = String, Header, description = "Service or system bearer token")),
    request_body(content = Value, description = "One lenso ModuleRootChange value", content_type = "application/json"),
    responses(
        (status = 200, description = "Immutable complete Module Change Plan", body = Value, content_type = "application/json"),
        (status = 400, description = "Change or planning input is invalid", body = ErrorResponse, content_type = "application/problem+json"),
        (status = 401, description = "Authentication is required", body = ErrorResponse, content_type = "application/problem+json"),
        (status = 403, description = "Service or system authentication is required", body = ErrorResponse, content_type = "application/problem+json"),
        (status = 409, description = "Planning context is not available", body = ErrorResponse, content_type = "application/problem+json"),
    )
)]
#[allow(clippy::result_large_err)]
async fn preview_change_plan(
    _admin: AdminActor,
    HttpRequestContext(request_context): HttpRequestContext,
    Json(body): Json<Value>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let change: ModuleRootChange = serde_json::from_value(body).map_err(|error| {
        api_error(
            ErrorCode::Validation,
            format!("Module change is invalid: {error}"),
            &request_context,
        )
    })?;
    let root = std::env::current_dir().map_err(|error| {
        api_error(
            ErrorCode::Internal,
            format!("Module management root is unavailable: {error}"),
            &request_context,
        )
    })?;
    let plan = WorkspaceModuleManagement::new(root)
        .preview(change, chrono::Utc::now())
        .map_err(|error| {
            let code = if matches!(
                error,
                lenso_module_management::WorkspaceModuleManagementError::PlanningUnavailable
            ) {
                ErrorCode::Conflict
            } else {
                ErrorCode::Validation
            };
            api_error(
                code,
                format!("Module planning failed: {error}"),
                &request_context,
            )
        })?;
    serde_json::to_value(plan).map(Json).map_err(|error| {
        api_error(
            ErrorCode::Internal,
            format!("Module plan serialization failed: {error}"),
            &request_context,
        )
    })
}

#[utoipa::path(post, path = "/admin/modules/operations", operation_id = "admin_modules_start_operation", tag = "module-management", request_body(content = Value, content_type = "application/json"), responses((status = 200, description = "Durable Module operation", body = Value), (status = 400, description = "Reviewed plan is stale or invalid", body = ErrorResponse), (status = 401, description = "Authentication is required", body = ErrorResponse), (status = 403, description = "Management authority is required", body = ErrorResponse), (status = 409, description = "Operation conflicts with target state", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn start_operation(
    admin: AdminActor,
    State(ctx): State<AppContext>,
    HttpRequestContext(request_context): HttpRequestContext,
    Json(body): Json<Value>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let body: StartOperationBody = decode(body, &request_context)?;
    let actor = management_actor(admin);
    let root = management_root(&request_context)?;
    let operator = WorkspaceModuleOperator::new(root, HostModuleEffectAdapter::new(&ctx));
    let operation = operator
        .start(
            &StartReviewedModulePlan {
                idempotency_key: body.idempotency_key,
                plan: body.plan,
            },
            &actor,
            MANAGEMENT_HOLDER,
            chrono::Utc::now(),
        )
        .map_err(|error| operator_error(error, &request_context))?;
    value(operation, &request_context)
}

#[utoipa::path(get, path = "/admin/modules/operations/{operation_id}", operation_id = "admin_modules_get_operation", tag = "module-management", params(("operation_id" = String, Path)), responses((status = 200, description = "Current durable Module operation", body = Value), (status = 404, description = "Operation was not found", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn get_operation(
    admin: AdminActor,
    State(ctx): State<AppContext>,
    HttpRequestContext(request_context): HttpRequestContext,
    AxumPath(operation_id): AxumPath<String>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let _actor = management_actor(admin);
    let operator = WorkspaceModuleOperator::new(
        management_root(&request_context)?,
        HostModuleEffectAdapter::new(&ctx),
    );
    value(
        operator
            .operation(&operation_id)
            .map_err(|error| operator_error(error, &request_context))?,
        &request_context,
    )
}

#[utoipa::path(get, path = "/admin/modules/operations/{operation_id}/journal", operation_id = "admin_modules_get_operation_journal", tag = "module-management", params(("operation_id" = String, Path)), responses((status = 200, description = "Hash-chained Module operation journal", body = Value), (status = 404, description = "Operation was not found", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn get_operation_journal(
    admin: AdminActor,
    State(ctx): State<AppContext>,
    HttpRequestContext(request_context): HttpRequestContext,
    AxumPath(operation_id): AxumPath<String>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let _actor = management_actor(admin);
    let operator = WorkspaceModuleOperator::new(
        management_root(&request_context)?,
        HostModuleEffectAdapter::new(&ctx),
    );
    value(
        operator
            .journal(&operation_id)
            .map_err(|error| operator_error(error, &request_context))?,
        &request_context,
    )
}

#[utoipa::path(post, path = "/admin/modules/operations/{operation_id}/apply", operation_id = "admin_modules_apply_operation", tag = "module-management", params(("operation_id" = String, Path)), responses((status = 200, description = "Applied, blocked, or completed Module operation", body = Value), (status = 409, description = "Operation cannot currently be applied", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn apply_operation(
    admin: AdminActor,
    State(ctx): State<AppContext>,
    HttpRequestContext(request_context): HttpRequestContext,
    AxumPath(operation_id): AxumPath<String>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let actor = management_actor(admin);
    run_effectful(ctx, request_context, move |operator| {
        operator.apply(&operation_id, &actor, chrono::Utc::now())
    })
    .await
}

#[utoipa::path(post, path = "/admin/modules/operations/{operation_id}/approvals", operation_id = "admin_modules_approve_operation", tag = "module-management", params(("operation_id" = String, Path)), request_body(content = Value, content_type = "application/json"), responses((status = 200, description = "Operation with plan-bound approval", body = Value), (status = 400, description = "Approval is invalid", body = ErrorResponse), (status = 409, description = "Approval is stale", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn approve_operation(
    admin: AdminActor,
    State(ctx): State<AppContext>,
    HttpRequestContext(request_context): HttpRequestContext,
    AxumPath(operation_id): AxumPath<String>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let body: ApprovalBody = decode(body, &request_context)?;
    let actor = management_actor(admin);
    let operator = WorkspaceModuleOperator::new(
        management_root(&request_context)?,
        HostModuleEffectAdapter::new(&ctx),
    );
    let operation = operator
        .approve(
            &operation_id,
            ApproveModuleOperation {
                expected_revision: body.expected_revision,
                boundary_id: body.boundary_id,
                reason: body.reason,
                nonce: body.nonce,
            },
            &actor,
            MANAGEMENT_HOLDER,
            chrono::Utc::now(),
        )
        .map_err(|error| operator_error(error, &request_context))?;
    value(operation, &request_context)
}

#[utoipa::path(post, path = "/admin/modules/operations/{operation_id}/cancel", operation_id = "admin_modules_cancel_operation", tag = "module-management", params(("operation_id" = String, Path)), request_body(content = Value, content_type = "application/json"), responses((status = 200, description = "Cancelled pre-mutation Module operation", body = Value), (status = 409, description = "Cancellation is unsafe", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn cancel_operation(
    admin: AdminActor,
    State(ctx): State<AppContext>,
    HttpRequestContext(request_context): HttpRequestContext,
    AxumPath(operation_id): AxumPath<String>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let body: RevisionBody = decode(body, &request_context)?;
    let actor = management_actor(admin);
    let operator = WorkspaceModuleOperator::new(
        management_root(&request_context)?,
        HostModuleEffectAdapter::new(&ctx),
    );
    value(
        operator
            .cancel(
                &operation_id,
                body.expected_revision,
                &actor,
                chrono::Utc::now(),
            )
            .map_err(|error| operator_error(error, &request_context))?,
        &request_context,
    )
}

#[utoipa::path(post, path = "/admin/modules/operations/{operation_id}/retry", operation_id = "admin_modules_retry_operation", tag = "module-management", params(("operation_id" = String, Path)), request_body(content = Value, content_type = "application/json"), responses((status = 200, description = "Retried Module operation", body = Value), (status = 409, description = "Operation cannot be retried", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn retry_operation(
    admin: AdminActor,
    State(ctx): State<AppContext>,
    HttpRequestContext(request_context): HttpRequestContext,
    AxumPath(operation_id): AxumPath<String>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let body: RevisionBody = decode(body, &request_context)?;
    let actor = management_actor(admin);
    run_effectful(ctx, request_context, move |operator| {
        operator.retry(
            &operation_id,
            body.expected_revision,
            &actor,
            chrono::Utc::now(),
        )
    })
    .await
}

#[utoipa::path(post, path = "/admin/modules/operations/{operation_id}/resume", operation_id = "admin_modules_resume_operation", tag = "module-management", params(("operation_id" = String, Path)), request_body(content = Value, content_type = "application/json"), responses((status = 200, description = "Crash-resumed Module operation", body = Value), (status = 409, description = "Safe continuation cannot be proven", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn resume_operation(
    admin: AdminActor,
    State(ctx): State<AppContext>,
    HttpRequestContext(request_context): HttpRequestContext,
    AxumPath(operation_id): AxumPath<String>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let body: RevisionBody = decode(body, &request_context)?;
    let actor = management_actor(admin);
    run_effectful(ctx, request_context, move |operator| {
        operator.resume(
            &operation_id,
            body.expected_revision,
            &actor,
            chrono::Utc::now(),
        )
    })
    .await
}

#[utoipa::path(get, path = "/admin/services/installations/{environment_id}", operation_id = "admin_services_installation_snapshot", tag = "service-management", params(("environment_id" = String, Path), ("system_id" = String, Query), ("authorization" = String, Header, description = "Service or system bearer token")), responses((status = 200, description = "Target-owned desired Service Installation Set", body = Value), (status = 401, description = "Authentication is required", body = ErrorResponse), (status = 403, description = "Service or system authentication is required", body = ErrorResponse), (status = 500, description = "Installation state cannot be read", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn service_installation_snapshot(
    _admin: AdminActor,
    HttpRequestContext(request_context): HttpRequestContext,
    AxumPath(environment_id): AxumPath<String>,
    Query(scope): Query<ServiceInstallationScope>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let manager = WorkspaceServiceInstallationManager::new(
        management_root(&request_context)?,
        scope.system_id,
        environment_id,
    );
    value(
        manager
            .snapshot()
            .map_err(|error| service_installation_error(error, &request_context))?,
        &request_context,
    )
}

#[utoipa::path(post, path = "/admin/services/installations/plans/preview", operation_id = "admin_services_preview_installation", tag = "service-management", request_body(content = Value, content_type = "application/json"), responses((status = 200, description = "Immutable Service Installation Plan", body = Value), (status = 400, description = "Installation change is invalid", body = ErrorResponse), (status = 401, description = "Authentication is required", body = ErrorResponse), (status = 403, description = "Service or system authentication is required", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn preview_service_installation(
    _admin: AdminActor,
    HttpRequestContext(request_context): HttpRequestContext,
    Json(body): Json<Value>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let body: PreviewServiceInstallationBody = decode(body, &request_context)?;
    let manager = WorkspaceServiceInstallationManager::new(
        management_root(&request_context)?,
        body.system_id,
        body.environment_id,
    );
    value(
        manager
            .preview(body.change, chrono::Utc::now())
            .map_err(|error| service_installation_error(error, &request_context))?,
        &request_context,
    )
}

#[utoipa::path(post, path = "/admin/services/installations/plans/{plan_id}/apply", operation_id = "admin_services_apply_installation", tag = "service-management", params(("plan_id" = String, Path)), request_body(content = Value, content_type = "application/json"), responses((status = 200, description = "Durable Service Installation Receipt", body = Value), (status = 400, description = "Installation plan is invalid", body = ErrorResponse), (status = 401, description = "Authentication is required", body = ErrorResponse), (status = 403, description = "service.manage authority is required", body = ErrorResponse), (status = 409, description = "Installation Set changed after preview", body = ErrorResponse)))]
#[allow(clippy::result_large_err)]
async fn apply_service_installation_plan(
    admin: AdminActor,
    HttpRequestContext(request_context): HttpRequestContext,
    AxumPath(plan_id): AxumPath<String>,
    Json(body): Json<Value>,
) -> Result<Json<Value>, ApiErrorResponse> {
    let body: ApplyServiceInstallationBody = decode(body, &request_context)?;
    if body.plan.plan_id != plan_id {
        return Err(api_error(
            ErrorCode::Validation,
            "Service Installation plan identity differs from request path".to_owned(),
            &request_context,
        ));
    }
    let actor = management_actor(admin);
    let manager = WorkspaceServiceInstallationManager::new(
        management_root(&request_context)?,
        &body.plan.system_id,
        &body.plan.environment_id,
    );
    value(
        manager
            .apply(
                &body.operation_id,
                &body.plan,
                &actor.actor_id,
                &actor.verified_authorities,
                chrono::Utc::now(),
            )
            .map_err(|error| service_installation_error(error, &request_context))?,
        &request_context,
    )
}

#[derive(Debug, Clone)]
struct HostModuleEffectAdapter {
    shutdown: Shutdown,
    db: DbPool,
    runtime: tokio::runtime::Handle,
}

impl HostModuleEffectAdapter {
    fn new(ctx: &AppContext) -> Self {
        Self {
            shutdown: ctx.shutdown.clone(),
            db: ctx.db.clone(),
            runtime: tokio::runtime::Handle::current(),
        }
    }
}

impl ModuleEffectAdapter for HostModuleEffectAdapter {
    fn execute(
        &self,
        workspace_root: &Path,
        operation: &lenso_module_management::ModuleOperation,
        effect: &ModulePlanEffect,
    ) -> Result<ModuleEffectExecution, ModuleEffectAdapterError> {
        let outcome = match effect {
            ModulePlanEffect::Validate {
                effect_id,
                command,
                expected_evidence,
            } if command == "cargo check --locked" => {
                let lock = fs::read(workspace_root.join("Cargo.lock"))
                    .map_err(|error| failed(effect_id, error))?;
                if sha256(&lock) != *expected_evidence {
                    return Err(failed(
                        effect_id,
                        "Cargo.lock no longer matches the reviewed candidate",
                    ));
                }
                let output = Command::new("cargo")
                    .args(["check", "--locked"])
                    .current_dir(workspace_root)
                    .output()
                    .map_err(|error| failed(effect_id, error))?;
                if !output.status.success() {
                    return Err(failed(effect_id, String::from_utf8_lossy(&output.stderr)));
                }
                ModuleEffectOutcome::Verified
            }
            ModulePlanEffect::Restart { target, .. } if target == "host" => {
                self.shutdown.signal();
                ModuleEffectOutcome::Applied
            }
            ModulePlanEffect::Activate {
                effect_id,
                target_lock_digest,
            } => {
                let bytes = fs::read(workspace_root.join("lenso.modules.lock.json"))
                    .map_err(|error| failed(effect_id, error))?;
                let lock =
                    serde_json::from_slice(&bytes).map_err(|error| failed(effect_id, error))?;
                if application_module_lock_digest(&lock)
                    .map_err(|error| failed(effect_id, error))?
                    != *target_lock_digest
                {
                    return Err(failed(
                        effect_id,
                        "application lock does not match reviewed target",
                    ));
                }
                ModuleEffectOutcome::Activated
            }
            ModulePlanEffect::Migration {
                effect_id,
                module_id,
                release_digest,
                migration_id,
                artifact_locator,
                artifact_digest,
                store_scope,
                execution,
                ..
            } if store_scope == "host" && *execution == MigrationExecutionMode::Transactional => {
                let artifact = verified_workspace_artifact(
                    workspace_root,
                    artifact_locator,
                    artifact_digest,
                    effect_id,
                )?;
                let sql = std::str::from_utf8(&artifact.bytes)
                    .map_err(|error| failed(effect_id, error))?;
                let name = format!("{module_id}/{release_digest}/{migration_id}");
                self.runtime
                    .block_on(apply_module_migration(
                        &self.db,
                        &name,
                        artifact_digest,
                        sql,
                    ))
                    .map_err(|error| failed(effect_id, error))?;
                return Ok(ModuleEffectExecution {
                    outcome: ModuleEffectOutcome::Applied,
                    evidence_references: vec![
                        artifact.reference,
                        write_effect_receipt(
                            workspace_root,
                            operation,
                            effect,
                            "module_migration_applied",
                        )?,
                    ],
                });
            }
            ModulePlanEffect::ServiceInstallation {
                installation_plan,
                action,
                ..
            } => {
                if let Some(receipt) = existing_effect_receipt(workspace_root, effect)? {
                    return Ok(ModuleEffectExecution {
                        outcome: ModuleEffectOutcome::Applied,
                        evidence_references: vec![receipt],
                    });
                }
                let mut references = Vec::new();
                if let Some(plan) = installation_plan {
                    references.push(apply_service_installation(
                        workspace_root,
                        operation,
                        effect,
                        plan,
                    )?);
                }
                let action =
                    action
                        .as_ref()
                        .ok_or_else(|| ModuleEffectAdapterError::Unsupported {
                            effect_id: effect.effect_id().to_owned(),
                            reason:
                                "desired Service installation was applied but no target-owned deployment action is available"
                                    .to_owned(),
                        })?;
                references.push(execute_service_action(workspace_root, effect, action)?);
                references.push(write_effect_receipt(
                    workspace_root,
                    operation,
                    effect,
                    "service_installation_and_deployment_applied",
                )?);
                return Ok(ModuleEffectExecution {
                    outcome: ModuleEffectOutcome::Applied,
                    evidence_references: references,
                });
            }
            ModulePlanEffect::ConsoleComposition { effect_id, .. } => {
                if let Some(receipt) = existing_effect_receipt(workspace_root, effect)? {
                    return Ok(ModuleEffectExecution {
                        outcome: ModuleEffectOutcome::Applied,
                        evidence_references: vec![receipt],
                    });
                }
                let management_url =
                    std::env::var("LENSO_CONSOLE_MANAGEMENT_URL").map_err(|_| {
                        ModuleEffectAdapterError::Unsupported {
                            effect_id: effect_id.clone(),
                            reason: "LENSO_CONSOLE_MANAGEMENT_URL is not configured".to_owned(),
                        }
                    })?;
                let management_token =
                    std::env::var("LENSO_CONSOLE_MANAGEMENT_TOKEN").map_err(|_| {
                        ModuleEffectAdapterError::Unsupported {
                            effect_id: effect_id.clone(),
                            reason: "LENSO_CONSOLE_MANAGEMENT_TOKEN is not configured".to_owned(),
                        }
                    })?;
                let management_url = reqwest::Url::parse(&management_url)
                    .map_err(|error| failed(effect_id, error))?;
                if management_url.scheme() != "https"
                    && !(management_url.scheme() == "http"
                        && management_url.host_str().is_some_and(|host| {
                            host == "localhost"
                                || host
                                    .parse::<std::net::IpAddr>()
                                    .is_ok_and(|ip| ip.is_loopback())
                        }))
                {
                    return Err(failed(
                        effect_id,
                        "Console management URL must use HTTPS or loopback HTTP",
                    ));
                }
                let endpoint = format!(
                    "{}/api/console/v1/artifacts/reconcile",
                    management_url.as_str().trim_end_matches('/')
                );
                let client = reqwest::Client::builder()
                    .redirect(reqwest::redirect::Policy::none())
                    .build()
                    .map_err(|error| failed(effect_id, error))?;
                let response = self
                    .runtime
                    .block_on(
                        client
                            .post(endpoint)
                            .bearer_auth(management_token)
                            .json(effect)
                            .send(),
                    )
                    .map_err(|error| failed(effect_id, error))?;
                if !response.status().is_success() {
                    let status = response.status();
                    let body = self.runtime.block_on(response.text()).unwrap_or_default();
                    return Err(failed(
                        effect_id,
                        format!("Console composition request failed with {status}: {body}"),
                    ));
                }
                let receipt = write_effect_receipt(
                    workspace_root,
                    operation,
                    effect,
                    "console_composition_reconciled",
                )?;
                return Ok(ModuleEffectExecution {
                    outcome: ModuleEffectOutcome::Applied,
                    evidence_references: vec![receipt],
                });
            }
            ModulePlanEffect::ServiceRemoval { action, .. }
            | ModulePlanEffect::ServiceRestart { action, .. } => {
                if let Some(receipt) = existing_effect_receipt(workspace_root, effect)? {
                    return Ok(ModuleEffectExecution {
                        outcome: ModuleEffectOutcome::Applied,
                        evidence_references: vec![receipt],
                    });
                }
                let action =
                    action
                        .as_ref()
                        .ok_or_else(|| ModuleEffectAdapterError::Unsupported {
                            effect_id: effect.effect_id().to_owned(),
                            reason: "reviewed plan has no target-owned Service deployment action"
                                .to_owned(),
                        })?;
                let evidence = execute_service_action(workspace_root, effect, action)?;
                let mut references = vec![evidence];
                references.push(write_effect_receipt(
                    workspace_root,
                    operation,
                    effect,
                    "service_deployment_action_applied",
                )?);
                return Ok(ModuleEffectExecution {
                    outcome: ModuleEffectOutcome::Applied,
                    evidence_references: references,
                });
            }
            _ => {
                return Err(ModuleEffectAdapterError::Unsupported {
                    effect_id: effect.effect_id().to_owned(),
                    reason: "no deployment or migration adapter is configured for this target"
                        .to_owned(),
                });
            }
        };
        Ok(ModuleEffectExecution {
            outcome,
            evidence_references: Vec::new(),
        })
    }
}

#[allow(clippy::result_large_err)]
async fn run_effectful(
    ctx: AppContext,
    request_context: platform_core::RequestContext,
    run: impl FnOnce(
        WorkspaceModuleOperator<HostModuleEffectAdapter>,
    )
        -> Result<lenso_module_management::ModuleOperation, WorkspaceModuleOperatorError>
    + Send
    + 'static,
) -> Result<Json<Value>, ApiErrorResponse> {
    let root = management_root(&request_context)?;
    let adapter = HostModuleEffectAdapter::new(&ctx);
    let result =
        tokio::task::spawn_blocking(move || run(WorkspaceModuleOperator::new(root, adapter)))
            .await
            .map_err(|error| {
                api_error(
                    ErrorCode::Internal,
                    format!("Module operation task failed: {error}"),
                    &request_context,
                )
            })?;
    value(
        result.map_err(|error| operator_error(error, &request_context))?,
        &request_context,
    )
}

struct VerifiedArtifact {
    bytes: Vec<u8>,
    reference: lenso_contracts::ArtifactReference,
}

fn verified_workspace_artifact(
    workspace_root: &Path,
    locator: &str,
    expected_digest: &str,
    effect_id: &str,
) -> Result<VerifiedArtifact, ModuleEffectAdapterError> {
    let root = workspace_root
        .canonicalize()
        .map_err(|error| failed(effect_id, error))?;
    let path = workspace_root
        .join(locator)
        .canonicalize()
        .map_err(|error| failed(effect_id, error))?;
    if !path.starts_with(&root) || !path.is_file() {
        return Err(failed(effect_id, "artifact escapes the managed workspace"));
    }
    let bytes = fs::read(&path).map_err(|error| failed(effect_id, error))?;
    if sha256(&bytes) != expected_digest {
        return Err(failed(
            effect_id,
            "artifact digest differs from the reviewed plan",
        ));
    }
    Ok(VerifiedArtifact {
        bytes,
        reference: lenso_contracts::ArtifactReference {
            locator: locator.to_owned(),
            digest: expected_digest.to_owned(),
        },
    })
}

fn execute_service_action(
    workspace_root: &Path,
    effect: &ModulePlanEffect,
    action: &ServiceDeploymentAction,
) -> Result<lenso_contracts::ArtifactReference, ModuleEffectAdapterError> {
    let effect_id = effect.effect_id();
    match action {
        ServiceDeploymentAction::Evidence { receipt } => Ok(verified_workspace_artifact(
            workspace_root,
            &receipt.locator,
            &receipt.digest,
            effect_id,
        )?
        .reference),
        ServiceDeploymentAction::Command {
            program,
            args,
            working_directory,
        } => {
            let executable = program.rsplit(['/', '\\']).next().unwrap_or_default();
            if matches!(
                executable.to_ascii_lowercase().as_str(),
                "sh" | "bash"
                    | "dash"
                    | "zsh"
                    | "fish"
                    | "cmd"
                    | "cmd.exe"
                    | "powershell"
                    | "powershell.exe"
                    | "pwsh"
                    | "pwsh.exe"
            ) {
                return Err(failed(
                    effect_id,
                    "shell programs are not valid deployment adapters",
                ));
            }
            let directory =
                command_directory(workspace_root, working_directory.as_deref(), effect_id)?;
            let output = Command::new(program)
                .args(args)
                .current_dir(directory)
                .output()
                .map_err(|error| failed(effect_id, error))?;
            if !output.status.success() {
                return Err(failed(
                    effect_id,
                    format!("Service deployment command exited with {}", output.status),
                ));
            }
            let digest = lenso_contracts::digest_json(&json!({
                "effect": effect,
                "program": program,
                "args": args,
                "exitCode": output.status.code(),
            }))
            .map_err(|error| failed(effect_id, error))?;
            Ok(lenso_contracts::ArtifactReference {
                locator: format!("command:{program}"),
                digest,
            })
        }
    }
}

fn command_directory(
    workspace_root: &Path,
    relative: Option<&str>,
    effect_id: &str,
) -> Result<PathBuf, ModuleEffectAdapterError> {
    let root = workspace_root
        .canonicalize()
        .map_err(|error| failed(effect_id, error))?;
    let directory = relative.map_or_else(|| root.clone(), |path| workspace_root.join(path));
    let directory = directory
        .canonicalize()
        .map_err(|error| failed(effect_id, error))?;
    if !directory.starts_with(&root) || !directory.is_dir() {
        return Err(failed(
            effect_id,
            "command working directory escapes the managed workspace",
        ));
    }
    Ok(directory)
}

fn write_effect_receipt(
    workspace_root: &Path,
    operation: &lenso_module_management::ModuleOperation,
    effect: &ModulePlanEffect,
    outcome: &str,
) -> Result<lenso_contracts::ArtifactReference, ModuleEffectAdapterError> {
    let effect_id = effect.effect_id();
    let effect_digest =
        lenso_contracts::digest_json(effect).map_err(|error| failed(effect_id, error))?;
    let relative = format!(
        ".lenso/module-management/effect-evidence/{}.json",
        effect_digest.trim_start_matches("sha256:")
    );
    let path = workspace_root.join(&relative);
    if let Some(existing) = existing_effect_receipt(workspace_root, effect)? {
        return Ok(existing);
    }
    let document = serde_json::to_vec_pretty(&json!({
        "protocol": "lenso.module-effect-evidence.v1",
        "operationId": operation.operation_id,
        "attempt": operation.attempt,
        "effectId": effect_id,
        "effectDigest": effect_digest,
        "outcome": outcome,
    }))
    .map_err(|error| failed(effect_id, error))?;
    let parent = path
        .parent()
        .ok_or_else(|| failed(effect_id, "receipt path has no parent"))?;
    fs::create_dir_all(parent).map_err(|error| failed(effect_id, error))?;
    let temporary = path.with_extension(format!("json.tmp-{}", std::process::id()));
    fs::write(&temporary, &document).map_err(|error| failed(effect_id, error))?;
    fs::rename(&temporary, &path).map_err(|error| failed(effect_id, error))?;
    Ok(lenso_contracts::ArtifactReference {
        locator: relative,
        digest: sha256(&document),
    })
}

fn existing_effect_receipt(
    workspace_root: &Path,
    effect: &ModulePlanEffect,
) -> Result<Option<lenso_contracts::ArtifactReference>, ModuleEffectAdapterError> {
    let effect_id = effect.effect_id();
    let effect_digest =
        lenso_contracts::digest_json(effect).map_err(|error| failed(effect_id, error))?;
    let relative = format!(
        ".lenso/module-management/effect-evidence/{}.json",
        effect_digest.trim_start_matches("sha256:")
    );
    let path = workspace_root.join(&relative);
    let bytes = match fs::read(&path) {
        Ok(bytes) => bytes,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(failed(effect_id, error)),
    };
    let document: Value =
        serde_json::from_slice(&bytes).map_err(|error| failed(effect_id, error))?;
    if document.get("protocol").and_then(Value::as_str) != Some("lenso.module-effect-evidence.v1")
        || document.get("effectDigest").and_then(Value::as_str) != Some(&effect_digest)
        || document.get("outcome").and_then(Value::as_str).is_none()
    {
        return Err(failed(
            effect_id,
            "persisted effect evidence does not match the reviewed effect",
        ));
    }
    Ok(Some(lenso_contracts::ArtifactReference {
        locator: relative,
        digest: sha256(&bytes),
    }))
}

fn management_actor(admin: AdminActor) -> ManagementActor {
    match admin {
        AdminActor::Service { service_id, scopes } => ManagementActor {
            actor_id: format!("service:{service_id}"),
            verified_authorities: scopes.into_iter().collect(),
        },
        AdminActor::User { user_id, scopes } => ManagementActor {
            actor_id: format!("user:{user_id}"),
            verified_authorities: scopes.into_iter().collect(),
        },
        AdminActor::System => ManagementActor {
            actor_id: "system".to_owned(),
            verified_authorities: BTreeSet::from([
                "module.manage".to_owned(),
                "module.migrate.destructive".to_owned(),
                "module.data.delete".to_owned(),
                "module.trust.override".to_owned(),
                "service.manage".to_owned(),
            ]),
        },
    }
}

fn apply_service_installation(
    workspace_root: &Path,
    operation: &lenso_module_management::ModuleOperation,
    effect: &ModulePlanEffect,
    plan: &ServiceInstallationPlan,
) -> Result<lenso_contracts::ArtifactReference, ModuleEffectAdapterError> {
    let effect_id = effect.effect_id();
    let effect_digest =
        lenso_contracts::digest_json(effect).map_err(|error| failed(effect_id, error))?;
    let service_operation_id = format!(
        "{}-{}",
        operation.operation_id,
        &effect_digest.trim_start_matches("sha256:")[..16]
    );
    let authorities = operation
        .verified_authorities
        .iter()
        .cloned()
        .collect::<BTreeSet<_>>();
    WorkspaceServiceInstallationManager::new(workspace_root, &plan.system_id, &plan.environment_id)
        .apply(
            &service_operation_id,
            plan,
            &operation.actor_id,
            &authorities,
            chrono::Utc::now(),
        )
        .map_err(|error| failed(effect_id, error))?;
    let relative = format!(
        ".lenso/environments/{}/service-install-receipts/{service_operation_id}.json",
        plan.environment_id
    );
    let bytes =
        fs::read(workspace_root.join(&relative)).map_err(|error| failed(effect_id, error))?;
    Ok(lenso_contracts::ArtifactReference {
        locator: relative,
        digest: sha256(&bytes),
    })
}

#[allow(clippy::result_large_err)]
fn management_root(
    context: &platform_core::RequestContext,
) -> Result<std::path::PathBuf, ApiErrorResponse> {
    std::env::current_dir().map_err(|error| {
        api_error(
            ErrorCode::Internal,
            format!("Module management root is unavailable: {error}"),
            context,
        )
    })
}
#[allow(clippy::result_large_err)]
fn decode<T: serde::de::DeserializeOwned>(
    body: Value,
    context: &platform_core::RequestContext,
) -> Result<T, ApiErrorResponse> {
    serde_json::from_value(body).map_err(|error| {
        api_error(
            ErrorCode::Validation,
            format!("Module operation request is invalid: {error}"),
            context,
        )
    })
}
#[allow(clippy::result_large_err)]
fn value<T: serde::Serialize>(
    body: T,
    context: &platform_core::RequestContext,
) -> Result<Json<Value>, ApiErrorResponse> {
    serde_json::to_value(body).map(Json).map_err(|error| {
        api_error(
            ErrorCode::Internal,
            format!("Module operation serialization failed: {error}"),
            context,
        )
    })
}
#[allow(clippy::needless_pass_by_value)]
fn operator_error(
    error: WorkspaceModuleOperatorError,
    context: &platform_core::RequestContext,
) -> ApiErrorResponse {
    let code = match &error {
        WorkspaceModuleOperatorError::Store(
            lenso_module_management::ModuleOperationStoreError::NotFound(_),
        ) => ErrorCode::NotFound,
        WorkspaceModuleOperatorError::Management(
            lenso_module_management::ModuleManagementError::MissingAuthority(_),
        ) => ErrorCode::Forbidden,
        WorkspaceModuleOperatorError::StalePlan
        | WorkspaceModuleOperatorError::CancellationUnsafe
        | WorkspaceModuleOperatorError::PolicyUnavailable
        | WorkspaceModuleOperatorError::Management(_)
        | WorkspaceModuleOperatorError::Store(_) => ErrorCode::Conflict,
        _ => ErrorCode::Internal,
    };
    api_error(code, format!("Module operation failed: {error}"), context)
}

#[allow(clippy::needless_pass_by_value)]
fn service_installation_error(
    error: lenso_module_management::ServiceInstallationError,
    context: &platform_core::RequestContext,
) -> ApiErrorResponse {
    let code = match error {
        lenso_module_management::ServiceInstallationError::InvalidContract(_)
        | lenso_module_management::ServiceInstallationError::UnsafeOperationIdentity
        | lenso_module_management::ServiceInstallationError::Json(_) => ErrorCode::Validation,
        lenso_module_management::ServiceInstallationError::MissingAuthority(_) => {
            ErrorCode::Forbidden
        }
        lenso_module_management::ServiceInstallationError::StaleState => ErrorCode::Conflict,
        lenso_module_management::ServiceInstallationError::Io(_) => ErrorCode::Internal,
    };
    api_error(
        code,
        format!("Service Installation operation failed: {error}"),
        context,
    )
}
fn failed(effect_id: &str, reason: impl std::fmt::Display) -> ModuleEffectAdapterError {
    ModuleEffectAdapterError::Failed {
        effect_id: effect_id.to_owned(),
        reason: reason.to_string(),
    }
}
fn sha256(bytes: &[u8]) -> String {
    let digest = Sha256::digest(bytes);
    let mut value = String::from("sha256:");
    for byte in digest {
        use std::fmt::Write as _;
        write!(&mut value, "{byte:02x}").expect("writing to a String cannot fail");
    }
    value
}

fn api_error(
    code: ErrorCode,
    message: String,
    context: &platform_core::RequestContext,
) -> ApiErrorResponse {
    ApiErrorResponse::with_context(AppError::new(code, message), context)
}

#[cfg(test)]
mod tests {
    use super::*;
    use lenso_module_management::ServiceDeploymentAdapterKind;
    use std::sync::atomic::{AtomicU64, Ordering};

    static NEXT_ROOT: AtomicU64 = AtomicU64::new(1);

    fn root(name: &str) -> PathBuf {
        let root = std::env::temp_dir().join(format!(
            "lenso-module-effect-{name}-{}-{}",
            std::process::id(),
            NEXT_ROOT.fetch_add(1, Ordering::Relaxed)
        ));
        fs::create_dir_all(&root).unwrap();
        root
    }

    fn effect(action: ServiceDeploymentAction) -> ModulePlanEffect {
        ModulePlanEffect::ServiceInstallation {
            effect_id: "service-install:test".to_owned(),
            service_id: "acme/support".to_owned(),
            service_release_digest: format!("sha256:{}", "a".repeat(64)),
            installation_plan: None,
            adapter: Some(ServiceDeploymentAdapterKind::Local),
            action: Some(action),
        }
    }

    #[test]
    fn evidence_action_rejects_digest_drift() {
        let root = root("evidence");
        fs::write(root.join("deployment.json"), b"observed").unwrap();
        let action = ServiceDeploymentAction::Evidence {
            receipt: lenso_contracts::ArtifactReference {
                locator: "deployment.json".to_owned(),
                digest: format!("sha256:{}", "0".repeat(64)),
            },
        };
        let error = execute_service_action(&root, &effect(action.clone()), &action).unwrap_err();
        assert!(error.to_string().contains("digest differs"));
        fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn command_action_uses_argv_without_a_shell() {
        let root = root("command");
        let action = ServiceDeploymentAction::Command {
            program: "rustc".to_owned(),
            args: vec!["--version".to_owned()],
            working_directory: None,
        };
        let reference = execute_service_action(&root, &effect(action.clone()), &action).unwrap();
        assert_eq!(reference.locator, "command:rustc");
        assert!(reference.digest.starts_with("sha256:"));
        fs::remove_dir_all(root).unwrap();
    }

    #[test]
    fn command_action_rejects_shell_programs() {
        let root = root("shell");
        let action = ServiceDeploymentAction::Command {
            program: "/bin/sh".to_owned(),
            args: vec!["-c".to_owned(), "exit 0".to_owned()],
            working_directory: None,
        };
        let error = execute_service_action(&root, &effect(action.clone()), &action).unwrap_err();
        assert!(error.to_string().contains("shell programs"));
        fs::remove_dir_all(root).unwrap();
    }
}