import sqlite3
import tempfile
import unittest
from concurrent.futures import ThreadPoolExecutor
from contextlib import closing
from pathlib import Path
from threading import Barrier

from inbox import EventConflict, initialise, record_event


class InboxTests(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.addCleanup(self.directory.cleanup)
        self.path = Path(self.directory.name) / "events.sqlite3"
        initialise(self.path)

    def rows(self):
        with closing(sqlite3.connect(self.path)) as connection:
            return connection.execute(
                "SELECT source, event_id, payload FROM inbox ORDER BY source, event_id"
            ).fetchall()

    def test_stored_event_is_visible_after_connection_reopens(self):
        self.assertEqual(record_event(self.path, "alpha", "1", b"hello"), "stored")
        self.assertEqual(self.rows(), [("alpha", "1", b"hello")])

    def test_retry_does_not_create_another_row(self):
        record_event(self.path, "alpha", "1", b"hello")
        self.assertEqual(record_event(self.path, "alpha", "1", b"hello"), "duplicate")
        self.assertEqual(len(self.rows()), 1)

    def test_conflict_preserves_original_and_next_write_works(self):
        record_event(self.path, "alpha", "1", b"hello")
        with self.assertRaises(EventConflict):
            record_event(self.path, "alpha", "1", b"changed")
        record_event(self.path, "alpha", "2", b"next")
        self.assertEqual(self.rows(), [("alpha", "1", b"hello"), ("alpha", "2", b"next")])

    def test_different_sources_can_use_the_same_event_identifier(self):
        record_event(self.path, "alpha", "1", b"hello")
        record_event(self.path, "beta", "1", b"other")
        self.assertEqual(len(self.rows()), 2)

    def test_concurrent_retries_store_one_event(self):
        barrier = Barrier(4)
        def submit(_):
            barrier.wait(timeout=5)
            return record_event(self.path, "alpha", "1", b"same")
        with ThreadPoolExecutor(max_workers=4) as workers:
            results = list(workers.map(submit, range(4)))
        self.assertEqual(results.count("stored"), 1)
        self.assertEqual(results.count("duplicate"), 3)
        self.assertEqual(len(self.rows()), 1)

    def test_locked_database_raises_and_does_not_claim_success(self):
        with closing(sqlite3.connect(self.path, isolation_level=None)) as lock:
            lock.execute("BEGIN IMMEDIATE")
            with self.assertRaises(sqlite3.OperationalError):
                record_event(self.path, "alpha", "1", b"hello", timeout=0.02)
            lock.rollback()
        self.assertEqual(self.rows(), [])

    def test_invalid_input_leaves_no_rows(self):
        for source, event_id, body, error in [
            ("", "1", b"hello", ValueError),
            ("alpha", "  ", b"hello", ValueError),
            ("alpha", 1, b"hello", ValueError),
            ("alpha", "1", "not bytes", TypeError),
        ]:
            with self.subTest(source=source, event_id=event_id):
                with self.assertRaises(error):
                    record_event(self.path, source, event_id, body)
        self.assertEqual(self.rows(), [])


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