From e20b20dbca19d3e50006f8ddeae891a005e26e16 Mon Sep 17 00:00:00 2001 From: Evan Yang Date: Mon, 9 Mar 2026 16:57:15 +0800 Subject: [PATCH] Restore dashboard analytics data path --- routstr/core/admin.py | 16 +- routstr/core/log_manager.py | 1017 ++++++++++++++---- routstr/core/usage_analytics_store.py | 1380 +++++++++++++++++++++++++ ui/app/page.tsx | 1028 +++++++++++++++++- ui/components/error-details-table.tsx | 78 ++ ui/components/usage-metrics-chart.tsx | 387 +++++++ ui/components/usage-summary-cards.tsx | 160 +++ ui/lib/api/client.ts | 6 +- ui/lib/api/services/admin.ts | 114 ++ 9 files changed, 3962 insertions(+), 224 deletions(-) create mode 100644 routstr/core/usage_analytics_store.py create mode 100644 ui/components/error-details-table.tsx create mode 100644 ui/components/usage-metrics-chart.tsx create mode 100644 ui/components/usage-summary-cards.tsx diff --git a/routstr/core/admin.py b/routstr/core/admin.py index 7776d5f8..202929e2 100644 --- a/routstr/core/admin.py +++ b/routstr/core/admin.py @@ -940,9 +940,7 @@ async def get_usage_metrics( interval: int = Query( default=15, ge=1, le=1440, description="Time interval in minutes" ), - hours: int = Query( - default=24, ge=1, le=168, description="Hours of history to analyze" - ), + hours: int = Query(default=24, ge=1, description="Hours of history to analyze"), ) -> dict: """Get usage metrics aggregated by time interval.""" return log_manager.get_usage_metrics(interval=interval, hours=hours) @@ -951,9 +949,7 @@ async def get_usage_metrics( @admin_router.get("/api/usage/summary", dependencies=[Depends(require_admin_api)]) async def get_usage_summary( request: Request, - hours: int = Query( - default=24, ge=1, le=168, description="Hours of history to analyze" - ), + hours: int = Query(default=24, ge=1, description="Hours of history to analyze"), ) -> dict: """Get summary statistics for the specified time period.""" return log_manager.get_usage_summary(hours=hours) @@ -962,9 +958,7 @@ async def get_usage_summary( @admin_router.get("/api/usage/error-details", dependencies=[Depends(require_admin_api)]) async def get_error_details( request: Request, - hours: int = Query( - default=24, ge=1, le=168, description="Hours of history to analyze" - ), + hours: int = Query(default=24, ge=1, description="Hours of history to analyze"), limit: int = Query( default=100, ge=1, le=1000, description="Maximum number of errors to return" ), @@ -978,9 +972,7 @@ async def get_error_details( ) async def get_revenue_by_model( request: Request, - hours: int = Query( - default=24, ge=1, le=168, description="Hours of history to analyze" - ), + hours: int = Query(default=24, ge=1, description="Hours of history to analyze"), limit: int = Query( default=20, ge=1, le=100, description="Maximum number of models to return" ), diff --git a/routstr/core/log_manager.py b/routstr/core/log_manager.py index 43920605..0444dcbf 100644 --- a/routstr/core/log_manager.py +++ b/routstr/core/log_manager.py @@ -1,17 +1,73 @@ import json +import time from collections import defaultdict from datetime import datetime, timedelta, timezone +from heapq import heappush, heapreplace from pathlib import Path -from typing import Any, Iterator +from threading import Lock +from typing import Any, Callable, Iterator, TypeVar from .logging import get_logger +from .usage_analytics_store import UsageAnalyticsStore logger = get_logger(__name__) +T = TypeVar("T") class LogManager: def __init__(self, logs_dir: Path = Path("logs")): self.logs_dir = logs_dir + self._usage_store = UsageAnalyticsStore(logs_dir=logs_dir) + self._analytics_cache_ttl_seconds = 30.0 + self._analytics_cache: dict[tuple[Any, ...], tuple[float, Any]] = {} + self._analytics_cache_lock = Lock() + self._cache_miss = object() + + def _get_cached(self, key: tuple[Any, ...]) -> Any: + now = time.time() + with self._analytics_cache_lock: + cached = self._analytics_cache.get(key) + if cached is None: + return self._cache_miss + + expires_at, value = cached + if expires_at <= now: + self._analytics_cache.pop(key, None) + return self._cache_miss + + return value + + def _set_cached( + self, key: tuple[Any, ...], value: Any, ttl_seconds: float | None = None + ) -> None: + ttl = ( + self._analytics_cache_ttl_seconds + if ttl_seconds is None + else max(1.0, ttl_seconds) + ) + expires_at = time.time() + ttl + with self._analytics_cache_lock: + self._analytics_cache[key] = (expires_at, value) + + def _cache_call( + self, + key: tuple[Any, ...], + compute: Callable[[], T], + ttl_seconds: float | None = None, + ) -> T: + cached = self._get_cached(key) + if cached is not self._cache_miss: + return cached + + value = compute() + self._set_cached(key, value, ttl_seconds=ttl_seconds) + return value + + def _get_cached_entries(self, hours: int) -> list[dict[str, Any]]: + return self._cache_call( + ("usage_entries", hours), + lambda: list(self._yield_log_entries(hours_back=hours)), + ) def _yield_log_entries( self, @@ -34,6 +90,7 @@ class LogManager: log_files = [] cutoff_date = None + cutoff_timestamp_str: str | None = None if specific_date: log_file = self.logs_dir / f"app_{specific_date}.log" @@ -47,6 +104,7 @@ class LogManager: # If we only care about hours back, we can optimize file selection if hours_back is not None: cutoff_date = datetime.now(timezone.utc) - timedelta(hours=hours_back) + cutoff_timestamp_str = cutoff_date.strftime("%Y-%m-%d %H:%M:%S") filtered_files = [] for log_path in log_files: try: @@ -69,27 +127,20 @@ class LogManager: for log_file in log_files: try: with open(log_file, "r") as f: - # For reverse search, we might want to read lines in reverse? - # But usually logs are append-only. - # If reverse_files is True, we iterate files newest to oldest. - # But lines within file are still oldest to newest unless we reverse them. - lines = f.readlines() - if reverse_files: - lines.reverse() + lines_iter = reversed(f.readlines()) if reverse_files else f - for line in lines: + for line in lines_iter: try: entry = json.loads(line.strip()) - if cutoff_date: + if cutoff_timestamp_str: timestamp_str = entry.get("asctime", "") - if not timestamp_str: + if ( + not isinstance(timestamp_str, str) + or len(timestamp_str) != 19 + ): continue - log_time = datetime.strptime( - timestamp_str, "%Y-%m-%d %H:%M:%S" - ) - log_time = log_time.replace(tzinfo=timezone.utc) - if log_time < cutoff_date: + if timestamp_str < cutoff_timestamp_str: continue yield entry @@ -166,7 +217,7 @@ class LogManager: methods: list[str] | None = None, endpoints: list[str] | None = None, ) -> bool: - if level and log_data.get("levelname", "").upper() != level.upper(): + if level and str(log_data.get("levelname", "")).upper() != level.upper(): return False if request_id and log_data.get("request_id") != request_id: @@ -216,207 +267,279 @@ class LogManager: return True + def _bucket_key_for_timestamp( + self, timestamp_str: str, interval_minutes: int + ) -> str | None: + if len(timestamp_str) != 19: + return None + if timestamp_str[10] != " ": + return None + + try: + hour = int(timestamp_str[11:13]) + minute = int(timestamp_str[14:16]) + except (TypeError, ValueError): + return None + + total_minutes = hour * 60 + minute + rounded_minutes = (total_minutes // interval_minutes) * interval_minutes + rounded_hour = rounded_minutes // 60 + rounded_minute = rounded_minutes % 60 + return f"{timestamp_str[:10]} {rounded_hour:02d}:{rounded_minute:02d}:00" + + def _extract_success_metrics( + self, entry: dict[str, Any], message: str + ) -> tuple[bool, float, int, int]: + # Use auth settlement logs as the canonical successful request signal. + logger_name = str(entry.get("name", "")) + if not logger_name.startswith("routstr.auth"): + return False, 0.0, 0, 0 + + input_tokens = self._parse_token_count(entry.get("input_tokens", 0)) + output_tokens = self._parse_token_count(entry.get("output_tokens", 0)) + + if "calculated token-based cost" in message: + token_cost = entry.get("token_cost", 0) + if isinstance(token_cost, (int, float)) and token_cost > 0: + return True, float(token_cost), input_tokens, output_tokens + return True, 0.0, input_tokens, output_tokens + + if "max cost payment finalized" in message: + charged_amount = entry.get("charged_amount", 0) + if isinstance(charged_amount, (int, float)) and charged_amount > 0: + return True, float(charged_amount), input_tokens, output_tokens + return True, 0.0, input_tokens, output_tokens + + return False, 0.0, 0, 0 + + def _parse_token_count(self, value: Any) -> int: + if isinstance(value, bool): + return 0 + if isinstance(value, int): + return max(0, value) + if isinstance(value, float): + return max(0, int(value)) + if isinstance(value, str): + try: + return max(0, int(float(value))) + except ValueError: + return 0 + return 0 + def get_usage_summary(self, hours: int = 24) -> dict: - entries = list(self._yield_log_entries(hours_back=hours)) - return self._calculate_summary_stats(entries) - - def get_usage_metrics(self, interval: int = 15, hours: int = 24) -> dict: - entries = list(self._yield_log_entries(hours_back=hours)) - return self._aggregate_metrics_by_time(entries, interval, hours) - - def get_error_details(self, hours: int = 24, limit: int = 100) -> dict: - errors: list[dict] = [] - # Iterate newest to oldest for errors? - # yield_log_entries sorts files by name (date) ascending by default. - # usage stats logic usually expects ascending time for aggregation (though dictionaries don't care). - # For error details "last N errors", we probably want newest first. - - # Using list() loads everything into memory, which is what PR 229 did. - # For optimization, we could use reverse iterator. - - # Let's just stick to PR 229 logic which filters 'ERROR' level. - - entries = self._yield_log_entries(hours_back=hours) # oldest to newest - - for entry in entries: - if entry.get("levelname", "").upper() == "ERROR": - timestamp_str = entry.get("asctime", "") - errors.append( - { - "timestamp": timestamp_str, - "message": entry.get("message", ""), - "error_type": entry.get("error_type", "unknown"), - "pathname": entry.get("pathname", ""), - "lineno": entry.get("lineno", 0), - "request_id": entry.get("request_id", ""), - } + def compute() -> dict: + try: + return self._usage_store.get_summary(hours_back=hours) + except Exception as e: + logger.error( + f"Usage analytics index failed, falling back to log scan: {e}" ) + return self._calculate_summary_stats(self._get_cached_entries(hours)) - # Sort reverse time - errors.sort(key=lambda x: x["timestamp"], reverse=True) - return {"errors": errors[:limit], "total_count": len(errors)} - - def get_revenue_by_model(self, hours: int = 24, limit: int = 20) -> dict: - entries = list(self._yield_log_entries(hours_back=hours)) - - model_stats: dict[str, dict[str, int | float]] = defaultdict( - lambda: { - "revenue_msats": 0, - "refunds_msats": 0, - "requests": 0, - "successful": 0, - "failed": 0, - } + return self._cache_call( + ("usage_summary", hours), + compute, ) - for entry in entries: + def get_usage_metrics(self, interval: int = 15, hours: int = 24) -> dict: + def compute() -> dict: try: - model = entry.get("model", "unknown") - if not isinstance(model, str): - model = "unknown" + return self._usage_store.get_metrics( + interval_minutes=interval, + hours_back=hours, + ) + except Exception as e: + logger.error( + f"Usage analytics index failed, falling back to log scan: {e}" + ) + return self._aggregate_metrics_by_time( + self._get_cached_entries(hours), interval, hours + ) - message = entry.get("message", "").lower() + return self._cache_call( + ("usage_metrics", interval, hours), + compute, + ) - if "received proxy request" in message: - model_stats[model]["requests"] += 1 + def get_usage_dashboard( + self, + interval: int = 15, + hours: int = 24, + error_limit: int = 100, + model_limit: int = 20, + ) -> dict: + # Large ranges are expensive to scan; keep cached longer. + if hours <= 24: + cache_ttl = 60.0 + elif hours <= 7 * 24: + cache_ttl = 300.0 + elif hours <= 30 * 24: + cache_ttl = 1800.0 + elif hours <= 90 * 24: + cache_ttl = 7200.0 + else: + cache_ttl = 21600.0 - if ( - "completed for streaming" in message - or "completed for non-streaming" in message - ): - model_stats[model]["successful"] += 1 - cost_data = entry.get("cost_data") - if isinstance(cost_data, dict): - actual_cost = cost_data.get("total_msats", 0) - if isinstance(actual_cost, (int, float)) and actual_cost > 0: - model_stats[model]["revenue_msats"] += actual_cost + def compute() -> dict: + try: + return self._usage_store.get_dashboard( + interval_minutes=interval, + hours_back=hours, + error_limit=error_limit, + model_limit=model_limit, + ) + except Exception as e: + logger.error( + f"Usage analytics index failed, falling back to log scan: {e}" + ) + return self._aggregate_dashboard( + interval_minutes=interval, + hours_back=hours, + error_limit=error_limit, + model_limit=model_limit, + ) - if "revert payment" in message or "upstream request failed" in message: - model_stats[model]["failed"] += 1 - if "revert payment" in message: - max_cost = entry.get("max_cost_for_model", 0) - if isinstance(max_cost, (int, float)) and max_cost > 0: - model_stats[model]["refunds_msats"] += max_cost + return self._cache_call( + ("usage_dashboard", interval, hours, error_limit, model_limit), + compute, + ttl_seconds=cache_ttl, + ) - except Exception: - continue + def get_error_details(self, hours: int = 24, limit: int = 100) -> dict: + def compute() -> dict: + try: + return self._usage_store.get_error_details(hours_back=hours, limit=limit) + except Exception as e: + logger.error( + f"Usage analytics index failed, falling back to log scan: {e}" + ) - models: list[dict[str, Any]] = [] - total_revenue = 0.0 + errors: list[dict] = [] + for entry in self._get_cached_entries(hours): + if str(entry.get("levelname", "")).upper() == "ERROR": + timestamp_str = entry.get("asctime", "") + errors.append( + { + "timestamp": timestamp_str, + "message": entry.get("message", ""), + "error_type": entry.get("error_type", "unknown"), + "pathname": entry.get("pathname", ""), + "lineno": entry.get("lineno", 0), + "request_id": entry.get("request_id", ""), + } + ) - for model, stats in model_stats.items(): - revenue_msats = float(stats["revenue_msats"]) - refunds_msats = float(stats["refunds_msats"]) + errors.sort(key=lambda x: x["timestamp"], reverse=True) + return {"errors": errors[:limit], "total_count": len(errors)} - revenue_sats = revenue_msats / 1000 - refunds_sats = refunds_msats / 1000 - net_revenue_sats = revenue_sats - refunds_sats + return self._cache_call(("error_details", hours, limit), compute) - total_revenue += net_revenue_sats + def get_revenue_by_model(self, hours: int = 24, limit: int = 20) -> dict: + def compute() -> dict: + try: + return self._usage_store.get_revenue_by_model( + hours_back=hours, limit=limit + ) + except Exception as e: + logger.error( + f"Usage analytics index failed, falling back to log scan: {e}" + ) - requests = int(stats["requests"]) - successful = int(stats["successful"]) + entries = self._get_cached_entries(hours) - models.append( - { - "model": model, - "revenue_sats": revenue_sats, - "refunds_sats": refunds_sats, - "net_revenue_sats": net_revenue_sats, - "requests": requests, - "successful": successful, - "failed": int(stats["failed"]), - "avg_revenue_per_request": ( - revenue_sats / successful if successful > 0 else 0 - ), + model_stats: dict[str, dict[str, int | float]] = defaultdict( + lambda: { + "revenue_msats": 0, + "refunds_msats": 0, + "requests": 0, + "successful": 0, + "failed": 0, } ) - models.sort(key=lambda x: float(x["net_revenue_sats"]), reverse=True) + for entry in entries: + try: + model = entry.get("model", "unknown") + if not isinstance(model, str): + model = "unknown" - return { - "models": models[:limit], - "total_revenue_sats": total_revenue, - "total_models": len(models), - } + message = str(entry.get("message", "")).lower() - def _calculate_summary_stats(self, entries: list[dict]) -> dict: - stats: dict[str, Any] = { - "total_entries": 0, - "total_requests": 0, - "successful_chat_completions": 0, - "failed_requests": 0, - "total_errors": 0, - "total_warnings": 0, - "payment_processed": 0, - "upstream_errors": 0, - "unique_models": set(), - "error_types": defaultdict(int), - "revenue_msats": 0.0, - "refunds_msats": 0.0, - } + completed, revenue_msats, _, _ = self._extract_success_metrics( + entry, message + ) + if completed: + model_stats[model]["requests"] += 1 + model_stats[model]["successful"] += 1 + if revenue_msats > 0: + model_stats[model]["revenue_msats"] += revenue_msats - for entry in entries: - try: - stats["total_entries"] += 1 + failed = ( + "revert payment" in message + or "upstream request failed" in message + ) + if failed: + model_stats[model]["requests"] += 1 + model_stats[model]["failed"] += 1 + if "revert payment" in message: + max_cost = entry.get("max_cost_for_model", 0) + if isinstance(max_cost, (int, float)) and max_cost > 0: + model_stats[model]["refunds_msats"] += max_cost - message = entry.get("message", "").lower() - level = entry.get("levelname", "").upper() + except Exception: + continue - if level == "ERROR": - stats["total_errors"] += 1 - if "error_type" in entry: - stats["error_types"][str(entry["error_type"])] += 1 - elif level == "WARNING": - stats["total_warnings"] += 1 + models: list[dict[str, Any]] = [] + total_revenue = 0.0 - if "received proxy request" in message: - stats["total_requests"] += 1 + for model, stats in model_stats.items(): + revenue_msats = float(stats["revenue_msats"]) + refunds_msats = float(stats["refunds_msats"]) - if ( - "completed for streaming" in message - or "completed for non-streaming" in message - ): - stats["successful_chat_completions"] += 1 + revenue_sats = revenue_msats / 1000 + refunds_sats = refunds_msats / 1000 + net_revenue_sats = revenue_sats - refunds_sats - if "upstream request failed" in message or "revert payment" in message: - stats["failed_requests"] += 1 + total_revenue += net_revenue_sats - if "payment processed successfully" in message: - stats["payment_processed"] += 1 + requests = int(stats["requests"]) + successful = int(stats["successful"]) - if "upstream" in message and level == "ERROR": - stats["upstream_errors"] += 1 + models.append( + { + "model": model, + "revenue_sats": revenue_sats, + "refunds_sats": refunds_sats, + "net_revenue_sats": net_revenue_sats, + "requests": requests, + "successful": successful, + "failed": int(stats["failed"]), + "avg_revenue_per_request": ( + revenue_sats / successful if successful > 0 else 0 + ), + } + ) - if "model" in entry: - model = entry["model"] - if isinstance(model, str) and model != "unknown": - stats["unique_models"].add(model) + models.sort(key=lambda x: float(x["net_revenue_sats"]), reverse=True) - if ( - "completed for streaming" in message - or "completed for non-streaming" in message - ): - cost_data = entry.get("cost_data") - if isinstance(cost_data, dict): - actual_cost = cost_data.get("total_msats", 0) - if isinstance(actual_cost, (int, float)) and actual_cost > 0: - stats["revenue_msats"] += float(actual_cost) + return { + "models": models[:limit], + "total_revenue_sats": total_revenue, + "total_models": len(models), + } - if "revert payment" in message: - max_cost = entry.get("max_cost_for_model", 0) - if isinstance(max_cost, (int, float)) and max_cost > 0: - stats["refunds_msats"] += float(max_cost) - - except Exception: - continue + return self._cache_call(("revenue_by_model", hours, limit), compute) + def _build_summary_response(self, stats: dict[str, Any]) -> dict[str, Any]: revenue_sats = stats["revenue_msats"] / 1000 refunds_sats = stats["refunds_msats"] / 1000 net_revenue_sats = revenue_sats - refunds_sats total_requests = stats["total_requests"] successful = stats["successful_chat_completions"] + input_tokens = stats["input_tokens"] + output_tokens = stats["output_tokens"] + total_tokens = stats["total_tokens"] return { "total_entries": stats["total_entries"], @@ -430,6 +553,18 @@ class LogManager: "unique_models_count": len(stats["unique_models"]), "unique_models": sorted(list(stats["unique_models"])), "error_types": dict(stats["error_types"]), + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "total_tokens": total_tokens, + "avg_input_tokens_per_completion": ( + input_tokens / successful if successful > 0 else 0 + ), + "avg_output_tokens_per_completion": ( + output_tokens / successful if successful > 0 else 0 + ), + "avg_total_tokens_per_completion": ( + total_tokens / successful if successful > 0 else 0 + ), "success_rate": (successful / total_requests * 100) if total_requests > 0 else 0, @@ -449,62 +584,536 @@ class LogManager: ), } + def _calculate_summary_stats(self, entries: list[dict]) -> dict: + stats: dict[str, Any] = { + "total_entries": 0, + "total_requests": 0, + "successful_chat_completions": 0, + "failed_requests": 0, + "total_errors": 0, + "total_warnings": 0, + "payment_processed": 0, + "upstream_errors": 0, + "unique_models": set(), + "error_types": defaultdict(int), + "revenue_msats": 0.0, + "refunds_msats": 0.0, + "input_tokens": 0, + "output_tokens": 0, + "total_tokens": 0, + } + + for entry in entries: + try: + stats["total_entries"] += 1 + + message = str(entry.get("message", "")).lower() + level = str(entry.get("levelname", "")).upper() + + if level == "ERROR": + stats["total_errors"] += 1 + if "error_type" in entry: + stats["error_types"][str(entry["error_type"])] += 1 + elif level == "WARNING": + stats["total_warnings"] += 1 + + completed, revenue_msats, input_tokens, output_tokens = ( + self._extract_success_metrics(entry, message) + ) + if completed: + stats["total_requests"] += 1 + stats["successful_chat_completions"] += 1 + stats["input_tokens"] += input_tokens + stats["output_tokens"] += output_tokens + stats["total_tokens"] += input_tokens + output_tokens + + failed = ( + "upstream request failed" in message + or "revert payment" in message + ) + if failed: + stats["total_requests"] += 1 + stats["failed_requests"] += 1 + + if "payment processed successfully" in message: + stats["payment_processed"] += 1 + + if "upstream" in message and level == "ERROR": + stats["upstream_errors"] += 1 + + if "model" in entry: + model = entry["model"] + if isinstance(model, str) and model != "unknown": + stats["unique_models"].add(model) + + if completed and revenue_msats > 0: + stats["revenue_msats"] += revenue_msats + + if "revert payment" in message: + max_cost = entry.get("max_cost_for_model", 0) + if isinstance(max_cost, (int, float)) and max_cost > 0: + stats["refunds_msats"] += float(max_cost) + + except Exception: + continue + + return self._build_summary_response(stats) + + def _aggregate_dashboard( + self, + interval_minutes: int, + hours_back: int, + error_limit: int, + model_limit: int, + ) -> dict[str, Any]: + time_buckets: dict[str, dict[str, Any]] = defaultdict( + lambda: { + "total_requests": 0, + "successful_chat_completions": 0, + "failed_requests": 0, + "errors": 0, + "warnings": 0, + "payment_processed": 0, + "upstream_errors": 0, + "revenue_msats": 0.0, + "refunds_msats": 0.0, + "input_tokens": 0, + "output_tokens": 0, + "total_tokens": 0, + } + ) + summary_stats: dict[str, Any] = { + "total_entries": 0, + "total_requests": 0, + "successful_chat_completions": 0, + "failed_requests": 0, + "total_errors": 0, + "total_warnings": 0, + "payment_processed": 0, + "upstream_errors": 0, + "unique_models": set(), + "error_types": defaultdict(int), + "revenue_msats": 0.0, + "refunds_msats": 0.0, + "input_tokens": 0, + "output_tokens": 0, + "total_tokens": 0, + } + model_stats: dict[str, dict[str, int | float]] = defaultdict( + lambda: { + "revenue_msats": 0, + "refunds_msats": 0, + "requests": 0, + "successful": 0, + "failed": 0, + } + ) + model_mix_buckets: dict[str, dict[str, int]] = defaultdict( + lambda: defaultdict(int) + ) + model_mix_revenue_buckets: dict[str, dict[str, float]] = defaultdict( + lambda: defaultdict(float) + ) + model_mix_token_buckets: dict[str, dict[str, int]] = defaultdict( + lambda: defaultdict(int) + ) + model_mix_totals: dict[str, int] = defaultdict(int) + model_mix_revenue_totals: dict[str, float] = defaultdict(float) + model_mix_token_totals: dict[str, int] = defaultdict(int) + latest_errors_heap: list[tuple[str, dict[str, Any]]] = [] + total_error_count = 0 + + for entry in self._yield_log_entries(hours_back=hours_back): + try: + summary_stats["total_entries"] += 1 + + timestamp_str = entry.get("asctime", "") + message = str(entry.get("message", "")).lower() + level = str(entry.get("levelname", "")).upper() + model = entry.get("model", "unknown") + if not isinstance(model, str): + model = "unknown" + + bucket_key = ( + self._bucket_key_for_timestamp(timestamp_str, interval_minutes) + if isinstance(timestamp_str, str) + else None + ) + bucket = time_buckets[bucket_key] if bucket_key else None + + if level == "ERROR": + summary_stats["total_errors"] += 1 + if bucket: + bucket["errors"] += 1 + if "error_type" in entry: + summary_stats["error_types"][str(entry["error_type"])] += 1 + + total_error_count += 1 + error_item = { + "timestamp": timestamp_str, + "message": entry.get("message", ""), + "error_type": entry.get("error_type", "unknown"), + "pathname": entry.get("pathname", ""), + "lineno": entry.get("lineno", 0), + "request_id": entry.get("request_id", ""), + } + if len(latest_errors_heap) < error_limit: + heappush(latest_errors_heap, (timestamp_str, error_item)) + elif timestamp_str > latest_errors_heap[0][0]: + heapreplace(latest_errors_heap, (timestamp_str, error_item)) + elif level == "WARNING": + summary_stats["total_warnings"] += 1 + if bucket: + bucket["warnings"] += 1 + + completed, revenue_msats, input_tokens, output_tokens = ( + self._extract_success_metrics(entry, message) + ) + if completed: + summary_stats["total_requests"] += 1 + summary_stats["successful_chat_completions"] += 1 + summary_stats["input_tokens"] += input_tokens + summary_stats["output_tokens"] += output_tokens + summary_stats["total_tokens"] += input_tokens + output_tokens + model_stats[model]["requests"] += 1 + model_stats[model]["successful"] += 1 + model_mix_totals[model] += 1 + if bucket: + bucket["total_requests"] += 1 + bucket["successful_chat_completions"] += 1 + bucket["input_tokens"] += input_tokens + bucket["output_tokens"] += output_tokens + bucket["total_tokens"] += input_tokens + output_tokens + if bucket_key: + model_mix_buckets[bucket_key][model] += 1 + if revenue_msats > 0: + model_mix_revenue_buckets[bucket_key][model] += revenue_msats + model_mix_revenue_totals[model] += revenue_msats + if input_tokens > 0 or output_tokens > 0: + token_total = input_tokens + output_tokens + model_mix_token_buckets[bucket_key][model] += token_total + model_mix_token_totals[model] += token_total + + if revenue_msats > 0: + summary_stats["revenue_msats"] += revenue_msats + model_stats[model]["revenue_msats"] += revenue_msats + if bucket: + bucket["revenue_msats"] += revenue_msats + + failed = ( + "upstream request failed" in message + or "revert payment" in message + ) + if failed: + summary_stats["total_requests"] += 1 + summary_stats["failed_requests"] += 1 + model_stats[model]["requests"] += 1 + model_stats[model]["failed"] += 1 + if bucket: + bucket["total_requests"] += 1 + bucket["failed_requests"] += 1 + + if "payment processed successfully" in message: + summary_stats["payment_processed"] += 1 + if bucket: + bucket["payment_processed"] += 1 + + if "upstream" in message and level == "ERROR": + summary_stats["upstream_errors"] += 1 + if bucket: + bucket["upstream_errors"] += 1 + + if model != "unknown": + summary_stats["unique_models"].add(model) + + if "revert payment" in message: + max_cost = entry.get("max_cost_for_model", 0) + if isinstance(max_cost, (int, float)) and max_cost > 0: + max_cost_float = float(max_cost) + summary_stats["refunds_msats"] += max_cost_float + model_stats[model]["refunds_msats"] += max_cost_float + if bucket: + bucket["refunds_msats"] += max_cost_float + except Exception: + continue + + metrics_result = [] + for bucket_key in sorted(time_buckets.keys()): + bucket = dict(time_buckets[bucket_key]) + bucket["requests"] = bucket["total_requests"] + metrics_result.append({"timestamp": bucket_key, **bucket}) + + models: list[dict[str, Any]] = [] + total_revenue = 0.0 + for model_name, stats in model_stats.items(): + revenue_msats = float(stats["revenue_msats"]) + refunds_msats = float(stats["refunds_msats"]) + revenue_sats = revenue_msats / 1000 + refunds_sats = refunds_msats / 1000 + net_revenue_sats = revenue_sats - refunds_sats + total_revenue += net_revenue_sats + + successful = int(stats["successful"]) + models.append( + { + "model": model_name, + "revenue_sats": revenue_sats, + "refunds_sats": refunds_sats, + "net_revenue_sats": net_revenue_sats, + "requests": int(stats["requests"]), + "successful": successful, + "failed": int(stats["failed"]), + "avg_revenue_per_request": ( + revenue_sats / successful if successful > 0 else 0 + ), + } + ) + + models.sort(key=lambda x: float(x["net_revenue_sats"]), reverse=True) + latest_errors = [ + item + for _, item in sorted( + latest_errors_heap, key=lambda x: x[0], reverse=True + ) + ] + top_model_limit = max(1, min(model_limit, 20)) + top_models_requests = [ + model_name + for model_name, _ in sorted( + ( + (name, count) + for name, count in model_mix_totals.items() + if name != "unknown" + ), + key=lambda item: item[1], + reverse=True, + )[:top_model_limit] + ] + top_models_revenue = [ + model_name + for model_name, _ in sorted( + ( + (name, amount) + for name, amount in model_mix_revenue_totals.items() + if name != "unknown" + ), + key=lambda item: item[1], + reverse=True, + )[:top_model_limit] + ] + top_models_tokens = [ + model_name + for model_name, _ in sorted( + ( + (name, token_count) + for name, token_count in model_mix_token_totals.items() + if name != "unknown" + ), + key=lambda item: item[1], + reverse=True, + )[:top_model_limit] + ] + selected_models: list[str] = [] + for model in top_models_requests + top_models_revenue + top_models_tokens: + if model not in selected_models: + selected_models.append(model) + top_model_set = set(selected_models) + + model_usage_mix_metrics: list[dict[str, Any]] = [] + mix_bucket_keys = sorted( + set(model_mix_buckets.keys()) + | set(model_mix_revenue_buckets.keys()) + | set(model_mix_token_buckets.keys()) + ) + for bucket_key in mix_bucket_keys: + counts = model_mix_buckets.get(bucket_key, {}) + revenue_counts = model_mix_revenue_buckets.get(bucket_key, {}) + token_counts = model_mix_token_buckets.get(bucket_key, {}) + others = 0 + others_revenue_msats = 0.0 + others_tokens = 0 + model_counts: dict[str, int] = {} + model_revenue_msats: dict[str, float] = {} + model_tokens: dict[str, int] = {} + for model_name, successful_count in counts.items(): + if model_name in top_model_set: + model_counts[model_name] = int(successful_count) + else: + others += int(successful_count) + for model_name, revenue_value in revenue_counts.items(): + if model_name in top_model_set: + model_revenue_msats[model_name] = float(revenue_value) + else: + others_revenue_msats += float(revenue_value) + for model_name, token_value in token_counts.items(): + if model_name in top_model_set: + model_tokens[model_name] = int(token_value) + else: + others_tokens += int(token_value) + + model_usage_mix_metrics.append( + { + "timestamp": bucket_key, + "total_successful": int(sum(counts.values())), + "total_revenue_msats": float(sum(revenue_counts.values())), + "total_tokens": int(sum(token_counts.values())), + "others": others, + "others_revenue_msats": others_revenue_msats, + "others_tokens": others_tokens, + "model_counts": model_counts, + "model_revenue_msats": model_revenue_msats, + "model_tokens": model_tokens, + } + ) + + return { + "metrics": { + "metrics": metrics_result, + "interval_minutes": interval_minutes, + "hours_back": hours_back, + "total_buckets": len(metrics_result), + }, + "summary": self._build_summary_response(summary_stats), + "error_details": { + "errors": latest_errors, + "total_count": total_error_count, + }, + "revenue_by_model": { + "models": models[:model_limit], + "total_revenue_sats": total_revenue, + "total_models": len(models), + }, + "model_usage_mix": { + "top_models": top_models_requests, + "top_models_by_metric": { + "requests": top_models_requests, + "revenue": top_models_revenue, + "tokens": top_models_tokens, + }, + "metrics": model_usage_mix_metrics, + "interval_minutes": interval_minutes, + "hours_back": hours_back, + "total_buckets": len(model_usage_mix_metrics), + }, + } + def _aggregate_metrics_by_time( self, entries: list[dict], interval_minutes: int, hours_back: int ) -> dict: time_buckets: dict[str, dict[str, Any]] = defaultdict( - lambda: {"requests": 0, "errors": 0, "revenue_msats": 0.0} + lambda: { + "total_requests": 0, + "successful_chat_completions": 0, + "failed_requests": 0, + "errors": 0, + "warnings": 0, + "payment_processed": 0, + "upstream_errors": 0, + "revenue_msats": 0.0, + "refunds_msats": 0.0, + "input_tokens": 0, + "output_tokens": 0, + "total_tokens": 0, + } ) for entry in entries: try: timestamp_str = entry.get("asctime", "") - if not timestamp_str: + if not isinstance(timestamp_str, str): continue - - log_time = datetime.strptime(timestamp_str, "%Y-%m-%d %H:%M:%S") - log_time = log_time.replace(tzinfo=timezone.utc) - - # Round down to nearest interval - minutes = log_time.minute - rounded_minutes = (minutes // interval_minutes) * interval_minutes - bucket_time = log_time.replace( - minute=rounded_minutes, second=0, microsecond=0 + bucket_key = self._bucket_key_for_timestamp( + timestamp_str, interval_minutes ) - bucket_key = bucket_time.strftime("%Y-%m-%d %H:%M:%S") + if not bucket_key: + continue bucket = time_buckets[bucket_key] - message = entry.get("message", "").lower() - level = entry.get("levelname", "").upper() + message = str(entry.get("message", "")).lower() + level = str(entry.get("levelname", "")).upper() - if "received proxy request" in message: - bucket["requests"] += 1 + completed, revenue_msats, input_tokens, output_tokens = ( + self._extract_success_metrics(entry, message) + ) + if completed: + bucket["total_requests"] += 1 + bucket["successful_chat_completions"] += 1 + bucket["input_tokens"] += input_tokens + bucket["output_tokens"] += output_tokens + bucket["total_tokens"] += input_tokens + output_tokens if level == "ERROR": bucket["errors"] += 1 + if "upstream" in message: + bucket["upstream_errors"] += 1 + elif level == "WARNING": + bucket["warnings"] += 1 - if ( - "completed for streaming" in message - or "completed for non-streaming" in message - ): - cost_data = entry.get("cost_data") - if isinstance(cost_data, dict): - actual_cost = cost_data.get("total_msats", 0) - if isinstance(actual_cost, (int, float)) and actual_cost > 0: - bucket["revenue_msats"] += float(actual_cost) + failed = ( + "upstream request failed" in message + or "revert payment" in message + ) + if failed: + bucket["total_requests"] += 1 + bucket["failed_requests"] += 1 + + if "payment processed successfully" in message: + bucket["payment_processed"] += 1 + + if completed and revenue_msats > 0: + bucket["revenue_msats"] += revenue_msats + + if "revert payment" in message: + max_cost = entry.get("max_cost_for_model", 0) + if isinstance(max_cost, (int, float)) and max_cost > 0: + bucket["refunds_msats"] += float(max_cost) except Exception: continue result = [] for bucket_key in sorted(time_buckets.keys()): - result.append({"timestamp": bucket_key, **time_buckets[bucket_key]}) + bucket = dict(time_buckets[bucket_key]) + # Backward-compatible alias for any callers still reading "requests". + bucket["requests"] = bucket["total_requests"] + result.append({"timestamp": bucket_key, **bucket}) + + totals = { + "total_requests": 0, + "successful_chat_completions": 0, + "failed_requests": 0, + "errors": 0, + "warnings": 0, + "payment_processed": 0, + "upstream_errors": 0, + "revenue_msats": 0.0, + "refunds_msats": 0.0, + "input_tokens": 0, + "output_tokens": 0, + "total_tokens": 0, + } + for bucket in result: + totals["total_requests"] += int(bucket["total_requests"]) + totals["successful_chat_completions"] += int( + bucket["successful_chat_completions"] + ) + totals["failed_requests"] += int(bucket["failed_requests"]) + totals["errors"] += int(bucket["errors"]) + totals["warnings"] += int(bucket["warnings"]) + totals["payment_processed"] += int(bucket["payment_processed"]) + totals["upstream_errors"] += int(bucket["upstream_errors"]) + totals["revenue_msats"] += float(bucket["revenue_msats"]) + totals["refunds_msats"] += float(bucket["refunds_msats"]) + totals["input_tokens"] += int(bucket["input_tokens"]) + totals["output_tokens"] += int(bucket["output_tokens"]) + totals["total_tokens"] += int(bucket["total_tokens"]) return { "metrics": result, "interval_minutes": interval_minutes, "hours_back": hours_back, "total_buckets": len(result), + "totals": totals, } diff --git a/routstr/core/usage_analytics_store.py b/routstr/core/usage_analytics_store.py new file mode 100644 index 00000000..7ba90e24 --- /dev/null +++ b/routstr/core/usage_analytics_store.py @@ -0,0 +1,1380 @@ +import json +import sqlite3 +import time +from collections import defaultdict +from datetime import datetime, timedelta, timezone +from pathlib import Path +from threading import Lock +from typing import Any + +from .logging import get_logger + +logger = get_logger(__name__) + + +class UsageAnalyticsStore: + """ + Incremental usage analytics index backed by SQLite. + + Instead of rescanning raw JSON log files for every dashboard request, we keep + a rolling minute-level aggregate that is updated from only newly appended log + bytes. + """ + + SCHEMA_VERSION = "4" + + def __init__(self, logs_dir: Path, db_path: Path | None = None): + self.logs_dir = logs_dir + self.db_path = db_path or (logs_dir / "usage_analytics.db") + self._lock = Lock() + self._conn: sqlite3.Connection | None = None + + def get_dashboard( + self, + *, + interval_minutes: int, + hours_back: int, + error_limit: int, + model_limit: int, + ) -> dict[str, Any]: + with self._lock: + conn = self._get_connection_locked() + self._ensure_up_to_date_locked(conn) + cutoff_timestamp = self._cutoff_timestamp(hours_back) + + summary = self._query_summary_locked(conn, cutoff_timestamp) + metrics = self._query_metrics_locked( + conn, + cutoff_timestamp=cutoff_timestamp, + interval_minutes=interval_minutes, + hours_back=hours_back, + ) + error_details = self._query_error_details_locked( + conn, + cutoff_timestamp=cutoff_timestamp, + limit=error_limit, + total_error_count=summary["total_errors"], + ) + revenue_by_model = self._query_revenue_by_model_locked( + conn, + cutoff_timestamp=cutoff_timestamp, + limit=model_limit, + ) + model_usage_mix = self._query_model_usage_mix_locked( + conn, + cutoff_timestamp=cutoff_timestamp, + interval_minutes=interval_minutes, + hours_back=hours_back, + limit=model_limit, + ) + + return { + "metrics": metrics, + "summary": summary, + "error_details": error_details, + "revenue_by_model": revenue_by_model, + "model_usage_mix": model_usage_mix, + } + + def get_summary(self, *, hours_back: int) -> dict[str, Any]: + with self._lock: + conn = self._get_connection_locked() + self._ensure_up_to_date_locked(conn) + cutoff_timestamp = self._cutoff_timestamp(hours_back) + return self._query_summary_locked(conn, cutoff_timestamp) + + def get_metrics( + self, + *, + interval_minutes: int, + hours_back: int, + ) -> dict[str, Any]: + with self._lock: + conn = self._get_connection_locked() + self._ensure_up_to_date_locked(conn) + cutoff_timestamp = self._cutoff_timestamp(hours_back) + return self._query_metrics_locked( + conn, + cutoff_timestamp=cutoff_timestamp, + interval_minutes=interval_minutes, + hours_back=hours_back, + ) + + def get_error_details(self, *, hours_back: int, limit: int) -> dict[str, Any]: + with self._lock: + conn = self._get_connection_locked() + self._ensure_up_to_date_locked(conn) + cutoff_timestamp = self._cutoff_timestamp(hours_back) + return self._query_error_details_locked( + conn, + cutoff_timestamp=cutoff_timestamp, + limit=limit, + ) + + def get_revenue_by_model(self, *, hours_back: int, limit: int) -> dict[str, Any]: + with self._lock: + conn = self._get_connection_locked() + self._ensure_up_to_date_locked(conn) + cutoff_timestamp = self._cutoff_timestamp(hours_back) + return self._query_revenue_by_model_locked( + conn, + cutoff_timestamp=cutoff_timestamp, + limit=limit, + ) + + def _get_connection_locked(self) -> sqlite3.Connection: + if self._conn is not None: + return self._conn + + self.db_path.parent.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect( + self.db_path, + timeout=30.0, + check_same_thread=False, + ) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA journal_mode=WAL") + conn.execute("PRAGMA synchronous=NORMAL") + conn.execute("PRAGMA temp_store=MEMORY") + conn.execute("PRAGMA cache_size=-20000") + self._initialize_schema_locked(conn) + self._conn = conn + return conn + + def _initialize_schema_locked(self, conn: sqlite3.Connection) -> None: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS analytics_meta ( + key TEXT PRIMARY KEY, + value TEXT NOT NULL + ) + """ + ) + + current_version_row = conn.execute( + "SELECT value FROM analytics_meta WHERE key = 'schema_version'" + ).fetchone() + current_version = current_version_row[0] if current_version_row else None + + conn.execute( + """ + CREATE TABLE IF NOT EXISTS analytics_file_state ( + path TEXT PRIMARY KEY, + inode INTEGER NOT NULL, + offset INTEGER NOT NULL, + size INTEGER NOT NULL, + updated_at REAL NOT NULL + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS analytics_minute ( + minute_ts TEXT PRIMARY KEY, + total_entries INTEGER NOT NULL DEFAULT 0, + total_requests INTEGER NOT NULL DEFAULT 0, + successful_chat_completions INTEGER NOT NULL DEFAULT 0, + failed_requests INTEGER NOT NULL DEFAULT 0, + errors INTEGER NOT NULL DEFAULT 0, + warnings INTEGER NOT NULL DEFAULT 0, + payment_processed INTEGER NOT NULL DEFAULT 0, + upstream_errors INTEGER NOT NULL DEFAULT 0, + revenue_msats REAL NOT NULL DEFAULT 0, + refunds_msats REAL NOT NULL DEFAULT 0, + input_tokens INTEGER NOT NULL DEFAULT 0, + output_tokens INTEGER NOT NULL DEFAULT 0, + total_tokens INTEGER NOT NULL DEFAULT 0 + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS analytics_model_minute ( + minute_ts TEXT NOT NULL, + model TEXT NOT NULL, + requests INTEGER NOT NULL DEFAULT 0, + successful INTEGER NOT NULL DEFAULT 0, + failed INTEGER NOT NULL DEFAULT 0, + revenue_msats REAL NOT NULL DEFAULT 0, + refunds_msats REAL NOT NULL DEFAULT 0, + input_tokens INTEGER NOT NULL DEFAULT 0, + output_tokens INTEGER NOT NULL DEFAULT 0, + total_tokens INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (minute_ts, model) + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS analytics_model_presence_minute ( + minute_ts TEXT NOT NULL, + model TEXT NOT NULL, + count INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (minute_ts, model) + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS analytics_error_type_minute ( + minute_ts TEXT NOT NULL, + error_type TEXT NOT NULL, + count INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (minute_ts, error_type) + ) + """ + ) + conn.execute( + """ + CREATE TABLE IF NOT EXISTS analytics_error_events ( + timestamp TEXT NOT NULL, + message TEXT NOT NULL, + error_type TEXT NOT NULL, + pathname TEXT NOT NULL, + lineno INTEGER NOT NULL, + request_id TEXT NOT NULL + ) + """ + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_analytics_model_minute_ts ON analytics_model_minute (minute_ts)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_analytics_model_minute_model_ts ON analytics_model_minute (model, minute_ts)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_analytics_model_presence_ts ON analytics_model_presence_minute (minute_ts)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_analytics_error_type_minute_ts ON analytics_error_type_minute (minute_ts)" + ) + conn.execute( + "CREATE INDEX IF NOT EXISTS idx_analytics_error_events_ts ON analytics_error_events (timestamp DESC)" + ) + self._migrate_schema_locked(conn) + if current_version != self.SCHEMA_VERSION: + conn.execute( + """ + INSERT OR REPLACE INTO analytics_meta (key, value) + VALUES ('schema_version', ?) + """, + (self.SCHEMA_VERSION,), + ) + conn.commit() + + def _migrate_schema_locked(self, conn: sqlite3.Connection) -> None: + self._ensure_column_locked( + conn, + "analytics_minute", + "input_tokens", + "INTEGER NOT NULL DEFAULT 0", + ) + self._ensure_column_locked( + conn, + "analytics_minute", + "output_tokens", + "INTEGER NOT NULL DEFAULT 0", + ) + self._ensure_column_locked( + conn, + "analytics_minute", + "total_tokens", + "INTEGER NOT NULL DEFAULT 0", + ) + self._ensure_column_locked( + conn, + "analytics_model_minute", + "input_tokens", + "INTEGER NOT NULL DEFAULT 0", + ) + self._ensure_column_locked( + conn, + "analytics_model_minute", + "output_tokens", + "INTEGER NOT NULL DEFAULT 0", + ) + self._ensure_column_locked( + conn, + "analytics_model_minute", + "total_tokens", + "INTEGER NOT NULL DEFAULT 0", + ) + + def _ensure_column_locked( + self, + conn: sqlite3.Connection, + table: str, + column: str, + column_definition: str, + ) -> None: + existing_columns = { + str(row["name"]) + for row in conn.execute(f"PRAGMA table_info({table})").fetchall() + } + if column in existing_columns: + return + + conn.execute( + f"ALTER TABLE {table} ADD COLUMN {column} {column_definition}" + ) + logger.info(f"Migrated analytics schema: added {table}.{column}") + + def _drop_index_tables_locked(self, conn: sqlite3.Connection) -> None: + conn.execute("DROP TABLE IF EXISTS analytics_file_state") + conn.execute("DROP TABLE IF EXISTS analytics_minute") + conn.execute("DROP TABLE IF EXISTS analytics_model_minute") + conn.execute("DROP TABLE IF EXISTS analytics_model_presence_minute") + conn.execute("DROP TABLE IF EXISTS analytics_error_type_minute") + conn.execute("DROP TABLE IF EXISTS analytics_error_events") + + def _ensure_up_to_date_locked(self, conn: sqlite3.Connection) -> None: + if not self.logs_dir.exists(): + return + + log_files = sorted(self.logs_dir.glob("app_*.log")) + if not log_files: + return + + requires_rebuild = False + + for log_file in log_files: + try: + self._process_log_file_locked(conn, log_file) + except RuntimeError: + requires_rebuild = True + break + except Exception as exc: + logger.error(f"Failed indexing usage analytics for {log_file}: {exc}") + continue + + if requires_rebuild: + logger.warning( + "Usage analytics index out-of-sync, rebuilding from all log files" + ) + self._rebuild_locked(conn, log_files) + return + + # Commit even when only file-state metadata changed + # (for example when we intentionally keep offset at the last full line). + conn.commit() + + def _rebuild_locked( + self, conn: sqlite3.Connection, log_files: list[Path] | None = None + ) -> None: + self._drop_index_tables_locked(conn) + self._initialize_schema_locked(conn) + + files = log_files if log_files is not None else sorted(self.logs_dir.glob("app_*.log")) + for log_file in files: + try: + self._process_log_file_locked(conn, log_file, force_full_read=True) + except Exception as exc: + logger.error(f"Failed rebuilding usage analytics for {log_file}: {exc}") + conn.commit() + + def _process_log_file_locked( + self, + conn: sqlite3.Connection, + log_file: Path, + force_full_read: bool = False, + ) -> bool: + stat = log_file.stat() + inode = int(getattr(stat, "st_ino", 0)) + file_size = int(stat.st_size) + log_file_path = str(log_file.resolve()) + + previous_offset = 0 + if not force_full_read: + row = conn.execute( + """ + SELECT inode, offset + FROM analytics_file_state + WHERE path = ? + """, + (log_file_path,), + ).fetchone() + if row is not None: + previous_inode = int(row["inode"]) + previous_offset = int(row["offset"]) + if previous_inode and inode and previous_inode != inode: + raise RuntimeError("inode changed") + if previous_offset > file_size: + raise RuntimeError("file shrunk") + + if previous_offset >= file_size and not force_full_read: + self._upsert_file_state_locked( + conn, + path=log_file_path, + inode=inode, + offset=file_size, + size=file_size, + ) + return False + + ( + end_offset, + minute_updates, + model_updates, + model_presence_updates, + error_type_updates, + error_events, + ) = self._collect_updates_from_file(log_file, previous_offset) + + self._apply_updates_locked( + conn=conn, + minute_updates=minute_updates, + model_updates=model_updates, + model_presence_updates=model_presence_updates, + error_type_updates=error_type_updates, + error_events=error_events, + ) + + latest_size = int(log_file.stat().st_size) + self._upsert_file_state_locked( + conn, + path=log_file_path, + inode=inode, + offset=end_offset, + size=latest_size, + ) + return end_offset != previous_offset + + def _upsert_file_state_locked( + self, + conn: sqlite3.Connection, + *, + path: str, + inode: int, + offset: int, + size: int, + ) -> None: + conn.execute( + """ + INSERT INTO analytics_file_state (path, inode, offset, size, updated_at) + VALUES (?, ?, ?, ?, ?) + ON CONFLICT(path) DO UPDATE SET + inode = excluded.inode, + offset = excluded.offset, + size = excluded.size, + updated_at = excluded.updated_at + """, + (path, inode, offset, size, time.time()), + ) + + def _collect_updates_from_file( + self, log_file: Path, start_offset: int + ) -> tuple[ + int, + dict[str, dict[str, float]], + dict[tuple[str, str], dict[str, float]], + dict[tuple[str, str], int], + dict[tuple[str, str], int], + list[tuple[str, str, str, str, int, str]], + ]: + minute_updates: dict[str, dict[str, float]] = defaultdict( + self._new_minute_stats + ) + model_updates: dict[tuple[str, str], dict[str, float]] = defaultdict( + self._new_model_stats + ) + model_presence_updates: dict[tuple[str, str], int] = defaultdict(int) + error_type_updates: dict[tuple[str, str], int] = defaultdict(int) + error_events: list[tuple[str, str, str, str, int, str]] = [] + + end_offset = start_offset + with open(log_file, "rb") as f: + f.seek(start_offset) + + while True: + line_start = f.tell() + raw_line = f.readline() + if not raw_line: + break + + # If the writer is appending and we catch a partial line at EOF, + # do not advance beyond it. We'll parse it on the next refresh. + if not raw_line.endswith(b"\n"): + f.seek(line_start) + break + + end_offset = f.tell() + if not raw_line.strip(): + continue + + try: + entry = json.loads(raw_line) + except Exception: + continue + + if not isinstance(entry, dict): + continue + + minute_key = self._minute_key(entry.get("asctime")) + if minute_key is None: + continue + + bucket = minute_updates[minute_key] + bucket["total_entries"] += 1 + + message_value = entry.get("message", "") + message = str(message_value).lower() + level = str(entry.get("levelname", "")).upper() + + model_raw = entry.get("model", "unknown") + model = model_raw if isinstance(model_raw, str) else "unknown" + + if level == "ERROR": + bucket["errors"] += 1 + error_type = str(entry.get("error_type", "unknown")) + error_type_updates[(minute_key, error_type)] += 1 + + lineno_value = entry.get("lineno", 0) + try: + lineno = int(lineno_value) + except (TypeError, ValueError): + lineno = 0 + + error_events.append( + ( + str(entry.get("asctime", "")), + str(message_value), + error_type, + str(entry.get("pathname", "")), + lineno, + str(entry.get("request_id", "")), + ) + ) + elif level == "WARNING": + bucket["warnings"] += 1 + + completed, revenue_msats, input_tokens, output_tokens = ( + self._extract_success_metrics(entry, message) + ) + if completed: + bucket["total_requests"] += 1 + bucket["successful_chat_completions"] += 1 + model_bucket = model_updates[(minute_key, model)] + model_bucket["requests"] += 1 + model_bucket["successful"] += 1 + bucket["input_tokens"] += input_tokens + bucket["output_tokens"] += output_tokens + bucket["total_tokens"] += input_tokens + output_tokens + model_bucket["input_tokens"] += input_tokens + model_bucket["output_tokens"] += output_tokens + model_bucket["total_tokens"] += input_tokens + output_tokens + + if revenue_msats > 0: + bucket["revenue_msats"] += revenue_msats + model_bucket["revenue_msats"] += revenue_msats + + failed = ( + "upstream request failed" in message + or "revert payment" in message + ) + if failed: + bucket["total_requests"] += 1 + bucket["failed_requests"] += 1 + model_bucket = model_updates[(minute_key, model)] + model_bucket["requests"] += 1 + model_bucket["failed"] += 1 + + if "payment processed successfully" in message: + bucket["payment_processed"] += 1 + + if level == "ERROR" and "upstream" in message: + bucket["upstream_errors"] += 1 + + if model != "unknown": + model_presence_updates[(minute_key, model)] += 1 + + if "revert payment" in message: + max_cost = entry.get("max_cost_for_model", 0) + if isinstance(max_cost, (int, float)) and max_cost > 0: + max_cost_float = float(max_cost) + bucket["refunds_msats"] += max_cost_float + model_updates[(minute_key, model)][ + "refunds_msats" + ] += max_cost_float + + return ( + end_offset, + minute_updates, + model_updates, + model_presence_updates, + error_type_updates, + error_events, + ) + + def _apply_updates_locked( + self, + *, + conn: sqlite3.Connection, + minute_updates: dict[str, dict[str, float]], + model_updates: dict[tuple[str, str], dict[str, float]], + model_presence_updates: dict[tuple[str, str], int], + error_type_updates: dict[tuple[str, str], int], + error_events: list[tuple[str, str, str, str, int, str]], + ) -> None: + if minute_updates: + rows = [ + ( + minute_ts, + int(stats["total_entries"]), + int(stats["total_requests"]), + int(stats["successful_chat_completions"]), + int(stats["failed_requests"]), + int(stats["errors"]), + int(stats["warnings"]), + int(stats["payment_processed"]), + int(stats["upstream_errors"]), + float(stats["revenue_msats"]), + float(stats["refunds_msats"]), + int(stats["input_tokens"]), + int(stats["output_tokens"]), + int(stats["total_tokens"]), + ) + for minute_ts, stats in minute_updates.items() + ] + conn.executemany( + """ + INSERT INTO analytics_minute ( + minute_ts, + total_entries, + total_requests, + successful_chat_completions, + failed_requests, + errors, + warnings, + payment_processed, + upstream_errors, + revenue_msats, + refunds_msats, + input_tokens, + output_tokens, + total_tokens + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(minute_ts) DO UPDATE SET + total_entries = total_entries + excluded.total_entries, + total_requests = total_requests + excluded.total_requests, + successful_chat_completions = successful_chat_completions + excluded.successful_chat_completions, + failed_requests = failed_requests + excluded.failed_requests, + errors = errors + excluded.errors, + warnings = warnings + excluded.warnings, + payment_processed = payment_processed + excluded.payment_processed, + upstream_errors = upstream_errors + excluded.upstream_errors, + revenue_msats = revenue_msats + excluded.revenue_msats, + refunds_msats = refunds_msats + excluded.refunds_msats, + input_tokens = input_tokens + excluded.input_tokens, + output_tokens = output_tokens + excluded.output_tokens, + total_tokens = total_tokens + excluded.total_tokens + """, + rows, + ) + + if model_updates: + model_rows = [ + ( + minute_ts, + model, + int(stats["requests"]), + int(stats["successful"]), + int(stats["failed"]), + float(stats["revenue_msats"]), + float(stats["refunds_msats"]), + int(stats["input_tokens"]), + int(stats["output_tokens"]), + int(stats["total_tokens"]), + ) + for (minute_ts, model), stats in model_updates.items() + ] + conn.executemany( + """ + INSERT INTO analytics_model_minute ( + minute_ts, + model, + requests, + successful, + failed, + revenue_msats, + refunds_msats, + input_tokens, + output_tokens, + total_tokens + ) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(minute_ts, model) DO UPDATE SET + requests = requests + excluded.requests, + successful = successful + excluded.successful, + failed = failed + excluded.failed, + revenue_msats = revenue_msats + excluded.revenue_msats, + refunds_msats = refunds_msats + excluded.refunds_msats, + input_tokens = input_tokens + excluded.input_tokens, + output_tokens = output_tokens + excluded.output_tokens, + total_tokens = total_tokens + excluded.total_tokens + """, + model_rows, + ) + + if model_presence_updates: + presence_rows = [ + (minute_ts, model, count) + for (minute_ts, model), count in model_presence_updates.items() + ] + conn.executemany( + """ + INSERT INTO analytics_model_presence_minute ( + minute_ts, + model, + count + ) + VALUES (?, ?, ?) + ON CONFLICT(minute_ts, model) DO UPDATE SET + count = count + excluded.count + """, + presence_rows, + ) + + if error_type_updates: + error_type_rows = [ + (minute_ts, error_type, count) + for (minute_ts, error_type), count in error_type_updates.items() + ] + conn.executemany( + """ + INSERT INTO analytics_error_type_minute ( + minute_ts, + error_type, + count + ) + VALUES (?, ?, ?) + ON CONFLICT(minute_ts, error_type) DO UPDATE SET + count = count + excluded.count + """, + error_type_rows, + ) + + if error_events: + conn.executemany( + """ + INSERT INTO analytics_error_events ( + timestamp, + message, + error_type, + pathname, + lineno, + request_id + ) + VALUES (?, ?, ?, ?, ?, ?) + """, + error_events, + ) + + def _query_metrics_locked( + self, + conn: sqlite3.Connection, + *, + cutoff_timestamp: str, + interval_minutes: int, + hours_back: int, + ) -> dict[str, Any]: + bucket_seconds = max(60, int(interval_minutes) * 60) + rows = conn.execute( + """ + SELECT + datetime( + (CAST(strftime('%s', minute_ts) AS INTEGER) / ?) * ?, + 'unixepoch' + ) AS bucket_ts, + COALESCE(SUM(total_requests), 0) AS total_requests, + COALESCE(SUM(successful_chat_completions), 0) AS successful_chat_completions, + COALESCE(SUM(failed_requests), 0) AS failed_requests, + COALESCE(SUM(errors), 0) AS errors, + COALESCE(SUM(warnings), 0) AS warnings, + COALESCE(SUM(payment_processed), 0) AS payment_processed, + COALESCE(SUM(upstream_errors), 0) AS upstream_errors, + COALESCE(SUM(revenue_msats), 0) AS revenue_msats, + COALESCE(SUM(refunds_msats), 0) AS refunds_msats, + COALESCE(SUM(input_tokens), 0) AS input_tokens, + COALESCE(SUM(output_tokens), 0) AS output_tokens, + COALESCE(SUM(total_tokens), 0) AS total_tokens + FROM analytics_minute + WHERE minute_ts >= ? + GROUP BY bucket_ts + ORDER BY bucket_ts + """, + (bucket_seconds, bucket_seconds, cutoff_timestamp), + ).fetchall() + + totals: dict[str, float] = { + "total_requests": 0.0, + "successful_chat_completions": 0.0, + "failed_requests": 0.0, + "errors": 0.0, + "warnings": 0.0, + "payment_processed": 0.0, + "upstream_errors": 0.0, + "revenue_msats": 0.0, + "refunds_msats": 0.0, + "input_tokens": 0.0, + "output_tokens": 0.0, + "total_tokens": 0.0, + } + + points: list[dict[str, Any]] = [] + for row in rows: + total_requests = int(row["total_requests"]) + successful = int(row["successful_chat_completions"]) + failed = int(row["failed_requests"]) + errors = int(row["errors"]) + warnings = int(row["warnings"]) + payment_processed = int(row["payment_processed"]) + upstream_errors = int(row["upstream_errors"]) + revenue_msats = float(row["revenue_msats"]) + refunds_msats = float(row["refunds_msats"]) + input_tokens = int(row["input_tokens"]) + output_tokens = int(row["output_tokens"]) + total_tokens = int(row["total_tokens"]) + + totals["total_requests"] += total_requests + totals["successful_chat_completions"] += successful + totals["failed_requests"] += failed + totals["errors"] += errors + totals["warnings"] += warnings + totals["payment_processed"] += payment_processed + totals["upstream_errors"] += upstream_errors + totals["revenue_msats"] += revenue_msats + totals["refunds_msats"] += refunds_msats + totals["input_tokens"] += input_tokens + totals["output_tokens"] += output_tokens + totals["total_tokens"] += total_tokens + + points.append( + { + "timestamp": str(row["bucket_ts"]), + "total_requests": total_requests, + "successful_chat_completions": successful, + "failed_requests": failed, + "errors": errors, + "warnings": warnings, + "payment_processed": payment_processed, + "upstream_errors": upstream_errors, + "revenue_msats": revenue_msats, + "refunds_msats": refunds_msats, + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "total_tokens": total_tokens, + "requests": total_requests, + } + ) + + normalized_totals: dict[str, int | float] = { + "total_requests": int(totals["total_requests"]), + "successful_chat_completions": int(totals["successful_chat_completions"]), + "failed_requests": int(totals["failed_requests"]), + "errors": int(totals["errors"]), + "warnings": int(totals["warnings"]), + "payment_processed": int(totals["payment_processed"]), + "upstream_errors": int(totals["upstream_errors"]), + "revenue_msats": float(totals["revenue_msats"]), + "refunds_msats": float(totals["refunds_msats"]), + "input_tokens": int(totals["input_tokens"]), + "output_tokens": int(totals["output_tokens"]), + "total_tokens": int(totals["total_tokens"]), + } + + return { + "metrics": points, + "interval_minutes": interval_minutes, + "hours_back": hours_back, + "total_buckets": len(points), + "totals": normalized_totals, + } + + def _query_summary_locked( + self, conn: sqlite3.Connection, cutoff_timestamp: str + ) -> dict[str, Any]: + totals = conn.execute( + """ + SELECT + COALESCE(SUM(total_entries), 0) AS total_entries, + COALESCE(SUM(total_requests), 0) AS total_requests, + COALESCE(SUM(successful_chat_completions), 0) AS successful_chat_completions, + COALESCE(SUM(failed_requests), 0) AS failed_requests, + COALESCE(SUM(errors), 0) AS total_errors, + COALESCE(SUM(warnings), 0) AS total_warnings, + COALESCE(SUM(payment_processed), 0) AS payment_processed, + COALESCE(SUM(upstream_errors), 0) AS upstream_errors, + COALESCE(SUM(revenue_msats), 0) AS revenue_msats, + COALESCE(SUM(refunds_msats), 0) AS refunds_msats, + COALESCE(SUM(input_tokens), 0) AS input_tokens, + COALESCE(SUM(output_tokens), 0) AS output_tokens, + COALESCE(SUM(total_tokens), 0) AS total_tokens + FROM analytics_minute + WHERE minute_ts >= ? + """, + (cutoff_timestamp,), + ).fetchone() + + unique_models = [ + str(row[0]) + for row in conn.execute( + """ + SELECT DISTINCT model + FROM analytics_model_presence_minute + WHERE minute_ts >= ? + ORDER BY model ASC + """, + (cutoff_timestamp,), + ).fetchall() + ] + + error_types = { + str(row[0]): int(row[1]) + for row in conn.execute( + """ + SELECT error_type, COALESCE(SUM(count), 0) AS total_count + FROM analytics_error_type_minute + WHERE minute_ts >= ? + GROUP BY error_type + """, + (cutoff_timestamp,), + ).fetchall() + } + + total_requests = int(totals["total_requests"]) + successful = int(totals["successful_chat_completions"]) + failed_requests = int(totals["failed_requests"]) + input_tokens = int(totals["input_tokens"]) + output_tokens = int(totals["output_tokens"]) + total_tokens = int(totals["total_tokens"]) + + revenue_msats = float(totals["revenue_msats"]) + refunds_msats = float(totals["refunds_msats"]) + net_revenue_msats = revenue_msats - refunds_msats + + revenue_sats = revenue_msats / 1000 + refunds_sats = refunds_msats / 1000 + net_revenue_sats = net_revenue_msats / 1000 + + return { + "total_entries": int(totals["total_entries"]), + "total_requests": total_requests, + "successful_chat_completions": successful, + "failed_requests": failed_requests, + "total_errors": int(totals["total_errors"]), + "total_warnings": int(totals["total_warnings"]), + "payment_processed": int(totals["payment_processed"]), + "upstream_errors": int(totals["upstream_errors"]), + "unique_models_count": len(unique_models), + "unique_models": unique_models, + "error_types": error_types, + "input_tokens": input_tokens, + "output_tokens": output_tokens, + "total_tokens": total_tokens, + "avg_input_tokens_per_completion": (input_tokens / successful) + if successful > 0 + else 0, + "avg_output_tokens_per_completion": (output_tokens / successful) + if successful > 0 + else 0, + "avg_total_tokens_per_completion": (total_tokens / successful) + if successful > 0 + else 0, + "success_rate": (successful / total_requests * 100) + if total_requests > 0 + else 0, + "revenue_msats": revenue_msats, + "refunds_msats": refunds_msats, + "revenue_sats": revenue_sats, + "refunds_sats": refunds_sats, + "net_revenue_msats": net_revenue_msats, + "net_revenue_sats": net_revenue_sats, + "avg_revenue_per_request_msats": (revenue_msats / successful) + if successful > 0 + else 0, + "refund_rate": (failed_requests / total_requests * 100) + if total_requests > 0 + else 0, + } + + def _query_error_details_locked( + self, + conn: sqlite3.Connection, + *, + cutoff_timestamp: str, + limit: int, + total_error_count: int | None = None, + ) -> dict[str, Any]: + rows = conn.execute( + """ + SELECT + timestamp, + message, + error_type, + pathname, + lineno, + request_id + FROM analytics_error_events + WHERE timestamp >= ? + ORDER BY timestamp DESC + LIMIT ? + """, + (cutoff_timestamp, limit), + ).fetchall() + + if total_error_count is None: + total_error_count_row = conn.execute( + """ + SELECT COALESCE(SUM(errors), 0) + FROM analytics_minute + WHERE minute_ts >= ? + """, + (cutoff_timestamp,), + ).fetchone() + total_error_count = int(total_error_count_row[0]) if total_error_count_row else 0 + + return { + "errors": [ + { + "timestamp": str(row["timestamp"]), + "message": str(row["message"]), + "error_type": str(row["error_type"]), + "pathname": str(row["pathname"]), + "lineno": int(row["lineno"]), + "request_id": str(row["request_id"]), + } + for row in rows + ], + "total_count": int(total_error_count), + } + + def _query_revenue_by_model_locked( + self, + conn: sqlite3.Connection, + *, + cutoff_timestamp: str, + limit: int, + ) -> dict[str, Any]: + rows = conn.execute( + """ + SELECT + model, + COALESCE(SUM(revenue_msats), 0) AS revenue_msats, + COALESCE(SUM(refunds_msats), 0) AS refunds_msats, + COALESCE(SUM(requests), 0) AS requests, + COALESCE(SUM(successful), 0) AS successful, + COALESCE(SUM(failed), 0) AS failed + FROM analytics_model_minute + WHERE minute_ts >= ? + GROUP BY model + ORDER BY (COALESCE(SUM(revenue_msats), 0) - COALESCE(SUM(refunds_msats), 0)) DESC + """, + (cutoff_timestamp,), + ).fetchall() + + models: list[dict[str, Any]] = [] + total_revenue_sats = 0.0 + + for row in rows: + revenue_msats = float(row["revenue_msats"]) + refunds_msats = float(row["refunds_msats"]) + revenue_sats = revenue_msats / 1000 + refunds_sats = refunds_msats / 1000 + net_revenue_sats = revenue_sats - refunds_sats + successful = int(row["successful"]) + + models.append( + { + "model": str(row["model"]), + "revenue_sats": revenue_sats, + "refunds_sats": refunds_sats, + "net_revenue_sats": net_revenue_sats, + "requests": int(row["requests"]), + "successful": successful, + "failed": int(row["failed"]), + "avg_revenue_per_request": (revenue_sats / successful) + if successful > 0 + else 0, + } + ) + total_revenue_sats += net_revenue_sats + + return { + "models": models[:limit], + "total_revenue_sats": total_revenue_sats, + "total_models": len(models), + } + + def _query_model_usage_mix_locked( + self, + conn: sqlite3.Connection, + *, + cutoff_timestamp: str, + interval_minutes: int, + hours_back: int, + limit: int, + ) -> dict[str, Any]: + top_limit = max(1, min(int(limit), 20)) + top_rows_requests = conn.execute( + """ + SELECT + model, + COALESCE(SUM(successful), 0) AS total_successful + FROM analytics_model_minute + WHERE minute_ts >= ? + AND model != 'unknown' + GROUP BY model + ORDER BY total_successful DESC + LIMIT ? + """, + (cutoff_timestamp, top_limit), + ).fetchall() + top_rows_revenue = conn.execute( + """ + SELECT + model, + COALESCE(SUM(revenue_msats), 0) AS total_revenue_msats + FROM analytics_model_minute + WHERE minute_ts >= ? + AND model != 'unknown' + GROUP BY model + ORDER BY total_revenue_msats DESC + LIMIT ? + """, + (cutoff_timestamp, top_limit), + ).fetchall() + top_rows_tokens = conn.execute( + """ + SELECT + model, + COALESCE(SUM(total_tokens), 0) AS total_tokens + FROM analytics_model_minute + WHERE minute_ts >= ? + AND model != 'unknown' + GROUP BY model + ORDER BY total_tokens DESC + LIMIT ? + """, + (cutoff_timestamp, top_limit), + ).fetchall() + + top_models_requests = [ + str(row["model"]) + for row in top_rows_requests + if int(row["total_successful"] or 0) > 0 + ] + top_models_revenue = [ + str(row["model"]) + for row in top_rows_revenue + if float(row["total_revenue_msats"] or 0.0) > 0 + ] + top_models_tokens = [ + str(row["model"]) + for row in top_rows_tokens + if int(row["total_tokens"] or 0) > 0 + ] + + selected_models: list[str] = [] + for model in top_models_requests + top_models_revenue + top_models_tokens: + if model not in selected_models: + selected_models.append(model) + + bucket_seconds = max(60, int(interval_minutes) * 60) + total_rows = conn.execute( + """ + SELECT + datetime( + (CAST(strftime('%s', minute_ts) AS INTEGER) / ?) * ?, + 'unixepoch' + ) AS bucket_ts, + COALESCE(SUM(successful), 0) AS total_successful, + COALESCE(SUM(revenue_msats), 0) AS total_revenue_msats, + COALESCE(SUM(total_tokens), 0) AS total_tokens + FROM analytics_model_minute + WHERE minute_ts >= ? + GROUP BY bucket_ts + ORDER BY bucket_ts + """, + (bucket_seconds, bucket_seconds, cutoff_timestamp), + ).fetchall() + + bucket_index: dict[str, dict[str, Any]] = {} + for row in total_rows: + total_successful = int(row["total_successful"]) + total_revenue_msats = float(row["total_revenue_msats"]) + total_tokens = int(row["total_tokens"]) + if ( + total_successful <= 0 + and total_revenue_msats <= 0 + and total_tokens <= 0 + ): + continue + + bucket_ts = str(row["bucket_ts"]) + bucket = bucket_index.setdefault( + bucket_ts, + { + "timestamp": bucket_ts, + "total_successful": 0, + "total_revenue_msats": 0.0, + "total_tokens": 0, + "others": 0, + "others_revenue_msats": 0.0, + "others_tokens": 0, + "model_counts": {}, + "model_revenue_msats": {}, + "model_tokens": {}, + }, + ) + bucket["total_successful"] = total_successful + bucket["total_revenue_msats"] = total_revenue_msats + bucket["total_tokens"] = total_tokens + bucket["others"] = total_successful + bucket["others_revenue_msats"] = total_revenue_msats + bucket["others_tokens"] = total_tokens + + if selected_models and bucket_index: + placeholders = ",".join("?" for _ in selected_models) + top_model_rows = conn.execute( + f""" + SELECT + datetime( + (CAST(strftime('%s', minute_ts) AS INTEGER) / ?) * ?, + 'unixepoch' + ) AS bucket_ts, + model, + COALESCE(SUM(successful), 0) AS successful, + COALESCE(SUM(revenue_msats), 0) AS revenue_msats, + COALESCE(SUM(total_tokens), 0) AS total_tokens + FROM analytics_model_minute + WHERE minute_ts >= ? + AND model IN ({placeholders}) + GROUP BY bucket_ts, model + ORDER BY bucket_ts + """, + (bucket_seconds, bucket_seconds, cutoff_timestamp, *selected_models), + ).fetchall() + + for row in top_model_rows: + bucket_ts = str(row["bucket_ts"]) + if bucket_ts not in bucket_index: + continue + bucket = bucket_index[bucket_ts] + + model = str(row["model"]) + successful = int(row["successful"]) + revenue_msats = float(row["revenue_msats"]) + total_tokens = int(row["total_tokens"]) + + model_counts = bucket["model_counts"] + model_counts[model] = successful + model_revenue_msats = bucket["model_revenue_msats"] + model_revenue_msats[model] = revenue_msats + model_tokens = bucket["model_tokens"] + model_tokens[model] = total_tokens + + bucket["others"] = max(0, int(bucket["others"]) - successful) + bucket["others_revenue_msats"] = max( + 0.0, + float(bucket["others_revenue_msats"]) - revenue_msats, + ) + bucket["others_tokens"] = max( + 0, + int(bucket["others_tokens"]) - total_tokens, + ) + + metrics = sorted(bucket_index.values(), key=lambda item: str(item["timestamp"])) + + return { + "top_models": top_models_requests, + "top_models_by_metric": { + "requests": top_models_requests, + "revenue": top_models_revenue, + "tokens": top_models_tokens, + }, + "metrics": metrics, + "hours_back": hours_back, + "interval_minutes": interval_minutes, + "total_buckets": len(metrics), + } + + def _cutoff_timestamp(self, hours_back: int) -> str: + cutoff = datetime.now(timezone.utc) - timedelta(hours=hours_back) + return cutoff.strftime("%Y-%m-%d %H:%M:%S") + + def _minute_key(self, timestamp: Any) -> str | None: + if not isinstance(timestamp, str) or len(timestamp) != 19: + return None + if timestamp[10] != " ": + return None + return f"{timestamp[:16]}:00" + + def _extract_success_metrics( + self, entry: dict[str, Any], message: str + ) -> tuple[bool, float, int, int]: + # These auth logs are emitted once per successful settlement across providers + # and avoid duplicate counting from provider-specific completion logs. + logger_name = str(entry.get("name", "")) + if not logger_name.startswith("routstr.auth"): + return False, 0.0, 0, 0 + + input_tokens = self._parse_token_count(entry.get("input_tokens", 0)) + output_tokens = self._parse_token_count(entry.get("output_tokens", 0)) + + if "calculated token-based cost" in message: + token_cost = entry.get("token_cost", 0) + if isinstance(token_cost, (int, float)) and token_cost > 0: + return True, float(token_cost), input_tokens, output_tokens + return True, 0.0, input_tokens, output_tokens + + if "max cost payment finalized" in message: + charged_amount = entry.get("charged_amount", 0) + if isinstance(charged_amount, (int, float)) and charged_amount > 0: + return True, float(charged_amount), input_tokens, output_tokens + return True, 0.0, input_tokens, output_tokens + + return False, 0.0, 0, 0 + + def _parse_token_count(self, value: Any) -> int: + if isinstance(value, bool): + return 0 + if isinstance(value, int): + return max(0, value) + if isinstance(value, float): + return max(0, int(value)) + if isinstance(value, str): + try: + return max(0, int(float(value))) + except ValueError: + return 0 + return 0 + + def _new_minute_stats(self) -> dict[str, float]: + return { + "total_entries": 0.0, + "total_requests": 0.0, + "successful_chat_completions": 0.0, + "failed_requests": 0.0, + "errors": 0.0, + "warnings": 0.0, + "payment_processed": 0.0, + "upstream_errors": 0.0, + "revenue_msats": 0.0, + "refunds_msats": 0.0, + "input_tokens": 0.0, + "output_tokens": 0.0, + "total_tokens": 0.0, + } + + def _new_model_stats(self) -> dict[str, float]: + return { + "requests": 0.0, + "successful": 0.0, + "failed": 0.0, + "revenue_msats": 0.0, + "refunds_msats": 0.0, + "input_tokens": 0.0, + "output_tokens": 0.0, + "total_tokens": 0.0, + } diff --git a/ui/app/page.tsx b/ui/app/page.tsx index ba948d25..380f465e 100644 --- a/ui/app/page.tsx +++ b/ui/app/page.tsx @@ -1,16 +1,520 @@ 'use client'; -import { useEffect, useState } from 'react'; +import { useEffect, useMemo, useState } from 'react'; +import { format } from 'date-fns'; import { useQuery } from '@tanstack/react-query'; -import { AppPageShell } from '@/components/app-page-shell'; +import { CalendarIcon, RefreshCw } from 'lucide-react'; +import { Bar, BarChart, CartesianGrid, XAxis, YAxis } from 'recharts'; +import type { DateRange } from 'react-day-picker'; +import { UsageMetricsChart } from '@/components/usage-metrics-chart'; +import { UsageSummaryCards } from '@/components/usage-summary-cards'; +import { ErrorDetailsTable } from '@/components/error-details-table'; import { DashboardBalanceSummary } from '@/components/dashboard-balance-summary'; -import { CheatSheet } from '@/components/landing/cheat-sheet'; +import { + AdminService, + type UsageMetricData, + type UsageSummary, +} from '@/lib/api/services/admin'; +import { Button } from '@/components/ui/button'; +import { Card, CardContent, CardHeader, CardTitle } from '@/components/ui/card'; +import { Calendar } from '@/components/ui/calendar'; +import { + ChartContainer, + ChartTooltip, + ChartTooltipContent, + type ChartConfig as UiChartConfig, +} from '@/components/ui/chart'; +import { + Empty, + EmptyDescription, + EmptyHeader, + EmptyMedia, + EmptyTitle, +} from '@/components/ui/empty'; +import { + Popover, + PopoverContent, + PopoverTrigger, +} from '@/components/ui/popover'; +import { + Select, + SelectContent, + SelectItem, + SelectTrigger, + SelectValue, +} from '@/components/ui/select'; import { Skeleton } from '@/components/ui/skeleton'; -import { fetchBtcUsdPrice, btcToSatsRate } from '@/lib/exchange-rate'; -import { ConfigurationService } from '@/lib/api/services/configuration'; import { useCurrencyStore } from '@/lib/stores/currency'; +import { fetchBtcUsdPrice, btcToSatsRate } from '@/lib/exchange-rate'; +import { CheatSheet } from '@/components/landing/cheat-sheet'; +import { ConfigurationService } from '@/lib/api/services/configuration'; +import { AppPageShell } from '@/components/app-page-shell'; +import { useIsMobile } from '@/hooks/use-mobile'; +import type { DisplayUnit } from '@/lib/types/units'; +import { cn } from '@/lib/utils'; + +type ChartDatum = Record & { timestamp: string }; + +type ChartKeyConfig = { + key: string; + name: string; + color: string; +}; + +type ChartConfig = { + id: string; + title: string; + mobileTitle?: string; + description: string; + data: ChartDatum[]; + dataKeys: ChartKeyConfig[]; + totals?: Partial>; + metricType: 'currency' | 'count'; +}; + +const TIME_RANGE_PRESETS = [ + { value: '24h', label: 'Last 24 Hours', hours: 24 }, + { value: '7d', label: 'Last 7 Days', hours: 7 * 24 }, + { value: '30d', label: 'Last 30 Days', hours: 30 * 24 }, + { value: '3m', label: 'Last 3 Months', hours: 90 * 24 }, + { value: '12m', label: 'Last 12 Months', hours: 365 * 24 }, +] as const; + +type TimeRangePresetValue = (typeof TIME_RANGE_PRESETS)[number]['value']; + +const DEFAULT_TIME_RANGE_PRESET = + TIME_RANGE_PRESETS.find((option) => option.value === '7d') ?? + TIME_RANGE_PRESETS[0]; + +function normalizeDateRange(range: DateRange): DateRange { + if (!range.from || !range.to) { + return range; + } + + if (range.from.getTime() <= range.to.getTime()) { + return range; + } + + return { + from: range.to, + to: range.from, + }; +} + +function getRangeHours(range?: DateRange): number | null { + if (!range?.from || !range.to) { + return null; + } + + const normalized = normalizeDateRange(range); + const fromTime = normalized.from?.getTime(); + const toTime = normalized.to?.getTime(); + + if (fromTime === undefined || toTime === undefined) { + return null; + } + + const diffMs = toTime - fromTime; + const diffHours = Math.ceil(diffMs / (1000 * 60 * 60)); + + return Math.max(1, diffHours); +} + +function formatCompactDateRangeLabel(range?: DateRange): string { + if (!range?.from || !range.to) { + return 'Custom range'; + } + + const normalized = normalizeDateRange(range); + const from = normalized.from; + const to = normalized.to; + + if (!from || !to) { + return 'Custom range'; + } + + const sameMonth = format(from, 'yyyy-MM') === format(to, 'yyyy-MM'); + if (sameMonth) { + return `${format(from, 'MMM d')} - ${format(to, 'd')}`; + } + + const sameYear = format(from, 'yyyy') === format(to, 'yyyy'); + if (sameYear) { + return `${format(from, 'MMM d')} - ${format(to, 'MMM d')}`; + } + + return `${format(from, 'MMM d, yyyy')} - ${format(to, 'MMM d, yyyy')}`; +} + +function getAutoIntervalMinutes(hours: number): number { + const totalMinutes = Math.max(60, Math.ceil(hours * 60)); + const targetPoints = 96; + const idealInterval = Math.ceil(totalMinutes / targetPoints); + const allowedIntervals = [5, 15, 30, 60, 120, 180, 240, 360, 480, 720, 1440]; + + return ( + allowedIntervals.find( + (intervalMinutes) => intervalMinutes >= idealInterval + ) ?? allowedIntervals[allowedIntervals.length - 1] + ); +} + +function getQueryErrorMessage(error: unknown): string { + if ( + error && + typeof error === 'object' && + 'message' in error && + typeof error.message === 'string' && + error.message.trim().length > 0 + ) { + return error.message; + } + + return 'The analytics request failed. Refresh and try again.'; +} + +function SectionLoading({ label }: { label: string }) { + if (label === 'summary') { + return ( +
+ {Array.from({ length: 14 }).map((_, index) => ( + + + + + + + + + + + ))} +
+ ); + } + + if (label === 'metrics') { + return ( + + +
+ {Array.from({ length: 4 }).map((_, index) => ( + + ))} +
+
+
+ + +
+ +
+ + +
+ {Array.from({ length: 4 }).map((_, index) => ( +
+ + + +
+ ))} +
+
+
+ {Array.from({ length: 3 }).map((_, index) => ( + + ))} +
+
+ {Array.from({ length: 10 }).map((_, index) => ( + + ))} +
+
+ + + ); + } + + if (label === 'revenue by model') { + return ( + + + + + + +
+
+
+ {Array.from({ length: 9 }).map((_, index) => ( + 0 && index < 5 && 'w-14 justify-self-end', + index === 5 && 'w-24', + index > 5 && 'w-16 justify-self-end' + )} + /> + ))} +
+ {Array.from({ length: 5 }).map((_, rowIndex) => ( +
+ + + + + +
+ + +
+ + + +
+ ))} +
+
+ + + ); + } + + if (label === 'errors') { + return ( + + + + + + +
+
+
+ + + + + +
+ {Array.from({ length: 6 }).map((_, rowIndex) => ( +
+ + + + + +
+ ))} +
+
+ + + ); + } + + return ( + + + + + + + ); +} + +function DashboardInsights({ + summary, + isMobile, +}: { + summary?: UsageSummary; + isMobile: boolean; +}) { + if (!summary) { + return null; + } + + const errorTypes = Object.entries(summary.error_types || {}) + .map(([type, count]) => ({ + type, + count: typeof count === 'number' ? count : Number(count || 0), + })) + .filter((entry) => Number.isFinite(entry.count) && entry.count > 0) + .sort((a, b) => b.count - a.count); + + const hasErrorTypes = errorTypes.length > 0; + + if (!hasErrorTypes) { + return null; + } + + const compactNumber = new Intl.NumberFormat('en-US', { + notation: 'compact', + maximumFractionDigits: 1, + }); + const truncateErrorType = (value: string): string => { + const maxLength = isMobile ? 16 : 26; + if (value.length <= maxLength) { + return value; + } + return `${value.slice(0, maxLength - 1)}…`; + }; + const chartData = errorTypes.slice(0, 10).map(({ type, count }) => ({ + type, + label: truncateErrorType(type), + count, + })); + const totalErrors = errorTypes.reduce((sum, entry) => sum + entry.count, 0); + const chartConfig: UiChartConfig = { + count: { + label: 'Errors', + color: 'var(--chart-4)', + }, + }; + + return ( +
+ {hasErrorTypes && ( + + + Error Types Distribution +

+ Top {chartData.length} categories in this range. +

+ + + + + + + compactNumber.format( + typeof value === 'number' ? value : Number(value || 0) + ) + } + /> + + + String(payload?.[0]?.payload?.type ?? '') + } + formatter={(value, name) => { + const numericValue = + typeof value === 'number' + ? value + : Number(value || 0); + return ( +
+ + {name} + + + {Number.isFinite(numericValue) + ? numericValue.toLocaleString() + : '-'} + +
+ ); + }} + /> + } + /> + + + +

+ Total errors: {totalErrors.toLocaleString()} +

+ + + )} +
+ ); +} export default function DashboardPage() { + const [selectedPreset, setSelectedPreset] = useState( + DEFAULT_TIME_RANGE_PRESET.value + ); + const [customRange, setCustomRange] = useState(); + const [pendingCustomRange, setPendingCustomRange] = useState(); + const [isCustomRangeActive, setIsCustomRangeActive] = useState(false); + const [isCustomRangePickerOpen, setIsCustomRangePickerOpen] = useState(false); + const [isManualRefreshing, setIsManualRefreshing] = useState(false); + const [activeChartId, setActiveChartId] = useState('revenue'); + const isMobile = useIsMobile(); const { displayUnit } = useCurrencyStore(); const [isAuthenticated, setIsAuthenticated] = useState(false); const [isAuthResolved, setIsAuthResolved] = useState(false); @@ -42,6 +546,236 @@ export default function DashboardPage() { }); const usdPerSat = btcUsdPrice ? btcToSatsRate(btcUsdPrice) : null; + const activePreset = + TIME_RANGE_PRESETS.find((option) => option.value === selectedPreset) ?? + DEFAULT_TIME_RANGE_PRESET; + const customRangeHours = getRangeHours(customRange); + const queryHours = + isCustomRangeActive && customRangeHours + ? customRangeHours + : activePreset.hours; + const autoInterval = getAutoIntervalMinutes(queryHours); + const usageRefetchIntervalMs = useMemo(() => { + if (queryHours > 90 * 24) { + return 4 * 60 * 60_000; + } + if (queryHours > 30 * 24) { + return 2 * 60 * 60_000; + } + if (queryHours > 7 * 24) { + return 30 * 60_000; + } + return 60_000; + }, [queryHours]); + const revenueDisplayUnit: DisplayUnit = useMemo(() => { + if (displayUnit === 'usd' && usdPerSat === null) { + // Keep revenue charts meaningful while the USD rate is unavailable. + return 'sat'; + } + return displayUnit; + }, [displayUnit, usdPerSat]); + const revenueUnitLabel = + revenueDisplayUnit === 'usd' + ? 'USD' + : revenueDisplayUnit === 'sat' + ? 'sats' + : revenueDisplayUnit === 'msat' + ? 'msats' + : revenueDisplayUnit; + + const { + data: metricsData, + isLoading: metricsLoading, + error: metricsError, + refetch: refetchMetrics, + } = useQuery({ + queryKey: ['usage-metrics', autoInterval, queryHours], + queryFn: () => AdminService.getUsageMetrics(autoInterval, queryHours), + enabled: isAuthenticated, + refetchInterval: usageRefetchIntervalMs, + staleTime: 30_000, + }); + + const { + data: summaryData, + isLoading: summaryLoading, + error: summaryError, + refetch: refetchSummary, + } = useQuery({ + queryKey: ['usage-summary', queryHours], + queryFn: () => AdminService.getUsageSummary(queryHours), + enabled: isAuthenticated, + refetchInterval: usageRefetchIntervalMs, + staleTime: 30_000, + }); + + const { + data: errorData, + isLoading: errorLoading, + error: errorDetailsError, + refetch: refetchErrors, + } = useQuery({ + queryKey: ['usage-errors', queryHours], + queryFn: () => AdminService.getErrorDetails(queryHours, 100), + enabled: isAuthenticated, + refetchInterval: usageRefetchIntervalMs, + staleTime: 30_000, + }); + + const chartConfigs = useMemo(() => { + if (!metricsData || metricsData.metrics.length === 0) { + return []; + } + + const metricPoints = metricsData.metrics as ChartDatum[]; + const convertRevenueMsats = (amountMsats: number): number => { + if (revenueDisplayUnit === 'msat') { + return amountMsats; + } + + const sats = amountMsats / 1000; + if (revenueDisplayUnit === 'usd') { + return sats * (usdPerSat ?? 0); + } + + return sats; + }; + const revenuePoints = metricsData.metrics.map( + (metric: UsageMetricData) => ({ + ...metric, + revenue_display: convertRevenueMsats(metric.revenue_msats), + }) + ) as ChartDatum[]; + + const hasTokenMetrics = metricPoints.some((metric) => + ['input_tokens', 'output_tokens', 'total_tokens'].some( + (key) => typeof metric[key] === 'number' + ) + ); + + return [ + { + id: 'revenue', + title: 'Revenue Over Time', + mobileTitle: 'Revenue', + description: 'Track collected revenue trends over time.', + data: revenuePoints, + metricType: 'currency', + dataKeys: [ + { + key: 'revenue_display', + name: 'Revenue', + color: 'var(--chart-1)', + }, + ], + }, + { + id: 'requests', + title: 'Request Volume', + mobileTitle: 'Requests', + description: 'Understand traffic and completion reliability over time.', + data: metricPoints, + metricType: 'count', + dataKeys: [ + { + key: 'total_requests', + name: 'Total Requests', + color: 'var(--chart-1)', + }, + { + key: 'successful_chat_completions', + name: 'Successful', + color: 'var(--chart-2)', + }, + { + key: 'failed_requests', + name: 'Failed', + color: 'var(--chart-5)', + }, + ], + }, + { + id: 'errors', + title: 'Error Tracking', + mobileTitle: 'Errors', + description: 'Monitor warnings, handled errors, and upstream failures.', + data: metricPoints, + metricType: 'count', + dataKeys: [ + { + key: 'errors', + name: 'Errors', + color: 'var(--chart-4)', + }, + { + key: 'warnings', + name: 'Warnings', + color: 'var(--chart-3)', + }, + { + key: 'upstream_errors', + name: 'Upstream Errors', + color: 'var(--chart-5)', + }, + ], + }, + { + id: 'payments', + title: 'Payment Activity', + mobileTitle: 'Payments', + description: 'Follow payment processing activity by interval.', + data: metricPoints, + metricType: 'count', + dataKeys: [ + { + key: 'payment_processed', + name: 'Payments Processed', + color: 'var(--chart-2)', + }, + ], + }, + ...(hasTokenMetrics + ? [ + { + id: 'tokens', + title: 'Token Usage', + mobileTitle: 'Tokens', + description: + 'Track input, output, and total token throughput over time.', + data: metricPoints, + metricType: 'count' as const, + dataKeys: [ + { + key: 'total_tokens', + name: 'Total Tokens', + color: 'var(--chart-1)', + }, + { + key: 'input_tokens', + name: 'Input Tokens', + color: 'var(--chart-2)', + }, + { + key: 'output_tokens', + name: 'Output Tokens', + color: 'var(--chart-3)', + }, + ], + }, + ] + : []), + ]; + }, [metricsData, revenueDisplayUnit, usdPerSat]); + + useEffect(() => { + if (chartConfigs.length === 0) { + return; + } + + if (!chartConfigs.some((config) => config.id === activeChartId)) { + setActiveChartId(chartConfigs[0].id); + } + }, [chartConfigs, activeChartId]); if (!isAuthResolved) { return ( @@ -51,7 +785,8 @@ export default function DashboardPage() { - + + ); @@ -61,6 +796,95 @@ export default function DashboardPage() { return ; } + const handleRefresh = async () => { + if (isManualRefreshing) { + return; + } + + setIsManualRefreshing(true); + try { + await Promise.allSettled([ + refetchMetrics(), + refetchSummary(), + refetchErrors(), + ]); + } finally { + setIsManualRefreshing(false); + } + }; + + const openRangePicker = () => { + // Force a fresh selection so the range is only applied + // after the user explicitly chooses both start and end. + setPendingCustomRange(undefined); + setIsCustomRangePickerOpen(true); + }; + + const handleCustomRangePickerChange = (open: boolean) => { + if (open) { + openRangePicker(); + return; + } + + setIsCustomRangePickerOpen(false); + }; + + const handleRangeSelectChange = (value: string) => { + if (value === 'custom') { + // Always require an explicit fresh range selection. + openRangePicker(); + return; + } + + const preset = TIME_RANGE_PRESETS.find((option) => option.value === value); + + if (!preset) { + return; + } + + setSelectedPreset(preset.value); + setIsCustomRangeActive(false); + }; + + const handleCustomRangeSelect = (nextRange: DateRange | undefined) => { + if (!nextRange?.from) { + setPendingCustomRange(undefined); + return; + } + + const normalized = normalizeDateRange(nextRange); + const from = normalized.from; + const to = normalized.to; + const hasPreviousStart = Boolean(pendingCustomRange?.from); + + if (!from) { + setPendingCustomRange(undefined); + return; + } + + // DayPicker may emit from===to on the first click in range mode. + // Keep waiting until the user explicitly picks a second (end) date. + const isSameDay = to ? from.getTime() === to.getTime() : false; + if (!hasPreviousStart || !to || isSameDay) { + setPendingCustomRange({ from, to: undefined }); + return; + } + + setPendingCustomRange(normalized); + setCustomRange(normalized); + setIsCustomRangeActive(true); + setIsCustomRangePickerOpen(false); + }; + + const activeChartConfig = + chartConfigs.find((config) => config.id === activeChartId) ?? + chartConfigs[0]; + const selectedRangeValue = + isCustomRangeActive && customRange?.from && customRange?.to + ? 'custom' + : selectedPreset; + const compactCustomRangeLabel = formatCompactDateRangeLabel(customRange); + return (
@@ -69,7 +893,7 @@ export default function DashboardPage() { Dashboard

- Node balances and wallet status. + Node balances, request health, and revenue trends.

@@ -77,6 +901,196 @@ export default function DashboardPage() { displayUnit={displayUnit} usdPerSat={usdPerSat} /> + +
+
+
+

+ Usage Analytics +

+
+

+ All cards and charts in this section update from the selected + range. +

+
+ +
+
+
+ + + + + + + + + +
+ + +
+
+ + +
+ + {metricsLoading ? ( + + ) : metricsError ? ( + + + + + + + + Unable to load analytics + + {getQueryErrorMessage(metricsError)} + + + + + + ) : activeChartConfig ? ( + ({ + id: config.id, + label: isMobile + ? (config.mobileTitle ?? config.title) + : config.title, + }))} + activeTabId={activeChartId} + onTabChange={setActiveChartId} + /> + ) : ( + + + + + + + + No data available + + No metrics data exists for this range yet. Try a broader + range. + + + + + + )} + + {summaryLoading ? ( + + ) : summaryError ? ( + + + + + Usage summary unavailable + + {getQueryErrorMessage(summaryError)} + + + + + + ) : summaryData ? ( + + ) : null} + + + + {errorLoading ? ( + + ) : errorDetailsError ? ( + + + + + Error details unavailable + + {getQueryErrorMessage(errorDetailsError)} + + + + + + ) : errorData ? ( + + ) : null} +
); diff --git a/ui/components/error-details-table.tsx b/ui/components/error-details-table.tsx new file mode 100644 index 00000000..3a630e5b --- /dev/null +++ b/ui/components/error-details-table.tsx @@ -0,0 +1,78 @@ +'use client'; + +import { Card, CardContent, CardHeader, CardTitle } from '@/components/ui/card'; +import { + Table, + TableBody, + TableCell, + TableHead, + TableHeader, + TableRow, +} from '@/components/ui/table'; +import { Badge } from '@/components/ui/badge'; +import { ErrorDetail } from '@/lib/api/services/admin'; + +interface ErrorDetailsTableProps { + errors: ErrorDetail[]; +} + +export function ErrorDetailsTable({ errors }: ErrorDetailsTableProps) { + if (errors.length === 0) { + return ( + + + Recent Errors + + +

+ No errors found in the selected time period +

+ + + ); + } + + return ( + + + Recent Errors ({errors.length}) + + +
+ + + + Timestamp + Type + Message + Location + Request ID + + + + {errors.map((error, index) => ( + + + {new Date(error.timestamp).toLocaleString()} + + + {error.error_type} + + + {error.message} + + + {error.pathname}:{error.lineno} + + + {error.request_id || '-'} + + + ))} + +
+
+ + + ); +} diff --git a/ui/components/usage-metrics-chart.tsx b/ui/components/usage-metrics-chart.tsx new file mode 100644 index 00000000..87d8e937 --- /dev/null +++ b/ui/components/usage-metrics-chart.tsx @@ -0,0 +1,387 @@ +'use client'; + +import { useEffect, useMemo, useRef, useState } from 'react'; +import { Card, CardContent, CardHeader, CardTitle } from '@/components/ui/card'; +import { Button } from '@/components/ui/button'; +import { Tabs, TabsList, TabsTrigger } from '@/components/ui/tabs'; +import { ExpandIcon, Minimize2Icon } from 'lucide-react'; +import { Area, AreaChart, XAxis, YAxis, CartesianGrid } from 'recharts'; +import { + ChartConfig, + ChartContainer, + ChartTooltip, + ChartTooltipContent, +} from '@/components/ui/chart'; +import { cn } from '@/lib/utils'; +import { useIsMobile } from '@/hooks/use-mobile'; + +interface UsageMetricsChartProps { + data: Array & { timestamp: string }>; + title: string; + description?: string; + dataKeys: Array<{ + key: string; + name: string; + color: string; + }>; + totals?: Partial>; + metricType?: 'currency' | 'count'; + currencyUnitLabel?: string; + tabs?: Array<{ id: string; label: string }>; + activeTabId?: string; + onTabChange?: (tabId: string) => void; +} + +export function UsageMetricsChart({ + data, + title, + description, + dataKeys, + totals, + metricType = 'count', + currencyUnitLabel = 'sats', + tabs, + activeTabId, + onTabChange, +}: UsageMetricsChartProps) { + const [hiddenSeries, setHiddenSeries] = useState>(new Set()); + const [isFullscreen, setIsFullscreen] = useState(false); + const isMobile = useIsMobile(); + const containerRef = useRef(null); + + const compactNumber = useMemo( + () => + new Intl.NumberFormat('en-US', { + notation: 'compact', + maximumFractionDigits: 1, + }), + [] + ); + + const hasMultipleDays = useMemo(() => { + const daySet = new Set( + data.map((item) => new Date(item.timestamp).toDateString()) + ); + return daySet.size > 1; + }, [data]); + + const formatAxisTick = (timestamp: string): string => { + const date = new Date(timestamp); + if (Number.isNaN(date.getTime())) { + return ''; + } + + if (hasMultipleDays) { + return date.toLocaleString([], { + month: 'short', + day: 'numeric', + hour: '2-digit', + }); + } + + return date.toLocaleTimeString([], { + hour: '2-digit', + minute: '2-digit', + }); + }; + + const formatMetricValue = (value: number): string => { + const formatted = compactNumber.format(value); + return metricType === 'currency' + ? `${formatted} ${currencyUnitLabel}` + : formatted; + }; + + const metricTotals = useMemo(() => { + const fallbackTotals = dataKeys.reduce>( + (acc, dataKey) => { + acc[dataKey.key] = 0; + return acc; + }, + {} + ); + + for (const point of data) { + for (const dataKey of dataKeys) { + const rawValue = point?.[dataKey.key]; + const value = + typeof rawValue === 'number' ? rawValue : Number(rawValue || 0); + if (Number.isFinite(value)) { + fallbackTotals[dataKey.key] += value; + } + } + } + + if (!totals) { + return fallbackTotals; + } + + const mergedTotals = { ...fallbackTotals }; + for (const dataKey of dataKeys) { + const rawTotal = totals[dataKey.key]; + if (typeof rawTotal === 'number' && Number.isFinite(rawTotal)) { + mergedTotals[dataKey.key] = rawTotal; + } + } + + return mergedTotals; + }, [data, dataKeys, totals]); + + const metricChips = useMemo( + () => + dataKeys.map((dataKey) => ({ + ...dataKey, + value: Number.isFinite(metricTotals[dataKey.key]) + ? metricTotals[dataKey.key] + : 0, + })), + [dataKeys, metricTotals] + ); + + const visibleDataKeys = dataKeys.filter( + (dataKey) => !hiddenSeries.has(dataKey.key) + ); + + const toggleSeries = (key: string) => { + setHiddenSeries((current) => { + const next = new Set(current); + if (next.has(key)) { + next.delete(key); + } else { + next.add(key); + } + return next; + }); + }; + + const chartConfig = dataKeys.reduce((acc, keyConfig) => { + acc[keyConfig.key] = { + label: keyConfig.name, + color: keyConfig.color, + }; + return acc; + }, {}); + + useEffect(() => { + const handleFullscreenChange = () => { + setIsFullscreen(document.fullscreenElement === containerRef.current); + }; + + document.addEventListener('fullscreenchange', handleFullscreenChange); + + return () => { + document.removeEventListener('fullscreenchange', handleFullscreenChange); + }; + }, []); + + const toggleFullscreen = async () => { + if (!containerRef.current) { + return; + } + + try { + if (document.fullscreenElement === containerRef.current) { + await document.exitFullscreen(); + } else { + await containerRef.current.requestFullscreen(); + } + } catch (error) { + console.error('Failed to toggle fullscreen analytics chart', error); + } + }; + + return ( +
+ + + {tabs && activeTabId && onTabChange ? ( + + + {tabs.map((tab) => ( + + {tab.label} + + ))} + + + ) : null} +
+
+ {title} + {description ? ( +

+ {description} +

+ ) : null} +
+ +
+ + +
+ {metricChips.map((metric) => { + const hidden = hiddenSeries.has(metric.key); + + return ( + + ); + })} +
+ + + + + {dataKeys.map((dataKey) => ( + + + + + ))} + + + + + compactNumber.format( + typeof value === 'number' ? value : Number(value || 0) + ) + } + tickLine={false} + axisLine={false} + width={isMobile ? 40 : 48} + /> + + new Date(String(label)).toLocaleString() + } + /> + } + /> + {visibleDataKeys.map((dataKey) => ( + + ))} + + + {visibleDataKeys.length === 0 ? ( +

+ Select at least one metric to display the chart. +

+ ) : null} + + +
+ ); +} diff --git a/ui/components/usage-summary-cards.tsx b/ui/components/usage-summary-cards.tsx new file mode 100644 index 00000000..e578748a --- /dev/null +++ b/ui/components/usage-summary-cards.tsx @@ -0,0 +1,160 @@ +'use client'; + +import { Card, CardContent, CardHeader, CardTitle } from '@/components/ui/card'; +import { UsageSummary } from '@/lib/api/services/admin'; +import { + CheckCircle2, + XCircle, + AlertTriangle, + Activity, + Database, + CreditCard, + TrendingUp, + DollarSign, + TrendingDown, + Coins, +} from 'lucide-react'; +import { useCurrencyStore } from '@/lib/stores/currency'; +import { useQuery } from '@tanstack/react-query'; +import { fetchBtcUsdPrice, btcToSatsRate } from '@/lib/exchange-rate'; +import { formatFromMsat } from '@/lib/currency'; + +interface UsageSummaryCardsProps { + summary: UsageSummary; +} + +export function UsageSummaryCards({ summary }: UsageSummaryCardsProps) { + const { displayUnit } = useCurrencyStore(); + const { data: btcUsdPrice } = useQuery({ + queryKey: ['btc-usd-price'], + queryFn: fetchBtcUsdPrice, + refetchInterval: 120_000, + staleTime: 60_000, + }); + const usdPerSat = btcUsdPrice ? btcToSatsRate(btcUsdPrice) : null; + + const formatAmount = (msat: number) => + formatFromMsat(msat, displayUnit, usdPerSat); + const hasTokenStats = + typeof summary.total_tokens === 'number' || + typeof summary.avg_total_tokens_per_completion === 'number'; + + const cards = [ + { + title: 'Total Requests', + value: summary.total_requests.toLocaleString(), + icon: Activity, + iconClassName: 'text-blue-600 dark:text-blue-300', + }, + { + title: 'Successful Completions', + value: summary.successful_chat_completions.toLocaleString(), + icon: CheckCircle2, + iconClassName: 'text-emerald-600 dark:text-emerald-300', + }, + ...(hasTokenStats + ? [ + { + title: 'Total Tokens', + value: Number(summary.total_tokens ?? 0).toLocaleString(), + icon: Database, + iconClassName: 'text-cyan-600 dark:text-cyan-300', + }, + { + title: 'Avg Tokens/Completion', + value: Number( + summary.avg_total_tokens_per_completion ?? 0 + ).toLocaleString(undefined, { + maximumFractionDigits: 1, + }), + icon: Activity, + iconClassName: 'text-indigo-600 dark:text-indigo-300', + }, + ] + : []), + { + title: 'Revenue', + value: formatAmount(summary.revenue_msats), + icon: Coins, + iconClassName: 'text-amber-600 dark:text-amber-300', + }, + { + title: 'Operational Net', + value: formatAmount(summary.net_revenue_msats), + icon: DollarSign, + iconClassName: 'text-lime-600 dark:text-lime-300', + }, + { + title: 'Reverted Holds', + value: formatAmount(summary.refunds_msats), + icon: TrendingDown, + iconClassName: 'text-rose-600 dark:text-rose-300', + }, + { + title: 'Avg Revenue/Request', + value: formatAmount(summary.avg_revenue_per_request_msats), + icon: CreditCard, + iconClassName: 'text-violet-600 dark:text-violet-300', + }, + { + title: 'Success Rate', + value: `${summary.success_rate.toFixed(1)}%`, + icon: TrendingUp, + iconClassName: 'text-teal-600 dark:text-teal-300', + }, + { + title: 'Refund Rate', + value: `${summary.refund_rate.toFixed(1)}%`, + icon: XCircle, + iconClassName: 'text-fuchsia-600 dark:text-fuchsia-300', + }, + { + title: 'Failed Requests', + value: summary.failed_requests.toLocaleString(), + icon: XCircle, + iconClassName: 'text-red-600 dark:text-red-300', + }, + { + title: 'Errors', + value: summary.total_errors.toLocaleString(), + icon: AlertTriangle, + iconClassName: 'text-orange-600 dark:text-orange-300', + }, + { + title: 'Unique Models', + value: summary.unique_models_count.toLocaleString(), + icon: Database, + iconClassName: 'text-cyan-600 dark:text-cyan-300', + }, + { + title: 'Upstream Errors', + value: summary.upstream_errors.toLocaleString(), + icon: AlertTriangle, + iconClassName: 'text-pink-600 dark:text-pink-300', + }, + ]; + + return ( +
+ {cards.map((card) => ( + + + + {card.title} + + + + + + +
+ {card.value} +
+ + + ))} +
+ ); +} diff --git a/ui/lib/api/client.ts b/ui/lib/api/client.ts index f2bd1295..0975bf32 100644 --- a/ui/lib/api/client.ts +++ b/ui/lib/api/client.ts @@ -13,7 +13,11 @@ class ApiClient { private handleAuthError(error: unknown): void { if (axios.isAxiosError(error)) { const axiosError = error as AxiosError; - if (axiosError.response?.status === 401) { + const status = axiosError.response?.status; + const requestUrl = axiosError.config?.url ?? ''; + const isAdminRequest = requestUrl.includes('/admin/'); + + if (status === 401 || (status === 403 && isAdminRequest)) { ConfigurationService.clearToken(); if ( typeof window !== 'undefined' && diff --git a/ui/lib/api/services/admin.ts b/ui/lib/api/services/admin.ts index 91213afa..3ccaf5ab 100644 --- a/ui/lib/api/services/admin.ts +++ b/ui/lib/api/services/admin.ts @@ -833,6 +833,39 @@ export class AdminService { ); } + static async getUsageMetrics( + interval: number = 15, + hours: number = 24 + ): Promise { + return await apiClient.get( + `/admin/api/usage/metrics?interval=${interval}&hours=${hours}` + ); + } + + static async getUsageSummary(hours: number = 24): Promise { + return await apiClient.get( + `/admin/api/usage/summary?hours=${hours}` + ); + } + + static async getErrorDetails( + hours: number = 24, + limit: number = 100 + ): Promise { + return await apiClient.get( + `/admin/api/usage/error-details?hours=${hours}&limit=${limit}` + ); + } + + static async getRevenueByModel( + hours: number = 24, + limit: number = 20 + ): Promise { + return await apiClient.get( + `/admin/api/usage/revenue-by-model?hours=${hours}&limit=${limit}` + ); + } + static async createProviderAccountByType(providerType: string): Promise<{ ok: boolean; account_data: Record; @@ -897,6 +930,87 @@ export const TemporaryBalanceSchema = z.object({ export type TemporaryBalance = z.infer; +export interface UsageMetricData { + timestamp: string; + total_requests: number; + successful_chat_completions: number; + failed_requests: number; + errors: number; + warnings: number; + payment_processed: number; + upstream_errors: number; + revenue_msats: number; + refunds_msats: number; + input_tokens?: number; + output_tokens?: number; + total_tokens?: number; + [key: string]: unknown; +} + +export interface UsageMetrics { + metrics: UsageMetricData[]; + interval_minutes: number; + hours_back: number; + total_buckets: number; + totals?: Partial>; +} + +export interface UsageSummary { + total_entries: number; + total_requests: number; + successful_chat_completions: number; + failed_requests: number; + total_errors: number; + total_warnings: number; + payment_processed: number; + upstream_errors: number; + unique_models_count: number; + unique_models: string[]; + error_types: Record; + success_rate: number; + revenue_msats: number; + refunds_msats: number; + revenue_sats: number; + refunds_sats: number; + net_revenue_msats: number; + net_revenue_sats: number; + avg_revenue_per_request_msats: number; + refund_rate: number; + total_tokens?: number; + avg_total_tokens_per_completion?: number; +} + +export interface ErrorDetail { + timestamp: string; + message: string; + error_type: string; + pathname: string; + lineno: number; + request_id: string; +} + +export interface ErrorDetails { + errors: ErrorDetail[]; + total_count: number; +} + +export interface ModelRevenueData { + model: string; + revenue_sats: number; + refunds_sats: number; + net_revenue_sats: number; + requests: number; + successful: number; + failed: number; + avg_revenue_per_request: number; +} + +export interface RevenueByModel { + models: ModelRevenueData[]; + total_revenue_sats: number; + total_models: number; +} + export interface LogEntry { asctime: string; name: string;