feat(v2): wire bitSpire invoice listener + settlement landing (P1a)

Replaces the no-op tasks.py stub with a real invoice listener that lands
bitSpire settlements idempotently into dca_settlements.

Architecture: satmachineadmin runs *inside* the LNbits process, so it
plugs into LNbits' canonical extension hook (register_invoice_listener
from lnbits.tasks) instead of going through the Nostr transport layer.
External clients like bitSpire use Nostr; internal extensions consume
the resulting Payment objects directly. One invoice_listener queue per
extension, dispatched by invoice_callback_dispatcher.

Flow:
  bitSpire ATM (Nostr kind-21000)
    → LNbits nostr_transport handler
    → core Payment system (create_invoice + status=SUCCESS on settle)
    → invoice_callback_dispatcher
    → satmachineadmin's invoice_queue
    → _handle_payment filters by wallet_id → active machine
    → bitspire.parse_settlement reads Payment.extra (or back-derives)
    → create_settlement_idempotent (keyed on payment_hash UNIQUE)

The parser (new bitspire.py module) is bitSpire-specific:

- Happy path (post-aiolabs/lamassu-next#44): Payment.extra carries
  {source:"bitspire", net_sats, fee_sats, fee_pct, exchange_rate,
   currency, txid, machine_npub, bills, cassettes}. Read directly,
  zero back-derivation.
- Fallback path (pre-#44): extra is absent. Back-derive the split
  using machine.fallback_commission_pct with the Lamassu-style
  formula (calculations.calculate_commission), mark
  used_fallback_split=true, log a WARNING that namechecks the
  upstream issue so it's findable in logs.

Two-stage commission split (super first, operator remainder) is
computed at land time so the audit row is complete:
  platform_fee_sats = round(commission_sats * super_fee_pct)
  operator_fee_sats = commission_sats - platform_fee_sats

The actual payout (LP DCA legs + super-fee leg + operator-split legs)
happens in a separate settlement-processor task in P2. P1 only LANDS
the settlement with status='pending'.

Smoke-tested both paths against real LNbits 1.4 (nostr-transport venv):
  happy:    266800 gross → 258835 net + 7965 commission
            (2390 super @ 30%, 5575 operator)
  fallback: 266800 gross → 254095 net + 12705 commission @ 5% default

Also adds crud.get_active_machine_by_wallet_id, the lookup that gates
inbound payments to known machine wallets.

Refs: aiolabs/satmachineadmin#9, aiolabs/lamassu-next#44

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tasks.py

@@ -1,27 +1,83 @@
-# Satoshi Machine v2 — task placeholders.
+# Satoshi Machine v2 — invoice listener (P1).
 #
-# The v1 SSH/PostgreSQL polling + invoice listener are intentionally absent.
-# They will be replaced in P1 (Nostr subscription manager: subscribes via
-# lnbits.core.services.nostr_transport to kind-21000 settlements + kind-30078
-# beacons + kind-30079 telemetry per registered machine, with auto-reconnect).
+# Subscribes to LNbits' invoice dispatcher (register_invoice_listener), then
+# for each successful inbound payment:
+#   1. Checks if wallet_id belongs to an active dca_machines row. If not, skip.
+#   2. Parses Payment.extra for bitSpire split metadata (post-lamassu-next#44).
+#      Falls back to machine.fallback_commission_pct if extra is absent.
+#   3. Computes the two-stage split (super_fee first, operator remainder).
+#   4. Inserts a dca_settlements row idempotently (keyed by payment_hash).
 #
-# These no-op stubs keep __init__.py importable in the interim so the
-# extension can be activated even before P1 lands.
+# The actual distribution of sats — paying out the LP DCA legs, the super-fee
+# leg, and the operator's commission-split legs — happens in a separate
+# settlement-processor task (P2). This listener only LANDS the settlement
+# row; status='pending' tells the processor it still needs to move the money.
 
 import asyncio
 
+from lnbits.core.models import Payment
+from lnbits.tasks import register_invoice_listener
 from loguru import logger
 
+from .bitspire import parse_settlement
+from .crud import (
+    create_settlement_idempotent,
+    get_active_machine_by_wallet_id,
+    get_super_config,
+)
+
+LISTENER_NAME = "ext_satmachineadmin"
+
 
 async def wait_for_paid_invoices() -> None:
-    """No-op placeholder pending P1 Nostr subscription manager."""
-    logger.debug(
-        "satmachineadmin v2: invoice listener stub running. "
-        "Real Nostr-transport subscription pending P1."
+    invoice_queue: asyncio.Queue = asyncio.Queue()
+    register_invoice_listener(invoice_queue, LISTENER_NAME)
+    logger.info(
+        "satmachineadmin v2: invoice listener registered as "
+        f"`{LISTENER_NAME}` — waiting for bitSpire settlements."
     )
-    # Sleep forever; the task system expects a long-lived coroutine.
     while True:
-        await asyncio.sleep(3600)
+        payment: Payment = await invoice_queue.get()
+        try:
+            await _handle_payment(payment)
+        except Exception as exc:  # listener must never die
+            logger.error(
+                f"satmachineadmin: error handling payment "
+                f"{payment.payment_hash[:12]}...: {exc}"
+            )
+
+
+async def _handle_payment(payment: Payment) -> None:
+    if not payment.is_in or not payment.success:
+        return
+    machine = await get_active_machine_by_wallet_id(payment.wallet_id)
+    if machine is None:
+        return
+    super_config = await get_super_config()
+    super_fee_pct = float(super_config.super_fee_pct) if super_config else 0.0
+    data, used_fallback = parse_settlement(
+        machine=machine,
+        payment_hash=payment.payment_hash,
+        gross_sats=payment.sat,
+        extra=payment.extra or {},
+        super_fee_pct=super_fee_pct,
+    )
+    settlement = await create_settlement_idempotent(data)
+    if settlement is None:
+        logger.error(
+            f"satmachineadmin: failed to insert settlement for "
+            f"payment_hash={payment.payment_hash[:12]}..."
+        )
+        return
+    fb = " (fallback split)" if used_fallback else ""
+    logger.info(
+        f"satmachineadmin: landed settlement {settlement.id} for "
+        f"machine={machine.machine_npub[:12]}... "
+        f"gross={data.gross_sats}sats net={data.net_sats}sats "
+        f"commission={data.commission_sats}sats "
+        f"(super_fee={data.platform_fee_sats} "
+        f"operator_fee={data.operator_fee_sats}){fb}"
+    )
 
 
 async def hourly_transaction_polling() -> None: