from datetime import UTC, datetime, timedelta

from fastapi.testclient import TestClient
from sqlmodel import Session

from fluksio.core.config import settings
from fluksio.models import EngineEvent, FlowRun, MetricBucket

PREFIX = f"{settings.API_V1_STR}/observability"
FLOW = "observability-test"


def _seed(db: Session) -> None:
    now = datetime.now(UTC).replace(second=0, microsecond=0)
    db.add(
        MetricBucket(
            flow=FLOW,
            node=f"{FLOW}.calc",
            bucket=now - timedelta(minutes=1),
            executions=4,
            errors=1,
            messages=6,
            duration_sum_ms=40.0,
            duration_max_ms=25.0,
            lag_sum_ms=100.0,
            lag_max_ms=60.0,
            items=4,
        )
    )
    db.add(
        EngineEvent(
            ts=now,
            type="node_error",
            flow=FLOW,
            node=f"{FLOW}.calc",
            detail="ValueError: bad input\nTraceback",
        )
    )
    db.add(
        EngineEvent(
            ts=now, type="audit", flow=FLOW, detail="published", actor="a@example.com"
        )
    )
    db.add(
        FlowRun(
            id="9-0",
            flow=FLOW,
            source="external",
            started_at=now,
            finished_at=now,
            status="ok",
            nodes=3,
            duration_ms=12.5,
        )
    )
    db.commit()


def test_observability_requires_authentication(client: TestClient) -> None:
    assert client.get(f"{PREFIX}/summary").status_code == 401


def test_the_summary_answers_even_when_degraded(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    response = client.get(f"{PREFIX}/summary", headers=superuser_token_headers)

    assert response.status_code == 200
    body = response.json()
    assert body["status"] in {"ok", "degraded"}
    assert set(body["flows"]) == {
        "total",
        "running",
        "paused",
        "quarantined",
        "invalid",
    }
    assert "error" in body["nodes"]
    # A windowed count comes from /flows?hours= now, which is what the tile
    # that used to read this actually sums.
    assert "failures_24h" not in body


def test_a_flow_that_cannot_run_makes_the_summary_degraded(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """A loop on the canvas has to reach the health screen.

    Validation runs on a build, not on this request, so what the engine already
    knows is what this reports — and reporting it is the whole point: a flow
    with a dependency loop cannot run, and the screen used to say "ok".
    """
    from fluksio.flow.pipeline import ValidationIssue

    controller = client.app.state.flow_controller
    before = controller.issues
    before_flows = controller.store.list_flows
    # The looping flow has to be one the store lists, otherwise "running"
    # counts nothing either way and the assertion below proves nothing.
    controller.store.list_flows = lambda: ["looping"]
    controller.issues = [
        ValidationIssue(
            code="cycle",
            message="These nodes depend on each other in a loop",
            flow="looping",
            nodes=["looping.a", "looping.b"],
        ),
        # Already counted as a node that failed to load, so not again here.
        ValidationIssue(
            code="node_error", message="boom", flow="broken", node="broken.x"
        ),
        ValidationIssue(
            code="unauthenticated_hook", message="open", flow="hooky", node="hooky.h"
        ),
    ]
    try:
        body = client.get(f"{PREFIX}/summary", headers=superuser_token_headers).json()
    finally:
        controller.issues = before
        controller.store.list_flows = before_flows

    assert body["status"] == "degraded"
    assert body["flows"]["invalid"] == 1
    assert body["flows"]["running"] == 0
    assert any("looping" in problem for problem in body["problems"])
    assert not any("broken" in problem for problem in body["problems"])
    assert not any("hooky" in problem for problem in body["problems"])


def test_the_history_reads_back(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    _seed(db)

    points = client.get(f"{PREFIX}/timeseries", headers=superuser_token_headers).json()
    mine = [point for point in points if point["executions"]]
    assert mine and mine[-1]["avg_ms"] > 0

    flows = client.get(f"{PREFIX}/flows", headers=superuser_token_headers).json()
    row = next(entry for entry in flows if entry["flow"] == FLOW)
    assert (row["executions"], row["errors"], row["messages"]) == (4, 1, 6)
    assert len(row["spark"]) == 60
    assert row["last_error_ts"] is not None

    runs = client.get(f"{PREFIX}/runs", headers=superuser_token_headers).json()
    assert any(run["id"] == "9-0" and run["status"] == "ok" for run in runs["data"])

    failures = client.get(f"{PREFIX}/events", headers=superuser_token_headers).json()
    assert any("Traceback" in event["detail"] for event in failures)
    assert all(event["type"] != "audit" for event in failures)

    audit = client.get(
        f"{PREFIX}/events", headers=superuser_token_headers, params={"kind": "audit"}
    ).json()
    assert any(event["actor"] == "a@example.com" for event in audit)

    dead = client.get(f"{PREFIX}/dead-letter", headers=superuser_token_headers)
    assert dead.status_code == 200
    assert isinstance(dead.json(), list)


def test_the_timeseries_folds_into_the_requested_bucket(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    """The fold moved into SQL, so the slices still have to line up as before.

    Slots are aligned to the epoch, which is what the old modulo did, and a
    slice only exists where a row does.
    """
    flow = "bucket-fold-test"
    # The start of the current quarter hour, half an hour back so both slices
    # sit inside a one hour window.
    now = datetime.now(UTC).timestamp()
    first = datetime.fromtimestamp(now // 900 * 900 - 1800, UTC)
    for offset, executions in ((0, 1), (2, 2), (5, 4), (15, 8)):
        db.add(
            MetricBucket(
                flow=flow,
                node=f"{flow}.calc",
                bucket=first + timedelta(minutes=offset),
                executions=executions,
            )
        )
    db.commit()

    points = client.get(
        f"{PREFIX}/timeseries",
        headers=superuser_token_headers,
        params={"flow": flow, "hours": 1, "bucket_s": 900},
    ).json()

    assert len(points) == 2
    assert points[0]["executions"] == 1 + 2 + 4
    assert points[1]["executions"] == 8
    assert points[0]["ts"] % 900 == 0


def test_the_bucket_still_filling_is_held_back(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    """The minute in progress is not drawn.

    A bucket is upserted every flush while its minute runs, so the newest one
    only ever holds part of a minute. Charted, it reads as a fall that never
    happened.
    """
    flow = "open-bucket-test"
    minute = int(datetime.now(UTC).timestamp()) // 60 * 60
    for offset, executions in ((-60, 3), (0, 1)):
        db.add(
            MetricBucket(
                flow=flow,
                node=f"{flow}.calc",
                bucket=datetime.fromtimestamp(minute + offset, UTC),
                executions=executions,
            )
        )
    db.commit()

    points = client.get(
        f"{PREFIX}/timeseries",
        headers=superuser_token_headers,
        params={"flow": flow, "hours": 1},
    ).json()
    flows = client.get(
        f"{PREFIX}/flows", headers=superuser_token_headers, params={"hours": 1}
    ).json()
    row = next(entry for entry in flows if entry["flow"] == flow)

    # Unless that minute closed while the requests were in flight — then both
    # buckets are complete and counting both is right.
    if int(datetime.now(UTC).timestamp()) // 60 * 60 == minute:
        assert [int(point["ts"]) for point in points] == [minute - 60]
        # The window ends on the last closed minute, so the final slice is a
        # whole one rather than however much of this minute has arrived.
        assert (row["executions"], row["spark"][-1]) == (3, 3)


def test_a_zero_hour_window_is_still_an_hour(
    client: TestClient, superuser_token_headers: dict[str, str]
) -> None:
    """`hours=0` divided by nothing and answered 500."""
    for path in ("timeseries", "flows"):
        response = client.get(
            f"{PREFIX}/{path}", headers=superuser_token_headers, params={"hours": 0}
        )
        assert response.status_code == 200


def test_the_runs_page_carries_its_total(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    """A full page says how much it left behind."""
    minute = datetime.now(UTC).replace(second=0, microsecond=0) - timedelta(hours=5)
    for index in range(3):
        db.add(
            FlowRun(
                id=f"paged-{index}",
                flow=FLOW,
                started_at=minute + timedelta(seconds=index),
                status="ok",
            )
        )
    db.commit()

    body = client.get(
        f"{PREFIX}/runs",
        headers=superuser_token_headers,
        params={
            "since": minute.isoformat(),
            "until": (minute + timedelta(minutes=1)).isoformat(),
            "limit": 2,
        },
    ).json()

    assert len(body["data"]) == 2
    assert body["count"] == 3


def test_runs_narrow_to_one_minute(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    """A minute picked off a chart reaches past what the recent list holds."""
    minute = datetime.now(UTC).replace(second=0, microsecond=0) - timedelta(hours=3)
    db.add(FlowRun(id="minute-in", flow=FLOW, started_at=minute, status="ok"))
    db.add(
        FlowRun(
            id="minute-after",
            flow=FLOW,
            started_at=minute + timedelta(minutes=1),
            status="ok",
        )
    )
    db.commit()

    runs = client.get(
        f"{PREFIX}/runs",
        headers=superuser_token_headers,
        params={
            "since": minute.isoformat(),
            "until": (minute + timedelta(minutes=1)).isoformat(),
        },
    ).json()

    # The upper bound is exclusive, so the run starting the next minute is not
    # in this one.
    assert [run["id"] for run in runs["data"]] == ["minute-in"]


def test_a_running_cascade_reports_how_long_it_has_been_going(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    """The collector writes a duration at the end; until then, time since."""
    started = datetime.now(UTC) - timedelta(seconds=30)
    db.add(FlowRun(id="in-flight", flow=FLOW, started_at=started, status="running"))
    db.commit()

    runs = client.get(
        f"{PREFIX}/runs", headers=superuser_token_headers, params={"status": "running"}
    ).json()

    # By id rather than by position: another module can leave a run of its own
    # still marked running, and the list is not this test's alone.
    in_flight = next(r for r in runs["data"] if r["id"] == "in-flight")
    assert in_flight["duration_ms"] >= 30_000


def test_events_narrow_to_one_minute(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    """The failures list reaches as far back as the charts beside it."""
    minute = datetime.now(UTC).replace(second=0, microsecond=0) - timedelta(hours=4)
    db.add(EngineEvent(ts=minute, type="node_error", flow=FLOW, detail="minute-in"))
    db.add(
        EngineEvent(
            ts=minute + timedelta(minutes=1),
            type="node_error",
            flow=FLOW,
            detail="minute-after",
        )
    )
    db.commit()

    events = client.get(
        f"{PREFIX}/events",
        headers=superuser_token_headers,
        params={
            "flow": FLOW,
            "since": minute.isoformat(),
            "until": (minute + timedelta(minutes=1)).isoformat(),
        },
    ).json()

    # The upper bound is exclusive, so the failure a minute later is not in it.
    assert [event["detail"] for event in events] == ["minute-in"]


def test_events_narrow_to_one_run(
    client: TestClient, superuser_token_headers: dict[str, str], db: Session
) -> None:
    """Reading one run's failures without filtering the engine-wide list."""
    ts = datetime.now(UTC) - timedelta(hours=6)
    db.add(
        EngineEvent(ts=ts, type="node_error", flow=FLOW, detail="mine", run="run-mine")
    )
    db.add(
        EngineEvent(
            ts=ts, type="node_error", flow=FLOW, detail="theirs", run="run-theirs"
        )
    )
    # What a live cascade unrelated to any run leaves, and what every row
    # written before the column existed looks like.
    db.add(EngineEvent(ts=ts, type="node_error", flow=FLOW, detail="neither"))
    db.commit()

    events = client.get(
        f"{PREFIX}/events",
        headers=superuser_token_headers,
        params={"flow": FLOW, "run": "run-mine"},
    ).json()

    assert [event["detail"] for event in events] == ["mine"]