This commit is contained in:
Shroominic
2025-08-01 12:51:31 -03:00
parent 27c5721cd8
commit 848af6992a
9 changed files with 168 additions and 605 deletions
+2 -2
View File
@@ -5,8 +5,8 @@ from fastapi import APIRouter, Request
from fastapi.responses import HTMLResponse
from sqlmodel import select
from .cashu import wallet
from .db import ApiKey, create_session
from .wallet import get_wallet_balance
admin_router = APIRouter(prefix="/admin")
@@ -112,7 +112,7 @@ async def dashboard(request: Request) -> str:
# avoid rounding issues.
total_user_balance = sum(key.balance for key in api_keys) // 1000
# Fetch balance from cashu
current_balance = await wallet().get_balance()
current_balance = await get_wallet_balance()
owner_balance = current_balance - total_user_balance
return f"""<!DOCTYPE html>
+2 -2
View File
@@ -4,9 +4,8 @@ from typing import Optional
from fastapi import HTTPException
from sqlmodel import col, update
from .cashu import credit_balance
from .db import ApiKey, AsyncSession
from .logging.logging_config import get_logger
from .logging import get_logger
from .payment.cost_caculation import (
CostData,
CostDataError,
@@ -14,6 +13,7 @@ from .payment.cost_caculation import (
calculate_cost,
)
from .payment.helpers import get_max_cost_for_model
from .wallet import credit_balance
logger = get_logger(__name__)
-536
View File
@@ -1,536 +0,0 @@
import asyncio
import os
import time
from typing import cast
from sixty_nuts import Wallet
from sixty_nuts.types import CurrencyUnit
from sqlmodel import col, func, select, update
from .db import ApiKey, AsyncSession, get_session
from .logging.logging_config import get_logger
logger = get_logger(__name__)
RECEIVE_LN_ADDRESS = os.environ["RECEIVE_LN_ADDRESS"]
MINT = os.environ.get("MINT", "https://mint.minibits.cash/Bitcoin")
MINIMUM_PAYOUT = int(os.environ.get("MINIMUM_PAYOUT", 100))
REFUND_PROCESSING_INTERVAL = int(os.environ.get("REFUND_PROCESSING_INTERVAL", 3600))
PAYOUT_INTERVAL = int(os.environ.get("PAYOUT_INTERVAL", 300)) # Default 5 minutes
DEV_LN_ADDRESS = "routstr@minibits.cash"
DEVS_DONATION_RATE = float(os.environ.get("DEVS_DONATION_RATE", 0.021)) # 2.1%
NSEC = os.environ["NSEC"] # Nostr private key for the wallet
logger.info(
"Cashu module initialized",
extra={
"mint": MINT,
"minimum_payout": MINIMUM_PAYOUT,
"refund_processing_interval": REFUND_PROCESSING_INTERVAL,
"payout_interval": PAYOUT_INTERVAL,
"devs_donation_rate": DEVS_DONATION_RATE,
},
)
wallet_instance: Wallet | None = None
async def init_wallet() -> None:
"""Initialize the Cashu wallet."""
global wallet_instance
try:
logger.info("Initializing Cashu wallet", extra={"mint": MINT})
wallet_instance = await Wallet.create(nsec=NSEC)
logger.info("Cashu wallet initialized successfully")
except Exception as e:
logger.error(
"Failed to initialize Cashu wallet",
extra={"error": str(e), "error_type": type(e).__name__, "mint": MINT},
)
raise
def wallet() -> Wallet:
"""Get the wallet instance."""
global wallet_instance
if wallet_instance is None:
logger.error("Wallet not initialized - call init_wallet() first")
raise ValueError("Wallet not initialized")
return wallet_instance
async def delete_key_if_zero_balance(key: ApiKey, session: AsyncSession) -> None:
"""Delete the given API key if its balance is zero."""
if key.balance == 0:
logger.info(
"Deleting API key with zero balance",
extra={"key_hash": key.hashed_key[:8] + "...", "balance": key.balance},
)
await session.delete(key)
await session.commit()
async def pay_out() -> None:
"""
Calculates the pay-out amount based on the spent balance, profit, and donation rate.
"""
try:
logger.debug("Starting payout process")
from .db import create_session
async with create_session() as session:
result = await session.exec(
select(func.sum(col(ApiKey.balance))).where(ApiKey.balance > 0)
)
balance = result.one_or_none()
if not balance:
logger.debug("No balance to pay out")
return
user_balance_sats = balance // 1000
wallet_balance_sats = await wallet().get_balance()
logger.debug(
"Payout calculation",
extra={
"user_balance_sats": user_balance_sats,
"wallet_balance_sats": wallet_balance_sats,
},
)
# Handle edge cases more gracefully
if wallet_balance_sats < user_balance_sats:
logger.warning(
"Insufficient wallet balance for payout",
extra={
"wallet_balance_sats": wallet_balance_sats,
"user_balance_sats": user_balance_sats,
"shortfall_sats": user_balance_sats - wallet_balance_sats,
},
)
return
if (revenue := wallet_balance_sats - user_balance_sats) <= MINIMUM_PAYOUT:
logger.debug(
"Revenue below minimum payout threshold",
extra={"revenue_sats": revenue, "minimum_payout": MINIMUM_PAYOUT},
)
return
devs_donation = int(revenue * DEVS_DONATION_RATE)
owners_draw = revenue - devs_donation
logger.info(
"Processing payout",
extra={
"revenue_sats": revenue,
"devs_donation_sats": devs_donation,
"owners_draw_sats": owners_draw,
"donation_rate": DEVS_DONATION_RATE,
},
)
# Send payouts
try:
await wallet().send_to_lnurl(RECEIVE_LN_ADDRESS, owners_draw)
logger.info(
"Owner payout sent successfully",
extra={
"amount_sats": owners_draw,
"address": RECEIVE_LN_ADDRESS[:10] + "...",
},
)
await wallet().send_to_lnurl(DEV_LN_ADDRESS, devs_donation)
logger.info(
"Developer donation sent successfully",
extra={"amount_sats": devs_donation, "address": DEV_LN_ADDRESS},
)
except Exception as payout_error:
logger.error(
"Failed to send payouts",
extra={
"error": str(payout_error),
"error_type": type(payout_error).__name__,
"owners_draw_sats": owners_draw,
"devs_donation_sats": devs_donation,
},
)
raise
except Exception as e:
logger.error(
"Error in payout process",
extra={"error": str(e), "error_type": type(e).__name__},
)
# Periodic payout task
async def periodic_payout() -> None:
"""Periodically process payouts."""
logger.info("Starting periodic payout task", extra={"interval_seconds": 300})
while True:
try:
await asyncio.sleep(300) # Run every 5 minutes
await pay_out()
except asyncio.CancelledError:
logger.info("Periodic payout task cancelled")
break
except Exception as e:
logger.error(
"Error in periodic payout",
extra={"error": str(e), "error_type": type(e).__name__},
)
# Continue running even if payout fails
async def credit_balance(cashu_token: str, key: ApiKey, session: AsyncSession) -> int:
"""Redeem a Cashu token and credit the amount to the API key balance."""
logger.debug(
"Starting token redemption",
extra={
"key_hash": key.hashed_key[:8] + "...",
"token_preview": cashu_token[:20] + "..."
if len(cashu_token) > 20
else cashu_token,
},
)
try:
amount, unit = await wallet().redeem(cashu_token)
logger.info(
"Token redeemed successfully",
extra={
"amount": amount,
"unit": unit,
"key_hash": key.hashed_key[:8] + "...",
},
)
except Exception as e:
logger.error(
"Token redemption failed",
extra={
"error": str(e),
"error_type": type(e).__name__,
"key_hash": key.hashed_key[:8] + "...",
"token_preview": cashu_token[:20] + "..."
if len(cashu_token) > 20
else cashu_token,
},
)
return 0
if amount <= 0:
logger.warning(
"Zero or negative amount redeemed",
extra={
"amount": amount,
"unit": unit,
"key_hash": key.hashed_key[:8] + "...",
},
)
return 0
if unit == "msat":
amount_msats = amount
else:
amount_msats = amount * 1000
logger.debug(
"Crediting balance",
extra={
"amount_msats": amount_msats,
"original_amount": amount,
"unit": unit,
"key_hash": key.hashed_key[:8] + "...",
},
)
# Apply the balance change atomically to avoid race conditions when topping
# up the same key concurrently.
stmt = (
update(ApiKey)
.where(col(ApiKey.hashed_key) == key.hashed_key)
.values(balance=col(ApiKey.balance) + amount_msats)
)
await session.exec(stmt) # type: ignore[call-overload]
await session.commit()
await session.refresh(key)
logger.info(
"Balance credited successfully",
extra={
"credited_msats": amount_msats,
"new_balance_msats": key.balance,
"key_hash": key.hashed_key[:8] + "...",
},
)
return amount_msats
async def check_for_refunds() -> None:
"""
Periodically checks for API keys that are eligible for refunds and processes them.
Raises:
Exception: If an error occurs during the refund check process.
"""
# Setting REFUND_PROCESSING_INTERVAL to 0 disables it
if REFUND_PROCESSING_INTERVAL == 0:
logger.info("Automatic refund processing is disabled")
return
logger.info(
"Starting refund monitoring task",
extra={"interval_seconds": REFUND_PROCESSING_INTERVAL},
)
while True:
try:
logger.debug("Checking for expired keys requiring refunds")
async for session in get_session():
result = await session.exec(select(ApiKey))
keys = result.all()
current_time = int(time.time())
expired_keys = []
for key in keys:
if (
key.balance > 0
and key.refund_address
and key.key_expiry_time
and key.key_expiry_time < current_time
):
expired_keys.append(key)
if expired_keys:
logger.info(
"Found expired keys for refund",
extra={
"expired_count": len(expired_keys),
"current_time": current_time,
},
)
for key in expired_keys:
logger.info(
"Processing refund for expired key",
extra={
"key_hash": key.hashed_key[:8] + "...",
"balance_msats": key.balance,
"expiry_time": key.key_expiry_time,
"current_time": current_time,
"expired_seconds": current_time
- (key.key_expiry_time or 0),
},
)
try:
await refund_balance(key.balance, key, session)
await delete_key_if_zero_balance(key, session)
logger.info(
"Refund processed successfully",
extra={"key_hash": key.hashed_key[:8] + "..."},
)
except Exception as refund_error:
logger.error(
"Failed to process refund",
extra={
"error": str(refund_error),
"error_type": type(refund_error).__name__,
"key_hash": key.hashed_key[:8] + "...",
"balance_msats": key.balance,
},
)
# Sleep for the specified interval before checking again
await asyncio.sleep(REFUND_PROCESSING_INTERVAL)
except asyncio.CancelledError:
logger.info("Refund monitoring task cancelled")
break
except Exception as e:
logger.error(
"Error during refund check",
extra={"error": str(e), "error_type": type(e).__name__},
)
async def refund_balance(amount_msats: int, key: ApiKey, session: AsyncSession) -> int:
"""Process a refund for an API key."""
if amount_msats <= 0:
amount_msats = key.balance
logger.info(
"Processing balance refund",
extra={
"amount_msats": amount_msats,
"key_hash": key.hashed_key[:8] + "...",
"refund_address": key.refund_address[:20] + "..."
if key.refund_address and len(key.refund_address) > 20
else key.refund_address,
},
)
# Convert msats to sats for cashu wallet
amount_sats = amount_msats // 1000
if amount_sats == 0:
logger.error(
"Amount too small to refund",
extra={
"amount_msats": amount_msats,
"amount_sats": amount_sats,
"key_hash": key.hashed_key[:8] + "...",
},
)
raise ValueError("Amount too small to refund (less than 1 sat)")
# Atomically deduct the balance to avoid race conditions when multiple
# refunds are triggered concurrently.
stmt = (
update(ApiKey)
.where(col(ApiKey.hashed_key) == key.hashed_key)
.where(col(ApiKey.balance) >= amount_msats)
.values(balance=col(ApiKey.balance) - amount_msats)
)
result = await session.exec(stmt) # type: ignore[call-overload]
await session.commit()
if result.rowcount == 0:
logger.error(
"Insufficient balance for refund",
extra={
"requested_msats": amount_msats,
"key_hash": key.hashed_key[:8] + "...",
"current_balance": key.balance,
},
)
raise ValueError("Insufficient balance.")
await session.refresh(key)
await delete_key_if_zero_balance(key, session)
if key.refund_address is None:
logger.error(
"Refund address not set", extra={"key_hash": key.hashed_key[:8] + "..."}
)
raise ValueError("Refund address not set.")
try:
result = await wallet().send_to_lnurl(key.refund_address, amount=amount_sats)
logger.info(
"Refund sent successfully",
extra={
"amount_sats": amount_sats,
"refund_address": key.refund_address[:20] + "..."
if len(key.refund_address) > 20
else key.refund_address,
"key_hash": key.hashed_key[:8] + "...",
"transaction_result": str(result),
},
)
return result
except Exception as e:
logger.error(
"Failed to send refund",
extra={
"error": str(e),
"error_type": type(e).__name__,
"amount_sats": amount_sats,
"refund_address": key.refund_address,
"key_hash": key.hashed_key[:8] + "...",
},
)
raise
async def x_cashu_refund(key: ApiKey, session: AsyncSession, unit: CurrencyUnit) -> str:
"""Process an X-Cashu refund token."""
logger.info(
"Processing X-Cashu refund",
extra={
"balance_msats": key.balance,
"unit": unit,
"key_hash": key.hashed_key[:8] + "...",
},
)
try:
refund_token = await wallet().send(key.balance, unit=unit)
logger.info(
"X-Cashu refund token created",
extra={
"amount": key.balance,
"unit": unit,
"key_hash": key.hashed_key[:8] + "...",
"token_preview": refund_token[:20] + "..."
if len(refund_token) > 20
else refund_token,
},
)
await session.delete(key)
await session.commit()
logger.info(
"X-Cashu refund completed", extra={"key_hash": key.hashed_key[:8] + "..."}
)
return refund_token
except Exception as e:
logger.error(
"Failed to create X-Cashu refund",
extra={
"error": str(e),
"error_type": type(e).__name__,
"balance": key.balance,
"unit": unit,
"key_hash": key.hashed_key[:8] + "...",
},
)
raise
async def redeem(cashu_token: str, lnurl: str) -> int:
"""Redeem a Cashu token and send to LNURL."""
logger.info(
"Starting token redemption for LNURL",
extra={
"token_preview": cashu_token[:20] + "..."
if len(cashu_token) > 20
else cashu_token,
"lnurl_preview": lnurl[:20] + "..." if len(lnurl) > 20 else lnurl,
},
)
try:
amount, unit = await wallet().redeem(cashu_token)
logger.info("Token redeemed for LNURL", extra={"amount": amount, "unit": unit})
unit = cast(CurrencyUnit, unit)
result = await wallet().send_to_lnurl(lnurl, amount=amount, unit=unit)
logger.info(
"Successfully sent to LNURL",
extra={
"amount": amount,
"unit": unit,
"lnurl_preview": lnurl[:20] + "..." if len(lnurl) > 20 else lnurl,
"transaction_result": str(result),
},
)
return amount
except Exception as e:
logger.error(
"Failed to redeem and send to LNURL",
extra={
"error": str(e),
"error_type": type(e).__name__,
"token_preview": cashu_token[:20] + "..."
if len(cashu_token) > 20
else cashu_token,
"lnurl_preview": lnurl[:20] + "..." if len(lnurl) > 20 else lnurl,
},
)
raise
+2 -2
View File
@@ -8,12 +8,12 @@ from fastapi.middleware.cors import CORSMiddleware
from .account import wallet_router
from .admin import admin_router
from .cashu import check_for_refunds, init_wallet, periodic_payout
from .db import init_db
from .discovery import providers_router
from .logging.logging_config import get_logger, setup_logging
from .logging import get_logger, setup_logging
from .models import MODELS, models_router, update_sats_pricing
from .proxy import proxy_router
from .wallet import check_for_refunds, init_wallet, periodic_payout
# Initialize logging first
setup_logging()
+7 -52
View File
@@ -1,21 +1,20 @@
import json
import traceback
from typing import AsyncGenerator, Literal, cast
from typing import AsyncGenerator
import httpx
from fastapi import BackgroundTasks, HTTPException, Request
from fastapi.responses import Response, StreamingResponse
from sixty_nuts.types import CurrencyUnit
from router.cashu import wallet
from router.logging.logging_config import get_logger
from router.payment.cost_caculation import (
from ..logging import get_logger
from ..wallet import CurrencyUnit, create_refund_token, redeem_token
from .cost_caculation import (
CostData,
CostDataError,
MaxCostData,
calculate_cost,
)
from router.payment.helpers import (
from .helpers import (
UPSTREAM_BASE_URL,
create_error_response,
get_max_cost_for_model,
@@ -454,7 +453,7 @@ async def handle_non_streaming_response(
# Emergency refund with small deduction for processing
emergency_refund = amount
refund_token = await wallet().send(emergency_refund)
refund_token = await create_refund_token(emergency_refund, unit=unit)
response.headers["X-Cashu"] = refund_token
logger.warning(
@@ -528,50 +527,6 @@ async def get_cost(response_data: dict) -> MaxCostData | CostData | None:
)
async def redeem_token(x_cashu_token: str) -> tuple[int, Literal["sat", "msat"]]:
"""Redeem X-Cashu token and return amount and unit."""
logger.debug(
"Redeeming X-Cashu token",
extra={
"token_preview": x_cashu_token[:20] + "..."
if len(x_cashu_token) > 20
else x_cashu_token
},
)
try:
result = await wallet().redeem(x_cashu_token)
amount, unit = cast(tuple[int, Literal["sat", "msat"]], result)
logger.info(
"X-Cashu token redeemed successfully",
extra={"amount": amount, "unit": unit},
)
return amount, unit
except Exception as e:
logger.error(
"X-Cashu token redemption failed",
extra={
"error": str(e),
"error_type": type(e).__name__,
"token_preview": x_cashu_token[:20] + "..."
if len(x_cashu_token) > 20
else x_cashu_token,
},
)
raise HTTPException(
status_code=401,
detail={
"error": {
"message": f"Invalid or expired Cashu key: {str(e)}",
"type": "invalid_request_error",
"code": "invalid_api_key",
}
},
)
async def send_refund(amount: int, unit: CurrencyUnit, mint: str | None = None) -> str:
"""Send a refund using Cashu tokens."""
logger.debug(
@@ -583,7 +538,7 @@ async def send_refund(amount: int, unit: CurrencyUnit, mint: str | None = None)
for attempt in range(max_retries):
try:
refund_token = await wallet().send(amount, unit=unit, mint_url=mint)
refund_token = await create_refund_token(amount, unit=unit, mint_url=mint)
logger.info(
"Refund token created successfully",
+1 -1
View File
@@ -3,7 +3,7 @@ import os
import httpx
from .logging.logging_config import get_logger
from .logging import get_logger
logger = get_logger(__name__)
+8 -9
View File
@@ -7,15 +7,6 @@ import httpx
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
from fastapi.responses import Response, StreamingResponse
from router.logging.logging_config import get_logger
from router.payment.helpers import (
UPSTREAM_BASE_URL,
check_token_balance,
create_error_response,
prepare_upstream_headers,
)
from router.payment.x_cashu import x_cashu_handler
from .auth import (
adjust_payment_for_tokens,
pay_for_request,
@@ -23,6 +14,14 @@ from .auth import (
validate_bearer_key,
)
from .db import ApiKey, AsyncSession, create_session, get_session
from .logging import get_logger
from .payment.helpers import (
UPSTREAM_BASE_URL,
check_token_balance,
create_error_response,
prepare_upstream_headers,
)
from .payment.x_cashu import x_cashu_handler
logger = get_logger(__name__)
proxy_router = APIRouter()
+144
View File
@@ -0,0 +1,144 @@
import os
from typing import Literal
from cashu.core.base import Token
from cashu.wallet.helpers import deserialize_token_from_string, redeem_universal
from cashu.wallet.wallet import Wallet
from .db import DATABASE_URL, ApiKey, AsyncSession
from .logging import get_logger
# from .cashu import (
# credit_balance,
# delete_key_if_zero_balance,
# refund_balance,
# wallet,
# )
# from .cashu import (
# check_for_refunds,
# init_wallet,
# periodic_payout,
# )
logger = get_logger(__name__)
RECEIVE_LN_ADDRESS = os.environ["RECEIVE_LN_ADDRESS"]
MINT = os.environ.get("MINT", "https://mint.minibits.cash/Bitcoin")
MINIMUM_PAYOUT = int(os.environ.get("MINIMUM_PAYOUT", 100))
REFUND_PROCESSING_INTERVAL = int(os.environ.get("REFUND_PROCESSING_INTERVAL", 3600))
PAYOUT_INTERVAL = int(os.environ.get("PAYOUT_INTERVAL", 300)) # Default 5 minutes
DEV_LN_ADDRESS = "routstr@minibits.cash"
DEVS_DONATION_RATE = float(os.environ.get("DEVS_DONATION_RATE", 0.021)) # 2.1%
NSEC = os.environ["NSEC"] # Nostr private key for the wallet
logger.info(
"Cashu module initialized",
extra={
"mint": MINT,
"minimum_payout": MINIMUM_PAYOUT,
"refund_processing_interval": REFUND_PROCESSING_INTERVAL,
"payout_interval": PAYOUT_INTERVAL,
"devs_donation_rate": DEVS_DONATION_RATE,
},
)
CurrencyUnit = Literal["sat", "msat"]
async def get_balance(unit: CurrencyUnit) -> int:
raise NotImplementedError
async def recieve_token(
token: str,
) -> tuple[int, CurrencyUnit, str]: # amount, unit, mint_url
raise NotImplementedError
async def send_token(
amount: int, unit: CurrencyUnit, mint_url: str | None = None
) -> str:
raise NotImplementedError
# insert initial token state here to reduce db calls
async def create_refund_token(
amount: int, unit: CurrencyUnit, mint_url: str | None = None
) -> str:
wallet = await Wallet.with_db(
mint_url, DATABASE_URL, load_all_keysets=True, unit=unit
)
if wallet.balance_per_minturl(unit=unit)[mint_url] < amount:
raise ValueError("Wallet has no balance")
if mint_url is None:
mint_url = wallet.mint_urls[0]
return await wallet._make_token(amount, unit=unit, mint_url=mint_url)
async def redeem_token(token: str) -> Token:
token_obj = deserialize_token_from_string(token)
wallet = await Wallet.with_db(
token_obj.mint,
DATABASE_URL,
load_all_keysets=True,
unit=token_obj.unit,
)
return await redeem_universal(wallet, token_obj)
async def delete_key_if_zero_balance(key: str) -> None:
raise NotImplementedError
async def credit_balance(cashu_token: str, key: ApiKey, session: AsyncSession) -> int:
raise NotImplementedError
async def check_for_refunds() -> None:
raise NotImplementedError
async def init_wallet() -> None:
raise NotImplementedError
async def periodic_payout() -> None:
raise NotImplementedError
async def get_wallet_balance() -> int:
raise NotImplementedError
# class Proof:
# """
# Represents an ecash bill
# """
# def redeem_to_proofs(self, token: str) -> list[Proof]:
# raise NotImplementedError
# class Payment:
# """
# Stores all cashu payment related data
# """
# def __init__(self, token: str) -> None:
# self.initial_token = token
# amount, unit, mint_url = self.parse_token(token)
# self.amount = amount
# self.unit = unit
# self.mint_url = mint_url
# self.claimed_proofs = redeem_to_proofs(token)
# def parse_token(self, token: str) -> tuple[int, CurrencyUnit, str]:
# raise NotImplementedError
# def refund_full(self) -> None:
# raise NotImplementedError
# def refund_partial(self, amount: int) -> None:
# raise NotImplementedError
+2 -1
View File
@@ -156,7 +156,8 @@ async def async_client(test_session: AsyncSession) -> AsyncGenerator[AsyncClient
mock_update.return_value = None
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
transport=ASGITransport(app=app), # type: ignore
base_url="http://test",
) as client:
yield client