overlord_event_system/logic/

chapters_management.rs

use crate::{
    entities,
    event::{OverlordEvent, PrepareFightType},
    game_config_helpers::GameConfigLookup,
    logic::handler::OverlordLogic,
    state::OverlordState,
};

use essences::{
    currency::{CurrencyConsumer, CurrencyUnit, check_can_decrease_currencies},
    dungeons::DungeonTemplateId,
    entity::{ActionWithDeadline, Entity, EntityId, EntityState},
    fighting::{
        ActiveDungeon, ActiveFight, EntityTeam, EntityType, FightTemplate, FightTemplateId,
        FightType,
    },
    flip::WorldSide,
    game::Chapter,
    pvp::PVPState,
};

use analytics::constants::METRICS_TARGET;
use configs::game_config::GameConfig;
use event_system::{event::EventPluginized, script::random::GameRng, system::EventHandleResult};
use rand::RngExt;

/// Minimum interval between party-ally DB refreshes, in game seconds
/// (matching the party action cooldown).
const PARTY_REFRESH_INTERVAL_SEC: i64 = 30;

/// How many campaign-boss first clears in a row have failed to roll a Mastery
/// Crystal. Lives in `custom_values` — one integer of progression state, which
/// does not deserve a column of its own.
const MASTERY_CRYSTAL_MISSES: &str = "mastery_crystal_misses";

/// The one-time Core Essence unlock packet for a fresh campaign crossing.
/// Keeping the crossing predicate and the authored payload together makes the
/// no-reconnect/no-double-grant invariant directly testable.
fn core_unlock_essence_grant_event(
    game_config: &GameConfig,
    prev_chapter_level: i64,
    next_chapter_level: i64,
) -> Option<EventPluginized<OverlordEvent, OverlordState>> {
    let settings = &game_config.cores_settings;
    if prev_chapter_level >= settings.unlock_chapter
        || next_chapter_level < settings.unlock_chapter
        || settings.unlock_essence_grant <= 0
    {
        return None;
    }

    Some(EventPluginized::now(OverlordEvent::CurrencyIncrease {
        currencies: vec![CurrencyUnit {
            currency_id: settings.upgrade_currency_id,
            amount: settings.unlock_essence_grant,
        }],
        currency_source: essences::currency::CurrencySource::ChapterReward,
    }))
}

impl OverlordLogic {
    /// Returns the `RefreshPartyMemberState` event if the player has a party
    /// ally and the TTL since the last refresh has elapsed. Updates the
    /// last-refresh time when it decides to emit.
    ///
    /// The refresh reloads the ally's whole character sheet from the DB, so
    /// emitting it after every fight is the dominant DB cost of a partied
    /// session. An ally's power drifts well under 1% in 30 seconds.
    fn maybe_refresh_party_event(
        &mut self,
        state: &OverlordState,
    ) -> Option<EventPluginized<OverlordEvent, OverlordState>> {
        state.character_state.character.party_character_id?;

        let now = ::time::utc_now();
        if self.last_party_refresh_at.is_some_and(|last| {
            now.signed_duration_since(last).num_seconds() < PARTY_REFRESH_INTERVAL_SEC
        }) {
            return None;
        }

        self.last_party_refresh_at = Some(now);
        Some(EventPluginized::now(
            OverlordEvent::RefreshPartyMemberState {},
        ))
    }

    /// Arena world switch (select-opponent screen): the side this player
    /// STARTS arena fights on — both as the attacker and as the snapshot
    /// opponent in other players' fights. Only meaningful once the flip is
    /// unlocked; before that there is no switch to set.
    pub fn handle_set_arena_world_side(
        &self,
        side: WorldSide,
        mut state: OverlordState,
    ) -> EventHandleResult<OverlordEvent, OverlordState> {
        let game_config = self.game_config.get();
        if !game_config
            .flip_settings
            .is_unlocked(state.character_state.character.current_chapter_level)
        {
            tracing::warn!(
                "SetArenaWorldSide rejected: flip locked at chapter {}",
                state.character_state.character.current_chapter_level
            );
            return EventHandleResult::fail(state);
        }

        state.character_state.character.arena_world_side = Some(side);
        EventHandleResult::ok(state)
    }

    pub fn is_battle_active(&self, state: &OverlordState) -> bool {
        if let Some(active_fight) = &state.active_fight {
            let has_player = active_fight
                .entities
                .iter()
                .any(|e| e.id == active_fight.player_id);

            let has_enemy = active_fight
                .entities
                .iter()
                .any(|e| e.team == EntityTeam::Enemy);

            has_player && has_enemy
        } else {
            false
        }
    }

    pub fn handle_prepare_fight(
        &mut self,
        prepare_fight_type: PrepareFightType,
        rand_gen: rand::rngs::StdRng,
        mut state: OverlordState,
    ) -> EventHandleResult<OverlordEvent, OverlordState> {
        let game_config = self.game_config.get();

        if state.pvp_state.is_some() {
            return EventHandleResult::fail(state);
        }

        let is_retry_boss_fight = matches!(prepare_fight_type, PrepareFightType::RetryBossFight);

        match prepare_fight_type {
            PrepareFightType::PVEFight => {
                if !self.validate_pve_fight(&state) {
                    return EventHandleResult::fail(state);
                }
            }
            PrepareFightType::PVPFight { .. } => {}
            PrepareFightType::RetryBossFight => {
                if !self.validate_prepare_retry_boss_fight(&state) {
                    return EventHandleResult::fail(state);
                }
            }
            PrepareFightType::DungeonFight {
                dungeon_id,
                difficulty,
            } => {
                if !self.validate_dungeon_fight(&state, dungeon_id, difficulty) {
                    return EventHandleResult::fail(state);
                }
            }
            PrepareFightType::ForfeitDungeonFight => {}
            PrepareFightType::SingleFight { .. } => {}
        };

        // Item TTL: remove expired items here, at the start of the next fight,
        // before its power snapshot is built — so they never drop mid-fight.
        // Snapshot the inventory only when something is removed, so a failed
        // prepare below can roll it back — else it would strip items without
        // emitting ItemsExpired and orphan their DB rows.
        //
        // Skipped for PvP: the fight is prepared twice — once by the
        // precalculation that decides it, once live for the player to watch —
        // and each run reads the wall clock separately. An item expiring in
        // between would change the power snapshot of only one of them. Expired
        // items are swept at the next PvE prepare.
        let is_pvp = matches!(prepare_fight_type, PrepareFightType::PVPFight { .. });
        let now = ::time::utc_now();
        let pre_sweep_inventory = (!is_pvp)
            .then(|| {
                state
                    .character_state
                    .inventory
                    .iter()
                    .any(|item| item.expires_at.is_some_and(|exp| exp <= now))
                    .then(|| state.character_state.inventory.clone())
            })
            .flatten();
        let expiry_events = if is_pvp {
            Vec::new()
        } else {
            super::items::sweep_expired_items(&mut state, now)
        };

        // The previous fight (if any) is over: drop its pending combat events
        // so the new fight schedules onto a clean clock. The saved clock is
        // restored on every failure exit below, so a failed PrepareFight can't
        // kill the running fight's heartbeat and pending scheduling.
        // The FightProgress heartbeat is re-installed below on success.
        let saved_clock = self.fight_clock.clone();
        self.fight_clock.clear();
        state.active_fight = None;

        let prepare_fight_events = match prepare_fight_type {
            PrepareFightType::PVEFight => self.handle_pve_prepare_fight(&mut state, rand_gen),
            PrepareFightType::ForfeitDungeonFight => {
                self.handle_pve_prepare_fight(&mut state, rand_gen)
            }
            PrepareFightType::PVPFight {
                fight_id,
                pvp_state,
            } => self.handle_pvp_prepare_fight(&mut state, fight_id, pvp_state, rand_gen),
            PrepareFightType::RetryBossFight => {
                match game_config
                    .require_chapter_by_level(state.character_state.character.current_chapter_level)
                {
                    Ok(chapter) => {
                        state.character_state.character.current_fight_number =
                            (chapter.fight_ids.len() - 1) as i64;
                        self.handle_pve_prepare_fight(&mut state, rand_gen)
                    }
                    Err(_) => {
                        tracing::error!(
                            "Failed to get chapter with chapter_level={}",
                            state.character_state.character.current_chapter_level
                        );
                        None
                    }
                }
            }
            PrepareFightType::DungeonFight {
                dungeon_id,
                difficulty,
            } => self.handle_dungeon_prepare_fight(&mut state, dungeon_id, difficulty, rand_gen),
            PrepareFightType::SingleFight { fight_templated_id } => {
                self.handle_single_prepare_fight(&mut state, fight_templated_id, rand_gen)
            }
        };

        let Some(events) = prepare_fight_events else {
            tracing::error!("Preparing fight failed, something went wrong");
            self.fight_clock = saved_clock;
            if let Some(inventory) = pre_sweep_inventory {
                state.character_state.inventory = inventory;
            }
            return EventHandleResult::fail(state);
        };

        if state.active_fight.is_none() {
            tracing::error!("No active fight after preparing fight");
            self.fight_clock = saved_clock;
            if let Some(inventory) = pre_sweep_inventory {
                state.character_state.inventory = inventory;
            }
            return EventHandleResult::fail(state);
        };

        if is_retry_boss_fight {
            // Consume the boss retry only once the prepare succeeded: failure
            // results still apply their state, and a prematurely-set flag
            // makes `validate_prepare_retry_boss_fight` reject every further
            // retry even though the boss fight was never restarted.
            state.character_state.character.last_boss_fight_won = true;
        }

        self.fight_clock.set_heartbeat(
            OverlordEvent::FightProgress {},
            game_config.game_settings.fight_progress_tick,
        );

        // Emit the chapter-start expiry events ahead of the fight's own events.
        let mut events = events;
        events.splice(0..0, expiry_events);

        EventHandleResult::ok_events(state, events)
    }

    fn validate_pve_fight(&self, state: &OverlordState) -> bool {
        if let Some(active_fight) = &state.active_fight
            && !active_fight.fight_ended
            && active_fight.dungeon.is_some()
        {
            tracing::error!("Dungeon fight is in progress");
            return false;
        };

        true
    }

    fn validate_prepare_retry_boss_fight(&self, state: &OverlordState) -> bool {
        let game_config = self.game_config.get();

        // These validation failures are expected in normal client-server
        // async play: the server state has already moved on while the
        // client's cached state is a few state_patches behind. They should
        // log at info, not error — otherwise they bury real bugs in error
        // metrics/alerts during every boss defeat or fight overlap.
        if state.character_state.character.last_boss_fight_won {
            tracing::info!("The last boss fight was won, so we can't try to retry boss fight");
            return false;
        }

        let Some(active_fight) = &state.active_fight else {
            return true;
        };

        if active_fight.dungeon.is_some() {
            tracing::info!("Dungeon fight is in progress");
            return false;
        }

        let Ok(fight) = game_config.require_fight_template(active_fight.fight_id) else {
            tracing::error!(
                "Failed to get fight_template with id {} ",
                active_fight.fight_id,
            );
            return false;
        };

        if fight.fight_type == FightType::CampaignBossFight && self.is_battle_active(state) {
            tracing::info!("The current fight is a boss fight, so we can't retry a boss fight");
            return false;
        };

        true
    }

    fn validate_dungeon_fight(
        &self,
        state: &OverlordState,
        dungeon_id: DungeonTemplateId,
        difficulty: i64,
    ) -> bool {
        let game_config = self.game_config.get();

        if *state
            .dungeons
            .completed_difficulties
            .get(&dungeon_id)
            .unwrap_or(&0)
            + 1
            < difficulty
        {
            tracing::error!(
                "Maximum available difficulty is {} ",
                *state
                    .dungeons
                    .completed_difficulties
                    .get(&dungeon_id)
                    .unwrap_or(&0)
                    + 1
            );
            return false;
        }

        let Ok(dungeon) = game_config.require_dungeon_template(dungeon_id) else {
            tracing::error!("Failed to get dungeon_template with id {} ", dungeon_id);
            return false;
        };

        if state.character_state.character.current_chapter_level < dungeon.chapter_level_unlock {
            tracing::error!(
                "Current chapter level is: {}, required is {} ",
                state.character_state.character.current_chapter_level,
                dungeon.chapter_level_unlock
            );
            return false;
        }

        if difficulty > dungeon.max_difficulty_level {
            tracing::error!(
                "Maximum difficulty for dungeon with id: {}, is {} ",
                dungeon_id,
                dungeon.max_difficulty_level
            );
            return false;
        }

        // Difficulties open one at a time (BAL-036 source audit). The server owns
        // that rule: without it a client could enter any difficulty directly, and
        // the first-clear bonus — which is keyed on the highest difficulty
        // cleared — would silently skip every difficulty jumped over.
        let completed_difficulty = *state
            .dungeons
            .completed_difficulties
            .get(&dungeon_id)
            .unwrap_or(&0);
        if difficulty > completed_difficulty + 1 {
            tracing::error!(
                "Dungeon {dungeon_id} difficulty {difficulty} is not open yet; \
                 highest cleared is {completed_difficulty}"
            );
            return false;
        }

        let keys_required = vec![CurrencyUnit {
            currency_id: dungeon.key_currency_id,
            amount: 1,
        }];

        if !check_can_decrease_currencies(&state.character_state.currencies, &keys_required) {
            tracing::error!("Not enough keys for running dungeon: {}", dungeon_id);
            return false;
        }

        true
    }

    fn generate_pve_fight_entities(
        &self,
        entities: &mut Vec<Entity>,
        fight: &FightTemplate,
        state: &mut OverlordState,
        rand_gen: &mut rand::rngs::StdRng,
    ) -> anyhow::Result<(Entity, Option<EntityId>)> {
        let game_config = self.game_config.get();

        fight.fight_entities.iter().for_each(|entity| {
            let created_entity = match entities::create_pve_entity(
                uuid::Builder::from_random_bytes(rand_gen.random()).into_uuid(),
                entity,
                &game_config,
                None,
            ) {
                Ok(entity) => entity,
                Err(err) => {
                    tracing::error!("Failed creating entity {}", err.to_string());
                    return;
                }
            };

            entities.push(created_entity);
        });

        // BAL-015: the durable gauge mirrors the fight-local one, so it resets
        // with it — the client never renders last fight's leftover charge.
        state.flip_state.progress = 0.0;
        let player = match entities::create_player_entity(
            &state.character_state,
            state.flip_state,
            uuid::Builder::from_random_bytes(rand_gen.random()).into_uuid(),
            &game_config,
        ) {
            Ok(entity) => entity,
            Err(err) => {
                anyhow::bail!("Failed creating player entity {}", err);
            }
        };

        entities.push(player.clone());

        let mut party_player_id = None;
        if let Some(party_state) = &state.party.party_state {
            match entities::create_party_entity(
                party_state,
                state.party.party_flip_state.unwrap_or_default(),
                uuid::Builder::from_random_bytes(rand_gen.random()).into_uuid(),
                &game_config,
            ) {
                Ok(party_ally) => {
                    party_player_id = Some(party_ally.id);
                    entities.push(party_ally);
                }
                Err(err) => {
                    tracing::error!("Failed creating party ally entity {}", err);
                }
            }
        }

        entities::stamp_gauge_hp_shares(entities);
        Ok((player, party_player_id))
    }

    fn generate_pvp_fight_entities(
        &self,
        entities: &mut Vec<Entity>,
        fight: &FightTemplate,
        pvp_state: &PVPState,
        state: &mut OverlordState,
        rand_gen: &mut rand::rngs::StdRng,
    ) -> anyhow::Result<(Entity, Option<EntityId>)> {
        let game_config = self.game_config.get();

        fight.fight_entities.iter().for_each(|entity| {
            let created_entity = match entity.entity_type {
                EntityType::PVEEntity {
                    entity_template_id: _,
                } => {
                    match entities::create_pve_entity(
                        uuid::Builder::from_random_bytes(rand_gen.random()).into_uuid(),
                        entity,
                        &game_config,
                        None,
                    ) {
                        Ok(entity) => entity,
                        Err(err) => {
                            tracing::error!("{}", err.to_string());
                            return;
                        }
                    }
                }
                EntityType::PVPEntity => {
                    match entities::create_pvp_entity(
                        &EntityState::Opponent(&pvp_state.opponent_state),
                        pvp_state.opponent_flip_state,
                        entity,
                        &game_config,
                    ) {
                        Ok(entity) => entity,
                        Err(err) => {
                            tracing::error!("{}", err.to_string());
                            return;
                        }
                    }
                }
            };

            entities.push(created_entity);
        });

        // BAL-015: the durable gauge mirrors the fight-local one, so it resets
        // with it — the client never renders last fight's leftover charge.
        state.flip_state.progress = 0.0;
        // Arena world switch: PvP fights start on the side the player picked on
        // the select-opponent screen, not on the durable flip side. Only the
        // fight-local seed is overridden — the durable state keeps evolving
        // through in-fight flips as usual.
        let mut player_flip_state = state.flip_state;
        if let Some(side) = state.character_state.character.arena_world_side {
            player_flip_state.active_side = side;
        }
        let player = match entities::create_player_entity(
            &state.character_state,
            player_flip_state,
            uuid::Builder::from_random_bytes(rand_gen.random()).into_uuid(),
            &game_config,
        ) {
            Ok(entity) => entity,
            Err(err) => {
                anyhow::bail!("Failed creating player entity {}", err);
            }
        };

        entities.push(player.clone());

        // No party ally in PvP — arena is strictly 1v1
        let party_player_id = None;

        entities::stamp_gauge_hp_shares(entities);
        Ok((player, party_player_id))
    }

    fn run_prepare_fight_script(
        &self,
        events: &mut Vec<EventPluginized<OverlordEvent, OverlordState>>,
        fight: &FightTemplate,
        active_fight: &ActiveFight,
        state: &mut OverlordState,
        rand_gen: rand::rngs::StdRng,
    ) -> anyhow::Result<()> {
        let game_config = self.game_config.get();
        let current_chapter = state.character_state.character.current_chapter_level;

        // Native `prepare_fight` interpreter: spawn the initial wave from the
        // typed `prepare_fight_waves` config (the native data source), replacing
        // migrated next-wave path in `logic::fighting`.
        //
        // `base_power` was a literal argument of the legacy `spawn_wave(...)`
        // script, transpiled at deploy time from the template's TOP-LEVEL
        // `power` field (`$.power`). The earlier port passed
        // `wave_data.power` — the waves-blob's inner `power`, an unrelated
        // value that runs 4-26x larger on live campaign templates — which
        // inflated every campaign mob's hp/attack by 2-5x vs the legacy
        // engine. Pass the template's own `power`, like the scripts did.
        // Templates that spawn their enemies directly via `fight_entities` carry
        // optional extra wave-spawn step, not a requirement). Treat an absent
        // wave config as "nothing to spawn here" instead of an error — the
        // initial entities are already built by `generate_pve_fight_entities`.
        let Some(waves_cfg) = fight.prepare_fight_waves.as_ref() else {
            return Ok(());
        };

        let wave_data = crate::mechanics::fight::wave_data_from_config(waves_cfg);
        let fight_type_str = format!("{:?}", fight.fight_type);
        let mut sink = crate::mechanics::fight::NativeSink::default();
        let rng = GameRng::new(rand_gen);
        if let Err(err) = crate::mechanics::fight::spawn_wave(
            &mut sink,
            &rng,
            &game_config,
            self.behaviors.lookups(),
            active_fight,
            &wave_data,
            fight.power.map(|p| p as f64).unwrap_or(0.0),
            current_chapter,
            &fight_type_str,
        ) {
            anyhow::bail!("Prepare fight wave spawn failed with error: {err:?}");
        }

        events.append(&mut sink.events.into_iter().map(EventPluginized::now).collect());

        Ok(())
    }

    /// Schedule `StartFight` for a freshly prepared fight, honoring the
    /// template's start delay override.
    fn schedule_start_fight(
        &mut self,
        fight: &FightTemplate,
        fight_id: uuid::Uuid,
        game_config: &GameConfig,
    ) {
        let start_fight_delay = fight
            .start_fight_delay_ticks
            .unwrap_or(game_config.fight_settings.start_fight_delay_ticks_default);

        self.fight_clock
            .schedule(OverlordEvent::StartFight { fight_id }, start_fight_delay);
    }

    fn handle_pve_prepare_fight(
        &mut self,
        state: &mut OverlordState,
        mut rand_gen: rand::rngs::StdRng,
    ) -> Option<Vec<EventPluginized<OverlordEvent, OverlordState>>> {
        let game_config = self.game_config.get();

        let mut events = vec![];
        let Ok(chapter) = game_config
            .require_chapter_by_level(state.character_state.character.current_chapter_level)
        else {
            tracing::error!(
                "Failed to get chapter with chapter_level={}",
                state.character_state.character.current_chapter_level
            );
            return None;
        };

        let Some(fight_id) = chapter
            .fight_ids
            .get(state.character_state.character.current_fight_number as usize)
            .cloned()
        else {
            tracing::error!(
                "Failed to get fight {} for chapter_level={}",
                state.character_state.character.current_fight_number,
                state.character_state.character.current_chapter_level
            );
            return None;
        };

        let Ok(fight) = game_config.require_fight_template(fight_id) else {
            tracing::error!("Failed to get fight_template with id {} ", fight_id,);
            return None;
        };

        let mut entities = vec![];

        let (player, party_player_id) =
            match self.generate_pve_fight_entities(&mut entities, fight, state, &mut rand_gen) {
                Ok(result) => result,
                Err(err) => {
                    tracing::error!("Got error, while creating PVE entities: {}", err);
                    return None;
                }
            };

        let active_fight = ActiveFight {
            id: uuid::Builder::from_random_bytes(rand_gen.random()).into_uuid(),
            fight_id,
            current_wave: 1,
            player_id: player.id,
            party_player_id,
            entities,
            max_duration_ticks: fight.max_duration_ticks,
            fight_stopped: false,
            fight_ended: false,
            dungeon: None,
            paused: false,
            pending_wave_spawns: Vec::new(),
            summoned_entity_ids: Vec::new(),
        };

        state.active_fight = Some(active_fight.clone());

        if let Err(err) =
            self.run_prepare_fight_script(&mut events, fight, &active_fight, state, rand_gen)
        {
            tracing::error!("Error running pve prepare_fight script: {err:?}");
            return None;
        }

        self.schedule_start_fight(fight, active_fight.id, &game_config);

        tracing::debug!(
            "Starting chapter_level={}, fight_number={}, fight_id={}",
            chapter.level,
            state.character_state.character.current_fight_number,
            fight_id,
        );

        Some(events)
    }

    fn handle_pvp_prepare_fight(
        &mut self,
        state: &mut OverlordState,
        fight_id: FightTemplateId,
        pvp_state: Box<PVPState>,
        mut rand_gen: rand::rngs::StdRng,
    ) -> Option<Vec<EventPluginized<OverlordEvent, OverlordState>>> {
        let game_config = self.game_config.get();
        // Applied to the state only after the fight is successfully built:
        // failure results still apply their state, and a half-set `pvp_state`
        // would make `handle_prepare_fight`'s entry guard reject every
        // subsequent PrepareFight for the session.
        let prepared_pvp_state = *pvp_state.clone();

        let mut events = vec![];

        let Ok(fight) = game_config.require_fight_template(fight_id) else {
            tracing::error!("Failed to get fight_template with id {} ", fight_id);
            return None;
        };

        let mut entities = vec![];

        let (player, party_player_id) = match self.generate_pvp_fight_entities(
            &mut entities,
            fight,
            &pvp_state,
            state,
            &mut rand_gen,
        ) {
            Ok(result) => result,
            Err(err) => {
                tracing::error!("Got error, while creating PVP entities: {}", err);
                return None;
            }
        };

        let active_fight = ActiveFight {
            id: uuid::Builder::from_random_bytes(rand_gen.random()).into_uuid(),
            fight_id,
            current_wave: 1,
            player_id: player.id,
            party_player_id,
            entities,
            max_duration_ticks: fight.max_duration_ticks,
            fight_stopped: false,
            fight_ended: false,
            dungeon: None,
            paused: false,
            pending_wave_spawns: Vec::new(),
            summoned_entity_ids: Vec::new(),
        };

        state.active_fight = Some(active_fight.clone());

        if let Err(err) =
            self.run_prepare_fight_script(&mut events, fight, &active_fight, state, rand_gen)
        {
            tracing::error!("Error running pvp prepare_fight script: {err:?}");
            return None;
        }

        state.pvp_state = Some(prepared_pvp_state);

        self.schedule_start_fight(fight, active_fight.id, &game_config);

        Some(events)
    }

    fn handle_dungeon_prepare_fight(
        &mut self,
        state: &mut OverlordState,
        dungeon_id: DungeonTemplateId,
        difficulty: i64,
        mut rand_gen: rand::rngs::StdRng,
    ) -> Option<Vec<EventPluginized<OverlordEvent, OverlordState>>> {
        let game_config = self.game_config.get();

        let Ok(dungeon) = game_config.require_dungeon_template(dungeon_id) else {
            tracing::error!("Failed to get dungeon_template with id {} ", dungeon_id);
            return None;
        };

        let Some(fight_template_id) = dungeon.fight_template_ids.get((difficulty - 1) as usize)
        else {
            tracing::error!(
                "Failed to get fight_template from dungeon with id {}, for difficulty: {}",
                dungeon_id,
                difficulty
            );
            return None;
        };

        let Ok(fight) = game_config.require_fight_template(*fight_template_id) else {
            tracing::error!(
                "Failed to get fight_template with id {} ",
                fight_template_id
            );
            return None;
        };

        let mut events = vec![];

        let mut entities = vec![];

        let (player, party_player_id) =
            match self.generate_pve_fight_entities(&mut entities, fight, state, &mut rand_gen) {
                Ok(result) => result,
                Err(err) => {
                    tracing::error!("Got error, while creating entities: {}", err);
                    return None;
                }
            };

        let active_fight = ActiveFight {
            id: uuid::Builder::from_random_bytes(rand_gen.random()).into_uuid(),
            fight_id: *fight_template_id,
            current_wave: 1,
            player_id: player.id,
            party_player_id,
            entities,
            max_duration_ticks: fight.max_duration_ticks,
            fight_stopped: false,
            fight_ended: false,
            dungeon: Some(ActiveDungeon {
                id: dungeon_id,
                difficulty,
            }),
            paused: false,
            pending_wave_spawns: Vec::new(),
            summoned_entity_ids: Vec::new(),
        };

        state.active_fight = Some(active_fight.clone());

        if let Err(err) =
            self.run_prepare_fight_script(&mut events, fight, &active_fight, state, rand_gen)
        {
            tracing::error!("Error running pvp prepare_fight script: {err:?}");
            return None;
        }

        self.schedule_start_fight(fight, active_fight.id, &game_config);

        Some(events)
    }

    fn handle_single_prepare_fight(
        &mut self,
        state: &mut OverlordState,
        fight_template_id: FightTemplateId,
        mut rand_gen: rand::rngs::StdRng,
    ) -> Option<Vec<EventPluginized<OverlordEvent, OverlordState>>> {
        let game_config = self.game_config.get();

        let Ok(fight) = game_config.require_fight_template(fight_template_id) else {
            tracing::error!(
                "Failed to get fight_template with id {} ",
                fight_template_id
            );
            return None;
        };

        let mut events = vec![];

        let mut entities = vec![];

        let (player, party_player_id) =
            match self.generate_pve_fight_entities(&mut entities, fight, state, &mut rand_gen) {
                Ok(result) => result,
                Err(err) => {
                    tracing::error!("Got error, while creating entities: {}", err);
                    return None;
                }
            };

        let active_fight = ActiveFight {
            id: uuid::Builder::from_random_bytes(rand_gen.random()).into_uuid(),
            fight_id: fight_template_id,
            current_wave: 1,
            player_id: player.id,
            party_player_id,
            entities,
            max_duration_ticks: fight.max_duration_ticks,
            fight_stopped: false,
            paused: false,
            fight_ended: false,
            dungeon: None,
            pending_wave_spawns: Vec::new(),
            summoned_entity_ids: Vec::new(),
        };

        state.active_fight = Some(active_fight.clone());

        if let Err(err) =
            self.run_prepare_fight_script(&mut events, fight, &active_fight, state, rand_gen)
        {
            tracing::error!("Error running single prepare_fight script: {err:?}");
            return None;
        }

        self.schedule_start_fight(fight, active_fight.id, &game_config);

        Some(events)
    }

    pub fn handle_start_fight(
        &mut self,
        event: OverlordEvent,
        fight_id: uuid::Uuid,
        // The native `fight_start` fns are RNG-free (init_fight + optional static
        // spawns), but the Team Die of the starting world is rolled from here.
        rand_gen: rand::rngs::StdRng,
        current_tick: u64,
        mut state: OverlordState,
    ) -> EventHandleResult<OverlordEvent, OverlordState> {
        let game_config = self.game_config.get();

        let Some(active_fight) = &mut state.active_fight else {
            tracing::error!("No active fight for start_fight");
            return EventHandleResult::fail(state);
        };

        if active_fight.id != fight_id {
            tracing::error!(
                "StartFight fight_id mismatch: expected {}, got {}",
                active_fight.id,
                fight_id
            );
            return EventHandleResult::fail(state);
        }

        let Ok(fight_template) = game_config.require_fight_template(active_fight.fight_id) else {
            tracing::error!("No fight template with fight_id: {}", active_fight.fight_id);
            return EventHandleResult::fail(state);
        };

        self.start_fight_tick = current_tick;

        let mut events = vec![];

        active_fight.entities.iter_mut().for_each(|e| {
            e.abilities.iter_mut().for_each(|ability| {
                e.actions_queue.push(&ActionWithDeadline::core(
                    self.make_start_cast_ability_action(e.id, ability.ability.template_id),
                    current_tick,
                ))
            })
        });

        active_fight.max_duration_ticks = fight_template.max_duration_ticks;

        let active_fight_for_native = active_fight.clone();
        let start_behavior = fight_template.start_behavior.clone();
        // scope; the native `fight_start` fns read it from `Fight`/`State`, so
        // the StartFight event itself is no longer an input.
        let _ = event;

        // Native `fight_start` port: look up the named fn on the registry and
        // drive it with `FightStartCtx`. Falls back to no events when no native
        // ref / fn is present.
        let Some(native_name) = start_behavior.as_deref() else {
            tracing::error!("Fight template {} has no start_behavior", fight_template.id);
            return EventHandleResult::fail(state);
        };
        let Some(native_fn) = self.behaviors.fight_start_fn(native_name) else {
            tracing::error!("No registered fight_start native fn named {native_name}");
            return EventHandleResult::fail(state);
        };
        match native_fn(&crate::behaviors::combat::fight_start::FightStartCtx {
            fight: &active_fight_for_native,
            state: &state,
            lookups: self.behaviors.lookups(),
        }) {
            Ok(start_fight_events) => {
                events.append(
                    &mut start_fight_events
                        .into_iter()
                        .map(EventPluginized::now)
                        .collect(),
                );
            }
            Err(err) => {
                tracing::error!("Start fight native fn failed with error: {err:?}");
                return EventHandleResult::fail(state);
            }
        };

        // Entering the starting world rolls the Team Die, the same die a flip
        // rolls. Last, so the outcome resolves against the fight the native
        // `fight_start` fn has finished building.
        let mut rng = rand_gen;
        events.extend(self.roll_team_die_on_fight_start(&mut state, &mut rng));

        EventHandleResult::ok_events(state, events)
    }

    pub fn get_prepare_fight_delay(
        &self,
        is_win: bool,
        chapter: &Chapter,
        state: &OverlordState,
    ) -> u64 {
        let game_config = self.game_config.get();

        let Some(fight_id) = chapter
            .fight_ids
            .get(state.character_state.character.current_fight_number as usize)
            .cloned()
        else {
            tracing::error!(
                "Failed to get fight {} for chapter_level={}",
                state.character_state.character.current_fight_number,
                state.character_state.character.current_chapter_level
            );
            return 0;
        };

        let Ok(fight) = game_config.require_fight_template(fight_id) else {
            tracing::error!("Failed to get fight_template with id {} ", fight_id,);
            return 0;
        };

        if is_win {
            fight.prepare_fight_win_duration_ticks.unwrap_or(
                game_config
                    .fight_settings
                    .prepare_fight_win_delay_ticks_default,
            )
        } else {
            fight.prepare_fight_lose_duration_ticks.unwrap_or(
                game_config
                    .fight_settings
                    .prepare_fight_lose_delay_ticks_default,
            )
        }
    }

    pub fn get_end_fight_delay(&self, fight_id: FightTemplateId) -> u64 {
        let game_config = self.game_config.get();

        let Ok(fight) = game_config.require_fight_template(fight_id) else {
            tracing::error!("Failed to get fight_template with id {} ", fight_id,);
            return 0;
        };

        fight
            .end_fight_delay_ticks
            .unwrap_or(game_config.fight_settings.end_fight_delay_ticks_default)
    }

    /// Shared end-of-fight tail: keep the loop going by scheduling the next
    /// PrepareFight, or stop fighting when the template says so.
    /// If (is_win && stop_on_win) or (!is_win && stop_on_lose) - we stop fighting
    fn continue_or_stop_fight_loop(
        &mut self,
        state: &mut OverlordState,
        events: &mut Vec<EventPluginized<OverlordEvent, OverlordState>>,
        current_fight: &FightTemplate,
        is_win: bool,
        prepare_fight_delay_ticks: u64,
    ) {
        if (!is_win || !current_fight.stop_on_win) && (is_win || !current_fight.stop_on_lose) {
            events.extend(self.maybe_refresh_party_event(state));
            self.fight_clock.schedule(
                OverlordEvent::PrepareFight {
                    prepare_fight_type: PrepareFightType::PVEFight,
                },
                prepare_fight_delay_ticks,
            );
        } else if let Some(fight) = state.active_fight.as_mut() {
            fight.fight_stopped = true;
        }
    }

    fn end_pvp_fight(
        &mut self,
        is_win: bool,
        pvp_state: &PVPState,
        prepare_fight_delay_ticks: u64,
        current_fight: &FightTemplate,
        mut state: OverlordState,
    ) -> EventHandleResult<OverlordEvent, OverlordState> {
        let mut events = vec![];
        if let Some(vassal) = &pvp_state.vassal {
            if is_win {
                state.character_state.vassals.push(vassal.clone());
                if let Some(bundle_id) = current_fight.bundle_reward_id {
                    events.push(EventPluginized::now(OverlordEvent::AddBundleGroup {
                        bundle_ids: vec![bundle_id],
                        source: essences::currency::CurrencySource::PvpVassalReward,
                    }));
                }
            } else {
                tracing::error!("Expected to add vassal, but PVP is lost");
            }
        }

        if let Some(rating_change) = &pvp_state.rating_change {
            if is_win {
                state.character_state.character.arena_rating +=
                    rating_change.winner_rating_increase;
            } else {
                state.character_state.character.arena_rating += rating_change.loser_rating_decrease;
            }
        }

        state.pvp_state = None;

        self.continue_or_stop_fight_loop(
            &mut state,
            &mut events,
            current_fight,
            is_win,
            prepare_fight_delay_ticks,
        );

        EventHandleResult::ok_events(state, events)
    }

    fn end_pve_fight(
        &mut self,
        is_win: bool,
        current_chapter: &Chapter,
        prepare_fight_delay_ticks: u64,
        current_fight: &FightTemplate,
        rand_gen: &mut rand::rngs::StdRng,
        mut state: OverlordState,
    ) -> EventHandleResult<OverlordEvent, OverlordState> {
        let game_config = self.game_config.get();

        let mut events = vec![];

        let prev_chapter_level = state.character_state.character.current_chapter_level;
        let mut next_chapter_level = prev_chapter_level;
        let mut next_fight_number = state.character_state.character.current_fight_number + 1;

        if is_win {
            if next_fight_number >= current_chapter.fight_ids.len() as i64 {
                next_fight_number = 0;
                if game_config
                    .chapter_by_level(next_chapter_level + 1)
                    .is_some()
                {
                    state.character_state.character.last_boss_fight_won = true;
                    next_chapter_level += 1;
                }
            } else {
                let next_fight_id = &current_chapter.fight_ids[next_fight_number as usize];

                let Ok(next_fight) = game_config.require_fight_template(*next_fight_id) else {
                    tracing::error!(
                        "Failed to get next fight_template with id={}",
                        next_fight_id
                    );
                    return EventHandleResult::fail(state);
                };

                // Uncomment to restore the old behavior where losing the boss
                // gates auto-progression wave -> boss until a RetryBossFight event.
                // if !state.character_state.character.last_boss_fight_won
                //     && next_fight.fight_type == FightType::CampaignBossFight
                // {
                //     next_fight_number = 0;
                // }
                let _ = next_fight;
            }

            state.character_state.character.current_chapter_level = next_chapter_level;
            state.character_state.character.current_fight_number = next_fight_number;

            // BAL-024/BAL-025: reaching an equipment gate guarantees its stone
            // package. The grant rides the ordinary PlayerNewStones path, and
            // the missing-check makes it a no-op for anything already owned.
            let (trigger_stones, effect_stones) = Self::missing_milestone_stones(
                &game_config,
                next_chapter_level,
                &state.character_state.stones,
            );
            if !trigger_stones.is_empty() || !effect_stones.is_empty() {
                events.push(EventPluginized::now(OverlordEvent::PlayerNewStones {
                    trigger_stones,
                    effect_stones,
                }));
            }

            // BAL-031: each newly-usable artifact socket family guarantees one
            // compatible L1 stone. Same converging missing-check as the
            // equipment milestones — a family the player already owns a stone
            // of needs nothing, so the grant is idempotent.
            let socket_stones = Self::missing_socket_artifact_stones(
                &game_config,
                next_chapter_level,
                &state.character_state.artifacts,
            );
            if !socket_stones.is_empty() {
                events.push(EventPluginized::now(OverlordEvent::PlayerNewArtifacts {
                    artifacts: Vec::new(),
                    artifact_stones: socket_stones,
                }));
            }

            // BAL-010: reaching the Cores gate grants BOTH cores at level 1 for
            // free -- the old level-1 price of 700 is gone from the ladder, which
            // now starts at level 2. Written directly rather than through an
            // event because it is a plain level floor, and the `<` checks make it
            // idempotent: it converges on every chapter advance and on reconnect,
            // and can never lower a core the player has already upgraded.
            let core_unlock_chapter = game_config.cores_settings.unlock_chapter;
            if next_chapter_level >= core_unlock_chapter {
                let cores = &mut state.character_state.cores;
                // Per side, and only for a side we actually raise: the level
                // alone is not the grant, because law unlocks are keyed to core
                // level and a granted core without its level-1 laws is one the
                // player cannot build on. Touching a side that was already above
                // 1 would backfill laws the ordinary upgrade path owns.
                for side in [
                    essences::flip::WorldSide::Real,
                    essences::flip::WorldSide::Fantasy,
                ] {
                    let level = match side {
                        essences::flip::WorldSide::Real => &mut cores.real_level,
                        essences::flip::WorldSide::Fantasy => &mut cores.fantasy_level,
                    };
                    if *level >= 1 {
                        continue;
                    }
                    *level = 1;
                    crate::mechanics::cores::grant_unlocked_laws(&game_config, cores, side);
                }
            }

            // D1 all-free supply floor: the grant belongs to the transition,
            // not merely to "being above" the gate. That makes it exactly once
            // for fresh progression, immune to reconnect/replayed convergence,
            // and deliberately non-retroactive for migrated accounts that were
            // already beyond ch21 before this knob existed.
            if let Some(grant) = core_unlock_essence_grant_event(
                &game_config,
                prev_chapter_level,
                next_chapter_level,
            ) {
                events.push(grant);
            }

            if self.should_reset_afk_timer_on_gating_unlock(prev_chapter_level, &state) {
                events.push(EventPluginized::now(
                    OverlordEvent::AfkRewardsGatingUnlocked {},
                ));
            }

            if !state.character_state.character.rate_us_shown
                && let Some(threshold) = game_config.game_settings.rate_us_chapter
                && prev_chapter_level < threshold
                && next_chapter_level >= threshold
            {
                events.push(EventPluginized::now(OverlordEvent::ShowRateUs {}));
                tracing::info!(
                    target: METRICS_TARGET,
                    event_type = "show_rate_us",
                    character_id = %state.character_state.character.id,
                    chapter_level = next_chapter_level,
                    trigger = "chapter_advance",
                    "Show rate us",
                );
            }

            if let Some(bundle_id) = current_fight.bundle_reward_id {
                events.push(EventPluginized::now(OverlordEvent::AddBundleGroup {
                    bundle_ids: vec![bundle_id],
                    source: essences::currency::CurrencySource::ChapterReward,
                }));
            }

            // BAL-035: Mastery is paid by campaign-boss FIRST clears only. A
            // chapter advance IS that first clear — `current_chapter_level` is
            // monotonic — so a farmed boss pays nothing and the schedule stays
            // tied to progression rather than to time spent.
            if next_chapter_level > prev_chapter_level {
                events.extend(Self::mastery_first_clear_rewards(
                    &game_config,
                    &mut state,
                    next_chapter_level,
                    rand_gen,
                ));
            }
        } else {
            if current_fight.fight_type == FightType::CampaignBossFight {
                state.character_state.character.last_boss_fight_won = false;
            }

            next_fight_number = 0;
            state.character_state.character.current_fight_number = next_fight_number;
        }

        // The starting world alternates on DEATH (restart) and on entering the
        // NEXT CHAPTER — wave→wave advances inside a chapter keep it, and the
        // chapter boss is always entered in the world its waves were fought in
        // (guard below; it also covers a death-restart into a boss-only
        // chapter). Only once the flip is unlocked: before that the player
        // stays on the default side. The durable side seeds the next fight's
        // entity at PrepareFight, and the EndFight ws handler already persists
        // `flip_state`.
        let world_changing_transition = !is_win || next_chapter_level != prev_chapter_level;
        if world_changing_transition && game_config.flip_settings.is_unlocked(next_chapter_level) {
            let upcoming_chapter = if next_chapter_level == prev_chapter_level {
                Some(current_chapter)
            } else {
                game_config.chapter_by_level(next_chapter_level)
            };
            let upcoming_is_boss = upcoming_chapter
                .and_then(|chapter| chapter.fight_ids.get(next_fight_number as usize))
                .and_then(|fight_id| game_config.fight_template(*fight_id))
                .is_some_and(|fight| fight.fight_type == FightType::CampaignBossFight);

            if !upcoming_is_boss {
                state.flip_state.active_side = state.flip_state.active_side.flipped();
                state.flip_state.revision = state.flip_state.revision.saturating_add(1);
            }
        }

        self.continue_or_stop_fight_loop(
            &mut state,
            &mut events,
            current_fight,
            is_win,
            prepare_fight_delay_ticks,
        );

        EventHandleResult::ok_events(state, events)
    }

    /// Shards for the `clear`-th campaign-boss first clear, plus the Crystal
    /// roll that rides the same event (BAL-035).
    ///
    /// The Crystal is a `25%` roll with a pity counter: the fourth consecutive
    /// miss pays out. The counter lives in `custom_values` — it is one small
    /// integer of progression state, and putting it there keeps the schedule
    /// off the characters table.
    pub(crate) fn mastery_first_clear_rewards(
        game_config: &configs::game_config::GameConfig,
        state: &mut OverlordState,
        clear: i64,
        rand_gen: &mut rand::rngs::StdRng,
    ) -> Vec<EventPluginized<OverlordEvent, OverlordState>> {
        let mastery = &game_config.game_settings.mastery;
        let mut rewards = Vec::new();

        let shards = mastery.shards_for_clear(clear);
        if shards > 0 {
            rewards.push(CurrencyUnit {
                currency_id: mastery.shard_currency_id,
                amount: shards,
            });
        }

        let misses = state
            .character_state
            .character
            .custom_values
            .0
            .get(MASTERY_CRYSTAL_MISSES)
            .copied()
            .unwrap_or(0);
        let pity_due = mastery.crystal_pity_clears > 0 && misses + 1 >= mastery.crystal_pity_clears;
        let won =
            pity_due || rand::RngExt::random_range(rand_gen, 0.0..1.0) < mastery.crystal_chance;

        if won {
            rewards.push(CurrencyUnit {
                currency_id: mastery.crystal_currency_id,
                amount: 1,
            });
        }

        let next_misses = if won { 0 } else { misses + 1 };
        state
            .character_state
            .character
            .custom_values
            .0
            .insert(MASTERY_CRYSTAL_MISSES.to_string(), next_misses);

        // The pity counter is account-wide (BAL-035), so the in-memory write
        // above is not enough: `characters.custom_values` is only written from
        // the `SetCustomValue` path, and without the event the counter resets
        // to its stored value on every reconnect. Emitted even when the clear
        // pays nothing — a miss still has to be counted.
        let mut events = vec![EventPluginized::now(OverlordEvent::SetCustomValue {
            key: MASTERY_CRYSTAL_MISSES.to_string(),
            value: next_misses,
        })];

        if !rewards.is_empty() {
            events.push(Self::currency_increase(
                &rewards,
                essences::currency::CurrencySource::ChapterReward,
            ));
        }
        events
    }

    fn end_dungeon_fight(
        &mut self,
        is_win: bool,
        prepare_fight_delay_ticks: u64,
        active_dungeon: &ActiveDungeon,
        current_fight: &FightTemplate,
        rand_gen: &mut rand::rngs::StdRng,
        mut state: OverlordState,
    ) -> EventHandleResult<OverlordEvent, OverlordState> {
        let game_config = self.game_config.get();

        let Ok(dungeon) = game_config.require_dungeon_template(active_dungeon.id) else {
            tracing::error!(
                "Failed to get dungeon_template with id {} ",
                active_dungeon.id
            );
            return EventHandleResult::fail(state);
        };

        let mut events = vec![];

        if is_win {
            let keys_price = vec![CurrencyUnit {
                currency_id: dungeon.key_currency_id,
                amount: 1,
            }];

            let Some(currency_event) =
                Self::currency_decrease(&state, &keys_price, CurrencyConsumer::DungeonFightEnd)
            else {
                return EventHandleResult::fail(state);
            };
            events.push(currency_event);

            // BAL-036: the first clear of a difficulty pays its resource bundle
            // TWICE, every later clear once. Difficulties open in order, so the
            // highest one cleared is an exact record of what has been paid.
            let is_first_clear = active_dungeon.difficulty
                > *state
                    .dungeons
                    .completed_difficulties
                    .get(&dungeon.id)
                    .unwrap_or(&0);

            if let Some(bundle_id) = current_fight.bundle_reward_id {
                // Two separate single-bundle grants rather than one grouped
                // pair: a dungeon reward stays an ordinary ungrouped bundle
                // row, and grouping keeps meaning "this came from a raid".
                let payouts = if is_first_clear { 2 } else { 1 };
                for _ in 0..payouts {
                    events.push(EventPluginized::now(OverlordEvent::AddBundleGroup {
                        bundle_ids: vec![bundle_id],
                        source: essences::currency::CurrencySource::DungeonReward,
                    }));
                }
            }

            // Artifact stones drop from chapter bosses and cleared dungeons — a
            // different faucet from the artifacts themselves, so the two
            // collections never compete. One roll per clear: the first-clear
            // bonus doubles the resource bundle and nothing else (BAL-036).
            let dungeon_chance = game_config.artifacts_settings.stone_drop.dungeon_chance;
            events.extend(self.roll_artifact_stone_drop(
                rand_gen,
                dungeon_chance,
                state.character_state.character.current_chapter_level,
            ));

            state
                .dungeons
                .completed_difficulties
                .entry(dungeon.id)
                .and_modify(|v| {
                    if active_dungeon.difficulty > *v {
                        *v = active_dungeon.difficulty;
                    }
                })
                .or_insert(active_dungeon.difficulty);
        }

        self.continue_or_stop_fight_loop(
            &mut state,
            &mut events,
            current_fight,
            is_win,
            prepare_fight_delay_ticks,
        );

        EventHandleResult::ok_events(state, events)
    }

    /// Pure predicate: is the chapter transition crossing the AFK rewards unlock
    /// threshold *and* is the elapsed time since the last claim shorter than
    /// `min_required_time_sec`? Callers use this to decide whether to emit
    /// `AfkRewardsGatingUnlocked`; the async handler for that event owns the
    /// actual state mutation.
    pub(crate) fn should_reset_afk_timer_on_gating_unlock(
        &self,
        prev_chapter_level: i64,
        state: &OverlordState,
    ) -> bool {
        let game_config = self.game_config.get();
        let unlock_chapter = game_config.gatings.afk_rewards_button_unlock_chapter;
        let new_chapter_level = state.character_state.character.current_chapter_level;

        if prev_chapter_level >= unlock_chapter || new_chapter_level < unlock_chapter {
            return false;
        }

        let min_required_time_sec = game_config.afk_rewards_settings.min_required_time_sec as i64;
        let elapsed = ::time::utc_now()
            .signed_duration_since(state.character_state.character.last_afk_reward_claimed_at)
            .num_seconds();

        elapsed < min_required_time_sec
    }

    pub fn handle_afk_rewards_gating_unlocked(
        &self,
        mut state: OverlordState,
    ) -> EventHandleResult<OverlordEvent, OverlordState> {
        let min_required_time_sec = self
            .game_config
            .get()
            .afk_rewards_settings
            .min_required_time_sec as i64;
        state.character_state.character.last_afk_reward_claimed_at =
            ::time::utc_now() - chrono::Duration::seconds(min_required_time_sec);
        EventHandleResult::ok(state)
    }

    /// `rand_gen` is threaded in for the placeholder artifact-stone drop on a
    /// cleared dungeon; nothing else in the end-of-fight pipeline rolls.
    pub fn handle_end_fight(
        &mut self,
        fight_id: uuid::Uuid,
        is_win: bool,
        pvp_state: Option<&PVPState>,
        mut rand_gen: rand::rngs::StdRng,
        mut state: OverlordState,
    ) -> EventHandleResult<OverlordEvent, OverlordState> {
        let game_config = self.game_config.get();

        let Some(active_fight) = &mut state.active_fight else {
            tracing::error!("No active fight for end_fight");
            return EventHandleResult::fail(state);
        };

        if active_fight.id != fight_id {
            tracing::error!(
                "EndFight fight_id mismatch: expected {}, got {}",
                active_fight.id,
                fight_id
            );
            return EventHandleResult::fail(state);
        }

        if self.ended_fight_id == Some(fight_id) {
            // A duplicate EndFight for the same fight is an expected race:
            // the max-duration timeout and an in-flight combat event (e.g. a
            // projectile killing the last entity) can each schedule one. The
            // end-of-fight pipeline (PvP rating, vassal, reward bundles,
            // chapter advance) must run exactly once per fight. (The
            // `fight_ended` flag can't be the guard here: producers set it
            // before scheduling the EndFight event.)
            tracing::info!("Fight {fight_id} already ended, ignoring duplicate EndFight");
            return EventHandleResult::fail(state);
        }

        let Ok(current_fight) = game_config.require_fight_template(active_fight.fight_id) else {
            tracing::error!(
                "Failed to get currrent fight_template with id={}",
                active_fight.fight_id
            );
            return EventHandleResult::fail(state);
        };

        active_fight.fight_ended = true;
        self.ended_fight_id = Some(fight_id);

        // Before the mode branch, so campaign, arena and dungeon all get the
        // summary from one place — and after the idempotency guard above, which
        // makes the write exactly-once for free. `SingleFight` returns early a
        // few lines down and would otherwise be the one mode without it.
        let duration_ticks = self.fight_clock.now().saturating_sub(self.start_fight_tick);
        let fight_template_id = active_fight.fight_id;
        // Always written, even when nothing was dealt: an empty summary for
        // THIS fight is a different statement from the previous fight's summary
        // left standing, and the client would have no way to tell them apart.
        state.last_fight_breakdown = Some(
            self.fight_breakdown_acc
                .take()
                .filter(|acc| acc.fight_instance_id() == fight_id)
                .map(|acc| acc.finish(is_win, duration_ticks))
                .unwrap_or_else(|| essences::fight_breakdown::FightBreakdown {
                    fight_instance_id: fight_id,
                    fight_id: fight_template_id,
                    is_win,
                    duration_ticks,
                    actors: Vec::new(),
                }),
        );

        if current_fight.fight_type == FightType::SingleFight {
            self.fight_clock.schedule(
                OverlordEvent::PrepareFight {
                    prepare_fight_type: PrepareFightType::PVEFight,
                },
                game_config
                    .fight_settings
                    .prepare_fight_win_delay_ticks_default,
            );
            let events = self.maybe_refresh_party_event(&state).into_iter().collect();
            return EventHandleResult::ok_events(state, events);
        }

        let active_fight = active_fight.clone();

        let Ok(current_chapter) = game_config
            .require_chapter_by_level(state.character_state.character.current_chapter_level)
            .cloned()
        else {
            tracing::error!(
                "Failed to get chapter with chapter_level={}",
                state.character_state.character.current_chapter_level
            );
            return EventHandleResult::fail(state);
        };

        let prepare_fight_delay_ticks =
            self.get_prepare_fight_delay(is_win, &current_chapter, &state);

        if let Some(pvp_state) = pvp_state {
            return self.end_pvp_fight(
                is_win,
                pvp_state,
                prepare_fight_delay_ticks,
                current_fight,
                state,
            );
        }

        if let Some(active_dungeon) = &active_fight.dungeon {
            return self.end_dungeon_fight(
                is_win,
                prepare_fight_delay_ticks,
                active_dungeon,
                current_fight,
                &mut rand_gen,
                state,
            );
        }

        self.end_pve_fight(
            is_win,
            &current_chapter,
            prepare_fight_delay_ticks,
            current_fight,
            &mut rand_gen,
            state,
        )
    }
}

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

    #[test]
    fn crossing_the_gate_emits_one_authored_chapter_reward_and_cannot_repeat() {
        let mut config = configs::tests_game_config::generate_game_config_for_tests();
        config.cores_settings.unlock_chapter = 21;
        config.cores_settings.unlock_essence_grant = 210;
        let core_id = config.cores_settings.upgrade_currency_id;

        let grant = core_unlock_essence_grant_event(&config, 20, 21)
            .expect("the fresh 20 -> 21 crossing grants once");
        match grant.event() {
            OverlordEvent::CurrencyIncrease {
                currencies,
                currency_source,
            } => {
                assert_eq!(
                    *currency_source,
                    essences::currency::CurrencySource::ChapterReward
                );
                assert_eq!(
                    currencies,
                    &vec![CurrencyUnit {
                        currency_id: core_id,
                        amount: 210,
                    }]
                );
            }
            other => panic!("unexpected unlock grant event: {other:?}"),
        }

        assert!(
            core_unlock_essence_grant_event(&config, 21, 22).is_none(),
            "ordinary progress/reconnect above the gate cannot repeat the grant"
        );
        assert!(
            core_unlock_essence_grant_event(&config, 22, 22).is_none(),
            "reprocessing the same above-gate chapter cannot double grant"
        );
    }
}

#[cfg(test)]
mod mastery_tests {
    use super::*;
    use configs::game_settings::{MasterySettings, MasteryShardBand};

    /// The authored schedule (BAL-035): ten clears per band.
    fn bands() -> Vec<MasteryShardBand> {
        [
            (10, 10),
            (20, 25),
            (30, 30),
            (40, 35),
            (50, 40),
            (60, 45),
            (70, 50),
            (80, 50),
            (90, 55),
            (100, 60),
            (300, 60),
        ]
        .into_iter()
        .map(|(up_to_clear, shards)| MasteryShardBand {
            up_to_clear,
            shards,
        })
        .collect()
    }

    fn settings(crystal_chance: f64, crystal_pity_clears: i64) -> MasterySettings {
        MasterySettings {
            shard_currency_id: uuid::Uuid::nil(),
            crystal_currency_id: uuid::Uuid::nil(),
            shard_bands: bands(),
            crystal_chance,
            crystal_pity_clears,
        }
    }

    /// The currencies one first clear grants, and the miss counter it leaves
    /// behind. `chance` drives the roll; `0.0` never wins on its own, so the
    /// only Crystal it can produce is the pity one.
    fn grant(
        config: &configs::game_config::GameConfig,
        state: &mut OverlordState,
        clear: i64,
    ) -> Vec<CurrencyUnit> {
        use rand::SeedableRng;
        let mut rng = rand::rngs::StdRng::seed_from_u64(1);
        let events = OverlordLogic::mastery_first_clear_rewards(config, state, clear, &mut rng);
        events
            .iter()
            .filter_map(|event| match event.event() {
                OverlordEvent::CurrencyIncrease { currencies, .. } => Some(currencies.clone()),
                _ => None,
            })
            .flatten()
            .collect()
    }

    fn misses(state: &OverlordState) -> i64 {
        state
            .character_state
            .character
            .custom_values
            .0
            .get(MASTERY_CRYSTAL_MISSES)
            .copied()
            .unwrap_or(0)
    }

    /// The Crystal is never certain until the pity clear, and the pity clear is
    /// certain — that pair is what makes the schedule bounded without making it
    /// deterministic.
    #[test]
    fn the_fourth_consecutive_miss_is_guaranteed_a_crystal() {
        let mut config = configs::tests_game_config::generate_game_config_for_tests();
        // A chance of zero isolates the pity: every win here is the guarantee.
        config.game_settings.mastery.crystal_chance = 0.0;
        let crystal_id = config.game_settings.mastery.crystal_currency_id;
        let shard_id = config.game_settings.mastery.shard_currency_id;

        let mut state = OverlordState::default();

        for clear in 1..=3 {
            let granted = grant(&config, &mut state, clear);
            assert!(
                granted.iter().all(|unit| unit.currency_id == shard_id),
                "clear {clear} pays Shards only while the pity is still counting"
            );
            assert_eq!(misses(&state), clear, "the miss counter walks up");
        }

        let fourth = grant(&config, &mut state, 4);
        assert_eq!(
            fourth
                .iter()
                .find(|unit| unit.currency_id == crystal_id)
                .map(|unit| unit.amount),
            Some(1),
            "the fourth consecutive miss pays exactly one Crystal"
        );
        assert_eq!(misses(&state), 0, "a Crystal resets the counter");
    }

    /// A natural win resets the same counter a pity win does, so the guarantee
    /// can never stack up behind a lucky streak.
    #[test]
    fn a_natural_crystal_resets_the_pity_counter() {
        let mut config = configs::tests_game_config::generate_game_config_for_tests();
        config.game_settings.mastery.crystal_chance = 1.0;
        let crystal_id = config.game_settings.mastery.crystal_currency_id;

        let mut state = OverlordState::default();
        state
            .character_state
            .character
            .custom_values
            .0
            .insert(MASTERY_CRYSTAL_MISSES.to_string(), 2);

        let granted = grant(&config, &mut state, 1);
        assert!(granted.iter().any(|unit| unit.currency_id == crystal_id));
        assert_eq!(misses(&state), 0);
    }

    /// The counter has to leave the tick as an event: a direct state write is
    /// dropped at reconnect, which would silently make the pity session-local.
    #[test]
    fn the_miss_counter_is_emitted_for_persistence() {
        use rand::SeedableRng;

        let mut config = configs::tests_game_config::generate_game_config_for_tests();
        config.game_settings.mastery.crystal_chance = 0.0;
        let mut state = OverlordState::default();
        let mut rng = rand::rngs::StdRng::seed_from_u64(1);

        let events = OverlordLogic::mastery_first_clear_rewards(&config, &mut state, 1, &mut rng);
        let written = events.iter().find_map(|event| match event.event() {
            OverlordEvent::SetCustomValue { key, value } if key == MASTERY_CRYSTAL_MISSES => {
                Some(*value)
            }
            _ => None,
        });

        assert_eq!(
            written,
            Some(misses(&state)),
            "the persisted counter must match the one the state carries"
        );
    }

    /// Every clear pays its band's Shards regardless of how the Crystal rolled —
    /// the two are one event, not two sources.
    #[test]
    fn shards_ride_the_same_clear_as_the_crystal_roll() {
        let config = configs::tests_game_config::generate_game_config_for_tests();
        let shard_id = config.game_settings.mastery.shard_currency_id;
        let mut state = OverlordState::default();

        let granted = grant(&config, &mut state, 1);
        assert_eq!(
            granted
                .iter()
                .find(|unit| unit.currency_id == shard_id)
                .map(|unit| unit.amount),
            Some(10),
            "the first band pays 10 Shards"
        );
    }

    #[test]
    fn each_band_pays_its_own_rate() {
        let mastery = settings(0.25, 4);
        // First and last clear of a band pay the same — the band is a step, not
        // a ramp.
        for (clear, expected) in [
            (1, 10),
            (10, 10),
            (11, 25),
            (20, 25),
            (50, 40),
            (51, 45),
            (100, 60),
        ] {
            assert_eq!(mastery.shards_for_clear(clear), expected, "clear {clear}");
        }
    }

    #[test]
    fn the_last_band_runs_on_past_its_own_edge() {
        let mastery = settings(0.25, 4);
        assert_eq!(mastery.shards_for_clear(300), 60);
        // A player deeper than the authored schedule keeps earning at the final
        // rate rather than dropping to nothing.
        assert_eq!(mastery.shards_for_clear(301), 60);
        assert_eq!(mastery.shards_for_clear(5_000), 60);
    }

    #[test]
    fn a_clear_before_the_first_one_pays_nothing() {
        let mastery = settings(0.25, 4);
        assert_eq!(mastery.shards_for_clear(0), 0);
        assert_eq!(mastery.shards_for_clear(-1), 0);
    }

    /// The three cumulative anchors the card names: the Shard schedule is what
    /// puts Class L3, L10 and L20 at chapters 10, 100 and 300.
    #[test]
    fn the_schedule_reproduces_the_cards_cumulative_anchors() {
        let mastery = settings(0.25, 4);
        let total_through = |last: i64| -> i64 {
            (1..=last)
                .map(|clear| mastery.shards_for_clear(clear))
                .sum()
        };
        assert_eq!(total_through(10), 100, "Class L3");
        assert_eq!(total_through(100), 4_000, "Class L10");
        assert_eq!(total_through(300), 16_000, "Class L20");
    }
}