import asyncio

from app.models.document import IngestRequest
from app.services import ingestion_queue_service


def _request(**overrides):
    payload = {
        "repo_id": "ad/repo",
        "doc_path": "README.md",
        "content": "# Intro",
        "commit_hash": "abc123",
        "snapshot_id": "snap_1",
    }
    payload.update(overrides)
    return IngestRequest(**payload)


def test_enqueue_ingestion_job_uses_selected_backend(monkeypatch):
    called = {}

    class _Publisher:
        async def publish(self, _event):
            called["published"] = True

    async def _mark_run_queued(run_id):
        called["queued"] = run_id

    monkeypatch.setattr(ingestion_queue_service, "queue_backend", lambda: "kafka")
    monkeypatch.setattr(ingestion_queue_service, "get_ingestion_queue_publisher", lambda _backend: _Publisher())
    monkeypatch.setattr(ingestion_queue_service.parse_run_repo, "mark_run_queued", _mark_run_queued)

    asyncio.run(ingestion_queue_service.enqueue_ingestion_job("run_1", _request()))

    assert called["published"] is True
    assert called["queued"] == "run_1"


def test_build_ingestion_job_event_is_deterministic():
    event1 = ingestion_queue_service.build_ingestion_job_event("run_1", _request())
    event2 = ingestion_queue_service.build_ingestion_job_event("run_1", _request())

    assert event1["run_id"] == "run_1"
    assert event1["repo_id"] == "ad/repo"
    assert event1["doc_path"] == "README.md"
    assert event1["payload"] == event2["payload"]


def test_consume_ingestion_jobs_acknowledges_success(monkeypatch):
    payload = ingestion_queue_service.build_ingestion_job_event("run_1", _request())

    called = {"jobs": 0}

    async def _run_job(_run_id, _request):
        called["jobs"] += 1

    class _Consumer:
        async def consume(self, callback, limit, concurrency):
            assert limit == 10
            assert concurrency == 3
            await callback(payload)
            return 1

    monkeypatch.setattr(ingestion_queue_service, "queue_enabled", lambda: True)
    monkeypatch.setattr(ingestion_queue_service, "queue_backend", lambda: "redis_stream")
    monkeypatch.setattr(ingestion_queue_service, "get_ingestion_queue_consumer", lambda _backend: _Consumer())
    monkeypatch.setattr(ingestion_queue_service, "run_ingestion_job", _run_job)

    consumed = asyncio.run(ingestion_queue_service.consume_ingestion_jobs(limit=10, concurrency=3))

    assert consumed == 1
    assert called["jobs"] == 1


def test_consume_ingestion_jobs_does_not_ack_failed_messages(monkeypatch):
    payload = ingestion_queue_service.build_ingestion_job_event("run_1", _request())

    async def _run_job(_run_id, _request):
        raise RuntimeError("boom")

    class _Consumer:
        async def consume(self, callback, limit, concurrency):
            assert concurrency == 2
            await callback(payload)
            return 1

    monkeypatch.setattr(ingestion_queue_service, "queue_enabled", lambda: True)
    monkeypatch.setattr(ingestion_queue_service, "queue_backend", lambda: "redis_stream")
    monkeypatch.setattr(ingestion_queue_service, "get_ingestion_queue_consumer", lambda _backend: _Consumer())
    monkeypatch.setattr(ingestion_queue_service, "run_ingestion_job", _run_job)

    try:
        asyncio.run(ingestion_queue_service.consume_ingestion_jobs(limit=10, concurrency=2))
        assert False
    except RuntimeError:
        assert True