from __future__ import annotations

from datetime import UTC, datetime, timedelta

import pytest
from backup_tool.db.models import (
    Backup,
    Execution,
    Job,
    NotificationDelivery,
    NotificationEvent,
    NotificationSubscription,
    Repository,
    Schedule,
    Source,
)
from backup_tool.execution import record_event
from backup_tool.notifications.events import EVENT_CATALOG
from backup_tool.scheduler import deliver_due
from backup_tool.worker import Worker
from sqlalchemy import select


@pytest.mark.asyncio
async def test_execution_catalog_events_are_produced_and_delivered(app_client) -> None:
    """Exercise the execution producer, not emit_event(), for each live execution type."""
    client, _ = app_client
    setup = await client.post(
        "/api/v2/setup",
        json={"username": "admin", "password": "correct horse battery staple"},
    )
    assert setup.status_code == 201
    app = client._transport.app
    expected = {
        "execution.queued",
        "execution.started",
        "execution.committed",
        "execution.failed",
        "execution.cancelled",
        "execution.retry_queued",
        "execution.worker_recovered",
    }
    assert expected <= set(EVENT_CATALOG)
    deferred_prefixes = ["source.", "gc.", "reconciliation."]
    assert not any(item.startswith(tuple(deferred_prefixes)) for item in EVENT_CATALOG)
    async with app.state.sessions() as db:
        repository = Repository(
            name="events-repository",
            root="/events-repository",
            format_version=1,
            compression="none",
            encryption="none",
        )
        source = Source(
            name="events-source",
            kind="local",
            public_config={"root": "/events-source"},
            secret_refs=[],
        )
        db.add_all([repository, source])
        await db.flush()
        job = Job(
            name="events-job",
            source_id=source.id,
            repository_id=repository.id,
            exclusions=[],
            retention={},
            requested_mode="full",
        )
        subscription = NotificationSubscription(
            channel="email",
            event_filters=["execution.*"],
            destination_config={"recipients": ["operator@example.test"]},
            rate_limit_per_minute=60,
            rate_tokens=60.0,
        )
        db.add_all([job, subscription])
        await db.flush()
        cases = (
            ("queued", 1, None, "execution.queued"),
            ("preparing", 1, None, "execution.started"),
            ("committed", 1, None, "execution.committed"),
            ("failed", 1, "transient_io", "execution.failed"),
            ("cancelled", 1, "cancellation_requested", "execution.cancelled"),
            ("queued", 2, None, "execution.retry_queued"),
            ("queued", 1, "worker_lost", "execution.worker_recovered"),
        )
        for state, attempt, reason, _event_type in cases:
            execution = Execution(
                job_id=job.id,
                trigger="manual",
                state=state,
                attempt=attempt,
                reason_code=reason,
                progress={},
            )
            db.add(execution)
            await db.flush()
            await record_event(db, execution)
            if state in {"queued", "preparing"}:
                execution.state = "failed"
                execution.reason_code = "test_cleanup"
            await db.flush()
        await db.commit()
        event_statement = select(NotificationEvent.type).where(NotificationEvent.type.in_(expected))
        event_types = set((await db.scalars(event_statement)).all())
        deliveries = await db.scalar(
            select(NotificationDelivery.id)
            .join(NotificationEvent, NotificationDelivery.event_id == NotificationEvent.id)
            .where(NotificationEvent.type.in_(expected))
            .limit(1)
        )
    assert event_types == expected
    assert deliveries is not None


@pytest.mark.asyncio
async def test_schedule_catalog_events_are_produced_and_delivered(app_client) -> None:
    client, _ = app_client
    setup = await client.post(
        "/api/v2/setup",
        json={"username": "admin", "password": "correct horse battery staple"},
    )
    assert setup.status_code == 201
    csrf = client.cookies["backup_tool_csrf"]
    app = client._transport.app
    expected = {
        "schedule.created",
        "schedule.updated",
        "schedule.deleted",
        "schedule.enabled",
        "schedule.disabled",
        "schedule.occurrence_enqueued",
        "schedule.occurrence_misfired",
        "schedule.occurrence_blocked",
    }
    assert expected <= set(EVENT_CATALOG)
    async with app.state.sessions() as db:
        repository = Repository(
            name="schedule-repository",
            root="/schedule-repository",
            format_version=1,
            compression="none",
            encryption="none",
        )
        source = Source(
            name="schedule-source",
            kind="local",
            public_config={"root": "/schedule-source"},
            secret_refs=[],
        )
        db.add_all([repository, source])
        await db.flush()
        jobs = [
            Job(
                name=f"schedule-job-{number}",
                source_id=source.id,
                repository_id=repository.id,
                exclusions=[],
                retention={},
                requested_mode="full",
                enabled=number not in {2, 3},
            )
            for number in range(1, 5)
        ]
        subscription = NotificationSubscription(
            channel="email",
            event_filters=["schedule.*"],
            destination_config={"recipients": ["operator@example.test"]},
            rate_limit_per_minute=60,
            rate_tokens=60.0,
        )
        db.add_all([*jobs, subscription])
        await db.commit()
        job_ids = [job.id for job in jobs]

    created = await client.post(
        f"/api/v2/jobs/{job_ids[0]}/schedule",
        json={"cron": "0 0 * * *", "timezone": "UTC", "enabled": True},
        headers={"X-CSRF-Token": csrf},
    )
    assert created.status_code == 201
    disabled = await client.patch(
        f"/api/v2/jobs/{job_ids[0]}/schedule",
        json={"cron": "0 0 * * *", "timezone": "UTC", "enabled": False},
        headers={"X-CSRF-Token": csrf},
    )
    assert disabled.status_code == 200
    enabled = await client.patch(
        f"/api/v2/jobs/{job_ids[0]}/schedule",
        json={"cron": "1 0 * * *", "timezone": "UTC", "enabled": True},
        headers={"X-CSRF-Token": csrf},
    )
    assert enabled.status_code == 200
    updated = await client.patch(
        f"/api/v2/jobs/{job_ids[0]}/schedule",
        json={"cron": "2 0 * * *", "timezone": "UTC", "enabled": True},
        headers={"X-CSRF-Token": csrf},
    )
    assert updated.status_code == 200
    assert (
        await client.delete(f"/api/v2/jobs/{job_ids[0]}/schedule", headers={"X-CSRF-Token": csrf})
    ).status_code == 204

    async with app.state.sessions() as db:
        now = datetime.now(UTC)
        db.add_all(
            [
                Schedule(
                    job_id=job_ids[1],
                    cron="* * * * *",
                    timezone="UTC",
                    misfire_grace_seconds=0,
                    enabled=True,
                    next_nominal_at=now - timedelta(hours=1),
                ),
                Schedule(
                    job_id=job_ids[2],
                    cron="* * * * *",
                    timezone="UTC",
                    misfire_grace_seconds=60,
                    enabled=True,
                    next_nominal_at=now,
                ),
                Schedule(
                    job_id=job_ids[3],
                    cron="* * * * *",
                    timezone="UTC",
                    misfire_grace_seconds=60,
                    enabled=True,
                    next_nominal_at=now,
                ),
            ]
        )
        await db.commit()
        assert await deliver_due(db, now=now) == 1
        statement = select(NotificationEvent.type).where(NotificationEvent.type.in_(expected))
        event_types = set((await db.scalars(statement)).all())
        delivery = await db.scalar(
            select(NotificationDelivery.id)
            .join(NotificationEvent, NotificationDelivery.event_id == NotificationEvent.id)
            .where(NotificationEvent.type.in_(expected))
            .limit(1)
        )
    assert event_types == expected
    assert delivery is not None


@pytest.mark.asyncio
async def test_backup_restore_and_retention_events_are_produced_and_delivered(
    app_client,
) -> None:
    client, settings = app_client
    source_root = settings.local_source_roots[0] / "notification-project"
    source_root.mkdir()
    (source_root / "data.txt").write_text("notification data\n", encoding="utf-8")
    setup = await client.post(
        "/api/v2/setup",
        json={"username": "admin", "password": "correct horse battery staple"},
    )
    assert setup.status_code == 201
    headers = {"X-CSRF-Token": client.cookies["backup_tool_csrf"]}
    subscription = await client.post(
        "/api/v2/notifications/subscriptions",
        json={
            "channel": "email",
            "event_filters": ["backup.*", "restore.*", "retention.*"],
            "destination": {"recipients": ["operator@example.test"]},
        },
        headers=headers,
    )
    assert subscription.status_code == 201
    repository = await client.post(
        "/api/v2/repositories",
        json={"name": "notification-repo", "relative_path": "notification-repo"},
        headers=headers,
    )
    source = await client.post(
        "/api/v2/sources",
        json={
            "name": "notification-source",
            "kind": "local",
            "public_config": {"root": str(source_root)},
        },
        headers=headers,
    )
    assert repository.status_code == source.status_code == 201
    job = await client.post(
        "/api/v2/jobs",
        json={
            "name": "notification-job",
            "source_id": source.json()["id"],
            "repository_id": repository.json()["id"],
            "requested_mode": "full",
            "exclusions": [],
            "retention": {"keep_last": 1},
            "allow_empty": False,
        },
        headers=headers,
    )
    assert job.status_code == 201
    execution = await client.post(f"/api/v2/jobs/{job.json()['id']}/executions", headers=headers)
    assert execution.status_code == 202
    worker = Worker(settings, owner="notification-backup-worker")
    try:
        assert await worker.run_once()
    finally:
        await worker.engine.dispose()
    app = client._transport.app
    async with app.state.sessions() as db:
        backup = await db.scalar(
            select(Backup).where(Backup.execution_id == execution.json()["id"])
        )
        assert backup is not None
        # A synthetic older catalog entry is valid business state; tombstoning is
        # performed only through the real retention producer below.
        older_execution = Execution(
            job_id=job.json()["id"],
            trigger="manual",
            state="committed",
            progress={},
        )
        db.add(older_execution)
        await db.flush()
        older = Backup(
            execution_id=older_execution.id,
            manifest_id="00000000-0000-7000-8000-000000000001",
            manifest_digest="0" * 64,
            logical_bytes=0,
            stored_bytes=0,
            integrity="verified",
            created_at=datetime.now(UTC) - timedelta(days=1),
        )
        db.add(older)
        await db.commit()

    retention_worker = Worker(settings, owner="notification-retention-worker")
    try:
        # Retention/GC is executed by worker maintenance, not a direct helper call.
        assert await retention_worker.run_once()
    finally:
        await retention_worker.engine.dispose()

    restore = await client.post(
        f"/api/v2/backups/{backup.id}/restores",
        json={
            "destination": str(settings.restore_roots[0] / "notification-restore"),
            "selection": [],
            "dry_run": True,
            "overwrite_policy": "fail",
        },
        headers=headers,
    )
    assert restore.status_code == 202
    restore_worker = Worker(settings, owner="notification-restore-worker")
    try:
        assert await restore_worker.run_once()
    finally:
        await restore_worker.engine.dispose()

    expected = {
        "backup.committed",
        "backup.verification_succeeded",
        "restore.queued",
        "restore.committed",
        "retention.tombstoned",
    }
    async with app.state.sessions() as db:
        types = set(
            (
                await db.scalars(
                    select(NotificationEvent.type).where(NotificationEvent.type.in_(expected))
                )
            ).all()
        )
        deliveries = list(
            (
                await db.scalars(
                    select(NotificationDelivery.id)
                    .join(
                        NotificationEvent,
                        NotificationDelivery.event_id == NotificationEvent.id,
                    )
                    .where(NotificationEvent.type.in_(expected))
                )
            ).all()
        )
    assert types == expected
    assert len(deliveries) >= len(expected)