"""Integration: lake publication, durable queue, incremental sync idempotency, resume, derive, compaction."""
from __future__ import annotations

from datetime import UTC, date, datetime

import pyarrow as pa
import pytest
from findata_core.db import connection, fetch_one, transaction
from findata_ingest import queue
from findata_ingest.lake import live_files
from findata_ingest.reader import apply_adjustment, read_bars, run_sql
from findata_ingest.tasks import backfill, compaction, derive, incremental, metadata

from tests.conftest import build_update_zip

DS = "market_bars.stock.1m.raw"


def _job(kind, dataset_id, params, run=True):
    jid = queue.enqueue(kind, dataset_id, params, created_by="test")
    job = queue.claim_next([kind])
    assert job and str(job["id"]) == jid
    return job


def _rows(ds=DS):
    with connection() as conn:
        return conn.execute("SELECT coalesce(sum(row_count),0) n, count(*) f FROM lake_files WHERE dataset_id=%s AND version_removed IS NULL", (ds,)).fetchone()


def test_queue_retry_backoff_and_dead_letter():
    jid = queue.enqueue("quality", None, {"x": 1}, max_attempts=2, dedupe_key="q-test")
    assert queue.enqueue("quality", None, {"x": 1}, dedupe_key="q-test") is None  # deduped while queued
    job = queue.claim_next(["quality"])
    assert job["attempts"] == 1
    queue.finish(job, False, error="boom userid=SECRETVALUE")
    r = fetch_one("SELECT status, last_error, next_run_at > now() later FROM ingest_jobs WHERE id=%s", (jid,))
    assert r["status"] == "queued" and r["later"] and "SECRETVALUE" not in r["last_error"]
    with transaction() as conn:
        conn.execute("UPDATE ingest_jobs SET next_run_at=now() WHERE id=%s", (jid,))
    job = queue.claim_next(["quality"])
    queue.finish(job, False, error="boom again")
    assert fetch_one("SELECT status FROM ingest_jobs WHERE id=%s", (jid,))["status"] == "dead"


def test_metadata_then_incremental_is_idempotent(vendor, lake, log):
    metadata.run(_job("metadata", None, {"types": ["stock", "etf"]}), lake, log)
    with connection() as conn:
        n = conn.execute("SELECT count(*) FROM instruments").fetchone()["count"]
        delisted = conn.execute("SELECT status FROM instruments WHERE symbol='TWTR-DELISTED'").fetchone()["status"]
        splits = conn.execute("SELECT count(*) FROM corporate_actions WHERE action_type='split'").fetchone()["count"]
    assert n == 5 and delisted == "delisted" and splits == 4

    # first incremental: 3 symbols, one session
    r1 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day"}), lake, log)
    assert r1["rows_published"] == 900 + 390 + 532 or r1["rows_published"] > 1500
    assert r1["version"] == 1 and r1["files_added"] == 1 and r1["symbols"] == 3
    v1 = _rows()
    cov = fetch_one("SELECT count(*) n FROM coverage WHERE dataset_id=%s", (DS,))["n"]
    assert cov == 3
    raw = fetch_one("SELECT status, path FROM raw_archives WHERE dataset_id=%s", (DS,))
    assert raw["status"] == "published" and (lake.root / raw["path"]).exists()  # immutable raw kept

    # re-run: vendor unchanged → skipped, nothing duplicated
    r2 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day"}), lake, log)
    assert r2.get("skipped") and _rows() == v1

    # forced re-fetch of the identical archive → recognised by sha256, no new version
    r3 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day", "force": True}), lake, log)
    assert r3.get("skipped") and r3["reason"] == "archive already ingested" and _rows() == v1

    # vendor republishes the same session with a corrected bar → supersede (no duplicates, version 2)
    fixed = (lake.root.parent / "x").exists()
    import pathlib
    txt = (pathlib.Path(__file__).parent / "fixtures" / "firstrate" / "AAPL_day_1min_UNADJUSTED.txt").read_text()
    txt = txt.replace("2026-10-30 09:30:00,346.63,348.51,346.62,348.26,1119437", "2026-10-30 09:30:00,346.63,348.51,346.62,348.27,1119437")
    vendor.archive_override = build_update_zip(mutate={"AAPL": txt})
    vendor.last_update = "2026-10-30"
    r4 = incremental.run(_job("incremental", DS, {"type": "stock", "period": "day", "force": True}), lake, log)
    assert not r4.get("skipped") and r4["version"] == 2 and r4["files_retired"] == 1
    v2 = _rows()
    assert v2["n"] == v1["n"] and v2["f"] == 1
    t, meta = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))
    assert t.num_rows == 900 and meta["version"] == 2
    row = t.filter(pa.compute.equal(t.column("timestamp_utc"), pa.scalar(datetime(2026, 10, 30, 13, 30, tzinfo=UTC), pa.timestamp("us", tz="UTC")))).to_pylist()[0]
    assert row["close"] == 348.27
    # snapshot reproducibility: version 1 still readable from retired file set
    t1, _ = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30), version=1)
    assert t1.num_rows == 900
    vendor.archive_override = None
    # no duplicate (instrument, ts) anywhere in the live set
    dup = run_sql("SELECT count(*) FROM (SELECT instrument_id, timestamp_utc, count(*) c FROM market_bars_stock_1m_raw GROUP BY 1,2 HAVING c > 1)", 10, 10,
                  {"market_bars_stock_1m_raw": [str(lake.abs(f["path"])) for f in _live()]})[0].to_pylist()[0]
    assert list(dup.values())[0] == 0


def _live():
    with connection() as conn:
        return live_files(conn, DS)


def test_derive_5m_from_published_session(vendor, lake, log):
    job = queue.claim_next(["derive"])  # enqueued by the incremental run
    assert job is not None
    r = derive.run(job, lake, log)
    assert r["rows"] > 0
    t, _ = read_bars("market_bars.stock.5m.raw", ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))
    assert 150 <= t.num_rows <= 192 and t.column("interval")[0].as_py() == "5m"
    one, _ = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))
    assert abs(sum(t.column("volume").to_pylist()) - sum(one.column("volume").to_pylist())) < 1e-6


def test_adjustment_applied_at_query_time(vendor, lake, log):
    with transaction() as conn:
        conn.execute("INSERT INTO corporate_actions(instrument_id, action_type, ex_date, value, source_id) VALUES ('firstrate:stock:AAPL','split','2026-10-31',2,'firstrate') ON CONFLICT DO NOTHING")
    raw, _ = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))
    adj, notes = apply_adjustment(raw, "split")
    assert abs(adj.column("close")[0].as_py() * 2 - raw.column("close")[0].as_py()) < 1e-9
    assert abs(adj.column("volume")[0].as_py() - raw.column("volume")[0].as_py() * 2) < 1e-6
    assert adj.column("adjustment_type")[0].as_py() == "split" and raw.column("adjustment_type")[0].as_py() == "raw"
    with transaction() as conn:
        conn.execute("DELETE FROM corporate_actions WHERE ex_date='2026-10-31'")


def test_backfill_resumes_from_checkpoint_and_respects_delta_dates(vendor, lake, log):
    job = _job("backfill", DS, {"type": "stock", "ticker_range": "A", "delete_raw_after_publish": False})
    # simulate a crash after AAL was published: pre-seed the checkpoint
    queue.set_checkpoint(f"job:{job['id']}", "state", {"done_symbols": ["AAL"]})
    r = backfill.run(job, lake, log)
    # all fixture rows fall on 2026-10-30 which already lives in a delta file → excluded, nothing duplicated
    assert r["symbols"] == 0 and r["skipped_rows_in_delta"] > 0
    ck = queue.get_checkpoint(f"job:{job['id']}", "state")
    assert set(ck["done_symbols"]) == {"A", "AAL", "AAPL"}
    assert _rows()["f"] == 1


def test_compaction_merges_delta_into_base_and_checks_invariant(vendor, lake, log):
    job = _job("compaction", DS, {"min_age_days": 0})
    r = compaction.run(job, lake, log)
    assert r["compacted_files"] == 1 and r["base_files_written"] == 3
    live = _live()
    assert all(f["layout"] == "base" for f in live) and len(live) == 3
    t, _ = read_bars(DS, ["AAPL"], date(2026, 10, 30), date(2026, 10, 30))
    assert t.num_rows == 900
    # the retired delta file was moved out of the live tree
    assert not any(p.name.endswith(".parquet") for p in (lake.root / "curated" / DS / "delta").rglob("*.parquet"))
    assert any((lake.root / "_retired").rglob("*.parquet"))


def test_run_sql_sandbox_rejects_mutations():
    views = {"market_bars_stock_1m_raw": []}
    for bad in ["DROP TABLE x", "SELECT 1; SELECT 2", "COPY (SELECT 1) TO '/tmp/x'", "INSTALL httpfs", "SELECT * FROM read_csv('/etc/passwd')", "ATTACH 'x.db'"]:
        with pytest.raises(ValueError):
            run_sql(bad, 10, 5, views)
    t, meta = run_sql("SELECT 1 AS one", 10, 5, views)
    assert t.to_pylist() == [{"one": 1}]