"""Read Prod ingress history and D3 diagnostics without changing the host.

Nginx is the external-request counting boundary. Docker output contributes only
search/permission diagnostics; internal HTTP calls are never added to volume.
"""
from __future__ import annotations

import glob
import gzip
import json
import math
import os
import re
import subprocess
from datetime import datetime, timezone
from pathlib import Path
from urllib.parse import urlsplit

from daily_prod_backend_log_report import HttpRequest, strip_ansi

DEFAULT_NGINX_GLOBS = (
    "/var/log/nginx/nexuscrm-prod-d3-*.access.log*",
    "/var/log/nginx/nexuscrm-prod.access.log*",
)
DEFAULT_PROD_ROOT = "/home/ubuntu/nexuscrm-deployment/online/envs/multi_prod"
COMBINED = re.compile(
    r'^\S+ \S+ \S+ \[(?P<time>[^]]+)\] "(?P<request>[^"\n]*)" '
    r'(?P<status>\d{3}) (?:\d+|-) "[^"\n]*" "[^"\n]*"\s*$'
)
ROTATION = re.compile(r"\.access\.log(?:\.\d+)?(?:\.gz)?$")
DIAGNOSTIC_MARKERS = ("DEBUG_SEARCH_SQL", "Permission diagnostics", "Permission read phase")


def parse_iso(value):
    # Python 3.9 accepts microseconds, while Docker timestamps carry nanoseconds.
    value = re.sub(r"(\.\d{6})\d+", r"\1", value)
    return datetime.fromisoformat(value.replace("Z", "+00:00"))


def iso(value):
    return value.isoformat().replace("+00:00", "Z") if value is not None else None


def parse_nginx(line):
    """Return (request, event identity, release identity); never invent latency."""
    if line.lstrip().startswith("{"):
        data = json.loads(line)
        if data.get("schema") != "nexuscrm.prod.http.v1":
            raise ValueError("unknown nginx JSON schema")
        timestamp = datetime.fromtimestamp(float(data["msec"]), timezone.utc)
        seconds = float(data["request_time"])
        if not math.isfinite(seconds) or seconds < 0:
            raise ValueError("invalid request_time")
        status = int(data["status"])
        method, path = data["method"], data["uri"]
        event_id = str(data["nginx_request_id"])
        if not event_id or event_id == "-":
            raise ValueError("missing nginx request identity")
        request_id = str(data.get("upstream_request_id", ""))
        if request_id == "-":
            request_id = ""
        latency = seconds * 1000
        release = str(data["release"])
    else:
        match = COMBINED.fullmatch(line.strip())
        if not match:
            raise ValueError("unrecognized nginx access record")
        timestamp = datetime.strptime(match["time"], "%d/%b/%Y:%H:%M:%S %z")
        request = match["request"].split(" ")
        if len(request) == 3 and request[2].startswith("HTTP/"):
            method, path = request[:2]
        else:
            # Nginx still records malformed client requests (including request '-').
            method, path = "UNKNOWN", "/__invalid_request__"
        status = int(match["status"])
        latency, event_id, request_id, release = None, "", "", ""
    if not 100 <= status <= 599 or not isinstance(path, str) or not isinstance(method, str):
        raise ValueError("invalid status or request target")
    if not re.fullmatch(r"[A-Z]+", method):
        method = "UNKNOWN"
    if path.startswith(("http://", "https://")):
        target = urlsplit(path)
        path = target.path or "/"
        if target.query:
            path += "?" + target.query
    elif not path.startswith("/"):
        path = "/__invalid_request__"

    # Historical combined logs and early JSON fixtures can contain raw query
    # arguments. They must not reach samples, SOQL groups, or report artifacts.
    path = urlsplit(path).path or "/"
    return HttpRequest(timestamp.astimezone(timezone.utc), method, path, status, latency, "", request_id, ""), event_id, release


def discover_files(patterns):
    """Resolve aliases by device/inode. Do not use mtime as event-time authority."""
    files, seen = [], set()
    for pattern in patterns:
        for value in sorted(glob.glob(pattern)):
            path = Path(value).resolve()
            if not path.is_file() or not ROTATION.search(path.name):
                continue
            stat = path.stat()
            identity = (stat.st_dev, stat.st_ino)
            if identity not in seen:
                seen.add(identity)
                files.append(path)
    return files


def nginx_entries(files, window, evidence):
    """Open the rotation set before scanning, preserving renamed open files."""
    from contextlib import ExitStack

    seen_ids = set()
    with ExitStack() as stack:
        opened = []
        for path in files:
            # Permission/rotation errors propagate: a silently skipped source is unsafe.
            raw = stack.enter_context(path.open("rb"))
            stat = os.fstat(raw.fileno())
            source = {"path": str(path), "device": stat.st_dev, "inode": stat.st_ino,
                      "size_at_open": stat.st_size, "mtime": stat.st_mtime,
                      "parsed_lines": 0, "malformed_lines": 0, "in_window": 0,
                      "duplicate_events": 0, "missing_latency": 0, "release_ids": [],
                      "first_event_at": None, "last_event_at": None,
                      "first_in_window_at": None, "last_in_window_at": None}
            evidence["sources"].append(source)
            opened.append((path, raw, source))
        for path, raw, source in opened:
            reader = gzip.GzipFile(fileobj=raw) if path.suffix == ".gz" else raw
            first = last = first_window = last_window = None
            releases = set()
            release_ranges = {}
            # Current files may grow, but the fixed completed window rejects later events.
            for encoded in reader:
                if not encoded.strip():
                    continue
                try:
                    request, event_id, release = parse_nginx(encoded.decode("utf-8"))
                except (ValueError, KeyError, TypeError, OverflowError, UnicodeError):
                    source["malformed_lines"] += 1
                    yield None, True, ""
                    continue
                source["parsed_lines"] += 1
                ts = request.timestamp
                first = min(first, ts) if first else ts
                last = max(last, ts) if last else ts
                if release:
                    bound = release_ranges.setdefault(release, {"first": ts, "last": ts, "in_window": 0})
                    bound["first"] = min(bound["first"], ts)
                    bound["last"] = max(bound["last"], ts)
                if not window.utc_start <= ts < window.utc_end:
                    continue
                first_window = min(first_window, ts) if first_window else ts
                last_window = max(last_window, ts) if last_window else ts
                if event_id:
                    key = (event_id, ts)
                    if key in seen_ids:
                        source["duplicate_events"] += 1
                        continue
                    seen_ids.add(key)
                source["in_window"] += 1
                source["missing_latency"] += request.latency_ms is None
                if release:
                    releases.add(release)
                    release_ranges[release]["in_window"] += 1
                yield request, True, ""
            source.update(first_event_at=iso(first), last_event_at=iso(last),
                          first_in_window_at=iso(first_window), last_in_window_at=iso(last_window),
                          release_ids=sorted(releases),
                          release_ranges=[{"release": release, "first_event_at": iso(bound["first"]),
                                           "last_event_at": iso(bound["last"]), "in_window": bound["in_window"]}
                                          for release, bound in sorted(release_ranges.items())])
            if reader is not raw:
                reader.close()


def docker_inventory():
    ids = subprocess.check_output(
        ["docker", "ps", "-aq", "--filter", "label=io.nexuscrm.serviceification-d3"],
        text=True, timeout=30).split()
    selected = []
    for container_id in ids:
        # Never emit inspect Config.Env (credentials) into report artifacts.
        raw = json.loads(subprocess.check_output(
            ["docker", "inspect", container_id], text=True, timeout=30))[0]
        name = raw["Name"].lstrip("/")
        if not name.startswith("nexuscrm-d3-prod-slot-"):
            continue
        role = next((role for role in ("runtime-api", "sfcrm", "integration")
                     if name.endswith("-" + role)), None)
        if not role:
            continue
        selected.append({"container_id": raw["Id"], "name": name, "role": role,
                         "image_id": raw["Image"], "created_at": raw["Created"],
                         "restart_count": raw["RestartCount"],
                         "state": {key: raw["State"].get(key) for key in ("Status", "StartedAt", "FinishedAt")},
                         "run_id": raw["Config"].get("Labels", {}).get("io.nexuscrm.serviceification-d3.run"),
                         "log_driver": raw["HostConfig"]["LogConfig"]["Type"],
                         "log_path": raw.get("LogPath"),
                         "retention": "unverified; Docker may have rotated or removed older events"})
    return selected


def diagnostic_entries(window, evidence):
    """Include retained stopped instances and restarts; never count their HTTP calls."""
    inventory = docker_inventory()
    evidence["diagnostic_sources"] = inventory
    for source in inventory:
        if parse_iso(source["created_at"]) >= window.utc_end:
            source["selection"] = "created_after_window"
            continue
        source.update(selection="retained_instance", diagnostic_lines=0,
                      first_event_at=None, last_event_at=None)
        command = ["docker", "logs", "--timestamps", "--since", iso(window.utc_start),
                   "--until", iso(window.utc_end), source["container_id"]]
        with subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                              text=True, errors="replace") as process:
            for line in process.stdout:
                docker_time, _, payload = line.partition(" ")
                try:
                    ts = parse_iso(docker_time)
                except ValueError:
                    continue
                if not window.utc_start <= ts < window.utc_end:
                    continue
                stamp = iso(ts)
                source["first_event_at"] = min(source["first_event_at"] or stamp, stamp, key=parse_iso)
                source["last_event_at"] = max(source["last_event_at"] or stamp, stamp, key=parse_iso)
                clean = strip_ansi(payload).strip()
                if any(marker in clean for marker in DIAGNOSTIC_MARKERS):
                    source["diagnostic_lines"] += 1
                    yield None, False, clean
            if process.wait() != 0:
                raise RuntimeError("cannot read diagnostic logs for " + source["container_id"])


def prod_entries(files, window, evidence, collect_diagnostics=True):
    yield from nginx_entries(files, window, evidence)
    if collect_diagnostics:
        yield from diagnostic_entries(window, evidence)


def coverage_evidence(prod_root):
    marker = Path(prod_root) / "d3/current.json"
    authority = json.loads(marker.read_text())
    if authority.get("schema") != "nexuscrm.serviceification-d3-prod-current/v2":
        raise ValueError("unrecognized Prod D3 authority")
    return {"schema": "nexuscrm.prod.http-coverage/v1", "http_source": "nginx_ingress",
            "authority_path": str(marker), "current_authority_snapshot": authority,
            "authority_scope": "collection-time snapshot, not historical event attribution",
            "sources": [], "diagnostic_sources": [],
            "completeness": "unproven: event presence alone does not establish continuous retention",
            "deduplication": "physical inode aliases; JSON nginx request ID + event time across segments",
            "legacy_identity_limit": "combined records have no request ID; overlapping legacy segments are inconclusive"}


def finish_coverage(report, evidence, diagnostics_enabled):
    reasons = []
    summary = report["summary"]
    categories = summary["request_categories"]
    if categories.get("interactive", 0) + categories.get("integration", 0) == 0:
        reasons.append("no interactive or integration requests")
    if not summary["latency_available"]:
        reasons.append("latency unavailable or incomplete; legacy nginx combined logs have no request_time")
    if summary["malformed_http_lines"]:
        reasons.append("unrecognized ingress records; event-time coverage cannot be established")
    if not any(s["in_window"] and "prod-d3-" in s["path"] for s in evidence["sources"]):
        reasons.append("no D3 ingress samples; Classic-only data cannot establish D3 coverage")
    selected_roles = {s["role"] for s in evidence["diagnostic_sources"]
                      if s.get("selection") == "retained_instance"}
    missing_roles = {"runtime-api", "sfcrm", "integration"} - selected_roles
    if not diagnostics_enabled or missing_roles:
        reasons.append("D3 diagnostic sources unavailable for: " + ", ".join(sorted(missing_roles)))
    # Legacy records cannot be deduplicated safely across overlapping rotated copies.
    legacy = [s for s in evidence["sources"] if s["missing_latency"]]
    for index, left in enumerate(legacy):
        for right in legacy[index + 1:]:
            if (max(left["first_in_window_at"], right["first_in_window_at"])
                    <= min(left["last_in_window_at"], right["last_in_window_at"])):
                reasons.append("overlapping legacy segments without event IDs; counts may overlap")
    ranges = [s for s in evidence["sources"] if s["first_event_at"]]
    first = min((s["first_event_at"] for s in ranges), key=parse_iso, default=None)
    last = max((s["last_event_at"] for s in ranges), key=parse_iso, default=None)
    evidence["observed_history_start"] = first
    evidence["observed_history_end"] = last
    if (not first or parse_iso(first) > parse_iso(report["utc_window_start"])
            or not last or parse_iso(last) < parse_iso(report["utc_window_end"])):
        reasons.append("retained ingress history does not bracket the requested window")
    # A Classic stream or a different slot must not establish this slot's
    # coverage. Without a durable cutover timeline, partial slot windows remain
    # inconclusive even if the union happens to bracket the requested window.
    streams = {}
    for source in ranges:
        match = re.search(r"nexuscrm-prod-d3-(slot-\d+)\.access\.log", source["path"])
        if match:
            if source.get("release_ranges"):
                for bound in source["release_ranges"]:
                    streams.setdefault((match[1], bound["release"]), []).append(bound)
            else:
                streams.setdefault((match[1], "legacy-unattributed"), []).append(source)
    evidence["d3_stream_bounds"] = []
    for (slot, release), segments in sorted(streams.items()):
        if not any(source["in_window"] for source in segments):
            continue
        first_slot = min((s["first_event_at"] for s in segments), key=parse_iso)
        last_slot = max((s["last_event_at"] for s in segments), key=parse_iso)
        brackets = (parse_iso(first_slot) <= parse_iso(report["utc_window_start"])
                    and parse_iso(last_slot) >= parse_iso(report["utc_window_end"]))
        evidence["d3_stream_bounds"].append({"slot": slot, "release": release, "first_event_at": first_slot,
                                             "last_event_at": last_slot, "brackets_window": brackets})
        if not brackets:
            reasons.append(f"D3 {slot}/{release} history does not independently bracket the window; cutover coverage is unproven")
    if len(evidence["d3_stream_bounds"]) > 1:
        reasons.append("multiple D3 slot/release streams observed without a durable cutover timeline")
    # Preserve useful term-shape/fieldset aggregates without publishing user input.
    for row in report["search_term_profile"]["by_input_type"]:
        row["samples"] = []
    report["search_term_profile"]["raw_samples_available"] = False
    report["search_term_profile"]["coverage"] = {
        "status": "OBSERVED" if report["search_term_profile"]["total_search_queries"] else "INCONCLUSIVE",
        "note": "Only retained DEBUG_SEARCH_SQL events are measured; zero events do not establish zero searches.",
    }
    evidence["status"] = "INCONCLUSIVE" if reasons else "OBSERVED"
    evidence["reasons"] = sorted(set(reasons))
    report["source_coverage"] = evidence
    summary["window_data_warning"] = bool(reasons) or summary["window_data_warning"]
    report["notes"][0:3] = [
        "External HTTP counts/status/latency come only from Prod nginx access logs, including retained rotations and slots.",
        "JSON v1 request_time is seconds converted to milliseconds. Legacy combined latency is unavailable, never zero.",
        "Docker logs supply search/permission diagnostics only. Internal HTTP requests are not added to ingress counts.",
        "OBSERVED confirms usable samples, not completeness or user health; inspect source ranges and retention limits.",
    ]