diff --git a/migrations/versions/d4e5f6a7b8c9_add_routstr_fees_table.py b/migrations/versions/d4e5f6a7b8c9_add_routstr_fees_table.py new file mode 100644 index 00000000..a7e38e6e --- /dev/null +++ b/migrations/versions/d4e5f6a7b8c9_add_routstr_fees_table.py @@ -0,0 +1,32 @@ +"""add routstr_fees table + +Revision ID: d4e5f6a7b8c9 +Revises: c3d4e5f6a7b8 +Create Date: 2026-03-20 00:00:00.000000 +""" + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision = "d4e5f6a7b8c9" +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") diff --git a/routstr/auth.py b/routstr/auth.py index 6009cb1d..df761938 100644 --- a/routstr/auth.py +++ b/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={ diff --git a/routstr/core/db.py b/routstr/core/db.py index 04ba32be..b2d855bb 100644 --- a/routstr/core/db.py +++ b/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: diff --git a/routstr/core/main.py b/routstr/core/main.py index 36c6c128..e959086c 100644 --- a/routstr/core/main.py +++ b/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 @@ -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 @@ -115,6 +116,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 +154,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 +181,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) diff --git a/routstr/wallet.py b/routstr/wallet.py index 9c27ca56..dee6625b 100644 --- a/routstr/wallet.py +++ b/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]