"""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:
        busy = conn.execute("SELECT count(*) AS n FROM ingest_jobs WHERE dataset_id=%s AND kind IN ('backfill','derive') AND status IN ('queued','running') AND id<>%s",
                            (dataset_id, job["id"])).fetchone()["n"]
        if busy:
            log.info("compaction.deferred", reason=f"{busy} backfill/derive job(s) pending on {dataset_id}")
            return {"compacted_files": 0, "deferred": True}
        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(config={"threads": 8, "memory_limit": "16GB"})
    # read every delta once (sorted), then slice in memory per (symbol, year) — avoids thousands of file scans
    all_rows = con.sql("SELECT * FROM read_parquet($p, hive_partitioning=false) ORDER BY symbol_as_reported, session_date, timestamp_utc",
                       params={"p": paths}).to_arrow_table().cast(BARS_SCHEMA)
    import polars as pl
    df = pl.from_arrow(all_rows).with_columns(pl.col("session_date").dt.year().alias("_y"))
    written, retire = [], [d["path"] for d in deltas]
    rows = 0
    groups = df.partition_by(["symbol_as_reported", "_y"], as_dict=True, maintain_order=True)
    log.info("compaction.loaded", delta_files=len(deltas), rows=df.height, groups=len(groups))
    for (s, y), new_df in groups.items():
        new = new_df.drop("_y").to_arrow().cast(BARS_SCHEMA)
        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]
        for o in old:
            parts.append(con.sql("SELECT * FROM read_parquet($p, hive_partitioning=false)", params={"p": [str(lake.abs(o["path"]))]}).to_arrow_table().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}