use crate::big_fish::tables::{
    big_fish_agent_message, big_fish_creation_session, big_fish_runtime_run,
};
use crate::runtime::{
    ProfilePlayedWorkUpsertInput, PublicWorkLikeRecordInput, PublicWorkPlayRecordInput,
    add_profile_observed_play_time, count_recent_public_work_plays, record_public_work_like,
    record_public_work_play, upsert_profile_played_work,
};
use crate::*;
use module_big_fish::{EvaluateBigFishPublishReadinessCommand, evaluate_publish_readiness};

const INITIAL_BIG_FISH_CREATION_PROGRESS_PERCENT: u32 = 0;

#[spacetimedb::procedure]
pub fn create_big_fish_session(
    ctx: &mut ProcedureContext,
    input: BigFishSessionCreateInput,
) -> BigFishSessionProcedureResult {
    match ctx.try_with_tx(|tx| create_big_fish_session_tx(tx, input.clone())) {
        Ok(session) => BigFishSessionProcedureResult {
            ok: true,
            session: Some(session),
            error_message: None,
        },
        Err(message) => BigFishSessionProcedureResult {
            ok: false,
            session: None,
            error_message: Some(message),
        },
    }
}

#[spacetimedb::procedure]
pub fn get_big_fish_session(
    ctx: &mut ProcedureContext,
    input: BigFishSessionGetInput,
) -> BigFishSessionProcedureResult {
    match ctx.try_with_tx(|tx| get_big_fish_session_tx(tx, input.clone())) {
        Ok(session) => BigFishSessionProcedureResult {
            ok: true,
            session: Some(session),
            error_message: None,
        },
        Err(message) => BigFishSessionProcedureResult {
            ok: false,
            session: None,
            error_message: Some(message),
        },
    }
}

#[spacetimedb::procedure]
pub fn list_big_fish_works(
    ctx: &mut ProcedureContext,
    input: BigFishWorksListInput,
) -> BigFishWorksProcedureResult {
    match ctx.try_with_tx(|tx| list_big_fish_works_tx(tx, input.clone())) {
        Ok(items) => match serde_json::to_string(&items) {
            Ok(items_json) => BigFishWorksProcedureResult {
                ok: true,
                items_json: Some(items_json),
                error_message: None,
            },
            Err(error) => BigFishWorksProcedureResult {
                ok: false,
                items_json: None,
                error_message: Some(error.to_string()),
            },
        },
        Err(message) => BigFishWorksProcedureResult {
            ok: false,
            items_json: None,
            error_message: Some(message),
        },
    }
}

#[spacetimedb::procedure]
pub fn delete_big_fish_work(
    ctx: &mut ProcedureContext,
    input: BigFishWorkDeleteInput,
) -> BigFishWorksProcedureResult {
    match ctx.try_with_tx(|tx| delete_big_fish_work_tx(tx, input.clone())) {
        Ok(items) => match serde_json::to_string(&items) {
            Ok(items_json) => BigFishWorksProcedureResult {
                ok: true,
                items_json: Some(items_json),
                error_message: None,
            },
            Err(error) => BigFishWorksProcedureResult {
                ok: false,
                items_json: None,
                error_message: Some(error.to_string()),
            },
        },
        Err(message) => BigFishWorksProcedureResult {
            ok: false,
            items_json: None,
            error_message: Some(message),
        },
    }
}

#[spacetimedb::procedure]
pub fn record_big_fish_play(
    ctx: &mut ProcedureContext,
    input: BigFishPlayRecordInput,
) -> BigFishWorksProcedureResult {
    match ctx.try_with_tx(|tx| record_big_fish_play_tx(tx, input.clone())) {
        Ok(items) => match serde_json::to_string(&items) {
            Ok(items_json) => BigFishWorksProcedureResult {
                ok: true,
                items_json: Some(items_json),
                error_message: None,
            },
            Err(error) => BigFishWorksProcedureResult {
                ok: false,
                items_json: None,
                error_message: Some(error.to_string()),
            },
        },
        Err(message) => BigFishWorksProcedureResult {
            ok: false,
            items_json: None,
            error_message: Some(message),
        },
    }
}

#[spacetimedb::procedure]
pub fn record_big_fish_like(
    ctx: &mut ProcedureContext,
    input: BigFishWorkLikeRecordInput,
) -> BigFishWorksProcedureResult {
    match ctx.try_with_tx(|tx| record_big_fish_like_tx(tx, input.clone())) {
        Ok(items) => match serde_json::to_string(&items) {
            Ok(items_json) => BigFishWorksProcedureResult {
                ok: true,
                items_json: Some(items_json),
                error_message: None,
            },
            Err(error) => BigFishWorksProcedureResult {
                ok: false,
                items_json: None,
                error_message: Some(error.to_string()),
            },
        },
        Err(message) => BigFishWorksProcedureResult {
            ok: false,
            items_json: None,
            error_message: Some(message),
        },
    }
}

#[spacetimedb::procedure]
pub fn remix_big_fish_work(
    ctx: &mut ProcedureContext,
    input: BigFishWorkRemixInput,
) -> BigFishSessionProcedureResult {
    match ctx.try_with_tx(|tx| remix_big_fish_work_tx(tx, input.clone())) {
        Ok(session) => BigFishSessionProcedureResult {
            ok: true,
            session: Some(session),
            error_message: None,
        },
        Err(message) => BigFishSessionProcedureResult {
            ok: false,
            session: None,
            error_message: Some(message),
        },
    }
}

#[spacetimedb::procedure]
pub fn submit_big_fish_message(
    ctx: &mut ProcedureContext,
    input: BigFishMessageSubmitInput,
) -> BigFishSessionProcedureResult {
    match ctx.try_with_tx(|tx| submit_big_fish_message_tx(tx, input.clone())) {
        Ok(session) => BigFishSessionProcedureResult {
            ok: true,
            session: Some(session),
            error_message: None,
        },
        Err(message) => BigFishSessionProcedureResult {
            ok: false,
            session: None,
            error_message: Some(message),
        },
    }
}

#[spacetimedb::procedure]
pub fn finalize_big_fish_agent_message_turn(
    ctx: &mut ProcedureContext,
    input: BigFishMessageFinalizeInput,
) -> BigFishSessionProcedureResult {
    match ctx.try_with_tx(|tx| finalize_big_fish_agent_message_turn_tx(tx, input.clone())) {
        Ok(session) => BigFishSessionProcedureResult {
            ok: true,
            session: Some(session),
            error_message: None,
        },
        Err(message) => BigFishSessionProcedureResult {
            ok: false,
            session: None,
            error_message: Some(message),
        },
    }
}

#[spacetimedb::procedure]
pub fn compile_big_fish_draft(
    ctx: &mut ProcedureContext,
    input: BigFishDraftCompileInput,
) -> BigFishSessionProcedureResult {
    match ctx.try_with_tx(|tx| compile_big_fish_draft_tx(tx, input.clone())) {
        Ok(session) => BigFishSessionProcedureResult {
            ok: true,
            session: Some(session),
            error_message: None,
        },
        Err(message) => BigFishSessionProcedureResult {
            ok: false,
            session: None,
            error_message: Some(message),
        },
    }
}

pub(crate) fn create_big_fish_session_tx(
    ctx: &ReducerContext,
    input: BigFishSessionCreateInput,
) -> Result<BigFishSessionSnapshot, String> {
    validate_session_create_input(&input).map_err(|error| error.to_string())?;
    if ctx
        .db
        .big_fish_creation_session()
        .session_id()
        .find(&input.session_id)
        .is_some()
    {
        return Err("big_fish_creation_session.session_id 已存在".to_string());
    }
    if ctx
        .db
        .big_fish_agent_message()
        .message_id()
        .find(&input.welcome_message_id)
        .is_some()
    {
        return Err("big_fish_agent_message.message_id 已存在".to_string());
    }

    let created_at = Timestamp::from_micros_since_unix_epoch(input.created_at_micros);
    let anchor_pack = infer_anchor_pack(&input.seed_text, None);
    let asset_coverage = build_asset_coverage(None, &[]);
    ctx.db
        .big_fish_creation_session()
        .insert(BigFishCreationSession {
            session_id: input.session_id.clone(),
            owner_user_id: input.owner_user_id.clone(),
            seed_text: input.seed_text.trim().to_string(),
            current_turn: 0,
            // 中文注释：欢迎语和初始锚点只建立工作台上下文，不能提前抬高创作进度。
            progress_percent: INITIAL_BIG_FISH_CREATION_PROGRESS_PERCENT,
            stage: BigFishCreationStage::CollectingAnchors,
            anchor_pack_json: serialize_anchor_pack(&anchor_pack)
                .map_err(|error| error.to_string())?,
            draft_json: None,
            asset_coverage_json: serialize_asset_coverage(&asset_coverage)
                .map_err(|error| error.to_string())?,
            last_assistant_reply: Some(input.welcome_message_text.clone()),
            publish_ready: false,
            play_count: 0,
            remix_count: 0,
            like_count: 0,
            published_at: None,
            created_at,
            updated_at: created_at,
        });
    ctx.db.big_fish_agent_message().insert(BigFishAgentMessage {
        message_id: input.welcome_message_id,
        session_id: input.session_id.clone(),
        role: BigFishAgentMessageRole::Assistant,
        kind: BigFishAgentMessageKind::Chat,
        text: input.welcome_message_text,
        created_at,
    });

    get_big_fish_session_tx(
        ctx,
        BigFishSessionGetInput {
            session_id: input.session_id,
            owner_user_id: input.owner_user_id,
        },
    )
}

pub(crate) fn get_big_fish_session_tx(
    ctx: &ReducerContext,
    input: BigFishSessionGetInput,
) -> Result<BigFishSessionSnapshot, String> {
    validate_session_get_input(&input).map_err(|error| error.to_string())?;
    let session = ctx
        .db
        .big_fish_creation_session()
        .session_id()
        .find(&input.session_id)
        .filter(|row| row.owner_user_id == input.owner_user_id)
        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;

    build_big_fish_session_snapshot(ctx, &session)
}

pub(crate) fn list_big_fish_works_tx(
    ctx: &ReducerContext,
    input: BigFishWorksListInput,
) -> Result<Vec<BigFishWorkSummarySnapshot>, String> {
    validate_works_list_input(&input).map_err(|error| error.to_string())?;
    let now_micros = ctx.timestamp.to_micros_since_unix_epoch();

    let mut items = ctx
        .db
        .big_fish_creation_session()
        .iter()
        .filter(|row| {
            if input.published_only {
                return row.stage == BigFishCreationStage::Published;
            }

            row.owner_user_id == input.owner_user_id && should_include_big_fish_work(ctx, row)
        })
        .map(|row| build_big_fish_work_summary(ctx, &row, now_micros))
        .collect::<Result<Vec<_>, _>>()?;

    items.sort_by(|left, right| {
        right
            .updated_at_micros
            .cmp(&left.updated_at_micros)
            .then_with(|| left.work_id.cmp(&right.work_id))
    });
    Ok(items)
}

fn should_include_big_fish_work(ctx: &ReducerContext, row: &BigFishCreationSession) -> bool {
    if big_fish_session_has_direct_work_content(row) {
        return true;
    }

    ctx.db.big_fish_agent_message().iter().any(|message| {
        message.session_id == row.session_id
            && matches!(message.role, BigFishAgentMessageRole::User)
    })
}

fn big_fish_session_has_direct_work_content(row: &BigFishCreationSession) -> bool {
    // 助手欢迎语和默认 anchorPack 只是工作台初始状态，不应被当成草稿作品。
    !row.seed_text.trim().is_empty()
        || row.draft_json.is_some()
        || row.stage == BigFishCreationStage::Published
}

pub(crate) fn delete_big_fish_work_tx(
    ctx: &ReducerContext,
    input: BigFishWorkDeleteInput,
) -> Result<Vec<BigFishWorkSummarySnapshot>, String> {
    validate_session_get_input(&BigFishSessionGetInput {
        session_id: input.session_id.clone(),
        owner_user_id: input.owner_user_id.clone(),
    })
    .map_err(|error| error.to_string())?;
    let session = ctx
        .db
        .big_fish_creation_session()
        .session_id()
        .find(&input.session_id)
        .filter(|row| row.owner_user_id == input.owner_user_id)
        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;

    // 中文注释：删除作品时同步清理 Agent 消息、素材槽和后端运行态快照，避免失去来源会话的 run 残留。
    ctx.db
        .big_fish_creation_session()
        .session_id()
        .delete(&session.session_id);
    for message in ctx
        .db
        .big_fish_agent_message()
        .iter()
        .filter(|row| row.session_id == input.session_id)
        .collect::<Vec<_>>()
    {
        ctx.db
            .big_fish_agent_message()
            .message_id()
            .delete(&message.message_id);
    }
    for slot in ctx
        .db
        .big_fish_asset_slot()
        .iter()
        .filter(|row| row.session_id == input.session_id)
        .collect::<Vec<_>>()
    {
        ctx.db.big_fish_asset_slot().slot_id().delete(&slot.slot_id);
    }
    for run in ctx
        .db
        .big_fish_runtime_run()
        .iter()
        .filter(|row| row.session_id == input.session_id)
        .collect::<Vec<_>>()
    {
        ctx.db.big_fish_runtime_run().run_id().delete(&run.run_id);
    }
    list_big_fish_works_tx(
        ctx,
        BigFishWorksListInput {
            owner_user_id: input.owner_user_id,
            published_only: false,
        },
    )
}

pub(crate) fn submit_big_fish_message_tx(
    ctx: &ReducerContext,
    input: BigFishMessageSubmitInput,
) -> Result<BigFishSessionSnapshot, String> {
    validate_message_submit_input(&input).map_err(|error| error.to_string())?;
    let session = ctx
        .db
        .big_fish_creation_session()
        .session_id()
        .find(&input.session_id)
        .filter(|row| row.owner_user_id == input.owner_user_id)
        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;
    if ctx
        .db
        .big_fish_agent_message()
        .message_id()
        .find(&input.user_message_id)
        .is_some()
    {
        return Err("big_fish_agent_message.user_message_id 已存在".to_string());
    }
    if ctx
        .db
        .big_fish_agent_message()
        .message_id()
        .find(&input.assistant_message_id)
        .is_some()
    {
        return Err("big_fish_agent_message.assistant_message_id 已存在".to_string());
    }

    let submitted_at = Timestamp::from_micros_since_unix_epoch(input.submitted_at_micros);
    ctx.db.big_fish_agent_message().insert(BigFishAgentMessage {
        message_id: input.user_message_id,
        session_id: input.session_id.clone(),
        role: BigFishAgentMessageRole::User,
        kind: BigFishAgentMessageKind::Chat,
        text: input.user_message_text.trim().to_string(),
        created_at: submitted_at,
    });

    let next_session = BigFishCreationSession {
        session_id: session.session_id.clone(),
        owner_user_id: session.owner_user_id.clone(),
        seed_text: session.seed_text.clone(),
        current_turn: session.current_turn,
        progress_percent: session.progress_percent,
        stage: BigFishCreationStage::CollectingAnchors,
        anchor_pack_json: session.anchor_pack_json.clone(),
        draft_json: session.draft_json.clone(),
        asset_coverage_json: session.asset_coverage_json.clone(),
        last_assistant_reply: session.last_assistant_reply.clone(),
        publish_ready: session.publish_ready,
        play_count: session.play_count,
        remix_count: session.remix_count,
        like_count: session.like_count,
        published_at: session.published_at,
        created_at: session.created_at,
        updated_at: submitted_at,
    };
    replace_big_fish_session(ctx, &session, next_session);

    get_big_fish_session_tx(
        ctx,
        BigFishSessionGetInput {
            session_id: input.session_id,
            owner_user_id: input.owner_user_id,
        },
    )
}

pub(crate) fn finalize_big_fish_agent_message_turn_tx(
    ctx: &ReducerContext,
    input: BigFishMessageFinalizeInput,
) -> Result<BigFishSessionSnapshot, String> {
    validate_message_finalize_input(&input).map_err(|error| error.to_string())?;
    let session = ctx
        .db
        .big_fish_creation_session()
        .session_id()
        .find(&input.session_id)
        .filter(|row| row.owner_user_id == input.owner_user_id)
        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;
    let updated_at = Timestamp::from_micros_since_unix_epoch(input.updated_at_micros);

    if let Some(error_message) = input
        .error_message
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
    {
        let next_session = BigFishCreationSession {
            session_id: session.session_id.clone(),
            owner_user_id: session.owner_user_id.clone(),
            seed_text: session.seed_text.clone(),
            current_turn: session.current_turn,
            progress_percent: session.progress_percent,
            stage: session.stage,
            anchor_pack_json: session.anchor_pack_json.clone(),
            draft_json: session.draft_json.clone(),
            asset_coverage_json: session.asset_coverage_json.clone(),
            last_assistant_reply: session.last_assistant_reply.clone(),
            publish_ready: session.publish_ready,
            play_count: session.play_count,
            remix_count: session.remix_count,
            like_count: session.like_count,
            published_at: session.published_at,
            created_at: session.created_at,
            updated_at,
        };
        replace_big_fish_session(ctx, &session, next_session);
        return Err(error_message.to_string());
    }

    let assistant_message_id = input
        .assistant_message_id
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| "big_fish assistant_message_id 不能为空".to_string())?
        .to_string();
    let assistant_reply_text = input
        .assistant_reply_text
        .as_deref()
        .map(str::trim)
        .filter(|value| !value.is_empty())
        .ok_or_else(|| "big_fish assistant_reply_text 不能为空".to_string())?
        .to_string();
    if ctx
        .db
        .big_fish_agent_message()
        .message_id()
        .find(&assistant_message_id)
        .is_some()
    {
        return Err("big_fish_agent_message.assistant_message_id 已存在".to_string());
    }
    let next_anchor_pack =
        deserialize_anchor_pack(&input.anchor_pack_json).map_err(|error| error.to_string())?;
    ctx.db.big_fish_agent_message().insert(BigFishAgentMessage {
        message_id: assistant_message_id,
        session_id: input.session_id.clone(),
        role: BigFishAgentMessageRole::Assistant,
        kind: BigFishAgentMessageKind::Chat,
        text: assistant_reply_text.clone(),
        created_at: updated_at,
    });

    let next_session = BigFishCreationSession {
        session_id: session.session_id.clone(),
        owner_user_id: session.owner_user_id.clone(),
        seed_text: session.seed_text.clone(),
        current_turn: session.current_turn.saturating_add(1),
        progress_percent: input.progress_percent.min(100),
        stage: input.stage,
        anchor_pack_json: serialize_anchor_pack(&next_anchor_pack)
            .map_err(|error| error.to_string())?,
        draft_json: session.draft_json.clone(),
        asset_coverage_json: session.asset_coverage_json.clone(),
        last_assistant_reply: Some(assistant_reply_text),
        publish_ready: session.publish_ready,
        play_count: session.play_count,
        remix_count: session.remix_count,
        like_count: session.like_count,
        published_at: session.published_at,
        created_at: session.created_at,
        updated_at,
    };
    replace_big_fish_session(ctx, &session, next_session);

    get_big_fish_session_tx(
        ctx,
        BigFishSessionGetInput {
            session_id: input.session_id,
            owner_user_id: input.owner_user_id,
        },
    )
}

pub(crate) fn compile_big_fish_draft_tx(
    ctx: &ReducerContext,
    input: BigFishDraftCompileInput,
) -> Result<BigFishSessionSnapshot, String> {
    validate_draft_compile_input(&input).map_err(|error| error.to_string())?;
    let session = ctx
        .db
        .big_fish_creation_session()
        .session_id()
        .find(&input.session_id)
        .filter(|row| row.owner_user_id == input.owner_user_id)
        .ok_or_else(|| "big_fish_creation_session 不存在".to_string())?;
    let anchor_pack =
        deserialize_anchor_pack(&session.anchor_pack_json).map_err(|error| error.to_string())?;
    let draft = input
        .draft_json
        .as_deref()
        .map(deserialize_draft)
        .transpose()
        .map_err(|error| format!("big_fish.draft_json 非法: {error}"))?
        .unwrap_or_else(|| compile_default_draft(&anchor_pack));
    let asset_slots = list_big_fish_asset_slots(ctx, &session.session_id);
    let readiness = evaluate_publish_readiness(
        EvaluateBigFishPublishReadinessCommand {
            session_id: session.session_id.clone(),
            owner_user_id: session.owner_user_id.clone(),
            draft: Some(draft.clone()),
            evaluated_at_micros: input.compiled_at_micros,
        },
        &asset_slots,
    )
    .map_err(|error| error.to_string())?;
    let coverage = build_asset_coverage(Some(&draft), &asset_slots);
    let compiled_at = Timestamp::from_micros_since_unix_epoch(input.compiled_at_micros);
    let reply = "第一版玩法草稿已编译完成，可以在结果页逐级生成主图、动作和场地背景。".to_string();

    let next_session = BigFishCreationSession {
        session_id: session.session_id.clone(),
        owner_user_id: session.owner_user_id.clone(),
        seed_text: session.seed_text.clone(),
        current_turn: session.current_turn,
        progress_percent: 80,
        stage: BigFishCreationStage::DraftReady,
        anchor_pack_json: session.anchor_pack_json.clone(),
        draft_json: Some(serialize_draft(&draft).map_err(|error| error.to_string())?),
        asset_coverage_json: serialize_asset_coverage(&coverage)
            .map_err(|error| error.to_string())?,
        last_assistant_reply: Some(reply.clone()),
        publish_ready: readiness.readiness.publish_ready,
        play_count: session.play_count,
        remix_count: session.remix_count,
        like_count: session.like_count,
        published_at: session.published_at,
        created_at: session.created_at,
        updated_at: compiled_at,
    };
    replace_big_fish_session(ctx, &session, next_session);
    for event in readiness.events {
        emit_big_fish_publish_readiness_event(ctx, event)?;
    }

    get_big_fish_session_tx(
        ctx,
        BigFishSessionGetInput {
            session_id: input.session_id,
            owner_user_id: input.owner_user_id,
        },
    )
}

pub(crate) fn record_big_fish_play_tx(
    ctx: &ReducerContext,
    input: BigFishPlayRecordInput,
) -> Result<Vec<BigFishWorkSummarySnapshot>, String> {
    validate_play_record_input(&input).map_err(|error| error.to_string())?;
    let session = ctx
        .db
        .big_fish_creation_session()
        .session_id()
        .find(&input.session_id)
        .filter(|row| row.stage == BigFishCreationStage::Published)
        .ok_or_else(|| "big_fish 已发布作品不存在".to_string())?;
    let played_at = Timestamp::from_micros_since_unix_epoch(input.played_at_micros);
    let draft = session
        .draft_json
        .as_deref()
        .map(deserialize_draft)
        .transpose()
        .map_err(|error| format!("big_fish.draft_json 非法: {error}"))?;
    let title = draft
        .as_ref()
        .map(|value| value.title.trim().to_string())
        .filter(|value| !value.is_empty())
        .unwrap_or_else(|| "大鱼吃小鱼".to_string());
    let subtitle = draft
        .as_ref()
        .and_then(|value| {
            let subtitle = value.subtitle.trim();
            if subtitle.is_empty() {
                let core_fun = value.core_fun.trim();
                (!core_fun.is_empty()).then(|| core_fun.to_string())
            } else {
                Some(subtitle.to_string())
            }
        })
        .unwrap_or_default();
    let world_key = format!("big-fish:{}", session.session_id);

    upsert_profile_played_work(
        ctx,
        ProfilePlayedWorkUpsertInput {
            user_id: input.user_id.clone(),
            world_key: world_key.clone(),
            owner_user_id: Some(session.owner_user_id.clone()),
            profile_id: Some(session.session_id.clone()),
            world_type: Some("BIG_FISH".to_string()),
            world_title: title,
            world_subtitle: subtitle,
            played_at_micros: input.played_at_micros,
        },
    )?;
    add_profile_observed_play_time(
        ctx,
        &input.user_id,
        &world_key,
        input.elapsed_ms,
        input.played_at_micros,
    )?;
    record_public_work_play(
        ctx,
        PublicWorkPlayRecordInput {
            source_type: "big-fish".to_string(),
            owner_user_id: session.owner_user_id.clone(),
            profile_id: session.session_id.clone(),
            played_at_micros: input.played_at_micros,
        },
    )?;
    let next_session = BigFishCreationSession {
        session_id: session.session_id.clone(),
        owner_user_id: session.owner_user_id.clone(),
        seed_text: session.seed_text.clone(),
        current_turn: session.current_turn,
        progress_percent: session.progress_percent,
        stage: session.stage,
        anchor_pack_json: session.anchor_pack_json.clone(),
        draft_json: session.draft_json.clone(),
        asset_coverage_json: session.asset_coverage_json.clone(),
        last_assistant_reply: session.last_assistant_reply.clone(),
        publish_ready: session.publish_ready,
        // 中文注释：正式进入已发布作品时同时累加作品播放数，用户侧去重由 profile_played_world 保证。
        play_count: session.play_count.saturating_add(1),
        remix_count: session.remix_count,
        like_count: session.like_count,
        published_at: session.published_at,
        created_at: session.created_at,
        updated_at: played_at,
    };
    replace_big_fish_session(ctx, &session, next_session);

    list_big_fish_works_tx(ctx, build_public_big_fish_gallery_list_input())
}

pub(crate) fn record_big_fish_like_tx(
    ctx: &ReducerContext,
    input: BigFishWorkLikeRecordInput,
) -> Result<Vec<BigFishWorkSummarySnapshot>, String> {
    let session_id = input.session_id.trim();
    let user_id = input.user_id.trim();
    if session_id.is_empty() || user_id.is_empty() {
        return Err("big_fish like 参数不能为空".to_string());
    }
    let session = ctx
        .db
        .big_fish_creation_session()
        .session_id()
        .find(&session_id.to_string())
        .filter(|row| row.stage == BigFishCreationStage::Published)
        .ok_or_else(|| "big_fish 已发布作品不存在，无法点赞".to_string())?;
    let inserted_like = record_public_work_like(
        ctx,
        PublicWorkLikeRecordInput {
            source_type: "big-fish".to_string(),
            owner_user_id: session.owner_user_id.clone(),
            profile_id: session.session_id.clone(),
            user_id: user_id.to_string(),
            liked_at_micros: input.liked_at_micros,
        },
    )?;

    if inserted_like {
        let liked_at = Timestamp::from_micros_since_unix_epoch(input.liked_at_micros);
        let next_session = BigFishCreationSession {
            session_id: session.session_id.clone(),
            owner_user_id: session.owner_user_id.clone(),
            seed_text: session.seed_text.clone(),
            current_turn: session.current_turn,
            progress_percent: session.progress_percent,
            stage: session.stage,
            anchor_pack_json: session.anchor_pack_json.clone(),
            draft_json: session.draft_json.clone(),
            asset_coverage_json: session.asset_coverage_json.clone(),
            last_assistant_reply: session.last_assistant_reply.clone(),
            publish_ready: session.publish_ready,
            play_count: session.play_count,
            remix_count: session.remix_count,
            like_count: session.like_count.saturating_add(1),
            published_at: session.published_at,
            created_at: session.created_at,
            updated_at: liked_at,
        };
        replace_big_fish_session(ctx, &session, next_session);
    }

    list_big_fish_works_tx(ctx, build_public_big_fish_gallery_list_input())
}

fn remix_big_fish_work_tx(
    ctx: &ReducerContext,
    input: BigFishWorkRemixInput,
) -> Result<BigFishSessionSnapshot, String> {
    let source_session_id = input.source_session_id.trim();
    let target_session_id = input.target_session_id.trim();
    let target_owner_user_id = input.target_owner_user_id.trim();
    let welcome_message_id = input.welcome_message_id.trim();
    if source_session_id.is_empty()
        || target_session_id.is_empty()
        || target_owner_user_id.is_empty()
        || welcome_message_id.is_empty()
    {
        return Err("big_fish remix 参数不能为空".to_string());
    }
    if ctx
        .db
        .big_fish_creation_session()
        .session_id()
        .find(&target_session_id.to_string())
        .is_some()
    {
        return Err("big_fish remix 目标 session 已存在".to_string());
    }
    if ctx
        .db
        .big_fish_agent_message()
        .message_id()
        .find(&welcome_message_id.to_string())
        .is_some()
    {
        return Err("big_fish remix 消息已存在".to_string());
    }

    let source = ctx
        .db
        .big_fish_creation_session()
        .session_id()
        .find(&source_session_id.to_string())
        .filter(|row| row.stage == BigFishCreationStage::Published)
        .ok_or_else(|| "big_fish 已发布源作品不存在".to_string())?;
    let remixed_at = Timestamp::from_micros_since_unix_epoch(input.remixed_at_micros);
    let next_source = BigFishCreationSession {
        session_id: source.session_id.clone(),
        owner_user_id: source.owner_user_id.clone(),
        seed_text: source.seed_text.clone(),
        current_turn: source.current_turn,
        progress_percent: source.progress_percent,
        stage: source.stage,
        anchor_pack_json: source.anchor_pack_json.clone(),
        draft_json: source.draft_json.clone(),
        asset_coverage_json: source.asset_coverage_json.clone(),
        last_assistant_reply: source.last_assistant_reply.clone(),
        publish_ready: source.publish_ready,
        play_count: source.play_count,
        remix_count: source.remix_count.saturating_add(1),
        like_count: source.like_count,
        published_at: source.published_at,
        created_at: source.created_at,
        updated_at: remixed_at,
    };
    replace_big_fish_session(ctx, &source, next_source);

    let target_session = BigFishCreationSession {
        session_id: target_session_id.to_string(),
        owner_user_id: target_owner_user_id.to_string(),
        seed_text: source.seed_text.clone(),
        current_turn: 1,
        progress_percent: 80,
        stage: BigFishCreationStage::DraftReady,
        anchor_pack_json: source.anchor_pack_json.clone(),
        draft_json: source.draft_json.clone(),
        asset_coverage_json: source.asset_coverage_json.clone(),
        last_assistant_reply: Some("已从公开作品 Remix 出新的大鱼吃小鱼草稿。".to_string()),
        publish_ready: source.publish_ready,
        play_count: 0,
        remix_count: 0,
        like_count: 0,
        published_at: None,
        created_at: remixed_at,
        updated_at: remixed_at,
    };
    ctx.db.big_fish_creation_session().insert(target_session);
    ctx.db.big_fish_agent_message().insert(BigFishAgentMessage {
        message_id: welcome_message_id.to_string(),
        session_id: target_session_id.to_string(),
        role: BigFishAgentMessageRole::Assistant,
        kind: BigFishAgentMessageKind::Summary,
        text: "已复制公开作品为你的草稿。".to_string(),
        created_at: remixed_at,
    });
    for slot in list_big_fish_asset_slots(ctx, &source.session_id) {
        upsert_big_fish_asset_slot(
            ctx,
            BigFishAssetSlotSnapshot {
                slot_id: slot.slot_id.replace(&source.session_id, target_session_id),
                session_id: target_session_id.to_string(),
                asset_kind: slot.asset_kind,
                level: slot.level,
                motion_key: slot.motion_key,
                status: slot.status,
                asset_url: slot.asset_url,
                prompt_snapshot: slot.prompt_snapshot,
                updated_at_micros: input.remixed_at_micros,
            },
        );
    }

    get_big_fish_session_tx(
        ctx,
        BigFishSessionGetInput {
            session_id: target_session_id.to_string(),
            owner_user_id: target_owner_user_id.to_string(),
        },
    )
}

pub(crate) fn build_big_fish_session_snapshot(
    ctx: &ReducerContext,
    row: &BigFishCreationSession,
) -> Result<BigFishSessionSnapshot, String> {
    let anchor_pack =
        deserialize_anchor_pack(&row.anchor_pack_json).unwrap_or_else(|_| empty_anchor_pack());
    let draft = row
        .draft_json
        .as_deref()
        .map(deserialize_draft)
        .transpose()
        .map_err(|error| format!("big_fish.draft_json 非法: {error}"))?;
    let asset_slots = list_big_fish_asset_slots(ctx, &row.session_id);
    let asset_coverage = build_asset_coverage(draft.as_ref(), &asset_slots);
    let mut messages = ctx
        .db
        .big_fish_agent_message()
        .iter()
        .filter(|message| message.session_id == row.session_id)
        .map(|message| BigFishAgentMessageSnapshot {
            message_id: message.message_id,
            session_id: message.session_id,
            role: message.role,
            kind: message.kind,
            text: message.text,
            created_at_micros: message.created_at.to_micros_since_unix_epoch(),
        })
        .collect::<Vec<_>>();
    messages.sort_by_key(|message| (message.created_at_micros, message.message_id.clone()));

    Ok(BigFishSessionSnapshot {
        session_id: row.session_id.clone(),
        owner_user_id: row.owner_user_id.clone(),
        seed_text: row.seed_text.clone(),
        current_turn: row.current_turn,
        progress_percent: row.progress_percent,
        stage: row.stage,
        anchor_pack,
        draft,
        asset_slots,
        asset_coverage,
        messages,
        last_assistant_reply: row.last_assistant_reply.clone(),
        publish_ready: row.publish_ready,
        created_at_micros: row.created_at.to_micros_since_unix_epoch(),
        updated_at_micros: row.updated_at.to_micros_since_unix_epoch(),
    })
}

pub(crate) fn build_big_fish_work_summary(
    ctx: &ReducerContext,
    row: &BigFishCreationSession,
    now_micros: i64,
) -> Result<BigFishWorkSummarySnapshot, String> {
    let draft = row
        .draft_json
        .as_deref()
        .map(deserialize_draft)
        .transpose()
        .map_err(|error| format!("big_fish.draft_json 非法: {error}"))?;
    let asset_slots = list_big_fish_asset_slots(ctx, &row.session_id);
    let coverage = build_asset_coverage(draft.as_ref(), &asset_slots);
    let cover_image_src = asset_slots
        .iter()
        .find(|slot| slot.asset_kind == BigFishAssetKind::StageBackground)
        .and_then(|slot| slot.asset_url.clone())
        .or_else(|| {
            asset_slots
                .iter()
                .find(|slot| slot.asset_kind == BigFishAssetKind::LevelMainImage)
                .and_then(|slot| slot.asset_url.clone())
        });
    let title = draft
        .as_ref()
        .map(|value| value.title.clone())
        .filter(|value| !value.trim().is_empty())
        .unwrap_or_else(|| "未命名大鱼草稿".to_string());
    let subtitle = draft
        .as_ref()
        .map(|value| value.subtitle.clone())
        .filter(|value| !value.trim().is_empty())
        .unwrap_or_else(|| "等待整理玩法草稿".to_string());
    let summary = draft
        .as_ref()
        .map(|value| value.core_fun.clone())
        .filter(|value| !value.trim().is_empty())
        .unwrap_or_else(|| {
            row.last_assistant_reply
                .clone()
                .unwrap_or_else(|| "继续补齐锚点后即可生成玩法草稿。".to_string())
        });

    Ok(BigFishWorkSummarySnapshot {
        work_id: format!("big-fish-work-{}", row.session_id),
        source_session_id: row.session_id.clone(),
        owner_user_id: row.owner_user_id.clone(),
        title,
        subtitle,
        summary,
        cover_image_src,
        status: if row.stage == BigFishCreationStage::Published {
            "published".to_string()
        } else {
            "draft".to_string()
        },
        updated_at_micros: row.updated_at.to_micros_since_unix_epoch(),
        publish_ready: coverage.publish_ready,
        level_count: draft
            .as_ref()
            .map(|value| value.runtime_params.level_count)
            .unwrap_or(BIG_FISH_DEFAULT_LEVEL_COUNT),
        level_main_image_ready_count: coverage.level_main_image_ready_count,
        level_motion_ready_count: coverage.level_motion_ready_count,
        background_ready: coverage.background_ready,
        play_count: row.play_count,
        remix_count: row.remix_count,
        like_count: row.like_count,
        recent_play_count_7d: count_recent_public_work_plays(
            ctx,
            "big-fish",
            &row.session_id,
            now_micros,
        ),
        published_at_micros: row
            .published_at
            .or_else(|| (row.stage == BigFishCreationStage::Published).then_some(row.updated_at))
            .map(|value| value.to_micros_since_unix_epoch()),
    })
}

fn build_public_big_fish_gallery_list_input() -> BigFishWorksListInput {
    BigFishWorksListInput {
        // 中文注释：published_only 分支不会按 owner 过滤；非空占位用于兼容旧部署模块的前置校验。
        owner_user_id: PUBLIC_BIG_FISH_GALLERY_OWNER_USER_ID.to_string(),
        published_only: true,
    }
}

pub(crate) fn replace_big_fish_session(
    ctx: &ReducerContext,
    current: &BigFishCreationSession,
    next: BigFishCreationSession,
) {
    ctx.db
        .big_fish_creation_session()
        .session_id()
        .delete(&current.session_id);
    ctx.db.big_fish_creation_session().insert(next);
}

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

    fn build_test_big_fish_session(
        seed_text: &str,
        draft_json: Option<&str>,
        stage: BigFishCreationStage,
    ) -> BigFishCreationSession {
        BigFishCreationSession {
            session_id: "big-fish-session-1".to_string(),
            owner_user_id: "user-1".to_string(),
            seed_text: seed_text.to_string(),
            current_turn: 0,
            progress_percent: 20,
            stage,
            anchor_pack_json: "{}".to_string(),
            draft_json: draft_json.map(str::to_string),
            asset_coverage_json: "{}".to_string(),
            last_assistant_reply: Some("欢迎来到大鱼吃小鱼共创。".to_string()),
            publish_ready: false,
            play_count: 0,
            remix_count: 0,
            like_count: 0,
            published_at: if stage == BigFishCreationStage::Published {
                Some(Timestamp::from_micros_since_unix_epoch(1))
            } else {
                None
            },
            created_at: Timestamp::from_micros_since_unix_epoch(1),
            updated_at: Timestamp::from_micros_since_unix_epoch(1),
        }
    }

    #[test]
    fn initial_big_fish_creation_progress_starts_from_zero() {
        assert_eq!(INITIAL_BIG_FISH_CREATION_PROGRESS_PERCENT, 0);
    }

    #[test]
    fn big_fish_direct_work_content_ignores_empty_created_session() {
        let empty_session =
            build_test_big_fish_session("", None, BigFishCreationStage::CollectingAnchors);
        let seeded_session = build_test_big_fish_session(
            "想做深海吞噬成长",
            None,
            BigFishCreationStage::CollectingAnchors,
        );
        let drafted_session = build_test_big_fish_session(
            "",
            Some(r#"{"title":"深海吞噬"}"#),
            BigFishCreationStage::DraftReady,
        );
        let published_session =
            build_test_big_fish_session("", None, BigFishCreationStage::Published);

        assert!(!big_fish_session_has_direct_work_content(&empty_session));
        assert!(big_fish_session_has_direct_work_content(&seeded_session));
        assert!(big_fish_session_has_direct_work_content(&drafted_session));
        assert!(big_fish_session_has_direct_work_content(&published_session));
    }
}