from __future__ import annotations

import fnmatch

from backend.services.operations import OperationsCache, OperationsEventStream


class FakeRedis:
    def __init__(self) -> None:
        self.values: dict[str, str] = {}
        self.expirations: dict[str, int] = {}
        self.stream_entries: list[tuple[str, dict[str, str]]] = []
        self.groups: set[str] = set()

    async def get(self, key: str):
        return self.values.get(key)

    async def set(self, key: str, value: str, *, ex: int):
        self.values[key] = value
        self.expirations[key] = ex
        return True

    async def unlink(self, *keys: str):
        removed = 0
        for key in keys:
            if key in self.values:
                removed += 1
                self.values.pop(key)
                self.expirations.pop(key, None)
        return removed

    async def scan_iter(self, *, match: str, count: int):
        del count
        for key in list(self.values):
            if fnmatch.fnmatch(key, match):
                yield key

    async def xadd(self, stream: str, fields, *, maxlen: int, approximate: bool):
        assert maxlen == 1000
        assert approximate is True
        message_id = f"1-{len(self.stream_entries)}"
        self.stream_entries.append((message_id, dict(fields)))
        return message_id

    async def xgroup_create(self, stream: str, group: str, *, id: str, mkstream: bool):
        assert stream == "ops:events"
        assert id == "0-0"
        assert mkstream is True
        self.groups.add(group)
        return True

    async def xreadgroup(self, group, consumer, streams, *, count: int, block: int):
        del group, consumer, count, block
        stream = next(iter(streams))
        return [(stream, list(self.stream_entries))]

    async def xack(self, stream: str, group: str, *message_ids: str):
        del stream, group
        return len(message_ids)


async def test_cache_round_trip_and_workspace_invalidation() -> None:
    redis = FakeRedis()
    cache = OperationsCache(
        redis,  # type: ignore[arg-type]
        prefix="ops",
        default_ttl_seconds=300,
    )
    key = cache.build_key("W1", "model-overview", {"month": "2026-07", "bank": "江城"})
    same_key = cache.build_key("W1", "model-overview", {"bank": "江城", "month": "2026-07"})
    other_key = cache.build_key("W2", "model-overview", {"bank": "江城"})
    assert key == same_key
    assert key != other_key
    assert await cache.set_json(key, {"total": 3, "name": "标准A卡"}) is True
    assert redis.expirations[key] == 300
    assert await cache.get_json(key) == {"total": 3, "name": "标准A卡"}
    await cache.set_json(other_key, {"total": 1})
    assert await cache.invalidate_resource("W1", "model-overview") == 1
    assert await cache.get_json(key) is None
    assert await cache.get_json(other_key) == {"total": 1}


async def test_cache_without_redis_is_a_noop() -> None:
    cache = OperationsCache(None, prefix="ops", default_ttl_seconds=300)
    assert await cache.get_json("missing") is None
    assert await cache.set_json("missing", {"value": 1}) is False
    assert await cache.delete("missing") == 0


async def test_stream_publish_group_read_and_ack() -> None:
    redis = FakeRedis()
    stream = OperationsEventStream(
        redis,  # type: ignore[arg-type]
        stream_name="ops:events",
        maxlen=1000,
    )
    assert await stream.ensure_group("notifications") is True
    message_id = await stream.publish(
        "monitor.review.due",
        {"review_id": "R1", "bank": "江城银行"},
        event_id="E1",
        idempotency_key="review:R1:due",
        trace_id="T1",
    )
    assert message_id == "1-0"
    messages = await stream.read_group("notifications", "worker-1")
    assert messages[0][0] == "1-0"
    assert messages[0][1]["event_type"] == "monitor.review.due"
    assert '"review_id":"R1"' in messages[0][1]["payload"]
    assert await stream.acknowledge("notifications", message_id) == 1