#!/usr/bin/env python3
"""Build a bounded, request-correlated Runtime phase and SQL ledger.

The summary intentionally excludes SQL text, parameters, record IDs and user
data. Elapsed values from nested events are reported as cumulative observations,
not added together as request wall time.
"""

from __future__ import annotations

import argparse
import json
import re
import shlex
from collections import Counter, defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any, Iterable


ANSI_ESCAPE = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]")
REQUEST_ID = re.compile(r"^[A-Za-z0-9._:-]{1,128}$")
TIMESTAMP = re.compile(r"(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?Z)")
SAFE_DIMENSIONS = (
    "stage",
    "operation",
    "object",
    "read_source",
    "table",
    "status",
    "phase",
    "field_kind",
)
FAILURE_STATUSES = {
    "canceled",
    "cancelled",
    "deadline_exceeded",
    "denied",
    "error",
    "failed",
    "failure",
    "incompatible",
    "rejected",
    "stale",
    "timeout",
    "unavailable",
}
MUTATION_PHASES = (
    "accept",
    "prepare",
    "operation_guard",
    "business_effect",
    "operation_complete",
    "finalize",
)


def _value(raw: str) -> Any:
    lowered = raw.lower()
    if lowered == "true":
        return True
    if lowered == "false":
        return False
    try:
        return int(raw)
    except ValueError:
        try:
            return float(raw)
        except ValueError:
            return raw


def _timestamp(raw: str) -> float | None:
    match = TIMESTAMP.search(raw)
    if not match:
        return None
    try:
        return datetime.fromisoformat(match.group(1).replace("Z", "+00:00")).timestamp()
    except ValueError:
        return None


def _event(line: str) -> dict[str, Any] | None:
    clean = ANSI_ESCAPE.sub("", line)
    start = clean.find("{")
    if start >= 0:
        try:
            decoded, _ = json.JSONDecoder().raw_decode(clean[start:])
        except json.JSONDecodeError:
            decoded = None
        if isinstance(decoded, dict) and decoded.get("event"):
            decoded = dict(decoded)
            decoded["_timestamp"] = _timestamp(clean)
            return decoded
    try:
        tokens = shlex.split(clean)
    except ValueError:
        return None
    fields: dict[str, Any] = {"_timestamp": _timestamp(clean)}
    for token in tokens:
        if "=" not in token:
            continue
        key, raw = token.split("=", 1)
        if key:
            fields[key] = _value(raw)
    return fields if fields.get("event") else None


def events(lines: Iterable[str]) -> Iterable[dict[str, Any]]:
    for line in lines:
        parsed = _event(line)
        if parsed is not None:
            yield parsed


def _top(counter: Counter[str], limit: int = 8) -> list[dict[str, Any]]:
    return [{"name": name, "count": count} for name, count in counter.most_common(limit)]


def _top_elapsed(
    rows: Iterable[dict[str, Any]],
    dimension: str,
    *,
    fallback: str | None = None,
    limit: int = 8,
) -> list[dict[str, Any]]:
    grouped: dict[str, list[float]] = defaultdict(list)
    for row in rows:
        value = row.get(dimension)
        if value in (None, "") and fallback is not None:
            value = row.get(fallback)
        grouped[str(value if value not in (None, "") else "unknown")].append(_elapsed(row))
    ranked = sorted(
        grouped.items(),
        key=lambda item: (-sum(item[1]), item[0]),
    )[:limit]
    return [
        {
            "name": name,
            "count": len(values),
            "elapsed_sum_ms": round(sum(values), 3),
            "elapsed_max_ms": round(max(values, default=0), 3),
        }
        for name, values in ranked
    ]


def _elapsed(row: dict[str, Any]) -> float:
    value = row.get("elapsed_ms", row.get("total_elapsed_ms", 0))
    return float(value) if isinstance(value, (int, float)) else 0.0


def _mutation(rows: list[dict[str, Any]]) -> dict[str, Any] | None:
    profiles = [
        row for row in rows
        if row.get("event") == "runtime_record_mutation_phase_profile"
    ]
    if not profiles:
        return None
    phases: list[dict[str, Any]] = []
    for phase in MUTATION_PHASES:
        phases.append({
            "phase": phase,
            "calls": sum(int(row.get(f"{phase}_calls", 0)) for row in profiles),
            "sql": sum(int(row.get(f"{phase}_sql", 0)) for row in profiles),
            "sql_elapsed_ms": round(sum(float(row.get(f"{phase}_sql_elapsed_ms", 0)) for row in profiles), 3),
            "elapsed_ms": round(sum(float(row.get(f"{phase}_elapsed_ms", 0)) for row in profiles), 3),
        })
    return {
        "samples": len(profiles),
        "total_elapsed_ms": round(sum(_elapsed(row) for row in profiles), 3),
        "total_sql": sum(int(row.get("total_sql", 0)) for row in profiles),
        "total_sql_elapsed_ms": round(sum(float(row.get("total_sql_elapsed_ms", 0)) for row in profiles), 3),
        "phases": phases,
    }


def _family(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
    grouped: dict[str, list[dict[str, Any]]] = defaultdict(list)
    for row in rows:
        grouped[str(row.get("event", "unknown"))].append(row)
    result: list[dict[str, Any]] = []
    for name, family_rows in grouped.items():
        elapsed = [_elapsed(row) for row in family_rows]
        dimensions: Counter[str] = Counter()
        dimension_rows: list[dict[str, Any]] = []
        for row in family_rows:
            parts = [f"{key}={row[key]}" for key in SAFE_DIMENSIONS if row.get(key) not in (None, "")]
            dimension = ", ".join(parts) or "unclassified"
            dimensions[dimension] += 1
            dimension_rows.append({"dimension": dimension, "elapsed_ms": _elapsed(row)})
        result.append(
            {
                "event": name,
                "samples": len(family_rows),
                "elapsed_sum_ms": round(sum(elapsed), 3),
                "elapsed_max_ms": round(max(elapsed, default=0), 3),
                "top_dimensions": _top(dimensions),
                "top_dimensions_by_elapsed": _top_elapsed(
                    dimension_rows, "dimension",
                ),
            }
        )
    return sorted(result, key=lambda row: (-row["elapsed_sum_ms"], row["event"]))


def _sql(rows: list[dict[str, Any]]) -> dict[str, Any]:
    sql_rows = [row for row in rows if row.get("event") == "repo_sql_timing"]
    elapsed = [_elapsed(row) for row in sql_rows]
    fields = [int(row.get("fields", 0)) for row in sql_rows if isinstance(row.get("fields", 0), int)]
    result_rows = [int(row.get("rows", 0)) for row in sql_rows if isinstance(row.get("rows", 0), int)]
    operations = Counter(str(row.get("operation", "unknown")) for row in sql_rows)
    sources = Counter(str(row.get("read_source", row.get("table", "unknown"))) for row in sql_rows)
    return {
        "count": len(sql_rows),
        "elapsed_sum_ms": round(sum(elapsed), 3),
        "elapsed_max_ms": round(max(elapsed, default=0), 3),
        "fields_sum": sum(fields),
        "fields_max": max(fields, default=0),
        "rows_sum": sum(result_rows),
        "top_operations": _top(operations),
        "top_sources": _top(sources),
        "top_operations_by_elapsed": _top_elapsed(sql_rows, "operation"),
        "top_sources_by_elapsed": _top_elapsed(
            sql_rows, "read_source", fallback="table",
        ),
    }


def summarize(raw_events: Iterable[dict[str, Any]]) -> dict[str, Any]:
    by_request: dict[str, list[dict[str, Any]]] = defaultdict(list)
    event_count = 0
    uncorrelated = 0
    for row in raw_events:
        event_count += 1
        request_id = str(row.get("request_id", ""))
        if not REQUEST_ID.fullmatch(request_id):
            uncorrelated += 1
            continue
        by_request[request_id].append(row)

    requests: list[dict[str, Any]] = []
    for request_id, rows in by_request.items():
        timestamps = [row["_timestamp"] for row in rows if isinstance(row.get("_timestamp"), float)]
        failures = Counter(
            f"{row.get('event')}:{row.get('stage', row.get('operation', '?'))}:{row.get('status')}"
            for row in rows
            if str(row.get("status", "")).lower() in FAILURE_STATUSES or row.get("has_error") is True
        )
        requests.append(
            {
                "request_id": request_id,
                "event_count": len(rows),
                # This is only the span between first and last observed Runtime
                # events. It is a lower bound, not the HTTP request wall time.
                "observed_event_span_ms": round((max(timestamps) - min(timestamps)) * 1000, 3)
                if len(timestamps) >= 2
                else 0.0,
                "sql": _sql(rows),
                "mutation": _mutation(rows),
                "families": _family(rows),
                "failures": _top(failures),
            }
        )
    requests.sort(
        key=lambda row: (
            -max(row["observed_event_span_ms"], row["sql"]["elapsed_sum_ms"]),
            row["request_id"],
        )
    )
    correlated = event_count - uncorrelated
    return {
        "schema": "nexuscrm.runtime-request-profile-summary/v1",
        "interpretation": {
            "observed_event_span_ms": "lower bound between first and last Runtime event, not HTTP wall time",
            "elapsed_sum_ms": "cumulative observations that may be nested or overlapping; do not add to wall time",
            "sql": "counts and timings only; SQL text, parameters, record IDs, and user data are excluded",
        },
        "event_count": event_count,
        "correlated_event_count": correlated,
        "uncorrelated_event_count": uncorrelated,
        "correlation_rate_pct": round((correlated / event_count) * 100, 2) if event_count else 0.0,
        "request_count": len(requests),
        "requests": requests,
    }


def markdown(summary: dict[str, Any]) -> str:
    lines = [
        "# Runtime request profile",
        "",
        f"Correlation: {summary['correlated_event_count']}/{summary['event_count']} events "
        f"({summary['correlation_rate_pct']}%), {summary['request_count']} requests.",
        "",
        "> Event span is a lower bound. Elapsed sums can contain nested/overlapping work and are not request wall time.",
        "",
        "| request | event span ms | events | SQL count | SQL cumulative ms | max fields | top SQL owner | top family |",
        "|---|---:|---:|---:|---:|---:|---|---|",
    ]
    for request in summary["requests"]:
        top_family = request["families"][0]["event"] if request["families"] else ""
        top_sql = request["sql"]["top_operations_by_elapsed"]
        top_sql_owner = (
            f"{top_sql[0]['name']} ({top_sql[0]['elapsed_sum_ms']}ms)"
            if top_sql else ""
        )
        lines.append(
            f"| {request['request_id']} | {request['observed_event_span_ms']} | {request['event_count']} | "
            f"{request['sql']['count']} | {request['sql']['elapsed_sum_ms']} | "
            f"{request['sql']['fields_max']} | {top_sql_owner} | {top_family} |"
        )
    return "\n".join(lines) + "\n"


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("log", type=Path)
    parser.add_argument("--json-output", type=Path)
    parser.add_argument("--markdown-output", type=Path)
    args = parser.parse_args()
    with args.log.open(encoding="utf-8", errors="replace") as handle:
        summary = summarize(events(handle))
    encoded = json.dumps(summary, indent=2, sort_keys=True) + "\n"
    if args.json_output:
        args.json_output.write_text(encoded, encoding="utf-8")
    else:
        print(encoded, end="")
    if args.markdown_output:
        args.markdown_output.write_text(markdown(summary), encoding="utf-8")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())