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

use axum::http::StatusCode;
use serde::Serialize;
use serde_json::{Value, json};
use sha2::{Digest, Sha256};
use shared_contracts::external_generation::{
    ExternalGenerationJobStatus, ExternalGenerationJobStatusRecord,
};
use shared_kernel::{build_prefixed_uuid_id, offset_datetime_to_unix_micros};
use spacetime_client::{ExternalGenerationJobEnqueueRecordInput, ExternalGenerationJobRecord};

use crate::{http_error::AppError, request_context::RequestContext, state::AppState};

pub(crate) const EDITOR_IMAGE_GENERATION_JOB_KIND: &str = "editor_image_generation";
pub(crate) const EDITOR_IMAGE_EDIT_JOB_KIND: &str = "editor_image_edit";
pub(crate) const EDITOR_BACKGROUND_REMOVAL_JOB_KIND: &str = "editor_background_removal";
pub(crate) const EDITOR_ICON_SPRITESHEET_GENERATION_JOB_KIND: &str =
    "editor_icon_spritesheet_generation";
pub(crate) const EDITOR_UI_DESIGN_ASSET_EXTRACTION_JOB_KIND: &str =
    "editor_ui_design_asset_extraction";
pub(crate) const EDITOR_CHARACTER_ANIMATION_GENERATION_JOB_KIND: &str =
    "editor_character_animation_generation";
pub(crate) const EDITOR_VIDEO_GENERATION_JOB_KIND: &str = "editor_video_generation";
pub(crate) const EDITOR_SOUND_EFFECT_GENERATION_JOB_KIND: &str = "editor_sound_effect_generation";
pub(crate) const EDITOR_BACKGROUND_MUSIC_GENERATION_JOB_KIND: &str =
    "editor_background_music_generation";

pub(crate) const EDITOR_GENERATION_QUEUE_SOURCE_MODULE: &str = "editor-canvas";
const EDITOR_GENERATION_QUEUE_PROVIDER: &str = "editor-generation-worker";
const MAX_EDITOR_GENERATION_JOB_PAYLOAD_BYTES: usize = 512 * 1024;
const EXTERNAL_API_GENERATION_DEDUPE_PREFIX: &str = "external-api-generation";
const EDITOR_API_REQUEST_GENERATION_DEDUPE_PREFIX: &str = "editor-api-request-generation";

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub(crate) struct EditorGenerationQueuedResponse {
    pub(crate) queue_state: ExternalGenerationJobStatusRecord,
}

pub(crate) async fn enqueue_editor_generation_job<T>(
    state: &AppState,
    request_context: &RequestContext,
    owner_user_id: &str,
    job_kind: &str,
    source_entity_id: impl Into<String>,
    request_label: impl Into<String>,
    price_mud_points: u64,
    payload: &T,
) -> Result<ExternalGenerationJobRecord, AppError>
where
    T: Serialize,
{
    let job_id = build_prefixed_uuid_id("task-");
    let dedupe_key = build_editor_generation_dedupe_key(
        EDITOR_API_REQUEST_GENERATION_DEDUPE_PREFIX,
        owner_user_id,
        job_kind,
        request_context.request_id().trim(),
    );
    enqueue_editor_generation_job_with_identity(
        state,
        owner_user_id,
        job_kind,
        source_entity_id,
        request_label,
        price_mud_points,
        payload,
        job_id.clone(),
        dedupe_key,
    )
    .await
}

fn build_editor_generation_dedupe_key(
    namespace: &str,
    owner_user_id: &str,
    job_kind: &str,
    stable_key: &str,
) -> String {
    let mut hasher = Sha256::new();
    hasher.update(owner_user_id.trim().as_bytes());
    hasher.update(b"\0");
    hasher.update(job_kind.trim().as_bytes());
    hasher.update(b"\0");
    hasher.update(stable_key.as_bytes());
    format!("{namespace}:{job_kind}:{:x}", hasher.finalize())
}

#[allow(clippy::too_many_arguments)]
pub(crate) async fn enqueue_editor_generation_job_with_identity<T>(
    state: &AppState,
    owner_user_id: &str,
    job_kind: &str,
    source_entity_id: impl Into<String>,
    request_label: impl Into<String>,
    price_mud_points: u64,
    payload: &T,
    job_id: String,
    dedupe_key: String,
) -> Result<ExternalGenerationJobRecord, AppError>
where
    T: Serialize,
{
    let request_payload_json = serialize_editor_generation_job_payload(payload)?;
    let job = enqueue_serialized_editor_generation_job_with_identity(
        state,
        owner_user_id,
        job_kind,
        source_entity_id,
        request_label,
        price_mud_points,
        request_payload_json.clone(),
        job_id,
        dedupe_key,
    )
    .await?;
    ensure_editor_generation_job_matches_request(
        job,
        owner_user_id,
        job_kind,
        request_payload_json.as_str(),
        "请求幂等键已用于不同的生成请求，请复用原请求参数或更换请求标识。",
    )
}

#[allow(clippy::too_many_arguments)]
pub(crate) async fn enqueue_external_api_editor_generation_job<T>(
    state: &AppState,
    owner_user_id: &str,
    job_kind: &str,
    source_entity_id: impl Into<String>,
    request_label: impl Into<String>,
    price_mud_points: u64,
    payload: &T,
    idempotency_key: &str,
) -> Result<ExternalGenerationJobRecord, AppError>
where
    T: Serialize,
{
    let request_payload_json = serialize_editor_generation_job_payload(payload)?;
    let dedupe_key = build_editor_generation_dedupe_key(
        EXTERNAL_API_GENERATION_DEDUPE_PREFIX,
        owner_user_id,
        job_kind,
        idempotency_key,
    );
    let requested_job_id = build_prefixed_uuid_id("task-");
    let job = enqueue_serialized_editor_generation_job_with_identity(
        state,
        owner_user_id,
        job_kind,
        source_entity_id,
        request_label,
        price_mud_points,
        request_payload_json.clone(),
        requested_job_id.clone(),
        dedupe_key,
    )
    .await?;

    ensure_external_api_generation_job_matches_request(
        job,
        owner_user_id,
        job_kind,
        request_payload_json.as_str(),
    )
}

fn ensure_external_api_generation_job_matches_request(
    job: ExternalGenerationJobRecord,
    owner_user_id: &str,
    job_kind: &str,
    request_payload_json: &str,
) -> Result<ExternalGenerationJobRecord, AppError> {
    ensure_editor_generation_job_matches_request(
        job,
        owner_user_id,
        job_kind,
        request_payload_json,
        "Idempotency-Key 已用于不同的生成请求，请复用原请求参数或更换幂等键。",
    )
}

fn ensure_editor_generation_job_matches_request(
    job: ExternalGenerationJobRecord,
    owner_user_id: &str,
    job_kind: &str,
    request_payload_json: &str,
    conflict_message: &str,
) -> Result<ExternalGenerationJobRecord, AppError> {
    if job.job_kind != job_kind
        || job.owner_user_id != owner_user_id
        || !editor_generation_request_payloads_match(
            job_kind,
            job.request_payload_json.as_str(),
            request_payload_json,
        )
    {
        return Err(
            AppError::from_status(StatusCode::CONFLICT).with_details(json!({
                "provider": EDITOR_GENERATION_QUEUE_PROVIDER,
                "message": conflict_message,
            })),
        );
    }
    Ok(job)
}

fn generation_input_references(value: &Value) -> Option<&Value> {
    value
        .as_object()
        .and_then(|payload| payload.get("generationInputs"))
        .and_then(Value::as_object)
        .and_then(|generation_inputs| generation_inputs.get("references"))
}

fn strip_untrusted_generation_input_references_from_payload(value: &mut Value) -> bool {
    let Some(generation_inputs) = value
        .as_object_mut()
        .and_then(|payload| payload.get_mut("generationInputs"))
        .and_then(Value::as_object_mut)
    else {
        return false;
    };
    generation_inputs.remove("references").is_some()
}

fn job_kind_migrated_away_from_client_generation_references(job_kind: &str) -> bool {
    matches!(
        job_kind,
        EDITOR_IMAGE_GENERATION_JOB_KIND
            | EDITOR_IMAGE_EDIT_JOB_KIND
            | EDITOR_BACKGROUND_REMOVAL_JOB_KIND
            | EDITOR_ICON_SPRITESHEET_GENERATION_JOB_KIND
            | EDITOR_UI_DESIGN_ASSET_EXTRACTION_JOB_KIND
    )
}

fn editor_generation_request_payloads_match(
    job_kind: &str,
    existing: &str,
    requested: &str,
) -> bool {
    if existing == requested {
        return true;
    }
    let (Ok(mut existing), Ok(requested)) = (
        serde_json::from_str::<Value>(existing),
        serde_json::from_str::<Value>(requested),
    ) else {
        return false;
    };
    // 只兼容部署前旧 payload 有 references、当前 sanitizer 已删除该字段的单向迁移。
    // 音频、视频、角色动作等仍会保留 references；如果当前请求也带该字段，就必须完整
    // 比较，不能把两个不同请求错误复用成同一任务。
    if !job_kind_migrated_away_from_client_generation_references(job_kind)
        || generation_input_references(&existing).is_none()
        || generation_input_references(&requested).is_some()
        || !strip_untrusted_generation_input_references_from_payload(&mut existing)
    {
        return false;
    }
    existing == requested
}

#[allow(clippy::too_many_arguments)]
pub(crate) async fn enqueue_editor_generation_job_for_caller<T>(
    state: &AppState,
    request_context: &RequestContext,
    owner_user_id: &str,
    job_kind: &str,
    source_entity_id: impl Into<String>,
    request_label: impl Into<String>,
    price_mud_points: u64,
    payload: &T,
    external_idempotency_key: Option<&str>,
) -> Result<ExternalGenerationJobRecord, AppError>
where
    T: Serialize,
{
    let source_entity_id = source_entity_id.into();
    let request_label = request_label.into();
    match external_idempotency_key {
        Some(idempotency_key) => {
            enqueue_external_api_editor_generation_job(
                state,
                owner_user_id,
                job_kind,
                source_entity_id,
                request_label,
                price_mud_points,
                payload,
                idempotency_key,
            )
            .await
        }
        None => {
            enqueue_editor_generation_job(
                state,
                request_context,
                owner_user_id,
                job_kind,
                source_entity_id,
                request_label,
                price_mud_points,
                payload,
            )
            .await
        }
    }
}

#[allow(clippy::too_many_arguments)]
async fn enqueue_serialized_editor_generation_job_with_identity(
    state: &AppState,
    owner_user_id: &str,
    job_kind: &str,
    source_entity_id: impl Into<String>,
    request_label: impl Into<String>,
    price_mud_points: u64,
    request_payload_json: String,
    job_id: String,
    dedupe_key: String,
) -> Result<ExternalGenerationJobRecord, AppError> {
    #[cfg(test)]
    if state.record_test_editor_generation_enqueue_attempt() {
        return Err(
            AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
                "provider": EDITOR_GENERATION_QUEUE_PROVIDER,
                "message": "测试已在 SpacetimeDB 写入前截获编辑器生成入队",
            })),
        );
    }
    let now_micros = current_utc_micros();
    state
        .spacetime_client()
        .enqueue_external_generation_job(ExternalGenerationJobEnqueueRecordInput {
            dedupe_key,
            job_id,
            job_kind: job_kind.to_string(),
            owner_user_id: owner_user_id.to_string(),
            source_module: EDITOR_GENERATION_QUEUE_SOURCE_MODULE.to_string(),
            source_entity_id: source_entity_id.into(),
            request_label: request_label.into(),
            request_payload_json,
            max_attempts: 1,
            available_at_micros: now_micros,
            created_at_micros: now_micros,
            price_mud_points,
        })
        .await
        .map_err(|error| {
            AppError::from_status(StatusCode::BAD_GATEWAY).with_details(json!({
                "provider": EDITOR_GENERATION_QUEUE_PROVIDER,
                "message": error.to_string(),
            }))
        })
}

fn serialize_editor_generation_job_payload<T>(payload: &T) -> Result<String, AppError>
where
    T: Serialize + ?Sized,
{
    let payload_value = serde_json::to_value(payload).map_err(payload_serialization_error)?;
    if contains_inline_media_reference(&payload_value) {
        return Err(
            AppError::from_status(StatusCode::BAD_REQUEST).with_details(json!({
                "provider": EDITOR_GENERATION_QUEUE_PROVIDER,
                "message": "编辑器生成任务参数禁止包含 data: 或 blob: 内联媒体引用，请先将媒体上传到对象存储并改传 objectKey 或 resourceId。",
            })),
        );
    }

    let request_payload_json =
        serde_json::to_string(&payload_value).map_err(payload_serialization_error)?;
    let payload_bytes = request_payload_json.len();
    if payload_bytes > MAX_EDITOR_GENERATION_JOB_PAYLOAD_BYTES {
        return Err(
            AppError::from_status(StatusCode::PAYLOAD_TOO_LARGE).with_details(json!({
                "provider": EDITOR_GENERATION_QUEUE_PROVIDER,
                "message": format!(
                    "编辑器生成任务 JSON 大小为 {payload_bytes} 字节，超过持久化上限 {MAX_EDITOR_GENERATION_JOB_PAYLOAD_BYTES} 字节；请移除冗余数据并改传 objectKey 或 resourceId。"
                ),
                "actualBytes": payload_bytes,
                "maxBytes": MAX_EDITOR_GENERATION_JOB_PAYLOAD_BYTES,
            })),
        );
    }

    Ok(request_payload_json)
}

#[cfg(test)]
pub(crate) fn serialize_editor_generation_job_payload_for_test<T>(
    payload: &T,
) -> Result<String, AppError>
where
    T: Serialize + ?Sized,
{
    serialize_editor_generation_job_payload(payload)
}

fn payload_serialization_error(error: serde_json::Error) -> AppError {
    AppError::from_status(StatusCode::INTERNAL_SERVER_ERROR).with_details(json!({
        "provider": EDITOR_GENERATION_QUEUE_PROVIDER,
        "message": format!("编辑器 worker 任务参数序列化失败：{error}"),
    }))
}

fn contains_inline_media_reference(value: &Value) -> bool {
    match value {
        Value::String(value) => is_inline_media_reference(value),
        Value::Array(values) => values.iter().any(contains_inline_media_reference),
        Value::Object(values) => values.iter().any(|(key, value)| {
            is_inline_media_reference(key) || contains_inline_media_reference(value)
        }),
        Value::Null | Value::Bool(_) | Value::Number(_) => false,
    }
}

fn is_inline_media_reference(value: &str) -> bool {
    let prefix = value.trim_start().as_bytes().get(..5);
    prefix.is_some_and(|prefix| {
        prefix.eq_ignore_ascii_case(b"data:") || prefix.eq_ignore_ascii_case(b"blob:")
    })
}

pub(crate) fn editor_generation_queue_state(
    job: ExternalGenerationJobRecord,
) -> ExternalGenerationJobStatusRecord {
    let (status, phase_detail, progress) = match job.status.as_str() {
        "completed" => (ExternalGenerationJobStatus::Completed, "生成已完成。", 100),
        "running" if job.phase.as_deref() == Some("processing") => {
            (ExternalGenerationJobStatus::Running, "正在处理。", 70)
        }
        "running" => (ExternalGenerationJobStatus::Running, "正在生成。", 35),
        "failed" | "cancelled" => (ExternalGenerationJobStatus::Failed, "生成失败。", 0),
        _ => (ExternalGenerationJobStatus::Queued, "排队中。", 8),
    };
    ExternalGenerationJobStatusRecord {
        operation_id: job.job_id,
        status,
        phase_label: job.request_label,
        phase_detail: phase_detail.to_string(),
        progress,
        error: job.last_error_message,
        updated_at_micros: job.updated_at_micros,
    }
}

pub(crate) fn editor_generation_source_entity_id(
    project_id: Option<&str>,
    fallback: &str,
) -> String {
    project_id
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or(fallback)
        .to_string()
}

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

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

    fn queue_job_fixture(status: &str, phase: Option<&str>) -> ExternalGenerationJobRecord {
        ExternalGenerationJobRecord {
            job_id: "task-queue-test".to_string(),
            dedupe_key: "editor-canvas:test:task-queue-test".to_string(),
            job_kind: EDITOR_IMAGE_GENERATION_JOB_KIND.to_string(),
            owner_user_id: "user-1".to_string(),
            source_module: EDITOR_GENERATION_QUEUE_SOURCE_MODULE.to_string(),
            source_entity_id: "project-1".to_string(),
            request_label: "图片画布生成图片".to_string(),
            request_payload_json: "{}".to_string(),
            status: status.to_string(),
            attempt: 0,
            max_attempts: 1,
            last_error_message: None,
            worker_id: None,
            lease_expires_at: None,
            available_at: "2026-07-31T00:00:00Z".to_string(),
            result_payload_json: None,
            created_at: "2026-07-31T00:00:00Z".to_string(),
            started_at: None,
            completed_at: None,
            updated_at: "2026-07-31T00:00:00Z".to_string(),
            updated_at_micros: 1_785_456_000_000_000,
            lease_token: None,
            price_mud_points: 2,
            refund_ledger_id: None,
            notification_acknowledged_at: None,
            notification_acknowledged_at_micros: None,
            phase: phase.map(ToOwned::to_owned),
        }
    }

    #[test]
    fn editor_api_request_dedupe_key_is_stable_and_namespaced() {
        let first = build_editor_generation_dedupe_key(
            EDITOR_API_REQUEST_GENERATION_DEDUPE_PREFIX,
            "user-1",
            EDITOR_IMAGE_GENERATION_JOB_KIND,
            "request-1",
        );
        let replay = build_editor_generation_dedupe_key(
            EDITOR_API_REQUEST_GENERATION_DEDUPE_PREFIX,
            " user-1 ",
            EDITOR_IMAGE_GENERATION_JOB_KIND,
            "request-1",
        );
        let other_owner = build_editor_generation_dedupe_key(
            EDITOR_API_REQUEST_GENERATION_DEDUPE_PREFIX,
            "user-2",
            EDITOR_IMAGE_GENERATION_JOB_KIND,
            "request-1",
        );
        let external = build_editor_generation_dedupe_key(
            EXTERNAL_API_GENERATION_DEDUPE_PREFIX,
            "user-1",
            EDITOR_IMAGE_GENERATION_JOB_KIND,
            "request-1",
        );

        assert_eq!(first, replay);
        assert_ne!(first, other_owner);
        assert_ne!(first, external);
        assert!(first.starts_with("editor-api-request-generation:editor_image_generation:"));
    }

    #[test]
    fn external_api_dedupe_key_preserves_legacy_hash_bytes() {
        let dedupe_key = build_editor_generation_dedupe_key(
            EXTERNAL_API_GENERATION_DEDUPE_PREFIX,
            " user-1 ",
            EDITOR_IMAGE_GENERATION_JOB_KIND,
            " request-1 ",
        );

        assert_eq!(
            dedupe_key,
            "external-api-generation:editor_image_generation:81e38a8eace5f098041b3c240f99053c555ccfa18139295576c9b4ba3d9acff6"
        );
    }

    #[test]
    fn replayed_editor_generation_job_must_match_original_request() {
        let mut job = queue_job_fixture("queued", None);
        job.owner_user_id = "user-1".to_string();
        job.job_kind = EDITOR_IMAGE_GENERATION_JOB_KIND.to_string();
        job.request_payload_json = r#"{"prompt":"same"}"#.to_string();

        assert!(
            ensure_editor_generation_job_matches_request(
                job.clone(),
                "user-1",
                EDITOR_IMAGE_GENERATION_JOB_KIND,
                r#"{"prompt":"same"}"#,
                "幂等冲突",
            )
            .is_ok()
        );
        let error = ensure_editor_generation_job_matches_request(
            job,
            "user-1",
            EDITOR_IMAGE_GENERATION_JOB_KIND,
            r#"{"prompt":"changed"}"#,
            "幂等冲突",
        )
        .expect_err("same request id must reject a different payload");
        assert_eq!(error.status_code(), StatusCode::CONFLICT);
    }

    #[test]
    fn replayed_legacy_external_job_ignores_only_removed_client_references() {
        let mut job = queue_job_fixture("queued", None);
        job.owner_user_id = "user-1".to_string();
        job.job_kind = EDITOR_IMAGE_GENERATION_JOB_KIND.to_string();
        job.request_payload_json = serde_json::to_string(&json!({
            "prompt": "same",
            "generationInputs": {
                "fields": [{"title": "提示词", "value": "same"}],
                "references": [{
                    "title": "旧客户端引用",
                    "refType": "asset",
                    "refId": "asset-forged"
                }]
            }
        }))
        .expect("legacy payload should serialize");
        let requested = serde_json::to_string(&json!({
            "prompt": "same",
            "generationInputs": {
                "fields": [{"title": "提示词", "value": "same"}]
            }
        }))
        .expect("current payload should serialize");

        assert!(
            ensure_editor_generation_job_matches_request(
                job.clone(),
                "user-1",
                EDITOR_IMAGE_GENERATION_JOB_KIND,
                requested.as_str(),
                "幂等冲突",
            )
            .is_ok()
        );

        let changed = requested.replace("same", "changed");
        let error = ensure_editor_generation_job_matches_request(
            job,
            "user-1",
            EDITOR_IMAGE_GENERATION_JOB_KIND,
            changed.as_str(),
            "幂等冲突",
        )
        .expect_err("non-reference payload changes must still conflict");
        assert_eq!(error.status_code(), StatusCode::CONFLICT);
    }

    #[test]
    fn replayed_jobs_with_references_on_both_sides_compare_them_strictly() {
        let mut job = queue_job_fixture("queued", None);
        job.owner_user_id = "user-1".to_string();
        job.job_kind = "editor_video_generation".to_string();
        job.request_payload_json = serde_json::to_string(&json!({
            "prompt": "same",
            "generationInputs": {
                "references": [{"refType": "asset", "refId": "asset-1"}]
            }
        }))
        .expect("existing video payload should serialize");
        let requested = serde_json::to_string(&json!({
            "prompt": "same",
            "generationInputs": {
                "references": [{"refType": "asset", "refId": "asset-2"}]
            }
        }))
        .expect("requested video payload should serialize");

        let error = ensure_editor_generation_job_matches_request(
            job,
            "user-1",
            "editor_video_generation",
            requested.as_str(),
            "幂等冲突",
        )
        .expect_err("different retained references must conflict");
        assert_eq!(error.status_code(), StatusCode::CONFLICT);
    }

    #[test]
    fn replayed_video_job_does_not_use_image_reference_migration_compatibility() {
        let mut job = queue_job_fixture("queued", None);
        job.owner_user_id = "user-1".to_string();
        job.job_kind = EDITOR_VIDEO_GENERATION_JOB_KIND.to_string();
        job.request_payload_json = serde_json::to_string(&json!({
            "prompt": "same",
            "generationInputs": {
                "references": [{"refType": "asset", "refId": "asset-1"}]
            }
        }))
        .expect("existing video payload should serialize");
        let requested = serde_json::to_string(&json!({
            "prompt": "same",
            "generationInputs": {}
        }))
        .expect("requested video payload should serialize");

        let error = ensure_editor_generation_job_matches_request(
            job,
            "user-1",
            EDITOR_VIDEO_GENERATION_JOB_KIND,
            requested.as_str(),
            "幂等冲突",
        )
        .expect_err("video payloads did not migrate away from references");
        assert_eq!(error.status_code(), StatusCode::CONFLICT);
    }

    #[test]
    fn serialize_payload_accepts_persistable_media_references() {
        let payload = json!({
            "sourceImageObjectKey": "users/user-1/editor/source.png",
            "resourceId": "resource-1",
            "nested": [{ "prompt": "保留 data 与 blob 这两个普通单词" }],
        });

        let serialized = serialize_editor_generation_job_payload(&payload)
            .expect("objectKey 和 resourceId 应允许进入持久任务 JSON");

        assert_eq!(
            serde_json::from_str::<serde_json::Value>(&serialized).expect("应生成有效 JSON"),
            payload
        );
    }

    #[test]
    fn serialize_payload_rejects_nested_data_url_case_insensitively() {
        let payload = json!({
            "input": {
                "references": [
                    { "url": "  \nDaTa:image/png;base64,AAAA" }
                ]
            }
        });

        let error = serialize_editor_generation_job_payload(&payload)
            .expect_err("任意层级的 Data URL 都必须被拒绝");

        assert_eq!(error.status_code(), StatusCode::BAD_REQUEST);
        assert!(error.body_text().contains("禁止包含 data: 或 blob:"));
    }

    #[test]
    fn serialize_payload_rejects_nested_blob_url_case_insensitively() {
        let payload = json!({
            "input": [{ "source": { "url": "\tBLOB:https://example.test/id" } }]
        });

        let error = serialize_editor_generation_job_payload(&payload)
            .expect_err("任意层级的 Blob URL 都必须被拒绝");

        assert_eq!(error.status_code(), StatusCode::BAD_REQUEST);
        assert!(error.body_text().contains("禁止包含 data: 或 blob:"));
    }

    #[test]
    fn serialize_payload_rejects_json_larger_than_persistence_limit() {
        let payload = json!({
            "prompt": "x".repeat(MAX_EDITOR_GENERATION_JOB_PAYLOAD_BYTES),
        });

        let error = serialize_editor_generation_job_payload(&payload)
            .expect_err("超过上限的任务 JSON 必须被拒绝");

        assert_eq!(error.status_code().as_u16(), 413);
        assert!(error.body_text().contains("超过持久化上限"));
    }

    #[test]
    fn external_background_music_idempotency_matches_the_raw_payload_exactly() {
        let raw_prompt = "\u{0085}\u{2003}森林音乐\u{00a0}";
        let raw_payload = shared_contracts::assets::EditorBackgroundMusicGenerateRequest {
            gpt_description_prompt: raw_prompt.to_string(),
            make_instrumental: false,
            project_id: Some("project-1".to_string()),
            canvas_completion: None,
            generation_inputs: None,
            asset_folder_id: None,
            asset_label: None,
        };
        let raw_payload_json = serialize_editor_generation_job_payload(&raw_payload)
            .expect("External BGM 原始 payload 应可序列化");
        let mut persisted_job = queue_job_fixture("pending", None);
        persisted_job.job_kind = EDITOR_BACKGROUND_MUSIC_GENERATION_JOB_KIND.to_string();
        persisted_job.request_payload_json = raw_payload_json.clone();

        ensure_external_api_generation_job_matches_request(
            persisted_job.clone(),
            "user-1",
            EDITOR_BACKGROUND_MUSIC_GENERATION_JOB_KIND,
            &raw_payload_json,
        )
        .expect("同一原始 payload 的幂等重放应复用既有任务");

        let canonical_payload = shared_contracts::assets::EditorBackgroundMusicGenerateRequest {
            gpt_description_prompt: "森林音乐".to_string(),
            ..raw_payload
        };
        let canonical_payload_json = serialize_editor_generation_job_payload(&canonical_payload)
            .expect("canonical 等价值应可序列化");
        let error = ensure_external_api_generation_job_matches_request(
            persisted_job,
            "user-1",
            EDITOR_BACKGROUND_MUSIC_GENERATION_JOB_KIND,
            &canonical_payload_json,
        )
        .expect_err("同一幂等键改用 canonical 等价值必须按既有精确语义冲突");

        assert_eq!(error.status_code(), StatusCode::CONFLICT);
        assert!(
            error
                .body_text()
                .contains("Idempotency-Key 已用于不同的生成请求")
        );
    }

    #[test]
    fn queue_state_maps_idempotent_replays_to_the_persisted_status() {
        let cases = [
            (
                "pending",
                None,
                ExternalGenerationJobStatus::Queued,
                "排队中。",
                8,
            ),
            (
                "running",
                Some("generating"),
                ExternalGenerationJobStatus::Running,
                "正在生成。",
                35,
            ),
            (
                "running",
                Some("processing"),
                ExternalGenerationJobStatus::Running,
                "正在处理。",
                70,
            ),
            (
                "completed",
                None,
                ExternalGenerationJobStatus::Completed,
                "生成已完成。",
                100,
            ),
            (
                "failed",
                None,
                ExternalGenerationJobStatus::Failed,
                "生成失败。",
                0,
            ),
        ];

        for (persisted_status, phase, expected_status, expected_detail, expected_progress) in cases
        {
            let state = editor_generation_queue_state(queue_job_fixture(persisted_status, phase));

            assert_eq!(state.operation_id, "task-queue-test");
            assert_eq!(state.status, expected_status, "status={persisted_status}");
            assert_eq!(
                state.phase_detail, expected_detail,
                "status={persisted_status}"
            );
            assert_eq!(
                state.progress, expected_progress,
                "status={persisted_status}"
            );
        }
    }

    #[test]
    fn queue_state_keeps_failed_error_for_idempotent_replay() {
        let mut job = queue_job_fixture("failed", None);
        job.last_error_message = Some("生成失败摘要".to_string());

        let state = editor_generation_queue_state(job);

        assert_eq!(state.status, ExternalGenerationJobStatus::Failed);
        assert_eq!(state.error.as_deref(), Some("生成失败摘要"));
    }
}