from __future__ import annotations

import hashlib
import json
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime, timedelta
from pathlib import Path
from sqlite3 import Connection, SQLITE_DELETE, SQLITE_DENY, SQLITE_INSERT, SQLITE_OK, SQLITE_UPDATE
from typing import Any, Callable

from jarvis_finance.api.schemas.market import QuoteRefreshRequest
from jarvis_finance.audit.log import record_audit_event
from jarvis_finance.market.providers import CoinGeckoClient, refresh_crypto_prices
from jarvis_finance.services.market_service import refresh_equity_quotes_batch
from jarvis_finance.services.modelled_wealth import build_modelled_wealth_development
from jarvis_finance.storage.database import connect

SOURCES = ("equity", "crypto", "fx")
PROTECTED_TABLES = (
    "accounts",
    "transactions",
    "crypto_holdings",
    "positions_snapshot",
    "postfinance_snapshot_positions",
    "truewealth_snapshot_positions",
)


@dataclass(frozen=True)
class SourceRunResult:
    stale_candidates: int = 0
    updated_count: int = 0
    fresh_unchanged_count: int = 0
    stale_remaining_count: int = 0
    failed_count: int = 0
    diagnostics: tuple[str, ...] = field(default_factory=tuple)

    def __iter__(self):
        yield self.stale_candidates
        yield self.updated_count


def _source_result(value: SourceRunResult | tuple[int, int]) -> SourceRunResult:
    if isinstance(value, SourceRunResult):
        return value
    return SourceRunResult(stale_candidates=int(value[0]), updated_count=int(value[1]))


def _deny_protected_dml(
    action: int,
    table: str | None,
    _column: str | None,
    _database: str | None,
    _trigger: str | None,
) -> int:
    if action in {SQLITE_INSERT, SQLITE_UPDATE, SQLITE_DELETE} and table in PROTECTED_TABLES:
        return SQLITE_DENY
    return SQLITE_OK


def _now() -> str:
    return datetime.now(UTC).isoformat()


def _database_path(conn: Connection) -> str:
    row = next((row for row in conn.execute("PRAGMA database_list") if str(row[1]) == "main"), None)
    if not row or not str(row[2] or ""):
        raise ValueError("asset_refresh_requires_persistent_database")
    return str(Path(str(row[2])).resolve())


def _protected_fingerprint(conn: Connection) -> str:
    payload: dict[str, list[dict[str, Any]]] = {}
    available = {
        str(row[0])
        for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
    }
    for table in PROTECTED_TABLES:
        if table not in available:
            continue
        rows = conn.execute(f'SELECT * FROM "{table}" ORDER BY rowid').fetchall()
        payload[table] = [dict(row) for row in rows]
    return hashlib.sha256(
        json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
    ).hexdigest()


def _status_payload(conn: Connection, job_id: str) -> dict[str, Any]:
    job = conn.execute("SELECT * FROM asset_price_refresh_jobs WHERE job_id=?", (job_id,)).fetchone()
    if not job:
        raise ValueError("asset_price_refresh_job_not_found")
    sources = [
        dict(row)
        for row in conn.execute(
            "SELECT * FROM asset_price_refresh_sources WHERE job_id=? ORDER BY CASE source WHEN 'equity' THEN 1 WHEN 'crypto' THEN 2 ELSE 3 END",
            (job_id,),
        ).fetchall()
    ]
    return {
        "job_id": str(job["job_id"]),
        "status": str(job["status"]),
        "requested_at": str(job["requested_at"]),
        "completed_at": str(job["completed_at"]) if job["completed_at"] else None,
        "stale_before": str(job["stale_before"]),
        "progress": {"completed": int(job["progress_completed"]), "total": int(job["progress_total"])},
        "sources": [
            {
                "source": str(row["source"]),
                "status": str(row["status"]),
                "stale_candidates": int(row["stale_candidates"]),
                "updated_count": int(row["updated_count"]),
                "error_code": str(row["error_code"]) if row["error_code"] else None,
                "fresh_unchanged_count": int(row["fresh_unchanged_count"]),
                "stale_remaining_count": int(row["stale_remaining_count"]),
                "failed_count": int(row["failed_count"]),
                "diagnostics": json.loads(str(row["diagnostics_json"] or "[]")),
                "started_at": str(row["started_at"]) if row["started_at"] else None,
                "completed_at": str(row["completed_at"]) if row["completed_at"] else None,
            }
            for row in sources
        ],
        "wealth_snapshot_created": bool(job["wealth_snapshot_id"]),
        "audit_recorded": bool(job["audit_id"]),
        "successful_assets": sum(int(row["updated_count"]) for row in sources),
        "fresh_unchanged_assets": sum(int(row["fresh_unchanged_count"]) for row in sources),
        "stale_assets": sum(int(row["stale_remaining_count"]) for row in sources),
        "failed_assets": sum(int(row["failed_count"]) for row in sources),
        "next_action": (
            "Diagnose prüfen und nur betroffene Quelle erneut versuchen."
            if any(int(row["failed_count"]) or int(row["stale_remaining_count"]) for row in sources)
            else "Keine Aktion nötig; alle verfügbaren Kurse sind aktuell."
        ),
        "provider_calls_on_read": False,
    }


def create_asset_price_refresh_job(conn: Connection, *, stale_hours: int = 24) -> tuple[dict[str, Any], str]:
    """Persist a queued job only. No provider call occurs before the HTTP response."""
    if conn.in_transaction:
        raise ValueError("asset_refresh_requires_clean_transaction")
    conn.execute("BEGIN IMMEDIATE")
    try:
        if conn.execute(
            "SELECT 1 FROM asset_price_refresh_jobs WHERE status IN ('queued','running') LIMIT 1"
        ).fetchone():
            raise ValueError("asset_price_refresh_job_already_running")
        now = datetime.now(UTC)
        job_id = f"asset-refresh-{uuid.uuid4().hex}"
        stale_before = (now - timedelta(hours=max(1, min(stale_hours, 720)))).isoformat()
        conn.execute(
            """INSERT INTO asset_price_refresh_jobs(
                 job_id,status,requested_at,stale_before,progress_total,progress_completed
               ) VALUES(?,'queued',?,?,3,0)""",
            (job_id, now.isoformat(), stale_before),
        )
        conn.executemany(
            """INSERT INTO asset_price_refresh_sources(
                 job_id,source,status,stale_candidates,updated_count
               ) VALUES(?,?,'pending',0,0)""",
            [(job_id, source) for source in SOURCES],
        )
        conn.commit()
    except Exception:
        if conn.in_transaction:
            conn.rollback()
        raise
    return _status_payload(conn, job_id), _database_path(conn)


def _equity_source(conn: Connection, stale_before: str) -> SourceRunResult:
    response = refresh_equity_quotes_batch(
        conn,
        QuoteRefreshRequest(
            provider="auto",
            only_missing=True,
            stale_before=stale_before,
            limit=500,
            max_retries=1,
            pacing_seconds=0.15,
        ),
    )
    candidates = max(0, int(response.total) - int(response.cached))
    failed = sum(1 for row in response.results if str(row.get("status")) in {"provider_error", "error"})
    stale = sum(1 for row in response.results if str(row.get("status")) in {"stale", "missing"})
    valuation_issues = [
        warning
        for warning in response.warnings
        if warning in {"portfolio_valuation_partial", "portfolio_valuation_failed"}
    ]
    return SourceRunResult(
        stale_candidates=candidates,
        updated_count=int(response.economic_updated),
        fresh_unchanged_count=int(response.cached + response.updated - response.economic_updated),
        stale_remaining_count=stale,
        failed_count=max(failed, len(response.errors)) + len(valuation_issues),
        diagnostics=tuple(sorted(set([*response.errors, *valuation_issues])))[:10],
    )


def _crypto_source(conn: Connection, stale_before: str) -> SourceRunResult:
    held_asset_ids = [
        str(row["asset_id"])
        for row in conn.execute(
            """SELECT a.asset_id FROM crypto_assets a
                 WHERE a.is_active=1 AND EXISTS(
                   SELECT 1 FROM crypto_holdings h WHERE h.asset_id=a.asset_id AND CAST(h.quantity AS REAL)<>0
                 ) ORDER BY a.asset_id"""
        ).fetchall()
    ]
    stale_asset_ids = [
        str(row["asset_id"])
        for row in conn.execute(
            """SELECT a.asset_id FROM crypto_assets a
                 WHERE a.is_active=1 AND EXISTS(
                   SELECT 1 FROM crypto_holdings h WHERE h.asset_id=a.asset_id AND CAST(h.quantity AS REAL)<>0
                 ) AND NOT EXISTS(
                   SELECT 1 FROM crypto_prices p
                    WHERE p.asset_id=a.asset_id AND p.fetched_at>=?
                      AND p.quality_status='fresh' AND p.price IS NOT NULL
                 ) ORDER BY a.asset_id""",
            (stale_before,),
        ).fetchall()
    ]
    if not stale_asset_ids:
        return SourceRunResult(fresh_unchanged_count=len(held_asset_ids))
    cutoff = datetime.fromisoformat(stale_before.replace("Z", "+00:00"))
    if cutoff.tzinfo is None:
        cutoff = cutoff.replace(tzinfo=UTC)
    max_age_seconds = max(1, int((datetime.now(UTC) - cutoff.astimezone(UTC)).total_seconds()))
    result = refresh_crypto_prices(
        conn,
        provider=CoinGeckoClient(),
        currency="CHF",
        max_age_seconds=max_age_seconds,
        asset_ids=stale_asset_ids,
        batch_size=100,
    )
    return SourceRunResult(
        stale_candidates=len(stale_asset_ids),
        updated_count=int(result.updated_count),
        fresh_unchanged_count=max(0, len(held_asset_ids) - len(stale_asset_ids)) + int(result.cached_count),
        stale_remaining_count=int(result.stale_count + result.missing_local_price_count),
        failed_count=int(result.error_count),
        diagnostics=tuple(result.errors[:10]),
    )


def _fx_source(conn: Connection, stale_before: str) -> SourceRunResult:
    from jarvis_finance.fx.providers import FrankfurterFxProvider, TwelveDataFxProvider
    from jarvis_finance.fx.rates import resolve_fx_rate_to_chf

    cutoff_date = stale_before[:10]
    all_currencies = [
        str(row["currency"]).upper()
        for row in conn.execute(
            """SELECT DISTINCT upper(i.currency) currency
                 FROM instruments i
                WHERE i.is_active=1 AND upper(COALESCE(i.currency,'CHF'))!='CHF'
                ORDER BY currency"""
        ).fetchall()
    ]
    currencies = [
        str(row["currency"]).upper()
        for row in conn.execute(
            """SELECT DISTINCT upper(i.currency) currency
                 FROM instruments i
                WHERE i.is_active=1 AND upper(COALESCE(i.currency,'CHF'))!='CHF'
                  AND NOT EXISTS(
                    SELECT 1 FROM fx_rates f
                     WHERE f.base_currency=upper(i.currency) AND f.quote_currency='CHF'
                       AND f.rate_date>=? AND f.quality_status IN ('fresh','ok')
                  )
                ORDER BY currency""",
            (cutoff_date,),
        ).fetchall()
    ]
    updated = 0
    provider_unchanged = 0
    failures = 0
    for currency in currencies:
        try:
            before = conn.execute(
                """SELECT rate_date,rate,provider,rate_type,quality_status
                     FROM fx_rates
                    WHERE base_currency=? AND quote_currency='CHF'
                    ORDER BY rate_date DESC,COALESCE(fetched_at,created_at) DESC LIMIT 1""",
                (currency,),
            ).fetchone()
            before_economic = tuple(before) if before is not None else None
            result = resolve_fx_rate_to_chf(
                conn,
                base_currency=currency,
                rate_date=None,
                providers=[FrankfurterFxProvider(), TwelveDataFxProvider()],
                persist=True,
                resolve_fixed=True,
            )
            after = conn.execute(
                """SELECT rate_date,rate,provider,rate_type,quality_status
                     FROM fx_rates
                    WHERE base_currency=? AND quote_currency='CHF'
                    ORDER BY rate_date DESC,COALESCE(fetched_at,created_at) DESC LIMIT 1""",
                (currency,),
            ).fetchone()
            after_economic = tuple(after) if after is not None else None
            if result.status == "ok" and after_economic != before_economic:
                updated += 1
            elif result.status == "ok":
                provider_unchanged += 1
        except Exception:
            failures += 1
    conn.commit()
    return SourceRunResult(
        stale_candidates=len(currencies),
        updated_count=updated,
        fresh_unchanged_count=max(0, len(all_currencies) - len(currencies)) + provider_unchanged,
        stale_remaining_count=failures,
        failed_count=failures,
    )


DEFAULT_RUNNERS: dict[str, Callable[[Connection, str], SourceRunResult | tuple[int, int]]] = {
    "equity": _equity_source,
    "crypto": _crypto_source,
    "fx": _fx_source,
}


def run_asset_price_refresh(
    db_path: str,
    job_id: str,
    *,
    runners: dict[str, Callable[[Connection, str], SourceRunResult | tuple[int, int]]] | None = None,
) -> None:
    """Background worker with source isolation, stored progress and mutation guard."""
    conn = connect(db_path)
    selected = runners or DEFAULT_RUNNERS
    try:
        conn.execute("BEGIN IMMEDIATE")
        claimed = conn.execute(
            """UPDATE asset_price_refresh_jobs
                  SET status='running'
                WHERE job_id=? AND status='queued'""",
            (job_id,),
        ).rowcount
        conn.commit()
        if claimed != 1:
            return
        stale_before = str(
            conn.execute(
                "SELECT stale_before FROM asset_price_refresh_jobs WHERE job_id=?",
                (job_id,),
            ).fetchone()[0]
        )
        protected_before = _protected_fingerprint(conn)
        conn.set_authorizer(_deny_protected_dml)
        completed = 0
        failures = 0
        total_updated = 0
        total_fresh_unchanged = 0
        total_candidates = 0
        for source in SOURCES:
            started = _now()
            conn.execute(
                "UPDATE asset_price_refresh_sources SET status='running',started_at=? WHERE job_id=? AND source=?",
                (started, job_id, source),
            )
            conn.commit()
            result = SourceRunResult()
            status = "complete"
            error_code = None
            try:
                result = _source_result(selected[source](conn, stale_before))
                if result.stale_candidates == 0 and result.fresh_unchanged_count == 0:
                    status = "skipped"
                if result.failed_count:
                    failures += 1
                    status = "failed"
                    error_code = f"{source}_instrument_failures"
                elif result.stale_remaining_count:
                    failures += 1
                    status = "failed"
                    error_code = f"{source}_stale_remaining"
            except Exception as exc:
                if conn.in_transaction:
                    conn.rollback()
                status = "failed"
                failures += 1
                result = SourceRunResult(failed_count=1, diagnostics=(type(exc).__name__,))
                error_code = f"{source}_refresh_failed"
            total_updated += result.updated_count
            total_fresh_unchanged += result.fresh_unchanged_count
            total_candidates += result.stale_candidates
            completed += 1
            conn.execute(
                """UPDATE asset_price_refresh_sources
                      SET status=?,stale_candidates=?,updated_count=?,error_code=?,completed_at=?,
                          fresh_unchanged_count=?,stale_remaining_count=?,failed_count=?,diagnostics_json=?
                    WHERE job_id=? AND source=?""",
                (
                    status, result.stale_candidates, result.updated_count, error_code, _now(),
                    result.fresh_unchanged_count, result.stale_remaining_count, result.failed_count,
                    json.dumps(list(result.diagnostics)), job_id, source,
                ),
            )
            conn.execute(
                "UPDATE asset_price_refresh_jobs SET progress_completed=? WHERE job_id=?",
                (completed, job_id),
            )
            conn.commit()
        if _protected_fingerprint(conn) != protected_before:
            raise RuntimeError("protected_holdings_or_transactions_mutated")

        usable_result = total_updated > 0 or total_fresh_unchanged > 0 or (total_candidates == 0 and failures == 0)
        wealth_snapshot_id = None
        if total_updated > 0:
            model = build_modelled_wealth_development(conn, period="1m")
            current = model.get("current") or {}
            wealth_snapshot_id = f"wealth-refresh-{uuid.uuid4().hex}"
            source_rows = [
                dict(row)
                for row in conn.execute(
                    """SELECT source,status,stale_candidates,updated_count,error_code,
                              fresh_unchanged_count,stale_remaining_count,failed_count
                         FROM asset_price_refresh_sources WHERE job_id=? ORDER BY source""",
                    (job_id,),
                ).fetchall()
            ]
            conn.execute(
                """INSERT INTO aggregated_wealth_refresh_snapshots(
                     wealth_snapshot_id,job_id,captured_at,known_wealth_chf,quality_status,source_status_json
                   ) VALUES(?,?,?,?,?,?)""",
                (
                    wealth_snapshot_id,
                    job_id,
                    _now(),
                    current.get("value_chf"),
                    "complete" if failures == 0 else "partial",
                    json.dumps(source_rows, sort_keys=True),
                ),
            )
        final_status = "complete" if failures == 0 else "partial" if usable_result else "failed"
        audit_id = record_audit_event(
            conn,
            source="asset_price_refresh_job_v1",
            action="asset_prices_refresh_completed",
            entity_type="asset_price_refresh_job",
            entity_id=job_id,
            old_values={},
            new_values={
                "status": final_status,
                "source_count": len(SOURCES),
                "failed_source_count": failures,
                "wealth_snapshot_created": bool(wealth_snapshot_id),
                "holdings_mutated": False,
                "transactions_mutated": False,
                "trades_created": 0,
            },
            created_by="system",
        )
        conn.execute(
            """UPDATE asset_price_refresh_jobs
                  SET status=?,completed_at=?,wealth_snapshot_id=?,audit_id=?
                WHERE job_id=?""",
            (final_status, _now(), wealth_snapshot_id, audit_id, job_id),
        )
        conn.commit()
    except Exception as exc:
        if conn.in_transaction:
            conn.rollback()
        audit_id = record_audit_event(
            conn,
            source="asset_price_refresh_job_v1",
            action="asset_prices_refresh_failed",
            entity_type="asset_price_refresh_job",
            entity_id=job_id,
            old_values={},
            new_values={"status": "failed", "error_code": str(exc)[:120]},
            created_by="system",
        )
        conn.execute(
            "UPDATE asset_price_refresh_jobs SET status='failed',completed_at=?,audit_id=? WHERE job_id=?",
            (_now(), audit_id, job_id),
        )
        conn.commit()
    finally:
        conn.close()


def asset_price_refresh_status(conn: Connection, job_id: str) -> dict[str, Any]:
    """Stored status only: no provider call, write or lazy refresh."""
    return _status_payload(conn, job_id)