manabrew_server/analytics/
mod.rs

1mod capture;
2mod event;
3mod sink;
4
5use std::collections::HashMap;
6use std::path::PathBuf;
7
8use serde_json::Value;
9use tokio::sync::mpsc;
10
11pub use event::{AnalyticsEvent, CardEntry, GameEndReason, SeatInfo};
12
13use crate::config::ServerConfig;
14use crate::lobby::StartedGame;
15use crate::metrics;
16use crate::replay::GameReplayCache;
17use crate::room::Room;
18use manabrew_protocol::deck_dto::Deck;
19
20const CHANNEL_CAPACITY: usize = 8192;
21
22#[derive(Clone)]
23pub struct AnalyticsHandle {
24    events: Option<mpsc::Sender<AnalyticsEvent>>,
25    capture: Option<std::sync::mpsc::SyncSender<capture::CaptureMessage>>,
26}
27
28impl AnalyticsHandle {
29    pub fn disabled() -> Self {
30        AnalyticsHandle {
31            events: None,
32            capture: None,
33        }
34    }
35
36    pub fn from_config(config: &ServerConfig) -> Self {
37        let hub = config
38            .hub_url
39            .clone()
40            .zip(config.hub_token.clone())
41            .map(|(url, token)| sink::HubTarget { url, token });
42        let spool = config.events_dir.clone().map(PathBuf::from);
43        let events = (hub.is_some() || spool.is_some()).then(|| {
44            let (tx, rx) = mpsc::channel(CHANNEL_CAPACITY);
45            sink::spawn(rx, spool, hub);
46            tx
47        });
48        let capture = config.capture_dir.clone().map(|dir| {
49            let (tx, rx) = std::sync::mpsc::sync_channel(CHANNEL_CAPACITY);
50            capture::spawn(rx, PathBuf::from(dir), config.capture_max_bytes());
51            tx
52        });
53        AnalyticsHandle { events, capture }
54    }
55
56    pub fn events_enabled(&self) -> bool {
57        self.events.is_some()
58    }
59
60    pub fn capture_enabled(&self) -> bool {
61        self.capture.is_some()
62    }
63
64    pub fn emit(&self, event: AnalyticsEvent) {
65        match &event {
66            AnalyticsEvent::GameStarted { game_id, .. } => {
67                self.send_capture(capture::CaptureMessage::Open {
68                    game_id: game_id.clone(),
69                    header: json_line(&event),
70                });
71            }
72            AnalyticsEvent::GameEnded { game_id, .. } => {
73                self.send_capture(capture::CaptureMessage::Close {
74                    game_id: game_id.clone(),
75                    footer: json_line(&event),
76                });
77            }
78            _ => {}
79        }
80        if let Some(tx) = &self.events {
81            if tx.try_send(event).is_err() {
82                metrics::record_analytics_dropped();
83            }
84        }
85    }
86
87    pub fn capture_envelope(
88        &self,
89        game_id: &str,
90        from_player: &str,
91        envelope: &Value,
92        client_rtt_ms: Option<u32>,
93    ) {
94        if self.capture.is_none() {
95            return;
96        }
97        let line = json_line(&event::CaptureLine {
98            ts: now_ts(),
99            from: from_player,
100            envelope,
101            client_rtt_ms,
102        });
103        self.send_capture(capture::CaptureMessage::Line {
104            game_id: game_id.to_string(),
105            line,
106        });
107    }
108
109    fn send_capture(&self, message: capture::CaptureMessage) {
110        if let Some(tx) = &self.capture {
111            if tx.try_send(message).is_err() {
112                metrics::record_analytics_dropped();
113            }
114        }
115    }
116}
117
118pub fn now_ts() -> String {
119    chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)
120}
121
122fn json_line<T: serde::Serialize>(value: &T) -> String {
123    serde_json::to_string(value).unwrap_or_default()
124}
125
126pub fn game_started_event(started: &StartedGame) -> AnalyticsEvent {
127    let players = started
128        .player_order
129        .iter()
130        .map(|username| {
131            let deck = started
132                .player_decks
133                .iter()
134                .find(|deck| &deck.username == username);
135            let is_bot = started
136                .room_info
137                .players
138                .iter()
139                .any(|player| &player.username == username && player.is_bot);
140            SeatInfo {
141                username: username.clone(),
142                is_bot,
143                deck_name: deck.map(|deck| deck.deck_name.clone()),
144                commander: deck.and_then(|deck| deck.commander_name.clone()),
145                published_deck_id: deck.and_then(|deck| deck.published_deck_id.clone()),
146                deck_fingerprint: deck
147                    .map(|deck| manabrew_protocol::deck_dto::deck_fingerprint(&deck.deck)),
148            }
149        })
150        .collect();
151    AnalyticsEvent::GameStarted {
152        ts: now_ts(),
153        game_id: started.game_id.clone(),
154        room_id: started.room_id.clone(),
155        format: started.room_info.format.clone(),
156        engine: started.room_info.engine,
157        hosted: started.room_info.hosted,
158        official: started.room_info.official,
159        table_style: started.room_info.table_style.clone(),
160        starting_life: started.starting_life,
161        players,
162    }
163}
164
165pub fn emit_game_ended(
166    handle: &AnalyticsHandle,
167    room: &Room,
168    replay: &GameReplayCache,
169    fallback: GameEndReason,
170) {
171    let reason = if replay.outcome.game_over {
172        GameEndReason::GameOver
173    } else if replay.outcome.fatal_message.is_some() {
174        GameEndReason::EngineFatal
175    } else {
176        fallback
177    };
178    metrics::record_game_ended(reason);
179    let winner = replay
180        .outcome
181        .winner_slot
182        .as_deref()
183        .and_then(|slot| replay.username_for_slot(slot));
184    let conceded = replay
185        .outcome
186        .conceded_slots
187        .iter()
188        .filter_map(|slot| replay.username_for_slot(slot))
189        .collect();
190    handle.emit(AnalyticsEvent::GameEnded {
191        ts: now_ts(),
192        game_id: replay.game_id.clone(),
193        room_id: room.room_id.clone(),
194        reason,
195        duration_s: replay.started_at.elapsed().as_secs_f64(),
196        game_over: replay.outcome.game_over,
197        winner,
198        conceded,
199        fatal_message: replay.outcome.fatal_message.clone(),
200        reported: replay.outcome.reported,
201        turns: replay.outcome.turns,
202    });
203}
204
205pub fn aggregate_deck_cards(deck: &Deck) -> Vec<CardEntry> {
206    let mut counts: HashMap<(String, String), u32> = HashMap::new();
207    let commanders = deck.commanders.iter().flatten();
208    for card in deck.cards.iter().chain(commanders) {
209        *counts
210            .entry((card.identity.name.clone(), card.identity.set_code.clone()))
211            .or_insert(0) += 1;
212    }
213    let mut cards: Vec<CardEntry> = counts
214        .into_iter()
215        .map(|((name, set_code), count)| CardEntry {
216            name,
217            set_code,
218            count,
219        })
220        .collect();
221    cards.sort_by(|a, b| a.name.cmp(&b.name));
222    cards
223}
manabrew_server/analytics/mod.rs

manabrew_server/analytics/
mod.rs
