lanspread/crates/lanspread-peer/src/call_to_play.rs

//! Replicated event history for Call to Play coordination.

use std::{
    collections::{BTreeSet, HashMap, HashSet},
    fmt,
    time::{SystemTime, UNIX_EPOCH},
};

use lanspread_proto::{CallToPlayAction, CallToPlayEvent};
use tokio::sync::mpsc::UnboundedSender;

use crate::{
    PeerEvent,
    context::Ctx,
    events,
    network::send_call_to_play_events,
    services::{HandshakeCtx, perform_handshake_with_peer},
};

const MAX_EVENTS: usize = 4_096;
const MAX_ID_CHARS: usize = 128;
const MAX_GAME_ID_CHARS: usize = 256;
const MAX_USERNAME_CHARS: usize = 24;
const MAX_MESSAGE_CHARS: usize = 500;
const EXPIRED_RETENTION_MS: i64 = 5 * 60_000;

#[derive(Debug, Default)]
pub(crate) struct CallToPlayStore {
    events: Vec<CallToPlayEvent>,
}

#[derive(Debug, PartialEq, Eq)]
pub(crate) struct BatchMerge {
    pub(crate) applied: Vec<CallToPlayEvent>,
    pub(crate) duplicates: usize,
    pub(crate) obsolete: usize,
    pub(crate) missing_call_ids: Vec<String>,
}

impl BatchMerge {
    pub(crate) fn needs_history(&self) -> bool {
        !self.missing_call_ids.is_empty()
    }
}

#[derive(Debug, PartialEq, Eq)]
pub(crate) enum MergeError {
    Invalid(&'static str),
    ConflictingEvent(String),
    HistoryFull,
}

impl fmt::Display for MergeError {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Invalid(reason) => formatter.write_str(reason),
            Self::ConflictingEvent(event_id) => {
                write!(formatter, "conflicting Call to Play event ID {event_id}")
            }
            Self::HistoryFull => formatter.write_str("Call to Play event history is full"),
        }
    }
}

impl CallToPlayStore {
    pub(crate) fn snapshot(&mut self) -> Vec<CallToPlayEvent> {
        self.snapshot_at(now_ms())
    }

    fn snapshot_at(&mut self, now: i64) -> Vec<CallToPlayEvent> {
        compact_history(&mut self.events, now);
        self.events.clone()
    }

    pub(crate) fn merge_batch(
        &mut self,
        incoming: Vec<CallToPlayEvent>,
    ) -> Result<BatchMerge, MergeError> {
        self.merge_batch_at(incoming, now_ms())
    }

    fn merge_batch_at(
        &mut self,
        incoming: Vec<CallToPlayEvent>,
        now: i64,
    ) -> Result<BatchMerge, MergeError> {
        for event in &incoming {
            validate_event(event).map_err(MergeError::Invalid)?;
        }

        let (incoming, mut duplicates) = deduplicate_batch(incoming)?;

        // Expiry is evaluated before accepting new actions. An action cannot keep
        // an already-retired root alive by itself; the sender must provide the
        // complete history in the same batch.
        let mut retained = self.events.clone();
        compact_history(&mut retained, now);

        let retained_by_id = retained
            .iter()
            .map(|event| (event.id.as_str(), event))
            .collect::<HashMap<_, _>>();
        let mut new_events = Vec::with_capacity(incoming.len());
        for event in incoming {
            if let Some(existing) = retained_by_id.get(event.id.as_str()) {
                if *existing != &event {
                    return Err(MergeError::ConflictingEvent(event.id));
                }
                duplicates += 1;
            } else {
                new_events.push(event);
            }
        }
        drop(retained_by_id);

        let rooted_calls = retained
            .iter()
            .chain(&new_events)
            .filter(|event| matches!(event.action, CallToPlayAction::Create { .. }))
            .map(|event| event.call_id.clone())
            .collect::<HashSet<_>>();
        let retained_tombstones = terminal_tombstone_call_ids(&retained);
        let mut applicable = Vec::with_capacity(new_events.len());
        let mut missing_call_ids = BTreeSet::new();
        let mut obsolete = 0;

        for event in new_events {
            if retained_tombstones.contains(event.call_id.as_str()) {
                obsolete += 1;
                continue;
            }
            if !matches!(event.action, CallToPlayAction::Create { .. })
                && !rooted_calls.contains(event.call_id.as_str())
            {
                missing_call_ids.insert(event.call_id);
                continue;
            }
            applicable.push(event);
        }

        let applicable_ids = applicable
            .iter()
            .map(|event| event.id.clone())
            .collect::<HashSet<_>>();
        retained.extend(applicable);
        compact_history(&mut retained, now);

        if unresolved_event_count(&retained) > MAX_EVENTS {
            return Err(MergeError::HistoryFull);
        }

        let retained_ids = retained
            .iter()
            .map(|event| event.id.clone())
            .collect::<HashSet<_>>();
        let applied = retained
            .iter()
            .filter(|event| applicable_ids.contains(&event.id))
            .cloned()
            .collect::<Vec<_>>();
        obsolete += applicable_ids
            .iter()
            .filter(|event_id| !retained_ids.contains(*event_id))
            .count();

        self.events = retained;
        Ok(BatchMerge {
            applied,
            duplicates,
            obsolete,
            missing_call_ids: missing_call_ids.into_iter().collect(),
        })
    }
}

fn deduplicate_batch(
    incoming: Vec<CallToPlayEvent>,
) -> Result<(Vec<CallToPlayEvent>, usize), MergeError> {
    let mut unique = Vec::<CallToPlayEvent>::with_capacity(incoming.len());
    let mut indexes = HashMap::<String, usize>::with_capacity(incoming.len());
    let mut duplicates = 0;

    for event in incoming {
        if let Some(index) = indexes.get(&event.id).copied() {
            if unique[index] != event {
                return Err(MergeError::ConflictingEvent(event.id));
            }
            duplicates += 1;
        } else {
            indexes.insert(event.id.clone(), unique.len());
            unique.push(event);
        }
    }

    Ok((unique, duplicates))
}

#[derive(Debug)]
struct HistoryIndex {
    creators: HashMap<String, CreateRecord>,
    terminal_events: HashMap<String, EventRecord>,
    extensions: HashMap<String, ExtensionRecord>,
}

#[derive(Clone, Debug)]
struct CreateRecord {
    at: i64,
    event_id: String,
    actor_id: String,
    deadline: i64,
}

impl CreateRecord {
    fn order_key(&self) -> (i64, &str) {
        (self.at, &self.event_id)
    }
}

#[derive(Clone, Debug)]
struct EventRecord {
    at: i64,
    event_id: String,
}

impl EventRecord {
    fn order_key(&self) -> (i64, &str) {
        (self.at, &self.event_id)
    }
}

#[derive(Clone, Debug)]
struct ExtensionRecord {
    event: EventRecord,
    deadline: i64,
}

impl HistoryIndex {
    fn build(events: &[CallToPlayEvent]) -> Self {
        let mut creators = HashMap::<String, CreateRecord>::new();
        for event in events {
            let CallToPlayAction::Create { deadline, .. } = event.action else {
                continue;
            };
            let candidate = CreateRecord {
                at: event.at,
                event_id: event.id.clone(),
                actor_id: event.actor_id.clone(),
                deadline,
            };
            creators
                .entry(event.call_id.clone())
                .and_modify(|current| {
                    if candidate.order_key() < current.order_key() {
                        current.clone_from(&candidate);
                    }
                })
                .or_insert(candidate);
        }

        let mut terminal_events = HashMap::<String, EventRecord>::new();
        let mut extensions = HashMap::<String, ExtensionRecord>::new();
        for event in events {
            let Some(creator) = creators.get(&event.call_id) else {
                continue;
            };
            if event.actor_id != creator.actor_id
                || (event.at, event.id.as_str()) <= creator.order_key()
            {
                continue;
            }

            if matches!(
                event.action,
                CallToPlayAction::Cancel | CallToPlayAction::Start
            ) {
                let candidate = EventRecord {
                    at: event.at,
                    event_id: event.id.clone(),
                };
                terminal_events
                    .entry(event.call_id.clone())
                    .and_modify(|current| {
                        if candidate.order_key() < current.order_key() {
                            current.clone_from(&candidate);
                        }
                    })
                    .or_insert(candidate);
            } else if let CallToPlayAction::AddTime { deadline } = event.action {
                let candidate = ExtensionRecord {
                    event: EventRecord {
                        at: event.at,
                        event_id: event.id.clone(),
                    },
                    deadline,
                };
                extensions
                    .entry(event.call_id.clone())
                    .and_modify(|current| {
                        if candidate.event.order_key() > current.event.order_key() {
                            current.clone_from(&candidate);
                        }
                    })
                    .or_insert(candidate);
            }
        }

        Self {
            creators,
            terminal_events,
            extensions,
        }
    }

    fn expired_call_ids(&self, now: i64) -> HashSet<&str> {
        self.creators
            .iter()
            .filter_map(|(call_id, creator)| {
                if self.terminal_events.contains_key(call_id) {
                    return None;
                }
                let deadline = self
                    .extensions
                    .get(call_id)
                    .map_or(creator.deadline, |extension| extension.deadline);
                (now - deadline > EXPIRED_RETENTION_MS).then_some(call_id.as_str())
            })
            .collect()
    }
}

fn compact_history(events: &mut Vec<CallToPlayEvent>, now: i64) {
    let index = HistoryIndex::build(events);
    let expired_calls = index.expired_call_ids(now);

    events.retain(|event| {
        if expired_calls.contains(event.call_id.as_str()) {
            return false;
        }
        index
            .terminal_events
            .get(&event.call_id)
            .is_none_or(|terminal| event.id == terminal.event_id)
    });
}

fn terminal_tombstone_call_ids(events: &[CallToPlayEvent]) -> HashSet<String> {
    let rooted_calls = events
        .iter()
        .filter(|event| matches!(event.action, CallToPlayAction::Create { .. }))
        .map(|event| event.call_id.clone())
        .collect::<HashSet<_>>();

    events
        .iter()
        .filter(|event| {
            matches!(
                event.action,
                CallToPlayAction::Cancel | CallToPlayAction::Start
            ) && !rooted_calls.contains(event.call_id.as_str())
        })
        .map(|event| event.call_id.clone())
        .collect()
}

fn unresolved_event_count(events: &[CallToPlayEvent]) -> usize {
    let index = HistoryIndex::build(events);
    events
        .iter()
        .filter(|event| {
            index.creators.contains_key(&event.call_id)
                && !index.terminal_events.contains_key(&event.call_id)
        })
        .count()
}

fn now_ms() -> i64 {
    i64::try_from(
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis(),
    )
    .unwrap_or(i64::MAX)
}

pub(crate) async fn publish(
    ctx: &Ctx,
    tx_notify_ui: &UnboundedSender<PeerEvent>,
    mut event: CallToPlayEvent,
) -> Result<(), String> {
    event.actor_id.clone_from(ctx.peer_id.as_ref());
    let merged = ctx
        .call_to_play
        .write()
        .await
        .merge_batch(vec![event.clone()])
        .map_err(|err| err.to_string())?;
    if merged.needs_history() {
        return Err("Call to Play history is missing".to_string());
    }
    if merged.applied.is_empty() {
        if merged.obsolete > 0 {
            return Err("Call to Play event is obsolete".to_string());
        }
        return Ok(());
    }

    events::send(tx_notify_ui, PeerEvent::CallToPlayEvents(merged.applied));

    let peer_addresses = ctx.peer_game_db.read().await.get_peer_addresses();
    let peer_id = ctx.peer_id.clone();
    let handshake_ctx = HandshakeCtx::from_ctx(ctx, tx_notify_ui);
    ctx.task_tracker.spawn(async move {
        let deliveries = peer_addresses.into_iter().map(|peer_addr| {
            let event = event.clone();
            let peer_id = peer_id.clone();
            let handshake_ctx = handshake_ctx.clone();
            async move {
                if let Err(err) =
                    send_call_to_play_events(peer_addr, peer_id.as_ref(), vec![event]).await
                {
                    log::warn!("Failed to send Call to Play event to {peer_addr}: {err}");
                    if let Err(resync_err) =
                        perform_handshake_with_peer(handshake_ctx, peer_addr, None).await
                    {
                        log::warn!(
                            "Failed to resync Call to Play history with {peer_addr}: {resync_err}"
                        );
                    }
                }
            }
        });
        futures::future::join_all(deliveries).await;
    });
    Ok(())
}

fn validate_event(event: &CallToPlayEvent) -> Result<(), &'static str> {
    validate_nonempty(&event.id, MAX_ID_CHARS, "invalid event id")?;
    validate_nonempty(&event.call_id, MAX_ID_CHARS, "invalid call id")?;
    validate_nonempty(&event.actor_id, MAX_ID_CHARS, "invalid actor id")?;
    validate_nonempty(&event.actor_name, MAX_USERNAME_CHARS, "invalid actor name")?;
    if event.at <= 0 {
        return Err("invalid event timestamp");
    }

    match &event.action {
        CallToPlayAction::Create {
            game_id,
            max_players,
            scheduled_for,
            deadline,
        } => {
            validate_nonempty(game_id, MAX_GAME_ID_CHARS, "invalid game id")?;
            if !(2..=64).contains(max_players) {
                return Err("max players must be between 2 and 64");
            }
            if *deadline <= event.at {
                return Err("deadline must be after creation");
            }
            if scheduled_for.is_some_and(|scheduled| scheduled != *deadline) {
                return Err("scheduled call deadline must match its start time");
            }
        }
        CallToPlayAction::Respond { ready_at } => {
            if ready_at.is_some_and(|ready| ready < event.at) {
                return Err("ready time cannot be before the response");
            }
        }
        CallToPlayAction::SendMessage { message_id, text } => {
            validate_nonempty(message_id, MAX_ID_CHARS, "invalid message id")?;
            validate_nonempty(text, MAX_MESSAGE_CHARS, "invalid message")?;
        }
        CallToPlayAction::AddTime { deadline } => {
            if *deadline <= event.at {
                return Err("extended deadline must be in the future");
            }
        }
        CallToPlayAction::Rsvp
        | CallToPlayAction::Leave
        | CallToPlayAction::Cancel
        | CallToPlayAction::Start => {}
    }

    Ok(())
}

fn validate_nonempty(
    value: &str,
    max_chars: usize,
    error: &'static str,
) -> Result<(), &'static str> {
    if value.trim().is_empty() || value.chars().count() > max_chars {
        return Err(error);
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use lanspread_proto::{CallToPlayAction, CallToPlayEvent};

    use super::{CallToPlayStore, MAX_EVENTS, MergeError};

    const TEST_NOW: i64 = 8_000_000_000_000;

    fn create_event(id: &str) -> CallToPlayEvent {
        create_event_for("call-1", id)
    }

    fn create_event_for(call_id: &str, id: &str) -> CallToPlayEvent {
        CallToPlayEvent {
            id: id.to_string(),
            call_id: call_id.to_string(),
            actor_id: "peer-alice".to_string(),
            actor_name: "Alice".to_string(),
            at: TEST_NOW,
            action: CallToPlayAction::Create {
                game_id: "game-1".to_string(),
                max_players: 4,
                scheduled_for: None,
                deadline: TEST_NOW + 60_000,
            },
        }
    }

    fn action_event(id: &str, call_id: &str, action: CallToPlayAction) -> CallToPlayEvent {
        CallToPlayEvent {
            id: id.to_string(),
            call_id: call_id.to_string(),
            actor_id: "peer-alice".to_string(),
            actor_name: "Alice".to_string(),
            at: TEST_NOW + 1_000,
            action,
        }
    }

    #[test]
    fn deduplicates_events_without_reordering_new_history() {
        let mut store = CallToPlayStore::default();
        let first = create_event("event-1");
        let second = create_event("event-2");
        let merged = store
            .merge_batch_at(vec![first.clone(), first.clone(), second.clone()], TEST_NOW)
            .expect("valid batch should merge");

        assert_eq!(merged.applied, [first, second]);
        assert_eq!(merged.duplicates, 1);
        assert_eq!(merged.obsolete, 0);
        assert!(!merged.needs_history());

        let duplicate = store
            .merge_batch_at(vec![create_event("event-1")], TEST_NOW)
            .expect("stored duplicate should be harmless");
        assert!(duplicate.applied.is_empty());
        assert_eq!(duplicate.duplicates, 1);

        let ids = event_ids(store.snapshot_at(TEST_NOW));
        assert_eq!(ids, ["event-1", "event-2"]);
    }

    #[test]
    fn invalid_batch_leaves_store_unchanged() {
        let mut store = CallToPlayStore::default();
        store
            .merge_batch_at(vec![create_event("create")], TEST_NOW)
            .expect("create should fit");
        let before = store.snapshot_at(TEST_NOW);

        let mut invalid = action_event(
            "invalid",
            "call-1",
            CallToPlayAction::SendMessage {
                message_id: "message-1".to_string(),
                text: "valid before corruption".to_string(),
            },
        );
        invalid.action = CallToPlayAction::SendMessage {
            message_id: "message-1".to_string(),
            text: "   ".to_string(),
        };

        assert_eq!(
            store.merge_batch_at(
                vec![
                    action_event("rsvp", "call-1", CallToPlayAction::Rsvp),
                    invalid,
                ],
                TEST_NOW,
            ),
            Err(MergeError::Invalid("invalid message"))
        );
        assert_eq!(store.snapshot_at(TEST_NOW), before);
    }

    #[test]
    fn conflicting_event_id_leaves_store_unchanged() {
        let mut store = CallToPlayStore::default();
        store
            .merge_batch_at(vec![create_event("create")], TEST_NOW)
            .expect("create should fit");
        let before = store.snapshot_at(TEST_NOW);
        let mut conflicting = create_event("create");
        conflicting.actor_name = "Mallory".to_string();

        assert_eq!(
            store.merge_batch_at(vec![conflicting], TEST_NOW),
            Err(MergeError::ConflictingEvent("create".to_string()))
        );
        assert_eq!(store.snapshot_at(TEST_NOW), before);
    }

    #[test]
    fn create_and_extension_revive_expired_history_in_any_batch_order() {
        let extended_deadline = TEST_NOW + 20 * 60_000;
        let extension = action_event(
            "extend",
            "call-1",
            CallToPlayAction::AddTime {
                deadline: extended_deadline,
            },
        );
        let after_recovery_window = TEST_NOW + 6 * 60_000 + 1;

        for history in [
            vec![create_event("create"), extension.clone()],
            vec![extension.clone(), create_event("create")],
        ] {
            let mut store = CallToPlayStore::default();
            store
                .merge_batch_at(vec![create_event("create")], TEST_NOW)
                .expect("initial create should fit");

            let missing = store
                .merge_batch_at(vec![extension.clone()], after_recovery_window)
                .expect("missing history is an acknowledged merge outcome");
            assert!(missing.applied.is_empty());
            assert_eq!(missing.missing_call_ids, ["call-1"]);
            assert!(store.snapshot_at(after_recovery_window).is_empty());

            let revived = store
                .merge_batch_at(history, after_recovery_window)
                .expect("complete history should revive the call atomically");
            assert_eq!(revived.applied.len(), 2);
            assert_eq!(
                event_ids(store.snapshot_at(after_recovery_window)),
                event_ids(revived.applied)
            );
        }
    }

    #[test]
    fn immediately_obsolete_event_is_not_applied() {
        let mut store = CallToPlayStore::default();
        let after_recovery_window = TEST_NOW + 6 * 60_000 + 1;
        let merged = store
            .merge_batch_at(vec![create_event("create")], after_recovery_window)
            .expect("obsolete is an acknowledged merge outcome");

        assert!(merged.applied.is_empty());
        assert_eq!(merged.obsolete, 1);
        assert!(store.snapshot_at(after_recovery_window).is_empty());
    }

    #[test]
    fn terminal_tombstone_prevents_stale_history_resurrection() {
        let mut store = CallToPlayStore::default();
        let start = action_event("start", "call-1", CallToPlayAction::Start);
        store
            .merge_batch_at(vec![create_event("create"), start.clone()], TEST_NOW)
            .expect("terminal history should merge");

        let stale = store
            .merge_batch_at(
                vec![
                    create_event("create"),
                    action_event(
                        "extend",
                        "call-1",
                        CallToPlayAction::AddTime {
                            deadline: TEST_NOW + 20 * 60_000,
                        },
                    ),
                ],
                TEST_NOW,
            )
            .expect("stale history is an obsolete merge outcome");

        assert!(stale.applied.is_empty());
        assert_eq!(stale.obsolete, 2);
        assert_eq!(store.snapshot_at(TEST_NOW), [start]);
    }

    #[test]
    fn terminal_actions_succeed_at_the_active_history_cap() {
        for terminal_action in [CallToPlayAction::Start, CallToPlayAction::Cancel] {
            let mut store = full_active_store();
            let terminal = action_event("terminal", "call-1", terminal_action);

            let merged = store
                .merge_batch_at(vec![terminal.clone()], TEST_NOW)
                .expect("terminal action should compact the full call");
            assert_eq!(merged.applied.as_slice(), std::slice::from_ref(&terminal));
            assert_eq!(store.snapshot_at(TEST_NOW), [terminal]);
        }
    }

    #[test]
    fn terminal_tombstones_do_not_consume_active_history_capacity() {
        let mut store = full_active_store();
        store
            .merge_batch_at(
                vec![action_event(
                    "call-1-start",
                    "call-1",
                    CallToPlayAction::Start,
                )],
                TEST_NOW,
            )
            .expect("start should compact active history");

        let created = create_event_for("call-2", "call-2-create");
        let merged = store
            .merge_batch_at(vec![created.clone()], TEST_NOW)
            .expect("terminal tombstone must not block a new active call");
        assert_eq!(merged.applied, [created]);
        assert_eq!(store.snapshot_at(TEST_NOW).len(), 2);
    }

    #[test]
    fn full_active_history_returns_an_error() {
        let mut store = full_active_store();
        let before = store.snapshot_at(TEST_NOW);

        assert_eq!(
            store.merge_batch_at(
                vec![action_event("overflow", "call-1", CallToPlayAction::Rsvp,)],
                TEST_NOW,
            ),
            Err(MergeError::HistoryFull)
        );
        assert_eq!(store.snapshot_at(TEST_NOW), before);
    }

    #[test]
    fn visible_call_retains_complete_chat_history() {
        let mut store = CallToPlayStore::default();
        let message = action_event(
            "message-event",
            "call-1",
            CallToPlayAction::SendMessage {
                message_id: "message-1".to_string(),
                text: "Ready when you are".to_string(),
            },
        );
        let merged = store
            .merge_batch_at(
                vec![
                    create_event("create"),
                    action_event("rsvp", "call-1", CallToPlayAction::Rsvp),
                    message.clone(),
                ],
                TEST_NOW,
            )
            .expect("visible history should merge");

        assert_eq!(merged.applied.len(), 3);
        assert!(store.snapshot_at(TEST_NOW).contains(&message));
    }

    fn full_active_store() -> CallToPlayStore {
        let mut history = Vec::with_capacity(MAX_EVENTS);
        history.push(create_event("create"));
        history.extend((1..MAX_EVENTS).map(|index| {
            action_event(&format!("event-{index}"), "call-1", CallToPlayAction::Rsvp)
        }));

        let mut store = CallToPlayStore::default();
        store
            .merge_batch_at(history, TEST_NOW)
            .expect("active history should fit through the cap");
        store
    }

    fn event_ids(events: Vec<CallToPlayEvent>) -> Vec<String> {
        events.into_iter().map(|event| event.id).collect()
    }
}