Compare commits

..
Author SHA1 Message Date
9qeklajc 1df66f48b8 better timeout handling 2026-01-04 14:03:28 +01:00
9qeklajc a9c5458660 set timeout 2026-01-03 23:17:51 +01:00
30 changed files with 10431 additions and 1599 deletions
+6 -11
View File
@@ -59,30 +59,25 @@ jobs:
- name: Checkout code
uses: actions/checkout@v4
- name: Setup pnpm
uses: pnpm/action-setup@v4
with:
version: 10
- name: Setup Node.js
uses: actions/setup-node@v4
with:
node-version: "18"
cache: "pnpm"
cache-dependency-path: ui/pnpm-lock.yaml
cache: "npm"
cache-dependency-path: ui/package-lock.json
- name: Install UI dependencies
working-directory: ./ui
run: pnpm install --frozen-lockfile
run: npm ci
- name: Run UI format check
working-directory: ./ui
run: pnpm run format-check
run: npm run format-check
- name: Run UI linting
working-directory: ./ui
run: pnpm run lint
run: npm run lint
- name: Run UI build
working-directory: ./ui
run: pnpm run build
run: npm run build
+1 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "routstr"
version = "0.2.2"
version = "0.2.1"
description = "Payment proxy for your LLM endpoint using cashu and nostr."
readme = "README.md"
requires-python = ">=3.11"
+41 -5
View File
@@ -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
View File
@@ -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,
+3 -26
View File
@@ -142,7 +142,6 @@ async def refund_wallet_endpoint(
authorization: Annotated[str, Header(...)],
session: AsyncSession = Depends(get_session),
) -> dict[str, str]:
logger.info("Refund request received", extra={"authorization": authorization})
if not authorization.startswith("Bearer "):
raise HTTPException(
status_code=401,
@@ -155,10 +154,8 @@ async def refund_wallet_endpoint(
return cached
key: ApiKey = await validate_bearer_key(bearer_value, session)
remaining_balance_msats: int = key.balance
remaining_balance_msats: int = key.total_balance
logger.info("Refunding key: %s", key.dict())
if key.refund_currency == "sat":
remaining_balance = remaining_balance_msats // 1000
else:
@@ -174,15 +171,6 @@ async def refund_wallet_endpoint(
if key.refund_address:
from .core.settings import settings as global_settings
logger.info(
"Sending refund to lnurl",
extra={
"remaining_balance": remaining_balance,
"refund_currency": key.refund_currency,
"refund_mint_url": key.refund_mint_url,
"refund_address": key.refund_address,
},
)
await send_to_lnurl(
remaining_balance,
key.refund_currency or "sat",
@@ -191,16 +179,7 @@ async def refund_wallet_endpoint(
)
result = {"recipient": key.refund_address}
else:
logger.info(
"Sending refund as token",
extra={
"remaining_balance": remaining_balance,
"refund_currency": key.refund_currency,
"refund_mint_url": key.refund_mint_url,
},
)
refund_currency = key.refund_currency or "sat"
token = await send_token(
remaining_balance, refund_currency, key.refund_mint_url
)
@@ -211,12 +190,10 @@ async def refund_wallet_endpoint(
else:
result["msats"] = str(remaining_balance_msats)
except HTTPException as e:
logger.error("Refund failed", extra={"exception": e})
except HTTPException:
# Re-raise HTTP exceptions (like 400 for balance too small)
raise
except Exception as e:
logger.error("Refund failed", extra={"exception": e})
# If refund fails, don't modify the database
error_msg = str(e)
if (
@@ -227,7 +204,7 @@ async def refund_wallet_endpoint(
):
raise HTTPException(status_code=503, detail="Mint service unavailable")
else:
raise HTTPException(status_code=500, detail="Refund failed: " + error_msg)
raise HTTPException(status_code=500, detail="Refund failed")
await _refund_cache_set(bearer_value, result)
-22
View File
@@ -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,
}
+1 -9
View File
@@ -6,7 +6,7 @@ from typing import AsyncGenerator
from alembic import command
from alembic.config import Config
from sqlalchemy.ext.asyncio.engine import create_async_engine
from sqlmodel import Field, Relationship, SQLModel, func, select, update
from sqlmodel import Field, Relationship, SQLModel, func, select
from sqlmodel.ext.asyncio.session import AsyncSession
from .logging import get_logger
@@ -53,14 +53,6 @@ class ApiKey(SQLModel, table=True): # type: ignore
return self.balance - self.reserved_balance
async def reset_all_reserved_balances(session: AsyncSession) -> None:
logger.info("Resetting all reserved balances to 0")
stmt = update(ApiKey).values(reserved_balance=0)
await session.exec(stmt) # type: ignore[call-overload]
await session.commit()
logger.info("Reserved balances reset successfully")
class ModelRow(SQLModel, table=True): # type: ignore
__tablename__ = "models"
id: str = Field(primary_key=True)
+1 -43
View File
@@ -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()
+2 -6
View File
@@ -33,9 +33,9 @@ setup_logging()
logger = get_logger(__name__)
if os.getenv("VERSION_SUFFIX") is not None:
__version__ = f"0.2.2-{os.getenv('VERSION_SUFFIX')}"
__version__ = f"0.2.1-{os.getenv('VERSION_SUFFIX')}"
else:
__version__ = "0.2.2"
__version__ = "0.2.1"
@asynccontextmanager
@@ -61,10 +61,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
# Initialize application settings (env -> computed -> DB precedence)
async with create_session() as session:
s = await SettingsService.initialize(session)
if s.reset_reserved_balance_on_startup:
from .db import reset_all_reserved_balances
await reset_all_reserved_balances(session)
# Apply app metadata from settings
try:
+8 -4
View File
@@ -6,7 +6,7 @@ import os
from datetime import datetime, timezone
from typing import Any
from pydantic.v1 import BaseModel, BaseSettings, Field
from pydantic.v1 import BaseModel, BaseSettings, Field, validator
from sqlmodel.ext.asyncio.session import AsyncSession
@@ -37,6 +37,13 @@ class Settings(BaseSettings):
# Cashu
cashu_mints: list[str] = Field(default_factory=list, env="CASHU_MINTS")
@validator("cashu_mints", pre=True, each_item=True)
def normalize_mint_url(cls, v: str) -> str:
if isinstance(v, str):
return v.rstrip("/")
return v
receive_ln_address: str = Field(default="", env="RECEIVE_LN_ADDRESS")
primary_mint: str = Field(default="", env="PRIMARY_MINT_URL")
primary_mint_unit: str = Field(default="sat", env="PRIMARY_MINT_UNIT")
@@ -54,9 +61,6 @@ class Settings(BaseSettings):
tolerance_percentage: float = Field(default=1.0, env="TOLERANCE_PERCENTAGE")
# Minimum per-request charge in millisatoshis when model pricing is free/zero
min_request_msat: int = Field(default=1, env="MIN_REQUEST_MSAT")
reset_reserved_balance_on_startup: bool = Field(
default=True, env="RESET_RESERVED_BALANCE_ON_STARTUP"
) # deactivate in horizontal scaling setups
# Network
cors_origins: list[str] = Field(default_factory=lambda: ["*"], env="CORS_ORIGINS")
+11 -22
View File
@@ -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)
+1 -1
View File
@@ -283,7 +283,7 @@ async def raw_send_to_lnurl(
f"({min_sendable_sat} - {max_sendable_sat} {unit})"
)
estimated_fees_sat = int(max(math.ceil((amount_msat / 1000) * 0.01), 2)) + 1
estimated_fees_sat = int(max(math.ceil((amount_msat / 1000) * 0.01), 2))
estimated_fees_msat = estimated_fees_sat * 1000
final_amount = amount_msat - estimated_fees_msat
+211 -58
View File
@@ -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
@@ -16,6 +17,7 @@ from .core.db import (
create_session,
get_session,
)
from .core.settings import settings
from .payment.helpers import (
calculate_discounted_max_cost,
check_token_balance,
@@ -25,6 +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, recieve_token
logger = get_logger(__name__)
proxy_router = APIRouter()
@@ -32,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)
@@ -73,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())
@@ -82,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
@@ -102,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,
)
)
@@ -146,14 +169,32 @@ async def proxy(
else:
model_id = request_body_dict.get("model", "unknown")
if "https://testnut.cashu.space" in settings.cashu_mints:
try:
token_str = None
if x_cashu_header := headers.get("x-cashu"):
token_str = x_cashu_header
elif auth_header := headers.get("authorization"):
parts = auth_header.split(" ")
if len(parts) > 1 and not parts[1].startswith("sk-"):
token_str = parts[1]
if token_str:
token_obj = deserialize_token_from_string(token_str)
if token_obj.mint == "https://testnut.cashu.space":
model_id = "mock/gpt-420-mock"
request_body_dict["model"] = model_id
except Exception:
pass
model_obj = get_model_instance(model_id)
if not model_obj:
return create_error_response(
"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}'",
@@ -170,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)
@@ -192,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
View File
@@ -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(
+265
View File
@@ -0,0 +1,265 @@
import asyncio
import json
import random
from typing import AsyncIterator
from fastapi import Request
from fastapi.responses import Response, StreamingResponse
from ..core.db import ApiKey, AsyncSession
from ..payment.models import Architecture, Model, Pricing
from .base import BaseUpstreamProvider
class MockUpstreamProvider(BaseUpstreamProvider):
"""Fack Mock Upstream provider specifically for Testing."""
provider_type = "mock"
async def forward_request(
self,
request: Request,
path: str,
headers: dict,
request_body: bytes | None,
key: ApiKey,
max_cost_for_model: int,
session: AsyncSession,
model_obj: Model,
) -> Response | StreamingResponse:
if path.endswith("chat/completions"):
is_streaming = False
if request_body:
request_data = json.loads(request_body)
is_streaming = request_data.get("stream", False)
if is_streaming:
async def fake_streaming_response(
chunk_size: int | None = None,
) -> AsyncIterator[bytes]:
suffix = random.randint(1000, 9999)
req_id = f"gen-mock-stream-{suffix}"
created = 1766138895
model = "mock/gpt-420-mock"
def make_chunk(
delta: dict,
finish_reason: str | None = None,
usage: dict | None = None,
) -> bytes:
chunk = {
"id": req_id,
"provider": "MockProvider",
"model": model,
"object": "chat.completion.chunk",
"created": created,
"choices": [
{
"index": 0,
"delta": delta,
"finish_reason": finish_reason,
"native_finish_reason": "completed"
if finish_reason
else None,
"logprobs": None,
}
],
}
if usage:
chunk["usage"] = usage
return f"data: {json.dumps(chunk)}\n\n".encode()
# 1. Initial chunk
yield make_chunk({"role": "assistant", "content": ""})
await asyncio.sleep(0.02)
# 2. Reasoning chunks
reasoning_tokens = ["Mock", " reason", "ing", "..."]
for token in reasoning_tokens:
delta = {
"role": "assistant",
"content": "",
"reasoning": token,
"reasoning_details": [
{
"type": "reasoning.summary",
"summary": token,
"format": "openai-responses-v1",
"index": 0,
}
],
}
yield make_chunk(delta)
await asyncio.sleep(0.03)
# 3. Content chunks
content_tokens = ["This", " is", " a", " mock", " stream", "."]
for token in content_tokens:
yield make_chunk({"role": "assistant", "content": token})
await asyncio.sleep(0.03)
# 4. Finish chunk
yield make_chunk(
{"role": "assistant", "content": ""}, finish_reason="stop"
)
# 5. Usage chunk
usage_data = {
"prompt_tokens": 10,
"completion_tokens": 20,
"total_tokens": 30,
"cost": 0.001,
"is_byok": False,
"prompt_tokens_details": {
"cached_tokens": 0,
"audio_tokens": 0,
"video_tokens": 0,
},
"cost_details": {
"upstream_inference_cost": None,
"upstream_inference_prompt_cost": 0,
"upstream_inference_completions_cost": 0.001,
},
"completion_tokens_details": {
"reasoning_tokens": 10,
"image_tokens": 0,
},
}
usage_chunk = {
"id": req_id,
"provider": "MockProvider",
"model": model,
"object": "chat.completion.chunk",
"created": created,
"choices": [
{
"index": 0,
"delta": {"role": "assistant", "content": ""},
"finish_reason": None,
"native_finish_reason": None,
"logprobs": None,
}
],
"usage": usage_data,
}
yield f"data: {json.dumps(usage_chunk)}\n\n".encode()
# 6. DONE
yield b"data: [DONE]\n\n"
# 7. Cost
cost_chunk = {
"cost": {
"base_msats": 0,
"input_msats": 2,
"output_msats": 10,
"total_msats": 12,
}
}
yield f"data: {json.dumps(cost_chunk)}\n\n".encode()
return StreamingResponse(
fake_streaming_response(),
200,
)
else:
suffix = random.randint(1000, 9999)
content_dict = {
"id": f"gen-mock-{suffix}",
"provider": "MockProvider",
"model": "mock/gpt-5-mini",
"object": "chat.completion",
"created": 1766138655,
"choices": [
{
"logprobs": None,
"finish_reason": "length",
"native_finish_reason": "max_output_tokens",
"index": 0,
"message": {
"role": "assistant",
"content": f"Mock Content {suffix}",
"refusal": None,
"reasoning": f"Mock Reasoning {suffix}",
"reasoning_details": [
{
"format": "openai-responses-v1",
"index": 0,
"type": "reasoning.summary",
"summary": f"Mock Summary {suffix}",
},
{
"id": f"rs_mock_{suffix}",
"format": "openai-responses-v1",
"index": 0,
"type": "reasoning.encrypted",
"data": "mock_encrypted_data",
},
],
},
}
],
"usage": {
"prompt_tokens": 10,
"completion_tokens": 10,
"total_tokens": 20,
"cost": 0,
"is_byok": False,
"prompt_tokens_details": {
"cached_tokens": 0,
"audio_tokens": 0,
"video_tokens": 0,
},
"cost_details": {
"upstream_inference_cost": None,
"upstream_inference_prompt_cost": 0,
"upstream_inference_completions_cost": 0,
},
"completion_tokens_details": {
"reasoning_tokens": 5,
"image_tokens": 0,
},
},
"cost": {
"base_msats": 0,
"input_msats": 0,
"output_msats": 0,
"total_msats": 0,
},
}
return Response(json.dumps(content_dict).encode(), 200)
elif path.endswith("embeddings"):
raise NotImplementedError
elif path.endswith("responses"):
raise NotImplementedError
else:
raise NotImplementedError
async def fetch_models(self) -> list[Model]:
return [
Model(
id="mock/gpt-420-mock",
name="mock/gpt-420-mock",
created=0,
description="mock model for testing",
context_length=8192,
architecture=Architecture(
modality="text",
input_modalities=["text"],
output_modalities=["text"],
tokenizer="",
instruct_type=None,
),
pricing=Pricing(prompt=0.01, completion=0.01),
),
]
def transform_model_name(self, model_id: str) -> str:
return "fake-model"
async def get_balance(self) -> float | None:
return 420.69
+1 -37
View File
@@ -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(),
+8
View File
@@ -218,6 +218,14 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
results = await asyncio.gather(*tasks)
upstreams = [p for p in results if p is not None]
if "https://testnut.cashu.space" in settings.cashu_mints:
from .fake import MockUpstreamProvider
mock_provider = MockUpstreamProvider("mock", "mock")
await mock_provider.refresh_models_cache()
upstreams.append(mock_provider)
logger.info("Initialized MockUpstreamProvider for testnut mint")
return upstreams
+4 -9
View File
@@ -40,23 +40,16 @@ async def recieve_token(
async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int, str]:
"""Internal send function - returns amount and serialized token"""
wallet: Wallet = await get_wallet(mint_url or settings.primary_mint, unit)
logger.info("Sending", extra={"amount": amount, "unit": unit, "mint_url": mint_url})
proofs = get_proofs_per_mint_and_unit(
wallet, mint_url or settings.primary_mint, unit
)
logger.info("Proofs", extra={"proofs": proofs})
logger.info(
"Selecting to send",
extra={"amount": amount, "unit": unit, "mint_url": mint_url},
)
send_proofs, _ = await wallet.select_to_send(
proofs, amount, set_reserved=True, include_fees=False
)
logger.info("Send proofs", extra={"send_proofs": send_proofs})
token = await wallet.serialize_proofs(
send_proofs, include_dleq=False, legacy=False, memo=None
)
logger.info("Token created", extra={"token": token})
return amount, token
@@ -88,7 +81,7 @@ async def swap_to_primary_mint(
amount_msat = token_amount
else:
raise ValueError("Invalid unit")
estimated_fee_sat = math.ceil(max(amount_msat // 1000 * 0.01, 2)) + 1
estimated_fee_sat = math.ceil(max(amount_msat // 1000 * 0.01, 2))
amount_msat_after_fee = amount_msat - estimated_fee_sat * 1000
primary_wallet = await get_wallet(settings.primary_mint, settings.primary_mint_unit)
@@ -320,6 +313,8 @@ async def periodic_payout() -> None:
try:
async with db.create_session() as session:
for mint_url in settings.cashu_mints:
if mint_url == "https://testnut.cashu.space":
continue
for unit in ["sat", "msat"]:
wallet = await get_wallet(mint_url, unit)
proofs = get_proofs_per_mint_and_unit(
-101
View File
@@ -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
+3 -8
View File
@@ -4,22 +4,17 @@ FROM base AS deps
RUN apk add --no-cache libc6-compat
WORKDIR /app
RUN corepack enable pnpm && corepack prepare pnpm@latest --activate
COPY package.json pnpm-lock.yaml* ./
RUN pnpm install --frozen-lockfile
COPY package.json yarn.lock* package-lock.json* pnpm-lock.yaml* .npmrc* ./
RUN npm i
FROM base AS builder
WORKDIR /app
RUN corepack enable pnpm && corepack prepare pnpm@latest --activate
COPY --from=deps /app/node_modules ./node_modules
COPY . .
ENV NEXT_TELEMETRY_DISABLED=1
RUN pnpm run build
RUN npm run build
FROM base AS runner
WORKDIR /app
+3 -5
View File
@@ -6,14 +6,12 @@ RUN apk add --no-cache libc6-compat
WORKDIR /app
# Copy package files
COPY package.json pnpm-lock.yaml* ./
RUN corepack enable pnpm && corepack prepare pnpm@latest --activate
RUN pnpm install --frozen-lockfile
COPY package.json package-lock.json* pnpm-lock.yaml* ./
RUN npm ci
# Build the UI
FROM base AS builder
WORKDIR /app
RUN corepack enable pnpm && corepack prepare pnpm@latest --activate
COPY --from=deps /app/node_modules ./node_modules
COPY . .
@@ -29,7 +27,7 @@ ENV NODE_ENV=production
ENV NEXT_TELEMETRY_DISABLED=1
# Build the application
RUN pnpm run build && \
RUN npm run build && \
echo "UI build completed at $(date)"
# Use the builder stage as the final stage
+2 -450
View File
@@ -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 &quot;{statusSearch}&quot;
</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 &quot;{methodSearch.toUpperCase()}&quot;
</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 &quot;{endpointSearch}&quot;
</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
View File
@@ -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>
-3
View File
@@ -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;
}
+1 -6
View File
@@ -208,12 +208,7 @@ function ProviderBalance({
return <Skeleton className='h-9 w-24' />;
}
if (
error ||
!balanceData?.ok ||
balanceData.balance_data === undefined ||
balanceData.balance_data === null
) {
if (error || !balanceData?.ok || !balanceData.balance_data) {
return null;
}
+9257
View File
File diff suppressed because it is too large Load Diff
+9 -10
View File
@@ -15,7 +15,6 @@
"@dnd-kit/core": "^6.3.1",
"@dnd-kit/modifiers": "^9.0.0",
"@dnd-kit/sortable": "^10.0.0",
"@dnd-kit/utilities": "^3.2.2",
"@hookform/resolvers": "^5.0.1",
"@radix-ui/react-alert-dialog": "^1.1.10",
"@radix-ui/react-avatar": "^1.1.6",
@@ -41,17 +40,17 @@
"@radix-ui/react-toggle": "^1.1.6",
"@radix-ui/react-toggle-group": "^1.1.6",
"@radix-ui/react-tooltip": "^1.2.3",
"@tanstack/react-query": "^5.90.16",
"@tanstack/react-query": "^5.74.4",
"@tanstack/react-table": "^8.21.3",
"axios": "^1.13.2",
"class-variance-authority": "^0.7.1",
"clsx": "^2.1.1",
"cmdk": "^1.1.1",
"date-fns": "^4.1.0",
"date-fns": "^3.6.0",
"embla-carousel-react": "^8.6.0",
"input-otp": "^1.4.2",
"lucide-react": "^0.562.0",
"next": "15.5.9",
"lucide-react": "^0.501.0",
"next": "15.3.1",
"next-themes": "^0.4.6",
"qrcode": "^1.5.4",
"qrcode.react": "^4.2.0",
@@ -68,21 +67,21 @@
"zustand": "^5.0.3"
},
"devDependencies": {
"@eslint/eslintrc": "^3.3.3",
"@tailwindcss/postcss": "^4.1.18",
"@tanstack/react-query-devtools": "^5.91.2",
"@eslint/eslintrc": "^3",
"@tailwindcss/postcss": "^4",
"@tanstack/react-query-devtools": "^5.74.4",
"@types/node": "^20",
"@types/qrcode": "^1.5.6",
"@types/react": "^19",
"@types/react-dom": "^19",
"eslint": "^9.25.0",
"eslint-config-next": "15.5.9",
"eslint-config-next": "15.3.1",
"eslint-config-prettier": "^10.1.2",
"eslint-plugin-prettier": "^5.2.6",
"eslint-plugin-react": "^7.37.5",
"prettier": "^3.5.3",
"prettier-plugin-tailwindcss": "^0.6.11",
"tailwindcss": "^4.1.18",
"tailwindcss": "^4",
"typescript": "^5"
}
}
+544 -544
View File
File diff suppressed because it is too large Load Diff
+1 -7
View File
@@ -22,12 +22,6 @@
"@/*": ["./*"]
}
},
"include": [
"next-env.d.ts",
"**/*.ts",
"**/*.tsx",
".next/types/**/*.ts",
".next/dev/types/**/*.ts"
],
"include": ["next-env.d.ts", "**/*.ts", "**/*.tsx", ".next/types/**/*.ts"],
"exclude": ["node_modules"]
}
Generated
+1 -1
View File
@@ -1878,7 +1878,7 @@ wheels = [
[[package]]
name = "routstr"
version = "0.2.2"
version = "0.2.1"
source = { editable = "." }
dependencies = [
{ name = "aiosqlite" },