feat(v2): LP auto-forward to LN address (P6 — closes #8)

Closes satmachineadmin#8 — operator-configured LP autoforward to an
external Lightning address. The data path was already in place from P0d
(autoforward_enabled + autoforward_ln_address on dca_clients); this
commit wires the actual outbound LN-address payment.

Flow (in distribution._attempt_autoforward, called from the DCA leg path):
  1. DCA leg lands in LP's LNbits wallet (regular internal transfer)
  2. If client.autoforward_enabled AND autoforward_ln_address set:
       a. Wrap address in lnurl.LnAddress
       b. Resolve to bolt11 via lnbits.core.services.lnurl.get_pr_from_lnurl
       c. Pay bolt11 from LP's wallet via pay_invoice
       d. Record a leg_type='autoforward' dca_payments row with
          destination_ln_address set
  3. On ANY failure (malformed addr, LNURL resolution fail, payment
     timeout): log warning, mark the autoforward leg 'failed', and
     leave sats in the LP's LNbits wallet — the explicit safety
     constraint from the original issue.

Audit: every autoforward attempt records a row (success or fail) so
operators can see in payment history which forwards landed externally
vs which left sats in LNbits. The destination_ln_address column on
dca_payments was already nullable to support this use case.

Safety guards:
- Skip autoforward if the DCA leg itself failed (nothing to forward).
- _attempt_autoforward never re-raises — failed forwarding must not
  abort subsequent DCA legs for other LPs at this machine.
- Sats only move from the LP's wallet (which they own), never from
  the operator's or super's wallets.

Refactor: extracted _pay_one_dca_leg from _pay_dca_distributions
to keep the outer function under the C901 complexity limit.

72/72 tests pass.

Refs: aiolabs/satmachineadmin#9, closes #8 (autoforward feature
request) — marked once verified end-to-end with a real LN address.

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

distribution.py

@@ -27,6 +27,8 @@ from datetime import datetime, timezone
 from typing import List
 
 from lnbits.core.services import create_invoice, pay_invoice
+from lnbits.core.services.lnurl import get_pr_from_lnurl
+from lnurl import LnAddress
 from loguru import logger
 
 from .calculations import (
@@ -470,35 +472,128 @@ async def _pay_dca_distributions(
         remaining_fiat = client_balances[client_id]
         cap_sats = int(remaining_fiat * float(settlement.exchange_rate))
         capped_allocations[client_id] = min(raw_sats, cap_sats)
-    # Pay each capped allocation.
     client_by_id = {c.id: c for c in clients}
     for client_id, amount_sats in capped_allocations.items():
-        if amount_sats <= 0:
-            continue
-        client = client_by_id[client_id]
-        amount_fiat = round(amount_sats / float(settlement.exchange_rate), 2)
-        memo = (
-            f"DCA: {amount_sats} sats • {amount_fiat:.2f} {settlement.fiat_code}"
-        )
-        await _pay_internal(
-            settlement=settlement,
-            machine=machine,
-            leg_type="dca",
-            client_id=client.id,
-            destination_wallet_id=client.wallet_id,
-            amount_sats=amount_sats,
-            amount_fiat=amount_fiat,
-            exchange_rate=float(settlement.exchange_rate),
-            memo=memo,
-            errors=errors,
+        await _pay_one_dca_leg(
+            settlement, machine, client_by_id[client_id], amount_sats, errors
         )
 
 
+async def _pay_one_dca_leg(
+    settlement: DcaSettlement,
+    machine: Machine,
+    client: DcaClient,
+    amount_sats: int,
+    errors: List[str],
+) -> None:
+    """Pay a single DCA leg + best-effort autoforward."""
+    if amount_sats <= 0:
+        return
+    amount_fiat = round(amount_sats / float(settlement.exchange_rate), 2)
+    memo = (
+        f"DCA: {amount_sats} sats • {amount_fiat:.2f} {settlement.fiat_code}"
+    )
+    dca_leg = await _pay_internal(
+        settlement=settlement,
+        machine=machine,
+        leg_type="dca",
+        client_id=client.id,
+        destination_wallet_id=client.wallet_id,
+        amount_sats=amount_sats,
+        amount_fiat=amount_fiat,
+        exchange_rate=float(settlement.exchange_rate),
+        memo=memo,
+        errors=errors,
+    )
+    # Best-effort auto-forward to LP's external LN address (closes
+    # satmachineadmin#8). Skip if the DCA leg failed (nothing to forward).
+    # If autoforward fails, sats stay in the LP's LNbits wallet — the
+    # explicit safety constraint.
+    if (
+        dca_leg is not None
+        and dca_leg.status == "completed"
+        and client.autoforward_enabled
+        and client.autoforward_ln_address
+    ):
+        await _attempt_autoforward(client, machine, settlement, amount_sats)
+
+
 # =============================================================================
 # Internal transfer helper
 # =============================================================================
 
 
+async def _attempt_autoforward(
+    client: DcaClient,
+    machine: Machine,
+    settlement: DcaSettlement,
+    amount_sats: int,
+) -> None:
+    """LP auto-forward (best-effort) — closes satmachineadmin#8.
+
+    Resolves the LP's configured LN address, requests a bolt11 invoice for
+    the DCA leg's sat amount, and pays it from the LP's LNbits wallet. Each
+    attempt records a dca_payments row with leg_type='autoforward' for
+    audit, regardless of outcome.
+
+    Safety: on any failure (malformed address, LNURL resolution fail,
+    payment timeout, etc.) we log a warning and leave the sats in the LP's
+    LNbits wallet. The LP can move them manually via the LNbits UI. We
+    never re-raise; failed forwarding must not block subsequent legs.
+    """
+    address = client.autoforward_ln_address
+    if not address:
+        return
+    leg = await create_dca_payment(
+        CreateDcaPaymentData(
+            settlement_id=settlement.id,
+            client_id=client.id,
+            machine_id=machine.id,
+            operator_user_id=machine.operator_user_id,
+            leg_type="autoforward",
+            destination_wallet_id=None,
+            destination_ln_address=address,
+            amount_sats=amount_sats,
+            amount_fiat=None,
+            exchange_rate=None,
+            transaction_time=datetime.now(timezone.utc),
+            external_payment_hash=None,
+        )
+    )
+    try:
+        lnaddr = LnAddress(address)
+        bolt11 = await get_pr_from_lnurl(
+            lnurl=lnaddr,
+            amount_msat=amount_sats * 1000,
+            comment=f"satmachine autoforward — {machine.machine_npub[:12]}",
+        )
+        paid = await pay_invoice(
+            wallet_id=client.wallet_id,
+            payment_request=bolt11,
+            description=f"satmachine autoforward → {address}",
+            tag=_payment_tag(machine),
+            extra={
+                "satmachine_leg": "autoforward",
+                "satmachine_settlement_id": settlement.id,
+                "satmachine_machine_npub": machine.machine_npub,
+                "satmachine_destination": address,
+            },
+        )
+        await update_payment_status(
+            leg.id, "completed", paid.payment_hash, None
+        )
+        logger.info(
+            f"distribution: autoforward {amount_sats} sats from client "
+            f"{client.id} → {address} OK"
+        )
+    except Exception as exc:
+        logger.warning(
+            f"distribution: autoforward FAILED for client {client.id} "
+            f"→ {address}: {exc}. Sats stay in LP's LNbits wallet."
+        )
+        await update_payment_status(leg.id, "failed", None, str(exc)[:512])
+
+
 async def _pay_internal(
     *,
     settlement: DcaSettlement,