"""Parallele Anfragen und doppelte Telegram-Nachrichten (Idempotenz)."""
import threading
import unittest

from aria_router.gateway import Handling

from tests.helpers import GatewayTestCase


def run_parallel(count, target):
    barrier = threading.Barrier(count)
    results = [None] * count

    def worker(index):
        barrier.wait()
        results[index] = target(index)

    threads = [threading.Thread(target=worker, args=(i,)) for i in range(count)]
    for thread in threads:
        thread.start()
    for thread in threads:
        thread.join(60)
    return results


class ConcurrencyTests(GatewayTestCase):
    def test_parallel_distinct_requests(self):
        gateway = self.gateway()
        results = run_parallel(12, lambda i: gateway.handle(self.request(100 + i, "#public Frage")))
        self.assertTrue(all(r.handling is Handling.REPLY and r.backend == "codex" for r in results))
        self.assertEqual(len(gateway.adapters["codex"].runner.calls), 12)
        self.assertEqual(gateway.status()["backends"]["codex"]["budget"]["jobs"], 12)
        self.assertEqual(len({r.job_id for r in results}), 12)

    def test_parallel_duplicate_requests_idempotent(self):
        gateway = self.gateway()
        results = run_parallel(10, lambda i: gateway.handle(self.request(7, "#public Frage")))
        handled = [r for r in results if r.handling is Handling.REPLY]
        dropped = [r for r in results if r.code == "DUPLICATE_MESSAGE"]
        self.assertEqual((len(handled), len(dropped)), (1, 9))
        self.assertEqual(len(gateway.adapters["codex"].runner.calls), 1)

    def test_sequential_duplicate_idempotent(self):
        gateway = self.gateway()
        first = gateway.handle(self.request(8, "hallo"))
        second = gateway.handle(self.request(8, "hallo"))
        self.assertEqual(first.handling, Handling.FORWARD_TO_CLAUDE_CHANNEL)
        self.assertEqual((second.handling, second.code), (Handling.DROP, "DUPLICATE_MESSAGE"))
        self.assertEqual(len(self.sink.delivered), 1)

    def test_duplicate_detected_after_restart(self):
        self.gateway().handle(self.request(9, "hallo"))
        again = self.gateway().handle(self.request(9, "hallo"))
        self.assertEqual(again.code, "DUPLICATE_MESSAGE")
        self.assertEqual(len(self.sink.delivered), 1)

    def test_parallel_budget_limit_holds(self):
        gateway = self.gateway({"budget": {"codex": {"daily_jobs": 3}}})
        results = run_parallel(10, lambda i: gateway.handle(self.request(200 + i, "/codex #public Frage")))
        codex_replies = [r for r in results if r.handling is Handling.REPLY]
        self.assertEqual(len(codex_replies), 3)
        self.assertEqual(len(gateway.adapters["codex"].runner.calls), 3)
        self.assertTrue(all(r.handling is Handling.BLOCKED for r in results if r not in codex_replies))


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