dotb_ai-platform/tests/test_webhook_dispatcher.py

import asyncio
import json
from datetime import UTC, datetime, timedelta
from pathlib import Path

import httpx
import pytest
from alembic import command
from alembic.config import Config
from pydantic import SecretStr
from sqlalchemy import select, update
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine

from lead_scoring.config import Settings
from lead_scoring.db.models.outbox_event import OutboxEvent
from lead_scoring.db.models.scoring_run import ScoringRun
from lead_scoring.main import create_app
from lead_scoring.schemas.scoring_run import ScoringRunStatus
from lead_scoring.schemas.webhook import (
    OutboxDeliveryStatus,
)
from lead_scoring.webhook.dispatcher import (
    DeliveryClassification,
    DeliveryResult,
    WebhookDispatcher,
    calculate_backoff,
    claim_outbox_events,
    dispatch_single_event,
    record_delivery_result,
    resolve_webhook_target_url,
)
from lead_scoring.webhook.signing import (
    TEST_SIGNING_SECRET,
    sanitize_secret,
    sign_webhook,
)

PROJECT_ROOT = Path(__file__).resolve().parent.parent


def get_alembic_config(database_url: str) -> Config:
    ini_path = PROJECT_ROOT / "alembic.ini"
    alembic_cfg = Config(str(ini_path))
    alembic_cfg.attributes["sqlalchemy_url"] = database_url
    return alembic_cfg


def test_calculate_backoff_deterministic() -> None:
    assert calculate_backoff(0, base_seconds=2.0, max_seconds=60.0) == 0.0
    assert calculate_backoff(1, base_seconds=2.0, max_seconds=60.0) == 2.0
    assert calculate_backoff(2, base_seconds=2.0, max_seconds=60.0) == 4.0
    assert calculate_backoff(3, base_seconds=2.0, max_seconds=60.0) == 8.0
    assert calculate_backoff(4, base_seconds=2.0, max_seconds=60.0) == 16.0
    assert calculate_backoff(5, base_seconds=2.0, max_seconds=60.0) == 32.0
    assert calculate_backoff(6, base_seconds=2.0, max_seconds=60.0) == 60.0
    assert calculate_backoff(10, base_seconds=2.0, max_seconds=60.0) == 60.0


def test_resolve_webhook_target_url() -> None:
    assert (
        resolve_webhook_target_url("http://localhost:8080")
        == "http://localhost:8080/api/v1/internal/lead-scoring/events"
    )
    assert (
        resolve_webhook_target_url("http://localhost:8080/")
        == "http://localhost:8080/api/v1/internal/lead-scoring/events"
    )
    assert (
        resolve_webhook_target_url("http://localhost:8080/api/v1/internal/lead-scoring/events")
        == "http://localhost:8080/api/v1/internal/lead-scoring/events"
    )
    with pytest.raises(ValueError, match="path must be empty"):
        resolve_webhook_target_url("https://omni.dotb.vn/custom/path")


def test_settings_rejects_arbitrary_webhook_endpoint_path() -> None:
    with pytest.raises(ValueError, match="path must be empty"):
        Settings(
            webhook_subscriber_url="https://omni.dotb.vn/custom/path",
            webhook_signing_secret=SecretStr("test-secret"),
        )


def test_sign_webhook_matches_released_contract_signature_vectors() -> None:
    vectors_file = (
        PROJECT_ROOT / "contracts" / "lead-scoring-webhook" / "v1" / "signature-vectors.json"
    )
    vectors_data = json.loads(vectors_file.read_text(encoding="utf-8"))
    test_secret = vectors_data["test_secret"]
    assert test_secret == TEST_SIGNING_SECRET

    for vector in vectors_data["vectors"]:
        body_file = PROJECT_ROOT / "contracts" / "lead-scoring-webhook" / "v1" / vector["body_file"]
        body_bytes = body_file.read_bytes()
        calculated_sig = sign_webhook(
            secret=test_secret,
            event_id=vector["event_id"],
            timestamp=vector["timestamp"],
            body=body_bytes,
        )
        assert calculated_sig == vector["signature"]


def test_signature_changes_with_fresh_timestamp_on_retry() -> None:
    event_id = "11111111-1111-4111-8111-111111111111"
    body = b'{"hello": "world"}\n'
    sig1 = sign_webhook(TEST_SIGNING_SECRET, event_id, "1787832600", body)
    sig2 = sign_webhook(TEST_SIGNING_SECRET, event_id, "1787832660", body)
    assert sig1 != sig2
    assert sig1.startswith("v1=")
    assert sig2.startswith("v1=")


def test_secret_sanitization() -> None:
    secret = "whsec_super_confidential_key_12345"
    raw_error = f"Failed to authenticate with secret {secret} against endpoint"
    sanitized = sanitize_secret(raw_error, secret)
    assert secret not in sanitized
    assert "[REDACTED]" in sanitized
    assert sanitize_secret("No secret here", None) == "No secret here"


@pytest.fixture
async def migrated_db(temp_postgres_db: str):
    alembic_cfg = get_alembic_config(temp_postgres_db)
    await asyncio.to_thread(command.upgrade, alembic_cfg, "head")
    engine = create_async_engine(temp_postgres_db)
    session_factory = async_sessionmaker(engine, expire_on_commit=False)
    try:
        yield session_factory, engine, temp_postgres_db
    finally:
        await engine.dispose()
        await asyncio.to_thread(command.downgrade, alembic_cfg, "base")


@pytest.mark.asyncio
async def test_claim_outbox_events_single_and_concurrent(migrated_db):
    session_factory, _, _ = migrated_db
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    # Insert 5 events (3 pending, 1 delivering with active lease, 1 delivering with expired lease)
    async with session_factory() as session:
        for i in range(1, 4):
            session.add(
                OutboxEvent(
                    id=f"00000000-0000-0000-0000-00000000000{i}",
                    event_type="lead_scoring.run.terminal",
                    event_version="1.0.0",
                    run_id=f"run-{i}",
                    payload={"status": "completed"},
                    occurred_at=now - timedelta(minutes=10 - i),
                    delivery_status=OutboxDeliveryStatus.PENDING.value,
                )
            )
        # Active delivering event
        session.add(
            OutboxEvent(
                id="00000000-0000-0000-0000-000000000004",
                event_type="lead_scoring.run.terminal",
                event_version="1.0.0",
                run_id="run-4",
                payload={"status": "completed"},
                occurred_at=now - timedelta(minutes=5),
                delivery_status=OutboxDeliveryStatus.DELIVERING.value,
                lease_expires_at=now + timedelta(seconds=20),
            )
        )
        # Expired delivering event (crash recovery)
        session.add(
            OutboxEvent(
                id="00000000-0000-0000-0000-000000000005",
                event_type="lead_scoring.run.terminal",
                event_version="1.0.0",
                run_id="run-5",
                payload={"status": "completed"},
                occurred_at=now - timedelta(minutes=4),
                delivery_status=OutboxDeliveryStatus.DELIVERING.value,
                lease_expires_at=now - timedelta(seconds=5),
            )
        )
        await session.commit()

    # Claim batch of 10
    async with session_factory() as session:
        claimed = await claim_outbox_events(
            session=session, batch_size=10, lease_seconds=30.0, now=now
        )
        await session.commit()

    claimed_ids = {e.id for e in claimed}
    # Should claim 1, 2, 3 (pending) and 5 (expired lease), but NOT 4 (active lease)
    assert len(claimed) == 4
    assert claimed_ids == {
        "00000000-0000-0000-0000-000000000001",
        "00000000-0000-0000-0000-000000000002",
        "00000000-0000-0000-0000-000000000003",
        "00000000-0000-0000-0000-000000000005",
    }
    for e in claimed:
        assert e.delivery_status == OutboxDeliveryStatus.DELIVERING.value
        assert e.lease_expires_at == now + timedelta(seconds=30.0)


@pytest.mark.asyncio
async def test_concurrent_dispatchers_no_duplicate_claims(migrated_db):
    session_factory, _, _ = migrated_db
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    # Insert 20 pending events
    async with session_factory() as session:
        for i in range(20):
            session.add(
                OutboxEvent(
                    id=f"00000000-0000-0000-0000-{i:012d}",
                    event_type="lead_scoring.run.terminal",
                    event_version="1.0.0",
                    run_id=f"run-{i}",
                    payload={"status": "completed"},
                    occurred_at=now + timedelta(seconds=i),
                    delivery_status=OutboxDeliveryStatus.PENDING.value,
                )
            )
        await session.commit()

    # 4 concurrent workers claim batches of 5 concurrently
    async def worker_claim():
        async with session_factory() as session:
            claimed = await claim_outbox_events(
                session=session, batch_size=5, lease_seconds=30.0, now=now
            )
            await session.commit()
            return [e.id for e in claimed]

    results = await asyncio.gather(
        worker_claim(),
        worker_claim(),
        worker_claim(),
        worker_claim(),
    )

    all_claimed_ids = [item for sublist in results for item in sublist]
    assert len(all_claimed_ids) == 20
    # No event claimed twice
    assert len(set(all_claimed_ids)) == 20


@pytest.mark.asyncio
async def test_dispatch_all_200_ingest_statuses_mark_delivered(migrated_db):
    session_factory, _, _ = migrated_db
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    for ingest_status in ["accepted", "duplicate", "stale", "ordering_conflict"]:
        event_id = f"00000000-0000-0000-0000-0000000000{ingest_status[:2]}"
        run_id = f"run-200-{ingest_status}"
        event = OutboxEvent(
            id=event_id,
            event_type="lead_scoring.run.terminal",
            event_version="1.0.0",
            run_id=run_id,
            payload={"event_id": event_id, "data": {"status": "completed"}},
            occurred_at=now,
            delivery_status=OutboxDeliveryStatus.DELIVERING.value,
        )

        mock_transport = httpx.MockTransport(
            lambda request, status=ingest_status: httpx.Response(
                200,
                json={"status": status, "event_id": "00000000-0000-0000-0000-000000000001"},
            )
        )
        async with httpx.AsyncClient(transport=mock_transport) as client:
            result = await dispatch_single_event(
                event=event,
                target_url="http://localhost:8080/api/v1/internal/lead-scoring/events",
                signing_key_id="test-key",
                signing_secret=TEST_SIGNING_SECRET,
                timeout_seconds=5.0,
                http_client=client,
                now=now,
            )

        assert result.classification == DeliveryClassification.DELIVERED
        assert result.http_status == 200
        assert result.ingest_status == ingest_status

        # Test recording result in DB
        async with session_factory() as session:
            session.add(event)
            await session.commit()

            await record_delivery_result(
                session=session,
                event=event,
                result=result,
                max_attempts=5,
                backoff_base_seconds=2.0,
                backoff_max_seconds=60.0,
                now=now,
            )
            await session.commit()

        async with session_factory() as check_session:
            updated = await check_session.get(OutboxEvent, event_id)
            assert updated is not None
            assert updated.delivery_status == OutboxDeliveryStatus.DELIVERED.value
            assert updated.delivered_at == now
            assert updated.lease_expires_at is None
            assert updated.last_http_status == 200
            assert updated.last_error_code is None


@pytest.mark.asyncio
async def test_dispatch_terminal_400_and_401_go_to_dlq(migrated_db):
    session_factory, _, _ = migrated_db
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    cases = [
        (400, "receiver_bad_request", "Bad JSON format"),
        (401, "receiver_unauthorized", "Invalid signature"),
    ]

    for status_code, expected_err, response_text in cases:
        event_id = f"00000000-0000-0000-0000-000000000{status_code}"
        run_id = f"run-err-{status_code}"
        event = OutboxEvent(
            id=event_id,
            event_type="lead_scoring.run.terminal",
            event_version="1.0.0",
            run_id=run_id,
            payload={"status": "completed"},
            occurred_at=now,
            delivery_status=OutboxDeliveryStatus.DELIVERING.value,
            attempt_count=1,
        )

        mock_transport = httpx.MockTransport(
            lambda request, sc=status_code, txt=response_text: httpx.Response(sc, text=txt)
        )
        async with httpx.AsyncClient(transport=mock_transport) as client:
            result = await dispatch_single_event(
                event=event,
                target_url="http://localhost:8080/api/v1/internal/lead-scoring/events",
                signing_key_id="test-key",
                signing_secret=TEST_SIGNING_SECRET,
                timeout_seconds=5.0,
                http_client=client,
                now=now,
            )

        assert result.classification == DeliveryClassification.DEAD_LETTER
        assert result.http_status == status_code
        assert result.error_code == expected_err

        async with session_factory() as session:
            session.add(event)
            await session.commit()

            await record_delivery_result(
                session=session,
                event=event,
                result=result,
                max_attempts=5,
                backoff_base_seconds=2.0,
                backoff_max_seconds=60.0,
                now=now,
            )
            await session.commit()

        async with session_factory() as check_session:
            updated = await check_session.get(OutboxEvent, event_id)
            assert updated is not None
            assert updated.delivery_status == OutboxDeliveryStatus.DEAD_LETTER.value
            assert updated.attempt_count == 1
            assert updated.last_http_status == status_code
            assert updated.last_error_code == expected_err
            assert updated.last_error_detail == response_text
            assert updated.lease_expires_at is None


@pytest.mark.asyncio
async def test_dispatch_terminal_404_tenant_unavailable_goes_to_dlq(migrated_db):
    session_factory, _, _ = migrated_db
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)
    event_id = "00000000-0000-0000-0000-000000000404"
    response_text = (
        '{"error_code":"subscriber_tenant_unavailable",'
        '"detail":"Webhook destination is not available"}'
    )
    event = OutboxEvent(
        id=event_id,
        event_type="lead_scoring.run.terminal",
        event_version="1.0.0",
        run_id="run-tenant-unavailable",
        payload={"status": "completed"},
        occurred_at=now,
        delivery_status=OutboxDeliveryStatus.DELIVERING.value,
        attempt_count=1,
    )

    mock_transport = httpx.MockTransport(lambda request: httpx.Response(404, text=response_text))
    async with httpx.AsyncClient(transport=mock_transport) as client:
        result = await dispatch_single_event(
            event=event,
            target_url="http://localhost:8080/api/v1/internal/lead-scoring/events",
            signing_key_id="test-key",
            signing_secret=TEST_SIGNING_SECRET,
            timeout_seconds=5.0,
            http_client=client,
            now=now,
        )

    assert result.classification == DeliveryClassification.DEAD_LETTER
    assert result.http_status == 404
    assert result.error_code == "receiver_client_error"

    async with session_factory() as session:
        session.add(event)
        await session.commit()
        recorded = await record_delivery_result(
            session=session,
            event=event,
            result=result,
            max_attempts=5,
            backoff_base_seconds=2.0,
            backoff_max_seconds=60.0,
            now=now,
        )
        await session.commit()

    assert recorded is True
    async with session_factory() as session:
        updated = await session.get(OutboxEvent, event_id)
        assert updated is not None
        assert updated.delivery_status == OutboxDeliveryStatus.DEAD_LETTER.value
        assert updated.attempt_count == 1
        assert updated.next_attempt_at is None
        assert updated.claim_token is None
        assert updated.lease_expires_at is None
        assert updated.last_http_status == 404
        assert updated.last_error_code == "receiver_client_error"
        assert updated.last_error_detail == response_text

        claimed = await claim_outbox_events(
            session=session,
            batch_size=1,
            lease_seconds=30.0,
            now=now + timedelta(minutes=1),
            max_attempts=5,
        )
        assert claimed == []


@pytest.mark.asyncio
async def test_dispatch_retryable_429_500_and_timeout(migrated_db):
    session_factory, _, _ = migrated_db
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    # 1. 429 Rate limited on attempt 1
    event_id = "00000000-0000-0000-0000-000000000429"
    event = OutboxEvent(
        id=event_id,
        event_type="lead_scoring.run.terminal",
        event_version="1.0.0",
        run_id="run-429",
        payload={"status": "completed"},
        occurred_at=now,
        delivery_status=OutboxDeliveryStatus.DELIVERING.value,
        claim_token="token-429",
        attempt_count=1,
    )
    mock_transport = httpx.MockTransport(
        lambda req: httpx.Response(429, text="Rate limit exceeded")
    )
    async with httpx.AsyncClient(transport=mock_transport) as client:
        result = await dispatch_single_event(
            event=event,
            target_url="http://localhost:8080/api/v1/internal/lead-scoring/events",
            signing_key_id="test-key",
            signing_secret=TEST_SIGNING_SECRET,
            timeout_seconds=5.0,
            http_client=client,
            now=now,
        )
    assert result.classification == DeliveryClassification.RETRYABLE
    assert result.http_status == 429

    async with session_factory() as session:
        session.add(event)
        await session.commit()
        await record_delivery_result(
            session=session,
            event=event,
            result=result,
            max_attempts=5,
            backoff_base_seconds=2.0,
            backoff_max_seconds=60.0,
            now=now,
            claim_token="token-429",
        )
        await session.commit()

    async with session_factory() as check_session:
        updated = await check_session.get(OutboxEvent, event_id)
        assert updated is not None
        assert updated.delivery_status == OutboxDeliveryStatus.PENDING.value
        assert updated.attempt_count == 1
        assert updated.next_attempt_at == now + timedelta(seconds=2.0)
        assert updated.last_error_code == "receiver_rate_limited"

    # 2. 500 Server error on attempt 2 -> backoff (4.0s)
    event_500_id = "00000000-0000-0000-0000-000000000500"
    event_500 = OutboxEvent(
        id=event_500_id,
        event_type="lead_scoring.run.terminal",
        event_version="1.0.0",
        run_id="run-500",
        payload={"status": "completed"},
        occurred_at=now,
        delivery_status=OutboxDeliveryStatus.DELIVERING.value,
        claim_token="token-500",
        attempt_count=2,
    )
    mock_transport_500 = httpx.MockTransport(
        lambda req: httpx.Response(500, text="Internal Server Error")
    )
    async with httpx.AsyncClient(transport=mock_transport_500) as client:
        result_500 = await dispatch_single_event(
            event=event_500,
            target_url="http://localhost:8080/api/v1/internal/lead-scoring/events",
            signing_key_id="test-key",
            signing_secret=TEST_SIGNING_SECRET,
            timeout_seconds=5.0,
            http_client=client,
            now=now,
        )
    assert result_500.classification == DeliveryClassification.RETRYABLE

    async with session_factory() as session:
        session.add(event_500)
        await session.commit()
        await record_delivery_result(
            session=session,
            event=event_500,
            result=result_500,
            max_attempts=5,
            backoff_base_seconds=2.0,
            backoff_max_seconds=60.0,
            now=now,
            claim_token="token-500",
        )
        await session.commit()

    async with session_factory() as check_session:
        updated_500 = await check_session.get(OutboxEvent, event_500_id)
        assert updated_500 is not None
        assert updated_500.delivery_status == OutboxDeliveryStatus.PENDING.value
        assert updated_500.attempt_count == 2
        assert updated_500.next_attempt_at == now + timedelta(seconds=4.0)

    # 3. Timeout on attempt 5 -> reaches max_attempts 5 -> DEAD_LETTER
    event_timeout_id = "00000000-0000-0000-0000-000000000504"
    event_timeout = OutboxEvent(
        id=event_timeout_id,
        event_type="lead_scoring.run.terminal",
        event_version="1.0.0",
        run_id="run-timeout",
        payload={"status": "completed"},
        occurred_at=now,
        delivery_status=OutboxDeliveryStatus.DELIVERING.value,
        claim_token="token-timeout",
        attempt_count=5,
    )
    timeout_result = DeliveryResult(
        classification=DeliveryClassification.RETRYABLE,
        error_code="delivery_timeout",
        detail="Webhook delivery timed out",
    )
    async with session_factory() as session:
        session.add(event_timeout)
        await session.commit()
        await record_delivery_result(
            session=session,
            event=event_timeout,
            result=timeout_result,
            max_attempts=5,
            backoff_base_seconds=2.0,
            backoff_max_seconds=60.0,
            now=now,
            claim_token="token-timeout",
        )
        await session.commit()

    async with session_factory() as check_session:
        updated_timeout = await check_session.get(OutboxEvent, event_timeout_id)
        assert updated_timeout is not None
        assert updated_timeout.delivery_status == OutboxDeliveryStatus.DEAD_LETTER.value
        assert updated_timeout.attempt_count == 5
        assert updated_timeout.last_error_code == "retries_exhausted"


@pytest.mark.asyncio
async def test_webhook_dispatcher_runtime_batch_processing(migrated_db):
    session_factory, _, _ = migrated_db
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    settings = Settings(
        webhook_subscriber_url="http://localhost:8080/api/v1/internal/lead-scoring/events",
        webhook_signing_secret=SecretStr(TEST_SIGNING_SECRET),
        webhook_signing_key_id="pilot-test-2026-01",
        webhook_max_attempts=3,
        webhook_batch_size=10,
    )

    received_requests = []

    def handle_request(request: httpx.Request):
        received_requests.append(
            {
                "url": str(request.url),
                "headers": dict(request.headers),
                "body": json.loads(request.content.decode("utf-8")),
            }
        )
        return httpx.Response(
            200,
            json={
                "status": "accepted",
                "event_id": request.headers.get(
                    "x-dotb-webhook-id", "00000000-0000-0000-0000-000000000000"
                ),
            },
        )

    mock_client = httpx.AsyncClient(transport=httpx.MockTransport(handle_request))

    # Insert 2 pending events
    async with session_factory() as session:
        session.add(
            OutboxEvent(
                id="00000000-0000-0000-0000-000000000101",
                event_type="lead_scoring.run.terminal",
                event_version="1.0.0",
                run_id="run-101",
                payload={
                    "event_id": "00000000-0000-0000-0000-000000000101",
                    "event_type": "lead_scoring.run.terminal",
                    "event_version": "1.0.0",
                    "tenant_id": "tenant_1",
                    "subject_ref": "lead_101",
                    "scoring_config_version": "1.0.0",
                    "run_id": "00000000-0000-0000-0000-000000000101",
                    "run_created_at": "2026-08-28T12:00:00Z",
                    "occurred_at": "2026-08-28T12:00:02Z",
                    "data": {"status": "completed", "rule_score": 85},
                },
                occurred_at=now,
                delivery_status=OutboxDeliveryStatus.PENDING.value,
            )
        )
        session.add(
            OutboxEvent(
                id="00000000-0000-0000-0000-000000000102",
                event_type="lead_scoring.run.reviewed",
                event_version="1.0.0",
                run_id="run-102",
                payload={
                    "event_id": "00000000-0000-0000-0000-000000000102",
                    "event_type": "lead_scoring.run.reviewed",
                    "event_version": "1.0.0",
                    "tenant_id": "tenant_1",
                    "subject_ref": "lead_102",
                    "scoring_config_version": "1.0.0",
                    "run_id": "00000000-0000-0000-0000-000000000102",
                    "run_created_at": "2026-08-28T12:00:00Z",
                    "occurred_at": "2026-08-28T12:05:00Z",
                    "data": {
                        "status": "completed",
                        "decision": "approve",
                        "calculated_score": 80,
                        "effective_score": 80,
                        "version": 2,
                        "reviewed_by": "admin_1",
                    },
                },
                occurred_at=now + timedelta(seconds=1),
                delivery_status=OutboxDeliveryStatus.PENDING.value,
            )
        )
        await session.commit()

    dispatcher = WebhookDispatcher(
        session_factory=session_factory,
        settings=settings,
        http_client=mock_client,
        clock=lambda: now,
    )

    processed_count = await dispatcher.dispatch_batch()
    assert processed_count == 2
    assert len(received_requests) == 2

    # Verify headers and HMAC signature on received request
    req1 = received_requests[0]
    assert req1["headers"]["x-dotb-webhook-id"] == "00000000-0000-0000-0000-000000000101"
    assert req1["headers"]["x-dotb-webhook-key-id"] == "pilot-test-2026-01"
    assert "x-dotb-webhook-signature" in req1["headers"]
    assert req1["headers"]["x-dotb-webhook-signature"].startswith("v1=")

    # Verify both rows in DB transitioned to delivered
    async with session_factory() as session:
        events = (await session.execute(select(OutboxEvent))).scalars().all()
        assert len(events) == 2
        for ev in events:
            assert ev.delivery_status == OutboxDeliveryStatus.DELIVERED.value
            assert ev.delivered_at is not None

    await dispatcher.stop()
    await mock_client.aclose()


@pytest.mark.asyncio
async def test_fastapi_lifespan_starts_and_stops_dispatcher(migrated_db):
    session_factory, engine, db_url = migrated_db
    settings = Settings(
        webhook_subscriber_url="http://localhost:8080/api/v1/internal/lead-scoring/events",
        webhook_signing_secret=SecretStr(TEST_SIGNING_SECRET),
        database_url=db_url,
    )

    mock_client = httpx.AsyncClient(
        transport=httpx.MockTransport(
            lambda req: httpx.Response(
                200, json={"status": "accepted", "event_id": "00000000-0000-0000-0000-000000000001"}
            )
        )
    )

    app = create_app(settings=settings, webhook_http_client=mock_client)

    # Trigger lifespan
    async with app.router.lifespan_context(app):
        # Check app state has dispatcher
        dispatcher = getattr(app.state, "webhook_dispatcher", None)
        assert dispatcher is not None
        assert isinstance(dispatcher, WebhookDispatcher)

        async with httpx.AsyncClient(
            transport=httpx.ASGITransport(app=app),
            base_url="http://testserver",
        ) as client:
            # Health endpoint should be healthy
            resp = await client.get("/health")
            assert resp.status_code == 200

    await mock_client.aclose()


@pytest.mark.asyncio
async def test_e2e_scoring_run_execution_outbox_to_webhook_delivery(migrated_db):
    session_factory, engine, db_url = migrated_db
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    from lead_scoring.config import ServiceCallerConfig
    from lead_scoring.execution import execute_scoring_run
    from lead_scoring.scoring import ScoringConfig

    scoring_config = ScoringConfig(
        version="1.0.0",
        base_score=85,
        rule_weight=1.0,
        ai_weight=0.0,
        confidence=0.9,
        confidence_threshold=0.8,
        hitl_policy_version="1.0.0",
        rules=(),
    )

    settings = Settings(
        webhook_subscriber_url="http://localhost:8080/api/v1/internal/lead-scoring/events",
        webhook_signing_secret=SecretStr(TEST_SIGNING_SECRET),
        webhook_signing_key_id="pilot-test-2026-01",
        database_url=db_url,
        scoring_configs=[scoring_config],
        service_callers=[
            ServiceCallerConfig(
                caller_id="test_caller",
                token=SecretStr("valid_token"),
                allowed_apps=["omni"],
                allowed_features=["scoring"],
                allowed_tenants=["tenant_1"],
            )
        ],
    )

    run_id = "00000000-0000-0000-0000-000000000999"

    # 1. Insert a pending scoring run
    async with session_factory() as session:
        session.add(
            ScoringRun(
                id=run_id,
                tenant_id="tenant_1",
                subject_ref="lead_999",
                scoring_config_version="1.0.0",
                idempotency_key="idem-999",
                payload_hash="hash-999",
                status=ScoringRunStatus.PENDING.value,
                snapshot_data={
                    "version": "1.0.0",
                    "captured_at": "2026-08-28T12:00:00Z",
                    "lead_profile": {
                        "ems_lead_id": "lead_999",
                        "name": "CEO",
                    },
                },
                effective_config=scoring_config.model_dump(mode="json"),
                confidence_threshold=scoring_config.confidence_threshold,
                created_at=now,
                updated_at=now,
            )
        )
        await session.commit()

    # 2. Execute scoring run -> writes terminal outbox event atomically
    executed = await execute_scoring_run(
        session_factory=session_factory,
        run_id=run_id,
        tenant_vkey="test-vkey",
        lease_seconds=30.0,
        clock=lambda: now,
    )
    assert executed is True

    # Verify outbox event written
    async with session_factory() as session:
        events = (
            (await session.execute(select(OutboxEvent).where(OutboxEvent.run_id == run_id)))
            .scalars()
            .all()
        )
        assert len(events) == 1
        outbox_entry = events[0]
        assert outbox_entry.event_type == "lead_scoring.run.terminal"
        assert outbox_entry.delivery_status == OutboxDeliveryStatus.PENDING.value

    # 3. Setup mock Omni subscriber that verifies HMAC signature and timestamp
    received_requests = []

    def mock_omni_receiver(request: httpx.Request):
        event_id = request.headers["x-dotb-webhook-id"]
        timestamp = request.headers["x-dotb-webhook-timestamp"]
        signature = request.headers["x-dotb-webhook-signature"]
        body = request.content

        expected_sig = sign_webhook(
            secret=TEST_SIGNING_SECRET,
            event_id=event_id,
            timestamp=timestamp,
            body=body,
        )
        assert signature == expected_sig

        parsed_body = json.loads(body.decode("utf-8"))
        received_requests.append(parsed_body)

        return httpx.Response(
            200,
            json={"status": "accepted", "event_id": event_id},
        )

    mock_client = httpx.AsyncClient(transport=httpx.MockTransport(mock_omni_receiver))

    dispatcher = WebhookDispatcher(
        session_factory=session_factory,
        settings=settings,
        http_client=mock_client,
        clock=lambda: now,
    )

    # 4. Dispatch batch
    dispatched_count = await dispatcher.dispatch_batch()
    assert dispatched_count == 1
    assert len(received_requests) == 1
    assert received_requests[0]["tenant_id"] == "tenant_1"
    assert received_requests[0]["subject_ref"] == "lead_999"
    assert received_requests[0]["data"]["status"] == "completed"
    assert received_requests[0]["data"]["rule_score"] == 85

    # 5. Verify outbox marked delivered
    async with session_factory() as session:
        updated_event = (
            await session.execute(select(OutboxEvent).where(OutboxEvent.run_id == run_id))
        ).scalar_one()
        assert updated_event.delivery_status == OutboxDeliveryStatus.DELIVERED.value
        assert updated_event.delivered_at == now
        assert updated_event.last_http_status == 200

    await dispatcher.stop()
    await mock_client.aclose()


@pytest.mark.asyncio
async def test_scoring_request_cannot_contain_callback_url(migrated_db):
    session_factory, engine, db_url = migrated_db

    from lead_scoring.config import ServiceCallerConfig
    from lead_scoring.scoring import ScoringConfig

    scoring_config = ScoringConfig(
        version="1.0.0",
        base_score=0,
        rule_weight=1.0,
        ai_weight=0.0,
        confidence=0.9,
        confidence_threshold=0.8,
        hitl_policy_version="1.0.0",
        rules=(),
    )

    settings = Settings(
        database_url=db_url,
        tenant_allowlist=["tenant_1"],
        scoring_configs=[scoring_config],
        service_callers=[
            ServiceCallerConfig(
                caller_id="test_caller",
                token=SecretStr("valid_token"),
                allowed_apps=["omni"],
                allowed_features=["scoring"],
                allowed_tenants=["tenant_1"],
            )
        ],
    )

    app = create_app(settings=settings)

    async with app.router.lifespan_context(app):
        async with httpx.AsyncClient(
            transport=httpx.ASGITransport(app=app),
            base_url="http://testserver",
        ) as client:
            headers = {
                "Authorization": "Bearer valid_token",
                "X-Tenant-Id": "tenant_1",
                "Idempotency-Key": "test-idem-callback",
                "X-DotB-App": "omni",
                "X-DotB-Feature": "scoring",
                "X-Tenant-VKey": "dummy-vkey",
            }
            # Attempt to inject callback_url in create request
            payload_with_callback = {
                "tenant_id": "tenant_1",
                "subject_ref": "lead_123",
                "scoring_config_version": "1.0.0",
                "snapshot": {
                    "version": "1.0.0",
                    "captured_at": "2026-08-28T12:00:00Z",
                    "lead_profile": {
                        "ems_lead_id": "lead_123",
                        "custom_fields": {"course_interest": "DotB Course"},
                    },
                },
                "callback_url": "https://attacker.com/webhook",
            }
            resp = await client.post(
                "/v1/scoring-runs", json=payload_with_callback, headers=headers
            )
            assert resp.status_code == 202
            run_id = resp.json()["run_id"]

            # Verify that in database, no callback_url is stored on scoring_runs
            async with session_factory() as session:
                run_row = await session.get(ScoringRun, run_id)
                assert run_row is not None
                assert "callback_url" not in run_row.snapshot_data
                assert "callback_url" not in (run_row.effective_config or {})


@pytest.mark.asyncio
async def test_stale_claim_cannot_overwrite_newer_delivery_state(migrated_db):
    session_factory, _, _ = migrated_db
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    event_id = "00000000-0000-0000-0000-000000000777"

    # Worker 1 claims the event with token_1
    worker_1_token = "claim-token-worker-1"

    async with session_factory() as session:
        event = OutboxEvent(
            id=event_id,
            event_type="lead_scoring.run.terminal",
            event_version="1.0.0",
            run_id="run-777",
            payload={"status": "completed"},
            occurred_at=now,
            delivery_status=OutboxDeliveryStatus.DELIVERING.value,
            claim_token=worker_1_token,
            lease_expires_at=now + timedelta(seconds=10),
            attempt_count=0,
        )
        session.add(event)
        await session.commit()

    # Time passes, Worker 1 stalls. Worker 2 reclaims and marks delivered
    async with session_factory() as session:
        # Worker 2 successfully delivers with token_2
        await session.execute(
            update(OutboxEvent)
            .where(OutboxEvent.id == event_id)
            .values(
                delivery_status=OutboxDeliveryStatus.DELIVERED.value,
                delivered_at=now + timedelta(seconds=15),
                claim_token=None,
                lease_expires_at=None,
                last_http_status=200,
            )
        )
        await session.commit()

    # Worker 1 wakes up and attempts to record a failure using stale token_1
    stale_result = DeliveryResult(
        classification=DeliveryClassification.DEAD_LETTER,
        error_code="receiver_bad_request",
        http_status=400,
        detail="Stale worker error",
    )

    async with session_factory() as session:
        recorded = await record_delivery_result(
            session=session,
            event=event,
            result=stale_result,
            max_attempts=5,
            backoff_base_seconds=2.0,
            backoff_max_seconds=60.0,
            now=now + timedelta(seconds=20),
            claim_token=worker_1_token,
        )
        await session.commit()
        # Should return False due to claim fence mismatch
        assert recorded is False

    # Verify event remains DELIVERED and was NOT overwritten by stale worker
    async with session_factory() as check_session:
        updated = await check_session.get(OutboxEvent, event_id)
        assert updated is not None
        assert updated.delivery_status == OutboxDeliveryStatus.DELIVERED.value
        assert updated.last_http_status == 200
        assert updated.last_error_code is None


@pytest.mark.asyncio
async def test_webhook_max_attempts_setting_controls_retry_limit(migrated_db):
    session_factory, _, _ = migrated_db
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    event_id = "00000000-0000-0000-0000-000000000888"
    token = "claim-token-max-attempts"

    async with session_factory() as session:
        event = OutboxEvent(
            id=event_id,
            event_type="lead_scoring.run.terminal",
            event_version="1.0.0",
            run_id="run-888",
            payload={"status": "completed"},
            occurred_at=now,
            delivery_status=OutboxDeliveryStatus.DELIVERING.value,
            claim_token=token,
            attempt_count=2,
        )
        session.add(event)
        await session.commit()

    # Setting max_attempts = 2: attempt_count 2 reaches max_attempts -> DEAD_LETTER
    retry_result = DeliveryResult(
        classification=DeliveryClassification.RETRYABLE,
        error_code="receiver_rate_limited",
        http_status=429,
        detail="Rate limited",
    )

    async with session_factory() as session:
        recorded = await record_delivery_result(
            session=session,
            event=event,
            result=retry_result,
            max_attempts=2,
            backoff_base_seconds=2.0,
            backoff_max_seconds=60.0,
            now=now,
            claim_token=token,
        )
        await session.commit()
        assert recorded is True

    async with session_factory() as check_session:
        updated = await check_session.get(OutboxEvent, event_id)
        assert updated is not None
        assert updated.delivery_status == OutboxDeliveryStatus.DEAD_LETTER.value
        assert updated.attempt_count == 2
        assert updated.last_error_code == "retries_exhausted"


@pytest.mark.asyncio
async def test_dispatch_non_200_2xx_classified_as_dlq():
    now = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    for status_code in [201, 202, 204]:
        event = OutboxEvent(
            id=f"00000000-0000-0000-0000-000000000{status_code}",
            event_type="lead_scoring.run.terminal",
            event_version="1.0.0",
            run_id=f"run-{status_code}",
            payload={"status": "completed"},
            occurred_at=now,
            delivery_status=OutboxDeliveryStatus.DELIVERING.value,
        )

        mock_transport = httpx.MockTransport(
            lambda request, sc=status_code: httpx.Response(sc, text="Non-200 2xx response")
        )
        async with httpx.AsyncClient(transport=mock_transport) as client:
            result = await dispatch_single_event(
                event=event,
                target_url="http://localhost:8080/api/v1/internal/lead-scoring/events",
                signing_key_id="test-key",
                signing_secret=TEST_SIGNING_SECRET,
                timeout_seconds=5.0,
                http_client=client,
                now=now,
            )

        assert result.classification == DeliveryClassification.DEAD_LETTER
        assert result.http_status == status_code
        assert result.error_code == "invalid_receiver_response"


@pytest.mark.asyncio
async def test_crash_reclaims_are_bounded_and_eventually_dlq(migrated_db):
    session_factory, _, _ = migrated_db
    t0 = datetime(2026, 8, 28, 12, 20, tzinfo=UTC)
    event_id = "00000000-0000-0000-0000-000000001201"
    expected_max_attempts = 3

    event = OutboxEvent(
        id=event_id,
        event_type="lead_scoring.run.terminal",
        event_version="1.0.0",
        run_id="run-crash-test",
        payload={
            "event_id": event_id,
            "event_type": "lead_scoring.run.terminal",
            "event_version": "1.0.0",
            "tenant_id": "tenant-1",
            "subject_ref": "lead-1",
            "scoring_config_version": "1.0.0",
            "run_id": event_id,
            "run_created_at": t0.isoformat().replace("+00:00", "Z"),
            "occurred_at": (t0 + timedelta(seconds=1)).isoformat().replace("+00:00", "Z"),
            "data": {"status": "completed", "rule_score": 80},
        },
        occurred_at=t0,
    )

    async with session_factory() as session:
        session.add(event)
        await session.commit()

    # Simulate 5 claim/crash cycles (process death before result recording)
    for index in range(5):
        async with session_factory() as session:
            await claim_outbox_events(
                session,
                batch_size=1,
                lease_seconds=1,
                now=t0 + timedelta(seconds=index * 2),
                max_attempts=expected_max_attempts,
            )
            await session.commit()

    async with session_factory() as session:
        stored = await session.get(OutboxEvent, event_id)
        assert stored is not None
        assert stored.delivery_status == OutboxDeliveryStatus.DEAD_LETTER.value
        assert stored.attempt_count >= expected_max_attempts
        assert stored.last_error_code == "retries_exhausted"
        assert stored.next_attempt_at is None
        assert stored.claim_token is None
        assert stored.lease_expires_at is None


@pytest.mark.asyncio
async def test_retry_followed_by_success_clears_next_attempt_at_and_errors(migrated_db):
    session_factory, _, _ = migrated_db
    t0 = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)
    event_id = "00000000-0000-0000-0000-000000000654"

    # Step 1: Initial event in PENDING state
    event = OutboxEvent(
        id=event_id,
        event_type="lead_scoring.run.terminal",
        event_version="1.0.0",
        run_id="run-retry-then-success",
        payload={"status": "completed"},
        occurred_at=t0,
        delivery_status=OutboxDeliveryStatus.PENDING.value,
    )
    async with session_factory() as session:
        session.add(event)
        await session.commit()

    # Step 2: Claim event (Attempt 1)
    async with session_factory() as session:
        claimed = await claim_outbox_events(
            session=session,
            batch_size=1,
            lease_seconds=30.0,
            now=t0,
            max_attempts=5,
        )
        await session.commit()
    assert len(claimed) == 1
    event_claimed = claimed[0]
    assert event_claimed.attempt_count == 1
    token_1 = event_claimed.claim_token

    # Step 3: Attempt 1 fails with 429 Retryable
    retry_result = DeliveryResult(
        classification=DeliveryClassification.RETRYABLE,
        http_status=429,
        error_code="receiver_rate_limited",
        detail="Rate limited",
    )
    async with session_factory() as session:
        recorded = await record_delivery_result(
            session=session,
            event=event_claimed,
            result=retry_result,
            max_attempts=5,
            backoff_base_seconds=2.0,
            backoff_max_seconds=60.0,
            now=t0,
            claim_token=token_1,
        )
        await session.commit()
    assert recorded is True

    # Check that next_attempt_at is set to t0 + 2s
    async with session_factory() as session:
        ev_after_retry = await session.get(OutboxEvent, event_id)
        assert ev_after_retry is not None
        assert ev_after_retry.delivery_status == OutboxDeliveryStatus.PENDING.value
        assert ev_after_retry.next_attempt_at == t0 + timedelta(seconds=2.0)
        assert ev_after_retry.last_error_code == "receiver_rate_limited"

    # Step 4: Claim event at next_attempt_at (Attempt 2)
    t1 = t0 + timedelta(seconds=2.0)
    async with session_factory() as session:
        claimed_2 = await claim_outbox_events(
            session=session,
            batch_size=1,
            lease_seconds=30.0,
            now=t1,
            max_attempts=5,
        )
        await session.commit()
    assert len(claimed_2) == 1
    event_claimed_2 = claimed_2[0]
    assert event_claimed_2.attempt_count == 2
    token_2 = event_claimed_2.claim_token

    # Step 5: Attempt 2 succeeds with HTTP 200 DELIVERED
    success_result = DeliveryResult(
        classification=DeliveryClassification.DELIVERED,
        http_status=200,
        ingest_status="accepted",
    )
    async with session_factory() as session:
        recorded_2 = await record_delivery_result(
            session=session,
            event=event_claimed_2,
            result=success_result,
            max_attempts=5,
            backoff_base_seconds=2.0,
            backoff_max_seconds=60.0,
            now=t1,
            claim_token=token_2,
        )
        await session.commit()
    assert recorded_2 is True

    # Step 6: Verify observable state: DELIVERED, next_attempt_at is None, error metadata cleared
    async with session_factory() as session:
        ev_final = await session.get(OutboxEvent, event_id)
        assert ev_final is not None
        assert ev_final.delivery_status == OutboxDeliveryStatus.DELIVERED.value
        assert ev_final.next_attempt_at is None
        assert ev_final.lease_expires_at is None
        assert ev_final.claim_token is None
        assert ev_final.last_error_code is None
        assert ev_final.last_error_detail is None
        assert ev_final.delivered_at == t1
        assert ev_final.last_http_status == 200


@pytest.mark.asyncio
async def test_slow_first_item_does_not_cause_lease_exhaustion_for_subsequent_items(migrated_db):
    session_factory, _, _ = migrated_db
    t0 = datetime(2026, 8, 28, 12, 0, 0, tzinfo=UTC)

    event_1_id = "00000000-0000-0000-0000-000000000901"
    event_2_id = "00000000-0000-0000-0000-000000000902"

    async with session_factory() as session:
        session.add(
            OutboxEvent(
                id=event_1_id,
                event_type="lead_scoring.run.terminal",
                event_version="1.0.0",
                run_id="run-slow-1",
                payload={"status": "completed"},
                occurred_at=t0,
                delivery_status=OutboxDeliveryStatus.PENDING.value,
            )
        )
        session.add(
            OutboxEvent(
                id=event_2_id,
                event_type="lead_scoring.run.terminal",
                event_version="1.0.0",
                run_id="run-slow-2",
                payload={"status": "completed"},
                occurred_at=t0 + timedelta(seconds=1),
                delivery_status=OutboxDeliveryStatus.PENDING.value,
            )
        )
        await session.commit()

    simulated_time = t0

    def current_time() -> datetime:
        return simulated_time

    settings = Settings(
        webhook_subscriber_url="http://localhost:8080/api/v1/internal/lead-scoring/events",
        webhook_signing_secret=SecretStr(TEST_SIGNING_SECRET),
        webhook_signing_key_id="test-key",
        webhook_lease_seconds=5.0,
        webhook_max_attempts=2,
        webhook_batch_size=2,
    )

    # Mock receiver where first item takes 10s (exceeding lease 5s of first item)
    def handle_request(request: httpx.Request) -> httpx.Response:
        nonlocal simulated_time
        body = json.loads(request.content.decode("utf-8"))
        if body.get("event_id") == event_1_id:
            # Advance time by 10s during dispatch of item 1
            simulated_time += timedelta(seconds=10.0)
        return httpx.Response(
            200,
            json={"status": "accepted", "event_id": request.headers.get("x-dotb-webhook-id", "")},
        )

    mock_client = httpx.AsyncClient(transport=httpx.MockTransport(handle_request))

    dispatcher = WebhookDispatcher(
        session_factory=session_factory,
        settings=settings,
        http_client=mock_client,
        clock=current_time,
    )

    processed = await dispatcher.dispatch_batch()
    assert processed == 2

    # Verify both events delivered successfully without premature exhaustion
    async with session_factory() as session:
        ev1 = await session.get(OutboxEvent, event_1_id)
        assert ev1 is not None
        assert ev1.delivery_status == OutboxDeliveryStatus.DELIVERED.value
        assert ev1.attempt_count == 1

        ev2 = await session.get(OutboxEvent, event_2_id)
        assert ev2 is not None
        assert ev2.delivery_status == OutboxDeliveryStatus.DELIVERED.value
        # Item 2 was claimed at t0 + 10s, attempt_count was 1 (not reclaimed/exhausted)
        assert ev2.attempt_count == 1
        assert ev2.next_attempt_at is None

    await dispatcher.stop()
    await mock_client.aclose()