"""DuckDB query plans: top-N-per-group queries must scan the Parquet files ONCE, with the datetime filter pushed
into the scan (no late-materialization re-scan), and the process must hold a single DuckDB database."""
from __future__ import annotations

import threading

import pytest


@pytest.fixture
def paths(app):
    import main
    idx = main.ticker_index("stock", "1min", "UNADJUSTED")
    return [idx[t] for t in ("AAPL", "MSFT", "SMCP")]


def _plan(sql, params):
    from core.duck import con, explain
    return explain(con(), sql, params)


def test_topn_plan_single_filtered_scan(paths):
    from core.duck import scan_count, topn_per_group
    lo, hi = "2025-06-30 09:30:00", "2025-06-30 09:34:59"
    sql = topn_per_group(None, paths, "datetime >= ? AND datetime <= ?", [lo, hi], "ticker", "datetime DESC", 100,
                         order_by="ticker, datetime DESC", limit=300)
    plan = _plan(sql, [paths, lo, hi, 100, 300])
    assert scan_count(plan) == 1, plan
    assert "Filters" in plan and "datetime>=" in plan.replace(" ", "") , plan
    assert "HASH_JOIN" not in plan, plan


def test_naive_qualify_would_double_scan(paths):
    """Documents WHY the helper exists: the bare form is rewritten into two scans + a join."""
    from core.duck import scan_count
    lo, hi = "2025-06-30 09:30:00", "2025-06-30 09:34:59"
    sql = ("SELECT * FROM read_parquet(?) WHERE datetime >= ? AND datetime <= ? "
           "QUALIFY row_number() OVER (PARTITION BY ticker ORDER BY datetime DESC) <= ? ORDER BY ticker, datetime DESC LIMIT ?")
    plan = _plan(sql, [paths, lo, hi, 100, 300])
    assert scan_count(plan) >= 2 or "HASH_JOIN" in plan, plan


def test_snapshot_plan_single_scan(paths):
    from core.duck import scan_count, topn_per_group
    sql = topn_per_group(None, paths, "datetime <= ?", ["2025-06-30 10:00:00"], "ticker", "datetime DESC", 1,
                         order_by="ticker", limit=3)
    plan = _plan(sql, [paths, "2025-06-30 10:00:00", 1, 3])
    assert scan_count(plan) == 1 and "HASH_JOIN" not in plan, plan


def test_futures_merged_plan_single_scan_per_file(app):
    from core.duck import scan_count
    from futures.lake import files_for, merged_sql
    files = files_for("ESH25", "1day")
    assert len(files) == 2, files          # update + archive overlap for 2025 contracts (daily: both cover 2024-12)
    sql = f"SELECT * EXCLUDE (ticker) FROM ({merged_sql(files, where='datetime >= ? AND datetime < ?')}) ORDER BY datetime LIMIT ?"
    plan = _plan(sql, [*files, "2024-12-01", "2024-12-31", 5001])
    assert scan_count(plan) == 2, plan     # one READ_PARQUET per file, none duplicated
    assert "HASH_JOIN" not in plan, plan
    from core.duck import con
    df = con().execute(sql, [*files, "2024-12-01", "2024-12-31", 5001]).df()
    assert len(df) == 21 and df["datetime"].is_unique       # dedup on datetime across the two buckets


def test_run_topn_results(paths):
    from core.duck import con, run_topn
    df = run_topn(con(), paths, "datetime >= ? AND datetime <= ?", ["2025-06-30 09:30:00", "2025-06-30 09:34:59"],
                  "ticker", "datetime DESC", 2, order_by="ticker, datetime DESC", limit=100)
    assert len(df) == 6 and list(df["ticker"].unique()) == ["AAPL", "MSFT", "SMCP"]
    assert str(df["datetime"].iloc[0]).startswith("2025-06-30 09:34")


def test_single_database_shared_across_threads(app):
    from core import duck
    seen = {}

    def worker(i):
        c = duck.con()
        assert duck.con() is c                                   # stable within the thread
        seen[i] = (c, c.execute("SELECT current_setting('memory_limit')").fetchone()[0])

    ts = [threading.Thread(target=worker, args=(i,)) for i in range(4)]
    for t in ts:
        t.start()
    for t in ts:
        t.join()
    assert len({id(v[0]) for v in seen.values()}) == 4      # one cursor per thread…
    assert len({v[1] for v in seen.values()}) == 1          # …one configured database
    limit = seen[0][1]
    assert limit.upper().startswith(("3.", "4", "3.7"))  # 4GB → DuckDB reports ~3.7 GiB
    threads = duck.con().execute("SELECT current_setting('threads')").fetchone()[0]
    assert int(threads) == duck.settings.duck_threads


def test_cache_has_ttl_and_maxsize():
    from core.cache import TTLCache
    c = TTLCache(maxsize=3, ttl=1000)
    for k in "abcd":
        c.set(k, k)
    assert len(c) == 3 and c.get("a") is None and c.get("d") == "d"
    calls = {"n": 0}

    def build():
        calls["n"] += 1
        return "v"
    assert c.get_or_build("x", build) == "v" and c.get_or_build("x", build) == "v" and calls["n"] == 1
    assert c.get_or_build("x", build, ttl=-1) == "v" and calls["n"] == 2      # caller-specific horizon expired
    assert c.invalidate("x") == 1 and c.get("x") is None