enforce-models-refresh-from-upstream

use correct field label

migrations/versions/02650cd6f028_add_routstr_fees_table.py

@@ -0,0 +1,32 @@
+"""add routstr_fees table
+
+Revision ID: 02650cd6f028
+Revises: c3d4e5f6a7b8
+Create Date: 2026-04-24 00:00:00.000000
+"""
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "02650cd6f028"
+down_revision = "c3d4e5f6a7b8"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "routstr_fees",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("accumulated_msats", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("total_paid_msats", sa.Integer(), nullable=False, server_default="0"),
+        sa.Column("last_paid_at", sa.Integer(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    # Seed with a single row
+    op.execute("INSERT INTO routstr_fees (id, accumulated_msats, total_paid_msats) VALUES (1, 0, 0)")
+
+
+def downgrade() -> None:
+    op.drop_table("routstr_fees")

migrations/versions/e8f9a0b1c2d3_merge_heads.py

@@ -0,0 +1,20 @@
+"""merge heads: routstr_fees + api_key_to_cashu_transactions
+
+Revision ID: e8f9a0b1c2d3
+Revises: 02650cd6f028, d4e5f6a7b8c9
+Create Date: 2026-04-24 00:00:00.000000
+"""
+
+# revision identifiers, used by Alembic.
+revision = "e8f9a0b1c2d3"
+down_revision = ("02650cd6f028", "d4e5f6a7b8c9")
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "routstr"
-version = "0.4.1"
+version = "0.4.3"
 description = "Payment proxy for your LLM endpoint using cashu and nostr."
 readme = "README.md"
 requires-python = ">=3.11"

routstr/auth.py

@@ -12,7 +12,7 @@ from sqlalchemy.exc import IntegrityError
 from sqlmodel import col, select, update
 
 from .core import get_logger
-from .core.db import ApiKey, AsyncSession
+from .core.db import ApiKey, AsyncSession, accumulate_routstr_fee
 from .core.settings import settings
 from .payment.cost_calculation import (
     CostData,
@@ -25,6 +25,12 @@ from .wallet import credit_balance, deserialize_token_from_string
 logger = get_logger(__name__)
 payments_logger = get_logger("routstr.payments")
 
+# Routstr platform fee constants
+ROUTSTR_FEE_PERCENT: float = 2.1
+ROUTSTR_LN_ADDRESS: str = "npub130mznv74rxs032peqym6g3wqavh472623mt3z5w73xq9r6qqdufs7ql29s@npub.cash"
+ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS: int = 900
+ROUTSTR_FEE_DEFAULT_PAYOUT: int = 200
+
 # TODO: implement prepaid api key (not like it was before)
 # PREPAID_API_KEY = os.environ.get("PREPAID_API_KEY", None)
 # PREPAID_BALANCE = int(os.environ.get("PREPAID_BALANCE", "0")) * 1000  # Convert to msats
@@ -741,6 +747,17 @@ async def adjust_payment_for_tokens(
                 },
             )
 
+    async def _accumulate_fee(total_cost_msats: int) -> None:
+        if total_cost_msats > 0 and ROUTSTR_FEE_PERCENT > 0:
+            fee_msats = math.ceil(total_cost_msats * ROUTSTR_FEE_PERCENT / 100)
+            try:
+                await accumulate_routstr_fee(session, fee_msats)
+            except Exception as e:
+                logger.warning(
+                    "Failed to accumulate Routstr fee",
+                    extra={"error": str(e), "fee_msats": fee_msats},
+                )
+
     match await calculate_cost(response_data, deducted_max_cost, session):
         case MaxCostData() as cost:
             logger.debug(
@@ -832,6 +849,7 @@ async def adjust_payment_for_tokens(
                         "model": model,
                     },
                 )
+                await _accumulate_fee(cost.total_msats)
                 payments_logger.info(
                     "FINALIZE",
                     extra={
@@ -935,6 +953,7 @@ async def adjust_payment_for_tokens(
                 await session.refresh(billing_key)
                 if billing_key.hashed_key != key.hashed_key:
                     await session.refresh(key)
+                await _accumulate_fee(total_cost_msats)
                 payments_logger.info(
                     "FINALIZE",
                     extra={
@@ -1005,6 +1024,7 @@ async def adjust_payment_for_tokens(
                             "model": model,
                         },
                     )
+                    await _accumulate_fee(total_cost_msats)
                     payments_logger.info(
                         "FINALIZE",
                         extra={
@@ -1130,6 +1150,7 @@ async def adjust_payment_for_tokens(
                             "model": model,
                         },
                     )
+                    await _accumulate_fee(total_cost_msats)
                     payments_logger.info(
                         "FINALIZE",
                         extra={

routstr/core/db.py

@@ -12,7 +12,7 @@ from alembic.util.exc import CommandError
 from sqlalchemy import UniqueConstraint
 from sqlalchemy.exc import OperationalError
 from sqlalchemy.ext.asyncio.engine import create_async_engine
-from sqlmodel import Field, Relationship, SQLModel, func, select, update
+from sqlmodel import Field, Relationship, SQLModel, col, func, select, update
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from .logging import get_logger
@@ -231,6 +231,50 @@ class UpstreamProviderRow(SQLModel, table=True):  # type: ignore
     )
 
 
+class RoutstrFee(SQLModel, table=True):  # type: ignore
+    __tablename__ = "routstr_fees"
+    id: int = Field(default=1, primary_key=True)
+    accumulated_msats: int = Field(default=0)
+    total_paid_msats: int = Field(default=0)
+    last_paid_at: int | None = Field(default=None)
+
+
+async def accumulate_routstr_fee(session: AsyncSession, amount_msats: int) -> None:
+    stmt = (
+        update(RoutstrFee)
+        .where(col(RoutstrFee.id) == 1)
+        .values(accumulated_msats=RoutstrFee.accumulated_msats + amount_msats)
+    )
+    result = await session.exec(stmt)  # type: ignore[call-overload]
+    if result.rowcount == 0:
+        session.add(RoutstrFee(id=1, accumulated_msats=amount_msats))
+    await session.commit()
+
+
+async def get_routstr_fee(session: AsyncSession) -> RoutstrFee:
+    fee = await session.get(RoutstrFee, 1)
+    if fee is None:
+        fee = RoutstrFee(id=1, accumulated_msats=0, total_paid_msats=0)
+        session.add(fee)
+        await session.commit()
+        await session.refresh(fee)
+    return fee
+
+
+async def reset_routstr_fee(session: AsyncSession, paid_msats: int) -> None:
+    stmt = (
+        update(RoutstrFee)
+        .where(col(RoutstrFee.id) == 1)
+        .values(
+            accumulated_msats=RoutstrFee.accumulated_msats - paid_msats,
+            total_paid_msats=RoutstrFee.total_paid_msats + paid_msats,
+            last_paid_at=int(time.time()),
+        )
+    )
+    await session.exec(stmt)  # type: ignore[call-overload]
+    await session.commit()
+
+
 async def balances_for_mint_and_unit(
     db_session: AsyncSession, mint_url: str, unit: str
 ) -> int:

routstr/core/main.py

@@ -22,7 +22,7 @@ from ..payment.models import models_router, update_sats_pricing
 from ..payment.price import update_prices_periodically
 from ..proxy import initialize_upstreams, proxy_router, refresh_model_maps_periodically
 from ..upstream.auto_topup import periodic_auto_topup
-from ..wallet import periodic_payout, periodic_refund_sweep
+from ..wallet import periodic_payout, periodic_refund_sweep, periodic_routstr_fee_payout
 from .admin import admin_router
 from .db import create_session, init_db, run_migrations
 from .exceptions import general_exception_handler, http_exception_handler
@@ -36,9 +36,9 @@ setup_logging()
 logger = get_logger(__name__)
 
 if os.getenv("VERSION_SUFFIX") is not None:
-    __version__ = f"0.4.1-{os.getenv('VERSION_SUFFIX')}"
+    __version__ = f"0.4.3-{os.getenv('VERSION_SUFFIX')}"
 else:
-    __version__ = "0.4.1"
+    __version__ = "0.4.3"
 
 
 @asynccontextmanager
@@ -56,6 +56,7 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
     key_reset_task = None
     auto_topup_task = None
     refund_sweep_task = None
+    routstr_fee_task = None
 
     try:
         # Run database migrations on startup
@@ -102,8 +103,11 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
         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:
+            # Pass the accessor (not its current value) so the loop sees providers
+            # added/changed via reinitialize_upstreams() instead of staying pinned
+            # to the startup snapshot.
             models_refresh_task = asyncio.create_task(
-                refresh_upstreams_models_periodically(get_upstreams())
+                refresh_upstreams_models_periodically(get_upstreams)
             )
         model_maps_refresh_task = asyncio.create_task(refresh_model_maps_periodically())
         payout_task = asyncio.create_task(periodic_payout())
@@ -115,6 +119,7 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
         key_reset_task = asyncio.create_task(periodic_key_reset())
         auto_topup_task = asyncio.create_task(periodic_auto_topup())
         refund_sweep_task = asyncio.create_task(periodic_refund_sweep())
+        routstr_fee_task = asyncio.create_task(periodic_routstr_fee_payout())
 
         yield
 
@@ -152,6 +157,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
             auto_topup_task.cancel()
         if refund_sweep_task is not None:
             refund_sweep_task.cancel()
+        if routstr_fee_task is not None:
+            routstr_fee_task.cancel()
 
         try:
             tasks_to_wait = []
@@ -177,6 +184,8 @@ async def lifespan(_: FastAPI) -> AsyncGenerator[None, None]:
                 tasks_to_wait.append(auto_topup_task)
             if refund_sweep_task is not None:
                 tasks_to_wait.append(refund_sweep_task)
+            if routstr_fee_task is not None:
+                tasks_to_wait.append(routstr_fee_task)
 
             if tasks_to_wait:
                 await asyncio.gather(*tasks_to_wait, return_exceptions=True)

routstr/upstream/helpers.py

@@ -3,7 +3,7 @@ from __future__ import annotations
 import asyncio
 import os
 import re
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Callable
 
 if TYPE_CHECKING:
     from ..core.settings import Settings
@@ -122,12 +122,16 @@ async def get_all_models_with_overrides(
 
 
 async def refresh_upstreams_models_periodically(
-    upstreams: list[BaseUpstreamProvider],
+    upstreams_provider: (
+        Callable[[], list[BaseUpstreamProvider]] | list[BaseUpstreamProvider]
+    ),
 ) -> None:
     """Background task to periodically refresh models cache for all providers.
 
     Args:
-        upstreams: List of upstream provider instances
+        upstreams_provider: Either a callable returning the live upstream list
+            (preferred — picks up providers added/changed via reinitialize_upstreams),
+            or a static list (legacy, will go stale after reinitialize_upstreams).
     """
     import asyncio
     import random
@@ -139,9 +143,14 @@ async def refresh_upstreams_models_periodically(
         logger.info("Provider models refresh disabled (interval <= 0)")
         return
 
+    def _resolve_upstreams() -> list[BaseUpstreamProvider]:
+        if callable(upstreams_provider):
+            return upstreams_provider()
+        return upstreams_provider
+
     while True:
         try:
-            for upstream in upstreams:
+            for upstream in _resolve_upstreams():
                 try:
                     await upstream.refresh_models_cache()
                 except Exception as e:

routstr/wallet.py

@@ -589,6 +589,46 @@ async def periodic_refund_sweep() -> None:
             )
 
 
+async def periodic_routstr_fee_payout() -> None:
+    from .auth import (
+        ROUTSTR_FEE_DEFAULT_PAYOUT,
+        ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS,
+        ROUTSTR_LN_ADDRESS,
+    )
+
+    if not ROUTSTR_LN_ADDRESS:
+        logger.info("ROUTSTR_LN_ADDRESS not set, skipping fee payout")
+        return
+    while True:
+        await asyncio.sleep(ROUTSTR_FEE_PAYOUT_INTERVAL_SECONDS)
+        try:
+            async with db.create_session() as session:
+                fee = await db.get_routstr_fee(session)
+                accumulated_sats = fee.accumulated_msats // 1000
+                if accumulated_sats >= ROUTSTR_FEE_DEFAULT_PAYOUT:
+                    wallet = await get_wallet(settings.primary_mint, "sat")
+                    proofs = get_proofs_per_mint_and_unit(
+                        wallet, settings.primary_mint, "sat", not_reserved=True
+                    )
+                    amount_received = await raw_send_to_lnurl(
+                        wallet, proofs, ROUTSTR_LN_ADDRESS, "sat", amount=accumulated_sats
+                    )
+                    paid_msats = accumulated_sats * 1000
+                    await db.reset_routstr_fee(session, paid_msats)
+                    logger.info(
+                        "Routstr fee payout sent",
+                        extra={
+                            "accumulated_sats": accumulated_sats,
+                            "amount_received": amount_received,
+                        },
+                    )
+        except Exception as e:
+            logger.error(
+                f"Error in Routstr fee payout: {type(e).__name__}",
+                extra={"error": str(e)},
+            )
+
+
 async def send_to_lnurl(amount: int, unit: str, mint: str, address: str) -> int:
     wallet = await get_wallet(mint, unit)
     proofs = wallet._get_proofs_per_keyset(wallet.proofs)[wallet.keyset_id]

tests/integration/test_wallet_refund.py

@@ -356,6 +356,70 @@ async def test_concurrent_refund_requests(
     assert len(successful) + len(failed) == 5
 
 
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_refund_rejects_concurrent_topup_on_same_key(
+    authenticated_client: AsyncClient,
+    testmint_wallet: Any,
+) -> None:
+    """Test refund returns 409 when a concurrent topup changes the balance first."""
+    from routstr import balance as balance_module
+
+    wallet_response = await authenticated_client.get("/v1/wallet/")
+    assert wallet_response.status_code == 200
+    initial_balance = wallet_response.json()["balance"]
+
+    topup_amount_sat = 500
+    topup_token = await testmint_wallet.mint_tokens(topup_amount_sat)
+
+    validate_called = asyncio.Event()
+    allow_refund_to_continue = asyncio.Event()
+    original_validate_bearer_key = balance_module.validate_bearer_key
+    delayed_once = False
+
+    async def delayed_validate_bearer_key(*args: Any, **kwargs: Any) -> ApiKey:
+        nonlocal delayed_once
+        key = await original_validate_bearer_key(*args, **kwargs)
+        if not delayed_once:
+            delayed_once = True
+            validate_called.set()
+            await allow_refund_to_continue.wait()
+        return key
+
+    async def issue_refund() -> Any:
+        return await authenticated_client.post("/v1/wallet/refund")
+
+    async def issue_topup() -> Any:
+        await validate_called.wait()
+        try:
+            return await authenticated_client.post(
+                "/v1/wallet/topup", params={"cashu_token": topup_token}
+            )
+        finally:
+            allow_refund_to_continue.set()
+
+    with patch(
+        "routstr.balance.validate_bearer_key", new=delayed_validate_bearer_key
+    ):
+        refund_response, topup_response = await asyncio.gather(
+            issue_refund(), issue_topup()
+        )
+
+    assert topup_response.status_code == 200
+    assert topup_response.json()["msats"] == topup_amount_sat * 1000
+    assert refund_response.status_code == 409
+    assert (
+        refund_response.json()["detail"]
+        == "Balance changed concurrently. Please retry the refund."
+    )
+
+    final_balance_response = await authenticated_client.get("/v1/wallet/")
+    assert final_balance_response.status_code == 200
+    assert final_balance_response.json()["balance"] == (
+        initial_balance + topup_amount_sat * 1000
+    )
+
+
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_refund_during_active_usage(

tests/unit/test_models_refresh_loop.py

@@ -0,0 +1,117 @@
+"""Regression tests for the periodic upstream models refresh loop."""
+
+from __future__ import annotations
+
+import asyncio
+import os
+from typing import cast
+from unittest.mock import AsyncMock
+
+import pytest
+
+os.environ.setdefault("UPSTREAM_BASE_URL", "http://test")
+os.environ.setdefault("UPSTREAM_API_KEY", "test")
+
+from routstr.upstream.base import BaseUpstreamProvider  # noqa: E402
+
+
+class _FakeUpstream:
+    """Minimal stand-in for BaseUpstreamProvider used by the refresh loop.
+
+    Only ``base_url`` (for error logging) and ``refresh_models_cache`` (the call
+    under test) are exercised; everything else stays unused.
+    """
+
+    def __init__(self, name: str) -> None:
+        self.base_url = f"http://{name}"
+        self.refresh_models_cache = AsyncMock()
+
+
+def _make_fake_upstream(name: str) -> BaseUpstreamProvider:
+    # The loop only uses duck-typed attributes — cast keeps the test type-clean
+    # without dragging in BaseUpstreamProvider's full constructor.
+    return cast(BaseUpstreamProvider, _FakeUpstream(name))
+
+
+@pytest.mark.asyncio
+async def test_refresh_loop_picks_up_providers_added_after_startup(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """If a provider is added after the loop starts (e.g. via reinitialize_upstreams),
+    the next loop iteration must refresh it. Previously the loop captured the upstream
+    list at startup and missed any later additions."""
+    from routstr.core.settings import settings as global_settings
+    from routstr.upstream.helpers import refresh_upstreams_models_periodically
+
+    # Tight interval so the test finishes quickly.
+    monkeypatch.setattr(
+        global_settings, "models_refresh_interval_seconds", 1, raising=False
+    )
+
+    initial_upstream = _make_fake_upstream("initial")
+    live_list: list[BaseUpstreamProvider] = [initial_upstream]
+
+    # Stub out the post-iteration sats-pricing refresh so the loop body has no DB deps.
+    async def _noop_pricing_refresh() -> None:  # pragma: no cover - trivial stub
+        return None
+
+    monkeypatch.setattr(
+        "routstr.payment.models._update_sats_pricing_once",
+        _noop_pricing_refresh,
+    )
+
+    task = asyncio.create_task(
+        refresh_upstreams_models_periodically(lambda: live_list)
+    )
+
+    try:
+        # Wait for the first iteration to refresh the initial upstream.
+        for _ in range(40):
+            if initial_upstream.refresh_models_cache.await_count >= 1:  # type: ignore[attr-defined]
+                break
+            await asyncio.sleep(0.05)
+        assert initial_upstream.refresh_models_cache.await_count >= 1, (  # type: ignore[attr-defined]
+            "loop did not refresh the initial upstream within the timeout"
+        )
+
+        # Simulate reinitialize_upstreams: replace the live list contents with new
+        # provider instances. The loop must observe the swap on its next tick.
+        new_upstream = _make_fake_upstream("added-after-startup")
+        live_list[:] = [new_upstream]
+
+        for _ in range(60):
+            if new_upstream.refresh_models_cache.await_count >= 1:  # type: ignore[attr-defined]
+                break
+            await asyncio.sleep(0.05)
+
+        assert new_upstream.refresh_models_cache.await_count >= 1, (  # type: ignore[attr-defined]
+            "loop did not refresh the upstream added after startup — "
+            "regression: list snapshot captured at startup"
+        )
+    finally:
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            pass
+
+
+@pytest.mark.asyncio
+async def test_refresh_loop_disabled_when_interval_non_positive(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from routstr.core.settings import settings as global_settings
+    from routstr.upstream.helpers import refresh_upstreams_models_periodically
+
+    monkeypatch.setattr(
+        global_settings, "models_refresh_interval_seconds", 0, raising=False
+    )
+
+    upstream = _make_fake_upstream("never-refreshed")
+
+    # Loop must return immediately without ever touching the upstream.
+    await asyncio.wait_for(
+        refresh_upstreams_models_periodically(lambda: [upstream]),
+        timeout=1.0,
+    )
+    upstream.refresh_models_cache.assert_not_awaited()  # type: ignore[attr-defined]

ui/app/transactions/page.tsx

@@ -339,7 +339,6 @@ export default function TransactionsPage() {
     setApikeyPage(0);
   }, [type, status, search]);
 
-  const activeQuery = activeTab === 'x-cashu' ? xcashuQuery : apikeyQuery;
   const isRefetching = xcashuQuery.isRefetching || apikeyQuery.isRefetching;
 
   const renderCardContent = (

ui/components/add-provider-model-dialog.tsx

@@ -540,7 +540,7 @@ export function AddProviderModelDialog({
                   };
                   return (
                     <FormItem>
-                      <FormLabel>Upstream Model ID</FormLabel>
+                      <FormLabel>Client Alias ID</FormLabel>
                       <FormControl>
                         <div className='flex gap-2'>
                           <Input
@@ -565,8 +565,8 @@ export function AddProviderModelDialog({
                         </div>
                       </FormControl>
                       <FormDescription>
-                        Model ID sent to the upstream provider. Defaults to the
-                        model&apos;s own ID.
+                        Alternate ID that clients can use to reference this
+                        model. Defaults to the model&apos;s own ID.
                       </FormDescription>
                       <FormMessage />
                     </FormItem>

uv.lock

@@ -1878,7 +1878,7 @@ wheels = [
 
 [[package]]
 name = "routstr"
-version = "0.4.1"
+version = "0.4.3"
 source = { editable = "." }
 dependencies = [
     { name = "aiosqlite" },