"""Cache the full-data report packet for the consult card.

The key is chart identity + ayanamsa + node mode + engine version
+ the reference date used for current dasha, transits and the year chart.
The file name is a hash. Birth data stays inside the scratch file, mode 0600.
Files older than two days are removed. The consult card also keeps the last
few parsed packets in memory with their shared domain base (issue #129).
"""

from __future__ import annotations

import hashlib
import json
import os
import threading
import time
from collections import OrderedDict
from datetime import date
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Callable

ROOT = Path(__file__).resolve().parents[1]
CACHE_SCHEMA = "consult-full-data-cache-v2"
SLOW_MISS_SECONDS = 20.0
KEEP_SECONDS = 2 * 24 * 60 * 60

PacketBuilder = Callable[[dict[str, Any]], dict[str, Any]]

# Parsed consult packets kept in this process, newest last. One entry is about
# 18 MB parsed plus its shared base copy (~30 MB together), so three stay well
# inside the 4 GB host. A hit skips the 5.5 MB file read and the base copy.
MEMORY_SLOTS = 3
_memory: "OrderedDict[str, tuple[dict[str, Any], dict[str, Any]]]" = OrderedDict()
_memory_guard = threading.Lock()

# One build per cache key. A later caller waits and reads the stored packet.
class _Flight:
    def __init__(self) -> None:
        self.done = threading.Event()
        self.error: BaseException | None = None


_flights: dict[str, _Flight] = {}
_flights_guard = threading.Lock()

# Chart-save warm has its own cap. When it is full the warm is skipped.
# A user request does not take this slot and does not receive 429 from it.
WARM_CONCURRENCY_LIMIT = 1
_warm_guard = threading.Lock()
_warm_inflight = 0


def _acquire_warm_slot() -> bool:
    global _warm_inflight
    with _warm_guard:
        if _warm_inflight >= WARM_CONCURRENCY_LIMIT:
            return False
        _warm_inflight += 1
        return True


def _release_warm_slot() -> None:
    global _warm_inflight
    with _warm_guard:
        if _warm_inflight > 0:
            _warm_inflight -= 1


def cache_dir(name: str = "consult_full_data_cache") -> Path:
    path = ROOT / "scratch" / "local" / name
    path.mkdir(parents=True, exist_ok=True)
    return path


def engine_version() -> str:
    try:
        from jyotish_vedic import __version__ as version
    except Exception:
        return "unknown"
    return str(version)


def _number(value: Any) -> int | float:
    number = float(value)
    if number == int(number):
        return int(number)
    return number


def reference_date_from_body(body: dict[str, Any], *, today: date | None = None) -> str:
    """One date for dasha, transits and the year chart. Request first, else the server day."""
    for key in ("today", "transit_date"):
        raw = body.get(key)
        if isinstance(raw, str) and len(raw) >= 10:
            try:
                return date.fromisoformat(raw[:10]).isoformat()
            except ValueError:
                continue
    return (today or date.today()).isoformat()


def identity_from_body(body: dict[str, Any], *, today: date | None = None) -> dict[str, Any]:
    """Birth fields are required. Missing keys raise KeyError so the caller can gap."""
    node = body.get("node_mode", body.get("nodeMode", "mean"))
    ayanamsa = body.get("ayanamsa", "lahiri")
    second = body.get("second", 0)
    return {
        "schema": CACHE_SCHEMA,
        "year": int(body["year"]),
        "month": int(body["month"]),
        "day": int(body["day"]),
        "hour": _number(body["hour"]),
        "minute": _number(body["minute"]),
        "second": _number(second if second is not None else 0),
        "lat": _number(body["lat"]),
        "lon": _number(body["lon"]),
        "tz": _number(body["tz"]),
        "ayanamsa": str(ayanamsa),
        "node_mode": str(node or "mean"),
        "engine_version": engine_version(),
        "reference_date": reference_date_from_body(body, today=today),
    }


def cache_key(identity: dict[str, Any]) -> str:
    canonical = json.dumps(identity, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def _cache_path(identity: dict[str, Any], name: str = "consult_full_data_cache") -> Path:
    return cache_dir(name) / f"{cache_key(identity)}.json"


def prune_cache(now: float | None = None, *, name: str = "consult_full_data_cache") -> int:
    """Drop cache files last written more than two days ago."""
    cutoff = (time.time() if now is None else now) - KEEP_SECONDS
    removed = 0
    folder = cache_dir(name)
    for path in folder.glob("*.json"):
        try:
            if path.stat().st_mtime < cutoff:
                path.unlink()
                removed += 1
        except OSError:
            continue
    return removed


def _write_private_text(path: Path, text: str) -> None:
    """Birth data: the file is created 0600, never readable by other users.

    Same bytes as before (no key sorting), so a packet read back is unchanged.
    """
    descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
    with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
        handle.write(text)
    os.chmod(path, 0o600)


def _read_cached(path: Path, started: float, now: Callable[[], float]) -> dict[str, Any]:
    packet = json.loads(path.read_text(encoding="utf-8"))
    return {"packet": packet, "hit": True, "elapsed_s": now() - started, "slow": False}


def get_or_build(
    identity: dict[str, Any],
    builder: PacketBuilder | None = None,
    *,
    clock: Callable[[], float] | None = None,
    cache_name: str = "consult_full_data_cache",
    json_default: Callable[..., Any] | None = None,
) -> dict[str, Any]:
    """Return the packet. Concurrent callers for one key share one build.

    ``cache_name`` selects the scratch subdirectory. The consult card keeps the
    default. A non-default name does not change the consult key or its files.
    """
    prune_cache(name=cache_name)
    now = clock or time.perf_counter
    started = now()
    path = _cache_path(identity, cache_name)
    if path.is_file():
        return _read_cached(path, started, now)
    key = f"{cache_name}\n{cache_key(identity)}"
    with _flights_guard:
        flight = _flights.get(key)
        owner = flight is None
        if owner:
            flight = _Flight()
            _flights[key] = flight
    assert flight is not None
    if not owner:
        flight.done.wait()
        if flight.error is not None:
            raise flight.error
        if path.is_file():
            return _read_cached(path, started, now)
        raise RuntimeError("consult_full_data_unavailable")
    try:
        if path.is_file():
            return _read_cached(path, started, now)
        if builder is None:
            builder = build_full_data_packet
        packet = builder(identity)
        temporary = path.with_suffix(".json.tmp")
        dump_kwargs: dict[str, Any] = {"ensure_ascii": False, "separators": (",", ":")}
        if json_default is not None:
            dump_kwargs["default"] = json_default
        _write_private_text(temporary, json.dumps(packet, **dump_kwargs))
        temporary.replace(path)
        elapsed = now() - started
        return {
            "packet": packet,
            "hit": False,
            "elapsed_s": elapsed,
            "slow": elapsed > SLOW_MISS_SECONDS,
        }
    except Exception as exc:
        flight.error = exc
        raise
    finally:
        flight.done.set()
        with _flights_guard:
            if _flights.get(key) is flight:
                _flights.pop(key, None)


def consult_packet_view(identity: dict[str, Any], builder: PacketBuilder | None = None) -> dict[str, Any]:
    """The consult card's packet and its shared domain base, both read only.

    Kept in memory for the last ``MEMORY_SLOTS`` charts. Callers must not
    mutate ``packet`` or ``base``: later turns read the same objects. Other
    users of ``get_or_build`` (report reference, marriage match) still get
    their own copy from the file.
    """
    try:
        from pl9_full_data_export import full_data_domain_base
    except ImportError:  # pragma: no cover - package import path
        from scripts.pl9_full_data_export import full_data_domain_base
    key = cache_key(identity)
    started = time.perf_counter()
    with _memory_guard:
        entry = _memory.get(key)
        if entry is not None:
            _memory.move_to_end(key)
    if entry is not None:
        return {"packet": entry[0], "base": entry[1], "hit": True, "memory": True,
                "elapsed_s": time.perf_counter() - started, "slow": False}
    cached = get_or_build(identity, builder)
    base = full_data_domain_base(cached["packet"])
    # Remember what a file hit returns: a fresh build is the builder's own
    # object, and its JSON round trip (lists for tuples, string keys) is what
    # every later turn used to read. This turn still answers from the build.
    remembered = (cached["packet"], base)
    if not cached["hit"]:
        path = _cache_path(identity)
        if path.is_file():
            parsed = json.loads(path.read_text(encoding="utf-8"))
            remembered = (parsed, full_data_domain_base(parsed))
    with _memory_guard:
        _memory[key] = remembered
        _memory.move_to_end(key)
        while len(_memory) > MEMORY_SLOTS:
            _memory.popitem(last=False)
    return {**cached, "base": base, "memory": False}


def clear_memory_cache() -> None:
    with _memory_guard:
        _memory.clear()


def _attach_graha_drishti(packet: dict[str, Any], reading: dict[str, Any]) -> dict[str, Any]:
    """Copy the engine's whole-sign aspects onto the consult packet. Do not recompute."""
    modules = reading.get("modules") if isinstance(reading, dict) else None
    aspects = modules.get("aspects") if isinstance(modules, dict) else None
    rows = aspects.get("house_aspects") if isinstance(aspects, dict) else None
    worksheets = packet.get("worksheets")
    if not isinstance(rows, list) or not isinstance(worksheets, dict):
        return packet
    if "graha_drishti" not in worksheets:
        worksheets["graha_drishti"] = {
            "source": "modules.aspects.house_aspects",
            "citation": "references/signs-and-houses.md",
            "house_aspects": rows,
        }
    return packet


def build_full_data_packet(identity: dict[str, Any]) -> dict[str, Any]:
    """Packet-only build. No markdown render and no sanitized rewrite."""
    from calculation_profile_contract import attach_calculation_profile
    from jyotish_engine import build_pl9_style_export_packet, cmd_full_reading

    ref = date.fromisoformat(str(identity["reference_date"]))
    args = SimpleNamespace(
        year=int(identity["year"]),
        month=int(identity["month"]),
        day=int(identity["day"]),
        hour=float(identity["hour"]),
        minute=float(identity["minute"]),
        second=float(identity["second"]),
        lat=float(identity["lat"]),
        lon=float(identity["lon"]),
        tz=float(identity["tz"]),
        ayanamsa=identity["ayanamsa"],
        node_mode=identity["node_mode"],
        today=ref.isoformat(),
        target_year=ref.year,
        age=ref.year - int(identity["year"]),
        transit_date=ref.isoformat(),
        birth_time_accuracy="confirmed",
    )
    reading = cmd_full_reading(args)
    if not isinstance(reading, dict) or reading.get("error"):
        raise RuntimeError("full_reading_unavailable")
    packet = build_pl9_style_export_packet(reading)
    if not isinstance(packet, dict):
        raise RuntimeError("full_reading_unavailable")
    _attach_graha_drishti(packet, reading)
    return attach_calculation_profile(packet, args)


def warm_consult_packet(body: dict[str, Any] | None) -> dict[str, Any]:
    """Build today's packet after a chart save. The response stays small.

    At most WARM_CONCURRENCY_LIMIT warms run at once. A full cap skips this
    warm. The skip is not a user-request 429, and a failure here does not
    fail the chart save (the save route does not wait on this response).
    """
    if not _acquire_warm_slot():
        return {
            "success": True,
            "endpoint": "consult_card_warm",
            "skipped": True,
            "hit": False,
            "elapsed_s": 0.0,
            "slow": False,
        }
    try:
        try:
            cached = consult_packet_view(identity_from_body(dict(body or {})))
        except Exception as exc:
            return {
                "success": False,
                "endpoint": "consult_card_warm",
                "error_type": type(exc).__name__,
            }
        return {
            "success": True,
            "endpoint": "consult_card_warm",
            "hit": bool(cached["hit"]),
            "elapsed_s": cached["elapsed_s"],
            "slow": bool(cached["slow"]),
        }
    finally:
        _release_warm_slot()


def _requested_domains(body: dict[str, Any]) -> list[str]:
    # Card domains (parents, children) are not the engine theme (family).
    raw = body.get("consult_card_domains")
    if raw is None:
        raw = body.get("themes", body.get("theme"))
    if raw is None:
        return []
    if isinstance(raw, str):
        return [raw]
    if isinstance(raw, list):
        return [str(item) for item in raw if isinstance(item, str)]
    return []


def attach_consult_full_data(result: dict[str, Any], body: dict[str, Any] | None) -> dict[str, Any]:
    """Thin hook for the consultation response. Failures do not break the chat."""
    payload = dict(body or {})
    if payload.get("include_consult_card_facts") is not True:
        return result
    try:
        from consult_card_domain_facts import build_consult_card_facts
    except ImportError:
        from scripts.consult_card_domain_facts import build_consult_card_facts
    try:
        identity = identity_from_body(payload)
        cached = consult_packet_view(identity)
        facts = build_consult_card_facts(
            cached["packet"],
            _requested_domains(payload),
            minute_confirmed=None,
            body=payload,
            base=cached["base"],
        )
        facts["timing"] = {
            "hit": bool(cached["hit"]),
            "elapsed_s": cached["elapsed_s"],
            "slow": bool(cached["slow"]),
        }
    except Exception as exc:
        facts = {
            "schema": "consult-card-facts-v1",
            "gaps": ["consult_full_data_unavailable"],
            "error_type": type(exc).__name__,
        }
    attached = dict(result)
    attached["consult_card_facts"] = facts
    return attached