modules/database.py

"""
modules/database.py — SQLite Database Interface for Payout Manager v3.0
─────────────────────────────────────────────────────────────────────────

Provides transactional, async-safe SQLite access for payout records,
gas price history, and ledger state. Implements connection pooling and
automatic migrations.
"""

import asyncio
import json
import logging
import sqlite3
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, List, Optional

logger = logging.getLogger(__name__)


@dataclass
class PayoutRecord:
    """Immutable snapshot written at the moment of every payout/sweep."""
    ts: str
    net_profit: float
    gross_profit: float
    total_loan_fees: float
    total_gas_cost: float
    buy_count: int
    scan_count: int
    wallet: str
    chain: str
    note: str = ""
    tx_hash: str = ""
    gas_price_gwei: float = 0.0
    sweep_amount_wei: int = 0
    sweep_type: str = "accounting"


@dataclass
class GasPriceEntry:
    """Historical gas price record for moving average calculation."""
    timestamp: str
    chain: str
    gas_price_gwei: float
    block_number: int
    tx_hash: str = ""


class PayoutDatabase:
    """
    SQLite-backed persistent storage for payout records, gas price history,
    and ledger state.

    Thread-safe via connection pooling and asyncio serialization.
    All I/O operations use asyncio.to_thread to avoid blocking the event loop.
    """

    def __init__(self, db_path: str = "/data/payout_ledger.db") -> None:
        """Initialize database connection and schema."""
        self._db_path = Path(db_path)
        self._lock = asyncio.Lock()
        self._ensure_writable_path()
        self._init_schema()
        logger.info("[PayoutDatabase] Initialized at %s", self._db_path)

    def _ensure_writable_path(self) -> None:
        """
        Ensure database directory exists and is writable.

        Tries, in order:
          1. The requested path as-is.
          2. Path.cwd()/data/<name> (legacy fallback).
          3. /tmp/garden_angel_data/<name> — always writable on HF Spaces
             and most sandboxed containers, used as the last resort so
             the bot can never crash on startup over a permissions issue.
        """
        candidates = [self._db_path]

        cwd_fallback = Path.cwd() / "data" / self._db_path.name
        if cwd_fallback != self._db_path:
            candidates.append(cwd_fallback)

        tmp_fallback = Path("/tmp/garden_angel_data") / self._db_path.name
        if tmp_fallback not in candidates:
            candidates.append(tmp_fallback)

        requested_path = self._db_path
        last_exc: Optional[Exception] = None
        for candidate in candidates:
            try:
                candidate.parent.mkdir(parents=True, exist_ok=True)
                probe = candidate.parent / f".write_test_{id(self)}"
                probe.write_text("ok", encoding="utf-8")
                probe.unlink()
                if candidate != requested_path:
                    # Falling back is expected/handled behavior on sandboxed
                    # hosts (HF Spaces etc.) where /data isn't writable — one
                    # summary line is enough, not a warning per failed try.
                    logger.warning(
                        "[PayoutDatabase] %s not writable — using %s instead",
                        requested_path, candidate,
                    )
                self._db_path = candidate
                return
            except Exception as exc:
                last_exc = exc
                # Individual candidate failures are expected noise during
                # fallback probing, not actionable on their own — the
                # summary line above (or the RuntimeError below, if every
                # candidate fails) carries the real signal.
                logger.debug(
                    "[PayoutDatabase] candidate %s not writable (%s)",
                    candidate.parent, exc,
                )

        raise RuntimeError(
            f"[PayoutDatabase] No writable location found for ledger db "
            f"(tried {[str(c) for c in candidates]}): {last_exc}"
        )

    def _init_schema(self) -> None:
        """Create tables if they don't exist."""
        conn = sqlite3.connect(str(self._db_path))
        try:
            cursor = conn.cursor()

            cursor.execute("""
                CREATE TABLE IF NOT EXISTS payouts (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    ts TEXT NOT NULL,
                    sweep_type TEXT NOT NULL,
                    tx_hash TEXT,
                    chain TEXT NOT NULL,
                    wallet TEXT NOT NULL,
                    net_profit REAL NOT NULL,
                    gross_profit REAL NOT NULL,
                    total_loan_fees REAL NOT NULL,
                    total_gas_cost REAL NOT NULL,
                    buy_count INTEGER NOT NULL,
                    scan_count INTEGER NOT NULL,
                    gas_price_gwei REAL NOT NULL,
                    sweep_amount_wei INTEGER NOT NULL,
                    note TEXT,
                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
                )
            """)

            cursor.execute("""
                CREATE TABLE IF NOT EXISTS gas_price_history (
                    id INTEGER PRIMARY KEY AUTOINCREMENT,
                    timestamp TEXT NOT NULL,
                    chain TEXT NOT NULL,
                    gas_price_gwei REAL NOT NULL,
                    block_number INTEGER,
                    tx_hash TEXT,
                    created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
                )
            """)

            # v18 — ledger_state is now keyed by `chain` instead of a fixed
            # id=1 row, so BSC and ETH (or any future chain) can each hold
            # independent running totals without cross-contaminating the
            # sweep-eligible net_profit. See _migrate_ledger_state_to_chain_key
            # for the one-time migration of pre-v18 single-row databases.
            cursor.execute("""
                CREATE TABLE IF NOT EXISTS ledger_state (
                    chain TEXT PRIMARY KEY,
                    gross_profit REAL NOT NULL DEFAULT 0.0,
                    total_loan_fees REAL NOT NULL DEFAULT 0.0,
                    total_gas_cost REAL NOT NULL DEFAULT 0.0,
                    buy_count INTEGER NOT NULL DEFAULT 0,
                    scan_count INTEGER NOT NULL DEFAULT 0,
                    total_swept_wei INTEGER NOT NULL DEFAULT 0,
                    anomaly_strikes INTEGER NOT NULL DEFAULT 0,
                    last_updated TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
                )
            """)

            self._migrate_ledger_state_to_chain_key(cursor)

            cursor.execute(
                "CREATE INDEX IF NOT EXISTS idx_payouts_ts ON payouts(ts DESC)"
            )
            cursor.execute(
                "CREATE INDEX IF NOT EXISTS idx_gas_history_chain_ts "
                "ON gas_price_history(chain, timestamp DESC)"
            )

            conn.commit()
            logger.debug("[PayoutDatabase] Schema initialized")
        finally:
            conn.close()

    def _migrate_ledger_state_to_chain_key(self, cursor: sqlite3.Cursor) -> None:
        """
        One-time migration: pre-v18 databases have a legacy `ledger_state`
        table with `id INTEGER PRIMARY KEY CHECK (id = 1)`. If that legacy
        shape is detected, copy its single row's totals into chain='BSC'
        under the new schema, since BSC has been the only chain this bot
        has ever run against. Safe to run every startup — it's a no-op
        once migrated because the legacy table won't exist anymore.
        """
        cursor.execute("PRAGMA table_info(ledger_state)")
        cols = {row[1] for row in cursor.fetchall()}
        if "id" not in cols:
            return  # already on the new chain-keyed schema, nothing to do

        logger.warning(
            "[PayoutDatabase] Legacy id=1 ledger_state detected — "
            "migrating single-row totals to chain='BSC' (one-time, non-destructive)."
        )
        cursor.execute("ALTER TABLE ledger_state RENAME TO ledger_state_legacy")
        cursor.execute("""
            CREATE TABLE ledger_state (
                chain TEXT PRIMARY KEY,
                gross_profit REAL NOT NULL DEFAULT 0.0,
                total_loan_fees REAL NOT NULL DEFAULT 0.0,
                total_gas_cost REAL NOT NULL DEFAULT 0.0,
                buy_count INTEGER NOT NULL DEFAULT 0,
                scan_count INTEGER NOT NULL DEFAULT 0,
                total_swept_wei INTEGER NOT NULL DEFAULT 0,
                anomaly_strikes INTEGER NOT NULL DEFAULT 0,
                last_updated TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP
            )
        """)
        cursor.execute("SELECT * FROM ledger_state_legacy WHERE id = 1")
        legacy_row = cursor.fetchone()
        if legacy_row:
            # legacy column order: id, gross_profit, total_loan_fees,
            # total_gas_cost, buy_count, scan_count, total_swept_wei,
            # anomaly_strikes, last_updated
            cursor.execute("""
                INSERT INTO ledger_state (
                    chain, gross_profit, total_loan_fees, total_gas_cost,
                    buy_count, scan_count, total_swept_wei, anomaly_strikes,
                    last_updated
                ) VALUES ('BSC', ?, ?, ?, ?, ?, ?, ?, ?)
            """, legacy_row[1:])
            logger.warning(
                "[PayoutDatabase] Migrated legacy totals into chain='BSC': "
                "history preserved, old table kept as 'ledger_state_legacy' "
                "for audit/rollback."
            )
        cursor.execute("INSERT OR IGNORE INTO ledger_state (chain) VALUES ('BSC')")

    async def save_payout(self, record: PayoutRecord) -> int:
        """Insert a payout record. Returns inserted row ID."""
        async with self._lock:
            return await asyncio.to_thread(self._save_payout_sync, record)

    def _save_payout_sync(self, record: PayoutRecord) -> int:
        """Synchronous payout save."""
        conn = sqlite3.connect(str(self._db_path))
        try:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT INTO payouts (
                    ts, sweep_type, tx_hash, chain, wallet,
                    net_profit, gross_profit, total_loan_fees, total_gas_cost,
                    buy_count, scan_count, gas_price_gwei, sweep_amount_wei, note
                ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """, (
                record.ts, record.sweep_type, record.tx_hash, record.chain,
                record.wallet, record.net_profit, record.gross_profit,
                record.total_loan_fees, record.total_gas_cost, record.buy_count,
                record.scan_count, record.gas_price_gwei, record.sweep_amount_wei,
                record.note,
            ))
            conn.commit()
            return cursor.lastrowid
        finally:
            conn.close()

    async def save_gas_price(self, entry: GasPriceEntry) -> int:
        """Record a gas price reading. Returns inserted row ID."""
        async with self._lock:
            return await asyncio.to_thread(self._save_gas_price_sync, entry)

    def _save_gas_price_sync(self, entry: GasPriceEntry) -> int:
        """Synchronous gas price save."""
        conn = sqlite3.connect(str(self._db_path))
        try:
            cursor = conn.cursor()
            cursor.execute("""
                INSERT INTO gas_price_history (
                    timestamp, chain, gas_price_gwei, block_number, tx_hash
                ) VALUES (?, ?, ?, ?, ?)
            """, (
                entry.timestamp, entry.chain, entry.gas_price_gwei,
                entry.block_number, entry.tx_hash,
            ))
            conn.commit()
            return cursor.lastrowid
        finally:
            conn.close()

    async def get_recent_gas_prices(
        self, chain: str, limit: int = 10
    ) -> List[GasPriceEntry]:
        """Fetch the last N gas price records for a chain."""
        async with self._lock:
            return await asyncio.to_thread(
                self._get_recent_gas_prices_sync, chain, limit
            )

    def _get_recent_gas_prices_sync(
        self, chain: str, limit: int = 10
    ) -> List[GasPriceEntry]:
        """Synchronous gas price fetch."""
        conn = sqlite3.connect(str(self._db_path))
        try:
            cursor = conn.cursor()
            cursor.execute("""
                SELECT timestamp, chain, gas_price_gwei, block_number, tx_hash
                FROM gas_price_history
                WHERE chain = ?
                ORDER BY timestamp DESC
                LIMIT ?
            """, (chain, limit))
            rows = cursor.fetchall()
            return [
                GasPriceEntry(
                    timestamp=row[0],
                    chain=row[1],
                    gas_price_gwei=row[2],
                    block_number=row[3],
                    tx_hash=row[4],
                )
                for row in rows
            ]
        finally:
            conn.close()

    async def get_payout_history(self, limit: int = 10) -> List[dict]:
        """Fetch the last N payout records."""
        async with self._lock:
            return await asyncio.to_thread(
                self._get_payout_history_sync, limit
            )

    def _get_payout_history_sync(self, limit: int = 10) -> List[dict]:
        """Synchronous payout history fetch."""
        conn = sqlite3.connect(str(self._db_path))
        conn.row_factory = sqlite3.Row
        try:
            cursor = conn.cursor()
            cursor.execute("""
                SELECT * FROM payouts
                ORDER BY ts DESC
                LIMIT ?
            """, (limit,))
            return [dict(row) for row in cursor.fetchall()]
        finally:
            conn.close()

    async def get_ledger_state(self, chain: str = "BSC") -> dict[str, Any]:
        """Fetch current ledger state for a specific chain."""
        async with self._lock:
            return await asyncio.to_thread(self._get_ledger_state_sync, chain)

    def _get_ledger_state_sync(self, chain: str) -> dict[str, Any]:
        """Synchronous ledger state fetch, scoped to one chain."""
        conn = sqlite3.connect(str(self._db_path))
        try:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM ledger_state WHERE chain = ?", (chain,))
            row = cursor.fetchone()
            if row:
                return {
                    "chain": row[0],
                    "gross_profit": row[1],
                    "total_loan_fees": row[2],
                    "total_gas_cost": row[3],
                    "buy_count": row[4],
                    "scan_count": row[5],
                    "total_swept_wei": row[6],
                    "anomaly_strikes": row[7],
                    "last_updated": row[8],
                }
            return {
                "chain": chain,
                "gross_profit": 0.0,
                "total_loan_fees": 0.0,
                "total_gas_cost": 0.0,
                "buy_count": 0,
                "scan_count": 0,
                "total_swept_wei": 0,
                "anomaly_strikes": 0,
                "last_updated": datetime.now(timezone.utc).isoformat(),
            }
        finally:
            conn.close()

    async def update_ledger_state(self, updates: dict[str, Any], chain: str = "BSC") -> None:
        """Update ledger state fields for a specific chain."""
        async with self._lock:
            await asyncio.to_thread(self._update_ledger_state_sync, updates, chain)

    def _update_ledger_state_sync(self, updates: dict[str, Any], chain: str) -> None:
        """Synchronous ledger state update, scoped to one chain."""
        conn = sqlite3.connect(str(self._db_path))
        try:
            cursor = conn.cursor()
            cursor.execute(
                "INSERT OR IGNORE INTO ledger_state (chain) VALUES (?)", (chain,)
            )
            allowed_fields = {
                "gross_profit", "total_loan_fees", "total_gas_cost",
                "buy_count", "scan_count", "total_swept_wei",
                "anomaly_strikes", "last_updated",
            }
            set_clause = ", ".join(
                f"{k} = ?" for k in updates.keys() if k in allowed_fields
            )
            values = [updates[k] for k in updates.keys() if k in allowed_fields]
            if set_clause:
                values.append(datetime.now(timezone.utc).isoformat())
                values.append(chain)
                cursor.execute(
                    f"UPDATE ledger_state SET {set_clause}, last_updated = ? WHERE chain = ?",
                    values,
                )
                conn.commit()
        finally:
            conn.close()

    async def reset_ledger(
        self, confirm: bool = False, chain: str = "BSC", full_reset: bool = False,
    ) -> bool:
        """
        Reset running totals for a specific chain (keeps payout history).

        By default only resets the profit/fee/gas/buy_count fields — the
        fields that feed the sweep-eligible net_profit calculation. This
        intentionally leaves total_swept_wei (lifetime swept, an audit
        figure) and anomaly_strikes/scan_count (activity history) untouched,
        since a caller resetting "the ledger" after a sweep is asking to
        zero out accumulated profit, not to erase how much has ever been
        swept or how many anomalies have ever fired.

        Pass full_reset=True to also zero total_swept_wei, anomaly_strikes,
        and scan_count — use this only when you actually want a clean-slate
        row (e.g. decommissioning a wallet), since it discards audit history
        that reset()'s default behavior preserves.
        """
        if not confirm:
            logger.warning("[PayoutDatabase] reset() no-op — pass confirm=True")
            return False
        async with self._lock:
            return await asyncio.to_thread(self._reset_ledger_sync, chain, full_reset)

    def _reset_ledger_sync(self, chain: str, full_reset: bool = False) -> bool:
        """Synchronous ledger reset, scoped to one chain."""
        conn = sqlite3.connect(str(self._db_path))
        try:
            cursor = conn.cursor()
            if full_reset:
                cursor.execute("""
                    UPDATE ledger_state
                    SET gross_profit = 0.0, total_loan_fees = 0.0,
                        total_gas_cost = 0.0, buy_count = 0,
                        scan_count = 0, total_swept_wei = 0,
                        anomaly_strikes = 0,
                        last_updated = CURRENT_TIMESTAMP
                    WHERE chain = ?
                """, (chain,))
                logger.warning(
                    "[PayoutDatabase] FULL ledger reset for chain=%s — "
                    "total_swept_wei/anomaly_strikes/scan_count cleared too.",
                    chain,
                )
            else:
                cursor.execute("""
                    UPDATE ledger_state
                    SET gross_profit = 0.0, total_loan_fees = 0.0,
                        total_gas_cost = 0.0, buy_count = 0,
                        last_updated = CURRENT_TIMESTAMP
                    WHERE chain = ?
                """, (chain,))
            conn.commit()
            logger.info("[PayoutDatabase] Ledger reset complete for chain=%s (full_reset=%s)", chain, full_reset)
            return True
        finally:
            conn.close()

    async def export_ledger_json(self) -> dict[str, Any]:
        """Export entire ledger (all chains) as JSON for backup/audit."""
        async with self._lock:
            return await asyncio.to_thread(self._export_ledger_json_sync)

    def _export_ledger_json_sync(self) -> dict[str, Any]:
        """Synchronous JSON export — all chains, not just one."""
        conn = sqlite3.connect(str(self._db_path))
        conn.row_factory = sqlite3.Row
        try:
            cursor = conn.cursor()
            cursor.execute("SELECT * FROM ledger_state")
            state_rows = cursor.fetchall()
            cursor.execute("SELECT * FROM payouts ORDER BY ts DESC")
            payouts = [dict(row) for row in cursor.fetchall()]
            cursor.execute(
                "SELECT * FROM gas_price_history ORDER BY timestamp DESC LIMIT 100"
            )
            gas_history = [dict(row) for row in cursor.fetchall()]

            return {
                "ledger_state_by_chain": {
                    row["chain"]: dict(row) for row in state_rows
                },
                "payouts": payouts,
                "gas_price_history": gas_history,
                "exported_at": datetime.now(timezone.utc).isoformat(),
            }
        finally:
            conn.close()