from __future__ import annotations

from datetime import UTC, datetime, timedelta
from decimal import Decimal
from pathlib import Path
from sqlite3 import Connection
import threading
from types import SimpleNamespace
from typing import Callable

from jarvis_finance.market.providers import PriceQuote
from jarvis_finance.fx.rates import upsert_fx_rate
from jarvis_finance.services import asset_price_refresh as asset_refresh
from jarvis_finance.services.asset_price_refresh import (
    asset_price_refresh_status,
    create_asset_price_refresh_job,
    run_asset_price_refresh,
)
from jarvis_finance.storage.database import connect
from jarvis_finance.storage.migrations import apply_migrations


def database(path: Path):
    conn = connect(path)
    apply_migrations(conn)
    return conn


def test_refresh_job_is_queued_then_isolates_source_failure_and_creates_one_snapshot(tmp_path):
    path = tmp_path / "finance.sqlite3"
    conn = database(path)
    queued, db_path = create_asset_price_refresh_job(conn, stale_hours=12)
    assert queued["status"] == "queued"
    assert all(row["status"] == "pending" for row in queued["sources"])
    assert queued["provider_calls_on_read"] is False
    job_id = queued["job_id"]
    conn.close()

    calls: list[str] = []
    cutoffs: list[str] = []

    def success(source: str):
        def runner(_conn, _stale_before):
            calls.append(source)
            cutoffs.append(_stale_before)
            return 2, 1
        return runner

    def failure(_conn, _stale_before):
        calls.append("crypto")
        cutoffs.append(_stale_before)
        _conn.execute("UPDATE transactions SET net_amount_chf='999.00'")
        raise AssertionError("protected-table authorizer should reject before this line")

    run_asset_price_refresh(
        db_path,
        job_id,
        runners={"equity": success("equity"), "crypto": failure, "fx": success("fx")},
    )

    conn = connect(path)
    status = asset_price_refresh_status(conn, job_id)
    assert calls == ["equity", "crypto", "fx"]
    assert cutoffs == [queued["stale_before"]] * 3
    assert status["status"] == "partial"
    assert status["progress"] == {"completed": 3, "total": 3}
    assert status["wealth_snapshot_created"] is True
    assert status["audit_recorded"] is True
    assert [row["status"] for row in status["sources"]] == ["complete", "failed", "complete"]
    assert conn.execute("SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots WHERE job_id=?", (job_id,)).fetchone()[0] == 1
    assert conn.execute("SELECT COUNT(*) FROM transactions").fetchone()[0] == 0
    conn.close()


def test_status_read_does_not_write_or_call_runner(tmp_path):
    path = tmp_path / "finance.sqlite3"
    conn = database(path)
    queued, _ = create_asset_price_refresh_job(conn)
    before = conn.total_changes
    first = asset_price_refresh_status(conn, queued["job_id"])
    second = asset_price_refresh_status(conn, queued["job_id"])
    assert conn.total_changes == before
    assert first == second
    assert first["provider_calls_on_read"] is False
    conn.close()


def test_concurrent_starts_create_exactly_one_active_job(tmp_path):
    path = tmp_path / "finance.sqlite3"
    database(path).close()
    barrier = threading.Barrier(2)
    outcomes: list[str] = []

    def worker() -> None:
        conn = connect(path)
        conn.execute("PRAGMA busy_timeout=5000")
        barrier.wait()
        try:
            create_asset_price_refresh_job(conn)
            outcomes.append("created")
        except ValueError as exc:
            outcomes.append(str(exc))
        finally:
            conn.close()

    first = threading.Thread(target=worker)
    second = threading.Thread(target=worker)
    first.start()
    second.start()
    first.join()
    second.join()

    assert sorted(outcomes) == ["asset_price_refresh_job_already_running", "created"]
    conn = connect(path)
    assert conn.execute(
        "SELECT COUNT(*) FROM asset_price_refresh_jobs WHERE status IN ('queued','running')"
    ).fetchone()[0] == 1
    conn.close()


def test_concurrent_workers_claim_a_queued_job_exactly_once(tmp_path):
    path = tmp_path / "finance.sqlite3"
    conn = database(path)
    queued, db_path = create_asset_price_refresh_job(conn)
    conn.close()
    calls: list[str] = []
    barrier = threading.Barrier(2)

    def runner(source: str) -> Callable[[Connection, str], tuple[int, int]]:
        def execute(_conn: Connection, _stale_before: str) -> tuple[int, int]:
            calls.append(source)
            return 0, 0

        return execute

    runners = {source: runner(source) for source in ("equity", "crypto", "fx")}

    def worker() -> None:
        barrier.wait()
        run_asset_price_refresh(db_path, queued["job_id"], runners=runners)

    first = threading.Thread(target=worker)
    second = threading.Thread(target=worker)
    first.start()
    second.start()
    first.join()
    second.join()

    assert sorted(calls) == ["crypto", "equity", "fx"]
    conn = connect(path)
    assert conn.execute(
        "SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots WHERE job_id=?",
        (queued["job_id"],),
    ).fetchone()[0] == 0
    conn.close()


def test_all_failed_sources_create_no_false_wealth_snapshot(tmp_path):
    path = tmp_path / "finance.sqlite3"
    conn = database(path)
    queued, db_path = create_asset_price_refresh_job(conn)
    conn.close()

    def failed(_conn, _stale_before):
        raise RuntimeError("internal provider detail")

    run_asset_price_refresh(db_path, queued["job_id"], runners={source: failed for source in ("equity", "crypto", "fx")})
    conn = connect(path)
    status = asset_price_refresh_status(conn, queued["job_id"])
    assert status["status"] == "failed"
    assert status["wealth_snapshot_created"] is False
    assert status["failed_assets"] == 3
    assert conn.execute("SELECT COUNT(*) FROM aggregated_wealth_refresh_snapshots").fetchone()[0] == 0
    assert "internal provider detail" not in str(status)
    conn.close()


def test_partial_instrument_failure_preserves_success_and_quality_counts(tmp_path):
    path = tmp_path / "finance.sqlite3"
    conn = database(path)
    queued, db_path = create_asset_price_refresh_job(conn)
    conn.close()

    def equity(_conn, _stale_before):
        return asset_refresh.SourceRunResult(
            stale_candidates=3,
            updated_count=2,
            stale_remaining_count=1,
            failed_count=1,
            diagnostics=("provider_error",),
        )

    def current(_conn, _stale_before):
        return asset_refresh.SourceRunResult(fresh_unchanged_count=2)

    run_asset_price_refresh(
        db_path,
        queued["job_id"],
        runners={"equity": equity, "crypto": current, "fx": current},
    )
    conn = connect(path)
    status = asset_price_refresh_status(conn, queued["job_id"])
    assert status["status"] == "partial"
    assert status["successful_assets"] == 2
    assert status["fresh_unchanged_assets"] == 4
    assert status["stale_assets"] == 1
    assert status["failed_assets"] == 1
    assert status["wealth_snapshot_created"] is True
    conn.close()


def test_unresolved_stale_asset_makes_job_partial(tmp_path):
    path = tmp_path / "finance.sqlite3"
    conn = database(path)
    queued, db_path = create_asset_price_refresh_job(conn)
    conn.close()

    def stale(_conn, _stale_before):
        return asset_refresh.SourceRunResult(
            stale_candidates=1,
            fresh_unchanged_count=1,
            stale_remaining_count=1,
        )

    def current(_conn, _stale_before):
        return asset_refresh.SourceRunResult(fresh_unchanged_count=1)

    run_asset_price_refresh(
        db_path,
        queued["job_id"],
        runners={"equity": stale, "crypto": current, "fx": current},
    )
    conn = connect(path)
    status = asset_price_refresh_status(conn, queued["job_id"])
    assert status["status"] == "partial"
    assert status["stale_assets"] == 1
    assert status["wealth_snapshot_created"] is False
    source = next(item for item in status["sources"] if item["source"] == "equity")
    assert source["status"] == "failed"
    assert source["error_code"] == "equity_stale_remaining"
    conn.close()


def test_equity_valuation_warning_marks_source_partial(tmp_path, monkeypatch):
    conn = database(tmp_path / "finance.sqlite3")
    response = type(
        "Response",
        (),
        {
            "total": 2,
            "cached": 0,
            "updated": 2,
            "economic_updated": 2,
            "results": [{"status": "fresh"}, {"status": "fresh"}],
            "errors": [],
            "warnings": ["portfolio_valuation_partial"],
        },
    )()
    monkeypatch.setattr(asset_refresh, "refresh_equity_quotes_batch", lambda *_args, **_kwargs: response)

    result = asset_refresh._equity_source(conn, "2026-08-26T00:00:00+00:00")

    assert result.updated_count == 2
    assert result.failed_count == 1
    assert result.diagnostics == ("portfolio_valuation_partial",)
    conn.close()


def test_crypto_refresh_fetches_only_stale_held_asset_ids(tmp_path, monkeypatch):
    path = tmp_path / "finance.sqlite3"
    conn = database(path)
    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','ethereum',1,'2026-01-01')"""
    )
    conn.execute(
        "INSERT INTO crypto_wallets(wallet_id,wallet_name,wallet_type,created_at) VALUES('wallet','Wallet','exchange','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('btc-held','btc','wallet','1','2026-08-27','verified','2026-01-01'),
                   ('eth-held','eth','wallet','1','2026-08-27','verified','2026-01-01')"""
    )
    now = datetime.now(UTC)
    conn.execute(
        """INSERT INTO crypto_prices(
             crypto_price_id,asset_id,coingecko_id,price_currency,price,provider,
             provider_timestamp,fetched_at,quality_status
           ) VALUES('eth-fresh','eth','ethereum','CHF','100','CoinGecko',?,?, 'fresh')""",
        (now.isoformat(), now.isoformat()),
    )
    conn.commit()

    class Provider:
        calls: list[tuple[str, ...]] = []

        def get_crypto_prices(self, coingecko_ids, currency="CHF"):
            ids = tuple(coingecko_ids)
            self.calls.append(ids)
            return {
                provider_id: PriceQuote(
                    provider_id,
                    currency,
                    Decimal("100") if provider_id == "ethereum" else Decimal("123.45"),
                    provider_timestamp=now.isoformat(),
                )
                for provider_id in ids
            }

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

    provider = Provider()
    monkeypatch.setattr(asset_refresh, "CoinGeckoClient", lambda: provider)
    candidates, updated = asset_refresh._crypto_source(
        conn,
        (now - timedelta(hours=24)).isoformat(),
    )

    assert (candidates, updated) == (1, 1)
    assert provider.calls == [("bitcoin",)]
    assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='eth'").fetchone()[0] == 1
    assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='btc'").fetchone()[0] == 1

    replay = asset_refresh._crypto_source(conn, "2099-01-01T00:00:00+00:00")
    assert replay.updated_count == 0
    assert replay.fresh_unchanged_count == 2
    assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='btc'").fetchone()[0] == 1
    assert conn.execute("SELECT COUNT(*) FROM crypto_prices WHERE asset_id='eth'").fetchone()[0] == 1
    conn.close()


def test_fx_metadata_replay_is_not_an_economic_update(tmp_path, monkeypatch):
    conn = database(tmp_path / "finance.sqlite3")
    conn.execute(
        "INSERT INTO instruments(instrument_id,asset_class,name,currency,is_active,created_at) "
        "VALUES('eur-asset','etf','Synthetic EUR','EUR',1,'2026-01-01')"
    )
    upsert_fx_rate(
        conn,
        base_currency="EUR",
        quote_currency="CHF",
        rate_date="2026-08-27",
        rate=Decimal("0.80448"),
        provider="mockfx",
        rate_type="close",
        fetched_at="2026-08-27T10:00:00+00:00",
    )
    conn.commit()

    def resolve_same(conn_arg, **_kwargs):
        upsert_fx_rate(
            conn_arg,
            base_currency="EUR",
            quote_currency="CHF",
            rate_date="2026-08-27",
            rate=Decimal("0.80448"),
            provider="mockfx",
            rate_type="close",
        )
        return SimpleNamespace(status="ok")

    monkeypatch.setattr("jarvis_finance.fx.rates.resolve_fx_rate_to_chf", resolve_same)
    result = asset_refresh._fx_source(conn, "2099-01-01T00:00:00+00:00")

    assert result.updated_count == 0
    assert result.fresh_unchanged_count == 1
    assert conn.execute("SELECT COUNT(*) FROM fx_rates WHERE base_currency='EUR'").fetchone()[0] == 1
    conn.close()