mirror of
https://github.com/Routstr/routstr-core.git
synced 2026-07-30 23:36:15 +00:00
Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
1df66f48b8 | ||
|
|
a9c5458660 |
+41
-5
@@ -1,5 +1,7 @@
|
||||
"""Model prioritization algorithm for selecting cheapest upstream providers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from .core.logging import get_logger
|
||||
@@ -157,15 +159,21 @@ def create_model_mappings(
|
||||
upstreams: list["BaseUpstreamProvider"],
|
||||
overrides_by_id: dict[str, tuple],
|
||||
disabled_model_ids: set[str],
|
||||
) -> tuple[dict[str, "Model"], dict[str, "BaseUpstreamProvider"], dict[str, "Model"]]:
|
||||
) -> tuple[
|
||||
dict[str, "Model"],
|
||||
dict[str, "BaseUpstreamProvider"],
|
||||
dict[str, "Model"],
|
||||
dict[str, list["BaseUpstreamProvider"]],
|
||||
]:
|
||||
"""Create optimal model mappings based on cost and provider preferences.
|
||||
|
||||
This is the main entry point for the algorithm. It processes all upstream providers
|
||||
and creates three mappings based on cost optimization:
|
||||
|
||||
1. model_instances: alias -> Model (all model aliases mapped to their Model objects)
|
||||
2. provider_map: alias -> UpstreamProvider (which provider to use for each alias)
|
||||
2. provider_map: alias -> UpstreamProvider (the BEST provider to use for each alias)
|
||||
3. unique_models: base_id -> Model (unique models without provider prefixes)
|
||||
4. provider_candidates_map: alias -> list[UpstreamProvider] (all providers offering the model, sorted by preference)
|
||||
|
||||
The algorithm:
|
||||
- Processes non-OpenRouter providers first (they're typically cheaper)
|
||||
@@ -178,7 +186,7 @@ def create_model_mappings(
|
||||
disabled_model_ids: Set of model IDs that should be excluded
|
||||
|
||||
Returns:
|
||||
Tuple of (model_instances, provider_map, unique_models)
|
||||
Tuple of (model_instances, provider_map, unique_models, provider_candidates_map)
|
||||
"""
|
||||
from .payment.models import _row_to_model
|
||||
from .upstream.helpers import resolve_model_alias
|
||||
@@ -186,9 +194,10 @@ def create_model_mappings(
|
||||
model_instances: dict[str, "Model"] = {}
|
||||
provider_map: dict[str, "BaseUpstreamProvider"] = {}
|
||||
unique_models: dict[str, "Model"] = {}
|
||||
provider_candidates_map: dict[str, list["BaseUpstreamProvider"]] = {}
|
||||
|
||||
# Separate OpenRouter from other providers
|
||||
openrouter: "BaseUpstreamProvider" | None = None
|
||||
openrouter: BaseUpstreamProvider | None = None
|
||||
other_upstreams: list["BaseUpstreamProvider"] = []
|
||||
|
||||
for upstream in upstreams:
|
||||
@@ -207,6 +216,13 @@ def create_model_mappings(
|
||||
) -> None:
|
||||
"""Set alias to model/provider if not set or if new model is preferred."""
|
||||
alias_lower = alias.lower()
|
||||
|
||||
# Add to candidates list, to be used later as fallback
|
||||
if alias_lower not in provider_candidates_map:
|
||||
provider_candidates_map[alias_lower] = [provider]
|
||||
else:
|
||||
provider_candidates_map[alias_lower].append(provider)
|
||||
|
||||
existing_model = model_instances.get(alias_lower)
|
||||
if not existing_model:
|
||||
# No existing mapping, set it
|
||||
@@ -276,6 +292,26 @@ def create_model_mappings(
|
||||
if openrouter:
|
||||
process_provider_models(openrouter, is_openrouter=True)
|
||||
|
||||
# Sort and filter provider candidates for each alias using provider_map as reference
|
||||
# We only keep entries that have more than one provider.
|
||||
final_candidates_map: dict[str, list["BaseUpstreamProvider"]] = {}
|
||||
for alias_lower, best_provider in provider_map.items():
|
||||
candidates = provider_candidates_map.get(alias_lower, [])
|
||||
# Remove duplicates
|
||||
unique_candidates = []
|
||||
seen = set()
|
||||
for c in candidates:
|
||||
if c not in seen:
|
||||
unique_candidates.append(c)
|
||||
seen.add(c)
|
||||
|
||||
if len(unique_candidates) > 1:
|
||||
# Keep the best one at the front, others follow.
|
||||
if best_provider in unique_candidates:
|
||||
unique_candidates.remove(best_provider)
|
||||
unique_candidates.insert(0, best_provider)
|
||||
final_candidates_map[alias_lower] = unique_candidates
|
||||
|
||||
# Log provider distribution
|
||||
provider_counts: dict[str, int] = {}
|
||||
for provider in provider_map.values():
|
||||
@@ -287,4 +323,4 @@ def create_model_mappings(
|
||||
extra={"provider_distribution": provider_counts},
|
||||
)
|
||||
|
||||
return model_instances, provider_map, unique_models
|
||||
return model_instances, provider_map, unique_models, provider_candidates_map
|
||||
|
||||
+20
-50
@@ -441,29 +441,6 @@ async def adjust_payment_for_tokens(
|
||||
},
|
||||
)
|
||||
|
||||
async def release_reservation_only() -> None:
|
||||
"""Fallback to release reservation without charging when main update fails."""
|
||||
try:
|
||||
release_stmt = (
|
||||
update(ApiKey)
|
||||
.where(col(ApiKey.hashed_key) == key.hashed_key)
|
||||
.values(reserved_balance=col(ApiKey.reserved_balance) - deducted_max_cost)
|
||||
)
|
||||
await session.exec(release_stmt) # type: ignore[call-overload]
|
||||
await session.commit()
|
||||
logger.warning(
|
||||
"Released reservation without charging (fallback)",
|
||||
extra={
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"deducted_max_cost": deducted_max_cost,
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Failed to release reservation in fallback",
|
||||
extra={"error": str(e), "key_hash": key.hashed_key[:8] + "..."},
|
||||
)
|
||||
|
||||
match await calculate_cost(response_data, deducted_max_cost, session):
|
||||
case MaxCostData() as cost:
|
||||
logger.debug(
|
||||
@@ -488,7 +465,7 @@ async def adjust_payment_for_tokens(
|
||||
await session.commit()
|
||||
if result.rowcount == 0:
|
||||
logger.error(
|
||||
"Failed to finalize max-cost payment - retrying reservation release",
|
||||
"Failed to finalize max-cost payment - insufficient reserved balance",
|
||||
extra={
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"deducted_max_cost": deducted_max_cost,
|
||||
@@ -497,7 +474,6 @@ async def adjust_payment_for_tokens(
|
||||
"model": model,
|
||||
},
|
||||
)
|
||||
await release_reservation_only()
|
||||
else:
|
||||
await session.refresh(key)
|
||||
logger.info(
|
||||
@@ -592,14 +568,13 @@ async def adjust_payment_for_tokens(
|
||||
)
|
||||
else:
|
||||
logger.warning(
|
||||
"Failed to finalize additional charge - releasing reservation",
|
||||
"Failed to finalize additional charge (concurrent operation)",
|
||||
extra={
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"attempted_charge": total_cost_msats,
|
||||
"model": model,
|
||||
},
|
||||
)
|
||||
await release_reservation_only()
|
||||
else:
|
||||
# Refund some of the base cost
|
||||
refund = abs(cost_difference)
|
||||
@@ -628,7 +603,7 @@ async def adjust_payment_for_tokens(
|
||||
|
||||
if result.rowcount == 0:
|
||||
logger.error(
|
||||
"Failed to finalize payment - releasing reservation",
|
||||
"Failed to finalize payment - insufficient reserved balance",
|
||||
extra={
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"deducted_max_cost": deducted_max_cost,
|
||||
@@ -637,27 +612,28 @@ async def adjust_payment_for_tokens(
|
||||
"model": model,
|
||||
},
|
||||
)
|
||||
await release_reservation_only()
|
||||
else:
|
||||
cost.total_msats = total_cost_msats
|
||||
await session.refresh(key)
|
||||
# Still return the cost data even if we couldn't properly finalize
|
||||
# The reservation was already made, so the user has paid
|
||||
|
||||
logger.info(
|
||||
"Refund processed successfully",
|
||||
extra={
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"refunded_amount": refund,
|
||||
"new_balance": key.balance,
|
||||
"final_cost": cost.total_msats,
|
||||
"model": model,
|
||||
},
|
||||
)
|
||||
cost.total_msats = total_cost_msats
|
||||
await session.refresh(key)
|
||||
|
||||
logger.info(
|
||||
"Refund processed successfully",
|
||||
extra={
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"refunded_amount": refund,
|
||||
"new_balance": key.balance,
|
||||
"final_cost": cost.total_msats,
|
||||
"model": model,
|
||||
},
|
||||
)
|
||||
|
||||
return cost.dict()
|
||||
|
||||
case CostDataError() as error:
|
||||
logger.error(
|
||||
"Cost calculation error during payment adjustment - releasing reservation",
|
||||
"Cost calculation error during payment adjustment",
|
||||
extra={
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"model": model,
|
||||
@@ -665,7 +641,6 @@ async def adjust_payment_for_tokens(
|
||||
"error_code": error.code,
|
||||
},
|
||||
)
|
||||
await release_reservation_only()
|
||||
|
||||
raise HTTPException(
|
||||
status_code=400,
|
||||
@@ -677,12 +652,7 @@ async def adjust_payment_for_tokens(
|
||||
}
|
||||
},
|
||||
)
|
||||
# Fallback: should not reach here, but release reservation just in case
|
||||
logger.error(
|
||||
"Unexpected fallback in adjust_payment_for_tokens - releasing reservation",
|
||||
extra={"key_hash": key.hashed_key[:8] + "...", "model": model},
|
||||
)
|
||||
await release_reservation_only()
|
||||
# Fallback return to satisfy type checker; execution should not reach here
|
||||
return {
|
||||
"base_msats": deducted_max_cost,
|
||||
"input_msats": 0,
|
||||
|
||||
@@ -3080,9 +3080,6 @@ async def get_logs_api(
|
||||
level: str | None = None,
|
||||
request_id: str | None = None,
|
||||
search: str | None = None,
|
||||
status_codes: str | None = Query(None, description="Comma-separated status codes"),
|
||||
methods: str | None = Query(None, description="Comma-separated HTTP methods"),
|
||||
endpoints: str | None = Query(None, description="Comma-separated endpoints"),
|
||||
limit: int = 100,
|
||||
) -> dict[str, object]:
|
||||
"""
|
||||
@@ -3093,32 +3090,16 @@ async def get_logs_api(
|
||||
level: Filter by log level
|
||||
request_id: Filter by request ID
|
||||
search: Search text in message and name fields (case-insensitive)
|
||||
status_codes: Comma-separated list of HTTP status codes
|
||||
methods: Comma-separated list of HTTP methods
|
||||
endpoints: Comma-separated list of endpoints
|
||||
limit: Maximum number of entries to return
|
||||
|
||||
Returns:
|
||||
Dict containing logs and filter metadata
|
||||
"""
|
||||
status_code_list = None
|
||||
if status_codes:
|
||||
try:
|
||||
status_code_list = [int(s.strip()) for s in status_codes.split(",")]
|
||||
except ValueError:
|
||||
pass
|
||||
|
||||
method_list = [m.strip() for m in methods.split(",")] if methods else None
|
||||
endpoint_list = [e.strip() for e in endpoints.split(",")] if endpoints else None
|
||||
|
||||
log_entries = log_manager.search_logs(
|
||||
date=date,
|
||||
level=level,
|
||||
request_id=request_id,
|
||||
search_text=search,
|
||||
status_codes=status_code_list,
|
||||
methods=method_list,
|
||||
endpoints=endpoint_list,
|
||||
limit=limit,
|
||||
)
|
||||
|
||||
@@ -3129,9 +3110,6 @@ async def get_logs_api(
|
||||
"level": level,
|
||||
"request_id": request_id,
|
||||
"search": search,
|
||||
"status_codes": status_codes,
|
||||
"methods": methods,
|
||||
"endpoints": endpoints,
|
||||
"limit": limit,
|
||||
}
|
||||
|
||||
|
||||
@@ -105,9 +105,6 @@ class LogManager:
|
||||
level: str | None = None,
|
||||
request_id: str | None = None,
|
||||
search_text: str | None = None,
|
||||
status_codes: list[int] | None = None,
|
||||
methods: list[str] | None = None,
|
||||
endpoints: list[str] | None = None,
|
||||
limit: int = 100,
|
||||
) -> list[dict[str, Any]]:
|
||||
"""
|
||||
@@ -137,13 +134,7 @@ class LogManager:
|
||||
|
||||
for log_data in iterator:
|
||||
if not self._matches_filters(
|
||||
log_data,
|
||||
level,
|
||||
request_id,
|
||||
search_text_lower,
|
||||
status_codes,
|
||||
methods,
|
||||
endpoints,
|
||||
log_data, level, request_id, search_text_lower
|
||||
):
|
||||
continue
|
||||
|
||||
@@ -162,9 +153,6 @@ class LogManager:
|
||||
level: str | None,
|
||||
request_id: str | None,
|
||||
search_text_lower: str | None,
|
||||
status_codes: list[int] | None = None,
|
||||
methods: list[str] | None = None,
|
||||
endpoints: list[str] | None = None,
|
||||
) -> bool:
|
||||
if level and log_data.get("levelname", "").upper() != level.upper():
|
||||
return False
|
||||
@@ -172,36 +160,6 @@ class LogManager:
|
||||
if request_id and log_data.get("request_id") != request_id:
|
||||
return False
|
||||
|
||||
if status_codes:
|
||||
entry_status = log_data.get("status_code")
|
||||
if entry_status is not None:
|
||||
try:
|
||||
if int(entry_status) not in status_codes:
|
||||
return False
|
||||
except (ValueError, TypeError):
|
||||
return False
|
||||
else:
|
||||
return False
|
||||
|
||||
if methods:
|
||||
entry_method = log_data.get("method", "").upper()
|
||||
if entry_method not in [m.upper() for m in methods]:
|
||||
return False
|
||||
|
||||
if endpoints:
|
||||
entry_path = log_data.get("path", "")
|
||||
matched = False
|
||||
for endpoint in endpoints:
|
||||
clean_endpoint = endpoint.lstrip("/")
|
||||
if entry_path.startswith(clean_endpoint):
|
||||
matched = True
|
||||
break
|
||||
if clean_endpoint in entry_path:
|
||||
matched = True
|
||||
break
|
||||
if not matched:
|
||||
return False
|
||||
|
||||
if search_text_lower:
|
||||
message = str(log_data.get("message", "")).lower()
|
||||
name = str(log_data.get("name", "")).lower()
|
||||
|
||||
@@ -48,6 +48,13 @@ async def calculate_cost( # todo: can be sync
|
||||
},
|
||||
)
|
||||
|
||||
cost_data = MaxCostData(
|
||||
base_msats=max_cost,
|
||||
input_msats=0,
|
||||
output_msats=0,
|
||||
total_msats=max_cost,
|
||||
)
|
||||
|
||||
if "usage" not in response_data or response_data["usage"] is None:
|
||||
logger.warning(
|
||||
"No usage data in response, using base cost only",
|
||||
@@ -56,12 +63,7 @@ async def calculate_cost( # todo: can be sync
|
||||
"model": response_data.get("model", "unknown"),
|
||||
},
|
||||
)
|
||||
return MaxCostData(
|
||||
base_msats=0,
|
||||
input_msats=0,
|
||||
output_msats=0,
|
||||
total_msats=0,
|
||||
)
|
||||
return cost_data
|
||||
|
||||
usage_data = response_data["usage"]
|
||||
|
||||
@@ -176,12 +178,7 @@ async def calculate_cost( # todo: can be sync
|
||||
"model": response_data.get("model", "unknown"),
|
||||
},
|
||||
)
|
||||
return MaxCostData(
|
||||
base_msats=max_cost,
|
||||
input_msats=0,
|
||||
output_msats=0,
|
||||
total_msats=max_cost,
|
||||
)
|
||||
return cost_data
|
||||
|
||||
input_tokens = usage_data.get("prompt_tokens", 0)
|
||||
output_tokens = usage_data.get("completion_tokens", 0)
|
||||
@@ -195,16 +192,8 @@ async def calculate_cost( # todo: can be sync
|
||||
)
|
||||
|
||||
# added for response api
|
||||
input_tokens = (
|
||||
input_tokens
|
||||
if input_tokens != 0
|
||||
else response_data.get("usage", {}).get("input_tokens", 0)
|
||||
)
|
||||
output_tokens = (
|
||||
output_tokens
|
||||
if output_tokens != 0
|
||||
else response_data.get("usage", {}).get("output_tokens", 0)
|
||||
)
|
||||
input_tokens = input_tokens if input_tokens != 0 else response_data.get("usage", {}).get("input_tokens", 0)
|
||||
output_tokens = output_tokens if output_tokens != 0 else response_data.get("usage", {}).get("output_tokens", 0)
|
||||
|
||||
input_msats = round(input_tokens / 1000 * MSATS_PER_1K_INPUT_TOKENS, 3)
|
||||
|
||||
|
||||
+192
-59
@@ -1,6 +1,7 @@
|
||||
import json
|
||||
from typing import Any
|
||||
|
||||
import httpx
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request
|
||||
from fastapi.responses import Response, StreamingResponse
|
||||
from sqlmodel import select
|
||||
@@ -26,7 +27,7 @@ from .payment.helpers import (
|
||||
from .payment.models import Model
|
||||
from .upstream import BaseUpstreamProvider
|
||||
from .upstream.helpers import init_upstreams
|
||||
from .wallet import deserialize_token_from_string
|
||||
from .wallet import deserialize_token_from_string, recieve_token
|
||||
|
||||
logger = get_logger(__name__)
|
||||
proxy_router = APIRouter()
|
||||
@@ -34,6 +35,9 @@ proxy_router = APIRouter()
|
||||
_upstreams: list[BaseUpstreamProvider] = []
|
||||
_model_instances: dict[str, Model] = {} # All aliases -> Model
|
||||
_provider_map: dict[str, BaseUpstreamProvider] = {} # All aliases -> Provider
|
||||
_provider_candidates_map: dict[
|
||||
str, list[BaseUpstreamProvider]
|
||||
] = {} # All aliases -> [Providers]
|
||||
_unique_models: dict[str, Model] = {} # Unique model.id -> Model (no duplicates)
|
||||
|
||||
|
||||
@@ -75,6 +79,21 @@ def get_provider_for_model(model_id: str) -> BaseUpstreamProvider | None:
|
||||
return _provider_map.get(model_id.lower())
|
||||
|
||||
|
||||
def get_providers_for_model(model_id: str) -> list[BaseUpstreamProvider]:
|
||||
"""Get list of prioritized UpstreamProviders for model ID from global cache.
|
||||
|
||||
If multiple providers are available, returns the sorted list.
|
||||
Otherwise, returns a list containing only the single best provider.
|
||||
"""
|
||||
candidates = _provider_candidates_map.get(model_id.lower(), [])
|
||||
if candidates:
|
||||
return candidates
|
||||
|
||||
# Fallback to the single best provider if no multi-provider candidates exist
|
||||
best = get_provider_for_model(model_id)
|
||||
return [best] if best else []
|
||||
|
||||
|
||||
def get_unique_models() -> list[Model]:
|
||||
"""Get list of unique models (no duplicates from aliases)."""
|
||||
return list(_unique_models.values())
|
||||
@@ -84,7 +103,7 @@ async def refresh_model_maps() -> None:
|
||||
"""Refresh global model and provider maps using the cost-based algorithm."""
|
||||
from sqlalchemy.orm import selectinload
|
||||
|
||||
global _model_instances, _provider_map, _unique_models
|
||||
global _model_instances, _provider_map, _unique_models, _provider_candidates_map
|
||||
|
||||
async with create_session() as session:
|
||||
# Fetch all providers with their models in a single logical operation
|
||||
@@ -104,10 +123,12 @@ async def refresh_model_maps() -> None:
|
||||
else:
|
||||
disabled_model_ids.add(model.id)
|
||||
|
||||
_model_instances, _provider_map, _unique_models = create_model_mappings(
|
||||
upstreams=_upstreams,
|
||||
overrides_by_id=overrides_by_id,
|
||||
disabled_model_ids=disabled_model_ids,
|
||||
_model_instances, _provider_map, _unique_models, _provider_candidates_map = (
|
||||
create_model_mappings(
|
||||
upstreams=_upstreams,
|
||||
overrides_by_id=overrides_by_id,
|
||||
disabled_model_ids=disabled_model_ids,
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
@@ -172,8 +193,8 @@ async def proxy(
|
||||
"invalid_model", f"Model '{model_id}' not found", 400, request=request
|
||||
)
|
||||
|
||||
upstream = get_provider_for_model(model_id)
|
||||
if not upstream:
|
||||
upstreams = get_providers_for_model(model_id)
|
||||
if not upstreams:
|
||||
return create_error_response(
|
||||
"invalid_model",
|
||||
f"No provider found for model '{model_id}'",
|
||||
@@ -190,14 +211,80 @@ async def proxy(
|
||||
check_token_balance(headers, request_body_dict, max_cost_for_model)
|
||||
|
||||
if x_cashu := headers.get("x-cashu", None):
|
||||
if is_responses_api:
|
||||
return await upstream.handle_x_cashu_responses(
|
||||
request, x_cashu, path, max_cost_for_model, model_obj
|
||||
)
|
||||
else:
|
||||
return await upstream.handle_x_cashu(
|
||||
request, x_cashu, path, max_cost_for_model, model_obj
|
||||
)
|
||||
# Redeem token once before trying any providers
|
||||
amount, unit, mint = await recieve_token(x_cashu)
|
||||
|
||||
# Fallback for X-Cashu payments
|
||||
last_exception = None
|
||||
for i, upstream in enumerate(upstreams):
|
||||
try:
|
||||
# Prepare headers for this specific upstream
|
||||
upstream_headers = upstream.prepare_headers(dict(request.headers))
|
||||
|
||||
if is_responses_api:
|
||||
return await upstream.forward_x_cashu_responses_request(
|
||||
request,
|
||||
path,
|
||||
upstream_headers,
|
||||
amount,
|
||||
unit,
|
||||
max_cost_for_model,
|
||||
model_obj,
|
||||
mint,
|
||||
)
|
||||
else:
|
||||
return await upstream.forward_x_cashu_request(
|
||||
request,
|
||||
path,
|
||||
upstream_headers,
|
||||
amount,
|
||||
unit,
|
||||
max_cost_for_model,
|
||||
model_obj,
|
||||
mint,
|
||||
)
|
||||
except (httpx.TimeoutException, httpx.ConnectError) as e:
|
||||
logger.warning(
|
||||
f"Upstream provider {i + 1}/{len(upstreams)} ({upstream.provider_type}) timed out, trying fallback",
|
||||
extra={
|
||||
"model": model_id,
|
||||
"error": str(e),
|
||||
"attempt": i + 1,
|
||||
},
|
||||
)
|
||||
last_exception = e
|
||||
continue
|
||||
|
||||
# If we get here, all providers failed
|
||||
# Since the token was already redeemed, we must issue a refund
|
||||
logger.error(
|
||||
"All providers failed for X-Cashu request, issuing emergency refund",
|
||||
extra={"amount": amount, "unit": unit, "mint": mint},
|
||||
)
|
||||
|
||||
# Try to use the first provider's refund mechanism
|
||||
refund_token = await upstreams[0].send_refund(amount - 60, unit, mint)
|
||||
|
||||
error_message = "All upstream providers timed out"
|
||||
if isinstance(last_exception, httpx.ConnectError):
|
||||
error_message = "Unable to connect to any upstream service"
|
||||
|
||||
error_response = Response(
|
||||
content=json.dumps(
|
||||
{
|
||||
"error": {
|
||||
"message": error_message,
|
||||
"type": "upstream_error",
|
||||
"code": 504,
|
||||
"refund_token": refund_token,
|
||||
}
|
||||
}
|
||||
),
|
||||
status_code=504,
|
||||
media_type="application/json",
|
||||
)
|
||||
error_response.headers["X-Cashu"] = refund_token
|
||||
return error_response
|
||||
|
||||
elif auth := headers.get("authorization", None):
|
||||
key = await get_bearer_token_key(headers, path, session, auth)
|
||||
@@ -212,56 +299,102 @@ async def proxy(
|
||||
)
|
||||
|
||||
logger.debug("Processing unauthenticated GET request", extra={"path": path})
|
||||
headers = upstream.prepare_headers(dict(request.headers))
|
||||
return await upstream.forward_get_request(request, path, headers)
|
||||
|
||||
# Try fallback for GET requests too
|
||||
last_exception = None
|
||||
for i, upstream in enumerate(upstreams):
|
||||
try:
|
||||
headers = upstream.prepare_headers(dict(request.headers))
|
||||
return await upstream.forward_get_request(request, path, headers)
|
||||
except (httpx.TimeoutException, httpx.ConnectError) as e:
|
||||
logger.warning(
|
||||
f"Upstream GET provider {i + 1}/{len(upstreams)} ({upstream.provider_type}) timed out, trying fallback",
|
||||
extra={"path": path, "error": str(e)},
|
||||
)
|
||||
last_exception = e
|
||||
continue
|
||||
|
||||
error_message = "Upstream service request timed out"
|
||||
if isinstance(last_exception, httpx.ConnectError):
|
||||
error_message = "Unable to connect to upstream service"
|
||||
|
||||
return create_error_response(
|
||||
"upstream_error", error_message, 502, request=request
|
||||
)
|
||||
|
||||
if request_body_dict:
|
||||
await pay_for_request(key, max_cost_for_model, session)
|
||||
|
||||
headers = upstream.prepare_headers(dict(request.headers))
|
||||
# Fallback for API Key payments
|
||||
last_exception = None
|
||||
for i, upstream in enumerate(upstreams):
|
||||
try:
|
||||
headers = upstream.prepare_headers(dict(request.headers))
|
||||
if is_responses_api:
|
||||
response = await upstream.forward_responses_request(
|
||||
request,
|
||||
path,
|
||||
headers,
|
||||
request_body,
|
||||
key,
|
||||
max_cost_for_model,
|
||||
session,
|
||||
model_obj,
|
||||
)
|
||||
else:
|
||||
response = await upstream.forward_request(
|
||||
request,
|
||||
path,
|
||||
headers,
|
||||
request_body,
|
||||
key,
|
||||
max_cost_for_model,
|
||||
session,
|
||||
model_obj,
|
||||
)
|
||||
|
||||
if is_responses_api:
|
||||
response = await upstream.forward_responses_request(
|
||||
request,
|
||||
path,
|
||||
headers,
|
||||
request_body,
|
||||
key,
|
||||
max_cost_for_model,
|
||||
session,
|
||||
model_obj,
|
||||
)
|
||||
else:
|
||||
response = await upstream.forward_request(
|
||||
request,
|
||||
path,
|
||||
headers,
|
||||
request_body,
|
||||
key,
|
||||
max_cost_for_model,
|
||||
session,
|
||||
model_obj,
|
||||
)
|
||||
if response.status_code != 200:
|
||||
# If it's a 429 (rate limit) or 503 (service unavailable), we might also want to fallback
|
||||
if response.status_code in (429, 503, 502) and i < len(upstreams) - 1:
|
||||
logger.warning(
|
||||
f"Upstream provider {i + 1}/{len(upstreams)} returned {response.status_code}, trying fallback",
|
||||
extra={"model": model_id, "status": response.status_code},
|
||||
)
|
||||
continue
|
||||
|
||||
if response.status_code != 200:
|
||||
await revert_pay_for_request(key, session, max_cost_for_model)
|
||||
logger.warning(
|
||||
"Upstream request failed, revert payment",
|
||||
extra={
|
||||
"status_code": response.status_code,
|
||||
"path": path,
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"key_balance": key.balance,
|
||||
"max_cost_for_model": max_cost_for_model,
|
||||
"upstream_headers": response.headers
|
||||
if hasattr(response, "headers")
|
||||
else None,
|
||||
},
|
||||
)
|
||||
# Return the mapped error response generated earlier rather than masking with 502
|
||||
return response
|
||||
await revert_pay_for_request(key, session, max_cost_for_model)
|
||||
logger.warning(
|
||||
"Upstream request failed, revert payment",
|
||||
extra={
|
||||
"status_code": response.status_code,
|
||||
"path": path,
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
"key_balance": key.balance,
|
||||
"max_cost_for_model": max_cost_for_model,
|
||||
"upstream_headers": response.headers
|
||||
if hasattr(response, "headers")
|
||||
else None,
|
||||
},
|
||||
)
|
||||
return response
|
||||
|
||||
return response
|
||||
return response
|
||||
|
||||
except (httpx.TimeoutException, httpx.ConnectError) as e:
|
||||
logger.warning(
|
||||
f"Upstream provider {i + 1}/{len(upstreams)} ({upstream.provider_type}) timed out, trying fallback",
|
||||
extra={"model": model_id, "error": str(e)},
|
||||
)
|
||||
last_exception = e
|
||||
continue
|
||||
|
||||
# All providers failed with timeout/connect error
|
||||
await revert_pay_for_request(key, session, max_cost_for_model)
|
||||
error_message = "Upstream service request timed out"
|
||||
if isinstance(last_exception, httpx.ConnectError):
|
||||
error_message = "Unable to connect to upstream service"
|
||||
|
||||
return create_error_response("upstream_error", error_message, 502, request=request)
|
||||
|
||||
|
||||
async def get_bearer_token_key(
|
||||
|
||||
+24
-76
@@ -37,6 +37,8 @@ from ..wallet import recieve_token, send_token
|
||||
|
||||
logger = get_logger(__name__)
|
||||
|
||||
DEFAULT_PROXY_TIMEOUT = 30.0
|
||||
|
||||
|
||||
class TopupData(BaseModel):
|
||||
"""Universal top-up data schema for Lightning Network invoices."""
|
||||
@@ -447,11 +449,6 @@ class BaseUpstreamProvider:
|
||||
async with create_session() as new_session:
|
||||
fresh_key = await new_session.get(key.__class__, key.hashed_key)
|
||||
if not fresh_key:
|
||||
logger.warning(
|
||||
"Key not found when finalizing streaming payment",
|
||||
extra={"key_hash": key.hashed_key[:8] + "..."},
|
||||
)
|
||||
usage_finalized = True
|
||||
return None
|
||||
try:
|
||||
fallback: dict = {
|
||||
@@ -480,7 +477,6 @@ class BaseUpstreamProvider:
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
},
|
||||
)
|
||||
usage_finalized = True
|
||||
return None
|
||||
|
||||
try:
|
||||
@@ -590,10 +586,8 @@ class BaseUpstreamProvider:
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
},
|
||||
)
|
||||
await finalize_without_usage()
|
||||
raise
|
||||
finally:
|
||||
if not usage_finalized:
|
||||
await finalize_without_usage()
|
||||
|
||||
# Remove inaccurate encoding headers from upstream response
|
||||
response_headers = dict(response.headers)
|
||||
@@ -698,6 +692,7 @@ class BaseUpstreamProvider:
|
||||
},
|
||||
)
|
||||
raise
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Error processing non-streaming chat completion",
|
||||
@@ -746,11 +741,6 @@ class BaseUpstreamProvider:
|
||||
async with create_session() as new_session:
|
||||
fresh_key = await new_session.get(key.__class__, key.hashed_key)
|
||||
if not fresh_key:
|
||||
logger.warning(
|
||||
"Key not found when finalizing Responses API streaming payment",
|
||||
extra={"key_hash": key.hashed_key[:8] + "..."},
|
||||
)
|
||||
usage_finalized = True
|
||||
return None
|
||||
try:
|
||||
fallback: dict = {
|
||||
@@ -779,7 +769,6 @@ class BaseUpstreamProvider:
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
},
|
||||
)
|
||||
usage_finalized = True
|
||||
return None
|
||||
|
||||
try:
|
||||
@@ -904,10 +893,8 @@ class BaseUpstreamProvider:
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
},
|
||||
)
|
||||
await finalize_without_usage()
|
||||
raise
|
||||
finally:
|
||||
if not usage_finalized:
|
||||
await finalize_without_usage()
|
||||
|
||||
# Remove inaccurate encoding headers from upstream response
|
||||
response_headers = dict(response.headers)
|
||||
@@ -1026,44 +1013,6 @@ class BaseUpstreamProvider:
|
||||
)
|
||||
raise
|
||||
|
||||
async def _finalize_generic_streaming_payment(
|
||||
self, key_hash: str, max_cost: int, path: str
|
||||
) -> None:
|
||||
"""Background task to finalize payment for generic streaming requests."""
|
||||
async with create_session() as session:
|
||||
key = await session.get(ApiKey, key_hash)
|
||||
if not key:
|
||||
logger.warning(
|
||||
"Key not found during background payment finalization",
|
||||
extra={"key_hash": key_hash[:8] + "..."},
|
||||
)
|
||||
return
|
||||
|
||||
try:
|
||||
# Finalize with "unknown" model and no usage to release reservation/charge max cost
|
||||
await adjust_payment_for_tokens(
|
||||
key,
|
||||
{"model": "unknown", "usage": None},
|
||||
session,
|
||||
max_cost,
|
||||
)
|
||||
logger.info(
|
||||
"Finalized generic streaming payment in background",
|
||||
extra={
|
||||
"path": path,
|
||||
"key_hash": key_hash[:8] + "...",
|
||||
},
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Error finalizing generic streaming payment in background",
|
||||
extra={
|
||||
"error": str(e),
|
||||
"key_hash": key_hash[:8] + "...",
|
||||
"path": path,
|
||||
},
|
||||
)
|
||||
|
||||
async def forward_request(
|
||||
self,
|
||||
request: Request,
|
||||
@@ -1110,7 +1059,7 @@ class BaseUpstreamProvider:
|
||||
|
||||
client = httpx.AsyncClient(
|
||||
transport=httpx.AsyncHTTPTransport(retries=1),
|
||||
timeout=None,
|
||||
timeout=DEFAULT_PROXY_TIMEOUT,
|
||||
)
|
||||
|
||||
try:
|
||||
@@ -1215,12 +1164,6 @@ class BaseUpstreamProvider:
|
||||
background_tasks = BackgroundTasks()
|
||||
background_tasks.add_task(response.aclose)
|
||||
background_tasks.add_task(client.aclose)
|
||||
background_tasks.add_task(
|
||||
self._finalize_generic_streaming_payment,
|
||||
key.hashed_key,
|
||||
max_cost_for_model,
|
||||
path,
|
||||
)
|
||||
|
||||
logger.debug(
|
||||
"Streaming non-chat response",
|
||||
@@ -1259,7 +1202,8 @@ class BaseUpstreamProvider:
|
||||
if isinstance(exc, httpx.ConnectError):
|
||||
error_message = "Unable to connect to upstream service"
|
||||
elif isinstance(exc, httpx.TimeoutException):
|
||||
error_message = "Upstream service request timed out"
|
||||
# Re-raise timeout exception to allow fallback handling in proxy layer
|
||||
raise
|
||||
elif isinstance(exc, httpx.NetworkError):
|
||||
error_message = "Network error while connecting to upstream service"
|
||||
else:
|
||||
@@ -1342,7 +1286,7 @@ class BaseUpstreamProvider:
|
||||
|
||||
client = httpx.AsyncClient(
|
||||
transport=httpx.AsyncHTTPTransport(retries=1),
|
||||
timeout=None,
|
||||
timeout=DEFAULT_PROXY_TIMEOUT,
|
||||
)
|
||||
|
||||
try:
|
||||
@@ -1424,15 +1368,9 @@ class BaseUpstreamProvider:
|
||||
background_tasks = BackgroundTasks()
|
||||
background_tasks.add_task(response.aclose)
|
||||
background_tasks.add_task(client.aclose)
|
||||
background_tasks.add_task(
|
||||
self._finalize_generic_streaming_payment,
|
||||
key.hashed_key,
|
||||
max_cost_for_model,
|
||||
path,
|
||||
)
|
||||
|
||||
logger.debug(
|
||||
"Streaming non-Responses API response",
|
||||
"Streaming non-chat response",
|
||||
extra={
|
||||
"path": path,
|
||||
"status_code": response.status_code,
|
||||
@@ -1468,7 +1406,8 @@ class BaseUpstreamProvider:
|
||||
if isinstance(exc, httpx.ConnectError):
|
||||
error_message = "Unable to connect to upstream service"
|
||||
elif isinstance(exc, httpx.TimeoutException):
|
||||
error_message = "Upstream service request timed out"
|
||||
# Re-raise timeout exception to allow fallback handling in proxy layer
|
||||
raise
|
||||
elif isinstance(exc, httpx.NetworkError):
|
||||
error_message = "Network error while connecting to upstream service"
|
||||
else:
|
||||
@@ -1531,7 +1470,7 @@ class BaseUpstreamProvider:
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=httpx.AsyncHTTPTransport(retries=1),
|
||||
timeout=None,
|
||||
timeout=DEFAULT_PROXY_TIMEOUT,
|
||||
) as client:
|
||||
try:
|
||||
response = await client.send(
|
||||
@@ -1562,6 +1501,9 @@ class BaseUpstreamProvider:
|
||||
status_code=response.status_code,
|
||||
headers=dict(response.headers),
|
||||
)
|
||||
except (httpx.TimeoutException, httpx.ConnectError):
|
||||
# Re-raise to allow fallback handling in proxy layer
|
||||
raise
|
||||
except Exception as exc:
|
||||
tb = traceback.format_exc()
|
||||
logger.error(
|
||||
@@ -2094,7 +2036,7 @@ class BaseUpstreamProvider:
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=httpx.AsyncHTTPTransport(retries=1),
|
||||
timeout=None,
|
||||
timeout=DEFAULT_PROXY_TIMEOUT,
|
||||
) as client:
|
||||
try:
|
||||
response = await client.send(
|
||||
@@ -2188,6 +2130,9 @@ class BaseUpstreamProvider:
|
||||
headers=dict(response.headers),
|
||||
background=background_tasks,
|
||||
)
|
||||
except (httpx.TimeoutException, httpx.ConnectError):
|
||||
# Re-raise to allow fallback handling in proxy layer
|
||||
raise
|
||||
except Exception as exc:
|
||||
tb = traceback.format_exc()
|
||||
logger.error(
|
||||
@@ -2355,7 +2300,7 @@ class BaseUpstreamProvider:
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
transport=httpx.AsyncHTTPTransport(retries=1),
|
||||
timeout=None,
|
||||
timeout=DEFAULT_PROXY_TIMEOUT,
|
||||
) as client:
|
||||
try:
|
||||
response = await client.send(
|
||||
@@ -2449,6 +2394,9 @@ class BaseUpstreamProvider:
|
||||
headers=dict(response.headers),
|
||||
background=background_tasks,
|
||||
)
|
||||
except (httpx.TimeoutException, httpx.ConnectError):
|
||||
# Re-raise to allow fallback handling in proxy layer
|
||||
raise
|
||||
except Exception as exc:
|
||||
tb = traceback.format_exc()
|
||||
logger.error(
|
||||
|
||||
@@ -187,44 +187,11 @@ class GeminiUpstreamProvider(BaseUpstreamProvider):
|
||||
)
|
||||
|
||||
async def stream_with_cost() -> AsyncGenerator[bytes, None]:
|
||||
payment_finalized = False
|
||||
|
||||
async def finalize_payment() -> None:
|
||||
nonlocal payment_finalized
|
||||
if payment_finalized:
|
||||
return
|
||||
from ..auth import adjust_payment_for_tokens
|
||||
from ..core.db import create_session
|
||||
|
||||
async with create_session() as new_session:
|
||||
fresh_key = await new_session.get(
|
||||
key.__class__, key.hashed_key
|
||||
)
|
||||
if fresh_key:
|
||||
try:
|
||||
await adjust_payment_for_tokens(
|
||||
fresh_key,
|
||||
{
|
||||
"model": model_obj.id,
|
||||
"usage": final_usage_data,
|
||||
},
|
||||
new_session,
|
||||
max_cost_for_model,
|
||||
)
|
||||
payment_finalized = True
|
||||
except Exception as cost_error:
|
||||
logger.error(
|
||||
"Error finalizing Gemini streaming payment in fallback",
|
||||
extra={
|
||||
"error": str(cost_error),
|
||||
"key_hash": key.hashed_key[:8] + "...",
|
||||
},
|
||||
)
|
||||
|
||||
try:
|
||||
async for chunk in response_generator:
|
||||
sse_data = f"data: {json.dumps(chunk)}\n\n"
|
||||
yield sse_data.encode()
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
"Error in Gemini streaming response",
|
||||
@@ -235,9 +202,6 @@ class GeminiUpstreamProvider(BaseUpstreamProvider):
|
||||
},
|
||||
)
|
||||
raise
|
||||
finally:
|
||||
if not payment_finalized:
|
||||
await finalize_payment()
|
||||
|
||||
return StreamingResponse(
|
||||
stream_with_cost(),
|
||||
|
||||
@@ -1,101 +0,0 @@
|
||||
import asyncio
|
||||
|
||||
import httpx
|
||||
|
||||
BASE_URL = input("Enter routstr URL: ")
|
||||
API_KEY = input("Enter key or token: ")
|
||||
|
||||
|
||||
async def get_balance(client: httpx.AsyncClient) -> int:
|
||||
response = await client.get("/v1/balance/info")
|
||||
response.raise_for_status()
|
||||
data = response.json()
|
||||
print(f"Current Balance Info: {data}")
|
||||
return data.get("reserved", 0)
|
||||
|
||||
|
||||
async def reproduce() -> None:
|
||||
headers = {"Authorization": f"Bearer {API_KEY}", "Content-Type": "application/json"}
|
||||
|
||||
async with httpx.AsyncClient(
|
||||
base_url=BASE_URL, headers=headers, timeout=30.0
|
||||
) as client:
|
||||
print("Checking initial balance...")
|
||||
try:
|
||||
initial_reserved = await get_balance(client)
|
||||
except Exception as e:
|
||||
print(f"Failed to get balance: {e}")
|
||||
return
|
||||
|
||||
print("\nStarting streaming request...")
|
||||
try:
|
||||
# Create a separate client for the stream so we can close it independently if needed,
|
||||
# but usually just breaking the loop and exiting the context manager is enough.
|
||||
# However, to be sure we simulate a harsh disconnect, we can just cancel the task or close the client.
|
||||
|
||||
async with client.stream(
|
||||
"POST",
|
||||
"/v1/chat/completions",
|
||||
json={
|
||||
"model": "gpt-5-nano",
|
||||
"messages": [
|
||||
{
|
||||
"role": "user",
|
||||
"content": "Write a long poem about the ocean.",
|
||||
}
|
||||
],
|
||||
"stream": True,
|
||||
},
|
||||
) as response:
|
||||
print(f"Stream status: {response.status_code}")
|
||||
if response.status_code != 200:
|
||||
err_bytes = await response.aread()
|
||||
try:
|
||||
err_str = err_bytes.decode()
|
||||
except Exception:
|
||||
err_str = repr(err_bytes)
|
||||
print(f"Error: {err_str}")
|
||||
return
|
||||
|
||||
print("Stream started. Reading a few chunks...")
|
||||
count = 0
|
||||
async for chunk in response.aiter_bytes():
|
||||
print(f"Received chunk: {len(chunk)} bytes")
|
||||
count += 1
|
||||
if count >= 3:
|
||||
print("Simulating client disconnect (breaking stream)...")
|
||||
break
|
||||
|
||||
except Exception as e:
|
||||
print(f"Stream interrupted (expected): {e}")
|
||||
|
||||
# Wait a bit for the server to realize we disconnected (though with asyncio it might be immediate or depend on keepalive)
|
||||
print("\nWaiting for server to process disconnect...")
|
||||
await asyncio.sleep(21)
|
||||
|
||||
print("\nChecking final balance...")
|
||||
try:
|
||||
final_reserved = await get_balance(client)
|
||||
except Exception:
|
||||
# Retry once if connection was closed
|
||||
async with httpx.AsyncClient(
|
||||
base_url=BASE_URL, headers=headers, timeout=30.0
|
||||
) as new_client:
|
||||
final_reserved = await get_balance(new_client)
|
||||
|
||||
if final_reserved > initial_reserved:
|
||||
print(
|
||||
f"\n[FAIL] Bug reproduced! Reserved balance increased: {initial_reserved} -> {final_reserved}"
|
||||
)
|
||||
print(f"Accumulated reserved balance: {final_reserved - initial_reserved}")
|
||||
else:
|
||||
print(
|
||||
f"\n[PASS] Reserved balance released correctly: {initial_reserved} -> {final_reserved}"
|
||||
)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
asyncio.run(reproduce())
|
||||
except KeyboardInterrupt:
|
||||
pass
|
||||
+2
-450
@@ -21,17 +21,7 @@ import {
|
||||
PopoverTrigger,
|
||||
} from '@/components/ui/popover';
|
||||
import { Calendar } from '@/components/ui/calendar';
|
||||
import { Badge } from '@/components/ui/badge';
|
||||
import {
|
||||
Command,
|
||||
CommandEmpty,
|
||||
CommandGroup,
|
||||
CommandInput,
|
||||
CommandItem,
|
||||
CommandList,
|
||||
} from '@/components/ui/command';
|
||||
import { Checkbox } from '@/components/ui/checkbox';
|
||||
import { CalendarIcon, Filter, X, Plus } from 'lucide-react';
|
||||
import { CalendarIcon, Filter, X } from 'lucide-react';
|
||||
import { useState, useEffect } from 'react';
|
||||
import { format } from 'date-fns';
|
||||
import { cn } from '@/lib/utils';
|
||||
@@ -41,17 +31,11 @@ interface LogFiltersProps {
|
||||
selectedLevel: string;
|
||||
requestId: string;
|
||||
searchText: string;
|
||||
selectedStatusCodes: string[];
|
||||
selectedMethods: string[];
|
||||
selectedEndpoints: string[];
|
||||
limit: number;
|
||||
onDateChange: (date: string) => void;
|
||||
onLevelChange: (level: string) => void;
|
||||
onRequestIdChange: (requestId: string) => void;
|
||||
onSearchTextChange: (searchText: string) => void;
|
||||
onStatusCodesChange: (statusCodes: string[]) => void;
|
||||
onMethodsChange: (methods: string[]) => void;
|
||||
onEndpointsChange: (endpoints: string[]) => void;
|
||||
onLimitChange: (limit: number) => void;
|
||||
onClearFilters: () => void;
|
||||
}
|
||||
@@ -59,87 +43,16 @@ interface LogFiltersProps {
|
||||
const LOG_LEVELS = ['TRACE', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'];
|
||||
const PRESET_LIMITS = ['25', '50', '100', '200', '500', '1000'];
|
||||
|
||||
const STATUS_CODE_OPTIONS = [
|
||||
'200',
|
||||
'201',
|
||||
'204',
|
||||
'400',
|
||||
'401',
|
||||
'402',
|
||||
'403',
|
||||
'404',
|
||||
'422',
|
||||
'429',
|
||||
'500',
|
||||
'502',
|
||||
'503',
|
||||
'504',
|
||||
];
|
||||
|
||||
const METHOD_OPTIONS = [
|
||||
'GET',
|
||||
'POST',
|
||||
'PUT',
|
||||
'DELETE',
|
||||
'PATCH',
|
||||
'OPTIONS',
|
||||
'HEAD',
|
||||
];
|
||||
|
||||
const ENDPOINT_OPTIONS = [
|
||||
'/chat/completions',
|
||||
'/v1/chat/completions',
|
||||
'/models',
|
||||
'/v1/models',
|
||||
'/responses',
|
||||
'/v1/responses',
|
||||
'v1/embeddings/models',
|
||||
'/embeddings/models',
|
||||
];
|
||||
|
||||
interface FilterBadgeProps {
|
||||
value: string;
|
||||
onRemove: (value: string) => void;
|
||||
}
|
||||
|
||||
function FilterBadge({ value, onRemove }: FilterBadgeProps) {
|
||||
return (
|
||||
<Badge
|
||||
variant='secondary'
|
||||
className='flex items-center gap-1 px-1 font-normal'
|
||||
>
|
||||
{value}
|
||||
<button
|
||||
type='button'
|
||||
onClick={(e) => {
|
||||
e.preventDefault();
|
||||
e.stopPropagation();
|
||||
onRemove(value);
|
||||
}}
|
||||
className='hover:bg-muted-foreground/20 rounded-full'
|
||||
>
|
||||
<X className='h-3 w-3' />
|
||||
</button>
|
||||
</Badge>
|
||||
);
|
||||
}
|
||||
|
||||
export function LogFilters({
|
||||
selectedDate,
|
||||
selectedLevel,
|
||||
requestId,
|
||||
searchText,
|
||||
selectedStatusCodes,
|
||||
selectedMethods,
|
||||
selectedEndpoints,
|
||||
limit,
|
||||
onDateChange,
|
||||
onLevelChange,
|
||||
onRequestIdChange,
|
||||
onSearchTextChange,
|
||||
onStatusCodesChange,
|
||||
onMethodsChange,
|
||||
onEndpointsChange,
|
||||
onLimitChange,
|
||||
onClearFilters,
|
||||
}: LogFiltersProps) {
|
||||
@@ -155,10 +68,6 @@ export function LogFilters({
|
||||
: undefined
|
||||
);
|
||||
|
||||
const [statusSearch, setStatusSearch] = useState('');
|
||||
const [methodSearch, setMethodSearch] = useState('');
|
||||
const [endpointSearch, setEndpointSearch] = useState('');
|
||||
|
||||
useEffect(() => {
|
||||
const currentIsPreset = PRESET_LIMITS.includes(limit.toString());
|
||||
setIsCustom(!currentIsPreset);
|
||||
@@ -220,31 +129,6 @@ export function LogFilters({
|
||||
}
|
||||
};
|
||||
|
||||
const toggleSelection = (
|
||||
current: string[],
|
||||
value: string,
|
||||
onChange: (val: string[]) => void
|
||||
) => {
|
||||
if (current.includes(value)) {
|
||||
onChange(current.filter((v) => v !== value));
|
||||
} else {
|
||||
onChange([...current, value]);
|
||||
}
|
||||
};
|
||||
|
||||
const handleQuickStatusCode = (range: '4xx' | '5xx') => {
|
||||
const codes = STATUS_CODE_OPTIONS.filter((c) => c.startsWith(range[0]));
|
||||
const newSelection = new Set([...selectedStatusCodes]);
|
||||
const allIncluded = codes.every((c) => selectedStatusCodes.includes(c));
|
||||
|
||||
if (allIncluded) {
|
||||
codes.forEach((c) => newSelection.delete(c));
|
||||
} else {
|
||||
codes.forEach((c) => newSelection.add(c));
|
||||
}
|
||||
onStatusCodesChange(Array.from(newSelection));
|
||||
};
|
||||
|
||||
return (
|
||||
<Card className='mb-6'>
|
||||
<CardHeader>
|
||||
@@ -253,8 +137,7 @@ export function LogFilters({
|
||||
Filters
|
||||
</CardTitle>
|
||||
<CardDescription>
|
||||
Filter logs by date, level, request ID, text search, status code,
|
||||
method, endpoint and limit
|
||||
Filter logs by date, level, request ID, text search, and limit
|
||||
</CardDescription>
|
||||
</CardHeader>
|
||||
<CardContent>
|
||||
@@ -314,337 +197,6 @@ export function LogFilters({
|
||||
</Select>
|
||||
</div>
|
||||
|
||||
<div className='space-y-2'>
|
||||
<Label>Status Codes</Label>
|
||||
<Popover>
|
||||
<PopoverTrigger asChild>
|
||||
<Button
|
||||
variant='outline'
|
||||
className='w-full justify-start text-left font-normal'
|
||||
>
|
||||
<div className='flex flex-wrap gap-1'>
|
||||
{selectedStatusCodes.length > 0 ? (
|
||||
selectedStatusCodes.map((code) => (
|
||||
<FilterBadge
|
||||
key={code}
|
||||
value={code}
|
||||
onRemove={(val) =>
|
||||
toggleSelection(
|
||||
selectedStatusCodes,
|
||||
val,
|
||||
onStatusCodesChange
|
||||
)
|
||||
}
|
||||
/>
|
||||
))
|
||||
) : (
|
||||
<span className='text-muted-foreground'>All codes</span>
|
||||
)}
|
||||
</div>
|
||||
</Button>
|
||||
</PopoverTrigger>
|
||||
<PopoverContent className='w-64 p-0' align='start'>
|
||||
<Command>
|
||||
<CommandInput
|
||||
placeholder='Search or add status code...'
|
||||
value={statusSearch}
|
||||
onValueChange={setStatusSearch}
|
||||
/>
|
||||
<CommandList>
|
||||
{selectedStatusCodes.length > 0 && (
|
||||
<CommandGroup heading='Selected'>
|
||||
{selectedStatusCodes.map((code) => (
|
||||
<CommandItem
|
||||
key={`selected-${code}`}
|
||||
onSelect={() =>
|
||||
toggleSelection(
|
||||
selectedStatusCodes,
|
||||
code,
|
||||
onStatusCodesChange
|
||||
)
|
||||
}
|
||||
>
|
||||
<Checkbox checked={true} className='mr-2' />
|
||||
{code}
|
||||
</CommandItem>
|
||||
))}
|
||||
</CommandGroup>
|
||||
)}
|
||||
{statusSearch &&
|
||||
!STATUS_CODE_OPTIONS.includes(statusSearch) &&
|
||||
!selectedStatusCodes.includes(statusSearch) && (
|
||||
<CommandGroup heading='Custom'>
|
||||
<CommandItem
|
||||
onSelect={() => {
|
||||
if (/^\d+$/.test(statusSearch)) {
|
||||
toggleSelection(
|
||||
selectedStatusCodes,
|
||||
statusSearch,
|
||||
onStatusCodesChange
|
||||
);
|
||||
setStatusSearch('');
|
||||
}
|
||||
}}
|
||||
>
|
||||
<Plus className='mr-2 h-4 w-4' />
|
||||
Add "{statusSearch}"
|
||||
</CommandItem>
|
||||
</CommandGroup>
|
||||
)}
|
||||
<CommandEmpty>No results found.</CommandEmpty>
|
||||
<CommandGroup heading='Quick Filters'>
|
||||
<CommandItem
|
||||
onSelect={() => handleQuickStatusCode('4xx')}
|
||||
>
|
||||
<Checkbox
|
||||
checked={STATUS_CODE_OPTIONS.filter((c) =>
|
||||
c.startsWith('4')
|
||||
).every((c) => selectedStatusCodes.includes(c))}
|
||||
className='mr-2'
|
||||
/>
|
||||
4xx Errors
|
||||
</CommandItem>
|
||||
<CommandItem
|
||||
onSelect={() => handleQuickStatusCode('5xx')}
|
||||
>
|
||||
<Checkbox
|
||||
checked={STATUS_CODE_OPTIONS.filter((c) =>
|
||||
c.startsWith('5')
|
||||
).every((c) => selectedStatusCodes.includes(c))}
|
||||
className='mr-2'
|
||||
/>
|
||||
5xx Errors
|
||||
</CommandItem>
|
||||
</CommandGroup>
|
||||
<CommandGroup heading='Common Codes'>
|
||||
{STATUS_CODE_OPTIONS.filter(
|
||||
(code) => !selectedStatusCodes.includes(code)
|
||||
).map((code) => (
|
||||
<CommandItem
|
||||
key={code}
|
||||
onSelect={() =>
|
||||
toggleSelection(
|
||||
selectedStatusCodes,
|
||||
code,
|
||||
onStatusCodesChange
|
||||
)
|
||||
}
|
||||
>
|
||||
<Checkbox checked={false} className='mr-2' />
|
||||
{code}
|
||||
</CommandItem>
|
||||
))}
|
||||
</CommandGroup>
|
||||
</CommandList>
|
||||
</Command>
|
||||
</PopoverContent>
|
||||
</Popover>
|
||||
</div>
|
||||
|
||||
<div className='space-y-2'>
|
||||
<Label>HTTP Methods</Label>
|
||||
<Popover>
|
||||
<PopoverTrigger asChild>
|
||||
<Button
|
||||
variant='outline'
|
||||
className='w-full justify-start text-left font-normal'
|
||||
>
|
||||
<div className='flex flex-wrap gap-1'>
|
||||
{selectedMethods.length > 0 ? (
|
||||
selectedMethods.map((method) => (
|
||||
<FilterBadge
|
||||
key={method}
|
||||
value={method}
|
||||
onRemove={(val) =>
|
||||
toggleSelection(
|
||||
selectedMethods,
|
||||
val,
|
||||
onMethodsChange
|
||||
)
|
||||
}
|
||||
/>
|
||||
))
|
||||
) : (
|
||||
<span className='text-muted-foreground'>All methods</span>
|
||||
)}
|
||||
</div>
|
||||
</Button>
|
||||
</PopoverTrigger>
|
||||
<PopoverContent className='w-64 p-0' align='start'>
|
||||
<Command>
|
||||
<CommandInput
|
||||
placeholder='Search or add method...'
|
||||
value={methodSearch}
|
||||
onValueChange={setMethodSearch}
|
||||
/>
|
||||
<CommandList>
|
||||
{selectedMethods.length > 0 && (
|
||||
<CommandGroup heading='Selected'>
|
||||
{selectedMethods.map((method) => (
|
||||
<CommandItem
|
||||
key={`selected-${method}`}
|
||||
onSelect={() =>
|
||||
toggleSelection(
|
||||
selectedMethods,
|
||||
method,
|
||||
onMethodsChange
|
||||
)
|
||||
}
|
||||
>
|
||||
<Checkbox checked={true} className='mr-2' />
|
||||
{method}
|
||||
</CommandItem>
|
||||
))}
|
||||
</CommandGroup>
|
||||
)}
|
||||
{methodSearch &&
|
||||
!METHOD_OPTIONS.includes(methodSearch.toUpperCase()) &&
|
||||
!selectedMethods.includes(methodSearch.toUpperCase()) && (
|
||||
<CommandGroup heading='Custom'>
|
||||
<CommandItem
|
||||
onSelect={() => {
|
||||
toggleSelection(
|
||||
selectedMethods,
|
||||
methodSearch.toUpperCase(),
|
||||
onMethodsChange
|
||||
);
|
||||
setMethodSearch('');
|
||||
}}
|
||||
>
|
||||
<Plus className='mr-2 h-4 w-4' />
|
||||
Add "{methodSearch.toUpperCase()}"
|
||||
</CommandItem>
|
||||
</CommandGroup>
|
||||
)}
|
||||
<CommandEmpty>No results found.</CommandEmpty>
|
||||
<CommandGroup>
|
||||
{METHOD_OPTIONS.filter(
|
||||
(method) => !selectedMethods.includes(method)
|
||||
).map((method) => (
|
||||
<CommandItem
|
||||
key={method}
|
||||
onSelect={() =>
|
||||
toggleSelection(
|
||||
selectedMethods,
|
||||
method,
|
||||
onMethodsChange
|
||||
)
|
||||
}
|
||||
>
|
||||
<Checkbox checked={false} className='mr-2' />
|
||||
{method}
|
||||
</CommandItem>
|
||||
))}
|
||||
</CommandGroup>
|
||||
</CommandList>
|
||||
</Command>
|
||||
</PopoverContent>
|
||||
</Popover>
|
||||
</div>
|
||||
|
||||
<div className='space-y-2'>
|
||||
<Label>Endpoints</Label>
|
||||
<Popover>
|
||||
<PopoverTrigger asChild>
|
||||
<Button
|
||||
variant='outline'
|
||||
className='w-full justify-start text-left font-normal'
|
||||
>
|
||||
<div className='flex flex-wrap gap-1 overflow-hidden'>
|
||||
{selectedEndpoints.length > 0 ? (
|
||||
selectedEndpoints.map((endpoint) => (
|
||||
<FilterBadge
|
||||
key={endpoint}
|
||||
value={endpoint}
|
||||
onRemove={(val) =>
|
||||
toggleSelection(
|
||||
selectedEndpoints,
|
||||
val,
|
||||
onEndpointsChange
|
||||
)
|
||||
}
|
||||
/>
|
||||
))
|
||||
) : (
|
||||
<span className='text-muted-foreground'>
|
||||
All endpoints
|
||||
</span>
|
||||
)}
|
||||
</div>
|
||||
</Button>
|
||||
</PopoverTrigger>
|
||||
<PopoverContent className='w-80 p-0' align='start'>
|
||||
<Command>
|
||||
<CommandInput
|
||||
placeholder='Search or add endpoint pattern...'
|
||||
value={endpointSearch}
|
||||
onValueChange={setEndpointSearch}
|
||||
/>
|
||||
<CommandList>
|
||||
{selectedEndpoints.length > 0 && (
|
||||
<CommandGroup heading='Selected'>
|
||||
{selectedEndpoints.map((endpoint) => (
|
||||
<CommandItem
|
||||
key={`selected-${endpoint}`}
|
||||
onSelect={() =>
|
||||
toggleSelection(
|
||||
selectedEndpoints,
|
||||
endpoint,
|
||||
onEndpointsChange
|
||||
)
|
||||
}
|
||||
>
|
||||
<Checkbox checked={true} className='mr-2' />
|
||||
{endpoint}
|
||||
</CommandItem>
|
||||
))}
|
||||
</CommandGroup>
|
||||
)}
|
||||
{endpointSearch &&
|
||||
!ENDPOINT_OPTIONS.includes(endpointSearch) &&
|
||||
!selectedEndpoints.includes(endpointSearch) && (
|
||||
<CommandGroup heading='Custom'>
|
||||
<CommandItem
|
||||
onSelect={() => {
|
||||
toggleSelection(
|
||||
selectedEndpoints,
|
||||
endpointSearch,
|
||||
onEndpointsChange
|
||||
);
|
||||
setEndpointSearch('');
|
||||
}}
|
||||
>
|
||||
<Plus className='mr-2 h-4 w-4' />
|
||||
Add "{endpointSearch}"
|
||||
</CommandItem>
|
||||
</CommandGroup>
|
||||
)}
|
||||
<CommandEmpty>No results found.</CommandEmpty>
|
||||
<CommandGroup heading='Common Endpoints'>
|
||||
{ENDPOINT_OPTIONS.filter(
|
||||
(endpoint) => !selectedEndpoints.includes(endpoint)
|
||||
).map((endpoint) => (
|
||||
<CommandItem
|
||||
key={endpoint}
|
||||
onSelect={() =>
|
||||
toggleSelection(
|
||||
selectedEndpoints,
|
||||
endpoint,
|
||||
onEndpointsChange
|
||||
)
|
||||
}
|
||||
>
|
||||
<Checkbox checked={false} className='mr-2' />
|
||||
{endpoint}
|
||||
</CommandItem>
|
||||
))}
|
||||
</CommandGroup>
|
||||
</CommandList>
|
||||
</Command>
|
||||
</PopoverContent>
|
||||
</Popover>
|
||||
</div>
|
||||
|
||||
<div className='space-y-2'>
|
||||
<Label htmlFor='request-id'>Request ID</Label>
|
||||
<Input
|
||||
|
||||
+2
-84
@@ -1,6 +1,6 @@
|
||||
'use client';
|
||||
|
||||
import { useState, useEffect } from 'react';
|
||||
import { useState } from 'react';
|
||||
import { useQuery } from '@tanstack/react-query';
|
||||
import { AppSidebar } from '@/components/app-sidebar';
|
||||
import { SiteHeader } from '@/components/site-header';
|
||||
@@ -22,66 +22,15 @@ import { LogFilters } from './log-filters';
|
||||
import { LogEntryCard } from './log-entry-card';
|
||||
import { LogDetailsDialog } from './log-details-dialog';
|
||||
|
||||
const STORAGE_KEY = 'routstr-log-filters';
|
||||
|
||||
export default function LogsPage() {
|
||||
const [selectedDate, setSelectedDate] = useState<string>('all');
|
||||
const [selectedLevel, setSelectedLevel] = useState<string>('all');
|
||||
const [requestId, setRequestId] = useState<string>('');
|
||||
const [searchText, setSearchText] = useState<string>('');
|
||||
const [selectedStatusCodes, setSelectedStatusCodes] = useState<string[]>([]);
|
||||
const [selectedMethods, setSelectedMethods] = useState<string[]>([]);
|
||||
const [selectedEndpoints, setSelectedEndpoints] = useState<string[]>([]);
|
||||
const [limit, setLimit] = useState<number>(100);
|
||||
const [selectedLog, setSelectedLog] = useState<LogEntry | null>(null);
|
||||
const [isDialogOpen, setIsDialogOpen] = useState<boolean>(false);
|
||||
|
||||
// Load filters from localStorage on mount
|
||||
useEffect(() => {
|
||||
const saved = localStorage.getItem(STORAGE_KEY);
|
||||
if (saved) {
|
||||
try {
|
||||
const parsed = JSON.parse(saved);
|
||||
if (parsed.selectedDate) setSelectedDate(parsed.selectedDate);
|
||||
if (parsed.selectedLevel) setSelectedLevel(parsed.selectedLevel);
|
||||
if (parsed.requestId) setRequestId(parsed.requestId);
|
||||
if (parsed.searchText) setSearchText(parsed.searchText);
|
||||
if (parsed.selectedStatusCodes)
|
||||
setSelectedStatusCodes(parsed.selectedStatusCodes);
|
||||
if (parsed.selectedMethods) setSelectedMethods(parsed.selectedMethods);
|
||||
if (parsed.selectedEndpoints)
|
||||
setSelectedEndpoints(parsed.selectedEndpoints);
|
||||
if (parsed.limit) setLimit(parsed.limit);
|
||||
} catch (e) {
|
||||
console.error('Failed to load filters from localStorage', e);
|
||||
}
|
||||
}
|
||||
}, []);
|
||||
|
||||
// Save filters to localStorage whenever they change
|
||||
useEffect(() => {
|
||||
const filters = {
|
||||
selectedDate,
|
||||
selectedLevel,
|
||||
requestId,
|
||||
searchText,
|
||||
selectedStatusCodes,
|
||||
selectedMethods,
|
||||
selectedEndpoints,
|
||||
limit,
|
||||
};
|
||||
localStorage.setItem(STORAGE_KEY, JSON.stringify(filters));
|
||||
}, [
|
||||
selectedDate,
|
||||
selectedLevel,
|
||||
requestId,
|
||||
searchText,
|
||||
selectedStatusCodes,
|
||||
selectedMethods,
|
||||
selectedEndpoints,
|
||||
limit,
|
||||
]);
|
||||
|
||||
const {
|
||||
data: logsData,
|
||||
refetch: refetchLogs,
|
||||
@@ -93,9 +42,6 @@ export default function LogsPage() {
|
||||
selectedLevel,
|
||||
requestId,
|
||||
searchText,
|
||||
selectedStatusCodes,
|
||||
selectedMethods,
|
||||
selectedEndpoints,
|
||||
limit,
|
||||
],
|
||||
queryFn: () =>
|
||||
@@ -104,16 +50,6 @@ export default function LogsPage() {
|
||||
level: selectedLevel === 'all' ? undefined : selectedLevel,
|
||||
request_id: requestId || undefined,
|
||||
search: searchText || undefined,
|
||||
status_codes:
|
||||
selectedStatusCodes.length > 0
|
||||
? selectedStatusCodes.join(',')
|
||||
: undefined,
|
||||
methods:
|
||||
selectedMethods.length > 0 ? selectedMethods.join(',') : undefined,
|
||||
endpoints:
|
||||
selectedEndpoints.length > 0
|
||||
? selectedEndpoints.join(',')
|
||||
: undefined,
|
||||
limit: limit,
|
||||
}),
|
||||
refetchInterval: 30000,
|
||||
@@ -124,9 +60,6 @@ export default function LogsPage() {
|
||||
setSelectedLevel('all');
|
||||
setRequestId('');
|
||||
setSearchText('');
|
||||
setSelectedStatusCodes([]);
|
||||
setSelectedMethods([]);
|
||||
setSelectedEndpoints([]);
|
||||
setLimit(100);
|
||||
};
|
||||
|
||||
@@ -167,17 +100,11 @@ export default function LogsPage() {
|
||||
selectedLevel={selectedLevel}
|
||||
requestId={requestId}
|
||||
searchText={searchText}
|
||||
selectedStatusCodes={selectedStatusCodes}
|
||||
selectedMethods={selectedMethods}
|
||||
selectedEndpoints={selectedEndpoints}
|
||||
limit={limit}
|
||||
onDateChange={setSelectedDate}
|
||||
onLevelChange={setSelectedLevel}
|
||||
onRequestIdChange={setRequestId}
|
||||
onSearchTextChange={setSearchText}
|
||||
onStatusCodesChange={setSelectedStatusCodes}
|
||||
onMethodsChange={setSelectedMethods}
|
||||
onEndpointsChange={setSelectedEndpoints}
|
||||
onLimitChange={setLimit}
|
||||
onClearFilters={handleClearFilters}
|
||||
/>
|
||||
@@ -195,22 +122,13 @@ export default function LogsPage() {
|
||||
{(selectedDate !== 'all' ||
|
||||
selectedLevel !== 'all' ||
|
||||
requestId ||
|
||||
searchText ||
|
||||
selectedStatusCodes.length > 0 ||
|
||||
selectedMethods.length > 0 ||
|
||||
selectedEndpoints.length > 0) && (
|
||||
searchText) && (
|
||||
<CardDescription className='text-xs sm:text-sm'>
|
||||
Showing logs
|
||||
{selectedDate !== 'all' && ` for ${selectedDate}`}
|
||||
{selectedLevel !== 'all' && ` with level ${selectedLevel}`}
|
||||
{requestId && ` with request ID ${requestId}`}
|
||||
{searchText && ` matching "${searchText}"`}
|
||||
{selectedStatusCodes.length > 0 &&
|
||||
` with status ${selectedStatusCodes.join(', ')}`}
|
||||
{selectedMethods.length > 0 &&
|
||||
` with method ${selectedMethods.join(', ')}`}
|
||||
{selectedEndpoints.length > 0 &&
|
||||
` with endpoint ${selectedEndpoints.join(', ')}`}
|
||||
</CardDescription>
|
||||
)}
|
||||
</CardHeader>
|
||||
|
||||
@@ -17,9 +17,6 @@ export interface LogsResponse {
|
||||
level: string | null;
|
||||
request_id: string | null;
|
||||
search: string | null;
|
||||
status_codes: string | null;
|
||||
methods: string | null;
|
||||
endpoints: string | null;
|
||||
limit: number;
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user