refactor(v2): hoist LP state (wallet, mode, autoforward) into dca_lp table

LP-level preferences were denormalised across every `dca_clients` row
of a given user. Every LP enrolment carried its own wallet_id /
dca_mode / fixed_mode_daily_limit / autoforward_ln_address /
autoforward_enabled — and satmachineclient's `update_lp_autoforward`
did a multi-row UPDATE to keep them in sync. That sync dance was the
smell: user-level intent stored at machine-enrolment granularity.

New shape:

  dca_lp  (user_id PK, dca_wallet_id, default_dca_mode,
           fixed_mode_daily_limit, autoforward_ln_address,
           autoforward_enabled, ...)
  dca_clients  (id, machine_id, user_id, username, status, ...)
        // pure (machine, LP) enrolment — wallet/mode/autoforward gone

Authority split:
  - LP writes dca_lp via satmachineclient (Phase 2, separate commit).
  - Operator writes dca_clients via satmachineadmin. They cannot
    choose the LP's destination wallet — it's resolved from dca_lp
    at distribution time. Better trust hygiene.

Onboarding gate:
  - `api_create_deposit` refuses (HTTP 422) when the target LP has
    no dca_lp row. Forces every LP through a "yes, I am here and
    this is where I want my sats" gesture via satmachineclient
    before any fiat starts accumulating against them.

Schema:
  - m001 canonical schema updated: slim `dca_clients`, new `dca_lp`.
    Fresh installs land here directly.
  - m004 idempotent migration for installs that already have the
    legacy `dca_clients.wallet_id` column: creates dca_lp,
    backfills from the latest dca_clients row per user (window
    function), then DROP COLUMN on the moved fields. Greg's live
    test data survives the upgrade.

Distribution:
  - `get_flow_mode_clients_for_machine` INNER JOINs dca_lp so
    un-onboarded LPs are filtered out (no destination wallet).
  - `_pay_one_dca_leg`, `_attempt_autoforward`, `settle_lp_balance`
    all fetch `dca_lp` via the new `get_dca_lp(user_id)` helper.
    Wallet + autoforward read from prefs, not from client.

Models:
  - `DcaClient` loses 5 fields. `CreateDcaClientData` reduces to
    (machine_id, user_id, username). `UpdateDcaClientData` keeps
    only operator-controlled fields (username, status).
  - New `DcaLpPreferences` + `UpsertDcaLpData` models for the
    per-user surface (satmachineclient writes these in Phase 2).

CRUD:
  - New: `get_dca_lp`, `lp_is_onboarded`, `upsert_dca_lp` (the
    latter takes a `fallback_wallet_id` for first-onboarding when
    satmachineclient auto-seeds from the LP's default LNbits wallet).
  - `create_dca_client` insert reduces to the new column set.

Tests: 86 unit tests still green.

Next:
  - Phase 1c (this repo): UI simplification for operator's
    Add/Edit LP dialogs + deposit-gating UX.
  - Phase 2 (satmachineclient): own dca_lp writes + auto-init with
    the LP's default LNbits wallet on first dashboard visit.

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

distribution.py

@@ -41,6 +41,7 @@ from .crud import (
     count_completed_legs_for_settlement,
     create_dca_payment,
     get_client_balance_summary,
+    get_dca_lp,
     get_effective_commission_splits,
     get_flow_mode_clients_for_machine,
     get_machine,
@@ -53,6 +54,7 @@ from .crud import (
 from .models import (
     CreateDcaPaymentData,
     DcaClient,
+    DcaLpPreferences,
     DcaPayment,
     DcaSettlement,
     Machine,
@@ -172,7 +174,17 @@ async def settle_lp_balance(
     machine and the funding wallet (API endpoint does this). The amount_fiat
     is capped at the LP's remaining balance — operators cannot accidentally
     over-pay via this path.
+
+    The destination wallet is the LP's own `dca_lp.dca_wallet_id` — the
+    operator can't redirect this; if the LP hasn't onboarded yet there's
+    no destination and we refuse.
     """
+    prefs = await get_dca_lp(client.user_id)
+    if prefs is None:
+        raise ValueError(
+            f"client {client.id} (user {client.user_id[:8]}...) has not "
+            f"onboarded via satmachineclient — no DCA wallet configured"
+        )
     summary = await get_client_balance_summary(client.id)
     if summary is None:
         raise ValueError(f"client {client.id} balance not available")
@@ -208,7 +220,7 @@ async def settle_lp_balance(
             machine_id=machine.id,
             operator_user_id=machine.operator_user_id,
             leg_type="settlement",
-            destination_wallet_id=client.wallet_id,
+            destination_wallet_id=prefs.dca_wallet_id,
             destination_ln_address=None,
             amount_sats=amount_sats,
             amount_fiat=amount_fiat,
@@ -225,7 +237,7 @@ async def settle_lp_balance(
     }
     try:
         new_invoice = await create_invoice(
-            wallet_id=client.wallet_id,
+            wallet_id=prefs.dca_wallet_id,
             amount=float(amount_sats),
             internal=True,
             memo=memo,
@@ -557,9 +569,20 @@ async def _pay_one_dca_leg(
     amount_sats: int,
     errors: List[str],
 ) -> None:
-    """Pay a single DCA leg + best-effort autoforward."""
+    """Pay a single DCA leg + best-effort autoforward.
+
+    Reads the LP's destination wallet + autoforward config from `dca_lp`.
+    Callers reach this through `get_flow_mode_clients_for_machine` which
+    INNER JOINs on `dca_lp`, so a `prefs is None` here would indicate a
+    race (LP deleted their dca_lp row between query and pay) — we
+    defensively skip.
+    """
     if amount_sats <= 0:
         return
+    prefs = await get_dca_lp(client.user_id)
+    if prefs is None:
+        errors.append(f"client {client.id}: dca_lp row disappeared mid-distribution")
+        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(
@@ -567,7 +590,7 @@ async def _pay_one_dca_leg(
         machine=machine,
         leg_type="dca",
         client_id=client.id,
-        destination_wallet_id=client.wallet_id,
+        destination_wallet_id=prefs.dca_wallet_id,
         amount_sats=amount_sats,
         amount_fiat=amount_fiat,
         exchange_rate=float(settlement.exchange_rate),
@@ -581,10 +604,10 @@ async def _pay_one_dca_leg(
     if (
         dca_leg is not None
         and dca_leg.status == "completed"
-        and client.autoforward_enabled
-        and client.autoforward_ln_address
+        and prefs.autoforward_enabled
+        and prefs.autoforward_ln_address
     ):
-        await _attempt_autoforward(client, machine, settlement, amount_sats)
+        await _attempt_autoforward(client, prefs, machine, settlement, amount_sats)
 
 
 # =============================================================================
@@ -594,6 +617,7 @@ async def _pay_one_dca_leg(
 
 async def _attempt_autoforward(
     client: DcaClient,
+    prefs: DcaLpPreferences,
     machine: Machine,
     settlement: DcaSettlement,
     amount_sats: int,
@@ -610,7 +634,7 @@ async def _attempt_autoforward(
     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
+    address = prefs.autoforward_ln_address
     if not address:
         return
     leg = await create_dca_payment(
@@ -637,7 +661,7 @@ async def _attempt_autoforward(
             comment=f"satmachine autoforward — {machine.machine_npub[:12]}",
         )
         paid = await pay_invoice(
-            wallet_id=client.wallet_id,
+            wallet_id=prefs.dca_wallet_id,
             payment_request=bolt11,
             description=f"satmachine autoforward → {address}",
             tag=_payment_tag(machine),