mnote/rust/crates/mnote-web/src/routes/media.rs

use crate::app::AppState;
use crate::context::RequestContext;
use crate::error::WebError;
use crate::routes::command_support::{build_runtime_command_plan, runtime_context};
use crate::transport::convex::{
    execute_convex_mutation_by_name, execute_convex_query_by_name,
    persist_runtime_command_artifacts,
};
use axum::extract::{Multipart, Query, State};
use axum::http::{header, HeaderMap};
use axum::response::{IntoResponse, Response};
use axum::{Extension, Json};
use base64::{engine::general_purpose::URL_SAFE_NO_PAD, Engine as _};
use bridge_runtime::{
    build_runtime_command_artifact_plan, RuntimeActorWire, RuntimeCommandEnvelopeWire,
    RuntimeSourceWire, RuntimeTargetWire,
};
use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{SystemTime, UNIX_EPOCH};
use time::format_description::well_known::Rfc3339;
use time::OffsetDateTime;

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

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct MediaSignQuery {
    asset_id: Option<String>,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct FileTreeUploadTargetPreflightPayload {
    workspace_id: Option<String>,
    target_document_id: Option<String>,
    target_row_id: Option<String>,
    focused_row_id: Option<String>,
    active_document_id: Option<String>,
    rows: Option<Vec<FileTreeUploadTargetRow>>,
    document_workspaces: Option<Vec<FileTreeDocumentWorkspace>>,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
#[allow(dead_code)]
pub struct FileTreeUploadTargetRow {
    row_id: Option<String>,
    row_kind: Option<String>,
    document_id: Option<String>,
    asset_id: Option<String>,
    asset_document_id: Option<String>,
}

#[derive(Debug, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct FileTreeDocumentWorkspace {
    document_id: Option<String>,
    workspace_id: Option<String>,
}

#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct FileTreeUploadTargetPlan {
    workspace_id: String,
    target_document_id: String,
    target_mindmap_id: Option<String>,
    target_sub_path: Option<String>,
}

#[derive(Debug)]
struct UploadFile {
    name: String,
    content_type: String,
    bytes: Vec<u8>,
}

async fn current_user_id(state: &AppState, context: &RequestContext) -> String {
    let actor_id = context.auth.actor_id.trim();
    if !actor_id.is_empty() && actor_id != "anonymous" {
        return actor_id.to_string();
    }
    if let Ok(user) = execute_convex_query_by_name(
        state.config(),
        context,
        "users:currentUser",
        json!({}),
        context.workspace.workspace_id.as_deref(),
        "media_current_user",
    )
    .await
    {
        for key in ["_id", "id"] {
            if let Some(user_id) = user
                .get(key)
                .and_then(Value::as_str)
                .map(str::trim)
                .filter(|value| !value.is_empty())
            {
                return user_id.to_string();
            }
        }
    }
    state.config().dev_user_id.clone()
}

fn now_millis() -> u128 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_millis()
}

fn new_asset_id() -> String {
    format!(
        "asset_{}_{}",
        now_millis(),
        UPLOAD_COUNTER.fetch_add(1, Ordering::Relaxed)
    )
}

fn now_iso_like() -> String {
    OffsetDateTime::now_utc()
        .format(&Rfc3339)
        .unwrap_or_else(|_| "1970-01-01T00:00:00Z".into())
}

fn asset_type(mime: &str) -> &'static str {
    if mime.starts_with("image/") {
        "image"
    } else if mime.starts_with("video/") {
        "video"
    } else if mime.starts_with("audio/") {
        "audio"
    } else {
        "file"
    }
}

async fn record_upload_artifacts(
    state: &AppState,
    context: &RequestContext,
    user_id: &str,
    workspace_id: &str,
    document_id: &str,
    asset_id: &str,
    file: &UploadFile,
    asset_kind: &str,
    target_sub_path: Option<&str>,
    created: &Value,
) -> Result<(), WebError> {
    let command = RuntimeCommandEnvelopeWire {
        name: "tree.resource.upload".into(),
        command_id: format!("resource_upload_{}", context.trace.request_id),
        idempotency_key: context.source.idempotency_key.clone(),
        actor: RuntimeActorWire {
            actor_type: context.auth.actor_type.clone(),
            actor_id: user_id.to_string(),
            session_id: context.auth.session_id.clone(),
        },
        source: RuntimeSourceWire {
            channel: context.source.channel.clone(),
            client: context.source.client.clone(),
            source_kind: None,
            root_uri: None,
            workspace_id: Some(workspace_id.to_string()),
            capabilities: Vec::new(),
        },
        target: Some(RuntimeTargetWire {
            workspace_id: Some(workspace_id.to_string()),
            page_id: Some(document_id.to_string()),
            block_id: Some(asset_id.to_string()),
        }),
        payload: json!({
            "assetId": asset_id,
            "workspaceId": workspace_id,
            "targetDocumentId": document_id,
            "targetSubPath": target_sub_path,
            "fileName": file.name,
            "fileSize": file.bytes.len(),
            "mimeType": file.content_type,
            "assetType": asset_kind,
        }),
        preflight_data: None,
        reason: Some("mnote-web media upload tree.resource.upload".into()),
        refs: vec!["file-tree-resource-upload".into()],
        dry_run: false,
        validate_only: false,
    };
    let runtime_context = runtime_context(context, Some(workspace_id));
    let plan = build_runtime_command_plan(context, Some(workspace_id), command.clone())?;
    let artifact_result = json!({
        "items": [created.clone()],
    });
    if let Some(artifacts) = build_runtime_command_artifact_plan(
        &runtime_context,
        &command,
        &plan,
        &artifact_result,
        &now_iso_like(),
    ) {
        persist_runtime_command_artifacts(state.config(), context, &artifacts).await?;
    }
    Ok(())
}

async fn read_upload_multipart(
    mut multipart: Multipart,
) -> Result<(UploadFile, String, String, Option<String>), WebError> {
    let mut file: Option<UploadFile> = None;
    let mut workspace_id = String::new();
    let mut document_id = String::new();
    let mut mindmap_id: Option<String> = None;

    while let Some(field) = multipart.next_field().await.map_err(|error| {
        WebError::bad_request_code(
            "media_upload_bad_multipart",
            format!("上传表单解析失败: {error}"),
        )
    })? {
        let name = field.name().unwrap_or_default().to_string();
        if name == "file" {
            let file_name = field
                .file_name()
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .unwrap_or("附件")
                .to_string();
            let content_type = field
                .content_type()
                .map(str::trim)
                .filter(|value| !value.is_empty())
                .unwrap_or("application/octet-stream")
                .to_string();
            let bytes = field
                .bytes()
                .await
                .map_err(|error| {
                    WebError::bad_request_code(
                        "media_upload_file_read_failed",
                        format!("读取上传文件失败: {error}"),
                    )
                })?
                .to_vec();
            file = Some(UploadFile {
                name: file_name,
                content_type,
                bytes,
            });
            continue;
        }

        let value = field.text().await.map_err(|error| {
            WebError::bad_request_code(
                "media_upload_field_read_failed",
                format!("读取上传字段失败: {error}"),
            )
        })?;
        match name.as_str() {
            "workspaceId" => workspace_id = value.trim().to_string(),
            "documentId" => document_id = value.trim().to_string(),
            "mindmapId" => {
                let trimmed = value.trim();
                if !trimmed.is_empty() {
                    mindmap_id = Some(trimmed.to_string());
                }
            }
            _ => {}
        }
    }

    let file =
        file.ok_or_else(|| WebError::bad_request_code("media_upload_file_missing", "缺少 file"))?;
    if file.bytes.is_empty() || workspace_id.is_empty() || document_id.is_empty() {
        return Err(WebError::bad_request_code(
            "media_upload_required_missing",
            "缺少必要参数",
        ));
    }
    Ok((file, workspace_id, document_id, mindmap_id))
}

fn absolute_origin(headers: &HeaderMap) -> String {
    let proto = headers
        .get("x-forwarded-proto")
        .and_then(|value| value.to_str().ok())
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or("http");
    let host = headers
        .get(header::HOST)
        .and_then(|value| value.to_str().ok())
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .unwrap_or("127.0.0.1:3000");
    format!("{proto}://{host}")
}

fn proxied_file_url(headers: &HeaderMap, raw: &str) -> String {
    let encoded = URL_SAFE_NO_PAD.encode(raw.as_bytes());
    format!(
        "{}/api/onlyoffice/proxy?u={encoded}",
        absolute_origin(headers)
    )
}

fn trim_string(value: Option<&String>) -> Option<String> {
    value
        .map(String::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
}

fn workspace_for_document(
    document_id: &str,
    fallback_workspace_id: Option<&str>,
    document_workspaces: &[FileTreeDocumentWorkspace],
) -> Option<String> {
    for item in document_workspaces {
        if trim_string(item.document_id.as_ref()).as_deref() == Some(document_id) {
            if let Some(workspace_id) = trim_string(item.workspace_id.as_ref()) {
                return Some(workspace_id);
            }
        }
    }
    fallback_workspace_id
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(ToOwned::to_owned)
}

fn document_for_target_row(
    target_row_id: Option<&str>,
    rows: &[FileTreeUploadTargetRow],
) -> Option<String> {
    let row_id = target_row_id?.trim();
    if row_id.is_empty() {
        return None;
    }
    for row in rows {
        if trim_string(row.row_id.as_ref()).as_deref() != Some(row_id) {
            continue;
        }
        if let Some(document_id) = trim_string(row.document_id.as_ref()) {
            return Some(document_id);
        }
        if let Some(document_id) = trim_string(row.asset_document_id.as_ref()) {
            return Some(document_id);
        }
    }
    None
}

pub async fn upload(
    State(state): State<AppState>,
    Extension(context): Extension<RequestContext>,
    multipart: Multipart,
) -> Result<Response, WebError> {
    let (file, workspace_id, document_id, mindmap_id) = read_upload_multipart(multipart).await?;
    let user_id = current_user_id(&state, &context).await;
    let upload_url = execute_convex_mutation_by_name(
        state.config(),
        &context,
        "mediaAssets:generateUploadUrl",
        json!({ "userId": user_id }),
        Some(&workspace_id),
        None,
        "media_upload_generate_url",
    )
    .await?;
    let upload_url = upload_url.as_str().ok_or_else(|| {
        WebError::bad_gateway_code("media_upload_bad_upload_url", "Convex 未返回上传 URL")
    })?;

    let client = reqwest::Client::new();
    let upload_response = client
        .post(upload_url)
        .header(header::CONTENT_TYPE, file.content_type.as_str())
        .body(file.bytes.clone())
        .send()
        .await
        .map_err(|error| {
            WebError::bad_gateway_code(
                "media_upload_storage_failed",
                format!("上传到 Convex Files 失败: {error}"),
            )
        })?;
    let upload_status = upload_response.status();
    let upload_json: Value = upload_response.json().await.map_err(|error| {
        WebError::bad_gateway_code(
            "media_upload_storage_bad_response",
            format!("Convex Files 响应解析失败: {error}"),
        )
        .with_header("x-upstream-status", upload_status.as_u16().to_string())
    })?;
    if !upload_status.is_success() {
        return Err(WebError::bad_gateway_code(
            "media_upload_storage_status",
            format!("上传到 Convex Files 失败: {upload_json}"),
        )
        .with_header("x-upstream-status", upload_status.as_u16().to_string()));
    }
    let storage_id = upload_json
        .get("storageId")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| {
            WebError::bad_gateway_code(
                "media_upload_storage_id_missing",
                "Convex Files 缺少 storageId",
            )
        })?;

    let target_sub_path = mindmap_id
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .map(|value| format!("mindmaps/{value}"));
    let id = new_asset_id();
    let kind = asset_type(&file.content_type);
    let created = execute_convex_mutation_by_name(
        state.config(),
        &context,
        "mediaAssets:createWithStorage",
        json!({
            "userId": user_id,
            "storageId": storage_id,
            "targetSubPath": target_sub_path,
            "asset": {
                "id": id,
                "workspace_id": workspace_id,
                "document_id": document_id,
                "asset_type": kind,
                "file_name": file.name,
                "file_size": file.bytes.len(),
                "mime_type": file.content_type,
            }
        }),
        Some(&workspace_id),
        None,
        "media_upload_create_asset",
    )
    .await?;

    let asset_id = created
        .get("id")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_string();
    if let Err(error) = record_upload_artifacts(
        &state,
        &context,
        &user_id,
        &workspace_id,
        &document_id,
        &asset_id,
        &file,
        &kind,
        target_sub_path.as_deref(),
        &created,
    )
    .await
    {
        tracing::warn!(
            error = %error.message(),
            asset_id = %asset_id,
            "media upload tree.resource.upload artifacts 记录失败，主上传结果继续返回"
        );
    }
    Ok(Json(json!({
        "asset": created,
        "mindmapUrl": format!("asset:{asset_id}"),
    }))
    .into_response())
}

pub async fn filetree_upload_target_preflight(
    Extension(context): Extension<RequestContext>,
    Json(payload): Json<FileTreeUploadTargetPreflightPayload>,
) -> Result<Response, WebError> {
    let rows = payload.rows.unwrap_or_default();
    let document_workspaces = payload.document_workspaces.unwrap_or_default();
    let fallback_workspace_id = trim_string(payload.workspace_id.as_ref());
    let target_row_document_id = document_for_target_row(
        trim_string(payload.target_row_id.as_ref()).as_deref(),
        &rows,
    );
    let document_id = trim_string(payload.target_document_id.as_ref())
        .or(target_row_document_id)
        .or_else(|| {
            trim_string(payload.focused_row_id.as_ref())
                .and_then(|row_id| document_for_target_row(Some(&row_id), &rows))
        })
        .or_else(|| trim_string(payload.active_document_id.as_ref()))
        .ok_or_else(|| {
            WebError::bad_request_code(
                "filetree_upload_target_document_missing",
                "请选择一个目标页面后再上传文件",
            )
        })?;
    let workspace_id = workspace_for_document(
        &document_id,
        fallback_workspace_id
            .as_deref()
            .or(context.workspace.workspace_id.as_deref()),
        &document_workspaces,
    )
    .ok_or_else(|| {
        WebError::bad_request_code(
            "filetree_upload_target_workspace_missing",
            "缺少 workspaceId",
        )
    })?;
    let plan = FileTreeUploadTargetPlan {
        workspace_id,
        target_document_id: document_id,
        target_mindmap_id: None,
        target_sub_path: None,
    };
    Ok(Json(json!({
        "requestId": context.trace.request_id,
        "traceId": context.trace.trace_id,
        "plan": plan,
    }))
    .into_response())
}

pub async fn sign(
    State(state): State<AppState>,
    Extension(context): Extension<RequestContext>,
    Query(query): Query<MediaSignQuery>,
    headers: HeaderMap,
) -> Result<Response, WebError> {
    let asset_id = query
        .asset_id
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| WebError::bad_request_code("media_sign_asset_missing", "缺少 assetId"))?;
    let user_id = current_user_id(&state, &context).await;
    let asset = execute_convex_query_by_name(
        state.config(),
        &context,
        "mediaAssets:getById",
        json!({ "userId": user_id, "id": asset_id }),
        None,
        "media_sign_get_asset",
    )
    .await?;
    if asset.is_null() {
        return Err(WebError::new(
            axum::http::StatusCode::NOT_FOUND,
            "media_asset_not_found",
            "资源不存在",
        ));
    }
    let refreshed = execute_convex_mutation_by_name(
        state.config(),
        &context,
        "mediaAssets:refreshUrl",
        json!({ "userId": user_id, "id": asset_id }),
        None,
        None,
        "media_sign_refresh_url",
    )
    .await?;
    let signed_url = refreshed
        .get("signedUrl")
        .and_then(Value::as_str)
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| WebError::bad_gateway_code("media_sign_url_missing", "生成签名链接失败"))?;

    Ok(Json(json!({
        "signedUrl": proxied_file_url(&headers, signed_url),
        "asset": {
            "id": asset.get("id").cloned().unwrap_or(Value::Null),
            "document_id": asset.get("document_id").cloned().unwrap_or(Value::Null),
            "workspace_id": asset.get("workspace_id").cloned().unwrap_or(Value::Null),
            "file_name": asset.get("file_name").cloned().unwrap_or(Value::Null),
            "mime_type": asset.get("mime_type").cloned().unwrap_or(Value::Null),
            "file_size": asset.get("file_size").cloned().unwrap_or(Value::Null),
            "storage_id": asset.get("storage_id").cloned().unwrap_or(Value::Null),
            "updated_at": asset.get("updated_at").cloned().unwrap_or(Value::Null),
        }
    }))
    .into_response())
}