hypercall_competition/

service.rs

use anyhow::anyhow;

use crate::{CompetitionError, Result};
use hypercall_db::{
    CompetitionFillInput, CompetitionFinalStatsInput, CompetitionRecord, CompetitionUpdateInput,
    CompetitionUpsertInput, CompetitionWriteError, CompetitionWriter, ProfileFillRecord,
    WalletUsernameRecord,
};
use hypercall_runtime_api::boundary::market_inputs::GreeksCacheReader;
use hypercall_runtime_api::valuation::get_symbol_theoretical_price;
use hypercall_types::ws_protocol::{
    WsCompetitionFinalStats, WsCompetitionPnlStanding, WsCompetitionPnlSummary, WsCompetitionUpdate,
};

use crate::{competition_state, CompetitionState};
use hypercall_types::{to_human_readable_decimal, WalletAddress};
use rust_decimal::Decimal;
use rust_decimal_macros::dec;
use std::cmp::Ordering;
use std::collections::{BTreeSet, HashMap, VecDeque};
use std::str::FromStr;
use std::sync::Arc;

#[derive(Debug, Clone, Copy)]
pub enum LeaderboardSortBy {
    Pnl,
    Volume,
    Efficiency,
}

impl LeaderboardSortBy {
    pub fn parse(input: &str) -> Option<Self> {
        match input {
            "pnl" => Some(Self::Pnl),
            "volume" => Some(Self::Volume),
            "efficiency" => Some(Self::Efficiency),
            _ => None,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CompetitionWinCondition {
    Pnl,
    Volume,
    Efficiency,
}

impl CompetitionWinCondition {
    fn parse(input: &str) -> Option<Self> {
        match input {
            "pnl" => Some(Self::Pnl),
            "volume" => Some(Self::Volume),
            "efficiency" => Some(Self::Efficiency),
            _ => None,
        }
    }

    fn sort_by(self) -> LeaderboardSortBy {
        match self {
            Self::Pnl => LeaderboardSortBy::Pnl,
            Self::Volume => LeaderboardSortBy::Volume,
            Self::Efficiency => LeaderboardSortBy::Efficiency,
        }
    }
}

#[derive(Debug, Clone, Copy)]
pub enum SortOrder {
    Asc,
    Desc,
}

impl SortOrder {
    pub fn parse(input: &str) -> Option<Self> {
        match input {
            "asc" => Some(Self::Asc),
            "desc" => Some(Self::Desc),
            _ => None,
        }
    }
}

#[derive(Debug, Clone)]
pub struct LeaderboardQuery {
    pub competition_id: i64,
    pub sort_by: LeaderboardSortBy,
    pub sort_order: SortOrder,
    pub limit: usize,
    pub offset: usize,
    pub wallet: Option<WalletAddress>,
}

#[derive(Debug, Clone)]
pub struct LeaderboardEntry {
    pub rank: usize,
    pub wallet: WalletAddress,
    pub username: String,
    pub pnl: Decimal,
    pub volume: Decimal,
    pub efficiency: Decimal,
    pub medal: Option<u8>,
}

#[derive(Debug, Clone)]
pub struct LeaderboardResult {
    pub rows: Vec<LeaderboardEntry>,
    pub connected_user: Option<LeaderboardEntry>,
}

#[derive(Debug, Clone)]
pub struct RealizedPnlBySymbol {
    pub symbol: String,
    pub realized_pnl: Decimal,
    pub event_count: i64,
}

#[derive(Debug, Clone, Copy, Default)]
pub struct PlatformMetricMedals {
    pub pnl: Option<u8>,
    pub volume: Option<u8>,
    pub efficiency: Option<u8>,
}

#[derive(Debug, Clone)]
pub struct CompetitionEngagementSnapshot {
    pub wallet: WalletAddress,
    pub competition_id: i64,
    pub rank: i64,
    pub pnl: Decimal,
    pub volume: Decimal,
    pub efficiency: Decimal,
    pub next_rank: Option<i64>,
    pub gap_metric_value: Option<Decimal>,
}

#[derive(Debug, Clone)]
struct WalletStats {
    realized: Decimal,
    unrealized: Decimal,
    fees: Decimal,
    volume: Decimal,
}

impl WalletStats {
    fn pnl(&self) -> Decimal {
        self.realized + self.unrealized - self.fees
    }

    fn efficiency(&self) -> Decimal {
        if self.volume == dec!(0) {
            dec!(0)
        } else {
            self.pnl() / self.volume
        }
    }
}

#[derive(Debug, Clone)]
struct Lot {
    qty: Decimal,
    price: Decimal,
    opened_in_window: bool,
}

#[derive(Debug, Clone, Default)]
struct SymbolLots {
    long_lots: VecDeque<Lot>,
    short_lots: VecDeque<Lot>,
    realized: Decimal,
    fees: Decimal,
    volume: Decimal,
}

impl SymbolLots {
    fn apply_fill(
        &mut self,
        side: &str,
        price: Decimal,
        size: Decimal,
        fee: Decimal,
        in_window: bool,
    ) -> Result<()> {
        if size <= dec!(0) {
            return Err(CompetitionError::BadRequest(
                "fill size must be positive".to_string(),
            ));
        }

        if in_window {
            self.fees += fee;
            self.volume += price * size;
        }

        match side {
            "Buy" => self.apply_buy(price, size, in_window),
            "Sell" => self.apply_sell(price, size, in_window),
            other => panic!(
                "invariant violated: invalid side '{}' in competition_fill_events",
                other
            ),
        }
    }

    fn apply_buy(&mut self, price: Decimal, size: Decimal, in_window: bool) -> Result<()> {
        let mut remaining = size;

        while remaining > dec!(0) {
            let Some(front) = self.short_lots.front_mut() else {
                break;
            };

            let matched = remaining.min(front.qty);
            if in_window {
                self.realized += (front.price - price) * matched;
            }
            front.qty -= matched;
            remaining -= matched;

            if front.qty == dec!(0) {
                self.short_lots.pop_front();
            }
        }

        if remaining > dec!(0) {
            self.long_lots.push_back(Lot {
                qty: remaining,
                price,
                opened_in_window: in_window,
            });
        }

        Ok(())
    }

    fn apply_sell(&mut self, price: Decimal, size: Decimal, in_window: bool) -> Result<()> {
        let mut remaining = size;

        while remaining > dec!(0) {
            let Some(front) = self.long_lots.front_mut() else {
                break;
            };

            let matched = remaining.min(front.qty);
            if in_window {
                self.realized += (price - front.price) * matched;
            }
            front.qty -= matched;
            remaining -= matched;

            if front.qty == dec!(0) {
                self.long_lots.pop_front();
            }
        }

        if remaining > dec!(0) {
            self.short_lots.push_back(Lot {
                qty: remaining,
                price,
                opened_in_window: in_window,
            });
        }

        Ok(())
    }

    fn has_in_window_lots(&self) -> bool {
        self.long_lots.iter().any(|lot| lot.opened_in_window)
            || self.short_lots.iter().any(|lot| lot.opened_in_window)
    }

    fn unrealized_with_mark(&self, mark: Decimal) -> Decimal {
        let mut total = dec!(0);
        for lot in &self.long_lots {
            if lot.opened_in_window {
                total += (mark - lot.price) * lot.qty;
            }
        }
        for lot in &self.short_lots {
            if lot.opened_in_window {
                total += (lot.price - mark) * lot.qty;
            }
        }
        total
    }
}

#[derive(Clone)]
pub struct CompetitionService {
    db: Arc<dyn CompetitionWriter>,
    greeks_cache: Arc<dyn GreeksCacheReader>,
}

impl CompetitionService {
    pub fn new(db: Arc<dyn CompetitionWriter>, greeks_cache: Arc<dyn GreeksCacheReader>) -> Self {
        Self { db, greeks_cache }
    }

    pub async fn list_competitions(
        &self,
        state: Option<CompetitionState>,
        from_ts_ms: Option<i64>,
        to_ts_ms: Option<i64>,
        limit: usize,
        offset: usize,
        now_ts_ms: i64,
    ) -> Result<Vec<CompetitionRecord>> {
        let state_str: Option<&str> = state.map(|s| s.as_str());
        Ok(self
            .db
            .list_competitions(
                state_str,
                from_ts_ms,
                to_ts_ms,
                now_ts_ms,
                limit as i64,
                offset as i64,
            )
            .await?)
    }

    pub async fn get_competition(&self, competition_id: i64) -> Result<CompetitionRecord> {
        self.db
            .get_competition_by_id(competition_id)
            .await?
            .ok_or_else(|| CompetitionError::NotFound(format!("competition {}", competition_id)))
    }

    pub async fn create_competition(
        &self,
        input: CompetitionUpsertInput,
    ) -> Result<CompetitionRecord> {
        Self::validate_competition_window(input.start_ts_ms, input.end_ts_ms)?;
        Self::validate_win_conditions(&input.win_conditions, &input.primary_win_condition)?;

        self.db
            .create_competition(&input)
            .await
            .map_err(|e| match e {
                CompetitionWriteError::OverlapViolation => CompetitionError::Conflict(
                    "competition window overlaps existing competition".to_string(),
                ),
                CompetitionWriteError::Internal(err) => {
                    CompetitionError::Internal(anyhow!("failed to create competition: {}", err))
                }
                other => CompetitionError::Internal(anyhow!("{}", other)),
            })
    }

    pub async fn update_competition(
        &self,
        competition_id: i64,
        input: CompetitionUpdateInput,
        now_ts_ms: i64,
    ) -> Result<CompetitionRecord> {
        let current = self.get_competition(competition_id).await?;
        if competition_state(&current, now_ts_ms) == CompetitionState::Post {
            return Err(CompetitionError::BadRequest(
                "post competitions are immutable".to_string(),
            ));
        }

        let name = input.name.unwrap_or(current.name);
        let description = input.description.unwrap_or(current.description);
        let rules_url = input.rules_url.unwrap_or(current.rules_url);
        let rules_content = input.rules_content.unwrap_or(current.rules_content);
        let win_conditions = input.win_conditions.unwrap_or(current.win_conditions);
        let primary_win_condition = input
            .primary_win_condition
            .unwrap_or(current.primary_win_condition);
        let start_ts_ms = input.start_ts_ms.unwrap_or(current.start_ts_ms);
        let end_ts_ms = input.end_ts_ms.unwrap_or(current.end_ts_ms);
        Self::validate_competition_window(start_ts_ms, end_ts_ms)?;
        Self::validate_win_conditions(&win_conditions, &primary_win_condition)?;

        self.db
            .update_competition(
                competition_id,
                &name,
                description.as_deref(),
                rules_url.as_deref(),
                rules_content.as_deref(),
                &win_conditions,
                &primary_win_condition,
                start_ts_ms,
                end_ts_ms,
            )
            .await
            .map_err(|e| match e {
                CompetitionWriteError::OverlapViolation => CompetitionError::Conflict(
                    "competition window overlaps existing competition".to_string(),
                ),
                CompetitionWriteError::NotFound(msg) => CompetitionError::NotFound(msg),
                CompetitionWriteError::Internal(err) => {
                    CompetitionError::Internal(anyhow!("failed to update competition: {}", err))
                }
                other => CompetitionError::Internal(anyhow!("{}", other)),
            })
    }

    pub async fn delete_competition(&self, competition_id: i64, now_ts_ms: i64) -> Result<()> {
        let competition = self.get_competition(competition_id).await?;
        if competition_state(&competition, now_ts_ms) != CompetitionState::Pre {
            return Err(CompetitionError::BadRequest(
                "only pre competitions can be deleted".to_string(),
            ));
        }

        let result = self.db.delete_competition(competition_id).await?;
        if result == 0 {
            return Err(CompetitionError::NotFound(format!(
                "competition {}",
                competition_id
            )));
        }

        Ok(())
    }

    pub async fn record_fill(&self, fill: &hypercall_types::Fill) -> Result<()> {
        let taker_size_human = to_human_readable_decimal(&fill.symbol, fill.size);
        let maker_side = match fill.taker_side {
            hypercall_types::Side::Buy => "Sell",
            hypercall_types::Side::Sell => "Buy",
        };

        self.db
            .record_competition_fill(&CompetitionFillInput {
                trade_id: fill.trade_id as i64,
                wallet: fill.taker_wallet_address,
                symbol: fill.symbol.clone(),
                side: format!("{:?}", fill.taker_side),
                price: fill.price,
                size: taker_size_human,
                fee: fill.fee,
                timestamp_ms: fill.timestamp as i64,
            })
            .await?;

        self.db
            .record_competition_fill(&CompetitionFillInput {
                trade_id: fill.trade_id as i64,
                wallet: fill.maker_wallet_address,
                symbol: fill.symbol.clone(),
                side: maker_side.to_string(),
                price: fill.price,
                size: taker_size_human,
                fee: dec!(0),
                timestamp_ms: fill.timestamp as i64,
            })
            .await?;

        Ok(())
    }

    pub async fn get_leaderboard(
        &self,
        query: LeaderboardQuery,
        now_ts_ms: i64,
    ) -> Result<LeaderboardResult> {
        let competition = self.get_competition(query.competition_id).await?;
        let state = competition_state(&competition, now_ts_ms);
        let mut used_frozen_rows = false;
        let cutoff_ts_ms = now_ts_ms.min(competition.end_ts_ms);

        let mut rows = if state == CompetitionState::Pre {
            Vec::new()
        } else if state == CompetitionState::Post {
            let finalized = self.get_finalized_leaderboard(query.competition_id).await?;
            if finalized.is_empty() {
                self.compute_live_leaderboard(
                    &competition,
                    query.sort_by,
                    query.sort_order,
                    true,
                    competition.end_ts_ms,
                )
                .await?
            } else {
                used_frozen_rows = true;
                finalized
            }
        } else {
            self.compute_live_leaderboard(
                &competition,
                query.sort_by,
                query.sort_order,
                false,
                cutoff_ts_ms,
            )
            .await?
        };

        if used_frozen_rows {
            Self::sort_entries(&mut rows, query.sort_by, query.sort_order);
            for (idx, row) in rows.iter_mut().enumerate() {
                row.rank = idx + 1;
                row.medal = match row.rank {
                    1 => Some(1),
                    2 => Some(2),
                    3 => Some(3),
                    _ => None,
                };
            }
        }

        let connected_user = query
            .wallet
            .and_then(|wallet| rows.iter().find(|r| r.wallet == wallet).cloned());

        rows = rows
            .into_iter()
            .skip(query.offset)
            .take(query.limit)
            .collect();

        Ok(LeaderboardResult {
            rows,
            connected_user,
        })
    }

    pub async fn get_realized_pnl_by_symbol(
        &self,
        wallet: WalletAddress,
        competition_id: Option<i64>,
    ) -> Result<Vec<RealizedPnlBySymbol>> {
        let (window_start, window_end) = if let Some(comp_id) = competition_id {
            let comp = self.get_competition(comp_id).await?;
            (Some(comp.start_ts_ms), Some(comp.end_ts_ms))
        } else {
            (None, None)
        };

        let records = self
            .db
            .get_realized_pnl_by_symbol(&wallet, window_start, window_end)
            .await?;

        Ok(records
            .into_iter()
            .map(|r| RealizedPnlBySymbol {
                symbol: r.symbol,
                realized_pnl: r.realized_pnl,
                event_count: r.event_count,
            })
            .collect())
    }

    pub async fn build_competition_update_for_wallet(
        &self,
        wallet: WalletAddress,
        now_ts_ms: i64,
    ) -> Result<Option<WsCompetitionUpdate>> {
        let active = self.db.get_active_competition(now_ts_ms).await?;
        let Some(competition) = active else {
            return Ok(None);
        };
        let primary_sort = CompetitionWinCondition::parse(&competition.primary_win_condition)
            .ok_or_else(|| {
                anyhow!(
                    "invalid competition primary_win_condition {}",
                    competition.primary_win_condition
                )
            })?
            .sort_by();

        let result = self
            .get_leaderboard(
                LeaderboardQuery {
                    competition_id: competition.id,
                    sort_by: primary_sort,
                    sort_order: SortOrder::Desc,
                    limit: 1,
                    offset: 0,
                    wallet: Some(wallet),
                },
                now_ts_ms,
            )
            .await?;

        let Some(me) = result.connected_user else {
            return Ok(None);
        };

        Ok(Some(WsCompetitionUpdate {
            wallet_address: wallet,
            competition_id: competition.id,
            rank: me.rank as i64,
            pnl: me.pnl,
            volume: me.volume,
            efficiency: me.efficiency,
            timestamp: now_ts_ms,
        }))
    }

    pub async fn build_competition_engagement_snapshot_for_wallet(
        &self,
        wallet: WalletAddress,
        now_ts_ms: i64,
    ) -> Result<Option<CompetitionEngagementSnapshot>> {
        let Some(competition) = self.db.get_active_competition(now_ts_ms).await? else {
            return Ok(None);
        };

        let primary = CompetitionWinCondition::parse(&competition.primary_win_condition)
            .ok_or_else(|| {
                anyhow!(
                    "invalid competition primary_win_condition {}",
                    competition.primary_win_condition
                )
            })?;

        let cutoff_ts_ms = now_ts_ms.min(competition.end_ts_ms);
        let rows = self
            .compute_live_leaderboard(
                &competition,
                primary.sort_by(),
                SortOrder::Desc,
                false,
                cutoff_ts_ms,
            )
            .await?;
        let Some((idx, me)) = rows
            .iter()
            .enumerate()
            .find(|(_, row)| row.wallet == wallet)
        else {
            return Ok(None);
        };

        let gap_metric_value = if idx == 0 {
            None
        } else {
            let ahead = &rows[idx - 1];
            let my_metric = Self::metric_value(me, primary.sort_by());
            let ahead_metric = Self::metric_value(ahead, primary.sort_by());
            Some(ahead_metric - my_metric)
        };

        Ok(Some(CompetitionEngagementSnapshot {
            wallet,
            competition_id: competition.id,
            rank: me.rank as i64,
            pnl: me.pnl,
            volume: me.volume,
            efficiency: me.efficiency,
            next_rank: if idx == 0 {
                None
            } else {
                Some(rows[idx - 1].rank as i64)
            },
            gap_metric_value,
        }))
    }

    pub async fn build_competition_pnl_summary_for_wallet(
        &self,
        wallet: WalletAddress,
        now_ts_ms: i64,
    ) -> Result<WsCompetitionPnlSummary> {
        let ledger_stats = self
            .db
            .compute_ledger_profile_stats(&wallet, now_ts_ms)
            .await?;
        let lifetime_realized_pnl = ledger_stats.lifetime_realized_pnl;

        let active_competition = match self.db.get_active_competition(now_ts_ms).await? {
            Some(competition) => {
                let primary_sort =
                    CompetitionWinCondition::parse(&competition.primary_win_condition)
                        .ok_or_else(|| {
                            anyhow!(
                                "invalid competition primary_win_condition {}",
                                competition.primary_win_condition
                            )
                        })?
                        .sort_by();

                let result = self
                    .get_leaderboard(
                        LeaderboardQuery {
                            competition_id: competition.id,
                            sort_by: primary_sort,
                            sort_order: SortOrder::Desc,
                            limit: 1,
                            offset: 0,
                            wallet: Some(wallet),
                        },
                        now_ts_ms,
                    )
                    .await?;

                Some(match result.connected_user {
                    Some(me) => WsCompetitionPnlStanding {
                        competition_id: competition.id,
                        competition_name: competition.name.clone(),
                        competition_state: competition_state(&competition, now_ts_ms)
                            .as_str()
                            .to_string(),
                        rank: Some(me.rank),
                        pnl: me.pnl,
                        volume: me.volume,
                        efficiency: me.efficiency,
                        medal: me.medal,
                    },
                    None => WsCompetitionPnlStanding {
                        competition_id: competition.id,
                        competition_name: competition.name.clone(),
                        competition_state: competition_state(&competition, now_ts_ms)
                            .as_str()
                            .to_string(),
                        rank: None,
                        pnl: dec!(0),
                        volume: dec!(0),
                        efficiency: dec!(0),
                        medal: None,
                    },
                })
            }
            None => None,
        };

        Ok(WsCompetitionPnlSummary {
            wallet_address: wallet,
            lifetime_realized_pnl,
            active_competition,
            timestamp: now_ts_ms,
        })
    }

    pub async fn finalize_ended_competitions(
        &self,
        now_ts_ms: i64,
    ) -> Result<Vec<WsCompetitionFinalStats>> {
        let candidates = self.db.get_competitions_to_finalize(now_ts_ms).await?;

        let mut notifications = Vec::new();

        for competition in candidates {
            let rows = self
                .compute_live_leaderboard(
                    &competition,
                    CompetitionWinCondition::parse(&competition.primary_win_condition)
                        .ok_or_else(|| {
                            anyhow!(
                                "invalid competition primary_win_condition {}",
                                competition.primary_win_condition
                            )
                        })?
                        .sort_by(),
                    SortOrder::Desc,
                    true,
                    competition.end_ts_ms,
                )
                .await?;

            if rows.is_empty() {
                continue;
            }

            let competition_id = competition.id;
            let stats: Vec<CompetitionFinalStatsInput> = rows
                .iter()
                .map(|row| CompetitionFinalStatsInput {
                    competition_id,
                    wallet: row.wallet,
                    rank: row.rank as i32,
                    pnl: row.pnl,
                    volume: row.volume,
                    efficiency: row.efficiency,
                    medal: row.medal.map(i32::from),
                })
                .collect();

            let inserted = self.db.finalize_competition(competition_id, &stats).await?;

            if inserted > 0 {
                for row in &rows {
                    notifications.push(WsCompetitionFinalStats {
                        wallet_address: row.wallet,
                        competition_id,
                        rank: row.rank as i64,
                        pnl: row.pnl,
                        volume: row.volume,
                        efficiency: row.efficiency,
                        medal: row.medal.map(i64::from),
                        timestamp: now_ts_ms,
                    });
                }
            }
        }

        Ok(notifications)
    }

    pub async fn get_display_username(&self, wallet: WalletAddress) -> Result<String> {
        let username = self.db.get_display_username(&wallet).await?;
        Ok(username.unwrap_or_else(|| truncate_wallet(wallet)))
    }

    pub async fn get_profile_image_url(&self, wallet: WalletAddress) -> Result<Option<String>> {
        Ok(self.db.get_profile_image_url(&wallet).await?)
    }

    pub async fn set_profile_image_url(
        &self,
        wallet: WalletAddress,
        profile_image_url: &str,
    ) -> Result<Option<String>> {
        Ok(self
            .db
            .set_profile_image_url(&wallet, profile_image_url)
            .await?)
    }

    async fn get_display_usernames(
        &self,
        wallets: impl IntoIterator<Item = WalletAddress>,
    ) -> Result<HashMap<WalletAddress, String>> {
        let unique_wallets: Vec<WalletAddress> = wallets
            .into_iter()
            .collect::<BTreeSet<_>>()
            .into_iter()
            .collect();

        let mut usernames = HashMap::with_capacity(unique_wallets.len());
        for wallet in &unique_wallets {
            usernames.insert(*wallet, truncate_wallet(*wallet));
        }

        if unique_wallets.is_empty() {
            return Ok(usernames);
        }

        let wallet_strings: Vec<String> = unique_wallets
            .iter()
            .map(|w| format!("{:#x}", w.0).to_lowercase())
            .collect();

        let rows: Vec<WalletUsernameRecord> =
            self.db.get_display_usernames_batch(&wallet_strings).await?;

        for row in rows {
            if let Ok(addr) = WalletAddress::from_str(&row.wallet_address) {
                usernames.insert(addr, row.username);
            }
        }

        Ok(usernames)
    }

    pub async fn get_profile_trade_history(
        &self,
        wallet: WalletAddress,
        competition_id: Option<i64>,
        from_ts_ms: Option<i64>,
        to_ts_ms: Option<i64>,
        symbol: Option<&str>,
        limit: usize,
        offset: usize,
    ) -> Result<Vec<crate::Fill>> {
        let (window_start, window_end) = if let Some(comp_id) = competition_id {
            let comp = self.get_competition(comp_id).await?;
            (Some(comp.start_ts_ms), Some(comp.end_ts_ms))
        } else {
            (None, None)
        };

        let rows: Vec<ProfileFillRecord> = self
            .db
            .get_profile_trade_history(
                &wallet,
                window_start,
                window_end,
                from_ts_ms,
                to_ts_ms,
                symbol,
                limit as i64,
                offset as i64,
            )
            .await?;

        rows.into_iter()
            .map(|r| {
                let fill_id = r.fill_id.ok_or_else(|| {
                    anyhow!("fill row has NULL fill_id (trade_id={})", r.trade_id)
                })?;
                let created_at = r
                    .created_at
                    .ok_or_else(|| anyhow!("fill row has NULL created_at (fill_id={})", fill_id))?;
                Ok(crate::Fill {
                    fill_id,
                    trade_id: r.trade_id,
                    wallet_address: r.wallet_address,
                    symbol: r.symbol,
                    price: r.price,
                    size: r.size,
                    fee: r.fee,
                    is_taker: r.is_taker,
                    timestamp: r.timestamp,
                    created_at,
                    builder_code_address: r.builder_code_address,
                    builder_code_fee: r.builder_code_fee,
                    realized_pnl: r.realized_pnl,
                })
            })
            .collect()
    }

    pub async fn compute_ledger_profile_stats(
        &self,
        wallet: WalletAddress,
        now_ts_ms: i64,
    ) -> Result<(Decimal, Decimal, Decimal, Decimal)> {
        let stats = self
            .db
            .compute_ledger_profile_stats(&wallet, now_ts_ms)
            .await?;
        Ok((
            stats.deposits,
            stats.withdrawals,
            stats.lifetime_realized_pnl,
            stats.pnl_24h,
        ))
    }

    pub async fn get_account_first_seen_ts_ms(&self, wallet: WalletAddress) -> Result<Option<i64>> {
        Ok(self.db.get_account_first_seen_ts_ms(&wallet).await?)
    }

    pub async fn get_platform_metric_medals(
        &self,
        wallet: WalletAddress,
    ) -> Result<PlatformMetricMedals> {
        let metrics = self.db.get_platform_wallet_metrics().await?;

        let rows: Vec<(WalletAddress, Decimal, Decimal, Decimal)> = metrics
            .into_iter()
            .map(|m| {
                let efficiency = if m.volume == dec!(0) {
                    dec!(0)
                } else {
                    m.realized / m.volume
                };
                (m.wallet, m.realized, m.volume, efficiency)
            })
            .collect();

        let pnl_rows: Vec<(WalletAddress, Decimal, Decimal)> =
            rows.iter().map(|row| (row.0, row.1, row.2)).collect();
        let volume_rows: Vec<(WalletAddress, Decimal, Decimal)> =
            rows.iter().map(|row| (row.0, row.2, row.2)).collect();
        let efficiency_rows: Vec<(WalletAddress, Decimal, Decimal)> =
            rows.iter().map(|row| (row.0, row.3, row.2)).collect();

        Ok(PlatformMetricMedals {
            pnl: Self::wallet_medal_for_metric(&pnl_rows, wallet, false),
            volume: Self::wallet_medal_for_metric(&volume_rows, wallet, false),
            efficiency: Self::wallet_medal_for_metric(&efficiency_rows, wallet, true),
        })
    }

    pub async fn get_active_competition(
        &self,
        now_ts_ms: i64,
    ) -> Result<Option<CompetitionRecord>> {
        Ok(self.db.get_active_competition(now_ts_ms).await?)
    }

    pub async fn get_latest_completed_competition(
        &self,
        now_ts_ms: i64,
    ) -> Result<Option<CompetitionRecord>> {
        Ok(self.db.get_latest_completed_competition(now_ts_ms).await?)
    }

    async fn get_finalized_leaderboard(
        &self,
        competition_id: i64,
    ) -> Result<Vec<LeaderboardEntry>> {
        let stats = self.db.get_finalized_stats(competition_id).await?;

        let usernames = self
            .get_display_usernames(stats.iter().map(|row| row.wallet))
            .await?;

        let mut entries = Vec::with_capacity(stats.len());
        for row in stats {
            entries.push(LeaderboardEntry {
                rank: row.rank as usize,
                wallet: row.wallet,
                username: usernames
                    .get(&row.wallet)
                    .cloned()
                    .unwrap_or_else(|| truncate_wallet(row.wallet)),
                pnl: row.pnl,
                volume: row.volume,
                efficiency: row.efficiency,
                medal: row.medal.and_then(|m| u8::try_from(m).ok()),
            });
        }

        Ok(entries)
    }

    async fn compute_live_leaderboard(
        &self,
        competition: &CompetitionRecord,
        sort_by: LeaderboardSortBy,
        sort_order: SortOrder,
        use_historical_theoretical_marks: bool,
        cutoff_ts_ms: i64,
    ) -> Result<Vec<LeaderboardEntry>> {
        let fills = self.db.get_competition_fills_before(cutoff_ts_ms).await?;

        if fills.is_empty() {
            return Ok(Vec::new());
        }

        let mut wallet_symbol_lots: HashMap<WalletAddress, HashMap<String, SymbolLots>> =
            HashMap::new();

        for fill in fills {
            let in_window =
                fill.timestamp_ms >= competition.start_ts_ms && fill.timestamp_ms < cutoff_ts_ms;
            let symbol_map = wallet_symbol_lots.entry(fill.wallet).or_default();
            let lots = symbol_map.entry(fill.symbol.clone()).or_default();
            lots.apply_fill(&fill.side, fill.price, fill.size, fill.fee, in_window)?;
        }

        let symbols_needing_marks: Vec<String> = wallet_symbol_lots
            .values()
            .flat_map(|m| {
                m.iter()
                    .filter(|&(_symbol, lots)| lots.has_in_window_lots())
                    .map(|(symbol, _lots)| symbol.clone())
            })
            .collect::<std::collections::BTreeSet<_>>()
            .into_iter()
            .collect();

        let marks = if use_historical_theoretical_marks {
            self.get_historical_theoretical_marks(&symbols_needing_marks, cutoff_ts_ms)
                .await?
        } else {
            self.get_live_theoretical_marks(&symbols_needing_marks)
                .await?
        };

        let usernames = self
            .get_display_usernames(wallet_symbol_lots.keys().copied())
            .await?;

        let mut leaderboard: Vec<LeaderboardEntry> = Vec::new();

        for (wallet, symbol_lots) in wallet_symbol_lots {
            let mut stats = WalletStats {
                realized: dec!(0),
                unrealized: dec!(0),
                fees: dec!(0),
                volume: dec!(0),
            };

            for (symbol, lots) in symbol_lots {
                stats.realized += lots.realized;
                stats.fees += lots.fees;
                stats.volume += lots.volume;

                if lots.has_in_window_lots() {
                    let mark = marks.get(&symbol).ok_or_else(|| {
                        CompetitionError::Unavailable(format!(
                            "missing mark for symbol {} while computing leaderboard",
                            symbol
                        ))
                    })?;
                    stats.unrealized += lots.unrealized_with_mark(*mark);
                }
            }

            if stats.realized == dec!(0)
                && stats.unrealized == dec!(0)
                && stats.fees == dec!(0)
                && stats.volume == dec!(0)
            {
                continue;
            }

            leaderboard.push(LeaderboardEntry {
                rank: 0,
                wallet,
                username: usernames
                    .get(&wallet)
                    .cloned()
                    .unwrap_or_else(|| truncate_wallet(wallet)),
                pnl: stats.pnl(),
                volume: stats.volume,
                efficiency: stats.efficiency(),
                medal: None,
            });
        }

        Self::sort_entries(&mut leaderboard, sort_by, sort_order);

        for (idx, row) in leaderboard.iter_mut().enumerate() {
            row.rank = idx + 1;
            row.medal = match row.rank {
                1 => Some(1),
                2 => Some(2),
                3 => Some(3),
                _ => None,
            };
        }

        Ok(leaderboard)
    }

    fn sort_entries(
        entries: &mut [LeaderboardEntry],
        sort_by: LeaderboardSortBy,
        sort_order: SortOrder,
    ) {
        entries.sort_by(|a, b| {
            let primary = match sort_by {
                LeaderboardSortBy::Pnl => a.pnl.partial_cmp(&b.pnl).unwrap_or(Ordering::Equal),
                LeaderboardSortBy::Volume => {
                    a.volume.partial_cmp(&b.volume).unwrap_or(Ordering::Equal)
                }
                LeaderboardSortBy::Efficiency => a
                    .efficiency
                    .partial_cmp(&b.efficiency)
                    .unwrap_or(Ordering::Equal),
            };

            let primary = match sort_order {
                SortOrder::Asc => primary,
                SortOrder::Desc => primary.reverse(),
            };

            if primary != Ordering::Equal {
                return primary;
            }

            let secondary = b.volume.partial_cmp(&a.volume).unwrap_or(Ordering::Equal);
            if secondary != Ordering::Equal {
                return secondary;
            }

            a.wallet.to_string().cmp(&b.wallet.to_string())
        });
    }

    fn metric_value(entry: &LeaderboardEntry, sort_by: LeaderboardSortBy) -> Decimal {
        match sort_by {
            LeaderboardSortBy::Pnl => entry.pnl,
            LeaderboardSortBy::Volume => entry.volume,
            LeaderboardSortBy::Efficiency => entry.efficiency,
        }
    }

    fn wallet_medal_for_metric(
        rows: &[(WalletAddress, Decimal, Decimal)],
        wallet: WalletAddress,
        require_positive_volume: bool,
    ) -> Option<u8> {
        let mut ordered = rows.to_vec();
        ordered.sort_by(|a, b| {
            let primary = b.1.partial_cmp(&a.1).unwrap_or(Ordering::Equal);
            if primary != Ordering::Equal {
                return primary;
            }
            let volume_tiebreak = b.2.partial_cmp(&a.2).unwrap_or(Ordering::Equal);
            if volume_tiebreak != Ordering::Equal {
                return volume_tiebreak;
            }
            a.0.to_string().cmp(&b.0.to_string())
        });

        let mut eligible_rank = 0_usize;
        for row in &ordered {
            if require_positive_volume && row.2 == dec!(0) {
                continue;
            }
            eligible_rank += 1;
            if row.0 == wallet {
                return match eligible_rank {
                    1 => Some(1),
                    2 => Some(2),
                    3 => Some(3),
                    _ => None,
                };
            }
        }
        None
    }

    async fn get_historical_theoretical_marks(
        &self,
        symbols: &[String],
        cutoff_ts_ms: i64,
    ) -> Result<HashMap<String, Decimal>> {
        if symbols.is_empty() {
            return Ok(HashMap::new());
        }

        let records = self
            .db
            .get_historical_theo_marks(symbols, cutoff_ts_ms)
            .await?;

        let mut map = HashMap::with_capacity(records.len());
        for record in records {
            map.insert(record.symbol, record.theoretical_price);
        }

        let missing_symbols: Vec<String> = symbols
            .iter()
            .filter(|symbol| !map.contains_key(symbol.as_str()))
            .cloned()
            .collect();

        if missing_symbols.is_empty() {
            return Ok(map);
        }

        tracing::warn!(
            cutoff_ts_ms,
            missing_symbols = ?missing_symbols,
            "Falling back to live theoretical marks for symbols missing historical snapshots"
        );

        let live_fallback_marks = self.get_live_theoretical_marks(&missing_symbols).await?;
        map.extend(live_fallback_marks);

        Ok(map)
    }

    async fn get_live_theoretical_marks(
        &self,
        symbols: &[String],
    ) -> Result<HashMap<String, Decimal>> {
        let mut map = HashMap::with_capacity(symbols.len());

        for symbol in symbols {
            let mark_f64 = get_symbol_theoretical_price(self.greeks_cache.as_ref(), symbol)
                .await
                .map_err(|error| CompetitionError::Unavailable(error.to_string()))?;
            let mark = Decimal::from_f64_retain(mark_f64)
                .ok_or_else(|| anyhow!("invalid theoretical price decimal for {}", symbol))?;
            map.insert(symbol.clone(), mark);
        }

        Ok(map)
    }

    fn validate_competition_window(start_ts_ms: i64, end_ts_ms: i64) -> Result<()> {
        if start_ts_ms >= end_ts_ms {
            return Err(CompetitionError::BadRequest(
                "start_ts_ms must be less than end_ts_ms".to_string(),
            ));
        }
        Ok(())
    }

    fn validate_win_conditions(
        win_conditions: &[String],
        primary_win_condition: &str,
    ) -> Result<()> {
        if win_conditions.is_empty() {
            return Err(CompetitionError::BadRequest(
                "win_conditions must be non-empty".to_string(),
            ));
        }
        if !win_conditions
            .iter()
            .all(|cond| CompetitionWinCondition::parse(cond).is_some())
        {
            return Err(CompetitionError::BadRequest(
                "win_conditions must contain only pnl|volume|efficiency".to_string(),
            ));
        }
        if CompetitionWinCondition::parse(primary_win_condition).is_none() {
            return Err(CompetitionError::BadRequest(
                "primary_win_condition must be one of pnl|volume|efficiency".to_string(),
            ));
        }
        if !win_conditions
            .iter()
            .any(|cond| cond == primary_win_condition)
        {
            return Err(CompetitionError::BadRequest(
                "primary_win_condition must be included in win_conditions".to_string(),
            ));
        }
        Ok(())
    }
}

fn truncate_wallet(wallet: WalletAddress) -> String {
    let full = wallet.to_string();
    if full.len() <= 10 {
        full
    } else {
        format!("{}...{}", &full[..6], &full[full.len() - 4..])
    }
}

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

    fn wallet(index: u8) -> WalletAddress {
        WalletAddress::from_str(&format!("0x{:040x}", index)).expect("valid wallet")
    }

    #[test]
    fn competition_state_transitions_follow_bounds() {
        assert_eq!(
            CompetitionState::from_bounds(2000, 3000, 1999),
            CompetitionState::Pre
        );
        assert_eq!(
            CompetitionState::from_bounds(2000, 3000, 2000),
            CompetitionState::Active
        );
        assert_eq!(
            CompetitionState::from_bounds(2000, 3000, 2999),
            CompetitionState::Active
        );
        assert_eq!(
            CompetitionState::from_bounds(2000, 3000, 3000),
            CompetitionState::Post
        );
    }

    #[test]
    fn lot_accounting_handles_long_close_and_reversal() {
        let mut lots = SymbolLots::default();
        lots.apply_fill("Buy", dec!(100), dec!(10), dec!(0), true)
            .expect("buy fill should apply");
        lots.apply_fill("Sell", dec!(120), dec!(4), dec!(0), true)
            .expect("partial close should apply");
        assert_eq!(lots.realized, dec!(80));
        assert_eq!(lots.unrealized_with_mark(dec!(110)), dec!(60));

        lots.apply_fill("Sell", dec!(90), dec!(10), dec!(0), true)
            .expect("reversal should apply");
        assert_eq!(lots.realized, dec!(20));
        assert_eq!(lots.unrealized_with_mark(dec!(80)), dec!(40));
    }

    #[test]
    fn sorting_uses_requested_metric_then_tiebreakers() {
        let mut entries = vec![
            LeaderboardEntry {
                rank: 0,
                wallet: wallet(1),
                username: "u1".to_string(),
                pnl: dec!(10),
                volume: dec!(100),
                efficiency: dec!(0.1),
                medal: None,
            },
            LeaderboardEntry {
                rank: 0,
                wallet: wallet(2),
                username: "u2".to_string(),
                pnl: dec!(10),
                volume: dec!(200),
                efficiency: dec!(0.05),
                medal: None,
            },
            LeaderboardEntry {
                rank: 0,
                wallet: wallet(3),
                username: "u3".to_string(),
                pnl: dec!(15),
                volume: dec!(150),
                efficiency: dec!(0.1),
                medal: None,
            },
        ];

        CompetitionService::sort_entries(&mut entries, LeaderboardSortBy::Pnl, SortOrder::Desc);
        assert_eq!(entries[0].wallet, wallet(3));
        assert_eq!(entries[1].wallet, wallet(2));
        assert_eq!(entries[2].wallet, wallet(1));
    }

    #[test]
    fn win_condition_validation_enforces_membership() {
        CompetitionService::validate_win_conditions(&["pnl".to_string()], "pnl")
            .expect("valid win condition");
        assert!(CompetitionService::validate_win_conditions(&[], "pnl").is_err());
        assert!(
            CompetitionService::validate_win_conditions(&["pnl".to_string()], "volume").is_err()
        );
    }
}