"""Compaction: merge session_date delta files older than `min_age_days` into base files per (symbol, year).
Asserts the (instrument_id, timestamp_utc) uniqueness invariant before publishing."""
from __future__ import annotations

from datetime import UTC, datetime, timedelta
from typing import Any

import duckdb
import pyarrow as pa
from findata_core.db import connection
from findata_schema import BARS_SCHEMA

from findata_ingest.lake import Lake


def run(job: dict[str, Any], lake: Lake, log) -> dict[str, Any]:
    p = job.get("params") or {}
    dataset_id = job["dataset_id"]
    min_age = int(p.get("min_age_days", 35))
    max_files = int(p.get("max_delta_files", 40))
    cutoff = (datetime.now(UTC) - timedelta(days=min_age)).date()
    with connection() as conn:
        deltas = conn.execute("SELECT path, date_min FROM lake_files WHERE dataset_id=%s AND layout='delta' AND version_removed IS NULL AND date_max <= %s ORDER BY date_min LIMIT %s",
                              (dataset_id, cutoff, max_files)).fetchall()
    if not deltas:
        return {"compacted_files": 0}
    paths = [str(lake.abs(d["path"])) for d in deltas]
    con = duckdb.connect()
    con.execute("SET threads TO 8")
    syms = [r[0] for r in con.sql("SELECT DISTINCT symbol_as_reported FROM read_parquet($p, hive_partitioning=false) ORDER BY 1", params={"p": paths}).fetchall()]
    years = sorted({d["date_min"].year for d in deltas})
    written, retire = [], [d["path"] for d in deltas]
    rows = 0
    for s in syms:
        for y in years:
            new = con.sql("SELECT * FROM read_parquet($p, hive_partitioning=false) WHERE symbol_as_reported=$s AND year(session_date)=$y", params={"p": paths, "s": s, "y": y}).to_arrow_table()
            if new.num_rows == 0:
                continue
            with connection() as conn:
                old = conn.execute("SELECT path FROM lake_files WHERE dataset_id=%s AND layout='base' AND version_removed IS NULL AND symbol_min=%s AND symbol_max=%s AND date_min>=%s AND date_max<=%s",
                                   (dataset_id, s, s, f"{y}-01-01", f"{y}-12-31")).fetchall()
            parts = [new.cast(BARS_SCHEMA)]
            for o in old:
                t = con.sql("SELECT * FROM read_parquet($p, hive_partitioning=false)", params={"p": [str(lake.abs(o["path"]))]}).to_arrow_table()
                parts.append(t.cast(BARS_SCHEMA))
                retire.append(o["path"])
            merged = pa.concat_tables(parts)
            con.register("m", merged)
            dup = con.sql("SELECT count(*) AS n FROM (SELECT instrument_id, timestamp_utc FROM m GROUP BY 1,2 HAVING count(*)>1)").fetchall()[0][0]
            con.unregister("m")
            if dup:
                raise RuntimeError(f"compaction invariant violated: {dup} duplicate (instrument_id, timestamp_utc) for {s} {y}")
            written.append(lake.write_table(dataset_id, merged, "base", {"symbol": s, "year": str(y)}))
            rows += merged.num_rows
    pub = lake.publish(dataset_id, written, sorted(set(retire)), job.get("run_id"), note=f"compaction {len(deltas)} delta files")
    log.info("compaction.done", deltas=len(deltas), base_files=len(written), rows=rows, **pub)
    return {"compacted_files": len(deltas), "base_files_written": len(written), "rows": rows, **pub}