Genarrative/server-rs/crates/spacetime-module/src/asset_metadata/objects.rs

use crate::*;

const ASSET_HISTORY_MAX_LIMIT: usize = 120;
const ASSET_HISTORY_CHARACTER_VISUAL_KIND: &str = "character_visual";
const ASSET_HISTORY_SCENE_IMAGE_KIND: &str = "scene_image";
const ASSET_HISTORY_PUZZLE_COVER_IMAGE_KIND: &str = "puzzle_cover_image";
const ASSET_HISTORY_SQUARE_HOLE_COVER_IMAGE_KIND: &str = "square_hole_cover_image";
const ASSET_HISTORY_SQUARE_HOLE_BACKGROUND_IMAGE_KIND: &str = "square_hole_background_image";
const ASSET_HISTORY_SQUARE_HOLE_SHAPE_IMAGE_KIND: &str = "square_hole_shape_image";
const ASSET_HISTORY_SQUARE_HOLE_HOLE_IMAGE_KIND: &str = "square_hole_hole_image";

/// 资产事件类型。
///
/// 事件表只承接订阅端和审计所需的轻量事实，正式资产状态仍以
/// `asset_object` 和 `asset_entity_binding` 为准。
#[derive(Clone, Copy, Debug, PartialEq, Eq, SpacetimeType)]
pub enum AssetEventKind {
    ObjectConfirmed,
    EntityBindingChanged,
}

#[spacetimedb::table(
    accessor = asset_event,
    public,
    event,
    index(accessor = by_asset_event_asset_object_id, btree(columns = [asset_object_id])),
    index(accessor = by_asset_event_owner_user_id, btree(columns = [owner_user_id])),
    index(accessor = by_asset_event_profile_id, btree(columns = [profile_id]))
)]
pub struct AssetEvent {
    #[primary_key]
    pub(crate) event_id: String,
    pub(crate) asset_object_id: String,
    pub(crate) binding_id: Option<String>,
    pub(crate) event_kind: AssetEventKind,
    pub(crate) asset_kind: String,
    pub(crate) owner_user_id: Option<String>,
    pub(crate) profile_id: Option<String>,
    pub(crate) entity_kind: Option<String>,
    pub(crate) entity_id: Option<String>,
    pub(crate) slot: Option<String>,
    pub(crate) occurred_at: Timestamp,
}

#[spacetimedb::table(
    accessor = asset_object,
    index(accessor = by_bucket_object_key, btree(columns = [bucket, object_key]))
)]
pub struct AssetObject {
    #[primary_key]
    asset_object_id: String,
    // 正式对象定位固定拆成 bucket + object_key 两列，避免后续再从单字符串路径做 schema 拆分。
    bucket: String,
    object_key: String,
    access_policy: AssetObjectAccessPolicy,
    content_type: Option<String>,
    content_length: u64,
    content_hash: Option<String>,
    version: u32,
    source_job_id: Option<String>,
    owner_user_id: Option<String>,
    profile_id: Option<String>,
    entity_id: Option<String>,
    #[index(btree)]
    asset_kind: String,
    created_at: Timestamp,
    updated_at: Timestamp,
}

// reducer 负责固定资产对象的正式写规则，供后续内部模块逻辑复用。
#[spacetimedb::reducer]
pub fn confirm_asset_object(
    ctx: &ReducerContext,
    input: AssetObjectUpsertInput,
) -> Result<(), String> {
    upsert_asset_object(ctx, input).map(|_| ())
}

// procedure 面向 Axum 同步确认接口，返回最终持久化后的对象记录，避免 HTTP 层再额外查询 private table。
#[spacetimedb::procedure]
pub fn confirm_asset_object_and_return(
    ctx: &mut ProcedureContext,
    input: AssetObjectUpsertInput,
) -> AssetObjectProcedureResult {
    match ctx.try_with_tx(|tx| upsert_asset_object(tx, input.clone())) {
        Ok(record) => AssetObjectProcedureResult {
            ok: true,
            record: Some(record),
            error_message: None,
        },
        Err(message) => AssetObjectProcedureResult {
            ok: false,
            record: None,
            error_message: Some(message),
        },
    }
}

// ACL 等安全判断必须读取事务内真相，不能把订阅 cache miss 当成对象不存在。
#[spacetimedb::procedure]
pub fn get_asset_object_by_location_and_return(
    ctx: &mut ProcedureContext,
    input: AssetObjectLocationInput,
) -> AssetObjectProcedureResult {
    let caller = ctx.sender();
    match ctx.try_with_tx(|tx| {
        crate::editor_project_storage::require_editor_generation_runtime_service_identity(
            tx, caller,
        )?;
        find_asset_object_by_location(tx, &input)
    }) {
        Ok(record) => AssetObjectProcedureResult {
            ok: true,
            record,
            error_message: None,
        },
        Err(message) => AssetObjectProcedureResult {
            ok: false,
            record: None,
            error_message: Some(message),
        },
    }
}

// 公开授权与资产 metadata 必须在同一事务快照中判断，不能依赖 API 连接池的订阅水位。
#[spacetimedb::procedure]
pub fn get_asset_read_access_by_location_and_return(
    ctx: &mut ProcedureContext,
    input: AssetObjectLocationInput,
) -> AssetObjectReadAccessProcedureResult {
    let caller = ctx.sender();
    match ctx.try_with_tx(|tx| {
        crate::editor_project_storage::require_editor_generation_runtime_service_identity(
            tx, caller,
        )?;
        let record = find_asset_object_by_location(tx, &input)?;
        let public_work_granted =
            crate::editor_project_storage::asset_location_has_public_showcase_read_grant(
                tx,
                input.object_key.as_str(),
                record.as_ref(),
            );
        Ok((record, public_work_granted))
    }) {
        Ok((record, public_work_granted)) => AssetObjectReadAccessProcedureResult {
            ok: true,
            record,
            public_work_granted,
            error_message: None,
        },
        Err(message) => AssetObjectReadAccessProcedureResult {
            ok: false,
            record: None,
            public_work_granted: false,
            error_message: Some(message),
        },
    }
}

#[spacetimedb::procedure]
pub fn get_asset_object_by_id_and_return(
    ctx: &mut ProcedureContext,
    asset_object_id: String,
) -> AssetObjectProcedureResult {
    let caller = ctx.sender();
    match ctx.try_with_tx(|tx| {
        crate::editor_project_storage::require_editor_generation_runtime_service_identity(
            tx, caller,
        )?;
        let asset_object_id = asset_object_id.trim();
        if asset_object_id.is_empty() {
            return Err("asset_object_id 不能为空".to_string());
        }
        Ok(tx
            .db
            .asset_object()
            .asset_object_id()
            .find(&asset_object_id.to_string())
            .map(asset_object_snapshot))
    }) {
        Ok(record) => AssetObjectProcedureResult {
            ok: true,
            record,
            error_message: None,
        },
        Err(message) => AssetObjectProcedureResult {
            ok: false,
            record: None,
            error_message: Some(message),
        },
    }
}

// 历史素材只返回编辑器复用所需的脱敏字段，asset_object 本表继续保持 private。
#[spacetimedb::procedure]
pub fn list_asset_history_and_return(
    ctx: &mut ProcedureContext,
    input: AssetHistoryListInput,
) -> AssetHistoryListResult {
    match ctx.try_with_tx(|tx| list_asset_history(tx, input.clone())) {
        Ok(entries) => AssetHistoryListResult {
            ok: true,
            entries,
            error_message: None,
        },
        Err(message) => AssetHistoryListResult {
            ok: false,
            entries: Vec::new(),
            error_message: Some(message),
        },
    }
}

pub(crate) fn upsert_asset_object(
    ctx: &ReducerContext,
    input: AssetObjectUpsertInput,
) -> Result<AssetObjectUpsertSnapshot, String> {
    validate_asset_object_fields(
        &input.bucket,
        &input.object_key,
        &input.asset_kind,
        input.version,
    )
    .map_err(|error| error.to_string())?;

    let current = ctx
        .db
        .asset_object()
        .by_bucket_object_key()
        .filter((input.bucket.as_str(), input.object_key.as_str()))
        .next();

    let snapshot = match current {
        Some(existing) => {
            require_matching_asset_object_owner(
                existing.owner_user_id.as_deref(),
                input.owner_user_id.as_deref(),
            )?;
            ctx.db
                .asset_object()
                .asset_object_id()
                .delete(&existing.asset_object_id);
            let snapshot = AssetObjectUpsertSnapshot {
                asset_object_id: existing.asset_object_id.clone(),
                bucket: input.bucket.clone(),
                object_key: input.object_key.clone(),
                access_policy: input.access_policy,
                content_type: input.content_type.clone(),
                content_length: input.content_length,
                content_hash: input.content_hash.clone(),
                version: input.version,
                source_job_id: input.source_job_id.clone(),
                owner_user_id: input.owner_user_id.clone(),
                profile_id: input.profile_id.clone(),
                entity_id: input.entity_id.clone(),
                asset_kind: input.asset_kind.clone(),
                created_at_micros: existing.created_at.to_micros_since_unix_epoch(),
                updated_at_micros: input.updated_at_micros,
            };
            ctx.db
                .asset_object()
                .insert(build_asset_object_row(&snapshot));
            snapshot
        }
        None => {
            let snapshot = AssetObjectUpsertSnapshot {
                asset_object_id: input.asset_object_id.clone(),
                bucket: input.bucket.clone(),
                object_key: input.object_key.clone(),
                access_policy: input.access_policy,
                content_type: input.content_type.clone(),
                content_length: input.content_length,
                content_hash: input.content_hash.clone(),
                version: input.version,
                source_job_id: input.source_job_id.clone(),
                owner_user_id: input.owner_user_id.clone(),
                profile_id: input.profile_id.clone(),
                entity_id: input.entity_id.clone(),
                asset_kind: input.asset_kind.clone(),
                created_at_micros: input.updated_at_micros,
                updated_at_micros: input.updated_at_micros,
            };
            ctx.db
                .asset_object()
                .insert(build_asset_object_row(&snapshot));
            snapshot
        }
    };

    emit_asset_object_confirmed_event(ctx, &snapshot);

    Ok(snapshot)
}

pub(crate) fn find_asset_object_by_location(
    ctx: &ReducerContext,
    input: &AssetObjectLocationInput,
) -> Result<Option<AssetObjectUpsertSnapshot>, String> {
    let bucket = input.bucket.trim();
    let object_key = input.object_key.trim().trim_start_matches('/');
    if bucket.is_empty() || object_key.is_empty() {
        return Err("bucket 与 object_key 不能为空".to_string());
    }

    let mut matches = ctx
        .db
        .asset_object()
        .by_bucket_object_key()
        .filter((bucket, object_key));
    resolve_unique_asset_object_location(&mut matches)
}

fn resolve_unique_asset_object_location(
    matches: &mut impl Iterator<Item = AssetObject>,
) -> Result<Option<AssetObjectUpsertSnapshot>, String> {
    let record = matches.next();
    if matches.next().is_some() {
        return Err("asset_object 的 bucket + object_key 存在重复记录".to_string());
    }
    Ok(record.map(asset_object_snapshot))
}

fn asset_object_snapshot(row: AssetObject) -> AssetObjectUpsertSnapshot {
    AssetObjectUpsertSnapshot {
        asset_object_id: row.asset_object_id,
        bucket: row.bucket,
        object_key: row.object_key,
        access_policy: row.access_policy,
        content_type: row.content_type,
        content_length: row.content_length,
        content_hash: row.content_hash,
        version: row.version,
        source_job_id: row.source_job_id,
        owner_user_id: row.owner_user_id,
        profile_id: row.profile_id,
        entity_id: row.entity_id,
        asset_kind: row.asset_kind,
        created_at_micros: row.created_at.to_micros_since_unix_epoch(),
        updated_at_micros: row.updated_at.to_micros_since_unix_epoch(),
    }
}

fn require_matching_asset_object_owner(
    existing_owner_user_id: Option<&str>,
    next_owner_user_id: Option<&str>,
) -> Result<(), String> {
    let existing_owner = existing_owner_user_id
        .map(str::trim)
        .filter(|value| !value.is_empty());
    let next_owner = next_owner_user_id
        .map(str::trim)
        .filter(|value| !value.is_empty());
    if existing_owner == next_owner {
        Ok(())
    } else {
        Err("已登记资产对象不能变更 owner_user_id".to_string())
    }
}

pub(crate) fn has_asset_object(ctx: &ReducerContext, asset_object_id: &str) -> bool {
    ctx.db
        .asset_object()
        .asset_object_id()
        .find(&asset_object_id.to_string())
        .is_some()
}

pub(crate) fn find_asset_object_snapshot_by_id(
    ctx: &ReducerContext,
    asset_object_id: &str,
) -> Option<AssetObjectUpsertSnapshot> {
    ctx.db
        .asset_object()
        .asset_object_id()
        .find(&asset_object_id.trim().to_string())
        .map(asset_object_snapshot)
}

fn list_asset_history(
    ctx: &ReducerContext,
    input: AssetHistoryListInput,
) -> Result<Vec<AssetHistoryEntrySnapshot>, String> {
    let asset_kind = input.asset_kind.trim();
    if asset_kind != ASSET_HISTORY_CHARACTER_VISUAL_KIND
        && asset_kind != ASSET_HISTORY_SCENE_IMAGE_KIND
        && asset_kind != ASSET_HISTORY_PUZZLE_COVER_IMAGE_KIND
        && asset_kind != ASSET_HISTORY_SQUARE_HOLE_COVER_IMAGE_KIND
        && asset_kind != ASSET_HISTORY_SQUARE_HOLE_BACKGROUND_IMAGE_KIND
        && asset_kind != ASSET_HISTORY_SQUARE_HOLE_SHAPE_IMAGE_KIND
        && asset_kind != ASSET_HISTORY_SQUARE_HOLE_HOLE_IMAGE_KIND
    {
        return Err(
            "历史素材类型只支持 character_visual、scene_image、puzzle_cover_image、square_hole_cover_image、square_hole_background_image、square_hole_shape_image 或 square_hole_hole_image".to_string(),
        );
    }

    let limit = usize::try_from(input.limit)
        .unwrap_or(ASSET_HISTORY_MAX_LIMIT)
        .clamp(1, ASSET_HISTORY_MAX_LIMIT);
    let mut entries = ctx
        .db
        .asset_object()
        .asset_kind()
        .filter(&asset_kind.to_string())
        .map(|row| AssetHistoryEntrySnapshot {
            asset_object_id: row.asset_object_id,
            asset_kind: row.asset_kind,
            image_src: object_key_to_legacy_image_src(row.object_key.as_str()),
            owner_user_id: row.owner_user_id,
            profile_id: row.profile_id,
            entity_id: row.entity_id,
            created_at_micros: row.created_at.to_micros_since_unix_epoch(),
            updated_at_micros: row.updated_at.to_micros_since_unix_epoch(),
        })
        .collect::<Vec<_>>();

    entries.sort_by(|left, right| {
        right
            .created_at_micros
            .cmp(&left.created_at_micros)
            .then_with(|| right.asset_object_id.cmp(&left.asset_object_id))
    });
    entries.truncate(limit);
    Ok(entries)
}

fn object_key_to_legacy_image_src(object_key: &str) -> String {
    let normalized = object_key.trim().trim_start_matches('/');
    if normalized.is_empty() {
        return String::new();
    }
    format!("/{normalized}")
}

fn build_asset_object_row(snapshot: &AssetObjectUpsertSnapshot) -> AssetObject {
    AssetObject {
        asset_object_id: snapshot.asset_object_id.clone(),
        bucket: snapshot.bucket.clone(),
        object_key: snapshot.object_key.clone(),
        access_policy: snapshot.access_policy,
        content_type: snapshot.content_type.clone(),
        content_length: snapshot.content_length,
        content_hash: snapshot.content_hash.clone(),
        version: snapshot.version,
        source_job_id: snapshot.source_job_id.clone(),
        owner_user_id: snapshot.owner_user_id.clone(),
        profile_id: snapshot.profile_id.clone(),
        entity_id: snapshot.entity_id.clone(),
        asset_kind: snapshot.asset_kind.clone(),
        created_at: Timestamp::from_micros_since_unix_epoch(snapshot.created_at_micros),
        updated_at: Timestamp::from_micros_since_unix_epoch(snapshot.updated_at_micros),
    }
}

pub(crate) fn emit_asset_object_confirmed_event(
    ctx: &ReducerContext,
    snapshot: &AssetObjectUpsertSnapshot,
) {
    let event = AssetObjectConfirmedEvent {
        asset_object_id: snapshot.asset_object_id.clone(),
        asset_kind: snapshot.asset_kind.clone(),
        owner_user_id: snapshot.owner_user_id.clone(),
        profile_id: snapshot.profile_id.clone(),
        entity_id: snapshot.entity_id.clone(),
        occurred_at_micros: snapshot.updated_at_micros,
    };

    ctx.db.asset_event().insert(AssetEvent {
        event_id: format!(
            "assetevt_{}_{}_confirmed",
            event.asset_object_id, event.occurred_at_micros
        ),
        asset_object_id: event.asset_object_id,
        binding_id: None,
        event_kind: AssetEventKind::ObjectConfirmed,
        asset_kind: event.asset_kind,
        owner_user_id: event.owner_user_id,
        profile_id: event.profile_id,
        entity_kind: None,
        entity_id: event.entity_id,
        slot: None,
        occurred_at: Timestamp::from_micros_since_unix_epoch(event.occurred_at_micros),
    });
}

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

    fn asset_object(asset_object_id: &str) -> AssetObject {
        AssetObject {
            asset_object_id: asset_object_id.to_string(),
            bucket: "genarrative-assets".to_string(),
            object_key: "generated-puzzle-assets/published.png".to_string(),
            access_policy: AssetObjectAccessPolicy::Private,
            content_type: Some("image/png".to_string()),
            content_length: 1,
            content_hash: None,
            version: 1,
            source_job_id: None,
            owner_user_id: Some("owner-a".to_string()),
            profile_id: Some("profile-a".to_string()),
            entity_id: None,
            asset_kind: "puzzle_cover_image".to_string(),
            created_at: Timestamp::from_micros_since_unix_epoch(1),
            updated_at: Timestamp::from_micros_since_unix_epoch(1),
        }
    }

    #[test]
    fn authoritative_location_lookup_fails_closed_on_duplicate_rows() {
        let record = asset_object("assetobj-first");
        let mut one = vec![record].into_iter();
        assert_eq!(
            resolve_unique_asset_object_location(&mut one)
                .expect("one exact row should resolve")
                .expect("one exact row should exist")
                .asset_object_id,
            "assetobj-first"
        );

        let mut duplicates = vec![
            asset_object("assetobj-first"),
            asset_object("assetobj-second"),
        ]
        .into_iter();
        assert!(resolve_unique_asset_object_location(&mut duplicates).is_err());
    }

    #[test]
    fn registered_asset_object_owner_is_immutable() {
        assert!(require_matching_asset_object_owner(Some("owner-a"), Some("owner-a")).is_ok());
        assert!(require_matching_asset_object_owner(None, None).is_ok());
        assert!(require_matching_asset_object_owner(Some("owner-a"), Some("owner-b")).is_err());
        assert!(require_matching_asset_object_owner(Some("owner-a"), None).is_err());
        assert!(require_matching_asset_object_owner(None, Some("owner-a")).is_err());
    }
}