"""TTL-aware S3 ingestion cache.

Registers a persistent-cache ``cloudpathlib.S3Client`` as the process-global
default for ``s3://`` URIs, so any ``CloudPath("s3://...")`` /
``AnyPath("s3://...")`` materializes through a project-local cache dir instead
of re-downloading every run. Source-change correctness is cloudpathlib's
built-in ETag check; disk hygiene is a wall-clock TTL enforced by a single
startup sweep.

No subclass, no per-access hook. Two functions set up the cache. The cache
dir, TTL, and (optional) region come from ``get_settings().s3_cache``, and both
functions accept explicit overrides for tests:

- ``evict_stale_s3_cache()`` — startup sweep. Walks the cache dir and unlinks
  any file whose access time is older than the TTL. This is the *only* TTL
  reclamation path: cloudpathlib's ETag check handles source correctness; the
  sweep handles "downloaded once, never touched again" files that would
  otherwise live in the cache forever.
- ``init_s3_cache()`` — package-import entrypoint. Runs the sweep, then
  constructs and registers the default S3 client. Called once from
  ``quber/__init__.py``.

Three more functions resolve inputs against it. ``resolve_document`` and
``resolve_artifacts`` read ``s3://`` sources through the cache. ``fetch_file``
copies a source to a caller-named path and bypasses the cache.

TTL is keyed off ``st_atime``, so re-processing a cached document resets its
window: actively used files stay warm, abandoned ones age out after the TTL.
On ``noatime``-mounted volumes atime is frozen and TTL degrades to "time since
download" — accepted rather than adding mount-detection complexity.
"""

from __future__ import annotations

import time
from pathlib import Path
from typing import Sequence

from cloudpathlib import AnyPath, S3Client, S3Path
from loguru import logger

from quber.settings import get_settings


def evict_stale_s3_cache(
    cache_dir: Path | None = None,
    ttl_seconds: float | None = None,
) -> int:
    """Delete cached files whose atime is older than ``ttl_seconds``.

    ``cache_dir`` / ``ttl_seconds`` default to ``get_settings().s3_cache`` when
    not passed explicitly. Walks ``cache_dir`` once with ``rglob('*')`` and
    unlinks regular files past the TTL. Returns the count evicted. Idempotent
    and safe to call when the cache directory does not exist yet (returns 0).
    Linear in the cached-file count; sub-second even at thousands.
    """
    s3 = get_settings().s3_cache
    cache_dir = cache_dir if cache_dir is not None else s3.cache_dir
    ttl_seconds = ttl_seconds if ttl_seconds is not None else s3.ttl_seconds
    if not cache_dir.exists():
        return 0
    cutoff = time.time() - ttl_seconds
    evicted = 0
    for p in cache_dir.rglob("*"):
        if p.is_file() and p.stat().st_atime < cutoff:
            p.unlink()
            evicted += 1
    return evicted


def init_s3_cache(
    cache_dir: Path | None = None,
    ttl_seconds: float | None = None,
) -> None:
    """Prune the stale cache, then register the default S3 client.

    Called once at package import (see ``quber/__init__.py``). ``cache_dir`` /
    ``ttl_seconds`` default to ``get_settings().s3_cache``. Two ordered steps:

    1. ``evict_stale_s3_cache()`` — bound disk usage for keys not re-accessed
       between runs.
    2. Construct a persistent-cache ``S3Client`` rooted at ``cache_dir`` and
       publish it via ``set_as_default_client()`` so plain
       ``CloudPath("s3://...")`` / ``AnyPath("s3://...")`` picks it up. When
       ``s3_cache.region`` is set, the client is bound to a region-scoped boto3
       session; otherwise region resolves from boto3's own chain.

    Constructing the client requires neither AWS credentials nor network access
    (boto3 resolves both lazily on first GET), so this is safe to run at import
    on any host, including CPU-only / credential-less dev environments.
    """
    s3 = get_settings().s3_cache
    cache_dir = cache_dir if cache_dir is not None else s3.cache_dir
    ttl_seconds = ttl_seconds if ttl_seconds is not None else s3.ttl_seconds
    evicted = evict_stale_s3_cache(cache_dir, ttl_seconds)
    if evicted:
        logger.debug("Evicted {} stale file(s) from S3 cache at {}", evicted, cache_dir)
    if s3.region:
        import boto3

        session = boto3.Session(region_name=s3.region)
        client = S3Client(local_cache_dir=cache_dir, file_cache_mode="persistent", boto3_session=session)
    else:
        client = S3Client(local_cache_dir=cache_dir, file_cache_mode="persistent")
    client.set_as_default_client()


def resolve_document(raw: str, cache_dir: Path | None = None) -> Path:
    """Coerce a user-supplied path string to a local filesystem Path.

    Local inputs pass straight through. For ``s3://`` inputs, log an INFO line
    when the document is already present in the local cache (cache hit), then
    materialize it (cloudpathlib downloads on miss) and return the local Path.
    ``cache_dir`` defaults to ``get_settings().s3_cache.cache_dir``.

    The cache-hit check mirrors cloudpathlib's on-disk layout
    (``<cache_dir>/<bucket>/<key>``) using only public ``S3Path`` attributes.
    Note: cloudpathlib still revalidates the S3 ETag on access, so a logged hit
    means "present locally"; if the source object changed upstream, cloudpathlib
    transparently re-downloads the fresh copy.
    """
    doc = AnyPath(raw)
    if isinstance(doc, S3Path):
        cache_dir = cache_dir if cache_dir is not None else get_settings().s3_cache.cache_dir
        cached = cache_dir / doc.bucket / doc.key
        if cached.exists():
            logger.info("S3 cache hit: {} served from {}", doc, cached)
        else:
            logger.debug("S3 cache miss: downloading {} to {}", doc, cached)
    return Path(doc)  # __fspath__ materializes / revalidates an S3Path; no-op for a local Path


def fetch_file(src: str, dest: Path) -> Path:
    """Copy a local path or s3:// URI to `dest`, bypassing the cache. Returns `dest`.

    Unlike `resolve_document`, the caller names the destination — used where a
    file must land at a specific staged path rather than wherever the cache
    keeps it.
    """
    dest.parent.mkdir(parents=True, exist_ok=True)
    src_path = AnyPath(src)
    if isinstance(src_path, S3Path):
        logger.info("Downloading {src} -> {dest}", src=src, dest=dest)
        src_path.download_to(dest)
    else:
        dest.write_bytes(Path(src).read_bytes())
    return dest


def resolve_artifacts(raw: str, filenames: Sequence[str], cache_dir: Path | None = None) -> Path:
    """Coerce an artifacts-directory string to a local directory Path.

    A local directory passes straight through. For an ``s3://`` prefix, each
    named artifact file is materialized through the persistent cache — exactly
    ``resolve_document`` per file, with the same hit/miss logging and ETag
    revalidation — and the returned Path is the local cache directory that
    mirrors the prefix, so a caller written against a directory of files
    (``ParseResult.load``) reads the cached copies in place.

    ``filenames`` must name every file the caller will read: only the named
    files are downloaded, nothing else under the prefix.
    """
    if not filenames:
        raise ValueError("resolve_artifacts: filenames must name at least one artifact file")
    directory = AnyPath(raw)
    if not isinstance(directory, S3Path):
        return Path(raw)
    files = [resolve_document(str(directory / name), cache_dir) for name in filenames]
    return files[0].parent