satmachineclient/crud.py

# Satoshi Machine Client v2 — CRUD over admin schema.
#
# Cross-extension reads of satoshimachine.dca_* tables, filtered by the
# LP's user_id. The admin extension owns writes to dca_clients/deposits/
# settlements; this extension owns writes to satoshimachine.dca_lp
# (the LP's per-user preferences row — wallet, mode, autoforward).

from datetime import datetime
from typing import List, Optional

from lnbits.db import Database
from lnbits.utils.exchange_rates import satoshis_amount_as_fiat
from loguru import logger

from .models import (
    ClientDashboardSummary,
    ClientTransaction,
    LpPreferences,
    PerMachinePosition,
    UpdateLpPreferences,
)

# Same DB schema as the admin extension — we share satoshimachine.* tables.
db = Database("ext_satoshimachine")


async def _fetch_user_clients(user_id: str) -> List[dict]:
    """All dca_clients rows for this LP, joined with their machines for
    per-machine display metadata.

    `dca_mode` lives on the LP's per-user `dca_lp` row now, not per
    enrolment — INNER JOIN dca_lp so positions only return when the LP
    has actually onboarded (otherwise distribution can't pay them either,
    so listing them in the dashboard would be misleading).
    """
    return await db.fetchall(
        """
        SELECT c.id AS client_id,
               c.machine_id,
               c.status,
               m.machine_npub,
               m.name AS machine_name,
               m.location AS machine_location,
               m.fiat_code AS machine_fiat_code,
               lp.default_dca_mode AS dca_mode
        FROM satoshimachine.dca_clients c
        JOIN satoshimachine.dca_machines m ON m.id = c.machine_id
        JOIN satoshimachine.dca_lp lp ON lp.user_id = c.user_id
        WHERE c.user_id = :user_id
        ORDER BY c.created_at DESC
        """,
        {"user_id": user_id},
    )


async def _position_for_client(client_row: dict) -> PerMachinePosition:
    cid = client_row["client_id"]
    confirmed_deposits = await db.fetchone(
        """
        SELECT COALESCE(SUM(amount), 0) AS total
        FROM satoshimachine.dca_deposits
        WHERE client_id = :cid AND status = 'confirmed'
        """,
        {"cid": cid},
    )
    # DCA + settlement legs both count against the LP's balance.
    distributed = await db.fetchone(
        """
        SELECT COALESCE(SUM(amount_fiat), 0) AS total_fiat,
               COALESCE(SUM(amount_sats), 0) AS total_sats,
               COUNT(*)                       AS tx_count,
               MAX(created_at)                AS last_tx
        FROM satoshimachine.dca_payments
        WHERE client_id = :cid
          AND leg_type IN ('dca', 'settlement')
          AND status = 'completed'
        """,
        {"cid": cid},
    )
    total_deposits = float(confirmed_deposits["total"]) if confirmed_deposits else 0.0
    total_fiat = float(distributed["total_fiat"]) if distributed else 0.0
    total_sats = int(distributed["total_sats"]) if distributed else 0
    tx_count = int(distributed["tx_count"]) if distributed else 0
    last_tx = distributed["last_tx"] if distributed else None
    return PerMachinePosition(
        machine_id=client_row["machine_id"],
        machine_npub=client_row["machine_npub"],
        machine_name=client_row["machine_name"],
        machine_location=client_row["machine_location"],
        currency=client_row["machine_fiat_code"],
        dca_mode=client_row["dca_mode"],
        status=client_row["status"],
        total_sats_accumulated=total_sats,
        total_fiat_invested=round(total_deposits, 2),
        current_fiat_balance=round(total_deposits - total_fiat, 2),
        total_transactions=tx_count,
        last_transaction_date=last_tx,
    )


async def get_client_dashboard_summary(
    user_id: str,
) -> Optional[ClientDashboardSummary]:
    """Aggregated LP dashboard. Returns None if the LP has no positions."""
    clients = await _fetch_user_clients(user_id)
    if not clients:
        return None
    positions = [await _position_for_client(c) for c in clients]
    total_sats = sum(p.total_sats_accumulated for p in positions)
    total_invested = round(sum(p.total_fiat_invested for p in positions), 2)
    total_balance = round(sum(p.current_fiat_balance for p in positions), 2)
    total_tx = sum(p.total_transactions for p in positions)
    last_tx = max(
        (p.last_transaction_date for p in positions if p.last_transaction_date),
        default=None,
    )
    # Pending deposits (across all clients of this LP).
    pending = await db.fetchone(
        """
        SELECT COALESCE(SUM(d.amount), 0) AS total
        FROM satoshimachine.dca_deposits d
        JOIN satoshimachine.dca_clients c ON c.id = d.client_id
        WHERE c.user_id = :uid AND d.status = 'pending'
        """,
        {"uid": user_id},
    )
    pending_fiat = float(pending["total"]) if pending else 0.0
    # Cost basis = sats / fiat-spent. Use the distributed fiat (deposits
    # less remaining balance) so we get the cost basis on what's actually
    # been DCA'd, not the gross deposits.
    fiat_spent = max(total_invested - total_balance, 0.0)
    cost_basis = total_sats / fiat_spent if fiat_spent > 0 else 0.0
    # Display currency: if all positions share one, use it; else "MIX".
    currencies = {p.currency for p in positions}
    currency = currencies.pop() if len(currencies) == 1 else "MIX"
    # Best-effort current fiat value of the LP's sats stack.
    current_value = 0.0
    if total_sats > 0 and currency != "MIX":
        try:
            current_value = await satoshis_amount_as_fiat(total_sats, currency)
        except Exception as exc:
            logger.warning(
                f"satmachineclient: could not fetch exchange rate for "
                f"{currency}: {exc}"
            )
    return ClientDashboardSummary(
        user_id=user_id,
        total_sats_accumulated=total_sats,
        total_fiat_invested=total_invested,
        current_fiat_balance=total_balance,
        pending_fiat_deposits=round(pending_fiat, 2),
        average_cost_basis=round(cost_basis, 4),
        current_sats_fiat_value=round(current_value, 2),
        total_transactions=total_tx,
        total_machines=len(positions),
        last_transaction_date=last_tx,
        currency=currency,
        positions=positions,
    )


async def get_client_transactions(
    user_id: str,
    limit: int = 50,
    offset: int = 0,
    machine_id: Optional[str] = None,
) -> List[ClientTransaction]:
    """LP's transaction history. Filters to 'dca' / 'settlement' /
    'autoforward' legs since those are the ones the LP cares about — super_fee
    and operator_split legs are operator-internal.

    Optional machine_id narrows to a single machine."""
    params: dict = {
        "uid": user_id,
        "lim": limit,
        "off": offset,
    }
    where = (
        "WHERE c.user_id = :uid "
        "AND p.leg_type IN ('dca', 'settlement', 'autoforward')"
    )
    if machine_id is not None:
        where += " AND p.machine_id = :mid"
        params["mid"] = machine_id
    rows = await db.fetchall(
        f"""
        SELECT p.id, p.machine_id, p.settlement_id, p.leg_type,
               p.amount_sats, p.amount_fiat, p.exchange_rate, p.status,
               p.created_at, p.transaction_time,
               m.machine_npub
        FROM satoshimachine.dca_payments p
        JOIN satoshimachine.dca_clients c ON c.id = p.client_id
        JOIN satoshimachine.dca_machines m ON m.id = p.machine_id
        {where}
        ORDER BY p.created_at DESC
        LIMIT :lim OFFSET :off
        """,
        params,
    )
    return [
        ClientTransaction(
            id=row["id"],
            machine_id=row["machine_id"],
            machine_npub=row["machine_npub"],
            settlement_id=row["settlement_id"],
            leg_type=row["leg_type"],
            amount_sats=row["amount_sats"],
            amount_fiat=row["amount_fiat"],
            exchange_rate=row["exchange_rate"],
            status=row["status"],
            created_at=row["created_at"],
            transaction_time=row["transaction_time"],
        )
        for row in rows
    ]


async def get_lp_preferences(user_id: str) -> Optional[LpPreferences]:
    """Read this LP's preferences row from `dca_lp`. Returns None if the
    LP hasn't onboarded yet (no row). Callers in this extension generally
    use `ensure_lp_preferences` instead, which auto-creates on first
    access."""
    return await db.fetchone(
        "SELECT * FROM satoshimachine.dca_lp WHERE user_id = :uid",
        {"uid": user_id},
        LpPreferences,
    )


async def ensure_lp_preferences(user_id: str, default_wallet_id: str) -> LpPreferences:
    """Get-or-create the LP's preferences row.

    First call (no row): seed with `default_wallet_id` (passed by the
    caller — typically the wallet the LP authenticated through). LP can
    change it later via `update_lp_preferences`.

    This is the structural enforcement of the "LP must onboard before
    deposits work" gate: the act of opening satmachineclient and hitting
    any endpoint creates the dca_lp row, which unlocks deposit creation
    on the operator side.
    """
    existing = await get_lp_preferences(user_id)
    if existing is not None:
        return existing
    now = datetime.now()
    await db.execute(
        """
        INSERT INTO satoshimachine.dca_lp
            (user_id, dca_wallet_id, default_dca_mode,
             autoforward_enabled, created_at, updated_at)
        VALUES (:uid, :wallet, 'flow', false, :now, :now)
        """,
        {"uid": user_id, "wallet": default_wallet_id, "now": now},
    )
    created = await get_lp_preferences(user_id)
    assert created is not None
    return created


async def update_lp_preferences(
    user_id: str, data: UpdateLpPreferences
) -> Optional[LpPreferences]:
    """LP-side update of their `dca_lp` row. Caller must ensure the row
    exists first (typically via `ensure_lp_preferences` on dashboard
    load). Operator cannot reach this path — it requires the LP's wallet
    admin key per the API auth dependency."""
    update_data = {k: v for k, v in data.dict().items() if v is not None}
    if not update_data:
        return await get_lp_preferences(user_id)
    update_data["updated_at"] = datetime.now()
    update_data["uid"] = user_id
    set_clause = ", ".join(f"{k} = :{k}" for k in update_data if k not in ("uid",))
    await db.execute(
        f"UPDATE satoshimachine.dca_lp SET {set_clause} WHERE user_id = :uid",
        update_data,
    )
    return await get_lp_preferences(user_id)