Compare commits

...

28 Commits

Author SHA1 Message Date
Shroominic
4974a22d0f reproducible cursor problems 2025-12-26 17:27:49 +01:00
shroominic
634a473f50 Merge pull request #276 from Routstr/force-sats-pricing
Force sats pricing
2025-12-26 12:36:02 +01:00
Shroominic
ea655b748b optimize startup time 2025-12-26 12:33:42 +01:00
9qeklajc
2d247ddc8b fmt 2025-12-26 10:29:51 +01:00
9qeklajc
c064452aea Merge branch 'v0.2.1' into force-sats-pricing 2025-12-26 10:27:11 +01:00
9qeklajc
1c6a603042 Merge pull request #271 from Routstr/embedding-with-aliases
Embedding with aliases
2025-12-26 10:26:38 +01:00
9qeklajc
84b0007b05 force sats pricig 2025-12-26 09:55:15 +01:00
Shroominic
525476ccfa fix alias not displaying correctly 2025-12-25 22:34:52 +01:00
Shroominic
b54812cb04 fix: alias_id not displayed correctly by dashboard 2025-12-25 22:28:26 +01:00
Shroominic
ee508cbb3a rm model cleanup 2025-12-25 21:50:09 +01:00
Shroominic
0c61fdee07 ruff format 2025-12-25 21:48:20 +01:00
Shroominic
b9418db31f rm model cleanup task 2025-12-25 21:47:17 +01:00
Shroominic
71e7c2171b add missing migration 2025-12-25 19:36:26 +01:00
Shroominic
a3e8d5fd38 ignore cloudflare headers from logs 2025-12-24 15:09:03 +01:00
9qeklajc
5b8e56f590 Merge branch 'v0.2.1' into embedding-with-aliases
# Conflicts:
#	routstr/upstream/base.py
2025-12-23 23:43:46 +01:00
9qeklajc
a6d0bd1a19 update aliases 2025-12-23 23:43:16 +01:00
shroominic
9594e9fb52 Merge pull request #251 from Routstr/embeddings-integration
Embeddings integration
2025-12-23 20:53:26 +01:00
9qeklajc
b01c7b2e56 Merge branch 'v0.2.1' into embeddings 2025-12-22 21:04:28 +01:00
shroominic
d203370f01 Merge pull request #265 from Routstr/update-cost-calculation
use cost field if available
2025-12-22 10:20:35 +01:00
Shroominic
e39742c429 rm completion_image pricing with manual overrides 2025-12-22 09:35:23 +01:00
Shroominic
301dd81215 include openrouters upstream_inference_cost + manually add gemini image completion cost 2025-12-21 09:55:25 +01:00
9qeklajc
5416cefd87 use cost field if available 2025-12-20 16:16:23 +01:00
shroominic
8edc3512c1 Merge pull request #258 from Routstr/remove-flaky-wallet-tests
Remove flaky wallet tests
2025-12-19 10:54:26 +01:00
Shroominic
590fb4bc2c ruff fix 2025-12-11 14:13:41 +08:00
Shroominic
5db9abc3ce remove flaky wallet tests 2025-12-11 14:12:34 +08:00
9qeklajc
8df0c17bc3 embedding integration 2025-12-03 22:20:31 +01:00
9qeklajc
7bc9ee0653 Merge branch 'gemini-upstream' into embeddings 2025-12-03 22:01:39 +01:00
9qeklajc
355f8601c1 embedding integration 2025-12-03 21:58:31 +01:00
26 changed files with 4188 additions and 475 deletions

View File

@@ -0,0 +1,44 @@
# Reproducing Cursor Problems
Each subdirectory contains a `request.json` (the request body) and `response.json` (the error response received).
## Using curl to reproduce
From the `routstr-core/` directory:
```bash
# OpenAI model error
curl -X POST https://staging.routstr.com/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $API_KEY" \
-d @cursor-problems/openai-model-error/request.json
# Anthropic internal error
curl -X POST https://staging.routstr.com/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $API_KEY" \
-d @cursor-problems/anthropic-internal-error/request.json
# Model not found error
curl -X POST https://staging.routstr.com/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $API_KEY" \
-d @cursor-problems/model-not-found-error/request.json
# Upstream rate limit error
curl -X POST https://staging.routstr.com/v1/chat/completions \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $API_KEY" \
-d @cursor-problems/upstream-rate-limit-error/request.json
```
## Generic pattern
```bash
curl -X POST <API_ENDPOINT> \
-H "Content-Type: application/json" \
-H "Authorization: Bearer $API_KEY" \
-d @cursor-problems/<directory>/request.json
```
The `-d @filename` syntax tells curl to read the request body from a file.

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,8 @@
{
"error": {
"message": "Internal Server Error",
"type": "upstream_error",
"code": 502
},
"request_id": "4a04e4f8-4a31-45f1-8189-455c86fc4e89"
}

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,8 @@
{
"error": {
"message": "Model 'claude-4.5-sonnet-thinking' not found",
"type": "invalid_model",
"code": 400
},
"request_id": "d410f512-3221-4047-a4ee-9be6e3fabe38"
}

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,8 @@
{
"error": {
"message": "Input required: specify \"prompt\" or \"messages\"",
"type": "invalid_request_error",
"code": 400
},
"request_id": "586e0aec-351f-413a-8641-ddda4a0cbadf"
}

File diff suppressed because one or more lines are too long

View File

@@ -0,0 +1,8 @@
{
"error": {
"message": "Upstream request failed",
"type": "rate_limit_exceeded",
"code": 429
},
"request_id": "80657fc6-4bca-4cb1-945d-ea65ec8a53c4"
}

View File

@@ -0,0 +1,37 @@
"""alias-ids
Revision ID: b9667ffc5701
Revises: lightning_invoices
Create Date: 2025-12-25 19:30:44.673350
"""
import sqlalchemy as sa
import sqlmodel
from alembic import op
# revision identifiers, used by Alembic.
revision = "b9667ffc5701"
down_revision = "lightning_invoices"
branch_labels = None
depends_on = None
def upgrade() -> None:
# ### commands auto generated by Alembic ###
op.add_column(
"models",
sa.Column("canonical_slug", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
)
op.add_column(
"models",
sa.Column("alias_ids", sqlmodel.sql.sqltypes.AutoString(), nullable=True),
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column("models", "alias_ids")
op.drop_column("models", "canonical_slug")
# ### end Alembic commands ###

View File

@@ -206,19 +206,20 @@ def create_model_mappings(
alias: str, model: "Model", provider: "BaseUpstreamProvider"
) -> None:
"""Set alias to model/provider if not set or if new model is preferred."""
existing_model = model_instances.get(alias)
alias_lower = alias.lower()
existing_model = model_instances.get(alias_lower)
if not existing_model:
# No existing mapping, set it
model_instances[alias] = model
provider_map[alias] = provider
model_instances[alias_lower] = model
provider_map[alias_lower] = provider
else:
# Check if candidate should replace existing
existing_provider = provider_map[alias]
existing_provider = provider_map[alias_lower]
if should_prefer_model(
model, provider, existing_model, existing_provider, alias
):
model_instances[alias] = model
provider_map[alias] = provider
model_instances[alias_lower] = model
provider_map[alias_lower] = provider
def process_provider_models(
upstream: "BaseUpstreamProvider", is_openrouter: bool = False

View File

@@ -1493,20 +1493,8 @@ class ModelCreate(BaseModel):
per_request_limits: dict[str, object] | None = None
top_provider: dict[str, object] | None = None
upstream_provider_id: int | None = None
enabled: bool = True
class ModelUpdate(BaseModel):
id: str
name: str
description: str
created: int
context_length: int
architecture: dict[str, object]
pricing: dict[str, object]
per_request_limits: dict[str, object] | None = None
top_provider: dict[str, object] | None = None
upstream_provider_id: int | None = None
canonical_slug: str | None = None
alias_ids: list[str] | None = None
enabled: bool = True
@@ -2416,44 +2404,80 @@ async def admin_upstream_providers(request: Request) -> str:
"/api/upstream-providers/{provider_id}/models",
dependencies=[Depends(require_admin_api)],
)
async def create_provider_model(
async def upsert_provider_model(
provider_id: int, payload: ModelCreate
) -> dict[str, object]:
print(payload)
logger.info(
f"UPSERT_PROVIDER_MODEL called: provider_id={provider_id}, model_id={payload.id}"
)
async with create_session() as session:
provider = await session.get(UpstreamProviderRow, provider_id)
if not provider:
raise HTTPException(status_code=404, detail="Provider not found")
exists = await session.get(ModelRow, (payload.id, provider_id))
if exists:
raise HTTPException(
status_code=409,
detail="Model with this ID already exists for this provider",
)
# Try to get existing model
existing_row = await session.get(ModelRow, (payload.id, provider_id))
row = ModelRow(
id=payload.id,
name=payload.name,
description=payload.description,
created=int(payload.created),
context_length=int(payload.context_length),
architecture=json.dumps(payload.architecture),
pricing=json.dumps(payload.pricing),
sats_pricing=None,
per_request_limits=(
if existing_row:
# Update existing model
logger.info(f"Updating existing model: {payload.id}")
existing_row.name = payload.name
existing_row.description = payload.description
existing_row.created = int(payload.created)
existing_row.context_length = int(payload.context_length)
existing_row.architecture = json.dumps(payload.architecture)
existing_row.pricing = json.dumps(payload.pricing)
existing_row.sats_pricing = None
existing_row.per_request_limits = (
json.dumps(payload.per_request_limits)
if payload.per_request_limits is not None
else None
),
top_provider=(
)
existing_row.top_provider = (
json.dumps(payload.top_provider) if payload.top_provider else None
),
upstream_provider_id=provider_id,
enabled=payload.enabled,
)
session.add(row)
await session.commit()
await session.refresh(row)
)
existing_row.canonical_slug = payload.canonical_slug
existing_row.alias_ids = (
json.dumps(payload.alias_ids) if payload.alias_ids else None
)
existing_row.enabled = payload.enabled
session.add(existing_row)
await session.commit()
await session.refresh(existing_row)
row = existing_row
else:
# Create new model
logger.info(f"Creating new model: {payload.id}")
row = ModelRow(
id=payload.id,
name=payload.name,
description=payload.description,
created=int(payload.created),
context_length=int(payload.context_length),
architecture=json.dumps(payload.architecture),
pricing=json.dumps(payload.pricing),
sats_pricing=None,
per_request_limits=(
json.dumps(payload.per_request_limits)
if payload.per_request_limits is not None
else None
),
top_provider=(
json.dumps(payload.top_provider) if payload.top_provider else None
),
canonical_slug=payload.canonical_slug,
alias_ids=(
json.dumps(payload.alias_ids) if payload.alias_ids else None
),
upstream_provider_id=provider_id,
enabled=payload.enabled,
)
session.add(row)
await session.commit()
await session.refresh(row)
await refresh_model_maps()
return _row_to_model(
@@ -2461,6 +2485,20 @@ async def create_provider_model(
).dict() # type: ignore
@admin_router.patch(
"/api/upstream-providers/{provider_id}/models/{model_id:path}",
dependencies=[Depends(require_admin_api)],
)
async def update_provider_model_legacy(
provider_id: int, model_id: str, payload: ModelCreate
) -> dict[str, object]:
"""Legacy PATCH endpoint - redirects to upsert POST endpoint for backward compatibility."""
logger.info(
f"LEGACY_PATCH_UPDATE called: provider_id={provider_id}, model_id={model_id}"
)
return await upsert_provider_model(provider_id, payload)
@admin_router.get(
"/api/upstream-providers/{provider_id}/models/{model_id:path}",
dependencies=[Depends(require_admin_api)],
@@ -2481,76 +2519,6 @@ async def get_provider_model(provider_id: int, model_id: str) -> dict[str, objec
).dict() # type: ignore
@admin_router.patch(
"/api/upstream-providers/{provider_id}/models/{model_id:path}",
dependencies=[Depends(require_admin_api)],
)
async def update_provider_model(
provider_id: int, model_id: str, payload: ModelUpdate
) -> dict[str, object]:
if payload.id != model_id:
raise HTTPException(status_code=400, detail="Path id does not match payload id")
async with create_session() as session:
provider = await session.get(UpstreamProviderRow, provider_id)
if not provider:
raise HTTPException(status_code=404, detail="Provider not found")
row = await session.get(ModelRow, (model_id, provider_id))
if not row:
raise HTTPException(
status_code=404, detail="Model not found for this provider"
)
row.name = payload.name
row.description = payload.description
row.created = int(payload.created)
row.context_length = int(payload.context_length)
row.architecture = json.dumps(payload.architecture)
row.pricing = json.dumps(payload.pricing)
row.sats_pricing = None
row.per_request_limits = (
json.dumps(payload.per_request_limits)
if payload.per_request_limits is not None
else None
)
row.top_provider = (
json.dumps(payload.top_provider) if payload.top_provider else None
)
was_disabled = not row.enabled
row.enabled = payload.enabled
session.add(row)
await session.commit()
await session.refresh(row)
if was_disabled and payload.enabled:
from ..payment.models import _cleanup_enabled_models_once
try:
await _cleanup_enabled_models_once()
except Exception as e:
logger.warning(
f"Failed to run model cleanup after enabling: {e}",
extra={"model_id": model_id, "error": str(e)},
)
await refresh_model_maps()
return _row_to_model(
row, apply_provider_fee=True, provider_fee=provider.provider_fee
).dict() # type: ignore
@admin_router.put(
"/api/upstream-providers/{provider_id}/models/{model_id:path}",
dependencies=[Depends(require_admin_api)],
)
async def update_provider_model_put(
provider_id: int, model_id: str, payload: ModelUpdate
) -> dict[str, object]:
return await update_provider_model(provider_id, model_id, payload)
@admin_router.delete(
"/api/upstream-providers/{provider_id}/models/{model_id:path}",
dependencies=[Depends(require_admin_api)],

View File

@@ -68,6 +68,10 @@ class ModelRow(SQLModel, table=True): # type: ignore
sats_pricing: str | None = Field(default=None)
per_request_limits: str | None = Field(default=None)
top_provider: str | None = Field(default=None)
canonical_slug: str | None = Field(default=None, description="Canonical model slug")
alias_ids: str | None = Field(
default=None, description="JSON array of model alias IDs"
)
enabled: bool = Field(default=True, description="Whether this model is enabled")
upstream_provider: "UpstreamProviderRow" = Relationship(back_populates="models")

View File

@@ -14,7 +14,6 @@ from ..balance import balance_router, deprecated_wallet_router
from ..discovery import providers_cache_refresher, providers_router
from ..nip91 import announce_provider
from ..payment.models import (
cleanup_enabled_models_periodically,
models_router,
update_sats_pricing,
)
@@ -49,7 +48,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
nip91_task = None
providers_task = None
models_refresh_task = None
models_cleanup_task = None
model_maps_refresh_task = None
try:
@@ -80,13 +78,17 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
_update_prices_task = asyncio.create_task(_update_prices())
_initialize_upstreams_task = asyncio.create_task(initialize_upstreams())
# ensure both setup tasks complete
await asyncio.gather(
_update_prices_task, _initialize_upstreams_task, return_exceptions=True
)
btc_price_task = asyncio.create_task(update_prices_periodically())
pricing_task = asyncio.create_task(update_sats_pricing())
if global_settings.models_refresh_interval_seconds > 0:
models_refresh_task = asyncio.create_task(
refresh_upstreams_models_periodically(get_upstreams())
)
models_cleanup_task = asyncio.create_task(cleanup_enabled_models_periodically())
model_maps_refresh_task = asyncio.create_task(refresh_model_maps_periodically())
payout_task = asyncio.create_task(periodic_payout())
if global_settings.nsec:
@@ -94,11 +96,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
if global_settings.providers_refresh_interval_seconds > 0:
providers_task = asyncio.create_task(providers_cache_refresher())
# ensure both setup tasks complete
await asyncio.gather(
_update_prices_task, _initialize_upstreams_task, return_exceptions=True
)
yield
except asyncio.CancelledError:
@@ -125,8 +122,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
providers_task.cancel()
if models_refresh_task is not None:
models_refresh_task.cancel()
if models_cleanup_task is not None:
models_cleanup_task.cancel()
if model_maps_refresh_task is not None:
model_maps_refresh_task.cancel()
@@ -144,8 +139,6 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
tasks_to_wait.append(providers_task)
if models_refresh_task is not None:
tasks_to_wait.append(models_refresh_task)
if models_cleanup_task is not None:
tasks_to_wait.append(models_cleanup_task)
if model_maps_refresh_task is not None:
tasks_to_wait.append(model_maps_refresh_task)

View File

@@ -55,7 +55,16 @@ class LoggingMiddleware(BaseHTTPMiddleware):
"headers": {
k: v
for k, v in request.headers.items()
if k.lower() not in ["authorization", "x-cashu", "cookie"]
if k.lower()
not in [
"authorization",
"x-cashu",
"cookie",
"cf-connecting-ip",
"cf-ipcountry",
"x-forwarded-for",
"x-real-ip",
]
},
"body_size": len(request_body) if request_body else 0,
},

View File

@@ -5,6 +5,7 @@ from pydantic.v1 import BaseModel
from ..core import get_logger
from ..core.db import AsyncSession
from ..core.settings import settings
from .price import sats_usd_price
logger = get_logger(__name__)
@@ -64,6 +65,56 @@ async def calculate_cost( # todo: can be sync
)
return cost_data
usage_data = response_data["usage"]
usd_cost = 0.0
# Prioritize cost_details.upstream_inference_cost
if "cost_details" in usage_data:
usd_cost = float(
usage_data["cost_details"].get("upstream_inference_cost", 0) or 0
)
# Fallback to cost field if upstream_inference_cost is 0
if usd_cost == 0 and "cost" in usage_data:
try:
usd_cost = float(usage_data.get("cost", 0) or 0)
except Exception:
pass
if usd_cost > 0:
try:
sats_per_usd = 1.0 / sats_usd_price()
cost_in_sats = usd_cost * sats_per_usd
cost_in_msats = math.ceil(cost_in_sats * 1000)
logger.info(
"Using cost from usage data/details",
extra={
"usd_cost": usd_cost,
"cost_in_sats": cost_in_sats,
"cost_in_msats": cost_in_msats,
"model": response_data.get("model", "unknown"),
},
)
return CostData(
base_msats=-1,
input_msats=-1, # Cost field doesn't break down by token type
output_msats=-1,
total_msats=cost_in_msats,
)
except Exception as e:
logger.warning(
"Error calculating cost from usage data",
extra={
"error": str(e),
"usd_cost": usd_cost,
"model": response_data.get("model", "unknown"),
},
)
# Fall through to token-based calculation
MSATS_PER_1K_INPUT_TOKENS: float = (
float(settings.fixed_per_1k_input_tokens) * 1000.0
)
@@ -129,10 +180,19 @@ async def calculate_cost( # todo: can be sync
)
return cost_data
input_tokens = response_data.get("usage", {}).get("prompt_tokens", 0)
output_tokens = response_data.get("usage", {}).get("completion_tokens", 0)
input_tokens = usage_data.get("prompt_tokens", 0)
output_tokens = usage_data.get("completion_tokens", 0)
# added for response api
input_tokens = (
input_tokens if input_tokens != 0 else usage_data.get("input_tokens", 0)
)
output_tokens = (
output_tokens if output_tokens != 0 else usage_data.get("output_tokens", 0)
)
input_msats = round(input_tokens / 1000 * MSATS_PER_1K_INPUT_TOKENS, 3)
output_msats = round(output_tokens / 1000 * MSATS_PER_1K_OUTPUT_TOKENS, 3)
token_based_cost = math.ceil(input_msats + output_msats)

View File

@@ -5,10 +5,9 @@ import random
import httpx
from fastapi import APIRouter, Depends
from pydantic.v1 import BaseModel
from sqlmodel import select
from sqlmodel.ext.asyncio.session import AsyncSession
from ..core.db import ModelRow, create_session, get_session
from ..core.db import ModelRow, get_session
from ..core.logging import get_logger
from ..core.settings import settings
from .price import sats_usd_price
@@ -185,6 +184,7 @@ def _row_to_model(
enabled=row.enabled,
upstream_provider_id=row.upstream_provider_id,
canonical_slug=getattr(row, "canonical_slug", None),
alias_ids=json.loads(row.alias_ids) if row.alias_ids else None,
)
if apply_provider_fee:
@@ -382,7 +382,7 @@ def _update_model_sats_pricing(model: Model, sats_to_usd: float) -> Model:
async def _update_sats_pricing_once() -> None:
"""Update sats pricing once for all provider models (in-memory only)."""
from ..proxy import get_upstreams
from ..proxy import get_upstreams, refresh_model_maps
upstreams = get_upstreams()
sats_to_usd = sats_usd_price()
@@ -399,6 +399,7 @@ async def _update_sats_pricing_once() -> None:
if updated_count > 0:
logger.info("Updated sats pricing", extra={"models_updated": updated_count})
await refresh_model_maps()
async def update_sats_pricing() -> None:
@@ -409,7 +410,13 @@ async def update_sats_pricing() -> None:
except Exception:
pass
await _update_sats_pricing_once()
try:
await _update_sats_pricing_once()
except Exception as e:
logger.warning(
"Initial sats pricing update failed (will retry in loop)",
extra={"error": str(e)},
)
while True:
try:
@@ -433,114 +440,6 @@ async def update_sats_pricing() -> None:
logger.error(f"Error updating sats pricing: {e}")
async def cleanup_enabled_models_periodically() -> None:
"""Background task to clean up enabled models that match upstream pricing.
When model is enabled (enabled=True), remove it from DB if it matches upstream pricing.
Keep it in DB only if pricing differs from upstream or if it's disabled.
"""
interval = getattr(
settings, "models_cleanup_interval_seconds", 300
) # 5 minutes default
if not interval or interval <= 0:
return
while True:
try:
await _cleanup_enabled_models_once()
except asyncio.CancelledError:
break
except Exception as e:
logger.error(
"Error during enabled models cleanup",
extra={"error": str(e), "error_type": type(e).__name__},
)
try:
jitter = max(0.0, float(interval) * 0.1)
await asyncio.sleep(interval + random.uniform(0, jitter))
except asyncio.CancelledError:
break
async def _cleanup_enabled_models_once() -> None:
"""Clean up enabled models that match upstream pricing."""
from ..proxy import get_upstreams
async with create_session() as session:
# Get all enabled models from DB
result = await session.exec(
select(ModelRow).where(
ModelRow.enabled, # Only enabled models
)
)
db_models = result.all()
if not db_models:
return
upstreams = get_upstreams()
models_to_remove = []
for db_model in db_models:
# Find corresponding upstream model
upstream_model = None
for upstream in upstreams:
upstream_model = upstream.get_cached_model_by_id(db_model.id)
if upstream_model:
break
if not upstream_model:
continue
# Compare pricing to see if they match
db_pricing = json.loads(db_model.pricing)
upstream_pricing = upstream_model.pricing.dict()
# Check if pricing matches (with small tolerance for float comparison)
pricing_matches = _pricing_matches(db_pricing, upstream_pricing)
if pricing_matches:
models_to_remove.append(db_model)
logger.info(
f"Removing enabled model {db_model.id} - matches upstream pricing",
extra={"model_id": db_model.id},
)
# Remove models that match upstream pricing
for model in models_to_remove:
await session.delete(model)
if models_to_remove:
await session.commit()
logger.info(
f"Cleaned up {len(models_to_remove)} enabled models that match upstream pricing"
)
def _pricing_matches(
db_pricing: dict, upstream_pricing: dict, tolerance: float = 0.0
) -> bool:
"""Check if pricing dictionaries match within tolerance."""
keys_to_compare = [
"prompt",
"completion",
"request",
"image",
"web_search",
"internal_reasoning",
]
for key in keys_to_compare:
db_val = int(float(db_pricing.get(key, 0.0)) * 1000000)
upstream_val = int(float(upstream_pricing.get(key, 0.0)) * 1000000)
if abs(db_val - upstream_val) > tolerance:
return False
return True
@models_router.get("/v1/models")
@models_router.get("/models", include_in_schema=False)
async def models(session: AsyncSession = Depends(get_session)) -> dict:

View File

@@ -70,7 +70,7 @@ def get_model_instance(model_id: str) -> Model | None:
def get_provider_for_model(model_id: str) -> BaseUpstreamProvider | None:
"""Get UpstreamProvider for model ID from global cache."""
return _provider_map.get(model_id)
return _provider_map.get(model_id.lower())
def get_unique_models() -> list[Model]:
@@ -80,31 +80,27 @@ def get_unique_models() -> list[Model]:
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
# Gather database overrides and disabled models
async with create_session() as session:
result = await session.exec(select(ModelRow).where(ModelRow.enabled))
override_rows = result.all()
provider_result = await session.exec(select(UpstreamProviderRow))
providers_by_id = {p.id: p for p in provider_result.all()}
overrides_by_id: dict[str, tuple[ModelRow, float]] = {
row.id: (
row,
providers_by_id[row.upstream_provider_id].provider_fee
if row.upstream_provider_id in providers_by_id
else 1.01,
)
for row in override_rows
if row.upstream_provider_id is not None
}
disabled_result = await session.exec(
select(ModelRow.id).where(ModelRow.enabled == False) # noqa: E712
# Fetch all providers with their models in a single logical operation
query = select(UpstreamProviderRow).options(
selectinload(UpstreamProviderRow.models) # type: ignore
)
disabled_model_ids = {row for row in disabled_result.all()}
result = await session.exec(query)
provider_rows = result.all()
overrides_by_id: dict[str, tuple[ModelRow, float]] = {}
disabled_model_ids: set[str] = set()
for provider in provider_rows:
for model in provider.models:
if model.enabled:
overrides_by_id[model.id] = (model, provider.provider_fee)
else:
disabled_model_ids.add(model.id)
_model_instances, _provider_map, _unique_models = create_model_mappings(
upstreams=_upstreams,

View File

@@ -1,5 +1,6 @@
from __future__ import annotations
import asyncio
import os
import re
from typing import TYPE_CHECKING
@@ -7,6 +8,7 @@ from typing import TYPE_CHECKING
if TYPE_CHECKING:
from ..core.settings import Settings
from sqlmodel import select
from ..core import get_logger
from ..core.db import AsyncSession, ModelRow, UpstreamProviderRow, create_session
@@ -145,6 +147,17 @@ async def refresh_upstreams_models_periodically(
f"Error refreshing models for {upstream.base_url}",
extra={"error": str(e), "error_type": type(e).__name__},
)
try:
from ..payment.models import _update_sats_pricing_once
await _update_sats_pricing_once()
except Exception as e:
logger.warning(f"Failed to update pricing after model refresh: {e}")
from ..proxy import refresh_model_maps
await refresh_model_maps()
except asyncio.CancelledError:
break
except Exception as e:
@@ -166,8 +179,6 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
Seeds database with providers from settings if empty, then loads and instantiates
provider instances from database records, and refreshes their models cache.
"""
from sqlmodel import select
from ..core.settings import settings
async with create_session() as session:
@@ -183,16 +194,16 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
result = await session.exec(select(UpstreamProviderRow))
existing_providers = result.all()
upstreams: list[BaseUpstreamProvider] = []
for provider_row in existing_providers:
async def _init_single_provider(
provider_row: UpstreamProviderRow,
) -> BaseUpstreamProvider | None:
if not provider_row.enabled:
logger.debug(f"Skipping disabled provider: {provider_row.base_url}")
continue
return None
provider = _instantiate_provider(provider_row)
if provider:
await provider.refresh_models_cache()
upstreams.append(provider)
logger.debug(
f"Initialized {provider_row.provider_type} provider",
extra={
@@ -200,6 +211,12 @@ async def init_upstreams() -> list[BaseUpstreamProvider]:
"models_cached": len(provider.get_cached_models()),
},
)
return provider
return None
tasks = [_init_single_provider(row) for row in existing_providers]
results = await asyncio.gather(*tasks)
upstreams = [p for p in results if p is not None]
return upstreams

View File

@@ -3,7 +3,6 @@ Integration tests for wallet authentication system including API key generation
Tests POST /v1/wallet/topup endpoint and authorization header validation.
"""
import hashlib
from datetime import datetime, timedelta
from typing import Any
@@ -15,7 +14,6 @@ from routstr.core.db import ApiKey
from .utils import (
CashuTokenGenerator,
ConcurrencyTester,
ResponseValidator,
)
@@ -389,69 +387,6 @@ async def test_api_key_with_expiry_time(
# The expiry time and refund address functionality is tested elsewhere
@pytest.mark.integration
@pytest.mark.asyncio
async def test_concurrent_token_submissions(
integration_client: AsyncClient, testmint_wallet: Any, integration_session: Any
) -> None:
"""Test concurrent submissions of different tokens"""
# Generate multiple unique tokens with known amounts
num_tokens = 10
tokens = []
expected_balances = {}
for i in range(num_tokens):
amount = 100 + i * 10
token = await testmint_wallet.mint_tokens(amount)
tokens.append(token)
# Store expected balance by token hash
hashed_key = hashlib.sha256(token.encode()).hexdigest()
expected_balances[hashed_key] = amount * 1000 # msats
# Create concurrent requests
requests = [
{
"method": "GET",
"url": "/v1/wallet/info",
"headers": {"Authorization": f"Bearer {token}"},
}
for token in tokens
]
# Execute concurrently
tester = ConcurrencyTester()
responses = await tester.run_concurrent_requests(
integration_client, requests, max_concurrent=5
)
# All should succeed
assert len(responses) == num_tokens
api_keys = set()
for response in responses:
assert response.status_code == 200
data = response.json()
api_key = data["api_key"]
api_keys.add(api_key)
# Verify balance matches the expected amount
hashed_key = api_key[3:] # Remove "sk-" prefix
assert data["balance"] == expected_balances[hashed_key]
# Should have created unique API keys
assert len(api_keys) == num_tokens
# Verify all keys exist in database
for api_key in api_keys:
hashed_key = api_key[3:] # Remove "sk-" prefix
result = await integration_session.execute(
select(ApiKey).where(ApiKey.hashed_key == hashed_key) # type: ignore[arg-type]
)
db_key = result.scalar_one()
assert db_key.balance == expected_balances[hashed_key]
@pytest.mark.integration
@pytest.mark.asyncio
async def test_authorization_with_cashu_token_directly(
@@ -504,48 +439,6 @@ async def test_x_cashu_header_support(
assert response.status_code == 200
@pytest.mark.integration
@pytest.mark.asyncio
@pytest.mark.slow
async def test_api_key_consistency_under_load(
integration_client: AsyncClient, testmint_wallet: Any, integration_session: Any
) -> None:
"""Test API key generation consistency under concurrent load"""
# Generate a single token
token = await testmint_wallet.mint_tokens(1000)
# First request to create the API key
integration_client.headers["Authorization"] = f"Bearer {token}"
initial_response = await integration_client.get("/v1/wallet/info")
assert initial_response.status_code == 200
expected_api_key = initial_response.json()["api_key"]
expected_balance = initial_response.json()["balance"]
# Try to use the same token concurrently multiple times
# All should return the same API key since it's already created
requests = [
{
"method": "GET",
"url": "/v1/wallet/info",
"headers": {"Authorization": f"Bearer {token}"},
}
for _ in range(20) # 20 concurrent attempts
]
tester = ConcurrencyTester()
responses = await tester.run_concurrent_requests(
integration_client, requests, max_concurrent=10
)
# All should succeed and return the same API key
for response in responses:
assert response.status_code == 200
data = response.json()
assert data["api_key"] == expected_api_key
assert data["balance"] == expected_balance
@pytest.mark.integration
@pytest.mark.asyncio
async def test_database_timestamp_accuracy(

View File

@@ -13,7 +13,7 @@ from sqlmodel import select, update
from routstr.core.db import ApiKey
from .utils import ConcurrencyTester, ResponseValidator
from .utils import ResponseValidator
@pytest.mark.integration
@@ -204,45 +204,6 @@ async def test_expired_api_key_behavior(
assert db_key.refund_address == "test@lightning.address"
@pytest.mark.integration
@pytest.mark.asyncio
async def test_concurrent_access_same_api_key(
integration_client: AsyncClient, authenticated_client: AsyncClient
) -> None:
"""Test concurrent access with the same API key"""
# Get the API key from authenticated client
response = await authenticated_client.get("/v1/wallet/")
api_key = response.json()["api_key"]
initial_balance = response.json()["balance"]
# Create multiple concurrent requests
requests = []
for i in range(20):
# Alternate between both endpoints
endpoint = "/v1/wallet/" if i % 2 == 0 else "/v1/wallet/info"
requests.append(
{
"method": "GET",
"url": endpoint,
"headers": {"Authorization": f"Bearer {api_key}"},
}
)
# Execute concurrently
tester = ConcurrencyTester()
responses = await tester.run_concurrent_requests(
integration_client, requests, max_concurrent=10
)
# All should succeed with consistent data
for response in responses:
assert response.status_code == 200
data = response.json()
assert data["api_key"] == api_key
assert data["balance"] == initial_balance
@pytest.mark.integration
@pytest.mark.asyncio
async def test_wallet_info_data_consistency(

View File

@@ -15,7 +15,6 @@ from routstr.core.db import ApiKey
from .utils import (
CashuTokenGenerator,
ConcurrencyTester,
ResponseValidator,
)
@@ -284,60 +283,6 @@ async def test_transaction_history_tracking( # type: ignore[no-untyped-def]
assert response.status_code == 400
@pytest.mark.integration
@pytest.mark.asyncio
async def test_concurrent_topups_same_api_key( # type: ignore[no-untyped-def]
integration_client: AsyncClient,
authenticated_client: AsyncClient,
testmint_wallet: Any,
) -> None:
"""Test concurrent top-ups to the same API key"""
# Get API key
response = await authenticated_client.get("/v1/wallet/")
api_key = response.json()["api_key"]
initial_balance = response.json()["balance"]
# Generate multiple unique tokens
num_tokens = 10
tokens = []
total_amount = 0
for i in range(num_tokens):
amount = 100 + i * 10 # Different amounts
token = await testmint_wallet.mint_tokens(amount)
tokens.append(token)
total_amount += amount
# Create concurrent top-up requests
requests = [
{
"method": "POST",
"url": "/v1/wallet/topup",
"params": {"cashu_token": token},
"headers": {"Authorization": f"Bearer {api_key}"},
}
for token in tokens
]
# Execute concurrently
tester = ConcurrencyTester()
responses = await tester.run_concurrent_requests(
integration_client, requests, max_concurrent=5
)
# All should succeed
for response in responses:
assert response.status_code == 200
assert "msats" in response.json()
# Verify final balance is correct
final_response = await authenticated_client.get("/v1/wallet/")
final_balance = final_response.json()["balance"]
expected_balance = initial_balance + (total_amount * 1000)
assert final_balance == expected_balance
@pytest.mark.integration
@pytest.mark.asyncio
async def test_topup_during_active_proxy_request( # type: ignore[no-untyped-def]

View File

@@ -41,6 +41,10 @@ export function ModelSearchFilter({
model.name.toLowerCase().includes(query) ||
model.full_name.toLowerCase().includes(query) ||
model.provider.toLowerCase().includes(query) ||
(model.alias_ids &&
model.alias_ids.some((alias) =>
alias.toLowerCase().includes(query)
)) ||
(model.description &&
model.description.toLowerCase().includes(query)) ||
model.modelType.toLowerCase().includes(query)

View File

@@ -542,6 +542,7 @@ export function ModelSelector({
top_provider: null,
upstream_provider_id: providerId,
enabled: model.isEnabled,
alias_ids: model.alias_ids || null,
};
setModelDialogState({
@@ -585,6 +586,7 @@ export function ModelSelector({
top_provider: null,
upstream_provider_id: providerId,
enabled: model.isEnabled,
alias_ids: model.alias_ids || null,
};
setModelDialogState({

View File

@@ -31,6 +31,7 @@ export const ModelSchema = z.object({
// API key type indicators
has_own_api_key: z.boolean(),
api_key_type: z.string(), // "individual" or "group"
alias_ids: z.array(z.string()).nullable().optional(),
});
// Schema for a model with additional provider-specific settings

View File

@@ -121,6 +121,7 @@ export interface AdminModelAsModel {
soft_deleted?: boolean;
has_own_api_key: boolean;
api_key_type: string;
alias_ids?: string[] | null;
}
export interface AdminModelGroup {
@@ -207,6 +208,7 @@ export class AdminService {
soft_deleted: !adminModel.enabled,
has_own_api_key: false,
api_key_type: 'group',
alias_ids: adminModel.alias_ids,
};
}
@@ -354,8 +356,9 @@ export class AdminService {
original: data.pricing,
converted: payload.pricing,
});
const model = await apiClient.patch<AdminModel>(
`/admin/api/upstream-providers/${providerId}/models/${encodeURIComponent(modelId)}`,
// Use the same POST endpoint for both create and update (upsert)
const model = await apiClient.post<AdminModel>(
`/admin/api/upstream-providers/${providerId}/models`,
payload
);
return {