refresh.py

"""Ingest orchestration (``stacks refresh``) and diagnostics (``stacks doctor``).

`refresh` walks the read-only ingest path for the configured sources (or the demo
world), unifies the result, and writes it to the persisted :class:`~ingest.store.Store`.
It skips re-ingesting when the source files' mtimes are unchanged, so the
dashboard stays cheap. `doctor` validates configuration and read-only access
without mutating anything — the human-facing preflight check.

Live kosync is only used when fully configured; tests drive the real-source path
with on-disk SQLite fixtures and no network.

Kosync progress used to be fetched one sequential HTTP GET per book, *inside*
:func:`ingest.unify.unify`, with every failure silently swallowed — an N+1 with
no visible outcome. :func:`fetch_progress` replaces that: it batches every
non-empty stat key into a single bounded-concurrency step, run once per
refresh, before ``unify`` ever runs, with a captured ok/no-progress/error
outcome per key. ``unify`` now just reads the resulting in-memory map.
"""

from __future__ import annotations

import concurrent.futures
import datetime as dt
import os
import tempfile
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional

from ingest.calibre import load_library
from ingest.config import KNOWN_STACKS_ENV, Config
from ingest.kobo import load_stats as load_kobo_stats
from ingest.koreader import load_daily_activity, load_stats
from ingest.kosync import FixtureKosync, ProgressSource
from ingest.models import DailyActivity, DeviceProgress, ReadingStat, ReadingState
from ingest.snapshot import columns, has_sidecar, open_snapshot
from ingest.store import ORIGIN_DEMO, ORIGIN_REAL, CatalogSourceUpdate, Store
from ingest.unify import unify


@dataclass(frozen=True)
class ProgressOutcome:
    """The captured result of fetching kosync progress for one stat key.

    Replaces the old blanket ``except Exception: return ()`` in ``unify`` —
    every key now has a visible outcome instead of a silent fallback.
    """

    key: str
    ok: bool  # False only on a transport/parse error fetching this key
    found: bool  # True if the source returned progress for this key
    error: str = ""  # non-empty iff ``ok`` is False


@dataclass(frozen=True)
class ProgressFetchResult:
    """The output of :func:`fetch_progress` — a resolved, in-memory progress map."""

    progress: dict[str, DeviceProgress] = field(default_factory=dict)
    outcomes: tuple[ProgressOutcome, ...] = ()
    fetched: int = 0
    errors: int = 0


def fetch_progress(
    source: Optional[ProgressSource],
    keys: Iterable[str],
    *,
    max_workers: int = 8,
) -> ProgressFetchResult:
    """Fetch kosync progress for ``keys`` with bounded concurrency.

    Each key gets its own :class:`ProgressOutcome` — ok/no-progress/error —
    instead of a blanket swallow. Dispatch and result assembly are both sorted
    by key, so the returned map and outcome order are deterministic regardless
    of which fetch happens to finish first (preserving the reproducibility
    gate even though fetches run concurrently).
    """
    sorted_keys = sorted({k for k in keys if k})
    if source is None or not sorted_keys:
        return ProgressFetchResult()

    def _fetch_one(key: str) -> tuple[str, Optional[DeviceProgress], Optional[str]]:
        try:
            dp = source.progress_for(key)
        except Exception as exc:  # noqa: BLE001 - captured as a visible outcome, not swallowed
            return key, None, f"{type(exc).__name__}: {exc}"
        return key, dp, None

    workers = max(1, min(max_workers, len(sorted_keys)))
    with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as pool:
        # .map yields results in input order regardless of completion order,
        # so assembly below stays deterministic under concurrency.
        results = list(pool.map(_fetch_one, sorted_keys))

    progress: dict[str, DeviceProgress] = {}
    outcomes: list[ProgressOutcome] = []
    fetched = 0
    errors = 0
    for key, dp, error in results:
        if error is not None:
            outcomes.append(ProgressOutcome(key=key, ok=False, found=False, error=error))
            errors += 1
        elif dp is not None:
            progress[key] = dp
            outcomes.append(ProgressOutcome(key=key, ok=True, found=True))
            fetched += 1
        else:
            outcomes.append(ProgressOutcome(key=key, ok=True, found=False))

    return ProgressFetchResult(
        progress=progress, outcomes=tuple(outcomes), fetched=fetched, errors=errors
    )


def _stat_signature(stat: ReadingStat) -> str:
    """A cheap fingerprint of a stat's local reading state, for cache invalidation.

    If this hasn't changed since the last successful kosync fetch for the same
    key, the cached progress is reused instead of re-fetching.
    """
    return f"{stat.last_read_ts}:{stat.pages_read}:{stat.read_time_seconds}:{stat.sessions}"


def _resolve_progress(
    source: Optional[ProgressSource],
    stats: list[ReadingStat],
    store: Optional[Store],
    now: int,
    ttl_seconds: int = 15 * 60,
) -> ProgressFetchResult:
    """Fetch fresh kosync progress only for keys whose stat changed; reuse the rest.

    Without a store (e.g. a bare call to :func:`ingest_states`), there is
    nothing to cache against, so every key is fetched fresh.
    """
    signatures = {stat.key: _stat_signature(stat) for stat in stats if stat.key}
    if not signatures:
        if store is not None:
            store.save_progress({}, {}, fetched_at=now, fetched_keys=set())
        return ProgressFetchResult()
    if store is None:
        return fetch_progress(source, signatures.keys())

    stale = store.stale_progress_keys(signatures, now=now, ttl_seconds=ttl_seconds)
    fresh = fetch_progress(source, stale) if stale else ProgressFetchResult()

    cached = store.cached_progress()
    reused = {key: dp for key, dp in cached.items() if key in signatures and key not in stale}
    errored = {outcome.key for outcome in fresh.outcomes if not outcome.ok}
    last_good_after_error = {
        key: cached[key] for key in errored if key in cached and key in signatures
    }
    merged = {**reused, **last_good_after_error, **fresh.progress}
    # A key whose fetch errored retains any last-good value but is persisted
    # with an explicit retry marker. This prevents an all-key outage from
    # becoming an empty cache that the top-level mtime guard treats as fresh.
    fetched_keys = {outcome.key for outcome in fresh.outcomes if outcome.ok}
    store.save_progress(
        merged,
        signatures,
        fetched_at=now,
        fetched_keys=fetched_keys,
        failed_keys=errored,
    )

    return ProgressFetchResult(
        progress=merged, outcomes=fresh.outcomes, fetched=len(merged), errors=fresh.errors
    )


@dataclass(frozen=True)
class RefreshResult:
    refreshed: bool
    n_states: int
    refreshed_at: int
    reason: str
    progress_fetched: int = 0
    progress_errors: int = 0
    progress_outcomes: tuple[ProgressOutcome, ...] = ()
    catalog_attempted: int = 0
    catalog_succeeded: int = 0
    catalog_errors: int = 0
    catalog_candidates: int = 0


def source_mtimes(config: Config) -> dict[str, int]:
    """Integer mtimes of the configured source files that currently exist."""
    out: dict[str, int] = {}
    for name, path in (
        ("calibre", config.calibre_db),
        ("koreader", config.koreader_db),
        ("kobo", config.kobo_db),
    ):
        if path is not None and path.is_file():
            out[name] = int(path.stat().st_mtime)
    return out


def _ingest_demo(
    config: Config, store: Optional[Store] = None, now: int = 0
) -> tuple[list[ReadingState], list[DailyActivity], ProgressFetchResult]:
    from ingest.demo import DEMO_RETRIEVED_AT, build_demo_dbs, demo_kosync

    demo_dir = config.data_dir / "demo"
    metadata_db, statistics_db = build_demo_dbs(demo_dir)
    snap = config.snapshot_dir
    books = load_library(metadata_db, snap, retrieved_at=DEMO_RETRIEVED_AT)
    stats = load_stats(statistics_db, snap)
    activity = load_daily_activity(statistics_db, snap)
    progress_result = _resolve_progress(
        demo_kosync(), stats, store, now, config.kosync_progress_ttl_seconds
    )
    states = unify(books, stats, FixtureKosync(progress_result.progress))
    return states, activity, progress_result


def _retrieval_date(now: int) -> str:
    """The UTC date stamped onto tags sourced from the real library.

    Sourced descriptors carry a citation, and a citation needs the date it was
    read. ``now`` is supplied by the caller (never read from the clock here) so
    ingest stays deterministic under test. A caller that passes ``0`` gets the
    epoch back, which is the honest "no retrieval time was supplied" sentinel
    rather than a plausible-looking guess.
    """
    return dt.datetime.fromtimestamp(now, tz=dt.UTC).strftime("%Y-%m-%d")


def _ingest_real(
    config: Config, store: Optional[Store] = None, now: int = 0
) -> tuple[list[ReadingState], list[DailyActivity], ProgressFetchResult]:
    snap = config.snapshot_dir
    retrieved_at = _retrieval_date(now)
    books = (
        load_library(config.calibre_db, snap, retrieved_at=retrieved_at)
        if config.calibre_db
        else []
    )
    stats = load_stats(config.koreader_db, snap) if config.koreader_db else []
    stats = stats + (load_kobo_stats(config.kobo_db, snap) if config.kobo_db else [])
    activity = load_daily_activity(config.koreader_db, snap) if config.koreader_db else []
    progress_result = _resolve_progress(
        _kosync(config), stats, store, now, config.kosync_progress_ttl_seconds
    )
    states = unify(books, stats, FixtureKosync(progress_result.progress))
    return states, activity, progress_result


def _kosync(config: Config):  # type: ignore[no-untyped-def]
    if not config.kosync_configured:
        return None
    from ingest.kosync import KosyncClient  # pragma: no cover - constructed only in real deploys

    return KosyncClient(  # pragma: no cover
        username=config.kosync_user or "",
        userkey_md5=config.kosync_key or "",
        host=config.kosync_host or "",
    )


def _ingest_with_progress(
    config: Config, store: Optional[Store] = None, now: int = 0
) -> tuple[list[ReadingState], list[DailyActivity], ProgressFetchResult]:
    return _ingest_demo(config, store, now) if config.demo else _ingest_real(config, store, now)


def ingest_states(
    config: Config, store: Optional[Store] = None, now: int = 0
) -> tuple[list[ReadingState], list[DailyActivity]]:
    """Run the read-only ingest path for the demo world or the real sources."""
    states, activity, _progress = _ingest_with_progress(config, store, now)
    return states, activity


def refresh(config: Config, store: Store, now: int, *, force: bool = False) -> RefreshResult:
    """Re-ingest into the store, skipping when source mtimes are unchanged."""
    from recommender.catalog_pool import (
        clear_legacy_response_cache,
        configured_source_ids,
        fetch_catalog_pool,
    )

    current = source_mtimes(config)
    catalog_ids = set(configured_source_ids(config))
    clear_legacy_response_cache(config)
    store.save_catalog_mode(config.catalog_outbound_mode, catalog_ids)
    progress_due = config.kosync_configured and store.progress_refresh_due(
        now, config.kosync_progress_ttl_seconds
    )
    stored_catalog_ids = {status.source_id for status in store.catalog_source_statuses()}
    catalog_due = (
        config.catalog_egress_enabled
        and bool(catalog_ids)
        and (
            catalog_ids != stored_catalog_ids
            or store.catalog_refresh_due(now, config.catalog_refresh_ttl_seconds)
        )
    )
    # The skip is only safe when the *stored* state also came from the real
    # libraries. A demo refresh writes to the same store path, so without the
    # origin check a real refresh that follows one sees a populated store whose
    # mtimes match and reports "sources unchanged" — leaving fixture books in
    # place and serving them as the reader's own library. Unrecorded origin
    # (a store written before FIX-STATE-ORIGIN) re-ingests once, which costs a
    # fraction of a second and cannot be wrong.
    unchanged = (
        not force
        and not config.demo
        and store.is_populated
        and store.state_origin() == ORIGIN_REAL
        and bool(current)
        and current == store.source_mtimes()
        and not progress_due
        and not catalog_due
    )
    if unchanged:
        states = store.load_states()
        return RefreshResult(
            refreshed=False,
            n_states=len(states),
            refreshed_at=store.refreshed_at() or 0,
            reason="sources unchanged since last refresh",
            catalog_candidates=store.catalog_pool_status().candidate_count,
        )

    states, activity, progress_result = _ingest_with_progress(config, store, now)
    # Demo states are stamped with no source mtimes at all: the real files'
    # mtimes describe libraries that had no part in producing these books, and
    # persisting them would assert a lineage the fixtures do not have.
    store.save(
        states,
        activity,
        refreshed_at=now,
        source_mtimes={} if config.demo else current,
        origin=ORIGIN_DEMO if config.demo else ORIGIN_REAL,
    )
    catalog_result = None
    if config.demo:
        from ingest.demo import demo_candidates

        demo_books = tuple(candidate.book for candidate in demo_candidates())
        store.save_catalog_refresh(
            (CatalogSourceUpdate(source_id="demo:built-in", books=demo_books),),
            active_source_ids={"demo:built-in"},
            attempted_at=now,
            outbound_mode="off",
        )
    elif config.catalog_egress_enabled and catalog_ids and (force or catalog_due):
        catalog_result = fetch_catalog_pool(config)
        store.save_catalog_refresh(
            catalog_result.updates,
            active_source_ids=set(catalog_result.active_source_ids),
            attempted_at=now,
            outbound_mode=config.catalog_outbound_mode,
        )
    pool_status = store.catalog_pool_status()
    return RefreshResult(
        refreshed=True,
        n_states=len(states),
        refreshed_at=now,
        reason="demo world" if config.demo else "ingested from sources",
        progress_fetched=progress_result.fetched,
        progress_errors=progress_result.errors,
        progress_outcomes=progress_result.outcomes,
        catalog_attempted=catalog_result.attempted if catalog_result else 0,
        catalog_succeeded=catalog_result.succeeded if catalog_result else 0,
        catalog_errors=catalog_result.errors if catalog_result else 0,
        catalog_candidates=pool_status.candidate_count,
    )


# --- doctor -----------------------------------------------------------------


@dataclass(frozen=True)
class Check:
    name: str
    ok: bool
    detail: str


def _check_source(label: str, path: Optional[Path], required_table: str) -> list[Check]:
    """Validate one source the way ``stacks refresh`` will actually read it.

    Snapshot-first, through :func:`~ingest.snapshot.open_snapshot`, into a
    temporary directory that is discarded immediately. This used to call
    ``open_readonly`` on the live path, whose ``immutable=1`` skips WAL recovery
    — so against a Calibre library that was open on the other screen (the setup
    the README describes, and the moment a reader runs ``doctor``) it reported a
    healthy library's main table as *missing*. The check was less accurate than
    the thing it was diagnosing.

    Going through the real entry point also means doctor now verifies that the
    source can be snapshotted consistently at all, which is the step ``refresh``
    would fail on.
    """
    checks: list[Check] = []
    if path is None:
        checks.append(Check(f"{label} configured", False, "no path set (demo or unconfigured)"))
        return checks
    if not path.is_file():
        checks.append(Check(f"{label} file", False, f"not found: {path}"))
        return checks
    checks.append(Check(f"{label} file", True, str(path)))
    if has_sidecar(path):
        # Not a problem — say so, or a reader will read the extra line as one.
        checks.append(
            Check(
                f"{label} in use",
                True,
                "a -wal/-journal sidecar is present, so the library is open in "
                "another program right now; reads go through a consistent "
                "snapshot, so this is fine",
            )
        )
    try:
        with (
            tempfile.TemporaryDirectory(prefix="stacks-doctor-") as tmp,
            open_snapshot(path, Path(tmp)) as conn,
        ):
            has_table = bool(columns(conn, required_table))
        checks.append(
            Check(
                f"{label} read-only access",
                has_table,
                "snapshotted and opened read-only; "
                f"'{required_table}' table {'found' if has_table else 'missing'}",
            )
        )
    except Exception as exc:  # noqa: BLE001 - surface any access problem to the user
        checks.append(Check(f"{label} read-only access", False, f"{type(exc).__name__}: {exc}"))
    return checks


def _check_env(env: Mapping[str, str]) -> list[Check]:
    """Flag unrecognized ``STACKS_*`` variables — typos are silently ignored otherwise."""
    unknown = sorted(k for k in env if k.startswith("STACKS_") and k not in KNOWN_STACKS_ENV)
    return [
        Check(f"env {key}", False, "unknown STACKS_* variable — ignored (typo?)") for key in unknown
    ]


def doctor(
    config: Config,
    store: Optional[Store] = None,
    env: Optional[Mapping[str, str]] = None,
) -> list[Check]:
    """Validate configuration + read-only access; never mutates anything."""
    resolved_env: Mapping[str, str] = os.environ if env is None else env
    checks: list[Check] = []
    checks.append(
        Check("mode", True, "demo (built-in offline library)" if config.demo else "real sources")
    )
    if not config.demo:
        checks.extend(_check_source("Calibre", config.calibre_db, "books"))
        checks.extend(_check_source("KOReader", config.koreader_db, "book"))
        if config.kobo_db is not None:
            checks.extend(_check_source("Kobo", config.kobo_db, "content"))
        checks.append(
            Check(
                "kosync",
                True,
                "configured (host + user + key present)"
                if config.kosync_configured
                else "not configured — progress will use KOReader stats only",
            )
        )
        if not config.catalog_egress_enabled:
            checks.append(
                Check(
                    "catalog egress",
                    True,
                    "off — no public catalog requests will be made",
                )
            )
        elif not config.catalog_sources_configured:
            checks.append(
                Check(
                    "catalog egress",
                    False,
                    "public-metadata consent enabled, but no broad subjects/"
                    "public lists configured",
                )
            )
        else:
            checks.append(
                Check(
                    "catalog egress",
                    True,
                    "public-metadata only; broad predeclared sources, "
                    "never reading-derived queries",
                )
            )
    checks.append(Check("data dir", True, str(config.data_dir)))
    if store is not None:
        if store.is_populated:
            checks.append(
                Check("app-state store", True, f"populated; refreshed_at={store.refreshed_at()}")
            )
        else:
            checks.append(Check("app-state store", True, "empty — run `stacks refresh`"))
        for status in store.catalog_source_statuses():
            has_fallback = status.candidate_count > 0 and status.fetched_at is not None
            if status.status == "error":
                detail = f"last attempt failed ({status.error})"
                if has_fallback:
                    detail += (
                        f"; serving {status.candidate_count} last-good candidates "
                        f"fetched_at={status.fetched_at}"
                    )
                checks.append(Check(f"catalog {status.source_id}", False, detail))
            else:
                checks.append(
                    Check(
                        f"catalog {status.source_id}",
                        True,
                        f"{status.candidate_count} candidates; fetched_at={status.fetched_at}",
                    )
                )
    checks.extend(_check_env(resolved_env))
    return checks