from __future__ import annotations

import asyncio
from pathlib import Path
from types import SimpleNamespace

import pytest

from app.gateway import api_read_cache


class FakeRedis:
    def __init__(self) -> None:
        self.values: dict[str, bytes] = {}
        self.expirations: dict[str, int] = {}

    async def get(self, key: str) -> bytes | None:
        return self.values.get(key)

    async def set(self, key: str, value: bytes | str, *, ex: int, nx: bool = False) -> bool | None:
        assert ex > 0
        if nx and key in self.values:
            return None
        self.values[key] = value if isinstance(value, bytes) else value.encode("utf-8")
        self.expirations[key] = ex
        return True

    async def incr(self, key: str) -> int:
        value = int(self.values.get(key, b"0")) + 1
        self.values[key] = str(value).encode("ascii")
        return value

    async def eval(self, _script: str, _num_keys: int, key: str, token: str) -> int:
        if self.values.get(key) == token.encode("utf-8"):
            del self.values[key]
            return 1
        return 0


def _config(*, enabled: bool = True) -> SimpleNamespace:
    return SimpleNamespace(
        api_read_cache=SimpleNamespace(
            enabled=enabled,
            url=None,
            password=None,
            namespace="test:homepage",
            ttl_seconds=300,
            max_value_bytes=524288,
            socket_connect_timeout_ms=50,
            socket_timeout_ms=50,
        ),
        concurrency=SimpleNamespace(
            redis=SimpleNamespace(
                enabled=enabled,
                url="redis://cache.test/0",
                password=None,
                namespace="test:deerflow",
                socket_connect_timeout_ms=50,
                socket_timeout_ms=50,
            )
        )
    )


def test_redis_cache_ttl_defaults_to_five_minutes_and_stays_bounded() -> None:
    assert api_read_cache._cache_ttl_seconds(SimpleNamespace(api_read_cache=SimpleNamespace())) == 300
    assert api_read_cache._cache_ttl_seconds(
        SimpleNamespace(api_read_cache=SimpleNamespace(ttl_seconds=9999))
    ) == 3600


@pytest.mark.asyncio
async def test_cache_round_trip_and_generation_invalidation(monkeypatch: pytest.MonkeyPatch) -> None:
    redis = FakeRedis()
    monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: None)
    monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
    monkeypatch.setattr(api_read_cache, "_client", lambda _config, **_kwargs: (redis, "test:cache"))
    config = _config()

    await api_read_cache.set_cached_json(config, "agents", "user:u1", {"items": [1, 2]})
    assert await api_read_cache.get_cached_json(config, "agents", "user:u1") == {"items": [1, 2]}
    payload = await api_read_cache.get_cached_json_payload(config, "agents", "user:u1")
    assert payload == b'{"items":[1,2]}'

    await api_read_cache.invalidate_cached_collection(config, "agents")
    assert await api_read_cache.get_cached_json(config, "agents", "user:u1") is None


@pytest.mark.asyncio
async def test_concurrent_cache_misses_wait_for_single_fill(monkeypatch: pytest.MonkeyPatch) -> None:
    redis = FakeRedis()
    monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: None)
    monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
    monkeypatch.setattr(api_read_cache, "_client", lambda _config, **_kwargs: (redis, "test:cache"))
    config = _config()
    _, owner_token = await api_read_cache.wait_for_cached_json_or_claim_fill(
        config, "agents", "user:u1", version="0"
    )
    assert owner_token is not None

    waiter = asyncio.create_task(
        api_read_cache.wait_for_cached_json_or_claim_fill(
            config, "agents", "user:u1", version="0", wait_ms=500
        )
    )
    await asyncio.sleep(0.05)
    await api_read_cache.set_cached_json(
        config, "agents", "user:u1", {"items": ["shared fill"]}, expected_version="0"
    )
    await api_read_cache.release_cache_fill(config, "agents", "user:u1", owner_token)
    cached, waiter_token = await waiter
    assert cached == {"items": ["shared fill"]}
    assert waiter_token is None
    assert not any(":fill:" in key for key in redis.values)

    stale_version = await api_read_cache.get_cache_version(config, "agents")
    await api_read_cache.invalidate_cached_collection(config, "agents")
    await api_read_cache.set_cached_json(
        config,
        "agents",
        "user:u1",
        {"items": ["stale"]},
        expected_version=stale_version,
    )
    assert await api_read_cache.get_cached_json(config, "agents", "user:u1") is None


@pytest.mark.asyncio
async def test_concurrent_http_payload_misses_wait_for_single_fill(monkeypatch: pytest.MonkeyPatch) -> None:
    redis = FakeRedis()
    monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: None)
    monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
    monkeypatch.setattr(api_read_cache, "_client", lambda _config, **_kwargs: (redis, "test:cache"))
    config = _config()
    _, owner_token = await api_read_cache.wait_for_cached_json_payload_or_claim_fill(
        config, "skills", "user:u1", version="0"
    )
    assert owner_token is not None

    waiter = asyncio.create_task(
        api_read_cache.wait_for_cached_json_payload_or_claim_fill(
            config, "skills", "user:u1", version="0", wait_ms=500
        )
    )
    await asyncio.sleep(0.05)
    await api_read_cache.set_cached_json(
        config,
        "skills",
        "user:u1",
        {"skills": [{"name": "one"}]},
        expected_version="0",
    )
    await api_read_cache.release_cache_fill(config, "skills", "user:u1", owner_token)
    payload, waiter_token = await waiter
    assert payload == b'{"skills":[{"name":"one"}]}'
    assert waiter_token is None


@pytest.mark.asyncio
async def test_disabled_redis_uses_the_database_path() -> None:
    config = _config(enabled=True)
    config.api_read_cache = SimpleNamespace(enabled=False)
    assert api_read_cache._redis_settings(config) is None
    api_read_cache.set_api_read_cache_enabled_local(False)
    assert await api_read_cache.get_cached_json(config, "skills", "user:u1") is None
    await api_read_cache.set_cached_json(config, "skills", "user:u1", {"items": []})


def test_homepage_cache_can_reuse_a_separate_redis_connection() -> None:
    config = _config(enabled=False)
    config.api_read_cache = SimpleNamespace(
        enabled=True,
        url="redis://homepage-cache.test/2",
        password="secret",
        namespace="test:homepage",
    )
    assert api_read_cache._redis_settings(config) == (
        "redis://homepage-cache.test/2",
        "secret",
        "test:homepage",
    )


def test_enabled_homepage_cache_reuses_concurrency_redis() -> None:
    config = _config(enabled=True)
    config.api_read_cache = SimpleNamespace(
        enabled=True,
        url=None,
        password=None,
        namespace="test:homepage",
    )
    assert api_read_cache._redis_settings(config) == (
        "redis://cache.test/0",
        None,
        "test:homepage",
    )


def test_separate_cache_redis_does_not_inherit_an_unrelated_password() -> None:
    config = _config(enabled=True)
    config.concurrency.redis.password = "concurrency-secret"
    config.api_read_cache = SimpleNamespace(
        enabled=True,
        url="redis://homepage-cache.test/2",
        password=None,
        namespace="test:homepage",
    )
    assert api_read_cache._redis_settings(config) == (
        "redis://homepage-cache.test/2",
        None,
        "test:homepage",
    )


@pytest.mark.asyncio
async def test_admin_cache_switch_and_ttl_are_persisted_and_survive_process_refresh(monkeypatch: pytest.MonkeyPatch) -> None:
    from deerflow.persistence.api_read_cache_settings.model import ApiReadCacheSettingRow

    class FakeSession:
        def __init__(self) -> None:
            self.row: ApiReadCacheSettingRow | None = None
            self.added: list[ApiReadCacheSettingRow] = []

        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return False

        async def get(self, _model, key: str, **_kwargs):
            assert key == "default"
            return self.row

        def add(self, row: ApiReadCacheSettingRow) -> None:
            self.added.append(row)
            self.row = row

        async def commit(self) -> None:
            return None

    session = FakeSession()
    monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: lambda: session)
    monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
    config = _config(enabled=False)

    assert await api_read_cache.get_api_read_cache_settings(config, force_refresh=True) == (False, 300)
    await api_read_cache.save_api_read_cache_settings(True, 900, config=config)
    assert session.row is not None and session.row.enabled is True
    assert session.row.ttl_seconds == 900
    assert await api_read_cache.get_api_read_cache_settings(config) == (True, 900)

    # A new process has no in-memory value and reads the persisted MySQL row.
    monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
    assert await api_read_cache.get_api_read_cache_settings(config, force_refresh=True) == (True, 900)
    await api_read_cache.save_api_read_cache_enabled(False)
    assert session.row.enabled is False
    assert session.row.ttl_seconds == 900


@pytest.mark.asyncio
async def test_cache_toggle_mysql_read_is_singleflight_under_concurrency(monkeypatch: pytest.MonkeyPatch) -> None:
    class FakeSession:
        def __init__(self) -> None:
            self.get_calls = 0

        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return False

        async def get(self, _model, _key):
            self.get_calls += 1
            await asyncio.sleep(0.01)
            return None

    session = FakeSession()
    monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: lambda: session)
    monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
    config = _config(enabled=False)

    results = await asyncio.gather(
        *(api_read_cache.is_api_read_cache_enabled(config) for _ in range(70))
    )

    assert results == [False] * 70
    assert session.get_calls == 1


@pytest.mark.asyncio
async def test_configured_mysql_ttl_is_used_for_agent_and_skill_cache_writes(monkeypatch: pytest.MonkeyPatch) -> None:
    redis = FakeRedis()

    class FakeSession:
        async def __aenter__(self):
            return self

        async def __aexit__(self, *_args):
            return False

        async def get(self, _model, _key):
            return SimpleNamespace(enabled=True, ttl_seconds=900)

    monkeypatch.setattr("deerflow.persistence.engine.get_session_factory", lambda: lambda: FakeSession())
    monkeypatch.setattr(api_read_cache, "_api_read_cache_settings_cache", None)
    monkeypatch.setattr(api_read_cache, "_client", lambda _config, **_kwargs: (redis, "test:cache"))
    config = _config()

    for collection in ("agents", "skills"):
        await api_read_cache.set_cached_json(config, collection, "user:u1", {"items": []})
        expirations = [
            expiry
            for key, expiry in redis.expirations.items()
            if f":{collection}:v0:" in key
        ]
        assert expirations == [900]


def test_existing_api_cache_setting_receives_five_minute_ttl(tmp_path: Path) -> None:
    import sqlalchemy as sa
    from alembic import command
    from alembic.config import Config as AlembicConfig

    database = tmp_path / "api-read-cache.db"
    engine = sa.create_engine(f"sqlite:///{database.resolve().as_posix()}")
    with engine.begin() as connection:
        connection.exec_driver_sql(
            "CREATE TABLE api_read_cache_settings ("
            "`key` VARCHAR(32) PRIMARY KEY, enabled BOOLEAN NOT NULL, updated_at DATETIME NOT NULL)"
        )
        connection.exec_driver_sql(
            "INSERT INTO api_read_cache_settings (`key`, enabled, updated_at) "
            "VALUES ('default', 1, CURRENT_TIMESTAMP)"
        )
        connection.exec_driver_sql(
            "CREATE TABLE alembic_version (version_num VARCHAR(32) NOT NULL PRIMARY KEY)"
        )
        connection.exec_driver_sql(
            "INSERT INTO alembic_version (version_num) VALUES ('20261026_02')"
        )
    engine.dispose()

    migrations = Path(__file__).parents[1] / "packages/harness/deerflow/persistence/migrations"
    config = AlembicConfig(str(migrations / "alembic.ini"))
    config.set_main_option(
        "sqlalchemy.url",
        f"sqlite+aiosqlite:///{database.resolve().as_posix()}",
    )
    command.upgrade(config, "head")

    engine = sa.create_engine(f"sqlite:///{database.resolve().as_posix()}")
    with engine.connect() as connection:
        setting = connection.execute(
            sa.text("SELECT enabled, ttl_seconds, collection_overrides, browser_status_enabled, cache_epoch FROM api_read_cache_settings WHERE `key`='default'")
        ).one()
        revision = connection.execute(sa.text("SELECT version_num FROM alembic_version")).scalar_one()
    engine.dispose()
    assert bool(setting.enabled) is True
    assert setting.ttl_seconds == 300
    assert setting.collection_overrides is None
    assert bool(setting.browser_status_enabled) is True
    assert setting.cache_epoch == "0"
    assert revision == "20261102_02"


@pytest.mark.asyncio
async def test_skill_catalog_projection_uses_compact_rows_without_long_fields(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from sqlalchemy import update
    from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

    from app.gateway import skill_catalog_cache
    from deerflow.persistence.skills.model import SkillRow

    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as connection:
        await connection.run_sync(SkillRow.__table__.create)
    session_factory = async_sessionmaker(engine, expire_on_commit=False)
    monkeypatch.setattr(skill_catalog_cache, "get_session_factory", lambda: session_factory)
    monkeypatch.setattr(skill_catalog_cache, "get_or_new_skill_storage", lambda **_kwargs: SimpleNamespace())
    monkeypatch.setattr(skill_catalog_cache, "_load_usage_entries", lambda: {})

    skill_file = tmp_path / "SKILL.md"
    skill_file.write_text("---\nname: catalog-demo\n---\n", encoding="utf-8")
    skill = SimpleNamespace(
        name="catalog-demo",
        description="compact metadata",
        license="MIT",
        category="public",
        enabled=True,
        always_on=False,
        skill_file=skill_file,
    )

    assert await skill_catalog_cache.refresh_skill_catalog_projection(skills=[skill]) == 1
    async with session_factory() as session:
        await session.execute(update(SkillRow).where(SkillRow.name == "catalog-demo").values(detail="x" * 5000))
        await session.commit()

    rows = await skill_catalog_cache.load_skill_catalog_projection(user_id="user-1")
    assert rows is not None and len(rows) == 1
    assert rows[0]["name"] == "catalog-demo"
    assert rows[0]["description"] == "compact metadata"
    assert "detail" not in rows[0]
    assert "detail" not in {column.key for column in skill_catalog_cache._SKILL_LIST_COLUMNS}
    await engine.dispose()


@pytest.mark.asyncio
async def test_skill_catalog_schema_adds_columns_to_existing_table_without_losing_data(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    from sqlalchemy import inspect
    from sqlalchemy.ext.asyncio import create_async_engine

    from app.gateway import skill_catalog_cache

    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as connection:
        await connection.exec_driver_sql(
            "CREATE TABLE skills ("
            "name VARCHAR(64) PRIMARY KEY, owner_user_id VARCHAR(64), "
            "published BOOLEAN NOT NULL DEFAULT 0, detail TEXT, name_zh VARCHAR(128), "
            "always_on BOOLEAN NOT NULL DEFAULT 0, square_id VARCHAR(64) NOT NULL DEFAULT '', "
            "created_at DATETIME NOT NULL, updated_at DATETIME NOT NULL)"
        )
        await connection.exec_driver_sql(
            "INSERT INTO skills (name, detail, created_at, updated_at) "
            "VALUES ('existing-skill', 'keep this annotation', CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)"
        )
    monkeypatch.setattr(skill_catalog_cache, "get_engine", lambda: engine)

    await skill_catalog_cache.ensure_skill_catalog_schema()
    # The migration must be safe to retry after process restarts.
    await skill_catalog_cache.ensure_skill_catalog_schema()

    async with engine.connect() as connection:
        columns = await connection.run_sync(
            lambda sync_connection: {column["name"] for column in inspect(sync_connection).get_columns("skills")}
        )
        preserved = (
            await connection.exec_driver_sql(
                "SELECT detail, available, category, detail_preview FROM skills WHERE name='existing-skill'"
            )
        ).one()
    assert "detail_preview" in columns
    assert preserved[0] == "keep this annotation"
    assert bool(preserved[1]) is True
    assert preserved[2] == "custom"
    assert preserved[3] == "keep this annotation"
    await engine.dispose()