"""Statistics labels, event identity and logging, as fixed in 2026.10.01.4.

Statistics run on SQLite and, with FFB_TEST_DATABASE_URL set, on Postgres
through the feedback tools' own synchronous engine (psycopg2).
"""

import asyncio
import contextlib
import json
import os
import subprocess
import sys
import tomllib
from pathlib import Path

import pytest
from fastmcp import Client, FastMCP
from fastmcp.exceptions import ToolError
from packaging.requirements import Requirement
from sqlalchemy import String, select, text, type_coerce
from sqlalchemy.engine import make_url

import fastmcp_feedback
from fastmcp_feedback import (
    FeedbackDatabase,
    ManagementMixin,
    RetrievalMixin,
    SubmissionMixin,
    SubmitFeedbackRequest,
    get_database_stats,
)
from fastmcp_feedback.database import enum_label
from fastmcp_feedback.instrumentation import REDACTED, MemorySink, instrument
from fastmcp_feedback.models import Feedback, FeedbackStatus, FeedbackType

ROOT = Path(__file__).resolve().parents[2]


# Statistics


@pytest.fixture
async def feedback_db(db_backend, db_url):
    """A FeedbackDatabase on an empty database of the current backend."""
    url = make_url(db_url("feedback"))
    if db_backend == "postgres":
        pytest.importorskip("psycopg2")
        url = url.set(drivername="postgresql+psycopg2")
    else:
        url = url.set(drivername="sqlite")
    db = FeedbackDatabase(url.render_as_string(hide_password=False))
    await db.initialize()
    yield db
    await db.close()


async def seed(db):
    """Two bugs (one in progress) and a feature, through the tools."""
    submit = SubmissionMixin(db)
    ids = []
    for kind in ("bug", "bug", "feature"):
        result = await submit.submit_feedback(
            SubmitFeedbackRequest(
                type=kind, title=f"a {kind}", description="d", submitter="t"
            )
        )
        assert result["success"], result
        ids.append(result["feedback_id"])
    updated = await ManagementMixin(db).update_feedback_status(
        ids[0], FeedbackStatus.IN_PROGRESS
    )
    assert updated["success"], updated
    return ids


class TestStatisticsLabels:
    async def test_database_stores_enum_names(self, feedback_db):
        # Why the mapping exists: the columns hold names on both backends.
        await seed(feedback_db)
        async with feedback_db.session() as session:
            stored = session.execute(
                select(type_coerce(Feedback.type, String)).distinct()
            ).scalars()
            assert set(stored) == {"BUG", "FEATURE"}

    async def test_statistics_keyed_by_value(self, feedback_db):
        await seed(feedback_db)
        stats = await RetrievalMixin(feedback_db).get_feedback_statistics()
        assert "error" not in stats, stats
        assert stats["total_count"] == 3
        assert stats["by_type"] == {"bug": 2, "feature": 1}
        assert stats["by_status"] == {"open": 2, "in_progress": 1}

    async def test_statistics_keys_match_list_feedback(self, feedback_db):
        await seed(feedback_db)
        retrieval = RetrievalMixin(feedback_db)
        stats = await retrieval.get_feedback_statistics()
        listed = (await retrieval.list_feedback(per_page=100))["feedback"]
        assert set(stats["by_type"]) == {f["type"] for f in listed}
        assert set(stats["by_status"]) == {f["status"] for f in listed}

    async def test_filter_by_statistics_key(self, feedback_db):
        await seed(feedback_db)
        retrieval = RetrievalMixin(feedback_db)
        stats = await retrieval.get_feedback_statistics()
        for kind, count in stats["by_type"].items():
            listed = await retrieval.list_feedback(type_filter=kind)
            assert listed["total_count"] == count

    async def test_database_stats_keyed_by_value(self, feedback_db):
        await seed(feedback_db)
        stats = await get_database_stats(feedback_db)
        assert stats["by_type"] == {"bug": 2, "feature": 1}
        assert stats["by_status"] == {"open": 2, "in_progress": 1}

    async def test_empty_database(self, feedback_db):
        stats = await RetrievalMixin(feedback_db).get_feedback_statistics()
        assert stats["by_type"] == {} and stats["by_status"] == {}


class TestEnumLabel:
    @pytest.mark.parametrize(
        ("enum_cls", "stored", "label"),
        [
            (FeedbackType, FeedbackType.BUG, "bug"),
            (FeedbackType, "BUG", "bug"),
            (FeedbackType, "bug", "bug"),
            (FeedbackStatus, "IN_PROGRESS", "in_progress"),
            (FeedbackStatus, "in_progress", "in_progress"),
            (FeedbackStatus, "archived", "archived"),
        ],
    )
    def test_label(self, enum_cls, stored, label):
        assert enum_label(enum_cls, stored) == label

    async def test_rows_written_as_values_are_merged(self, tmp_path):
        # A row holding the value instead of the name (written by hand, or by
        # another tool) is counted under the same key, not dropped.
        db = FeedbackDatabase(f"sqlite:///{tmp_path}/legacy.db")
        await db.initialize()
        try:
            await seed(db)
            async with db.session() as session:
                session.execute(
                    text(
                        "INSERT INTO feedback (type, title, description, submitter,"
                        " status, created_at, updated_at) VALUES ('bug', 't', 'd',"
                        " 's', 'open', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
                    )
                )
                session.commit()
            stats = await RetrievalMixin(db).get_feedback_statistics()
            assert stats["by_type"] == {"bug": 3, "feature": 1}
            assert stats["by_status"] == {"open": 3, "in_progress": 1}
        finally:
            await db.close()


# Event identity


def make_app(resolver=None, **kwargs):
    sink = MemorySink()
    app = FastMCP("identity")
    holder = {"spawned": []}

    @app.tool
    async def submit(job_id: str) -> dict:
        holder["mw"].record_event("job.submitted", key=job_id)
        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 submit_as(job_id: str) -> dict:
        holder["mw"].record_event(
            "job.submitted",
            key=job_id,
            user_sub="given-user",
            caller_kind="given-kind",
            client_id="given-client",
        )
        return {"job_id": job_id}

    @app.tool
    async def chat() -> str:
        holder["mw"].record_llm_call(model="m", provider="p")
        return "ok"

    @app.tool
    async def fails() -> str:
        holder["mw"].record_event("job.failed")
        raise RuntimeError("no")

    @app.tool
    async def spawn(job_id: str) -> dict:
        """Start a task that records after this 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)

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

    mw = instrument(app, [sink], identity_resolver=resolver, **kwargs)
    holder["mw"] = mw
    return app, sink, mw, holder["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()


class Resolver:
    """Counts its invocations; returns a different user each time."""

    def __init__(self, extra=None):
        self.calls = 0
        self.extra = extra or {}

    def __call__(self, context):
        self.calls += 1
        return {
            "user_sub": f"user-{self.calls}",
            "caller_kind": "pat",
            "client_id": "addon",
            "org": "acme",
            **self.extra,
        }


class AsyncResolver(Resolver):
    async def __call__(self, context):
        await asyncio.sleep(0)
        return Resolver.__call__(self, context)


IDENTITY = ("user-1", "pat", "addon")


def ident(record):
    return (record.user_sub, record.caller_kind, record.client_id)


class TestEventIdentity:
    @pytest.mark.parametrize("resolver_cls", [Resolver, AsyncResolver])
    async def test_event_in_call_gets_resolver_identity(self, resolver_cls):
        who = resolver_cls()
        app, sink, mw, _ = make_app(who)
        await call(app, mw, ("submit", {"job_id": "j1"}))
        [rec] = sink.calls
        [ev] = sink.events
        assert ident(ev) == IDENTITY
        assert ident(rec) == IDENTITY
        assert ev.call_id == rec.id
        assert rec.identity == {"org": "acme"}

    async def test_sync_tool_event_gets_identity(self):
        app, sink, mw, _ = make_app(Resolver())
        await call(app, mw, ("submit_sync", {"job_id": "j1"}))
        await asyncio.sleep(0.01)
        await mw.flush()
        [ev] = sink.events
        assert ident(ev) == IDENTITY

    async def test_llm_call_event_gets_identity(self):
        app, sink, mw, _ = make_app(Resolver())
        await call(app, mw, ("chat", {}))
        [ev] = sink.events
        assert ev.kind == "llm.call" and ident(ev) == IDENTITY

    async def test_failing_call_event_and_record_share_identity(self):
        app, sink, mw, _ = make_app(Resolver())
        await call(app, mw, ("fails", {}))
        [rec] = sink.calls
        [ev] = sink.events
        assert rec.outcome == "error"
        assert ident(ev) == ident(rec) == IDENTITY

    async def test_resolver_runs_once_per_call(self):
        who = Resolver()
        app, sink, mw, _ = make_app(who)
        await call(
            app,
            mw,
            ("submit", {"job_id": "a"}),
            ("chat", {}),
            ("fails", {}),
            ("submit", {"job_id": "b"}),
        )
        assert who.calls == 4
        assert [r.user_sub for r in sink.calls] == [
            "user-1",
            "user-2",
            "user-3",
            "user-4",
        ]
        # Each event carries its own call's identity.
        assert [(e.user_sub, e.call_id) for e in sink.events] == [
            (r.user_sub, r.id) for r in sink.calls
        ]

    async def test_explicit_identity_wins(self):
        app, sink, mw, _ = make_app(Resolver())
        await call(app, mw, ("submit_as", {"job_id": "j1"}))
        [ev] = sink.events
        assert ident(ev) == ("given-user", "given-kind", "given-client")
        assert ident(sink.calls[0]) == IDENTITY

    async def test_task_after_call_gets_no_identity(self):
        who = Resolver()
        app, sink, mw, spawned = make_app(who)
        await call(app, mw, ("spawn", {"job_id": "j1"}))
        await asyncio.gather(*spawned)
        await mw.flush()
        [rec] = sink.calls
        [ev] = sink.events
        assert ident(rec) == IDENTITY
        assert ev.kind == "job.started"
        assert ev.call_id is None and ev.session_id is None
        assert ident(ev) == (None, None, None)
        assert who.calls == 1

    async def test_outside_call_unchanged(self):
        who = Resolver()
        app, sink, mw, _ = make_app(who)
        mw.record_event("job.finished", key="j1")
        mw.record_event("job.finished", key="j2", user_sub="u9", client_id="c9")
        await mw.flush()
        first, second = sink.events
        assert ident(first) == (None, None, None) and first.call_id is None
        assert ident(second) == ("u9", None, "c9")
        assert who.calls == 0

    async def test_identity_redacted_for_events(self):
        token = "bmcp_abcdefgh12345678"
        app, sink, mw, _ = make_app(Resolver({"caller_kind": f"pat:{token}"}))
        await call(app, mw, ("submit", {"job_id": "j1"}))
        [ev] = sink.events
        assert ev.caller_kind == f"pat:{REDACTED}"
        assert sink.calls[0].caller_kind == f"pat:{REDACTED}"

    async def test_failing_resolver_leaves_identity_empty(self):
        calls = []

        def boom(context):
            calls.append(1)
            raise RuntimeError("resolver bug")

        app, sink, mw, _ = make_app(boom)
        await call(app, mw, ("submit", {"job_id": "j1"}))
        [rec] = sink.calls
        [ev] = sink.events
        assert rec.ok and ev.call_id == rec.id
        assert ev.user_sub is None and rec.user_sub is None
        assert calls == [1]

    async def test_slow_resolver_cut_off_before_tool(self):
        async def slow(context):
            await asyncio.sleep(5)
            return {"user_sub": "late"}

        app, sink, mw, _ = make_app(slow, hook_timeout=0.05)
        await call(app, mw, ("submit", {"job_id": "j1"}))
        [rec] = sink.calls
        [ev] = sink.events
        assert rec.user_sub is None and ev.user_sub is None and rec.ok

    async def test_mode_off_runs_no_resolver(self):
        who = Resolver()
        app, sink, mw, _ = make_app(who, mode="off")
        await call(app, mw, ("submit", {"job_id": "j1"}))
        assert who.calls == 0 and sink.calls == [] and sink.events == []

    async def test_excluded_tool_runs_no_resolver(self):
        who = Resolver()
        app, sink, mw, _ = make_app(who, exclude_tools={"submit"})
        await call(app, mw, ("submit", {"job_id": "j1"}))
        [ev] = sink.events
        assert who.calls == 0 and ev.user_sub is None and ev.call_id is None


# Logging


def run_python(code, *args, env=None):
    return subprocess.run(
        [sys.executable, "-c", code, *args],
        capture_output=True,
        text=True,
        timeout=60,
        env=env,
    )


class TestLogging:
    def test_import_configures_nothing_and_prints_nothing(self):
        proc = run_python(
            # FastMCP 2.x and 3.0 reset warning filters on import and authlib
            # then warns; drop warnings so stderr holds only logging output.
            "import warnings\nwarnings.showwarning = lambda *a, **k: None\n"
            "import logging, fastmcp_feedback, fastmcp_feedback.instrumentation\n"
            "root = logging.getLogger()\n"
            "print(len(root.handlers), root.level)\n"
            "print([type(h).__name__ for h in"
            " logging.getLogger('fastmcp_feedback').handlers])\n"
        )
        assert proc.returncode == 0, proc.stderr
        assert proc.stderr == ""
        handlers, names = proc.stdout.splitlines()
        assert handlers == f"0 {30}"  # no handler, level still WARNING
        assert names == "['NullHandler']"

    def test_import_logs_nothing_above_debug(self):
        # A host that set up logging first sees only the DEBUG line.
        proc = run_python(
            "import logging, json\n"
            "seen = []\n"
            "class H(logging.Handler):\n"
            "    def emit(self, r):\n"
            "        if r.name.startswith('fastmcp_feedback'):\n"
            "            seen.append((r.levelname, r.getMessage()))\n"
            "logging.basicConfig(level=logging.DEBUG, handlers=[H()])\n"
            "import fastmcp_feedback\n"
            "print(json.dumps(seen))\n"
        )
        assert proc.returncode == 0, proc.stderr
        seen = json.loads(proc.stdout.splitlines()[-1])
        assert [level for level, _ in seen] == ["DEBUG"]
        assert "initialized" in seen[0][1]

    def test_setup_logging_is_opt_in(self):
        proc = run_python("import fastmcp_feedback\nfastmcp_feedback.setup_logging()\n")
        assert proc.returncode == 0
        assert "fastmcp_feedback - INFO - FastMCP Feedback v" in proc.stderr

    def test_package_info_matches_pyproject(self):
        pyproject = ROOT / "pyproject.toml"
        if not pyproject.exists():
            pytest.skip("not running from a source checkout")
        project = tomllib.loads(pyproject.read_text())["project"]
        info = fastmcp_feedback.PACKAGE_INFO
        assert info["version"] == fastmcp_feedback.__version__ == project["version"]

        def reqs(items):
            return {str(Requirement(r)) for r in items}

        assert reqs(info["dependencies"]) == reqs(project["dependencies"])
        extras = project["optional-dependencies"]
        assert set(info["optional_dependencies"]) == set(extras)
        for name, items in extras.items():
            if any(r.startswith("fastmcp-feedback") for r in items):
                continue  # self-references are expanded in the metadata
            assert reqs(info["optional_dependencies"][name]) == reqs(items), name
        assert info["classifiers"] == sorted(project["classifiers"])


class TestBackfillCliLogging:
    @pytest.fixture
    def stored_db(self, tmp_path):
        """A calls database with one error, written by the CLI's own sink."""
        url = f"sqlite+aiosqlite:///{tmp_path}/calls.db"
        code = (
            "import asyncio, sys\n"
            "from datetime import UTC, datetime\n"
            "from fastmcp_feedback.instrumentation import DatabaseSink,"
            " ToolCallRecord\n"
            "async def main():\n"
            "    sink = DatabaseSink(sys.argv[1], create_tables=True)\n"
            "    await sink.write([ToolCallRecord(tool='t',"
            " started_at=datetime.now(UTC), duration_ms=1, ok=False,"
            " outcome='error', mode='meta', error_type='E',"
            " error_message='broke')])\n"
            "    await sink.ensure_embeddings()\n"
            "    await sink.engine.dispose()\n"
            "asyncio.run(main())\n"
        )
        proc = run_python(code, url)
        assert proc.returncode == 0, proc.stderr
        return url

    def backfill(self, url, *flags):
        env = {k: v for k, v in os.environ.items() if k != "FFB_EMBED_API_KEY"}
        return subprocess.run(
            [
                sys.executable,
                "-m",
                "fastmcp_feedback.instrumentation.backfill",
                "--database-url",
                url,
                "--dry-run",
                *flags,
            ],
            capture_output=True,
            text=True,
            env=env,
            timeout=60,
        )

    def test_verbose_logs_at_info(self, stored_db):
        proc = self.backfill(stored_db, "-v")
        assert proc.returncode == 0, proc.stderr
        assert "INFO fastmcp_feedback.instrumentation.backfill: Backfill dry run" in (
            proc.stderr
        )
        assert json.loads(proc.stdout.splitlines()[-1])["call_error"] == 1

    def test_quiet_without_verbose(self, stored_db):
        proc = self.backfill(stored_db)
        assert proc.returncode == 0, proc.stderr
        assert "INFO" not in proc.stderr
        assert "initialized" not in proc.stderr