From ed7f86ce01f2d6b8cd85693ab657a62e85ecfdb2 Mon Sep 17 00:00:00 2001 From: Shroominic Date: Fri, 1 Aug 2025 13:44:57 -0300 Subject: [PATCH] nutshell impl --- router/account.py | 18 +- router/admin.py | 4 +- router/oldcashu.py | 536 ++++++++++++++++++++++++++++++++++++++ router/payment/x_cashu.py | 17 +- router/wallet.py | 114 ++++---- 5 files changed, 605 insertions(+), 84 deletions(-) create mode 100644 router/oldcashu.py diff --git a/router/account.py b/router/account.py index f9b4e6b9..c4389a2a 100644 --- a/router/account.py +++ b/router/account.py @@ -3,13 +3,8 @@ from typing import Annotated, NoReturn from fastapi import APIRouter, Depends, Header, HTTPException from .auth import validate_bearer_key -from .cashu import ( - credit_balance, - delete_key_if_zero_balance, - refund_balance, - wallet, -) from .db import ApiKey, AsyncSession, get_session +from .wallet import credit_balance, send_to_lnurl, send_token wallet_router = APIRouter(prefix="/v1/wallet") @@ -66,8 +61,8 @@ async def refund_wallet_endpoint( # Perform refund operation first, before modifying balance if key.refund_address: - await refund_balance(remaining_balance_msats, key, session) - result = {"recipient": key.refund_address, "msats": remaining_balance_msats} + await send_to_lnurl(remaining_balance_msats, "msat", key.refund_address) + result = {"recipient": key.refund_address, "msat": remaining_balance_msats} else: # Convert msats to sats for cashu wallet remaining_balance_sats = remaining_balance_msats // 1000 @@ -77,15 +72,12 @@ async def refund_wallet_endpoint( ) # TODO: choose currency and mint based on what user has configured - token = await wallet().send(remaining_balance_sats) + token = await send_token(remaining_balance_sats, "sat") result = {"msats": remaining_balance_msats, "recipient": None, "token": token} - # Only after successful refund, zero out the balance - key.balance = 0 - session.add(key) + await session.delete(key) await session.commit() - await delete_key_if_zero_balance(key, session) return result diff --git a/router/admin.py b/router/admin.py index b6717921..bc0255f5 100644 --- a/router/admin.py +++ b/router/admin.py @@ -6,7 +6,7 @@ from fastapi.responses import HTMLResponse from sqlmodel import select from .db import ApiKey, create_session -from .wallet import get_wallet_balance +from .wallet import get_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 get_wallet_balance() + current_balance = await get_balance("sat") owner_balance = current_balance - total_user_balance return f""" diff --git a/router/oldcashu.py b/router/oldcashu.py new file mode 100644 index 00000000..fea13c3d --- /dev/null +++ b/router/oldcashu.py @@ -0,0 +1,536 @@ +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 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 diff --git a/router/payment/x_cashu.py b/router/payment/x_cashu.py index 89ca495c..30ce674c 100644 --- a/router/payment/x_cashu.py +++ b/router/payment/x_cashu.py @@ -7,13 +7,8 @@ from fastapi import BackgroundTasks, HTTPException, Request from fastapi.responses import Response, StreamingResponse from ..logging import get_logger -from ..wallet import CurrencyUnit, create_refund_token, redeem_token -from .cost_caculation import ( - CostData, - CostDataError, - MaxCostData, - calculate_cost, -) +from ..wallet import CurrencyUnit, recieve_token, send_token +from .cost_caculation import CostData, CostDataError, MaxCostData, calculate_cost from .helpers import ( UPSTREAM_BASE_URL, create_error_response, @@ -41,12 +36,12 @@ async def x_cashu_handler( try: headers = dict(request.headers) - amount, unit = await redeem_token(x_cashu_token) + amount, unit, mint = await recieve_token(x_cashu_token) headers = prepare_upstream_headers(dict(request.headers)) logger.info( "X-Cashu token redeemed successfully", - extra={"amount": amount, "unit": unit, "path": path}, + extra={"amount": amount, "unit": unit, "path": path, "mint": mint}, ) return await forward_to_upstream(request, path, headers, amount, unit) @@ -453,7 +448,7 @@ async def handle_non_streaming_response( # Emergency refund with small deduction for processing emergency_refund = amount - refund_token = await create_refund_token(emergency_refund, unit=unit) + refund_token = await send_token(emergency_refund, unit=unit) response.headers["X-Cashu"] = refund_token logger.warning( @@ -538,7 +533,7 @@ async def send_refund(amount: int, unit: CurrencyUnit, mint: str | None = None) for attempt in range(max_retries): try: - refund_token = await create_refund_token(amount, unit=unit, mint_url=mint) + refund_token = await send_token(amount, unit=unit, mint_url=mint) logger.info( "Refund token created successfully", diff --git a/router/wallet.py b/router/wallet.py index 2a4d91aa..92191b3a 100644 --- a/router/wallet.py +++ b/router/wallet.py @@ -1,12 +1,11 @@ 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.core.settings import settings # type: ignore +from cashu.wallet.helpers import deserialize_token_from_string, receive, send from cashu.wallet.wallet import Wallet from .db import DATABASE_URL, ApiKey, AsyncSession -from .logging import get_logger # from .cashu import ( # credit_balance, @@ -20,94 +19,93 @@ from .logging import get_logger # 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"] +PRIMARY_MINT_URL = os.environ["PRIMARY_MINT_URL"] + async def get_balance(unit: CurrencyUnit) -> int: - raise NotImplementedError + wallet = await Wallet.with_db( + PRIMARY_MINT_URL, + # DATABASE_URL, + db=os.path.join(settings.cashu_dir, "temp"), + load_all_keysets=True, + unit=unit, + ) + wallet.load_proofs() + return wallet.available_balance.amount async def recieve_token( token: str, ) -> tuple[int, CurrencyUnit, str]: # amount, unit, mint_url - raise NotImplementedError + # trusted_mints = os.environ["CASHU_MINTS"].split(",") + 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 + ) + await receive(wallet, token_obj) + return token_obj.amount, token_obj.unit, token_obj.mint async def send_token( amount: int, unit: CurrencyUnit, mint_url: str | None = None ) -> str: - raise NotImplementedError + wallet = await Wallet.with_db( + mint_url or PRIMARY_MINT_URL, + DATABASE_URL, + load_all_keysets=True, + unit=unit, + ) + balance, token = await send(wallet, amount=amount) + return token # 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 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 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 credit_balance(cashu_token: str, key: ApiKey, session: AsyncSession) -> int: raise NotImplementedError -async def check_for_refunds() -> None: +async def send_to_lnurl(amount: int, unit: CurrencyUnit, lnurl: str) -> dict[str, int]: raise NotImplementedError +async def check_for_refunds() -> None: + print("check_for_refunds, temp not implemented") + + async def init_wallet() -> None: - raise NotImplementedError + balance = await get_balance("sat") + print(f"init_wallet, balance: {balance}") async def periodic_payout() -> None: - raise NotImplementedError - - -async def get_wallet_balance() -> int: - raise NotImplementedError + print("periodic_payout, temp not implemented") # class Proof: