import json
import unittest

from app.models import SMTPResult
from app.validator.retry_queue import (
    MemoryRetryStore,
    RedisRetryStore,
    RetryEntry,
    RetryQueue,
    drain_retry_queue,
)


class FakeRedis:
    """Minimal async Redis subset (hash + zset) used by RedisRetryStore."""

    def __init__(self):
        self.hashes: dict[str, dict[str, str]] = {}
        self.zsets: dict[str, dict[str, float]] = {}

    async def hset(self, key, field, value):
        self.hashes.setdefault(key, {})[field] = value

    async def hget(self, key, field):
        return self.hashes.get(key, {}).get(field)

    async def hdel(self, key, field):
        self.hashes.get(key, {}).pop(field, None)

    async def zadd(self, key, mapping):
        self.zsets.setdefault(key, {}).update(mapping)

    async def zrem(self, key, member):
        self.zsets.get(key, {}).pop(member, None)

    async def zrangebyscore(self, key, min_, max_, start=0, num=None):
        items = [
            (m, s) for m, s in self.zsets.get(key, {}).items() if min_ <= s <= max_
        ]
        items.sort(key=lambda kv: kv[1])
        members = [m for m, _ in items]
        return members[start : (start + num) if num else None]

    async def zcard(self, key):
        return len(self.zsets.get(key, {}))

    async def hlen(self, key):
        return len(self.hashes.get(key, {}))


class RedisRetryStoreTests(unittest.IsolatedAsyncioTestCase):
    """Production needs the queue to survive a restart and be shared across
    workers — an in-process dict does neither."""

    async def test_add_persists_entry(self):
        store = RedisRetryStore(client=FakeRedis())
        entry = RetryEntry(email="a@corp.test", mx_host="mx.corp.test", due_at=500.0)
        await store.add(entry)
        got = await store.get("a@corp.test")
        self.assertIsNotNone(got)
        self.assertEqual(got.mx_host, "mx.corp.test")
        self.assertEqual(got.due_at, 500.0)

    async def test_due_returns_only_matured_entries(self):
        store = RedisRetryStore(client=FakeRedis())
        await store.add(RetryEntry(email="early@corp.test", mx_host="mx", due_at=100.0))
        await store.add(RetryEntry(email="late@corp.test", mx_host="mx", due_at=9999.0))
        due = await store.due(now=200.0, limit=10)
        self.assertEqual([e.email for e in due], ["early@corp.test"])

    async def test_due_is_ordered_by_time(self):
        store = RedisRetryStore(client=FakeRedis())
        await store.add(RetryEntry(email="b@corp.test", mx_host="mx", due_at=300.0))
        await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=200.0))
        due = await store.due(now=400.0, limit=10)
        self.assertEqual([e.email for e in due], ["a@corp.test", "b@corp.test"])

    async def test_remove_clears_both_indexes(self):
        redis = FakeRedis()
        store = RedisRetryStore(client=redis)
        await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=1.0))
        await store.remove("a@corp.test")
        self.assertIsNone(await store.get("a@corp.test"))
        self.assertEqual(await store.count(), 0)

    async def test_readd_updates_rather_than_duplicates(self):
        store = RedisRetryStore(client=FakeRedis())
        await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=1.0))
        await store.add(RetryEntry(email="a@corp.test", mx_host="mx", due_at=99.0, attempts=1))
        self.assertEqual(await store.count(), 1)
        self.assertEqual((await store.get("a@corp.test")).attempts, 1)

    async def test_corrupt_entry_is_skipped_not_raised(self):
        redis = FakeRedis()
        store = RedisRetryStore(client=redis)
        await store.add(RetryEntry(email="ok@corp.test", mx_host="mx", due_at=1.0))
        # Simulate a half-written/garbage payload for another member.
        redis.hashes.setdefault("vx:retry:entries", {})["bad@corp.test"] = "{not json"
        redis.zsets.setdefault("vx:retry:due", {})["bad@corp.test"] = 1.0
        due = await store.due(now=5.0, limit=10)
        self.assertEqual([e.email for e in due], ["ok@corp.test"])


class DrainWorkerTests(unittest.IsolatedAsyncioTestCase):
    """A queue that only fills is a leak. Due entries must actually be retried,
    and a definitive answer must clear the entry."""

    async def test_drain_reprobes_due_entries_and_resolves_them(self):
        clock = {"t": 0.0}
        queue = RetryQueue(
            store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"]
        )
        await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
        clock["t"] = 100.0

        probed: list[tuple[str, str]] = []

        async def prober(email, mx_host):
            probed.append((email, mx_host))
            return SMTPResult(verified=True, smtp_code=250)

        stats = await drain_retry_queue(prober, queue=queue, limit=10)

        self.assertEqual(probed, [("a@corp.test", "mx.corp.test")])
        self.assertEqual(stats["resolved"], 1)
        self.assertEqual(stats["requeued"], 0)
        self.assertEqual(await queue.size(), 0)

    async def test_still_greylisted_entry_is_requeued_with_backoff(self):
        clock = {"t": 0.0}
        queue = RetryQueue(
            store=MemoryRetryStore(), backoff_seconds=10, max_attempts=3,
            now=lambda: clock["t"],
        )
        await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
        clock["t"] = 100.0

        async def prober(email, mx_host):
            return SMTPResult(verified=None, smtp_code=451, smtp_message="still greylisted")

        stats = await drain_retry_queue(prober, queue=queue, limit=10)

        self.assertEqual(stats["requeued"], 1)
        self.assertEqual(await queue.size(), 1)
        # Backed off: not immediately due again.
        self.assertEqual(await queue.due(), [])

    async def test_rejected_entry_is_resolved_and_removed(self):
        clock = {"t": 0.0}
        queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"])
        await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
        clock["t"] = 100.0

        async def prober(email, mx_host):
            return SMTPResult(verified=False, smtp_code=550, smtp_message="550 no such user")

        stats = await drain_retry_queue(prober, queue=queue, limit=10)

        self.assertEqual(stats["resolved"], 1)
        self.assertEqual(await queue.size(), 0)

    async def test_prober_error_requeues_without_crashing(self):
        clock = {"t": 0.0}
        queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"])
        await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
        clock["t"] = 100.0

        async def prober(email, mx_host):
            raise RuntimeError("mx exploded")

        stats = await drain_retry_queue(prober, queue=queue, limit=10)

        self.assertEqual(stats["errors"], 1)
        self.assertEqual(await queue.size(), 1, "must stay queued for another try")

    async def test_empty_queue_is_a_noop(self):
        queue = RetryQueue(store=MemoryRetryStore(), now=lambda: 0.0)

        async def prober(email, mx_host):
            raise AssertionError("should not be called")

        stats = await drain_retry_queue(prober, queue=queue, limit=10)
        self.assertEqual(
            stats, {"checked": 0, "resolved": 0, "requeued": 0, "errors": 0, "results": []}
        )

    async def test_results_are_returned_for_persistence(self):
        clock = {"t": 0.0}
        queue = RetryQueue(store=MemoryRetryStore(), backoff_seconds=10, now=lambda: clock["t"])
        await queue.enqueue("a@corp.test", "mx.corp.test", code=451)
        clock["t"] = 100.0

        async def prober(email, mx_host):
            return SMTPResult(verified=True, smtp_code=250)

        stats = await drain_retry_queue(prober, queue=queue, limit=10)
        self.assertEqual(stats["results"][0][0], "a@corp.test")
        self.assertTrue(stats["results"][0][1].verified)


if __name__ == "__main__":
    unittest.main()