"""Read-only access to the parsed bucket, plus the caching that keeps it quick.

Everything here is GetObject/ListObjectsV2 against bsd-wpe-logs-parsed. Nothing
in this app writes to S3, creates AWS resources, or touches bucket settings.

The shape we read:
    summary/date=YYYY-MM-DD.json                    small, one per day
    events/date=YYYY-MM-DD/site=<domain>/<f>.jsonl  one JSON object per line

HOW THIS STAYS FAST

The overview only ever reads the summary files (9 small objects), so it is
effectively instant.

The detail table works out which shards can possibly match the filters and
fetches only those. A date narrows it to that day's ~110 shards; a site narrows
it to one. Shards are fetched in parallel and cached in memory, so paging
through results costs nothing after the first load. The worst case -- every day,
no site filter -- is ~160MB and takes a while the first time, then is cached.

Events are cached as plain tuples rather than dicts, and the repeated strings
are interned, which keeps the whole dataset in memory at a sane size.
"""

import array
import collections
import gzip
import json
import os
import sys
import threading
import time
from concurrent.futures import ThreadPoolExecutor
from datetime import date, timedelta

import boto3
from botocore.config import Config
from botocore.exceptions import ClientError

from categories import category_for
from content import is_content
from sites import install_from_source_key, install_from_stem

BUCKET = os.environ.get("PARSED_BUCKET", "bsd-wpe-logs-parsed")
REGION = os.environ.get("AWS_REGION", "us-east-1")

# Written by the parser; lists every source log file it has processed, including
# the empty ones. This is where "sites we are capturing logs for" comes from --
# see tracked_installs().
MANIFEST_KEY = "manifest/processed.json"

# How many shard reads run at once. This is the detail page's main cost -- it has
# to fetch the actual event rows -- and the work is I/O-bound (each thread spends
# almost all its time waiting on an S3 GET, releasing the GIL), so more readers
# scale it up cleanly. Measured on the latest day (151 shards): 16 -> 4.9s,
# 32 -> 3.2s, 48 -> 2.1s, 64 -> 2.1s. 48 is the knee; past it S3 latency, not
# concurrency, is the floor. One shared pool caps this for the whole process no
# matter how many requests are in flight (see _executor), so this is the total
# S3 concurrency, not per-request. Override on a small box if it feels like too
# much for the droplet.
_WORKERS = int(os.environ.get("VIEWER_FETCH_WORKERS", "48"))
# Bounds memory. Must comfortably exceed the working set of one query, or the
# cache thrashes: a query would warm N shards, evict most of them, then re-fetch
# them while iterating. After the 2026-07-17 backfill tripled the data, one
# week is ~2,300 shards -- the old cap of 500 turned a 20s page into minutes.
# The iteration below no longer depends on this being big enough, but paging
# through results still wants the range to stay resident.
#
# By 2026-07-20 a week had grown to ~3,250 shards and crossed the old 3,000 cap,
# so This Week thrashed again -- two overlapping scans evicting each other's
# shards. Keep this above a week's working set with headroom for growth.
_MAX_CACHED_SHARDS = 5000

# Field order for a cached event row.
TS, SITE, PLATFORM, GROUP, URL, STATUS = range(6)

# The set of dates and sites only changes when a nightly parse lands, but
# listing them costs several seconds -- far too much to repeat on every page
# load. Cache them briefly so new data still appears without a restart.
_LIST_TTL = 300  # seconds

_lock = threading.Lock()
_shard_cache = collections.OrderedDict()  # key -> [tuple, ...]
_summary_cache = {}                       # date -> dict
_ttl_cache = {}                           # name -> (expires_at, value)
_client_holder = threading.local()

# ONE pool for the whole process, shared by every request. The fan-out used to
# be a fresh ThreadPoolExecutor(_WORKERS) per call, so N concurrent requests ran
# N*_WORKERS threads against a connection pool sized for _WORKERS -- they starved
# each other and the whole server stalled. A single shared pool caps total S3
# concurrency at _WORKERS however many requests are in flight; the excess just
# queues. Created lazily, never at import, because a pool built before gunicorn
# forks its worker would hand the child dead threads and deadlock on first use.
_pool = None
_pool_lock = threading.Lock()


def _executor():
    """The process-wide fetch pool, built on first use (after any fork)."""
    global _pool
    if _pool is None:
        with _pool_lock:
            if _pool is None:
                _pool = ThreadPoolExecutor(max_workers=_WORKERS,
                                           thread_name_prefix="s3fetch")
    return _pool


def _cached(name, ttl, produce):
    """Memoise `produce()` under `name` for `ttl` seconds."""
    now = time.time()
    with _lock:
        hit = _ttl_cache.get(name)
        if hit and hit[0] > now:
            return hit[1]
    value = produce()
    with _lock:
        _ttl_cache[name] = (now + ttl, value)
    return value


def client():
    # botocore clients are thread-safe, but a client per thread avoids any
    # contention on the shared connection pool during a parallel fetch.
    got = getattr(_client_holder, "c", None)
    if got is None:
        got = boto3.client(
            "s3",
            region_name=REGION,
            config=Config(max_pool_connections=_WORKERS + 4, retries={"max_attempts": 3}),
        )
        _client_holder.c = got
    return got


def check_access():
    """Confirm we can read the bucket. Returns (ok, message)."""
    try:
        client().list_objects_v2(Bucket=BUCKET, Prefix="summary/", MaxKeys=1)
        return True, "ok"
    except ClientError as exc:
        return False, exc.response.get("Error", {}).get("Message", str(exc))
    except Exception as exc:  # credentials missing, etc.
        return False, str(exc)


def _list_keys(prefix, delimiter=None):
    out, prefixes = [], []
    paginator = client().get_paginator("list_objects_v2")
    kwargs = {"Bucket": BUCKET, "Prefix": prefix}
    if delimiter:
        kwargs["Delimiter"] = delimiter
    for page in paginator.paginate(**kwargs):
        out.extend(o["Key"] for o in page.get("Contents", []))
        prefixes.extend(p["Prefix"] for p in page.get("CommonPrefixes", []))
    return out, prefixes


def available_dates():
    """Every date with a summary file, newest first."""
    def produce():
        keys, _ = _list_keys("summary/date=")
        dates = []
        for k in keys:
            stem = k[len("summary/date="):]
            if stem.endswith(".json"):
                dates.append(stem[:-len(".json")])
        return sorted(dates, reverse=True)
    return _cached("dates", _LIST_TTL, produce)


def summary(day):
    """The daily summary dict, cached. None if that day has no summary."""
    with _lock:
        if day in _summary_cache:
            return _summary_cache[day]
    try:
        body = client().get_object(Bucket=BUCKET, Key="summary/date=%s.json" % day)["Body"].read()
    except ClientError:
        return None
    doc = json.loads(body)
    # by_url used to be ~99% of a summary (a per-URL count for every URL on every
    # site) and nothing here ever read it -- the detail page reads URLs from the
    # event shards. The parser no longer writes it, so this pop only trims a
    # summary written before that change and not yet rebuilt. The overview reads
    # summaries for every range now, so keeping them small is what keeps it fast.
    doc.pop("by_url", None)
    with _lock:
        _summary_cache[day] = doc
    return doc


def summaries(days):
    """Summaries for many days, fetched in parallel. Skips days with none."""
    if not days:
        return []
    with _lock:
        missing = [d for d in days if d not in _summary_cache]
    if missing:
        list(_executor().map(summary, missing))
    return [(d, summary(d)) for d in days if summary(d)]


def sites_for(days):
    """Sites that have data on the given days, for the filter dropdown.

    Scoped to the selected dates rather than the whole bucket: one delimited
    list per day, cached per day. Listing all nine days to populate a dropdown
    made the first detail page load take tens of seconds.
    """
    seen = set()
    for day in days:
        def produce(d=day):
            _, prefixes = _list_keys("events/date=%s/" % d, delimiter="/")
            out = []
            for p in prefixes:
                tail = p[len("events/date=%s/" % d):].rstrip("/")
                if tail.startswith("site="):
                    out.append(tail[len("site="):])
            return out
        seen.update(_cached("sites:" + day, _LIST_TTL, produce))
    return sorted(seen)


def all_sites():
    """Every site across every day. Slow (one list per day); prefer sites_for."""
    return sites_for(available_dates())


def tracked_installs():
    """Every install we are capturing logs for, whether or not it has hits.

    Read from the parser's manifest rather than by listing the source bucket:
    the manifest keys ARE the source filenames, so this is still "derived from
    the log filenames", but it keeps the viewer's reach inside the parsed bucket.
    That matters -- the source bucket holds raw request logs (IP plus browsed
    URL), and the viewer has no business being able to read them.

    A file with no AI hits still has a manifest entry, which is exactly what
    makes a zero-hit site nameable: an empty log has no vhost to read.

    Caveat, deliberately: this counts sites whose logs reach us. A site that has
    never had log export switched on has no file, no manifest entry, and cannot
    appear here. See the README.
    """
    def produce():
        try:
            body = client().get_object(Bucket=BUCKET, Key=MANIFEST_KEY)["Body"].read()
        except ClientError:
            return []
        try:
            doc = json.loads(body)
        except ValueError:
            return []
        found = set()
        for key in doc.get("processed", {}):
            install = install_from_source_key(key)
            if install:
                found.add(install)
        return sorted(found)
    return _cached("installs", _LIST_TTL, produce)


def _day_keys(day):
    """Every object key under one day's events/ prefix. Cached per day and shared
    by install_domains() and shard_keys(), so a range is listed once per TTL."""
    prefix = "events/date=%s/" % day
    return _cached("keys:" + prefix, _LIST_TTL, lambda p=prefix: _list_keys(p)[0])


def _warm_day_keys(days):
    """List every day in the range in parallel. One S3 list per day, and served
    serially a 90-day range was ~a minute of round-trips before the first paint
    -- the dominant cost now that the overview reads summaries, not shards. Warm
    (within the TTL) this is a no-op; cold it collapses the wait to one round."""
    if days:
        list(_executor().map(_day_keys, days))


def install_domains(days):
    """{install: {domain, ...}} for the given days, read from the event keys.

    Both halves of the pairing are in the key already -- the partition is the
    vhost, the stem is the install -- so this observes the mapping rather than
    guessing it. Listing is per-day and cached, and shares its cache entry with
    shard_keys(), so the overview pays for it once.
    """
    out = collections.defaultdict(set)
    _warm_day_keys(days)
    for day in days:
        prefix = "events/date=%s/" % day
        keys = _day_keys(day)
        for key in keys:
            if not key.endswith(".jsonl"):
                continue
            rest = key[len(prefix):]
            if not rest.startswith("site="):
                continue
            domain, _, stem = rest[len("site="):].partition("/")
            install = install_from_stem(stem[: -len(".jsonl")])
            if install:
                out[install].add(domain)
    return dict(out)


def domain_to_install(days):
    """{domain: install}, the inverse of install_domains().

    Safe as a plain dict: no domain is served by more than one install (checked
    against the whole bucket), so this cannot collide.
    """
    out = {}
    for install, domains in install_domains(days).items():
        for domain in domains:
            out[domain] = install
    return out


def dates_in_range(start, end):
    """Dates that exist AND fall in [start, end]."""
    have = set(available_dates())
    out = []
    try:
        d0 = date.fromisoformat(start)
        d1 = date.fromisoformat(end)
    except ValueError:
        return []
    if d1 < d0:
        d0, d1 = d1, d0
    cur = d0
    while cur <= d1:
        iso = cur.isoformat()
        if iso in have:
            out.append(iso)
        cur += timedelta(days=1)
    return out


def install_of_key(key):
    """The install a shard came from, read from its stem. None if unreadable."""
    stem = key.rsplit("/", 1)[-1]
    if not stem.endswith(".jsonl"):
        return None
    return install_from_stem(stem[: -len(".jsonl")])


def shard_keys(days, sites=None):
    """Only the shards that can match: scoped by date, and by install if given.

    `sites` holds installs, not domains -- one install can serve several vhosts
    and so several site= partitions, and all of them are the same site. Matching
    on the stem picks up every one of them without needing to know the domains.

    This lists one prefix per day and filters, rather than listing each install's
    partition. It is the same number of S3 calls (a day is well under one page)
    and the listing is then shared with install_domains() and the overview.
    """
    want = set(sites) if sites else None
    keys = []
    _warm_day_keys(days)
    for day in days:
        prefix = "events/date=%s/" % day
        got = _day_keys(day)
        for key in got:
            if not key.endswith(".jsonl"):
                continue
            if want is not None and install_of_key(key) not in want:
                continue
            keys.append(key)
    return keys


def _parse_shard(key):
    body = client().get_object(Bucket=BUCKET, Key=key)["Body"].read().decode("utf-8", "replace")
    rows = []
    for line in body.splitlines():
        if not line.strip():
            continue
        try:
            e = json.loads(line)
        except ValueError:
            continue  # a truncated line should not take down the page
        platform = sys.intern(str(e.get("ai_platform", "")))
        # The group is derived from the bot name, NOT read from e["category"] --
        # stored events predate the ai_search split. See categories.py.
        group = sys.intern(category_for(platform, e.get("category")))
        rows.append((
            e.get("timestamp", ""),
            sys.intern(str(e.get("site", ""))),
            platform,
            group,
            e.get("url", ""),
            e.get("status_code"),
        ))
    return rows


def load_shard(key):
    with _lock:
        hit = _shard_cache.get(key)
        if hit is not None:
            _shard_cache.move_to_end(key)
            return hit
    rows = _parse_shard(key)
    with _lock:
        _shard_cache[key] = rows
        _shard_cache.move_to_end(key)
        while len(_shard_cache) > _MAX_CACHED_SHARDS:
            _shard_cache.popitem(last=False)
    return rows


def _shard_rows(keys):
    """Yield (key, rows) per shard, fetched in parallel, each shard exactly once.

    pool.map keeps the results in order and hands them over as they finish, so
    callers can consume and discard rather than holding every shard at once. The
    key rides along because it carries the install -- the rows only know the
    vhost, which is not the site identity.
    """
    if not keys:
        return
    for key, rows in zip(keys, _executor().map(load_shard, keys)):
        yield key, rows


def load_shards(keys):
    """Fetch many shards in parallel, returning one flat list of rows."""
    out = []
    for _, rows in _shard_rows(keys):
        out.extend(rows)
    return out


def query(days, sites=None, platforms=None, groups=None, url_query="",
          content_only=False, limit=None):
    """Matching event rows, newest first, plus the totals for the whole match.

    Filters shard-by-shard rather than materialising every row and filtering
    afterwards, so a wide date range costs roughly what it matches, not what it
    scans. Returns (rows, total, by_group, by_platform).

    `content_only` drops assets, APIs, robots.txt and scanner probes -- see
    content.py. It filters here, before counting, so the counts and the table
    always agree.
    """
    keys = shard_keys(days, sites)
    if not keys:
        return [], 0, {}, {}

    want_groups = set(groups) if groups else None
    want_platforms = set(platforms) if platforms else None
    needle = (url_query or "").strip().lower()

    matched = []
    by_group = collections.Counter()
    by_platform = collections.Counter()
    # Consume each shard's rows as they arrive rather than warming the cache and
    # re-reading it. Re-reading meant a range larger than the cache re-fetched
    # every evicted shard mid-loop -- fetching the same data several times over.
    # This way each shard is fetched exactly once however big the range is.
    for _, rows in _shard_rows(keys):
        for row in rows:
            if want_groups is not None and row[GROUP] not in want_groups:
                continue
            if want_platforms is not None and row[PLATFORM] not in want_platforms:
                continue
            if content_only and not is_content(row[URL]):
                continue
            if needle and needle not in row[URL].lower():
                continue
            by_group[row[GROUP]] += 1
            by_platform[row[PLATFORM]] += 1
            matched.append(row)

    total = len(matched)
    matched.sort(key=lambda r: r[TS], reverse=True)
    if limit is not None:
        matched = matched[:limit]
    return matched, total, dict(by_group), dict(by_platform)


def rollup(days, groups=None, content_only=False):
    """Overview totals from the event shards: (by_platform, by_install).

    The daily summaries are pre-aggregated and carry no URL detail, so they
    cannot answer "how many of these were real pages?". When the content filter
    is on we have to count the events themselves. That is slower than reading a
    summary, which is why the overview still uses summaries when the filter is
    off, and why startup pre-warms the default range.

    Counts land on the install, not row[SITE]: a hit on buckfirelaw.com and one
    on buckfirelaw.wpengine.com are the same site and must add up, not appear as
    two rows.
    """
    keys = shard_keys(days)
    if not keys:
        return {}, {}

    want = set(groups) if groups else None
    by_platform = collections.Counter()
    by_install = collections.Counter()
    for key, rows in _shard_rows(keys):
        install = install_of_key(key)
        for row in rows:
            if want is not None and row[GROUP] not in want:
                continue
            if content_only and not is_content(row[URL]):
                continue
            by_platform[row[PLATFORM]] += 1
            if install:
                by_install[install] += 1
    return dict(by_platform), dict(by_install)


def cache_stats():
    with _lock:
        shards = len(_shard_cache)
        events = sum(len(v) for v in _shard_cache.values())
    return {"shards": shards, "events": events}


# --- per-URL daily aggregates, for the trend page ---------------------------
#
# The parser writes urlagg/date=YYYY-MM-DD.json.gz next to each summary (see
# handler.rebuild_summary): per-day hit counts keyed by URL, with (install,
# platform) triples underneath. One small gzipped file per day is what lets the
# trend page answer "how often was this URL fetched each day of the last 90
# days?" without touching the event shards -- the same trick the overview plays
# with the summaries.

# A day is held as parallel arrays over small integer ids, not per-row tuples.
# The tuple version of this cache measured ~40MB per day (a busy day is 140k
# URLs and 200k triples; per-triple tuples and their string refs add up) --
# 1.8GB resident for the current 41 days, times however many gunicorn workers.
# Columnar it is ~5MB a day: URLs are interned once process-wide (they repeat
# heavily day to day), and installs/platforms shrink to array('i') ids.
#
#   day -> (urls,     list of interned url strings, one per distinct URL
#           flags,    bytearray, 1 = is_content(url)
#           counts,   array, how many triples belong to urls[j]
#           installs, array of install ids   \  flattened triples, in
#           platforms,array of platform ids   } urls order, counts[j]
#           hits)     array of hit counts    /  entries per url
_MAX_CACHED_URLAGG_DAYS = 400
_urlagg_cache = collections.OrderedDict()

# id <-> name tables, shared by every cached day. Appended under _lock; readers
# index them without one (lists never shrink). Platform ids also memoise the
# bot's display group, so classification happens once per bot, not per row --
# and by the same category_for the detail page uses, so the two views can never
# classify the same bot differently.
_install_names, _install_ids = [], {}
_platform_names, _platform_ids, _platform_groups = [], {}, []


def _install_id(name):
    got = _install_ids.get(name)
    if got is None:
        with _lock:
            got = _install_ids.get(name)
            if got is None:
                got = len(_install_names)
                _install_names.append(name)
                _install_ids[name] = got
    return got


def _platform_id(name):
    got = _platform_ids.get(name)
    if got is None:
        with _lock:
            got = _platform_ids.get(name)
            if got is None:
                got = len(_platform_names)
                _platform_names.append(name)
                _platform_groups.append(category_for(name))
                _platform_ids[name] = got
    return got


def urlagg(day):
    """One day's URL aggregate in the columnar shape above, cached.
    None if that day has no aggregate file yet."""
    with _lock:
        if day in _urlagg_cache:
            _urlagg_cache.move_to_end(day)
            return _urlagg_cache[day]
    try:
        body = client().get_object(
            Bucket=BUCKET, Key="urlagg/date=%s.json.gz" % day)["Body"].read()
    except ClientError:
        return None
    doc = json.loads(gzip.decompress(body))
    urls, flags = [], bytearray()
    counts = array.array("i")
    installs, platforms, hits = array.array("i"), array.array("i"), array.array("l")
    for url, triples in doc.get("urls", {}).items():
        url = sys.intern(url)
        urls.append(url)
        flags.append(1 if is_content(url) else 0)
        counts.append(len(triples))
        for install, platform, n in triples:
            installs.append(_install_id(str(install)))
            platforms.append(_platform_id(str(platform)))
            hits.append(int(n))
    data = (urls, flags, counts, installs, platforms, hits)
    with _lock:
        _urlagg_cache[day] = data
        _urlagg_cache.move_to_end(day)
        while len(_urlagg_cache) > _MAX_CACHED_URLAGG_DAYS:
            _urlagg_cache.popitem(last=False)
    return data


def trend(days, url_query="", sites=None, platforms=None, groups=None,
          content_only=False, top_urls=25):
    """Daily counts of matching URL hits, straight from the urlagg files.

    Returns (per_day, by_url, by_platform, missing_days):
      per_day    [(day, hits), ...] in the given order, 0 for quiet days
      by_url     the `top_urls` most-hit matching URLs across the range,
                 as [(url, hits), ...]
      by_platform {platform: hits} across the range
      missing_days days that have no aggregate file (parsed before the
                 aggregate existed and not yet rebuilt); counted as 0.

    `url_query` matches case-insensitively anywhere in the URL, exactly like
    the detail page's URL filter. Empty matches everything, which makes the
    default view "all AI traffic over time" for the chosen site/filters.
    """
    needle = (url_query or "").strip().lower()

    # Warm the cache in parallel; serial GETs would make a 90-day range crawl.
    with _lock:
        cold = [d for d in days if d not in _urlagg_cache]
    if cold:
        list(_executor().map(urlagg, cold))

    # Translate the name filters into id predicates once, AFTER the warm --
    # the id tables only know names seen in loaded days. A site or platform
    # the tables have never seen simply yields an empty allow-set, which
    # correctly matches nothing rather than everything.
    want_installs = (None if not sites else
                     {_install_ids[s] for s in sites if s in _install_ids})
    allowed_platforms = None
    if platforms or groups:
        want_p = set(platforms) if platforms else None
        want_g = set(groups) if groups else None
        allowed_platforms = {
            pid for pid, name in enumerate(_platform_names)
            if (want_p is None or name in want_p)
            and (want_g is None or _platform_groups[pid] in want_g)}

    per_day = []
    by_url = collections.Counter()
    by_platform = collections.Counter()   # keyed by platform id
    missing = []
    for day in days:
        data = urlagg(day)
        if data is None:
            missing.append(day)
            per_day.append((day, 0))
            continue
        urls, flags, counts, installs, platforms_a, hits = data
        day_hits = 0
        k = 0
        for j, url in enumerate(urls):
            n = counts[j]
            if (content_only and not flags[j]) or \
               (needle and needle not in url.lower()):
                k += n
                continue
            url_hits = 0
            for i in range(k, k + n):
                if want_installs is not None and installs[i] not in want_installs:
                    continue
                if allowed_platforms is not None and platforms_a[i] not in allowed_platforms:
                    continue
                h = hits[i]
                url_hits += h
                by_platform[platforms_a[i]] += h
            if url_hits:
                day_hits += url_hits
                by_url[url] += url_hits
            k += n
        per_day.append((day, day_hits))

    named_platforms = {_platform_names[pid]: n for pid, n in by_platform.items()}
    return per_day, by_url.most_common(top_urls), named_platforms, missing