from __future__ import annotations

import fcntl
import importlib
import sqlite3
import threading
from datetime import UTC, datetime
from decimal import Decimal
from pathlib import Path

import pytest

from jarvis_finance.market.providers import PriceQuote
from jarvis_finance.services.crypto_market_recovery import (
    SOURCE_KEY,
    activate_crypto_market_source,
    build_crypto_market_dry_run,
    is_crypto_market_source_activated,
    run_crypto_market_one_shot,
)
from jarvis_finance.services.performance_activation import build_daily_valuation_job_status
from jarvis_finance.storage.database import connect, connect_memory, connect_read_only
from jarvis_finance.storage.migrations import MIGRATION_VERSION, apply_migrations

NOW = datetime(2026, 8, 3, 20, 0, tzinfo=UTC)
DAY = "2026-08-03"


class BatchProvider:
    provider_key = "coingecko"

    def __init__(self, *, missing: set[str] | None = None, stale: set[str] | None = None) -> None:
        self.missing = missing or set()
        self.stale = stale or set()
        self.calls: list[tuple[tuple[str, ...], str]] = []

    def get_crypto_price(self, coingecko_id: str, currency: str = "CHF") -> PriceQuote:
        return self.get_crypto_prices([coingecko_id], currency)[coingecko_id]

    def get_crypto_prices(self, coingecko_ids, currency: str = "CHF"):
        ids = tuple(coingecko_ids)
        self.calls.append((ids, currency))
        result = {}
        for index, provider_id in enumerate(ids, 1):
            if provider_id in self.missing:
                result[provider_id] = PriceQuote(provider_id, currency, None, quality_status="missing", error_message="missing")
            else:
                ts = "2026-08-03T14:00:00+00:00" if provider_id in self.stale else "2026-08-03T19:55:00+00:00"
                result[provider_id] = PriceQuote(provider_id, currency, Decimal(str(index * 100)), provider_timestamp=ts)
        return result


def db(*, missing_mapping: bool = False):
    conn = connect_memory()
    apply_migrations(conn)
    conn.execute(
        """INSERT INTO crypto_assets(asset_id,coin_name,symbol,coingecko_id,is_active,created_at)
           VALUES('btc','Bitcoin','BTC','bitcoin',1,'2026-01-01'),
                 ('eth','Ethereum','ETH',?,1,'2026-01-01')""",
        (None if missing_mapping else "ethereum",),
    )
    conn.execute(
        """INSERT INTO crypto_wallets(wallet_id,wallet_name,wallet_type,created_at)
           VALUES('wallet-a','Wallet A','exchange','2026-01-01'),
                 ('wallet-b','Wallet B','hardware','2026-01-01')"""
    )
    conn.execute(
        """INSERT INTO crypto_holdings(
               crypto_holding_id,asset_id,wallet_id,quantity,last_verified_at,
               verification_status,created_at
           ) VALUES('h1','btc','wallet-a','0.5','2025-12-31','verified','2026-01-01'),
                   ('h2','btc','wallet-b','0.5','2025-12-31','verified','2026-01-01'),
                   ('h3','eth','wallet-a','2','2025-12-31','verified','2026-01-01')"""
    )
    conn.commit()
    return conn


def test_live_dry_run_uses_canonical_ids_values_every_recorded_holding_and_writes_nothing() -> None:
    conn = db()
    provider = BatchProvider()
    before = conn.total_changes

    result = build_crypto_market_dry_run(conn, provider=provider, now=NOW)

    assert result["status"] == "complete"
    assert result["asset_count"] == result["priced_count"] == 2
    assert result["missing_count"] == 0
    assert result["total_value_chf"] == "500.0"
    assert provider.calls == [(('bitcoin', 'ethereum'), 'CHF')]
    assert result["planned_writes"] == {
        "crypto_prices": 2,
        "fx_rates": 0,
        "portfolio_valuation_snapshots": 1,
        "market_data_runs": 1,
        "audit_log": 1,
    }
    assert result["persistence_performed"] is False
    assert conn.total_changes == before
    assert conn.execute("SELECT COUNT(*) FROM crypto_prices").fetchone()[0] == 0


def test_unapproved_provider_is_blocked_before_network_call() -> None:
    provider = BatchProvider()
    provider.provider_key = "other"
    result = build_crypto_market_dry_run(db(), provider=provider, now=NOW)
    assert result["status"] == "blocked"
    assert "crypto_provider_not_approved" in result["reason_codes"]
    assert provider.calls == []


@pytest.mark.parametrize(
    ("quote", "reason"),
    [
        (
            PriceQuote("ethereum", "CHF", Decimal("100"), provider_timestamp="2026-08-03T19:55:00+00:00"),
            "crypto_provider_identity_mismatch",
        ),
        (
            PriceQuote("bitcoin", "USD", Decimal("100"), provider_timestamp="2026-08-03T19:55:00+00:00"),
            "crypto_provider_currency_mismatch",
        ),
        (
            PriceQuote(
                "bitcoin",
                "CHF",
                Decimal("100"),
                provider="NotCoinGecko",
                provider_timestamp="2026-08-03T19:55:00+00:00",
            ),
            "crypto_provider_not_approved",
        ),
    ],
)
def test_quote_identity_currency_and_provider_mismatch_fail_closed(
    quote: PriceQuote, reason: str
) -> None:
    class MismatchedQuoteProvider(BatchProvider):
        def get_crypto_prices(self, coingecko_ids, currency: str = "CHF"):
            result = super().get_crypto_prices(coingecko_ids, currency)
            result["bitcoin"] = quote
            return result

    result = build_crypto_market_dry_run(db(), provider=MismatchedQuoteProvider(), now=NOW)
    assert result["status"] == "partial"
    assert result["total_value_chf"] is None
    assert result["planned_writes"]["crypto_prices"] == 0
    assert reason in result["reason_codes"]


def test_missing_mapping_blocks_complete_total_without_affecting_other_sources() -> None:
    conn = db(missing_mapping=True)
    result = build_crypto_market_dry_run(conn, provider=BatchProvider(), now=NOW)
    assert result["status"] in {"partial", "blocked"}
    assert result["total_value_chf"] is None
    assert "crypto_provider_mapping_missing" in result["reason_codes"]
    assert result["planned_writes"]["crypto_prices"] == 0


def test_missing_or_old_quote_never_gets_stored_as_current(tmp_path: Path) -> None:
    conn = db()
    stale = run_crypto_market_one_shot(
        conn,
        provider=BatchProvider(stale={"ethereum"}),
        as_of=DAY,
        now=NOW,
        lock_path=tmp_path / "stale.lock",
    )
    assert stale.status == "partial"
    assert "crypto_provider_timestamp_stale" in stale.reason_codes
    assert conn.execute("SELECT COUNT(*) FROM crypto_prices").fetchone()[0] == 0
    assert conn.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots").fetchone()[0] == 0
    status = conn.execute("SELECT status,price_stored,valuation_stored FROM market_data_runs WHERE source_key=?", (SOURCE_KEY,)).fetchone()
    assert tuple(status) == ("partial", 0, 0)
    assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE source=?", (SOURCE_KEY,)).fetchone()[0] == 1


def test_productive_one_shot_is_atomic_canonical_and_idempotent(tmp_path: Path) -> None:
    conn = db()
    first = run_crypto_market_one_shot(
        conn,
        provider=BatchProvider(),
        as_of=DAY,
        now=NOW,
        lock_path=tmp_path / "one-shot.lock",
    )
    second_provider = BatchProvider()
    second = run_crypto_market_one_shot(
        conn,
        provider=second_provider,
        as_of=DAY,
        now=NOW,
        lock_path=tmp_path / "one-shot.lock",
    )
    assert first.status == "complete" and first.price_stored == 2 and first.valuation_stored == 1
    assert second.idempotent is True and second.run_id == first.run_id
    assert second_provider.calls == []
    assert conn.execute("SELECT COUNT(*) FROM crypto_prices").fetchone()[0] == 2
    stored_timestamps = {row[0] for row in conn.execute("SELECT provider_timestamp FROM crypto_prices")}
    assert stored_timestamps == {"2026-08-03T19:55:00+00:00"}
    assert conn.execute("SELECT COUNT(*) FROM market_data_runs WHERE source_key=?", (SOURCE_KEY,)).fetchone()[0] == 1
    assert conn.execute("SELECT COUNT(*) FROM portfolio_valuation_snapshots WHERE source=?", (SOURCE_KEY,)).fetchone()[0] == 1
    assert conn.execute("SELECT COUNT(*) FROM audit_log WHERE source=?", (SOURCE_KEY,)).fetchone()[0] == 1
    assert conn.execute("SELECT COUNT(*) FROM crypto_holdings").fetchone()[0] == 3
    assert conn.execute("SELECT COUNT(*) FROM crypto_transactions").fetchone()[0] == 0


def test_current_holdings_cannot_be_backdated(tmp_path: Path) -> None:
    with pytest.raises(ValueError, match="cannot backdate"):
        run_crypto_market_one_shot(
            db(),
            provider=BatchProvider(),
            as_of="2026-01-01",
            now=NOW,
            lock_path=tmp_path / "backdate.lock",
        )


def test_parallel_run_is_rejected(tmp_path: Path) -> None:
    lock = tmp_path / "parallel.lock"
    lock.touch()
    with lock.open("a+") as handle:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        with pytest.raises(RuntimeError, match="market_job_already_running"):
            run_crypto_market_one_shot(db(), provider=BatchProvider(), as_of=DAY, now=NOW, lock_path=lock)


def test_market_source_activation_is_separate_audited_and_idempotent() -> None:
    conn = db()
    assert is_crypto_market_source_activated(conn) is False
    first = activate_crypto_market_source(conn, confirmation_id="crypto-market-activation-1")
    second = activate_crypto_market_source(conn, confirmation_id="crypto-market-activation-1")
    assert first["idempotent"] is False and second["idempotent"] is True
    assert is_crypto_market_source_activated(conn) is True
    assert conn.execute("SELECT COUNT(*) FROM performance_scope_classifications").fetchone()[0] == 0
    assert conn.execute("SELECT COUNT(*) FROM performance_cashflow_coverage").fetchone()[0] == 0


def test_daily_path_requires_source_activation_but_manual_one_shot_does_not(tmp_path: Path) -> None:
    conn = db()
    blocked = run_crypto_market_one_shot(
        conn,
        provider=BatchProvider(),
        as_of=DAY,
        now=NOW,
        lock_path=tmp_path / "daily.lock",
        require_activation=True,
    )
    assert blocked.status == "not_activated"
    assert conn.execute("SELECT COUNT(*) FROM crypto_prices").fetchone()[0] == 0


def test_existing_manual_run_does_not_bypass_daily_source_activation(tmp_path: Path) -> None:
    conn = db()
    manual = run_crypto_market_one_shot(
        conn,
        provider=BatchProvider(),
        as_of=DAY,
        now=NOW,
        lock_path=tmp_path / "manual.lock",
    )
    assert manual.status == "complete"
    provider = BatchProvider()
    scheduled = run_crypto_market_one_shot(
        conn,
        provider=provider,
        as_of=DAY,
        now=NOW,
        lock_path=tmp_path / "manual.lock",
        require_activation=True,
    )
    assert scheduled.status == "not_activated"
    assert provider.calls == []


@pytest.mark.parametrize(
    "mutation",
    [
        "UPDATE crypto_holdings SET quantity='9' WHERE crypto_holding_id='h3'",
        "UPDATE crypto_assets SET coingecko_id='ethereum-classic' WHERE asset_id='eth'",
        "UPDATE crypto_prices SET price='999' WHERE asset_id='eth'",
    ],
)
def test_same_day_replay_fails_closed_on_consumed_input_drift(
    tmp_path: Path, mutation: str
) -> None:
    conn = db()
    first = run_crypto_market_one_shot(
        conn,
        provider=BatchProvider(),
        as_of=DAY,
        now=NOW,
        lock_path=tmp_path / "drift.lock",
    )
    assert first.status == "complete"
    conn.execute(mutation)
    conn.commit()
    provider = BatchProvider()
    with pytest.raises(RuntimeError, match="crypto_existing_run_input_drift"):
        run_crypto_market_one_shot(
            conn,
            provider=provider,
            as_of=DAY,
            now=NOW,
            lock_path=tmp_path / "drift.lock",
        )
    assert provider.calls == []
    assert conn.execute(
        "SELECT COUNT(*) FROM market_data_runs WHERE source_key=?", (SOURCE_KEY,)
    ).fetchone()[0] == 1


def test_source_activation_is_idempotent_under_concurrency(tmp_path: Path) -> None:
    path = tmp_path / "activation.sqlite3"
    seed = connect(path)
    apply_migrations(seed)
    seed.close()
    results: list[dict] = []

    def worker() -> None:
        conn = connect(path)
        try:
            results.append(activate_crypto_market_source(conn, confirmation_id="sprint20e-prod"))
        finally:
            conn.close()

    threads = [threading.Thread(target=worker) for _ in range(2)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join()
    check = connect(path)
    try:
        assert check.execute("SELECT COUNT(*) FROM audit_log WHERE source='market_source_activation_v1'").fetchone()[0] == 1
    finally:
        check.close()
    assert sorted(result["idempotent"] for result in results) == [False, True]


def test_live_read_only_connection_sees_committed_wal_rows(tmp_path: Path) -> None:
    path = tmp_path / "wal.sqlite3"
    writer = sqlite3.connect(path)
    writer.execute("CREATE TABLE sample(value INTEGER)")
    writer.commit()
    writer.execute("PRAGMA journal_mode=WAL")
    writer.execute("PRAGMA wal_autocheckpoint=0")
    writer.execute("INSERT INTO sample VALUES(1)")
    writer.commit()
    live_reader = connect_read_only(path, immutable=False)
    stale_reader = connect_read_only(path, immutable=True)
    try:
        assert live_reader.execute("SELECT COUNT(*) FROM sample").fetchone()[0] == 1
        assert stale_reader.execute("SELECT COUNT(*) FROM sample").fetchone()[0] == 0
    finally:
        live_reader.close()
        stale_reader.close()
        writer.close()


def test_partial_attempt_reports_fetched_and_missing_assets_from_same_run(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    conn = db()
    activate_crypto_market_source(conn, confirmation_id="partial-status")
    result = run_crypto_market_one_shot(
        conn,
        provider=BatchProvider(missing={"ethereum"}),
        as_of=DAY,
        now=NOW,
        lock_path=tmp_path / "partial.lock",
    )
    assert result.status == "partial"
    monkeypatch.setenv("JARVIS_FINANCE_DAILY_VALUATION_ENABLED", "1")
    status = next(
        source
        for source in build_daily_valuation_job_status(conn)["sources"]
        if source["source_key"] == SOURCE_KEY
    )
    assert status["operational_status"] == "partial"
    assert status["valued_assets"] == 1
    assert status["missing_assets"] == 1
    assert status["last_successful_at"] is None
    assert status["last_run_at"] is not None


def test_disabled_timer_is_reported_as_paused_not_unknown_data_error(monkeypatch: pytest.MonkeyPatch) -> None:
    conn = db()
    monkeypatch.setenv("JARVIS_FINANCE_DAILY_VALUATION_ENABLED", "0")
    source = next(
        item
        for item in build_daily_valuation_job_status(conn)["sources"]
        if item["source_key"] == SOURCE_KEY
    )
    assert source["operational_status"] == "paused"
    assert source["source_enabled"] is False
    assert source["next_action"] == "Quelle kontrolliert aktivieren"
    assert source["last_successful_at"] is None


def test_enabled_source_awaiting_first_run_is_active_not_paused(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    conn = db()
    activate_crypto_market_source(conn, confirmation_id="awaiting-first-run")
    monkeypatch.setenv("JARVIS_FINANCE_DAILY_VALUATION_ENABLED", "1")

    source = next(
        item
        for item in build_daily_valuation_job_status(conn)["sources"]
        if item["source_key"] == SOURCE_KEY
    )

    assert source["source_enabled"] is True
    assert source["status"] == "never_run"
    assert source["operational_status"] == "active"
    assert source["next_action"] == "Ersten abgesicherten Krypto-Lauf ausführen"


def test_complete_status_reports_price_and_fx_provenance(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    conn = db()
    activate_crypto_market_source(conn, confirmation_id="provider-status")
    result = run_crypto_market_one_shot(
        conn,
        provider=BatchProvider(),
        as_of=DAY,
        now=NOW,
        lock_path=tmp_path / "provider-status.lock",
    )
    assert result.status == "complete"
    monkeypatch.setenv("JARVIS_FINANCE_DAILY_VALUATION_ENABLED", "1")

    status = next(
        source
        for source in build_daily_valuation_job_status(conn)["sources"]
        if source["source_key"] == SOURCE_KEY
    )

    assert status["price_provider"] == "CoinGecko"
    assert status["price_currency"] == "CHF"
    assert status["fx_provider"] == "Direkte CHF-Notierung (kein FX-Lauf)"


def test_status_hides_all_provenance_if_one_bound_price_timestamp_is_missing(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    conn = db()
    activate_crypto_market_source(conn, confirmation_id="incomplete-provider-status")
    run_crypto_market_one_shot(
        conn,
        provider=BatchProvider(),
        as_of=DAY,
        now=NOW,
        lock_path=tmp_path / "incomplete-provider-status.lock",
    )
    conn.execute("UPDATE crypto_prices SET provider_timestamp=NULL WHERE asset_id='eth'")
    conn.commit()
    monkeypatch.setenv("JARVIS_FINANCE_DAILY_VALUATION_ENABLED", "1")

    status = next(
        source
        for source in build_daily_valuation_job_status(conn)["sources"]
        if source["source_key"] == SOURCE_KEY
    )

    assert status["price_as_of"] is None
    assert status["price_provider"] is None
    assert status["price_currency"] is None
    assert status["fx_provider"] is None


def test_daily_dispatcher_selects_crypto_and_truewealth_sources(
    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
    path = tmp_path / "dispatcher.sqlite3"
    conn = connect(path)
    apply_migrations(conn)
    conn.close()
    module = importlib.import_module("jarvis_finance.cli.main")
    selected: list[str] = []

    def capture(workers):
        selected.extend(name for name, _worker in workers)
        return []

    monkeypatch.setenv("JARVIS_FINANCE_DAILY_VALUATION_ENABLED", "1")
    monkeypatch.setenv("JARVIS_FINANCE_DB_PATH", str(path))
    monkeypatch.setenv("JARVIS_FINANCE_RUNTIME_DIR", str(tmp_path / "runtime"))
    monkeypatch.setattr(module, "run_isolated_daily_sources", capture)
    assert module.main(["run-daily-market-valuation"]) == 0
    assert selected == ["crypto", "truewealth"]


def test_schema_remains_49() -> None:
    assert MIGRATION_VERSION == 50