"""Tests for events: record_event, record_llm_call, events_for and their storage.

Real FastMCP servers through fastmcp.Client, as in test_instrumentation.py.
"""

import asyncio
import contextlib
import io
import json
import threading
from datetime import UTC, datetime, timedelta

import pytest
from fastmcp import Client, FastMCP
from fastmcp.exceptions import ToolError
from sqlalchemy import MetaData, func, select
from sqlalchemy.exc import DBAPIError

from fastmcp_feedback.instrumentation import (
    REDACTED,
    CallbackSink,
    DatabaseSink,
    EventRecord,
    JsonLinesSink,
    MemorySink,
    ToolCallRecord,
    build_metadata,
    instrument,
)


@pytest.fixture
async def db(db_url):
    """DatabaseSinks whose engines are always disposed, or the process hangs."""
    made = []

    def factory(name="calls", **kwargs):
        kwargs.setdefault("create_tables", True)
        sink = DatabaseSink(db_url(name), **kwargs)
        made.append(sink)
        return sink

    yield factory
    for sink in made:
        await sink.engine.dispose()


def make_app(sinks=None, **kwargs):
    """A server whose tools record events through the middleware returned."""
    sink = MemorySink()
    app = FastMCP("events")
    holder = {}
    spawned = []

    @app.tool
    async def submit(job_id: str) -> dict:
        mw = holder["mw"]
        mw.record_event("job.submitted", key=job_id, attrs={"case": "wing", "nproc": 4})
        return {"job_id": job_id}

    @app.tool
    def submit_sync(job_id: str) -> dict:
        holder["mw"].record_event("job.submitted", key=job_id)
        return {"job_id": job_id}

    @app.tool
    async def spawn(job_id: str) -> dict:
        """Start a watcher task that records after the call has returned."""
        started = asyncio.Event()

        async def watcher():
            started.set()
            await asyncio.sleep(0.05)
            holder["mw"].record_event("job.started", key=job_id)

        spawned.append(asyncio.get_running_loop().create_task(watcher()))
        await started.wait()
        return {"job_id": job_id}

    @app.tool
    async def chat(prompt: str) -> str:
        holder["mw"].record_llm_call(
            model="claude-opus-4",
            provider="anthropic",
            duration_ms=812.5,
            input_tokens=1200,
            output_tokens=90,
            attrs={"cache_read_input_tokens": 1000},
        )
        return "ok"

    @app.tool
    async def chat_fails(prompt: str) -> str:
        err = RuntimeError(
            "upstream 401: api_key=sk-abcdefghijklmnopqrstuv " + "x" * 800
        )
        holder["mw"].record_llm_call(model="gpt-x", provider="openai", error=err)
        raise err

    @app.tool
    def poll() -> dict:
        return {"status": "idle"}

    @app.tool
    def poll_with_event() -> dict:
        holder["mw"].record_event("poll.saw_job", key="j9")
        return {"status": "idle"}

    mw = instrument(app, sinks if sinks is not None else [sink], **kwargs)
    holder["mw"] = mw
    return app, sink, mw, spawned


async def call(app, mw, *calls):
    async with Client(app) as client:
        for tool, args in calls:
            with contextlib.suppress(ToolError):
                await client.call_tool(tool, args)
    await mw.flush()


async def table_rows(sink, table):
    async with sink.engine.connect() as conn:
        return [dict(r) for r in (await conn.execute(select(table))).mappings()]


def event(kind="job.step", key="j1", minutes_ago=0, **kwargs):
    return EventRecord(
        kind=kind,
        occurred_at=datetime.now(UTC) - timedelta(minutes=minutes_ago),
        key=key,
        **kwargs,
    )


class TestLinking:
    async def test_event_in_tool_links_to_stored_call(self, db):
        store = db()
        app, _, mw, _ = make_app([store])
        await call(app, mw, ("submit", {"job_id": "j1"}))
        [call_row] = await table_rows(store, store.table)
        [ev] = await table_rows(store, store.events)
        assert ev["kind"] == "job.submitted"
        assert ev["call_id"] == call_row["id"]
        assert ev["session_id"] == call_row["session_id"]
        assert ev["call_id"] == mw.recent[-1].id

    async def test_sync_tool_links_too(self):
        app, sink, mw, _ = make_app()
        await call(app, mw, ("submit_sync", {"job_id": "j2"}))
        await asyncio.sleep(0.01)
        await mw.flush()
        [ev] = sink.events
        [rec] = sink.calls
        assert ev.call_id == rec.id

    async def test_event_outside_call_has_no_call_id(self):
        app, sink, mw, _ = make_app()
        assert mw.record_event("job.finished", key="j1") is True
        await mw.flush()
        [ev] = sink.events
        assert ev.call_id is None and ev.session_id is None

    async def test_task_outliving_call_is_not_linked(self):
        app, sink, mw, spawned = make_app()
        await call(app, mw, ("spawn", {"job_id": "j3"}))
        await asyncio.gather(*spawned)
        await mw.flush()
        [ev] = sink.events
        assert ev.kind == "job.started" and ev.call_id is None

    async def test_explicit_call_id_wins(self):
        app, sink, mw, _ = make_app()
        mw.record_event("x.y", call_id="given")
        await mw.flush()
        assert sink.events[0].call_id == "given"

    async def test_sampled_out_call_kept_when_it_recorded_event(self):
        app, sink, mw, _ = make_app(
            sample_rates={"poll_with_event": 0.01, "poll": 0.01},
            random=lambda: 0.99,
        )
        await call(app, mw, ("poll", {}), ("poll_with_event", {}))
        [rec] = sink.calls
        [ev] = sink.events
        assert rec.tool == "poll_with_event" and rec.sample_rate is None
        assert ev.call_id == rec.id

    async def test_excluded_tool_events_unlinked(self):
        app, sink, mw, _ = make_app(exclude_tools={"poll_with_event"})
        await call(app, mw, ("poll_with_event", {}))
        assert sink.calls == []
        [ev] = sink.events
        assert ev.call_id is None


class TestRecordEvent:
    async def test_key_attrs_and_server_version_stored(self, db):
        store = db()
        app, _, mw, _ = make_app([store], server_version="1.2.3")
        mw.record_event(
            "step.finished",
            key=42,
            attrs={"step": "snappyHexMesh", "cells": 1200000, "mesh_ok": True},
            user_sub="u1",
        )
        await mw.flush()
        [ev] = await table_rows(store, store.events)
        assert ev["key"] == "42"
        assert ev["attrs"] == {
            "step": "snappyHexMesh",
            "cells": 1200000,
            "mesh_ok": True,
        }
        assert ev["user_sub"] == "u1" and ev["server_version"] == "1.2.3"

    async def test_attrs_redacted(self):
        app, sink, mw, _ = make_app()
        mw.record_event(
            "job.failed",
            attrs={
                "token": "abc123",
                "nested": {"db_password": "hunter2"},
                "error_excerpt": "curl -H 'Authorization: Bearer abcdefghijklmnop'",
            },
        )
        await mw.flush()
        attrs = sink.events[0].attrs
        assert attrs["token"] == REDACTED
        assert attrs["nested"]["db_password"] == REDACTED
        assert "abcdefghijklmnop" not in attrs["error_excerpt"]

    async def test_attrs_made_json_safe(self, db):
        store = db()
        app, _, mw, _ = make_app([store])
        when = datetime(2026, 9, 29, tzinfo=UTC)
        mw.record_event(
            "solver.progress",
            key="j1",
            attrs={"residuals": {"p": float("nan"), "U": 1e-5}, "at": when, 3: (1, 2)},
        )
        await mw.flush()
        [ev] = await table_rows(store, store.events)
        assert ev["attrs"] == {
            "residuals": {"p": "nan", "U": 1e-5},
            "at": when.isoformat(),
            "3": [1, 2],
        }

    @pytest.mark.parametrize(
        "kind",
        ["", "   ", None, 5, "k" * 129],
        ids=["empty", "blank", "none", "int", "long"],
    )
    async def test_invalid_kind_rejected(self, kind):
        app, sink, mw, _ = make_app()
        assert mw.record_event(kind) is False
        await mw.flush()
        assert sink.events == [] and list(mw.recent_events) == []

    async def test_invalid_key_and_attrs_rejected(self):
        app, sink, mw, _ = make_app()
        assert mw.record_event("a.b", key="k" * 256) is False
        assert mw.record_event("a.b", attrs=["not", "a", "dict"]) is False
        await mw.flush()
        assert sink.events == []

    async def test_mode_off_is_noop(self):
        app, sink, mw, _ = make_app(mode="off")
        assert mw.record_event("job.submitted", key="j1") is False
        assert mw.record_llm_call(model="m") is False
        await call(app, mw, ("submit", {"job_id": "j1"}))
        assert sink.records == [] and list(mw.recent_events) == []
        assert mw.dispatcher.dropped == 0

    async def test_queue_full_returns_false_and_counts(self):
        app, sink, mw, _ = make_app(max_queue=1)
        assert mw.record_event("a.one") is True
        assert mw.record_event("a.two") is False
        assert mw.dispatcher.dropped == 1
        await mw.flush()
        assert [e.kind for e in sink.events] == ["a.one"]
        assert [e.kind for e in mw.recent_events] == ["a.one"]

    async def test_broken_sink_never_raises(self):
        def boom(records):
            raise RuntimeError("sink down")

        app, _, mw, _ = make_app([CallbackSink(boom)])
        assert mw.record_event("a.b", key="j") is True
        await mw.flush()
        assert mw.dispatcher.sink_errors == 1

    async def test_broken_redactor_never_raises(self):
        app, sink, mw, _ = make_app()

        class Broken:
            def redact(self, value):
                raise RuntimeError("bad pattern")

        mw.redactor = Broken()
        assert mw.record_event("a.b", attrs={"x": 1}) is False


class TestThreads:
    async def test_record_from_thread_while_server_runs(self, db):
        store = db()
        app, _, mw, _ = make_app([store])
        results = []
        async with Client(app) as client:
            await client.call_tool("submit", {"job_id": "j1"})

            def watcher():
                for kind in ("step.started", "job.finished"):
                    results.append(
                        mw.record_event(kind, key="j1", attrs={"case": "wing"})
                    )

            t = threading.Thread(target=watcher)
            t.start()
            await asyncio.to_thread(t.join)
            await asyncio.sleep(0)
        await mw.flush()
        assert results == [True, True]
        rows = await table_rows(store, store.events)
        assert sorted(r["kind"] for r in rows) == [
            "job.finished",
            "job.submitted",
            "step.started",
        ]
        threaded = [r for r in rows if r["kind"] != "job.submitted"]
        assert all(r["call_id"] is None for r in threaded)
        assert [e.kind for e in mw.recent_events] == [
            "job.submitted",
            "step.started",
            "job.finished",
        ]

    async def test_thread_with_its_own_loop_hands_over(self):
        app, sink, mw, _ = make_app()
        mw.record_event("a.first")  # binds the dispatcher to this loop
        out = []
        t = threading.Thread(
            target=lambda: out.append(asyncio.run(self._record_in_loop(mw)))
        )
        t.start()
        await asyncio.to_thread(t.join)
        await asyncio.sleep(0)
        await mw.flush()
        assert out == [True]
        assert [e.kind for e in sink.events] == ["a.first", "a.second"]
        assert mw.dispatcher._loop is asyncio.get_running_loop()

    @staticmethod
    async def _record_in_loop(mw):
        return mw.record_event("a.second")

    def test_no_loop_and_nothing_recorded_drops(self):
        app, sink, mw, _ = make_app()
        assert mw.record_event("job.started", key="j1") is False
        assert mw.dispatcher.dropped == 1
        assert list(mw.recent_events) == []

    async def test_thread_before_anything_recorded_drops(self):
        app, sink, mw, _ = make_app()
        out = []
        t = threading.Thread(target=lambda: out.append(mw.record_event("a.b")))
        t.start()
        await asyncio.to_thread(t.join)
        assert out == [False] and mw.dispatcher.dropped == 1


class TestLlmCall:
    async def test_attrs_and_linking(self, db):
        store = db()
        app, _, mw, _ = make_app([store])
        await call(app, mw, ("chat", {"prompt": "hi"}))
        [call_row] = await table_rows(store, store.table)
        [ev] = await table_rows(store, store.events)
        assert ev["kind"] == "llm.call" and ev["call_id"] == call_row["id"]
        assert ev["attrs"] == {
            "cache_read_input_tokens": 1000,
            "model": "claude-opus-4",
            "provider": "anthropic",
            "duration_ms": 812.5,
            "input_tokens": 1200,
            "output_tokens": 90,
            "ok": True,
        }

    async def test_error_redacted_truncated_and_not_ok(self):
        app, sink, mw, _ = make_app(max_error_chars=100)
        await call(app, mw, ("chat_fails", {"prompt": "hi"}))
        [ev] = sink.events
        attrs = ev.attrs
        assert attrs["ok"] is False and attrs["error_type"] == "RuntimeError"
        assert "sk-abcdefghijklmnopqrstuv" not in attrs["error"]
        assert len(attrs["error"]) == 100
        assert ev.call_id == sink.calls[0].id and sink.calls[0].outcome == "error"

    async def test_named_fields_win_and_ok_explicit(self):
        app, sink, mw, _ = make_app()
        mw.record_llm_call(
            model="m", ok=False, key="turn-1", attrs={"model": "spoofed", "round": 2}
        )
        await mw.flush()
        [ev] = sink.events
        assert ev.key == "turn-1"
        assert ev.attrs == {"model": "m", "round": 2, "ok": False}


class TestSinks:
    async def test_database_sink_mixed_batch(self, db):
        store = db()
        rec = ToolCallRecord(
            tool="t",
            started_at=datetime.now(UTC),
            duration_ms=1.0,
            ok=True,
            mode="meta",
        )
        ev = event(call_id=rec.id, attrs={"a": 1})
        await store.write([rec, ev])
        assert [r["id"] for r in await table_rows(store, store.table)] == [rec.id]
        [row] = await table_rows(store, store.events)
        assert row["id"] == ev.id and row["call_id"] == rec.id

    async def test_hand_built_event_attrs_made_json_safe(self, db):
        store = db()
        await store.write([event(attrs={"p": float("nan"), "q": float("-inf")})])
        [row] = await table_rows(store, store.events)
        assert row["attrs"] == {"p": "nan", "q": "-inf"}

    async def test_database_sink_events_only_and_calls_only(self, db):
        store = db()
        await store.write([event()])
        assert len(await table_rows(store, store.events)) == 1
        assert await table_rows(store, store.table) == []

    async def test_calls_stored_when_events_table_missing(self, db):
        store = db("old", create_tables=False)
        old = MetaData()
        build_metadata("ffb_", old)
        old.remove(old.tables["ffb_events"])
        async with store.engine.begin() as conn:
            await conn.run_sync(old.create_all)
        rec = ToolCallRecord(
            tool="t",
            started_at=datetime.now(UTC),
            duration_ms=1.0,
            ok=True,
            mode="meta",
        )
        # SQLite raises OperationalError, Postgres ProgrammingError.
        with pytest.raises(DBAPIError, match="no such table|does not exist"):
            await store.write([rec, event()])
        assert len(await table_rows(store, store.table)) == 1

    async def test_json_lines_record_field(self):
        stream = io.StringIO()
        app, _, mw, _ = make_app([JsonLinesSink(stream)])
        await call(app, mw, ("submit", {"job_id": "j1"}))
        lines = [json.loads(line) for line in stream.getvalue().splitlines()]
        by_type = {line["record"]: line for line in lines}
        assert set(by_type) == {"tool_call", "event"}
        assert by_type["event"]["kind"] == "job.submitted"
        assert by_type["event"]["call_id"] == by_type["tool_call"]["id"]
        assert by_type["tool_call"]["tool"] == "submit"

    async def test_memory_sink_calls_and_events(self):
        app, sink, mw, _ = make_app()
        await call(app, mw, ("submit", {"job_id": "j1"}), ("poll", {}))
        assert [r.tool for r in sink.calls] == ["submit", "poll"]
        assert [e.kind for e in sink.events] == ["job.submitted"]
        assert len(sink.records) == 3


class TestEventsFor:
    async def test_merges_database_and_memory_in_time_order(self, db):
        store = db()
        app, _, mw, _ = make_app([store])
        # Written by another worker: only in the database.
        await store.write([event("job.submitted", minutes_ago=30)])
        # Recorded here and flushed: in both, returned once.
        mw.record_event("step.started", key="j1")
        await mw.flush()
        # Recorded here, not yet written: only in memory.
        pending = event("step.finished", minutes_ago=-1)
        mw.recent_events.append(pending)
        # Another job, and a kind filtered out below.
        mw.record_event("job.submitted", key="j2")
        await mw.flush()

        rows = await mw.events_for("j1")
        assert [r["kind"] for r in rows] == [
            "job.submitted",
            "step.started",
            "step.finished",
        ]
        assert all(r["occurred_at"].tzinfo is not None for r in rows)
        assert rows[-1]["id"] == pending.id
        assert len({r["id"] for r in rows}) == 3

        steps = await mw.events_for("j1", kinds=["step.started", "step.finished"])
        assert [r["kind"] for r in steps] == ["step.started", "step.finished"]
        last = await mw.events_for("j1", limit=2)
        assert [r["kind"] for r in last] == ["step.started", "step.finished"]

    async def test_memory_only_without_database(self):
        app, sink, mw, _ = make_app()
        await call(app, mw, ("submit", {"job_id": "j1"}))
        mw.record_event("job.finished", key="j1")
        rows = await mw.events_for("j1")
        assert [r["kind"] for r in rows] == ["job.submitted", "job.finished"]
        assert rows[0]["call_id"] == sink.calls[0].id

    async def test_database_failure_falls_back_to_memory(self, db):
        store = db("nodb", create_tables=False)
        app, _, mw, _ = make_app([store])
        mw.recent_events.append(event("job.submitted"))
        rows = await mw.events_for("j1")
        assert [r["kind"] for r in rows] == ["job.submitted"]


class TestPrune:
    async def test_prune_deletes_old_events_in_batches(self, db):
        store = db(retention=timedelta(days=30), prune_batch=2)
        old_call = ToolCallRecord(
            tool="old",
            started_at=datetime.now(UTC) - timedelta(days=40),
            duration_ms=1.0,
            ok=True,
            mode="meta",
        )
        store._last_prune = float("inf")  # keep write() from pruning first
        await store.write(
            [old_call]
            + [event(f"old.{i}", minutes_ago=60 * 24 * 40) for i in range(5)]
            + [event("new.one"), event("new.two", minutes_ago=60 * 24 * 29)]
        )
        statements = []
        from sqlalchemy import event as sa_event

        def count_deletes(conn, cursor, statement, *args):
            if statement.lstrip().upper().startswith("DELETE FROM FFB_EVENTS"):
                statements.append(statement)

        sa_event.listen(
            store.engine.sync_engine, "before_cursor_execute", count_deletes
        )
        deleted = await store.prune()
        assert deleted == 6  # 1 call + 5 events
        assert len(statements) == 3  # 2 + 2 + 1
        kinds = sorted(r["kind"] for r in await table_rows(store, store.events))
        assert kinds == ["new.one", "new.two"]

    async def test_auto_prune_covers_events(self, db):
        store = db(retention=timedelta(days=30))
        await store.write([event("old", minutes_ago=60 * 24 * 31), event("new")])
        assert [r["kind"] for r in await table_rows(store, store.events)] == ["new"]

    async def test_prune_without_events_table(self, db):
        store = db("legacy", create_tables=False)
        old = MetaData()
        build_metadata("ffb_", old)
        old.remove(old.tables["ffb_events"])
        async with store.engine.begin() as conn:
            await conn.run_sync(old.create_all)
        rec = ToolCallRecord(
            tool="old",
            started_at=datetime.now(UTC) - timedelta(days=40),
            duration_ms=1.0,
            ok=True,
            mode="meta",
        )
        await store.write([rec])
        assert await store.prune(datetime.now(UTC) - timedelta(days=30)) == 1
        async with store.engine.connect() as conn:
            n = (
                await conn.execute(select(func.count()).select_from(store.table))
            ).scalar()
        assert n == 0