zebflow 0.11.14

//! The 0.11 migration over HTTP, for a project's owner (or a superadmin):
//!
//! - `GET  /api/projects/{owner}/{project}/migration/0.11/plan` — the plan:
//!   JSON, or its report alone with `?format=markdown`. Reads only.
//! - `POST /api/projects/{owner}/{project}/migration/0.11/apply` with
//!   `{ "fingerprint": "<the plan's>" }` — applies that plan; refused when
//!   the project changed since, or the plan is not ready.
//!
//! And at server start, [`migrate_projects_at_start`]: the same apply for
//! every project this office holds whose plan is ready and holds nothing to
//! review; every other project with 0.10 files left is logged and noticed.

use async_trait::async_trait;
use axum::Json;
use axum::extract::{Path, Query, State};
use axum::http::{HeaderMap, Method, StatusCode, Uri, header};
use axum::response::{IntoResponse, Response};
use axum::body::Bytes;
use serde::Deserialize;
use serde_json::json;

use std::sync::Arc;

use super::{PlatformAppState, internal_error, maybe_forward_project_api_to_worker, require_project_api_capability, run_lifecycle_hooks};
use crate::infra::mem::subscriber::KvSubscriber;
use crate::infra::scheduler::PipelineScheduler;
use crate::infra::ws_client::WsClientManager;
use crate::platform::services::PlatformService;
use crate::platform::error::PlatformError;
use crate::platform::model::ProjectCapability;
use crate::platform::services::migration::{MigrationService, PipelineSwitch, StartupOutcome};

/// Activation as the pipeline API does it: the runtime registry, the
/// schedules, subscriptions and socket clients, and the lifecycle hooks.
/// Attached to the project service when the router is built, so every door
/// that switches pipelines (HTTP, MCP) switches them the same way.
pub struct WebSwitch {
    pub platform: Arc<PlatformService>,
    pub scheduler: Arc<PipelineScheduler>,
    pub kv_subscriber: Arc<KvSubscriber>,
    pub ws_client_manager: Arc<WsClientManager>,
}

#[async_trait]
impl PipelineSwitch for WebSwitch {
    async fn activate(&self, owner: &str, project: &str, file_rel_path: &str) -> Result<(), PlatformError> {
        let meta = self.platform.projects.activate_pipeline_definition(owner, project, file_rel_path)?;
        self.arm(owner, project, &meta.file_rel_path).await
    }

    async fn deactivate(&self, owner: &str, project: &str, file_rel_path: &str) -> Result<(), PlatformError> {
        let meta = self.platform.projects.deactivate_pipeline_definition(owner, project, file_rel_path)?;
        self.disarm(owner, project, &meta.file_rel_path).await
    }

    async fn arm(&self, owner: &str, project: &str, file_rel_path: &str) -> Result<(), PlatformError> {
        self.platform.pipeline_runtime.refresh_pipeline(owner, project, file_rel_path)?;
        self.scheduler.sync_pipeline(owner, project, file_rel_path).await;
        self.kv_subscriber.sync_pipeline(owner, project, file_rel_path).await;
        self.ws_client_manager.sync_pipeline(owner, project, file_rel_path).await;
        run_lifecycle_hooks(&self.platform, owner, project, file_rel_path, "on_activate").await;
        Ok(())
    }

    async fn disarm(&self, owner: &str, project: &str, file_rel_path: &str) -> Result<(), PlatformError> {
        run_lifecycle_hooks(&self.platform, owner, project, file_rel_path, "on_deactivate").await;
        self.platform.pipeline_runtime.evict(owner, project, file_rel_path);
        self.scheduler.sync_pipeline(owner, project, file_rel_path).await;
        self.kv_subscriber.sync_pipeline(owner, project, file_rel_path).await;
        self.ws_client_manager.sync_pipeline(owner, project, file_rel_path).await;
        Ok(())
    }
}

#[derive(Debug, Deserialize, Default)]
pub struct PlanQuery {
    #[serde(default)]
    pub format: Option<String>,
}

#[derive(Debug, Deserialize)]
pub struct ApplyRequest {
    pub fingerprint: String,
}

/// Owner-only: the capability only a project's owner (and a superadmin)
/// holds.
const CAPABILITY: ProjectCapability = ProjectCapability::ProjectDelete;

pub async fn api_migration_plan(
    State(state): State<PlatformAppState>,
    headers: HeaderMap,
    Path((owner, project)): Path<(String, String)>,
    uri: Uri,
    Query(query): Query<PlanQuery>,
) -> Response {
    if let Err(response) = require_project_api_capability(&state, &headers, &owner, &project, CAPABILITY) {
        return response;
    }
    match maybe_forward_project_api_to_worker(&state, &uri, &Method::GET, &headers, Bytes::new(), &owner, &project).await {
        Ok(Some(response)) => return response,
        Ok(None) => {}
        Err(err) => return internal_error(err),
    }
    let service = MigrationService::new(state.platform.clone());
    match service.plan_project(&owner, &project) {
        Ok(plan) if query.format.as_deref() == Some("markdown") => {
            ([(header::CONTENT_TYPE, "text/markdown; charset=utf-8")], plan.report).into_response()
        }
        Ok(plan) => Json(json!({ "ok": true, "plan": plan })).into_response(),
        Err(err) => internal_error(err),
    }
}

pub async fn api_migration_apply(
    State(state): State<PlatformAppState>,
    headers: HeaderMap,
    Path((owner, project)): Path<(String, String)>,
    uri: Uri,
    Json(request): Json<ApplyRequest>,
) -> Response {
    if let Err(response) = require_project_api_capability(&state, &headers, &owner, &project, CAPABILITY) {
        return response;
    }
    let body = Bytes::from(serde_json::to_vec(&json!({ "fingerprint": request.fingerprint })).unwrap_or_default());
    match maybe_forward_project_api_to_worker(&state, &uri, &Method::POST, &headers, body, &owner, &project).await {
        Ok(Some(response)) => return response,
        Ok(None) => {}
        Err(err) => return internal_error(err),
    }
    let service = MigrationService::new(state.platform.clone());
    match service.apply_plan(&owner, &project, &request.fingerprint).await {
        Ok(report) => {
            let status = if report.ok { StatusCode::OK } else { StatusCode::MULTI_STATUS };
            (status, Json(json!({ "ok": report.ok, "report": report }))).into_response()
        }
        Err(err) if err.code == "PLATFORM_MIGRATION_STALE" || err.code == "PLATFORM_MIGRATION_BLOCKED" => (
            StatusCode::CONFLICT,
            Json(json!({ "ok": false, "error": { "code": err.code, "message": err.message } })),
        )
            .into_response(),
        Err(err) => internal_error(err),
    }
}

/// The 0.11 migration as part of an upgrade (`services/migration/startup.rs`),
/// over every project this office holds — any role: a project placed on
/// another office is that office's to migrate at its own start. Runs after
/// the pipeline switch is attached and before the first request.
pub async fn migrate_projects_at_start(state: &PlatformAppState) {
    let platform = &state.platform;
    let mut projects = std::collections::BTreeSet::new();
    for user in platform.users.list_users().unwrap_or_default() {
        for item in platform.data.list_projects(&user.owner).unwrap_or_default() {
            projects.insert((item.owner, item.project));
        }
    }
    let service = MigrationService::new(platform.clone());
    let (mut applied, mut named) = (0usize, 0usize);
    for (owner, project) in projects {
        if matches!(super::remote_project_worker_id(state, &owner, &project), Ok(Some(_))) {
            continue;
        }
        match service.migrate_at_start(&owner, &project).await {
            Ok(StartupOutcome::Nothing) => {}
            Ok(StartupOutcome::Applied(report)) => {
                applied += 1;
                println!(
                    "✅ 0.11 migration applied: {owner}/{project} — {} step(s); the record is archive/0.10/MIGRATION.md",
                    report.done.len()
                );
            }
            Ok(StartupOutcome::Named(notice)) => {
                named += 1;
                eprintln!("⚠️  {}", notice.line);
            }
            Err(err) => {
                named += 1;
                eprintln!("⚠️  0.11 migration could not be planned: {owner}/{project} — {}: {}", err.code, err.message);
            }
        }
    }
    if applied + named > 0 {
        println!("0.11 migration at start: {applied} project(s) migrated, {named} left for their owner");
    }
}