manabrew_server/

cleanup.rs

use std::sync::Arc;
use std::time::{Duration, Instant};

use tracing::{info, warn};

use crate::analytics::{self, GameEndReason};
use crate::connection::{broadcast_room_transport, broadcast_to_room, emit_to};
use crate::lobby;
use crate::protocol::{RoomStatus, ServerMessage};
use crate::room::Room;
use crate::state::ServerState;

const CLEANUP_INTERVAL: Duration = Duration::from_secs(60);
const STALE_CONNECTED_TIMEOUT: Duration = Duration::from_secs(180);
const IN_GAME_DISCONNECTED_GRACE: Duration = Duration::from_secs(3600);
const RECONNECT_ABORT_MARGIN: Duration = Duration::from_secs(5);

pub async fn cleanup_loop(state: Arc<ServerState>) {
    let mut ticker = tokio::time::interval(CLEANUP_INTERVAL);
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    loop {
        ticker.tick().await;
        cleanup_stale_state(&state);
    }
}

fn cleanup_stale_state(state: &Arc<ServerState>) {
    let now = Instant::now();
    let stale_players = state
        .players
        .iter()
        .filter_map(|entry| {
            let player = entry.value();
            (player.connected && now.duration_since(player.last_seen) >= STALE_CONNECTED_TIMEOUT)
                .then(|| {
                    (
                        entry.key().clone(),
                        player.username.clone(),
                        player.generation,
                    )
                })
        })
        .collect::<Vec<_>>();

    for (player_id, username, generation) in stale_players {
        warn!(
            "[cleanup] '{}' had no websocket frames for {}s -- marking disconnected",
            username,
            STALE_CONNECTED_TIMEOUT.as_secs()
        );
        mark_disconnected(state, &player_id, generation);
    }

    let mut humanless_rooms = Vec::new();
    for mut entry in state.rooms.iter_mut() {
        let room = entry.value_mut();
        if room.status != RoomStatus::InGame || room.has_connected_human() {
            room.humanless_since = None;
            continue;
        }
        let since = *room.humanless_since.get_or_insert(now);
        let grace = Duration::from_secs(room.reconnect_timeout_s as u64) + RECONNECT_ABORT_MARGIN;
        if now.duration_since(since) >= grace {
            humanless_rooms.push(entry.key().clone());
        }
    }
    for room_id in humanless_rooms {
        abort_humanless_room(state, &room_id);
    }

    let rooms_to_remove = state
        .rooms
        .iter()
        .filter_map(|entry| room_is_stale(state, entry.value(), now).then(|| entry.key().clone()))
        .collect::<Vec<_>>();

    for room_id in rooms_to_remove {
        info!(
            "[cleanup] removing stale room {}",
            &room_id[..8.min(room_id.len())]
        );
        remove_room_and_clear_sessions(state, &room_id, GameEndReason::StaleExpired);
    }
}

fn room_is_stale(state: &Arc<ServerState>, room: &Room, now: Instant) -> bool {
    if room.parent_limited_room.is_some() {
        return !paired_room_has_participants(state, room, now);
    }
    match room.status {
        RoomStatus::Lobby => room.all_disconnected(),
        RoomStatus::InGame => in_game_room_expired(state, room, now),
    }
}

pub(crate) fn paired_room_has_participants(
    state: &Arc<ServerState>,
    room: &Room,
    now: Instant,
) -> bool {
    let grace = Duration::from_secs(room.reconnect_timeout_s as u64) + RECONNECT_ABORT_MARGIN;
    room.players
        .iter()
        .filter(|seat| !seat.is_bot)
        .map(|seat| seat.player_id.as_str())
        .chain(
            room.observers
                .iter()
                .map(|observer| observer.player_id.as_str()),
        )
        .any(|player_id| {
            state.players.get(player_id).is_some_and(|player| {
                player.room_id.as_deref() == Some(room.room_id.as_str())
                    && (player.connected
                        || player
                            .disconnected_at
                            .is_some_and(|since| now.duration_since(since) < grace))
            })
        })
}

fn abort_humanless_room(state: &Arc<ServerState>, room_id: &str) {
    let Some((info, notify)) = lobby::reset_room_to_lobby(state, room_id, GameEndReason::Abandoned)
    else {
        return;
    };
    info!(
        "[cleanup] in-game room {} had no connected human players -- reset to lobby",
        &room_id[..8.min(room_id.len())]
    );
    broadcast_to_room(state, room_id, &ServerMessage::RoomUpdate { room: info });
    let aborted = ServerMessage::GameAborted {
        room_id: room_id.to_string(),
    };
    if let Ok(json) = serde_json::to_string(&aborted) {
        for pid in &notify {
            emit_to(state, pid, &aborted, &json);
        }
    }
}

pub fn schedule_host_resume_abort(
    state: Arc<ServerState>,
    room_id: String,
    host_player_id: String,
) {
    let Some(timeout_s) = state
        .rooms
        .get(&room_id)
        .map(|room| room.reconnect_timeout_s)
    else {
        return;
    };
    let timeout = Duration::from_secs(timeout_s as u64);

    tokio::spawn(async move {
        tokio::time::sleep(timeout + RECONNECT_ABORT_MARGIN).await;

        // ResumeRoom rotates host_player_id and session reclaim flips the
        // player back to connected; either one disarms this abort.
        let host_still_gone = state
            .rooms
            .get(&room_id)
            .map(|room| room.status == RoomStatus::InGame && room.host_player_id == host_player_id)
            .unwrap_or(false)
            && state
                .players
                .get(&host_player_id)
                .map(|player| !player.connected)
                .unwrap_or(true);
        if !host_still_gone {
            return;
        }

        info!(
            "[cleanup] hosted in-game room {} host never resumed -- removing",
            &room_id[..8.min(room_id.len())]
        );
        broadcast_to_room(
            &state,
            &room_id,
            &ServerMessage::GameAborted {
                room_id: room_id.clone(),
            },
        );
        remove_room_and_clear_sessions(&state, &room_id, GameEndReason::HostLost);
    });
}

pub fn schedule_peer_host_loss(state: Arc<ServerState>, room_id: String, player_id: String) {
    let Some(timeout_s) = state
        .rooms
        .get(&room_id)
        .filter(|room| room.parent_limited_room.is_some() || !room.is_limited_session())
        .map(|room| room.reconnect_timeout_s)
    else {
        return;
    };
    let timeout = Duration::from_secs(timeout_s as u64);
    let disconnected_at = state
        .players
        .get(&player_id)
        .and_then(|p| p.disconnected_at);

    tokio::spawn(async move {
        tokio::time::sleep(timeout + RECONNECT_ABORT_MARGIN).await;

        let still_hosting = state
            .rooms
            .get(&room_id)
            .map(|room| room.status == RoomStatus::InGame && room.host_player_id == player_id)
            .unwrap_or(false);
        let still_gone = state
            .players
            .get(&player_id)
            .map(|player| !player.connected && player.disconnected_at == disconnected_at)
            .unwrap_or(true);
        if !still_hosting || !still_gone {
            return;
        }

        let Some((info, notify)) =
            lobby::reset_room_to_lobby(&state, &room_id, GameEndReason::HostLost)
        else {
            return;
        };
        info!(
            "[cleanup] room {} lost the seat holding its engine -- game ended, room reset to lobby",
            &room_id[..8.min(room_id.len())]
        );
        broadcast_to_room(&state, &room_id, &ServerMessage::RoomUpdate { room: info });
        let aborted = ServerMessage::GameAborted {
            room_id: room_id.clone(),
        };
        if let Ok(json) = serde_json::to_string(&aborted) {
            for pid in &notify {
                emit_to(&state, pid, &aborted, &json);
            }
        }
    });
}

pub fn schedule_seat_forfeit(state: Arc<ServerState>, room_id: String, player_id: String) {
    let Some(timeout_s) = state
        .rooms
        .get(&room_id)
        .filter(|room| room.parent_limited_room.is_some() || !room.is_limited_session())
        .map(|room| room.reconnect_timeout_s)
    else {
        return;
    };
    let timeout = Duration::from_secs(timeout_s as u64);
    let disconnected_at = state
        .players
        .get(&player_id)
        .and_then(|p| p.disconnected_at);

    tokio::spawn(async move {
        tokio::time::sleep(timeout + RECONNECT_ABORT_MARGIN).await;

        // A reconnect flips the player back to connected; a rejoin re-keys the
        // seat's player_id (disarming via the seat check below). No player
        // entry at all means a pending-rejoin seat resurrected after a relay
        // restart — nobody is connected under that id, so it counts as gone.
        let still_gone = state
            .players
            .get(&player_id)
            .map(|player| {
                !player.connected
                    && player.room_id.as_deref() == Some(room_id.as_str())
                    && player.disconnected_at == disconnected_at
            })
            .unwrap_or(disconnected_at.is_none());
        let retained_seat = state
            .rooms
            .get(&room_id)
            .map(|room| {
                (room.status == RoomStatus::InGame || room.parent_limited_room.is_some())
                    && room.players.iter().any(|slot| slot.player_id == player_id)
            })
            .unwrap_or(false);
        if !still_gone || !retained_seat {
            return;
        }
        forfeit_seat(&state, &room_id, &player_id);
    });
}

fn forfeit_seat(state: &Arc<ServerState>, room_id: &str, player_id: &str) {
    let (username, info, parent_id) = {
        let Some(mut room) = state.rooms.get_mut(room_id) else {
            return;
        };
        let Some(slot) = room.remove_player(player_id) else {
            return;
        };
        (
            slot.username,
            room.to_room_info(),
            room.parent_limited_room.clone(),
        )
    };
    release_room_session(state, room_id, player_id, parent_id.as_deref());
    let abandoned = state.rooms.get(room_id).is_some_and(|room| {
        room.parent_limited_room.is_some()
            && !paired_room_has_participants(state, &room, Instant::now())
    });
    if abandoned {
        remove_room_and_clear_sessions(state, room_id, GameEndReason::Abandoned);
        return;
    }
    info!(
        "[cleanup] '{}' did not reconnect within grace -- forfeiting seat in room {}",
        username,
        &room_id[..8.min(room_id.len())]
    );
    broadcast_to_room(
        state,
        room_id,
        &ServerMessage::PlayerLeft {
            room_id: room_id.to_string(),
            username,
        },
    );
    broadcast_to_room(state, room_id, &ServerMessage::RoomUpdate { room: info });
}

fn in_game_room_expired(state: &Arc<ServerState>, room: &Room, now: Instant) -> bool {
    if !room.all_disconnected() {
        return false;
    }

    room.players
        .iter()
        .map(|slot| slot.player_id.as_str())
        .chain(
            room.observers
                .iter()
                .map(|observer| observer.player_id.as_str()),
        )
        .filter_map(|player_id| state.players.get(player_id).and_then(|p| p.disconnected_at))
        .max()
        .is_some_and(|latest| now.duration_since(latest) >= IN_GAME_DISCONNECTED_GRACE)
}

pub fn mark_disconnected(state: &Arc<ServerState>, player_id: &str, our_generation: u64) {
    mark_disconnected_inner(state, player_id, our_generation);
    crate::connection::broadcast_player_list(state);
}

fn mark_disconnected_inner(state: &Arc<ServerState>, player_id: &str, our_generation: u64) {
    let (username, room_id) = {
        if let Some(mut player) = state.players.get_mut(player_id) {
            if player.generation != our_generation {
                info!(
                    "[disconnect] '{}' old connection cleaned up (session reclaimed by new connection)",
                    player.username
                );
                return;
            }
            player.connected = false;
            player.disconnected_at = Some(Instant::now());
            (player.username.clone(), player.room_id.clone())
        } else {
            return;
        }
    };
    let parent_id = room_id.as_ref().and_then(|id| {
        crate::draft_clock::pause_disconnected_host(state, id, player_id);
        state
            .rooms
            .get(id)
            .and_then(|room| room.parent_limited_room.clone())
    });
    if let Some(parent_id) = parent_id {
        let lost_host = state
            .rooms
            .get_mut(&parent_id)
            .map(|mut room| {
                room.set_connected(player_id, false);
                room.is_host(player_id)
            })
            .unwrap_or(false);
        let info = state.rooms.get(&parent_id).map(|room| room.to_room_info());
        if let Some(room) = info {
            broadcast_to_room(state, &parent_id, &ServerMessage::RoomUpdate { room });
        }
        if lost_host {
            schedule_peer_host_loss(state.clone(), parent_id, player_id.to_string());
        }
    }

    if let Some(rid) = &room_id {
        let room_status = state.rooms.get(rid).map(|r| r.status.clone());

        match room_status {
            Some(RoomStatus::InGame) => {
                let host_without_player = state
                    .rooms
                    .get(rid)
                    .map(|room| room.is_host(player_id) && !room.host_is_player())
                    .unwrap_or(false);
                if host_without_player {
                    // Give the host the same reconnect grace as a player: it can
                    // come back via session reclaim or ResumeRoom. Killing the
                    // room here would abort every guest's game on a host blip.
                    if let Some(mut room) = state.rooms.get_mut(rid) {
                        room.set_connected(player_id, false);
                    } else {
                        return;
                    }
                    info!(
                        "[disconnect] hosted in-game room {} lost its non-playing host -- awaiting resume",
                        &rid[..8]
                    );
                    broadcast_to_room(
                        state,
                        rid,
                        &ServerMessage::PlayerDisconnected {
                            username: username.clone(),
                        },
                    );
                    schedule_host_resume_abort(state.clone(), rid.clone(), player_id.to_string());
                    return;
                }

                let peer_host_lost = state
                    .rooms
                    .get(rid)
                    .map(|room| !room.hosted && room.is_host(player_id) && room.host_is_player())
                    .unwrap_or(false);

                let paired_limited = state
                    .rooms
                    .get(rid)
                    .is_some_and(|room| room.parent_limited_room.is_some());
                let holds_seat = {
                    if let Some(mut room) = state.rooms.get_mut(rid) {
                        room.set_connected(player_id, false);
                        room.players.iter().any(|slot| slot.player_id == player_id)
                    } else {
                        return;
                    }
                };

                if peer_host_lost {
                    info!(
                        "[disconnect] in-game room {} lost the seat holding its engine -- awaiting resume",
                        &rid[..8]
                    );
                    broadcast_to_room(
                        state,
                        rid,
                        &ServerMessage::PlayerDisconnected {
                            username: username.clone(),
                        },
                    );
                    schedule_peer_host_loss(state.clone(), rid.clone(), player_id.to_string());
                    if paired_limited {
                        schedule_seat_forfeit(state.clone(), rid.clone(), player_id.to_string());
                    }
                    return;
                }

                info!(
                    "[disconnect] '{}' marked disconnected in in-game room {} (session preserved)",
                    username,
                    &rid[..8]
                );
                if holds_seat {
                    schedule_seat_forfeit(state.clone(), rid.clone(), player_id.to_string());
                }
                broadcast_to_room(
                    state,
                    rid,
                    &ServerMessage::PlayerDisconnected {
                        username: username.clone(),
                    },
                );

                let info = state.rooms.get(rid).map(|room| room.to_room_info());
                if let Some(room) = info {
                    broadcast_to_room(state, rid, &ServerMessage::RoomUpdate { room });
                }
            }
            Some(RoomStatus::Lobby) => {
                let paired_limited = state
                    .rooms
                    .get(rid)
                    .is_some_and(|room| room.parent_limited_room.is_some());
                if paired_limited {
                    if let Some(mut room) = state.rooms.get_mut(rid) {
                        room.set_connected(player_id, false);
                    }
                    schedule_seat_forfeit(state.clone(), rid.clone(), player_id.to_string());
                    broadcast_to_room(
                        state,
                        rid,
                        &ServerMessage::PlayerDisconnected {
                            username: username.clone(),
                        },
                    );
                    let info = state.rooms.get(rid).map(|room| room.to_room_info());
                    if let Some(info) = info {
                        broadcast_to_room(state, rid, &ServerMessage::RoomUpdate { room: info });
                    }
                    return;
                }
                info!(
                    "[disconnect] '{}' disconnected from lobby room {} -- treating as leave",
                    username,
                    &rid[..8]
                );

                let remove_hosted_room = state
                    .rooms
                    .get(rid)
                    .map(|room| room.is_host(player_id) && !room.host_is_player())
                    .unwrap_or(false);
                if remove_hosted_room {
                    info!(
                        "[cleanup] hosted lobby room {} lost its non-playing host -- removing",
                        &rid[..8]
                    );
                    remove_room_and_clear_sessions(state, rid, GameEndReason::HostLost);
                    return;
                }

                let room_empty = {
                    if let Some(mut room) = state.rooms.get_mut(rid) {
                        room.remove_participant(player_id);
                        let empty = room.is_empty();
                        if !empty && room.players.is_empty() {
                            room.reset_lobby_settings();
                        }
                        empty
                    } else {
                        false
                    }
                };

                if let Some(mut player) = state.players.get_mut(player_id) {
                    player.room_id = None;
                }

                if room_empty {
                    info!(
                        "[cleanup] lobby room {} is now empty -- removing",
                        &rid[..8]
                    );
                    remove_room_and_clear_sessions(state, rid, GameEndReason::Abandoned);
                } else {
                    broadcast_to_room(
                        state,
                        rid,
                        &ServerMessage::PlayerLeft {
                            room_id: rid.clone(),
                            username: username.clone(),
                        },
                    );
                    if let Some(room) = state.rooms.get(rid) {
                        broadcast_to_room(
                            state,
                            rid,
                            &ServerMessage::RoomUpdate {
                                room: room.to_room_info(),
                            },
                        );
                    }
                    broadcast_room_transport(state, rid);
                }

                info!("[cleanup] '{}' removed (disconnected from lobby)", username);
                state.players.remove(player_id);
            }
            None => {
                info!("[cleanup] '{}' removed (room no longer exists)", username);
                state.players.remove(player_id);
            }
        }
    } else {
        info!("[cleanup] '{}' removed (was not in a room)", username);
        state.players.remove(player_id);
    }
}

pub fn remove_room_and_clear_sessions(
    state: &Arc<ServerState>,
    room_id: &str,
    reason: GameEndReason,
) {
    let Some((_, room)) = state.rooms.remove_if(room_id, |_, room| {
        reason != GameEndReason::StaleExpired || room_is_stale(state, room, Instant::now())
    }) else {
        return;
    };
    clear_removed_room(state, room, reason);
}

fn clear_removed_room(state: &Arc<ServerState>, room: Room, reason: GameEndReason) {
    if let Some(replay) = room.replay.as_ref() {
        analytics::emit_game_ended(&state.analytics, &room, replay, reason);
    }
    if let Some(parent_id) = &room.parent_limited_room {
        if let Some(mut parent) = state.rooms.get_mut(parent_id) {
            parent.limited_matches.retain(|id| id != &room.room_id);
        }
    }
    for child_id in &room.limited_matches {
        let active = state
            .rooms
            .get(child_id)
            .is_some_and(|child| child.status == RoomStatus::InGame);
        if active {
            broadcast_to_room(
                state,
                child_id,
                &ServerMessage::GameAborted {
                    room_id: child_id.clone(),
                },
            );
        }
        if let Some((_, child)) = state.rooms.remove(child_id) {
            clear_removed_room(state, child, reason);
        }
    }
    release_room_sessions(state, &room.room_id, room.parent_limited_room.as_deref());
}

fn release_room_sessions(state: &Arc<ServerState>, room_id: &str, parent_id: Option<&str>) {
    let player_ids = state
        .players
        .iter()
        .filter_map(|entry| {
            entry
                .value()
                .room_id
                .as_deref()
                .is_some_and(|rid| rid == room_id)
                .then(|| entry.key().clone())
        })
        .collect::<Vec<_>>();
    for player_id in player_ids {
        release_room_session(state, room_id, &player_id, parent_id);
    }
}

fn release_room_session(
    state: &Arc<ServerState>,
    room_id: &str,
    player_id: &str,
    parent_id: Option<&str>,
) {
    let parent = parent_id.and_then(|id| state.rooms.get(id)).filter(|room| {
        room.is_limited_session() && room.players.iter().any(|seat| seat.player_id == player_id)
    });
    state.players.remove_if_mut(player_id, |_, player| {
        if player.room_id.as_deref() != Some(room_id) {
            return false;
        }
        if let Some(parent) = &parent {
            player.room_id = Some(parent.room_id.clone());
            false
        } else if player.connected {
            player.room_id = None;
            false
        } else {
            true
        }
    });
}