inventory_cache.py

"""
Persistent SQLite inventory cache with mtime-based per-coin invalidation.

Location: data/logs/inventory_cache.db

For each (exchange, dataset, coin):
  - Stores all table display columns (n_files, total_bytes, oldest_day, ...,
    hl_minutes, other_minutes, missing_minutes, expected_hours, coverage_pct, ...)
  - Stores dir_mtime + optionally src_mtime (for source-index changes)
  - On load: reads all coin-dir mtimes (< 1ms for 227 coins), recomputes only
    changed coins, writes back — typical page load < 50ms regardless of coin count

Supported datasets:
  - "1m" / "candles_1m"      → data/ohlcv/{exchange}/1m/{coin}/*.npz + source index
  - "1m_api" / "candles_1m_api" → data/ohlcv/{exchange}/1m_api/{coin}/*.npz
  - "l2Book"                 → data/ohlcv/{exchange}/l2Book/{coin}/*.lz4
  - "pb7_cache:{tf}"         → pb7/caches/ohlcv/{exchange}/{tf}/{coin}/*.npy
  - "pb8_cache:{tf}"         → pb8/caches/ohlcv/{exchange}/{tf}/{coin}/*.npy

Public API:
  get_inventory(exchange, dataset, *, lag_minutes, tradfi_type_fn, force_refresh)
      → list[dict]  (sorted by coin)
  invalidate_exchange(exchange)   → remove all rows for this exchange (full recompute)
  invalidate_coin(exchange, dataset, coin) → remove specific coin row
"""

from __future__ import annotations

import errno
import fcntl
import os
import sqlite3
import stat
import threading
import time
from contextlib import contextmanager
from datetime import date, datetime, timedelta
from pathlib import Path
from typing import Any, Callable

from market_data import (
    _parse_day_hour_from_filename,
    _parse_pb7_cache_day_from_name,
    get_exchange_raw_root_dir,
    _get_pb7_root_dir,
    _get_pb8_root_dir,
    is_l2book_archive_enabled,
    load_l2book_archive_dir,
)
from market_data_sources import (
    get_daily_source_counts_for_range,
    get_source_codes_for_day,
    get_source_index_path,
)
from logging_helpers import human_log as _log

SERVICE = "MarketData"

# ---------------------------------------------------------------------------
# DB setup
# ---------------------------------------------------------------------------

_DB_FILENAME = "inventory_cache.db"
_CONNECT_TIMEOUT_SECONDS = 10


def _db_path() -> Path:
    p = Path(__file__).resolve().parent / "data" / "logs" / _DB_FILENAME
    p.parent.mkdir(parents=True, exist_ok=True)
    return p


# Setup (WAL and schema) is serialized across threads and processes. Normal
# connections only read whether it is complete. The lock budget matches the
# SQLite busy timeout; SQLite statements afterwards may wait additionally.
_SETUP_LOCK_TIMEOUT_SECONDS = _CONNECT_TIMEOUT_SECONDS
_WAL_ATTEMPT_LIMIT = 64
_setup_thread_lock = threading.Lock()
_state_lock = threading.Lock()
# Resolved database paths whose WAL switch completed in this process (any
# resulting mode). Bounded; after eviction one more attempt may happen.
_wal_attempted: set[str] = set()
_open_lock_fds: set[int] = set()


def _reset_after_fork() -> None:
    """Give a forked child fresh locks and state; a parent thread may hold the old ones."""
    global _setup_thread_lock, _state_lock, _wal_attempted, _open_lock_fds
    inherited = list(_open_lock_fds)
    _setup_thread_lock = threading.Lock()
    _state_lock = threading.Lock()
    _wal_attempted = set()
    _open_lock_fds = set()
    # POSIX record locks belong to the parent and are not inherited; closing the
    # child's descriptor copies only prevents leaks and never releases them.
    for fd in inherited:
        try:
            os.close(fd)
        except OSError:
            pass


if hasattr(os, "register_at_fork"):
    os.register_at_fork(after_in_child=_reset_after_fork)


def _open_setup_lock_file(path: Path) -> int:
    """Open the owner-only setup lock file without following links."""
    flags = os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0) | getattr(os, "O_CLOEXEC", 0)
    fd = os.open(path, flags, 0o600)
    try:
        info = os.fstat(fd)
        if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid():
            raise sqlite3.OperationalError(f"Unsafe inventory cache setup lock file: {path}")
        os.fchmod(fd, 0o600)
        with _state_lock:
            _open_lock_fds.add(fd)
    except BaseException:
        os.close(fd)
        raise
    return fd


def _close_setup_lock_file(fd: int) -> None:
    """Close a lock descriptor; closing also releases this process's record lock."""
    with _state_lock:
        _open_lock_fds.discard(fd)
    os.close(fd)


@contextmanager
def _setup_lock(db_path: Path):
    """Serialize setup between threads and processes within a bounded wait."""
    deadline = time.monotonic() + _SETUP_LOCK_TIMEOUT_SECONDS
    thread_lock = _setup_thread_lock
    if not thread_lock.acquire(timeout=_SETUP_LOCK_TIMEOUT_SECONDS):
        raise sqlite3.OperationalError("inventory cache setup lock timed out")
    try:
        fd = _open_setup_lock_file(db_path.with_name(db_path.name + ".lock"))
        try:
            while True:
                try:
                    fcntl.lockf(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
                    break
                except OSError as exc:
                    if exc.errno not in (errno.EACCES, errno.EAGAIN):
                        raise
                    if time.monotonic() >= deadline:
                        raise sqlite3.OperationalError("inventory cache setup lock timed out") from None
                    time.sleep(0.01)
            try:
                yield
            finally:
                fcntl.lockf(fd, fcntl.LOCK_UN)
        finally:
            _close_setup_lock_file(fd)
    finally:
        thread_lock.release()


def _table_exists(conn: sqlite3.Connection) -> bool:
    """Read whether the schema exists; read errors propagate unchanged."""
    return conn.execute(
        "SELECT 1 FROM sqlite_master WHERE type='table' AND name='inventory'"
    ).fetchone() is not None


def _setup_complete(conn: sqlite3.Connection, key: str) -> bool:
    """Return whether this connection can skip setup, using only reads."""
    if not _table_exists(conn):
        return False
    with _state_lock:
        if key in _wal_attempted:
            return True
    return str(conn.execute("PRAGMA journal_mode").fetchone()[0]).lower() == "wal"


def _remember_wal_attempt(key: str) -> None:
    """Record a completed WAL attempt within the bounded process state."""
    with _state_lock:
        if len(_wal_attempted) >= _WAL_ATTEMPT_LIMIT:
            _wal_attempted.clear()
        _wal_attempted.add(key)


def _run_setup(conn: sqlite3.Connection, key: str) -> None:
    """Create the schema and try WAL; caller holds the setup lock.

    An existing schema gets one WAL attempt per process. A missing schema (new,
    removed or replaced file) always gets one, so a recreated file is not left
    in rollback mode by an earlier marker.
    """
    wal_result = None
    with _state_lock:
        attempted = key in _wal_attempted
    if not attempted or not _table_exists(conn):
        attempted = False
    if str(conn.execute("PRAGMA journal_mode").fetchone()[0]).lower() != "wal" and not attempted:
        try:
            wal_result = str(conn.execute("PRAGMA journal_mode=WAL").fetchone()[0]).lower()
        except sqlite3.OperationalError as exc:
            # Readers outside this lock (another PBGui version or tool) can keep
            # the switch busy. The cache works in rollback mode; retry later.
            if (getattr(exc, "sqlite_errorcode", 0) & 0xFF) != sqlite3.SQLITE_BUSY:
                raise
    _init_db(conn)
    if wal_result is not None:
        if wal_result != "wal":
            _log(SERVICE, f"Inventory cache uses journal mode '{wal_result}' because WAL is unavailable", level="WARNING")
        _remember_wal_attempt(key)


@contextmanager
def _get_conn():
    db_path = _db_path()
    key = str(db_path.resolve())
    conn = sqlite3.connect(str(db_path), timeout=_CONNECT_TIMEOUT_SECONDS, check_same_thread=False)
    try:
        # The schema check runs per connection, so a removed or replaced
        # database file is set up again on its next access.
        if not _setup_complete(conn, key):
            with _setup_lock(db_path):
                if not _setup_complete(conn, key):
                    _run_setup(conn, key)
        conn.execute("PRAGMA synchronous=NORMAL")
        yield conn
    finally:
        conn.close()


def _init_db(conn: sqlite3.Connection) -> None:
    conn.execute(
        """
        CREATE TABLE IF NOT EXISTS inventory (
            exchange         TEXT NOT NULL,
            dataset          TEXT NOT NULL,
            coin             TEXT NOT NULL,
            dir_mtime        REAL NOT NULL DEFAULT 0,
            src_mtime        REAL NOT NULL DEFAULT 0,
            n_files          INTEGER NOT NULL DEFAULT 0,
            total_bytes      INTEGER NOT NULL DEFAULT 0,
            oldest_day       TEXT NOT NULL DEFAULT '',
            newest_day       TEXT NOT NULL DEFAULT '',
            n_days           INTEGER NOT NULL DEFAULT 0,
            expected_hours   REAL NOT NULL DEFAULT 0,
            coverage_pct     REAL NOT NULL DEFAULT 0,
            missing_days_count INTEGER NOT NULL DEFAULT 0,
            missing_days_sample TEXT NOT NULL DEFAULT '',
            hl_minutes       INTEGER NOT NULL DEFAULT 0,
            other_minutes    INTEGER NOT NULL DEFAULT 0,
            missing_minutes  INTEGER NOT NULL DEFAULT 0,
            PRIMARY KEY (exchange, dataset, coin)
        )
        """
    )


# ---------------------------------------------------------------------------
# mtime helpers
# ---------------------------------------------------------------------------

def _dir_mtime(path: Path) -> float:
    try:
        return float(path.stat().st_mtime)
    except Exception:
        return 0.0


def _src_mtime(exchange: str, coin: str) -> float:
    """mtime of the source index file (1m_src/.../sources.idx)."""
    try:
        return float(get_source_index_path(exchange, coin).stat().st_mtime)
    except Exception:
        return 0.0


# ---------------------------------------------------------------------------
# Per-coin computation
# ---------------------------------------------------------------------------

def _scan_npz_coin_dir(coin_path: Path) -> dict[str, Any]:
    """Fast scan of a coin dir containing .npz files. Returns base fields."""
    n_files = 0
    total_bytes = 0
    oldest_day = ""
    newest_day = ""

    try:
        with os.scandir(str(coin_path)) as it:
            for fe in it:
                if not fe.name.endswith(".npz"):
                    continue
                day = _parse_day_hour_from_filename(fe.name)
                if not day:
                    continue
                day_s = day[0] if isinstance(day, tuple) else day
                if not day_s:
                    continue
                n_files += 1
                try:
                    total_bytes += fe.stat().st_size
                except Exception:
                    pass
                if not oldest_day or day_s < oldest_day:
                    oldest_day = day_s
                if not newest_day or day_s > newest_day:
                    newest_day = day_s
    except Exception:
        pass

    n_days = 0
    try:
        if oldest_day and newest_day:
            dt0 = datetime.strptime(oldest_day, "%Y%m%d").date()
            dt1 = datetime.strptime(newest_day, "%Y%m%d").date()
            today = date.today()
            if today > dt1:
                dt1 = today
            if dt1 >= dt0:
                n_days = (dt1 - dt0).days + 1
    except Exception:
        pass

    return {
        "n_files": n_files,
        "total_bytes": total_bytes,
        "oldest_day": oldest_day,
        "newest_day": newest_day,
        "n_days": n_days,
    }


def _compute_1m_coin(
    exchange: str,
    coin: str,
    coin_path: Path,
    lag_minutes: int,
    tradfi_type_fn: Callable[[str], str] | None,
    expected_minutes_fn: "Callable[[str, Any], set[int]] | None" = None,
) -> dict[str, Any]:
    base = _scan_npz_coin_dir(coin_path)

    hl_minutes = 0
    other_minutes = 0
    missing_minutes = 0
    expected_hours = int(base["n_days"]) * 24
    coverage_pct = 0.0
    missing_days_count = 0
    missing_days_sample = ""

    oldest_day = base["oldest_day"]
    newest_day = base["newest_day"]

    if oldest_day and newest_day:
        end_day = date.today().strftime("%Y%m%d")
        try:
            effective_now = datetime.utcnow()
            if lag_minutes > 0:
                effective_now = effective_now - timedelta(minutes=lag_minutes)

            counts = get_daily_source_counts_for_range(
                exchange=exchange,
                coin=coin,
                start_day=oldest_day,
                end_day=end_day,
                lag_minutes=lag_minutes,
                cutoff_ts_ms=None,
            )

            if counts:
                for day_data in counts.values():
                    hl_minutes += int(day_data.get("api") or 0) + int(day_data.get("l2Book_mid") or 0)
                    other_minutes += int(day_data.get("other_exchange") or 0)

                tradfi_type = tradfi_type_fn(coin) if tradfi_type_fn else ""

                if tradfi_type and expected_minutes_fn is not None:
                    # TradFi: use session-aware expected minutes via caller-provided callback
                    expected_minutes_total = 0
                    covered_minutes_total = 0
                    missing_days: list[str] = []
                    try:
                        dt0 = datetime.strptime(oldest_day, "%Y%m%d").date()
                        dt1 = datetime.strptime(end_day, "%Y%m%d").date()
                        effective_day_utc = effective_now.date()
                        effective_minute_idx = effective_now.hour * 60 + effective_now.minute
                        cur = dt0
                        while cur <= dt1:
                            expected_indices: set[int] = expected_minutes_fn(tradfi_type, cur)
                            if expected_indices:
                                if cur > effective_day_utc:
                                    expected_indices = set()
                                elif cur == effective_day_utc:
                                    expected_indices = {mi for mi in expected_indices if mi <= effective_minute_idx}
                            expected_cnt = len(expected_indices)
                            if expected_cnt > 0:
                                expected_minutes_total += expected_cnt
                                day_s = cur.strftime("%Y%m%d")
                                day_codes = get_source_codes_for_day(exchange=exchange, coin=coin, day=day_s)
                                covered_cnt = 0
                                if isinstance(day_codes, list) and day_codes:
                                    for minute_idx in expected_indices:
                                        if minute_idx < len(day_codes) and int(day_codes[minute_idx]) != 0:
                                            covered_cnt += 1
                                covered_minutes_total += covered_cnt
                                if covered_cnt < expected_cnt:
                                    missing_days.append(day_s)
                            cur = cur + timedelta(days=1)

                        missing_missing = max(0, expected_minutes_total - covered_minutes_total)
                        missing_minutes = missing_missing
                        expected_hours = round(float(expected_minutes_total) / 60.0, 2)
                        coverage_pct = (
                            round((float(covered_minutes_total) / float(expected_minutes_total)) * 100.0, 2)
                            if expected_minutes_total > 0
                            else 0.0
                        )
                        missing_days_count = len(missing_days)
                        if missing_days:
                            sample = ",".join(missing_days[:10])
                            if len(missing_days) > 10:
                                sample += ",..."
                            missing_days_sample = sample
                    except Exception:
                        pass
                else:
                    # Non-TradFi: missing = total missing from source index
                    total_missing = sum(int(d.get("missing") or 0) for d in counts.values())
                    missing_minutes = total_missing
                    total_present = hl_minutes + other_minutes
                    total_counted = total_present + total_missing
                    if total_counted > 0:
                        coverage_pct = round((total_present / total_counted) * 100.0, 2)
                    # Missing days (< 1440 total minutes)
                    m_days = []
                    for day_s, day_data in sorted(counts.items()):
                        day_total = sum(day_data.values())
                        if day_total < 1440:
                            m_days.append(day_s)
                    missing_days_count = len(m_days)
                    if m_days:
                        sample = ",".join(m_days[:10])
                        if len(m_days) > 10:
                            sample += ",..."
                        missing_days_sample = sample
        except Exception:
            pass

    return {
        **base,
        "expected_hours": expected_hours,
        "coverage_pct": coverage_pct,
        "missing_days_count": missing_days_count,
        "missing_days_sample": missing_days_sample,
        "hl_minutes": hl_minutes,
        "other_minutes": other_minutes,
        "missing_minutes": missing_minutes,
    }


def _compute_1m_api_coin(coin_path: Path) -> dict[str, Any]:
    base = _scan_npz_coin_dir(coin_path)
    n_days = base["n_days"]
    n_files = base["n_files"]
    coverage_pct = 0.0
    if n_days > 0:
        coverage_pct = round(min(100.0, (n_files / float(n_days)) * 100.0), 2)
    expected_hours = int(n_days) * 24
    return {
        **base,
        "expected_hours": expected_hours,
        "coverage_pct": coverage_pct,
        "missing_days_count": 0,
        "missing_days_sample": "",
        "hl_minutes": 0,
        "other_minutes": 0,
        "missing_minutes": 0,
    }


def _compute_l2book_coin(coin_path: Path, archive_path: Path | None = None) -> dict[str, Any]:
    n_files = 0
    total_bytes = 0
    oldest_day = ""
    newest_day = ""
    hours_present: set[tuple[str, str]] = set()

    # Scan both local and archive directories
    scan_dirs = [coin_path]
    if archive_path is not None and archive_path.is_dir():
        scan_dirs.append(archive_path)

    for scan_dir in scan_dirs:
        try:
            with os.scandir(str(scan_dir)) as it:
                for fe in it:
                    nm = fe.name.lower()
                    if not (nm.endswith(".lz4") or nm.endswith(".jsonl")):
                        continue
                    parsed = _parse_day_hour_from_filename(fe.name)
                    if not parsed:
                        continue
                    day_s, hour = parsed
                    if not day_s:
                        continue
                    n_files += 1
                    try:
                        total_bytes += fe.stat().st_size
                    except Exception:
                        pass
                    if hour is not None:
                        hours_present.add((day_s, str(hour)))
                    if not oldest_day or day_s < oldest_day:
                        oldest_day = day_s
                    if not newest_day or day_s > newest_day:
                        newest_day = day_s
        except Exception:
            pass

    n_days = 0
    expected_hours = 0
    coverage_pct = 0.0
    missing_days_count = 0
    missing_days_sample = ""

    try:
        if oldest_day and newest_day:
            dt0 = datetime.strptime(oldest_day, "%Y%m%d").date()
            dt1 = datetime.strptime(newest_day, "%Y%m%d").date()
            n_days = (dt1 - dt0).days + 1
            expected_hours = n_days * 24
            if expected_hours > 0:
                coverage_pct = round((len(hours_present) / float(expected_hours)) * 100.0, 2)

            hours_by_day: dict[str, int] = {}
            for day_s, _ in hours_present:
                hours_by_day[day_s] = hours_by_day.get(day_s, 0) + 1

            m_days: list[str] = []
            cur = dt0
            while cur <= dt1:
                day_s = cur.strftime("%Y%m%d")
                if hours_by_day.get(day_s, 0) < 24:
                    m_days.append(day_s)
                cur = cur + timedelta(days=1)
            missing_days_count = len(m_days)
            if m_days:
                sample = ",".join(m_days[:10])
                if len(m_days) > 10:
                    sample += ",..."
                missing_days_sample = sample
    except Exception:
        pass

    return {
        "n_files": n_files,
        "total_bytes": total_bytes,
        "oldest_day": oldest_day,
        "newest_day": newest_day,
        "n_days": n_days,
        "expected_hours": expected_hours,
        "coverage_pct": coverage_pct,
        "missing_days_count": missing_days_count,
        "missing_days_sample": missing_days_sample,
        "hl_minutes": 0,
        "other_minutes": 0,
        "missing_minutes": 0,
    }


def _compute_pb7_coin(coin_path: Path) -> dict[str, Any]:
    n_files = 0
    total_bytes = 0
    oldest_day = ""
    newest_day = ""

    try:
        with os.scandir(str(coin_path)) as it:
            for fe in it:
                if not fe.name.endswith(".npy"):
                    continue
                day_s = _parse_pb7_cache_day_from_name(fe.name)
                if not day_s:
                    continue
                n_files += 1
                try:
                    total_bytes += fe.stat().st_size
                except Exception:
                    pass
                if not oldest_day or day_s < oldest_day:
                    oldest_day = day_s
                if not newest_day or day_s > newest_day:
                    newest_day = day_s
    except Exception:
        pass

    n_days = 0
    try:
        if oldest_day and newest_day:
            dt0 = datetime.strptime(oldest_day, "%Y%m%d").date()
            dt1 = datetime.strptime(newest_day, "%Y%m%d").date()
            if dt1 >= dt0:
                n_days = (dt1 - dt0).days + 1
    except Exception:
        pass

    return {
        "n_files": n_files,
        "total_bytes": total_bytes,
        "oldest_day": oldest_day,
        "newest_day": newest_day,
        "n_days": n_days,
        "expected_hours": 0,
        "coverage_pct": 0.0,
        "missing_days_count": 0,
        "missing_days_sample": "",
        "hl_minutes": 0,
        "other_minutes": 0,
        "missing_minutes": 0,
    }


def _compute_coin(
    exchange: str,
    dataset: str,
    coin: str,
    coin_path: Path,
    lag_minutes: int,
    tradfi_type_fn: Callable[[str], str] | None,
    expected_minutes_fn: "Callable[[str, Any], set[int]] | None" = None,
    archive_path: Path | None = None,
) -> dict[str, Any]:
    """Dispatch per-coin computation to the right function."""
    ds_l = str(dataset).strip().lower()
    if ds_l in ("1m", "candles_1m"):
        return _compute_1m_coin(exchange, coin, coin_path, lag_minutes, tradfi_type_fn, expected_minutes_fn)
    elif ds_l in ("1m_api", "candles_1m_api"):
        return _compute_1m_api_coin(coin_path)
    elif ds_l in ("l2book",):
        return _compute_l2book_coin(coin_path, archive_path=archive_path)
    elif ds_l.startswith(("pb7_cache:", "pb8_cache:")):
        return _compute_pb7_coin(coin_path)
    else:
        # Generic NPZ fallback
        return _compute_1m_api_coin(coin_path)


# ---------------------------------------------------------------------------
# Directory discovery
# ---------------------------------------------------------------------------

def _coin_dirs_for_dataset(exchange: str, dataset: str) -> dict[str, Path]:
    """Return {coin_name: coin_path} for all coin dirs of this exchange+dataset."""
    ds_l = str(dataset).strip().lower()

    if ds_l.startswith(("pb7_cache:", "pb8_cache:")):
        prefix = "pb8_cache:" if ds_l.startswith("pb8_cache:") else "pb7_cache:"
        tf = ds_l[len(prefix):]
        root = _get_pb8_root_dir() if prefix == "pb8_cache:" else _get_pb7_root_dir()
        if root is None:
            return {}
        base = root / "caches" / "ohlcv" / str(exchange) / tf
        if not base.is_dir():
            return {}
        result: dict[str, Path] = {}
        with os.scandir(str(base)) as it:
            for e in it:
                if e.is_dir(follow_symlinks=False):
                    result[e.name] = Path(e.path)
        return result

    # Normalise dataset name for disk lookup
    ds_disk = ds_l
    if ds_l == "l2book":
        # On disk it might be l2Book (capital B) — check both
        base_lower = get_exchange_raw_root_dir(exchange) / "l2book"
        base_upper = get_exchange_raw_root_dir(exchange) / "l2Book"
        base = base_upper if base_upper.is_dir() else base_lower
    else:
        base = get_exchange_raw_root_dir(exchange) / ds_disk

    if not base.is_dir():
        # Try without "candles_" prefix mapping
        for alt in ("1m", "candles_1m", "1m_api", "candles_1m_api"):
            alt_base = get_exchange_raw_root_dir(exchange) / alt
            if alt_base.is_dir() and alt == ds_disk:
                base = alt_base
                break

    if not base.is_dir():
        return {}

    result = {}
    with os.scandir(str(base)) as it:
        for e in it:
            if e.is_dir(follow_symlinks=False) and not e.name.endswith("_src"):
                result[e.name] = Path(e.path)

    # For l2book: also include coins that exist only in the NAS archive
    if ds_l == "l2book" and is_l2book_archive_enabled():
        archive_dir = load_l2book_archive_dir()
        if archive_dir:
            archive_base = Path(archive_dir)
            if archive_base.is_dir():
                try:
                    with os.scandir(str(archive_base)) as it:
                        for e in it:
                            if e.is_dir(follow_symlinks=False) and e.name not in result:
                                # Coin exists only in archive — create an empty local
                                # placeholder path (won't be scanned, archive scan
                                # handles the files)
                                result[e.name] = base / e.name
                except Exception:
                    pass

    return result


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def get_inventory(
    exchange: str,
    dataset: str,
    *,
    lag_minutes: int = 0,
    tradfi_type_fn: Callable[[str], str] | None = None,
    expected_minutes_fn: "Callable[[str, Any], set[int]] | None" = None,
    force_refresh: bool = False,
) -> list[dict]:
    """
    Return inventory rows for (exchange, dataset).

    On first call (cold DB): computes all coins — same runtime as before.
    On subsequent calls: only recomputes coins whose dir mtime changed.
    force_refresh=True: recompute all coins regardless of mtime.
    """
    ex = str(exchange or "").strip().lower()
    ds = str(dataset or "").strip().lower()

    coin_dirs = _coin_dirs_for_dataset(ex, ds)
    if not coin_dirs:
        return []

    # For l2book: build archive path lookup
    is_l2book = ds in ("l2book",)
    l2book_archive_base: Path | None = None
    if is_l2book and is_l2book_archive_enabled():
        adir = load_l2book_archive_dir()
        if adir:
            l2book_archive_base = Path(adir)
            if not l2book_archive_base.is_dir():
                l2book_archive_base = None

    # Fast mtime read for all coin dirs (< 1ms for 227 dirs)
    live_mtimes: dict[str, float] = {}
    for coin, cpath in coin_dirs.items():
        mt = _dir_mtime(cpath)
        # For l2book with archive: combine local + archive mtime
        if l2book_archive_base is not None:
            archive_coin = l2book_archive_base / coin
            mt = max(mt, _dir_mtime(archive_coin))
        live_mtimes[coin] = mt

    # For 1m datasets: also check source-index mtime
    is_1m = ds in ("1m", "candles_1m")
    live_src_mtimes: dict[str, float] = {}
    if is_1m:
        for coin in coin_dirs:
            live_src_mtimes[coin] = _src_mtime(ex, coin)

    with _get_conn() as conn:
        # Load all cached rows for this exchange+dataset
        cur = conn.execute(
            "SELECT coin, dir_mtime, src_mtime, n_files, total_bytes, oldest_day, newest_day, "
            "n_days, expected_hours, coverage_pct, missing_days_count, missing_days_sample, "
            "hl_minutes, other_minutes, missing_minutes "
            "FROM inventory WHERE exchange=? AND dataset=?",
            (ex, ds),
        )
        cached: dict[str, dict] = {}
        for row in cur.fetchall():
            (coin, dir_mt, src_mt, n_files, total_bytes, oldest_day, newest_day,
             n_days, expected_hours, coverage_pct, missing_days_count, missing_days_sample,
             hl_minutes, other_minutes, missing_minutes) = row
            cached[coin] = {
                "dir_mtime": float(dir_mt),
                "src_mtime": float(src_mt),
                "n_files": int(n_files),
                "total_bytes": int(total_bytes),
                "oldest_day": oldest_day,
                "newest_day": newest_day,
                "n_days": int(n_days),
                "expected_hours": float(expected_hours),
                "coverage_pct": float(coverage_pct),
                "missing_days_count": int(missing_days_count),
                "missing_days_sample": missing_days_sample,
                "hl_minutes": int(hl_minutes),
                "other_minutes": int(other_minutes),
                "missing_minutes": int(missing_minutes),
            }

        # Find coins that need recomputation.
        # UI design principle: the GUI only reads from cache; background workers
        # are responsible for keeping the cache up to date via refresh_coin().
        # Therefore: cached coins are NEVER recomputed here (unless force_refresh=True).
        # Only coins that are not yet in the cache get computed (cold-start).
        coins_to_recompute: list[str] = []
        coins_to_delete: list[str] = []

        for coin in coin_dirs:
            if force_refresh or coin not in cached:
                coins_to_recompute.append(coin)

        # Coins in cache but no longer on disk → remove
        for coin in list(cached.keys()):
            if coin not in coin_dirs:
                coins_to_delete.append(coin)

        # Recompute changed coins
        upsert_params: list[tuple] = []
        for coin in coins_to_recompute:
            cpath = coin_dirs[coin]
            # Resolve archive path for l2book coins
            arc_path: Path | None = None
            if l2book_archive_base is not None:
                arc_path = l2book_archive_base / coin
            try:
                computed = _compute_coin(ex, ds, coin, cpath, lag_minutes, tradfi_type_fn, expected_minutes_fn, archive_path=arc_path)
            except Exception:
                computed = {
                    "n_files": 0, "total_bytes": 0, "oldest_day": "", "newest_day": "",
                    "n_days": 0, "expected_hours": 0, "coverage_pct": 0.0,
                    "missing_days_count": 0, "missing_days_sample": "",
                    "hl_minutes": 0, "other_minutes": 0, "missing_minutes": 0,
                }
            lmt = live_mtimes[coin]
            lsmt = live_src_mtimes.get(coin, 0.0)
            upsert_params.append((
                ex, ds, coin,
                lmt, lsmt,
                int(computed["n_files"]),
                int(computed["total_bytes"]),
                str(computed["oldest_day"]),
                str(computed["newest_day"]),
                int(computed["n_days"]),
                float(computed["expected_hours"]),
                float(computed["coverage_pct"]),
                int(computed["missing_days_count"]),
                str(computed["missing_days_sample"]),
                int(computed["hl_minutes"]),
                int(computed["other_minutes"]),
                int(computed["missing_minutes"]),
            ))
            # Update local cache dict so we return fresh data
            cached[coin] = {
                "dir_mtime": lmt,
                "src_mtime": lsmt,
                **computed,
            }

        if upsert_params:
            conn.executemany(
                "INSERT OR REPLACE INTO inventory VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
                upsert_params,
            )
        if coins_to_delete:
            conn.executemany(
                "DELETE FROM inventory WHERE exchange=? AND dataset=? AND coin=?",
                [(ex, ds, c) for c in coins_to_delete],
            )
            for c in coins_to_delete:
                cached.pop(c, None)
        if upsert_params or coins_to_delete:
            conn.commit()

    # Build result list — same schema as summarize_raw_inventory rows
    # + hl_minutes, other_minutes, missing_minutes columns
    rows: list[dict] = []
    for coin, data in sorted(cached.items()):
        if coin not in coin_dirs:
            continue  # was deleted
        rows.append(
            {
                "exchange": ex,
                "dataset": ds,
                "coin": coin,
                "n_files": data["n_files"],
                "total_bytes": data["total_bytes"],
                "oldest_day": data["oldest_day"],
                "newest_day": data["newest_day"],
                "n_days": data["n_days"],
                "expected_hours": data["expected_hours"],
                "coverage_pct": data["coverage_pct"],
                "missing_days_count": data["missing_days_count"],
                "missing_days_sample": data["missing_days_sample"],
                "hl_minutes": data["hl_minutes"],
                "other_minutes": data["other_minutes"],
                "missing_minutes": data["missing_minutes"],
            }
        )
    return rows


def invalidate_exchange(exchange: str) -> None:
    """Remove all cached rows for an exchange — triggers full recompute on next get_inventory."""
    ex = str(exchange or "").strip().lower()
    with _get_conn() as conn:
        conn.execute("DELETE FROM inventory WHERE exchange=?", (ex,))
        conn.commit()


def invalidate_coin(exchange: str, dataset: str, coin: str) -> None:
    """Remove a single cached row — triggers recompute for that coin on next get_inventory."""
    ex = str(exchange or "").strip().lower()
    ds = str(dataset or "").strip().lower()
    cn = str(coin or "").strip()
    with _get_conn() as conn:
        conn.execute(
            "DELETE FROM inventory WHERE exchange=? AND dataset=? AND coin=?",
            (ex, ds, cn),
        )
        conn.commit()


def sweep_cache_mtimes() -> dict:
    """Check all cached (exchange, dataset, coin) rows against current dir mtimes.

    For each cached coin:
      - If the directory no longer exists → remove from cache (invalidate).
      - If dir_mtime (or src_mtime for 1m) changed → recompute via refresh_coin().
      - Otherwise → leave as-is (unchanged).

    Returns a dict: {"refreshed": int, "deleted": int, "unchanged": int}
    """
    refreshed = 0
    deleted = 0
    unchanged = 0

    # Load all cached rows grouped by (exchange, dataset)
    with _get_conn() as conn:
        cur = conn.execute(
            "SELECT exchange, dataset, coin, dir_mtime, src_mtime FROM inventory"
        )
        rows = cur.fetchall()

    # Group by (exchange, dataset) to avoid calling _coin_dirs_for_dataset repeatedly
    from collections import defaultdict as _defaultdict
    groups: dict[tuple, list] = _defaultdict(list)
    for ex, ds, coin, dir_mt, src_mt in rows:
        groups[(ex, ds)].append((coin, float(dir_mt), float(src_mt)))

    for (ex, ds), coins in groups.items():
        is_1m = ds in ("1m", "candles_1m")
        coin_dirs = _coin_dirs_for_dataset(ex, ds)

        for coin, cached_dir_mt, cached_src_mt in coins:
            cpath = coin_dirs.get(coin)

            if cpath is None:
                # Directory gone — remove from cache
                invalidate_coin(ex, ds, coin)
                deleted += 1
                continue

            cur_dir_mt = _dir_mtime(cpath)
            cur_src_mt = _src_mtime(ex, coin) if is_1m else 0.0

            mtime_changed = (
                abs(cur_dir_mt - cached_dir_mt) > 0.001
                or (is_1m and abs(cur_src_mt - cached_src_mt) > 0.001)
            )
            if mtime_changed:
                try:
                    refresh_coin(ex, ds, coin)
                    refreshed += 1
                except Exception:
                    pass
            else:
                unchanged += 1

    return {"refreshed": refreshed, "deleted": deleted, "unchanged": unchanged}


def refresh_coin(
    exchange: str,
    dataset: str,
    coin: str,
    lag_minutes: int = 0,
    tradfi_type_fn: "Callable[[str], str] | None" = None,
    expected_minutes_fn: "Callable[[str, Any], set[int]] | None" = None,
) -> None:
    """Recompute and persist inventory for a single coin immediately.

    Called by the task worker after finishing a coin so the UI reads
    fresh data without needing its own recompute.
    """
    ex = str(exchange or "").strip().lower()
    ds = str(dataset or "").strip().lower()
    cn = str(coin or "").strip()
    coin_dirs = _coin_dirs_for_dataset(ex, ds)
    cpath = coin_dirs.get(cn)
    if cpath is None:
        return
    lmt = _dir_mtime(cpath)
    is_1m = ds in ("1m", "candles_1m")
    lsmt = _src_mtime(ex, cn) if is_1m else 0.0
    # Resolve archive path for l2book
    arc_path: Path | None = None
    if ds in ("l2book",) and is_l2book_archive_enabled():
        adir = load_l2book_archive_dir()
        if adir:
            arc_base = Path(adir)
            if arc_base.is_dir():
                arc_path = arc_base / cn
                lmt = max(lmt, _dir_mtime(arc_path))
    try:
        computed = _compute_coin(ex, ds, cn, cpath, lag_minutes, tradfi_type_fn, expected_minutes_fn, archive_path=arc_path)
    except Exception:
        return
    with _get_conn() as conn:
        conn.execute(
            """
            INSERT OR REPLACE INTO inventory
              (exchange, dataset, coin, dir_mtime, src_mtime,
               n_files, total_bytes, oldest_day, newest_day, n_days,
               expected_hours, coverage_pct, missing_days_count, missing_days_sample,
               hl_minutes, other_minutes, missing_minutes)
            VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
            """,
            (
                ex, ds, cn, lmt, lsmt,
                int(computed["n_files"]), int(computed["total_bytes"]),
                str(computed["oldest_day"]), str(computed["newest_day"]),
                int(computed["n_days"]), float(computed["expected_hours"]),
                float(computed["coverage_pct"]), int(computed["missing_days_count"]),
                str(computed["missing_days_sample"]),
                int(computed["hl_minutes"]), int(computed["other_minutes"]),
                int(computed["missing_minutes"]),
            ),
        )
        conn.commit()