Add deposit crediting from confirmed UTXOs

Credits a user's internal balance once a UTXO on their deposit address
reaches 1 confirmation, keeping utxo_events as the source of truth and
cached_balance_sats as a derived read cache.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

app/deposits/service.py

@@ -0,0 +1,54 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.audit.log import write_audit_log
+from app.db.models import UtxoEvent
+from app.wallet.balance import recompute_balance
+
+
+async def credit_confirmed_utxos(session: AsyncSession, user_id: int, entries: list[dict]) -> int:
+    """Insert utxo_events for newly-confirmed entries from an Electrum
+    `listunspent` response (idempotent on txid+vout), refresh the user's cached
+    balance. Returns the number of newly-credited UTXOs.
+
+    entries: [{"tx_hash": ..., "tx_pos": ..., "height": ..., "value": ...}, ...]
+    height <= 0 means unconfirmed (mempool) per the Electrum protocol convention —
+    skipped, since the spec requires 1 confirmation before crediting.
+    """
+    existing_keys = {
+        (txid, vout)
+        for txid, vout in (
+            await session.execute(select(UtxoEvent.txid, UtxoEvent.vout).where(UtxoEvent.user_id == user_id))
+        ).all()
+    }
+
+    newly_credited = 0
+    for entry in entries:
+        if entry["height"] <= 0:
+            continue
+        key = (entry["tx_hash"], entry["tx_pos"])
+        if key in existing_keys:
+            continue
+        session.add(
+            UtxoEvent(
+                user_id=user_id,
+                txid=entry["tx_hash"],
+                vout=entry["tx_pos"],
+                amount_sats=entry["value"],
+                confirmed_height=entry["height"],
+            )
+        )
+        await write_audit_log(
+            session,
+            "deposit_credited",
+            {"txid": entry["tx_hash"], "vout": entry["tx_pos"], "amount_sats": entry["value"]},
+            user_id=user_id,
+        )
+        newly_credited += 1
+
+    if newly_credited:
+        await session.flush()
+        await recompute_balance(session, user_id)
+        await session.commit()
+
+    return newly_credited

tests/unit/test_deposits.py

@@ -0,0 +1,54 @@
+import pytest
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.db.base import Base
+from app.db.models import User
+from app.deposits.service import credit_confirmed_utxos
+
+
+@pytest.fixture
+async def session_factory():
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    yield async_sessionmaker(engine, expire_on_commit=False)
+    await engine.dispose()
+
+
+@pytest.fixture
+async def user_id(session_factory):
+    async with session_factory() as session:
+        user = User(username="alice", password_hash="x", derivation_index=0, address="plm1qxxx")
+        session.add(user)
+        await session.commit()
+        return user.id
+
+
+async def test_credits_confirmed_utxo_and_updates_balance(session_factory, user_id):
+    entries = [{"tx_hash": "aa" * 32, "tx_pos": 0, "height": 100, "value": 10_000_000}]
+    async with session_factory() as session:
+        credited = await credit_confirmed_utxos(session, user_id, entries)
+        assert credited == 1
+        user = await session.get(User, user_id)
+        assert user.cached_balance_sats == 10_000_000
+
+
+async def test_unconfirmed_entry_is_ignored(session_factory, user_id):
+    entries = [{"tx_hash": "bb" * 32, "tx_pos": 0, "height": 0, "value": 5_000_000}]
+    async with session_factory() as session:
+        credited = await credit_confirmed_utxos(session, user_id, entries)
+        assert credited == 0
+        user = await session.get(User, user_id)
+        assert user.cached_balance_sats == 0
+
+
+async def test_idempotent_on_repeated_notification(session_factory, user_id):
+    entries = [{"tx_hash": "cc" * 32, "tx_pos": 0, "height": 100, "value": 7_000_000}]
+    async with session_factory() as session:
+        first = await credit_confirmed_utxos(session, user_id, entries)
+    async with session_factory() as session:
+        second = await credit_confirmed_utxos(session, user_id, entries)
+        user = await session.get(User, user_id)
+        assert first == 1
+        assert second == 0
+        assert user.cached_balance_sats == 7_000_000