diff --git a/routstr/wallet.py b/routstr/wallet.py index 5b78c8ec..59764caf 100644 --- a/routstr/wallet.py +++ b/routstr/wallet.py @@ -237,7 +237,9 @@ async def send(amount: int, unit: str, mint_url: str | None = None) -> tuple[int all_mint_urls = list({k.mint_url for k in wallet.keysets.values()}) proof_summary = { - f"{k.mint_url}/{k.unit.name}": sum(p.amount for p in wallet.proofs if p.id == k.id) + f"{k.mint_url}/{k.unit.name}": sum( + p.amount for p in wallet.proofs if p.id == k.id + ) for k in wallet.keysets.values() } # Show ALL proofs in DB by keyset_id, regardless of whether the loaded wallet @@ -591,11 +593,16 @@ async def swap_to_primary_mint( # advance the counter so the next request derives fresh secrets. logger.warning( "swap_to_primary_mint: outputs already signed — recovering orphaned proofs", - extra={"mint_quote_id": mint_quote.quote, "minted_amount": minted_amount}, + extra={ + "mint_quote_id": mint_quote.quote, + "minted_amount": minted_amount, + }, ) try: for keyset_id in primary_wallet.keysets: - await primary_wallet.restore_tokens_for_keyset(keyset_id, to=1, batch=25) + await primary_wallet.restore_tokens_for_keyset( + keyset_id, to=1, batch=25 + ) await primary_wallet.load_proofs(reload=True) post_recovery_balance = primary_wallet.available_balance.amount balance_gained = post_recovery_balance - pre_mint_balance @@ -854,7 +861,9 @@ async def fetch_all_balances( "unit": unit, "wallet_balance": proofs_balance, "user_balance": user_balance, - "owner_balance": proofs_balance - user_balance if proofs_balance != 0 else 0, + "owner_balance": proofs_balance - user_balance + if proofs_balance != 0 + else 0, } return result except Exception as e: @@ -994,54 +1003,59 @@ async def periodic_payout() -> None: ) +async def refund_sweep_once() -> None: + """Sweep eligible uncollected refund tokens once.""" + cutoff = int(time.time()) - settings.refund_sweep_ttl_seconds + async with db.create_session() as session: + stmt = select(db.CashuTransaction).where( + db.CashuTransaction.type == "out", + db.CashuTransaction.collected == False, # noqa: E712 + db.CashuTransaction.swept == False, # noqa: E712 + db.CashuTransaction.created_at < cutoff, + ) + results = await session.exec(stmt) + refunds = results.all() + + for refund in refunds: + try: + await recieve_token(refund.token) + refund.swept = True + session.add(refund) + logger.info( + "Swept uncollected refund", + extra={ + "id": refund.id, + "amount": refund.amount, + "unit": refund.unit, + }, + ) + except Exception as e: + error_msg = str(e).lower() + if "already spent" in error_msg: + refund.collected = True + session.add(refund) + logger.info( + "Refund already spent (client collected), marking swept", + extra={ + "id": refund.id, + }, + ) + else: + logger.warning( + "Failed to sweep refund", + extra={ + "id": refund.id, + "error": str(e), + }, + ) + await session.commit() + + async def periodic_refund_sweep() -> None: while True: await asyncio.sleep(60 * 60) # every hour try: - cutoff = int(time.time()) - settings.refund_sweep_ttl_seconds - async with db.create_session() as session: - stmt = select(db.CashuTransaction).where( - db.CashuTransaction.type == "out", - db.CashuTransaction.collected == False, # noqa: E712 - db.CashuTransaction.swept == False, # noqa: E712 - db.CashuTransaction.created_at < cutoff, - ) - results = await session.exec(stmt) - refunds = results.all() - - for refund in refunds: - try: - await recieve_token(refund.token) - refund.swept = True - session.add(refund) - logger.info( - "Swept uncollected refund", - extra={ - "id": refund.id, - "amount": refund.amount, - "unit": refund.unit, - }, - ) - except Exception as e: - error_msg = str(e).lower() - if "already spent" in error_msg: - refund.collected = True - session.add(refund) - logger.info( - "Refund already spent (client collected), marking swept", - extra={ - "id": refund.id, - }, - ) - else: - logger.warning( - "Failed to sweep refund", - extra={ - "id": refund.id, - "error": str(e), - }, - ) - await session.commit() + await refund_sweep_once() except Exception as e: logger.error( "Error in periodic refund sweep", @@ -1071,7 +1085,11 @@ async def periodic_routstr_fee_payout() -> None: wallet, settings.primary_mint, "sat", not_reserved=True ) amount_received = await raw_send_to_lnurl( - wallet, proofs, ROUTSTR_LN_ADDRESS, "sat", amount=accumulated_sats + wallet, + proofs, + ROUTSTR_LN_ADDRESS, + "sat", + amount=accumulated_sats, ) paid_msats = accumulated_sats * 1000 await db.reset_routstr_fee(session, paid_msats) diff --git a/tests/unit/test_refund_sweep.py b/tests/unit/test_refund_sweep.py new file mode 100644 index 00000000..e82c2111 --- /dev/null +++ b/tests/unit/test_refund_sweep.py @@ -0,0 +1,118 @@ +from collections.abc import AsyncIterator +from pathlib import Path +from unittest.mock import AsyncMock, patch + +import pytest +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlmodel import SQLModel, select +from sqlmodel.ext.asyncio.session import AsyncSession + +from routstr.core.db import CashuTransaction +from routstr.wallet import refund_sweep_once + + +@pytest.fixture +async def session_factory( + tmp_path: Path, +) -> AsyncIterator[async_sessionmaker[AsyncSession]]: + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'refunds.db'}") + async with engine.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + factory = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + yield factory + await engine.dispose() + + +async def _insert( + factory: async_sessionmaker[AsyncSession], *rows: CashuTransaction +) -> None: + async with factory() as session: + session.add_all(rows) + await session.commit() + + +async def _load( + factory: async_sessionmaker[AsyncSession], +) -> dict[str, CashuTransaction]: + async with factory() as session: + result = await session.exec(select(CashuTransaction)) + return {row.token: row for row in result.all()} + + +@pytest.mark.asyncio +async def test_refund_sweep_only_processes_expired_eligible_outgoing_tokens( + session_factory: async_sessionmaker[AsyncSession], +) -> None: + rows = [ + CashuTransaction( + token="eligible", amount=1, unit="sat", type="out", created_at=800 + ), + CashuTransaction( + token="fresh", amount=1, unit="sat", type="out", created_at=950 + ), + CashuTransaction( + token="boundary", amount=1, unit="sat", type="out", created_at=900 + ), + CashuTransaction( + token="collected", + amount=1, + unit="sat", + type="out", + created_at=800, + collected=True, + ), + CashuTransaction( + token="swept", amount=1, unit="sat", type="out", created_at=800, swept=True + ), + CashuTransaction( + token="incoming", amount=1, unit="sat", type="in", created_at=800 + ), + ] + await _insert(session_factory, *rows) + receive = AsyncMock() + with ( + patch("routstr.wallet.db.create_session", side_effect=session_factory), + patch("routstr.wallet.settings.refund_sweep_ttl_seconds", 100), + patch("routstr.wallet.time.time", return_value=1000), + patch("routstr.wallet.recieve_token", receive), + ): + await refund_sweep_once() + + receive.assert_awaited_once_with("eligible") + loaded = await _load(session_factory) + assert loaded["eligible"].swept is True + assert loaded["fresh"].swept is False + assert loaded["boundary"].swept is False + assert loaded["collected"].swept is False + assert loaded["swept"].swept is True + assert loaded["incoming"].swept is False + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("error", "collected"), + [ + (RuntimeError("token already spent"), True), + (RuntimeError("mint unavailable"), False), + ], +) +async def test_refund_sweep_records_terminal_but_not_transient_failures( + session_factory: async_sessionmaker[AsyncSession], error: Exception, collected: bool +) -> None: + await _insert( + session_factory, + CashuTransaction( + token="refund", amount=1, unit="sat", type="out", created_at=800 + ), + ) + with ( + patch("routstr.wallet.db.create_session", side_effect=session_factory), + patch("routstr.wallet.settings.refund_sweep_ttl_seconds", 100), + patch("routstr.wallet.time.time", return_value=1000), + patch("routstr.wallet.recieve_token", AsyncMock(side_effect=error)), + ): + await refund_sweep_once() + + refund = (await _load(session_factory))["refund"] + assert refund.collected is collected + assert refund.swept is False