"""ObjectScale / S3 storage API for Command Center."""

from __future__ import annotations

import json as _json
import os
import random as _rnd
import threading as _threading
import time
from collections import deque
from datetime import datetime, timezone
from typing import Any

import boto3
from botocore.client import Config
from botocore.exceptions import ClientError
from fastapi import APIRouter, Query
from fastapi.responses import JSONResponse, StreamingResponse

S3_ENDPOINT = os.getenv("S3_ENDPOINT", "http://10.0.20.111:9020")
S3_ACCESS_KEY = os.getenv("S3_ACCESS_KEY", "object_admin1")
S3_SECRET_KEY = os.getenv("S3_SECRET_KEY", "ChangeMeChangeMeChangeMeChangeMeChangeMe")
S3_REGION = os.getenv("S3_REGION", "us-east-1")

router = APIRouter(prefix="/api/storage/s3", tags=["storage"])

# Bounded scan so the dashboard stays responsive on big / fast-growing buckets.
SCAN_MAX_OBJECTS = int(os.getenv("S3_SCAN_MAX_OBJECTS", "80000"))
SCAN_DEADLINE_S = float(os.getenv("S3_SCAN_DEADLINE_S", "20"))
# Per top-level prefix cap so a single huge prefix (e.g. kafka/ CDC json) cannot
# starve the others — this keeps the type/size composition representative.
SCAN_PER_PREFIX = int(os.getenv("S3_SCAN_PER_PREFIX", "20000"))

# In-process activity log — every S3 op routed through this API is recorded so
# the dashboard can show live "what is happening on the storage" timelines.
_activity: deque[tuple[float, str]] = deque(maxlen=20000)
_analytics_cache: dict[str, Any] = {"ts": 0.0, "data": None}
_ANALYTICS_TTL = 45.0

# ── Pipeline → S3 archiver ────────────────────────────────────────────────────
# When data is generated, the streaming pipeline must actually land objects in
# S3 so the dashboard reflects it (Last write / growth / activity). We mirror two
# real stages straight into the object store:
#   • Kafka → S3 CDC archive   -> cdc-archive/dt=YYYY-MM-DD/events-*.json (raw NDJSON)
#   • Spark → S3 curated layer -> curated/sales_orders_masked/dt=…/part-*.json (PII masked)
# Writes are buffered like a Kafka-Connect S3 sink (flush on size or interval) so
# we don't create a flood of tiny objects, and force-flushed on a manual burst.
S3_ARCHIVE_BUCKET = os.getenv("S3_ARCHIVE_BUCKET", "data")
S3_FLUSH_INTERVAL = float(os.getenv("S3_ARCHIVE_FLUSH_S", "12"))
S3_FLUSH_SIZE = int(os.getenv("S3_ARCHIVE_FLUSH_SIZE", "400"))

_arch_lock = _threading.Lock()
_arch_buf: dict[str, list] = {"cdc": [], "curated": []}
_arch_since: dict[str, float] = {"t": 0.0}
_last_write: dict[str, Any] = {}
_arch_stats: dict[str, int] = {"objects": 0, "bytes": 0, "rows": 0}


def _iso(v: Any) -> str:
    return v.isoformat() if hasattr(v, "isoformat") else str(v)


def _mask_cust(cid: Any) -> str:
    h = abs(hash(("cust", cid))) % 0xFFFFFF
    return f"cust_{h:06x}***"


def _order_to_cdc(t: tuple, ts: str) -> dict[str, Any]:
    cid, pid, region, channel, ots, amt, curr, status = t
    return {"op": "c", "source": "postgres", "db": "sales", "table": "sales_orders", "ts": ts,
            "after": {"customer_id": cid, "product_id": pid, "region": region, "sales_channel": channel,
                      "amount": amt, "currency": curr, "order_status": status, "order_ts": _iso(ots)}}


def _order_to_curated(t: tuple, ts: str) -> dict[str, Any]:
    cid, pid, region, channel, ots, amt, curr, status = t
    return {"customer_ref": _mask_cust(cid), "product_id": pid, "region": region, "sales_channel": channel,
            "amount": amt, "currency": curr, "order_status": status, "order_ts": _iso(ots),
            "ingested_ts": ts, "pii_masked": True}


def _hr_to_cdc(t: tuple, ts: str) -> dict[str, Any]:
    eid, dept, role, region, evt, sal, ets = t
    return {"op": "c", "source": "mysql", "db": "hr", "table": "employee_events", "ts": ts,
            "after": {"employee_id": eid, "department": dept, "role_name": role, "region": region,
                      "event_type": evt, "salary_change": sal, "event_ts": _iso(ets)}}


def _supply_to_cdc(d: dict, ts: str) -> dict[str, Any]:
    return {"op": "c", "source": "mongodb", "db": "supplychain", "table": "events", "ts": ts, "after": dict(d)}


def _tel_to_cdc(t: tuple, ts: str) -> dict[str, Any]:
    dev, mts, mtype, mval, _payload = t
    return {"op": "c", "source": "cassandra", "db": "telemetry", "table": "device_metrics", "ts": ts,
            "after": {"device_id": dev, "metric_ts": _iso(mts), "metric_type": mtype, "metric_value": mval}}


def _put(s3, bucket: str, key: str, body: bytes, content_type: str) -> None:
    s3.put_object(Bucket=bucket, Key=key, Body=body, ContentType=content_type)
    _track("write")
    _last_write.update({"ts": datetime.now(timezone.utc).isoformat(), "mono": time.time(),
                        "bucket": bucket, "key": key, "bytes": len(body)})
    _arch_stats["objects"] += 1
    _arch_stats["bytes"] += len(body)


def _flush_locked() -> list[dict[str, Any]] | None:
    cdc = _arch_buf["cdc"]
    cur = _arch_buf["curated"]
    if not cdc and not cur:
        return None
    s3 = _client()
    now = datetime.now(timezone.utc)
    day = now.strftime("%Y-%m-%d")
    ms = int(now.timestamp() * 1000)
    rid = _rnd.randint(1000, 9999)
    written: list[dict[str, Any]] = []
    if cdc:
        body = ("\n".join(_json.dumps(e, default=str) for e in cdc) + "\n").encode()
        key = f"cdc-archive/dt={day}/events-{ms}-{rid}.json"
        _put(s3, S3_ARCHIVE_BUCKET, key, body, "application/x-ndjson")
        written.append({"stage": "kafka→s3", "key": key, "rows": len(cdc), "bytes": len(body)})
    if cur:
        body = ("\n".join(_json.dumps(e, default=str) for e in cur) + "\n").encode()
        key = f"curated/sales_orders_masked/dt={day}/part-{ms}-{rid}.json"
        _put(s3, S3_ARCHIVE_BUCKET, key, body, "application/x-ndjson")
        written.append({"stage": "spark→s3", "key": key, "rows": len(cur), "bytes": len(body)})
    _arch_buf["cdc"] = []
    _arch_buf["curated"] = []
    _arch_since["t"] = time.time()
    return written


def archive_generated_batch(orders_rows=None, hr_rows=None, supply_docs=None, tel_rows=None,
                            *, force: bool = False) -> list[dict[str, Any]] | dict[str, Any] | None:
    """Stage a freshly generated batch into S3 (Kafka→S3 CDC archive + Spark→S3
    curated masked). Buffered; flushes on size/interval or when force=True."""
    try:
        with _arch_lock:
            ts = datetime.now(timezone.utc).isoformat()
            for t in (orders_rows or []):
                _arch_buf["cdc"].append(_order_to_cdc(t, ts))
                _arch_buf["curated"].append(_order_to_curated(t, ts))
                _arch_stats["rows"] += 1
            for t in (hr_rows or []):
                _arch_buf["cdc"].append(_hr_to_cdc(t, ts)); _arch_stats["rows"] += 1
            for d in (supply_docs or []):
                _arch_buf["cdc"].append(_supply_to_cdc(d, ts)); _arch_stats["rows"] += 1
            for t in (tel_rows or []):
                _arch_buf["cdc"].append(_tel_to_cdc(t, ts)); _arch_stats["rows"] += 1
            if _arch_since["t"] == 0.0:
                _arch_since["t"] = time.time()
            buffered = len(_arch_buf["cdc"]) + len(_arch_buf["curated"])
            age = time.time() - _arch_since["t"]
            if force or buffered >= S3_FLUSH_SIZE or age >= S3_FLUSH_INTERVAL:
                return _flush_locked()
    except Exception as exc:  # never break the generator on an S3 hiccup
        return {"error": str(exc)}
    return None


def put_object_bytes(key: str, body: bytes, content_type: str = "application/octet-stream",
                     bucket: str | None = None) -> dict[str, Any]:
    """Write raw bytes to S3 (used by the ETL offload agent for Parquet parts).
    Tracks last-write + activity so the storage dashboard reflects it live."""
    b = bucket or S3_ARCHIVE_BUCKET
    s3 = _client()
    _put(s3, b, key, body, content_type)
    return {"ok": True, "bucket": b, "key": key, "bytes": len(body)}


def archive_active(window_s: float = 25.0) -> bool:
    """True if the pipeline wrote to S3 recently — drives the kafka→S3 edge pulse."""
    return (time.time() - float(_last_write.get("mono") or 0.0)) < window_s


def archive_info() -> dict[str, Any]:
    return {"last_write": dict(_last_write) or None, "objects": _arch_stats["objects"],
            "bytes": _arch_stats["bytes"], "rows": _arch_stats["rows"]}


def _client():
    return boto3.client(
        "s3",
        endpoint_url=S3_ENDPOINT,
        aws_access_key_id=S3_ACCESS_KEY,
        aws_secret_access_key=S3_SECRET_KEY,
        region_name=S3_REGION,
        config=Config(signature_version="s3v4", s3={"addressing_style": "path"}),
    )


def _human_size(n: float) -> str:
    for unit in ("B", "KB", "MB", "GB", "TB"):
        if n < 1024:
            return f"{n:.0f} {unit}" if unit == "B" else f"{n:.1f} {unit}"
        n /= 1024
    return f"{n:.1f} PB"


def _track(op: str) -> None:
    try:
        _activity.append((time.time(), op))
    except Exception:
        pass


def _ext(key: str) -> str:
    base = key.rsplit("/", 1)[-1]
    if "." in base:
        e = base.rsplit(".", 1)[-1].lower()
        if 1 <= len(e) <= 8 and e.isalnum():
            return e
    return "(none)"


def _size_class(n: int) -> str:
    kb, mb, gb = 1024, 1024 ** 2, 1024 ** 3
    if n < kb:
        return "<1 KB"
    if n < mb:
        return "1 KB–1 MB"
    if n < 10 * mb:
        return "1–10 MB"
    if n < 100 * mb:
        return "10–100 MB"
    if n < gb:
        return "100 MB–1 GB"
    return ">1 GB"


@router.get("/health")
async def s3_health():
    try:
        _track("health")
        s3 = _client()
        buckets = s3.list_buckets()
        names = [b["Name"] for b in buckets.get("Buckets", [])]
        return {
            "ok": True,
            "endpoint": S3_ENDPOINT,
            "buckets": len(names),
            "bucket_names": names,
        }
    except Exception as exc:
        return JSONResponse({"ok": False, "endpoint": S3_ENDPOINT, "error": str(exc)}, status_code=502)


@router.get("/buckets")
async def list_buckets():
    try:
        _track("list_buckets")
        s3 = _client()
        resp = s3.list_buckets()
        items = []
        for b in resp.get("Buckets", []):
            name = b["Name"]
            try:
                loc = s3.list_objects_v2(Bucket=name, MaxKeys=1)
                count_hint = loc.get("KeyCount", 0)
            except ClientError:
                count_hint = None
            items.append({
                "name": name,
                "created": b.get("CreationDate", "").isoformat() if b.get("CreationDate") else None,
                "has_objects": bool(count_hint),
            })
        return {"ok": True, "buckets": items, "endpoint": S3_ENDPOINT}
    except Exception as exc:
        return JSONResponse({"ok": False, "error": str(exc)}, status_code=502)


@router.get("/buckets/{bucket}/objects")
async def list_objects(
    bucket: str,
    prefix: str = Query("", alias="prefix"),
    max_keys: int = Query(200, le=1000),
    recursive: bool = Query(False),
    q: str = Query(""),
    ext: str = Query(""),
    token: str = Query(""),
):
    """Browse a bucket.

    - Folder mode (default): one level, returns sub-folders + objects.
    - recursive / q / ext: flat search across the whole prefix subtree,
      paginated via `token` (returned as `next_token`).
    """
    try:
        _track("list_objects")
        s3 = _client()
        ql = q.lower().strip()
        ext_l = ext.lower().lstrip(".").strip()
        flat = bool(recursive or ql or ext_l)
        folders: list[dict[str, Any]] = []
        objects: list[dict[str, Any]] = []
        cont = token or None
        pages = 0
        scanned = 0
        next_token: str | None = None
        PAGE_BUDGET = 40

        while True:
            kw: dict[str, Any] = {"Bucket": bucket, "Prefix": prefix, "MaxKeys": 1000}
            if not flat:
                kw["Delimiter"] = "/"
            if cont:
                kw["ContinuationToken"] = cont
            r = s3.list_objects_v2(**kw)
            pages += 1

            for p in r.get("CommonPrefixes", []):
                folders.append({
                    "type": "prefix",
                    "name": p["Prefix"][len(prefix):].rstrip("/"),
                    "prefix": p["Prefix"],
                })

            for o in r.get("Contents", []):
                key = o["Key"]
                if key == prefix:
                    continue
                scanned += 1
                if ql and ql not in key.lower():
                    continue
                if ext_l and _ext(key) != ext_l:
                    continue
                objects.append({
                    "type": "object",
                    "key": key,
                    "name": key[len(prefix):] if key.startswith(prefix) else key,
                    "ext": _ext(key),
                    "size": o.get("Size", 0),
                    "size_human": _human_size(o.get("Size", 0)),
                    "modified": o.get("LastModified", "").isoformat() if o.get("LastModified") else None,
                })

            cont = r.get("NextContinuationToken")
            more = r.get("IsTruncated")

            if not flat:
                next_token = cont if more else None
                break
            if len(objects) >= max_keys or not more or pages >= PAGE_BUDGET:
                next_token = cont if more else None
                break

        return {
            "ok": True,
            "bucket": bucket,
            "prefix": prefix,
            "flat": flat,
            "folders": folders,
            "objects": objects[:max_keys] if flat else objects,
            "scanned": scanned,
            "next_token": next_token,
            "truncated": bool(next_token),
        }
    except ClientError as exc:
        return JSONResponse({"ok": False, "error": str(exc)}, status_code=403)
    except Exception as exc:
        return JSONResponse({"ok": False, "error": str(exc)}, status_code=502)


_MEDIA_BY_EXT = {
    "jpeg": "image/jpeg", "jpg": "image/jpeg", "png": "image/png", "gif": "image/gif",
    "webp": "image/webp", "bmp": "image/bmp", "svg": "image/svg+xml",
}


@router.get("/buckets/{bucket}/download")
async def download_object(bucket: str, key: str = Query(...), inline: bool = Query(False)):
    try:
        _track("download")
        s3 = _client()
        obj = s3.get_object(Bucket=bucket, Key=key)
        body = obj["Body"]
        filename = key.split("/")[-1] or "download"
        ext_media = _MEDIA_BY_EXT.get(_ext(key))
        stored = obj.get("ContentType")
        # ECS often stores everything as octet-stream — trust the extension for
        # known image types so previews render inline instead of downloading.
        media = ext_media or stored or "application/octet-stream"
        if ext_media and (not stored or stored == "application/octet-stream"):
            media = ext_media

        def stream():
            while chunk := body.read(1024 * 256):
                yield chunk

        disp = "inline" if inline else "attachment"
        return StreamingResponse(
            stream(),
            media_type=media,
            headers={
                "Content-Disposition": f'{disp}; filename="{filename}"',
                "Cache-Control": "private, max-age=300",
            },
        )
    except ClientError as exc:
        return JSONResponse({"ok": False, "error": str(exc)}, status_code=404)


@router.get("/buckets/{bucket}/preview")
async def preview_object(bucket: str, key: str = Query(...), max_bytes: int = Query(131072, le=1048576)):
    """Return the first chunk of an object as text for inline preview."""
    try:
        _track("preview")
        s3 = _client()
        head = None
        try:
            head = s3.head_object(Bucket=bucket, Key=key)
        except Exception:
            pass
        total = int(head.get("ContentLength", 0)) if head else 0
        obj = s3.get_object(Bucket=bucket, Key=key, Range=f"bytes=0-{max_bytes - 1}")
        raw = obj["Body"].read()
        truncated = (total > len(raw)) or (len(raw) >= max_bytes)
        text = raw.decode("utf-8", errors="replace")
        return {
            "ok": True,
            "bucket": bucket,
            "key": key,
            "ext": _ext(key),
            "size": total or len(raw),
            "size_human": _human_size(total or len(raw)),
            "bytes_read": len(raw),
            "truncated": truncated,
            "content": text,
        }
    except ClientError as exc:
        return JSONResponse({"ok": False, "error": str(exc)}, status_code=404)
    except Exception as exc:
        return JSONResponse({"ok": False, "error": str(exc)}, status_code=502)


# ── Analytics ────────────────────────────────────────────────────────────────


def _accumulate(agg: dict[str, Any], bucket: str, o: dict[str, Any]) -> None:
    sz = int(o.get("Size", 0) or 0)
    key = o["Key"]
    lm = o.get("LastModified")
    agg["objects"] += 1
    agg["bytes"] += sz
    bb = agg["by_bucket"].setdefault(bucket, [0, 0])
    bb[0] += 1
    bb[1] += sz
    e = agg["by_ext"].setdefault(_ext(key), [0, 0])
    e[0] += 1
    e[1] += sz
    sc = _size_class(sz)
    agg["by_size"][sc] = agg["by_size"].get(sc, 0) + 1
    if lm:
        d = lm.astimezone(timezone.utc).strftime("%Y-%m-%d")
        dd = agg["by_day"].setdefault(d, [0, 0])
        dd[0] += 1
        dd[1] += sz
        agg["recent"].append((lm.isoformat(), bucket, key, sz))
        if len(agg["recent"]) > 400:
            agg["recent"] = sorted(agg["recent"], reverse=True)[:60]
    rest = key[len(bucket) + 1:] if key.startswith(bucket + "/") else key
    top = rest.split("/", 1)[0] if "/" in rest else "(root)"
    pp = agg["by_prefix"].setdefault(f"{bucket}/{top}", [0, 0])
    pp[0] += 1
    pp[1] += sz
    agg["largest"].append((sz, bucket, key, lm.isoformat() if lm else None))
    if len(agg["largest"]) > 400:
        agg["largest"] = sorted(agg["largest"], reverse=True)[:60]


def _scan_prefix(s3, bucket: str, prefix: str, cap: int, deadline: float, agg: dict[str, Any]) -> bool:
    """Scan one prefix subtree (flat). Returns True if capped/cut short."""
    tok = None
    n = 0
    while True:
        if time.time() > deadline or n >= cap or agg["objects"] >= SCAN_MAX_OBJECTS:
            return True
        kw: dict[str, Any] = {"Bucket": bucket, "MaxKeys": 1000}
        if prefix:
            kw["Prefix"] = prefix
        if tok:
            kw["ContinuationToken"] = tok
        try:
            r = s3.list_objects_v2(**kw)
        except Exception:
            return False
        for o in r.get("Contents", []):
            _accumulate(agg, bucket, o)
            n += 1
            if n >= cap or agg["objects"] >= SCAN_MAX_OBJECTS:
                return True
        if r.get("IsTruncated"):
            tok = r.get("NextContinuationToken")
            if not tok:
                return False
        else:
            return False


def _scan_analytics() -> dict[str, Any]:
    s3 = _client()
    deadline = time.time() + SCAN_DEADLINE_S
    resp = s3.list_buckets()
    bucket_objs = resp.get("Buckets", [])
    created_map = {
        b["Name"]: (b.get("CreationDate").isoformat() if b.get("CreationDate") else None)
        for b in bucket_objs
    }
    names = [b["Name"] for b in bucket_objs]

    agg: dict[str, Any] = {
        "objects": 0, "bytes": 0, "by_bucket": {}, "by_ext": {}, "by_size": {},
        "by_day": {}, "by_prefix": {}, "largest": [], "recent": [],
    }
    truncated = False

    for name in names:
        if time.time() > deadline:
            truncated = True
            break
        try:
            r = s3.list_objects_v2(Bucket=name, Delimiter="/", MaxKeys=1000)
        except Exception:
            continue
        # root-level objects
        for o in r.get("Contents", []):
            _accumulate(agg, name, o)
        units = [p["Prefix"] for p in r.get("CommonPrefixes", [])]
        agg["by_bucket"].setdefault(name, [0, 0])
        if not units:
            if _scan_prefix(s3, name, "", SCAN_MAX_OBJECTS, deadline, agg):
                truncated = True
        else:
            # breadth-first: every top-level prefix gets its own budget so a
            # single huge prefix can't hide the rest of the data.
            for pre in units:
                if time.time() > deadline:
                    truncated = True
                    break
                if _scan_prefix(s3, name, pre, SCAN_PER_PREFIX, deadline, agg):
                    truncated = True

    total_bytes = agg["bytes"]
    total_objects = agg["objects"]

    growth = []
    cum_b = cum_o = 0
    for d in sorted(agg["by_day"]):
        c, b = agg["by_day"][d]
        cum_o += c
        cum_b += b
        growth.append({"date": d, "objects": c, "bytes": b, "cum_objects": cum_o, "cum_bytes": cum_b})

    buckets_out = sorted(
        [{"name": n, "objects": v[0], "bytes": v[1], "size_human": _human_size(v[1]),
          "pct": round(100 * v[1] / total_bytes, 1) if total_bytes else 0,
          "created": created_map.get(n)} for n, v in agg["by_bucket"].items()],
        key=lambda x: x["bytes"], reverse=True)

    types_out = sorted(
        [{"ext": k, "objects": v[0], "bytes": v[1], "size_human": _human_size(v[1])}
         for k, v in agg["by_ext"].items()], key=lambda x: x["objects"], reverse=True)[:12]

    size_order = ["<1 KB", "1 KB–1 MB", "1–10 MB", "10–100 MB", "100 MB–1 GB", ">1 GB"]
    size_out = [{"label": k, "count": agg["by_size"].get(k, 0)} for k in size_order]

    prefixes_out = sorted(
        [{"prefix": k, "objects": v[0], "bytes": v[1], "size_human": _human_size(v[1])}
         for k, v in agg["by_prefix"].items()], key=lambda x: x["bytes"], reverse=True)[:12]

    largest = sorted(agg["largest"], reverse=True)[:10]
    largest_out = [{"bucket": b, "key": k, "bytes": s, "size_human": _human_size(s), "modified": m}
                   for s, b, k, m in largest]

    recent = sorted(agg["recent"], reverse=True)[:15]
    recent_out = [{"modified": m, "bucket": b, "key": k, "bytes": s, "size_human": _human_size(s)}
                  for m, b, k, s in recent]

    return {
        "summary": {
            "buckets": len(names),
            "objects": total_objects,
            "bytes": total_bytes,
            "size_human": _human_size(total_bytes),
            "avg_object_bytes": int(total_bytes / total_objects) if total_objects else 0,
            "avg_object_human": _human_size(int(total_bytes / total_objects) if total_objects else 0),
            "largest_human": largest_out[0]["size_human"] if largest_out else "0 B",
            "newest": recent_out[0]["modified"] if recent_out else None,
            "oldest": growth[0]["date"] if growth else None,
            "truncated": truncated,
            "scanned": total_objects,
        },
        "buckets": buckets_out,
        "types": types_out,
        "size_histogram": size_out,
        "growth": growth,
        "top_prefixes": prefixes_out,
        "largest_objects": largest_out,
        "recent": recent_out,
    }


def _activity_view() -> dict[str, Any]:
    now = time.time()
    cutoff = now - 3600
    mins = [0] * 60
    by_op: dict[str, int] = {}
    total = 0
    for ts, op in list(_activity):
        if ts < cutoff:
            continue
        total += 1
        by_op[op] = by_op.get(op, 0) + 1
        idx = int((now - ts) // 60)
        if 0 <= idx < 60:
            mins[59 - idx] += 1
    last_ts = _activity[-1][0] if _activity else None
    return {
        "per_minute": mins,
        "by_op": [{"op": k, "count": v} for k, v in sorted(by_op.items(), key=lambda x: -x[1])],
        "total_last_hour": total,
        "last_activity": datetime.fromtimestamp(last_ts, tz=timezone.utc).isoformat() if last_ts else None,
    }


@router.get("/analytics")
async def analytics(refresh: bool = Query(False)):
    """Aggregated storage analytics for the dashboard (cached ~45s)."""
    _track("analytics")
    now = time.time()
    if not refresh and _analytics_cache["data"] is not None and now - _analytics_cache["ts"] < _ANALYTICS_TTL:
        data = _analytics_cache["data"]
    else:
        try:
            data = _scan_analytics()
            _analytics_cache["data"] = data
            _analytics_cache["ts"] = now
        except Exception as exc:
            return JSONResponse({"ok": False, "error": str(exc), "endpoint": S3_ENDPOINT}, status_code=502)

    # Overlay the live last-write so the dashboard reflects pipeline writes
    # immediately, without waiting for the (bounded, cached) full rescan.
    summary = dict(data.get("summary") or {})
    recent = list(data.get("recent") or [])
    lw = dict(_last_write)
    if lw.get("ts"):
        if not summary.get("newest") or lw["ts"] > summary["newest"]:
            summary["newest"] = lw["ts"]
        recent = ([{"modified": lw["ts"], "bucket": lw.get("bucket"), "key": lw.get("key"),
                    "bytes": lw.get("bytes", 0), "size_human": _human_size(lw.get("bytes", 0))}]
                  + [r for r in recent if r.get("key") != lw.get("key")])[:15]
    return {"ok": True, "endpoint": S3_ENDPOINT, "generated_at": datetime.now(timezone.utc).isoformat(),
            "activity": _activity_view(), **data, "summary": summary, "recent": recent,
            "archive": archive_info()}