Genarrative/server-rs/crates/api-server/src/external_generation.rs

use axum::{
    Json,
    extract::{Extension, Path, Query, State},
    http::StatusCode,
    response::Response,
};
use serde::Deserialize;
use serde_json::json;
use shared_contracts::external_generation::{
    ExternalGenerationJobStatus, ExternalGenerationJobStatusDetailRecord,
    ExternalGenerationJobStatusRecord, ExternalGenerationJobStatusResponse,
    ExternalGenerationQueueOverview, ExternalGenerationQueueOverviewResponse,
    ExternalGenerationTaskAcknowledgeRequest, ExternalGenerationTaskAcknowledgeResponse,
    ExternalGenerationTaskListResponse, ExternalGenerationTaskRecord,
};
use spacetime_client::{
    ExternalGenerationJobAcknowledgeRecordInput, ExternalGenerationJobGetRecordInput,
    ExternalGenerationJobListRecordInput, ExternalGenerationJobSummaryListRecord,
    ExternalGenerationJobSummaryRecord, SpacetimeClientError,
};

use crate::editor_generation_queue::{
    EDITOR_BACKGROUND_MUSIC_GENERATION_JOB_KIND, EDITOR_BACKGROUND_REMOVAL_JOB_KIND,
    EDITOR_CHARACTER_ANIMATION_GENERATION_JOB_KIND, EDITOR_SOUND_EFFECT_GENERATION_JOB_KIND,
};
use crate::{
    api_response::json_success_body, auth::AuthenticatedAccessToken, http_error::AppError,
    request_context::RequestContext, state::AppState,
};

const EXTERNAL_GENERATION_PROVIDER: &str = "external_generation";

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct ExternalGenerationTaskListQuery {
    limit: Option<u32>,
    include_acknowledged_terminal: Option<bool>,
    statuses: Option<String>,
}

pub async fn get_external_generation_queue_overview(
    State(state): State<AppState>,
    Extension(request_context): Extension<RequestContext>,
    Extension(authenticated): Extension<AuthenticatedAccessToken>,
) -> Result<Json<serde_json::Value>, Response> {
    let owner_user_id = authenticated.claims().user_id().to_string();
    let list = state
        .spacetime_client()
        .list_external_generation_job_summaries(ExternalGenerationJobListRecordInput {
            owner_user_id,
            limit: 1,
            include_acknowledged_terminal: false,
            statuses: Vec::new(),
        })
        .await
        .map_err(|error| external_generation_error_response(&request_context, error))?;

    Ok(json_success_body(
        Some(&request_context),
        ExternalGenerationQueueOverviewResponse {
            overview: map_external_generation_queue_overview(&list),
        },
    ))
}

pub async fn list_external_generation_tasks(
    State(state): State<AppState>,
    Extension(request_context): Extension<RequestContext>,
    Extension(authenticated): Extension<AuthenticatedAccessToken>,
    Query(query): Query<ExternalGenerationTaskListQuery>,
) -> Result<Json<serde_json::Value>, Response> {
    let owner_user_id = authenticated.claims().user_id().to_string();
    let requested_limit = query.limit.unwrap_or(50).clamp(1, 100);
    let status_filter = parse_external_generation_status_filter(query.statuses.as_deref());
    let statuses = external_generation_status_filter_input(&status_filter);
    let list = state
        .spacetime_client()
        .list_external_generation_job_summaries(ExternalGenerationJobListRecordInput {
            owner_user_id,
            limit: requested_limit,
            include_acknowledged_terminal: query.include_acknowledged_terminal.unwrap_or(false),
            statuses,
        })
        .await
        .map_err(|error| external_generation_error_response(&request_context, error))?;
    let overview = map_external_generation_queue_overview(&list);
    let mut tasks: Vec<_> = list
        .jobs
        .into_iter()
        .map(map_external_generation_task_record)
        .filter(|task| {
            status_filter.is_empty() || status_filter.iter().any(|status| status == &task.status)
        })
        .collect();
    tasks.truncate(requested_limit as usize);

    Ok(json_success_body(
        Some(&request_context),
        ExternalGenerationTaskListResponse { overview, tasks },
    ))
}

pub async fn acknowledge_external_generation_tasks(
    State(state): State<AppState>,
    Extension(request_context): Extension<RequestContext>,
    Extension(authenticated): Extension<AuthenticatedAccessToken>,
    Json(payload): Json<ExternalGenerationTaskAcknowledgeRequest>,
) -> Result<Json<serde_json::Value>, Response> {
    let owner_user_id = authenticated.claims().user_id().to_string();
    let acknowledged = state
        .spacetime_client()
        .acknowledge_external_generation_job_summaries(
            ExternalGenerationJobAcknowledgeRecordInput {
                owner_user_id,
                job_ids: payload.job_ids,
                acknowledged_at_micros: current_utc_micros(),
            },
        )
        .await
        .map_err(|error| external_generation_error_response(&request_context, error))?;

    Ok(json_success_body(
        Some(&request_context),
        ExternalGenerationTaskAcknowledgeResponse {
            acknowledged_tasks: acknowledged
                .jobs
                .into_iter()
                .map(map_external_generation_task_record)
                .collect(),
        },
    ))
}

pub async fn get_external_generation_job_status(
    State(state): State<AppState>,
    Extension(request_context): Extension<RequestContext>,
    Extension(authenticated): Extension<AuthenticatedAccessToken>,
    Path(job_id): Path<String>,
) -> Result<Json<serde_json::Value>, Response> {
    let owner_user_id = authenticated.claims().user_id().to_string();
    let input = ExternalGenerationJobGetRecordInput {
        job_id,
        owner_user_id,
    };
    let job = state
        .spacetime_client()
        .get_external_generation_job_summary(input.clone())
        .await
        .map_err(|error| external_generation_error_response(&request_context, error))?;
    let mut detail = map_external_generation_job_status_detail(job);
    if detail.status.status == ExternalGenerationJobStatus::Completed {
        let artifacts = state
            .spacetime_client()
            .get_external_generation_job_generated_artifacts(input)
            .await
            .map_err(|error| external_generation_error_response(&request_context, error))?;
        detail.result =
            external_generation_completed_result(artifacts.result_payload_json.as_deref());
    }

    Ok(json_success_body(
        Some(&request_context),
        ExternalGenerationJobStatusResponse { job: detail },
    ))
}

fn map_external_generation_queue_overview(
    list: &ExternalGenerationJobSummaryListRecord,
) -> ExternalGenerationQueueOverview {
    ExternalGenerationQueueOverview {
        pending_count: list.pending_count,
        running_count: list.running_count,
        unacknowledged_terminal_count: list.unacknowledged_terminal_count,
        updated_at_micros: list.now_micros,
    }
}

fn map_external_generation_job_status(
    job: ExternalGenerationJobSummaryRecord,
) -> ExternalGenerationJobStatusRecord {
    let (status, phase_detail, progress, error) = match job.status.as_str() {
        "completed" => (
            ExternalGenerationJobStatus::Completed,
            "生成已完成。",
            100,
            None,
        ),
        "running" if job.phase.as_deref() == Some("processing") => {
            (ExternalGenerationJobStatus::Running, "正在处理。", 70, None)
        }
        "running" => (ExternalGenerationJobStatus::Running, "正在生成。", 35, None),
        "failed" => (
            ExternalGenerationJobStatus::Failed,
            "生成失败。",
            0,
            user_visible_external_generation_error(
                job.job_kind.as_str(),
                job.last_error_message.clone(),
            ),
        ),
        _ => (ExternalGenerationJobStatus::Queued, "排队中。", 8, None),
    };

    ExternalGenerationJobStatusRecord {
        operation_id: job.job_id.clone(),
        status,
        phase_label: job.request_label.clone(),
        phase_detail: phase_detail.to_string(),
        progress,
        error,
        updated_at_micros: job.updated_at_micros,
    }
}

/// 外部任务表保存的失败文本同时服务于 worker 诊断和用户通知。抠图、角色动作透明化和音效
/// 任务的原始失败文本可能包含内部 provider、分割模型、请求端点或 HTTP 传输层细节，因此
/// 只能在普通用户读取边界替换成稳定文案。
/// 原始值仍留在任务记录、tracing 和后台审计路径中。
pub(crate) fn user_visible_external_generation_error(
    job_kind: &str,
    error: Option<String>,
) -> Option<String> {
    if job_kind == EDITOR_BACKGROUND_REMOVAL_JOB_KIND && error.is_some() {
        return Some("去除背景失败，请稍后重试。".to_string());
    }
    if job_kind == EDITOR_CHARACTER_ANIMATION_GENERATION_JOB_KIND && error.is_some() {
        return Some("角色动作生成失败，请稍后重试。".to_string());
    }
    // 音效失败文本来自 ElevenLabs / reqwest，会带上请求端点、底层错误链和上游状态码。
    if job_kind == EDITOR_SOUND_EFFECT_GENERATION_JOB_KIND && error.is_some() {
        return Some("音效生成失败，请稍后重试。".to_string());
    }
    if job_kind == EDITOR_BACKGROUND_MUSIC_GENERATION_JOB_KIND && error.is_some() {
        return Some("背景音乐生成失败，请稍后重试。".to_string());
    }

    error
}

pub(crate) fn map_external_generation_job_status_detail(
    job: ExternalGenerationJobSummaryRecord,
) -> ExternalGenerationJobStatusDetailRecord {
    let warning = job.warning_message.clone();
    ExternalGenerationJobStatusDetailRecord {
        status: map_external_generation_job_status(job),
        warning,
        result: None,
    }
}

fn external_generation_completed_result(
    result_payload_json: Option<&str>,
) -> Option<serde_json::Value> {
    result_payload_json
        .and_then(|payload| serde_json::from_str::<serde_json::Value>(payload).ok())
        .and_then(|payload| payload.get("result").cloned())
}

fn map_external_generation_task_record(
    job: ExternalGenerationJobSummaryRecord,
) -> ExternalGenerationTaskRecord {
    let status_record = map_external_generation_job_status(job.clone());
    ExternalGenerationTaskRecord {
        job_id: job.job_id,
        job_kind: job.job_kind,
        source_module: job.source_module,
        source_entity_id: job.source_entity_id,
        request_label: job.request_label,
        request_prompt: job.request_prompt,
        status: status_record.status,
        phase_label: status_record.phase_label,
        phase_detail: status_record.phase_detail,
        progress: status_record.progress,
        error: status_record.error,
        warning: job.warning_message,
        price_mud_points: job.price_mud_points,
        refund_ledger_id: job.refund_ledger_id,
        notification_acknowledged_at: job.notification_acknowledged_at,
        created_at: job.created_at,
        started_at: job.started_at,
        completed_at: job.completed_at,
        updated_at: job.updated_at,
        updated_at_micros: job.updated_at_micros,
    }
}

fn parse_external_generation_status_filter(
    statuses: Option<&str>,
) -> Vec<ExternalGenerationJobStatus> {
    statuses
        .unwrap_or_default()
        .split(',')
        .filter_map(|status| match status.trim() {
            "queued" => Some(ExternalGenerationJobStatus::Queued),
            "running" => Some(ExternalGenerationJobStatus::Running),
            "completed" => Some(ExternalGenerationJobStatus::Completed),
            "failed" => Some(ExternalGenerationJobStatus::Failed),
            _ => None,
        })
        .collect()
}

fn external_generation_status_filter_input(
    statuses: &[ExternalGenerationJobStatus],
) -> Vec<String> {
    statuses
        .iter()
        .map(|status| match status {
            ExternalGenerationJobStatus::Queued => "queued",
            ExternalGenerationJobStatus::Running => "running",
            ExternalGenerationJobStatus::Completed => "completed",
            ExternalGenerationJobStatus::Failed => "failed",
        })
        .map(str::to_string)
        .collect()
}

fn current_utc_micros() -> i64 {
    shared_kernel::offset_datetime_to_unix_micros(time::OffsetDateTime::now_utc())
}

fn external_generation_error_response(
    request_context: &RequestContext,
    error: SpacetimeClientError,
) -> Response {
    AppError::from_status(StatusCode::BAD_GATEWAY)
        .with_details(json!({
            "provider": EXTERNAL_GENERATION_PROVIDER,
            "message": error.to_string(),
        }))
        .into_response_with_context(Some(request_context))
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn parses_external_generation_status_filter() {
        assert_eq!(
            parse_external_generation_status_filter(Some("running,queued,unknown")),
            vec![
                ExternalGenerationJobStatus::Running,
                ExternalGenerationJobStatus::Queued
            ]
        );
    }

    #[test]
    fn completed_generation_result_uses_compact_result_payload() {
        let result = external_generation_completed_result(Some(
            r#"{"result":{"resourceId":"resource-1","objectKey":"generated/1.png"}}"#,
        ))
        .expect("completed result should be readable");

        assert_eq!(result["resourceId"], "resource-1");
        assert!(external_generation_completed_result(Some(r#"{"data":{}}"#)).is_none());
        assert!(external_generation_completed_result(Some("not-json")).is_none());
    }

    #[test]
    fn maps_task_from_payload_free_summary_projection() {
        let summary = ExternalGenerationJobSummaryRecord {
            job_id: "task-1".to_string(),
            job_kind: "editor_image_generation".to_string(),
            owner_user_id: "user-1".to_string(),
            source_module: "editor-canvas".to_string(),
            source_entity_id: "project-1".to_string(),
            request_label: "图片生成".to_string(),
            request_prompt: Some("发光主视觉".to_string()),
            status: "completed".to_string(),
            last_error_message: Some("上一次尝试失败".to_string()),
            created_at: "2026-07-10T08:00:00Z".to_string(),
            started_at: Some("2026-07-10T08:00:01Z".to_string()),
            completed_at: Some("2026-07-10T08:00:10Z".to_string()),
            updated_at: "2026-07-10T08:00:10Z".to_string(),
            updated_at_micros: 1_000,
            price_mud_points: 4,
            refund_ledger_id: None,
            notification_acknowledged_at: None,
            notification_acknowledged_at_micros: None,
            phase: None,
            warning_message: Some("连通域数量不足".to_string()),
        };
        let status = map_external_generation_job_status_detail(summary.clone());
        let task = map_external_generation_task_record(summary);

        assert_eq!(status.warning.as_deref(), Some("连通域数量不足"));
        assert!(status.status.error.is_none());
        assert_eq!(task.request_prompt.as_deref(), Some("发光主视觉"));
        assert_eq!(task.status, ExternalGenerationJobStatus::Completed);
        assert_eq!(task.warning.as_deref(), Some("连通域数量不足"));
        assert!(task.error.is_none());
    }

    #[test]
    fn maps_running_processing_phase_from_backend_projection() {
        let status = map_external_generation_job_status(ExternalGenerationJobSummaryRecord {
            job_id: "task-processing".to_string(),
            job_kind: "editor_character_animation_generation".to_string(),
            owner_user_id: "user-1".to_string(),
            source_module: "editor-canvas".to_string(),
            source_entity_id: "project-1".to_string(),
            request_label: "角色动作生成".to_string(),
            request_prompt: None,
            status: "running".to_string(),
            last_error_message: None,
            created_at: "2026-07-13T08:00:00Z".to_string(),
            started_at: Some("2026-07-13T08:00:01Z".to_string()),
            completed_at: None,
            updated_at: "2026-07-13T08:00:10Z".to_string(),
            updated_at_micros: 1_000,
            price_mud_points: 4,
            refund_ledger_id: None,
            notification_acknowledged_at: None,
            notification_acknowledged_at_micros: None,
            phase: Some("processing".to_string()),
            warning_message: None,
        });

        assert_eq!(status.status, ExternalGenerationJobStatus::Running);
        assert_eq!(status.phase_detail, "正在处理。");
        assert_eq!(status.progress, 70);
    }

    #[test]
    fn maps_legacy_running_job_without_phase_as_generating() {
        let mut job = ExternalGenerationJobSummaryRecord {
            job_id: "task-legacy".to_string(),
            job_kind: "editor_image_generation".to_string(),
            owner_user_id: "user-1".to_string(),
            source_module: "editor-canvas".to_string(),
            source_entity_id: "project-1".to_string(),
            request_label: "图片生成".to_string(),
            request_prompt: None,
            status: "running".to_string(),
            last_error_message: None,
            created_at: "2026-07-13T08:00:00Z".to_string(),
            started_at: Some("2026-07-13T08:00:01Z".to_string()),
            completed_at: None,
            updated_at: "2026-07-13T08:00:10Z".to_string(),
            updated_at_micros: 1_000,
            price_mud_points: 4,
            refund_ledger_id: None,
            notification_acknowledged_at: None,
            notification_acknowledged_at_micros: None,
            phase: None,
            warning_message: None,
        };

        let legacy = map_external_generation_job_status(job.clone());
        assert_eq!(legacy.phase_detail, "正在生成。");
        job.phase = Some("generating".to_string());
        let generating = map_external_generation_job_status(job);
        assert_eq!(generating.phase_detail, "正在生成。");
    }

    #[test]
    fn background_removal_failure_hides_internal_processing_details_from_owner() {
        let job = ExternalGenerationJobSummaryRecord {
            job_id: "task-matting-failed".to_string(),
            job_kind: EDITOR_BACKGROUND_REMOVAL_JOB_KIND.to_string(),
            owner_user_id: "user-1".to_string(),
            source_module: "editor-canvas".to_string(),
            source_entity_id: "project-1".to_string(),
            request_label: "去除背景".to_string(),
            request_prompt: None,
            status: "failed".to_string(),
            last_error_message: Some(
                "BgFilter birefnet provider 返回失败，model=segment-common-image".to_string(),
            ),
            created_at: "2026-07-31T08:00:00Z".to_string(),
            started_at: Some("2026-07-31T08:00:01Z".to_string()),
            completed_at: Some("2026-07-31T08:00:10Z".to_string()),
            updated_at: "2026-07-31T08:00:10Z".to_string(),
            updated_at_micros: 1_000,
            price_mud_points: 0,
            refund_ledger_id: None,
            notification_acknowledged_at: None,
            notification_acknowledged_at_micros: None,
            phase: None,
            warning_message: None,
        };

        let status = map_external_generation_job_status(job.clone());
        let task = map_external_generation_task_record(job);

        assert_eq!(status.error.as_deref(), Some("去除背景失败，请稍后重试。"));
        assert_eq!(task.error.as_deref(), Some("去除背景失败，请稍后重试。"));
        for value in [status.error, task.error].into_iter().flatten() {
            let lower = value.to_ascii_lowercase();
            assert!(!lower.contains("bgfilter"));
            assert!(!lower.contains("birefnet"));
            assert!(!lower.contains("segment-common-image"));
        }
    }

    #[test]
    fn character_animation_failure_hides_internal_matting_details_from_owner() {
        let message = user_visible_external_generation_error(
            EDITOR_CHARACTER_ANIMATION_GENERATION_JOB_KIND,
            Some("解析 BgFilter 输入动作帧失败，seg_model=birefnet，provider=BgFilter".to_string()),
        );

        assert_eq!(message.as_deref(), Some("角色动作生成失败，请稍后重试。"));
        let lower = message
            .expect("失败任务应返回稳定文案")
            .to_ascii_lowercase();
        assert!(!lower.contains("bgfilter"));
        assert!(!lower.contains("birefnet"));
        assert!(!lower.contains("provider"));
    }

    #[test]
    fn sound_effect_failure_hides_provider_and_transport_details_from_owner() {
        let job = ExternalGenerationJobSummaryRecord {
            job_id: "task-sound-effect-failed".to_string(),
            job_kind: EDITOR_SOUND_EFFECT_GENERATION_JOB_KIND.to_string(),
            owner_user_id: "user-1".to_string(),
            source_module: "editor-canvas".to_string(),
            source_entity_id: "project-1".to_string(),
            request_label: "音效生成".to_string(),
            request_prompt: None,
            status: "failed".to_string(),
            last_error_message: Some(
                "请求 ElevenLabs 音效生成失败：error sending request for url \
                 (https://api.elevenlabs.io/v1/sound-generation?output_format=mp3_44100_128): \
                 connection error: Connection refused (os error 111)"
                    .to_string(),
            ),
            created_at: "2026-08-07T08:00:00Z".to_string(),
            started_at: Some("2026-08-07T08:00:01Z".to_string()),
            completed_at: Some("2026-08-07T08:00:10Z".to_string()),
            updated_at: "2026-08-07T08:00:10Z".to_string(),
            updated_at_micros: 1_000,
            price_mud_points: 5,
            refund_ledger_id: None,
            notification_acknowledged_at: None,
            notification_acknowledged_at_micros: None,
            phase: None,
            warning_message: None,
        };

        let status = map_external_generation_job_status(job.clone());
        let task = map_external_generation_task_record(job);

        assert_eq!(status.error.as_deref(), Some("音效生成失败，请稍后重试。"));
        assert_eq!(task.error.as_deref(), Some("音效生成失败，请稍后重试。"));
        for value in [status.error, task.error].into_iter().flatten() {
            let lower = value.to_ascii_lowercase();
            assert!(!lower.contains("elevenlabs"));
            assert!(!lower.contains("api.elevenlabs.io"));
            assert!(!lower.contains("sound-generation"));
            assert!(!lower.contains("os error"));
        }
    }

    #[test]
    fn background_music_failure_hides_provider_and_transport_details_from_owner() {
        let message = user_visible_external_generation_error(
            EDITOR_BACKGROUND_MUSIC_GENERATION_JOB_KIND,
            Some(
                "提交 Vector Engine 背景音乐任务失败：error sending request for url \
                 (https://vector-engine.internal/v1/audio/generations): connection error: \
                 Connection refused (os error 111)"
                    .to_string(),
            ),
        );

        assert_eq!(message.as_deref(), Some("背景音乐生成失败，请稍后重试。"));
        let lower = message
            .expect("失败任务应返回稳定文案")
            .to_ascii_lowercase();
        for forbidden in ["vector engine", "vector-engine", "https://", "os error"] {
            assert!(
                !lower.contains(forbidden),
                "普通用户文案不应包含 {forbidden}"
            );
        }
    }

    #[test]
    fn unrelated_generation_failure_keeps_its_user_visible_message() {
        let message = user_visible_external_generation_error(
            "editor_image_generation",
            Some("生成服务暂时不可用。".to_string()),
        );

        assert_eq!(message.as_deref(), Some("生成服务暂时不可用。"));
    }
}