"""Worker entrypoint — APScheduler running the reminder planner and the message sender
(tasks 5.1-5.3, design D6).

There is no broker. The planner writes `message_job` rows; the sender polls for due ones.
Both are idempotent, so a worker that was down for any length of time self-heals on its
next tick rather than needing missed-run recovery.
"""

import asyncio
import logging
import signal
from datetime import UTC, datetime, timedelta
from typing import Any

from apscheduler.schedulers.asyncio import AsyncIOScheduler
from sqlalchemy import func, select

from app.config import get_settings
from app.db import get_session_factory
from app.models import MessageJob
from app.models.enums import Channel, MessageJobStatus
from app.services import heartbeat, messaging, planner, providers, rate_limit, templating

logger = logging.getLogger(__name__)
settings = get_settings()

#: A job stuck in `sending` means the worker died mid-send. Nothing would ever pick it up
#: again, so it is returned to the queue after this long.
STUCK_AFTER = timedelta(minutes=15)


async def _beat_failure(name: str, exc: BaseException) -> None:
    """Record a failed tick on a fresh session — the tick's own was rolled back."""
    try:
        async with get_session_factory()() as session:
            await heartbeat.beat(session, name, error=f"{type(exc).__name__}: {exc}"[:500])
            await session.commit()
    except Exception:
        logger.exception("could not record heartbeat failure for %s", name)


async def plan_reminders() -> None:
    """Hourly: compute due waves and insert jobs idempotently."""
    async with get_session_factory()() as session:
        try:
            plans = await planner.plan_all(session)
            await heartbeat.beat(session, heartbeat.PLANNER)
            await session.commit()
        except Exception as exc:
            await session.rollback()
            logger.exception("planner tick failed")
            # The heartbeat records the failure in its own transaction: a planner that is
            # running but erroring must not look identical to one that is not running.
            await _beat_failure(heartbeat.PLANNER, exc)
            return

    created = sum(p.created for p in plans)
    skipped = sum(1 for p in plans if p.skipped_past)
    if created or skipped:
        logger.info("planner: %d job(s) queued, %d past wave(s) skipped", created, skipped)


async def send_due_messages() -> None:
    """Every 30s: claim due jobs and send them."""
    async with get_session_factory()() as session:
        try:
            jobs = await messaging.claim_due_jobs(session, settings.sender_batch_size)
            if not jobs:
                # An idle poll is still a live poll — beat before the early return, or a
                # healthy worker with an empty queue would report as dead.
                await heartbeat.beat(session, heartbeat.SENDER)
                await session.commit()
                return

            outcome = messaging.SendOutcome()
            for job in jobs:
                result = await messaging.process_job(session, job)
                if result == MessageJobStatus.SENT:
                    outcome.sent += 1
                elif result == MessageJobStatus.SKIPPED:
                    outcome.skipped += 1
                elif result == MessageJobStatus.FAILED:
                    outcome.failed += 1
                else:
                    outcome.deferred += 1
            await heartbeat.beat(session, heartbeat.SENDER)
            await session.commit()
        except Exception as exc:
            await session.rollback()
            logger.exception("sender tick failed")
            await _beat_failure(heartbeat.SENDER, exc)
            return

    logger.info(
        "sender: %d sent, %d skipped, %d failed, %d deferred",
        outcome.sent,
        outcome.skipped,
        outcome.failed,
        outcome.deferred,
    )


async def requeue_stuck_jobs() -> None:
    """Scheduler entry point — see `messaging.requeue_stuck` for the behaviour."""
    async with get_session_factory()() as session:
        stuck = await messaging.requeue_stuck(session, STUCK_AFTER)
        await session.commit()
    if stuck:
        logger.warning("requeued %d job(s) abandoned mid-send", stuck)


async def cleanup_rate_limits() -> None:
    """Rate-limit windows accumulate forever otherwise."""
    async with get_session_factory()() as session:
        removed = await rate_limit.cleanup(session)
        await session.commit()
    if removed:
        logger.debug("pruned %d expired rate-limit window(s)", removed)


#: Suppress repeat alerts so a sustained outage does not send one email every 15 minutes.
_ALERT_COOLDOWN = timedelta(hours=3)
_last_alert_at: datetime | None = None


async def alert_on_failed_waves() -> None:
    """Alert when a send wave fails badly (task 5.6, FR §7.7: >5% failures).

    Emailed *and* logged. The log line is for whatever monitor watches stdout; the email is
    because the whole point of this system is that nobody is watching a dashboard during
    wedding week.
    """
    global _last_alert_at

    since = datetime.now(UTC) - timedelta(hours=1)
    async with get_session_factory()() as session:
        rows = (
            await session.execute(
                select(MessageJob.status, func.count(MessageJob.id))
                .where(MessageJob.created_at > since)
                .group_by(MessageJob.status)
            )
        ).all()

    # `dict(rows)` is what ruff would prefer, but Row is only tuple-*like* to mypy, so
    # the comprehension is what makes the element types checkable.
    counts: dict[MessageJobStatus, int] = {status: count for status, count in rows}  # noqa: C416
    # Skipped jobs are a deliberate outcome (no email, opted out), not a failure — counting
    # them would make a wave to guests without addresses look like an outage.
    considered = sum(c for s, c in counts.items() if s is not MessageJobStatus.SKIPPED)
    failed = counts.get(MessageJobStatus.FAILED, 0)

    # A 1-of-2 failure is 50% and means nothing. The floor stops the alert firing on noise.
    if considered < settings.failure_alert_min_sample:
        return
    ratio = failed / considered
    if ratio <= settings.failure_alert_threshold:
        return

    logger.error(
        "ALERT: %d of %d messages in the last hour failed (%.1f%%)",
        failed,
        considered,
        ratio * 100,
    )

    now = datetime.now(UTC)
    if _last_alert_at and now - _last_alert_at < _ALERT_COOLDOWN:
        return
    _last_alert_at = now

    recipient = settings.alert_email_to
    if not recipient:
        logger.warning("ALERT_EMAIL_TO is unset — the failure alert was logged only")
        return

    body = (
        f"{failed} of {considered} messages queued in the last hour failed "
        f"({ratio * 100:.1f}%, threshold {settings.failure_alert_threshold * 100:.0f}%).\n\n"
        "Check the message log for the provider error, then use Retry failed once the "
        "cause is fixed. Reminder waves are not resent automatically."
    )
    try:
        await providers.get_provider(Channel.EMAIL).send(
            providers.SendRequest(
                to=recipient,
                subject=f"[RSVP] {ratio * 100:.0f}% of messages are failing",
                html=templating.to_html(body),
                text=body,
                # Time-bucketed so a retry of this alert cannot double-send, but a genuine
                # new incident tomorrow still gets through.
                idempotency_key=f"alert:{now:%Y-%m-%dT%H}",
            )
        )
    except Exception:
        logger.exception("failure alert could not be delivered")


def build_scheduler() -> AsyncIOScheduler:
    scheduler = AsyncIOScheduler(timezone=settings.timezone)
    scheduler.add_job(
        plan_reminders,
        trigger="cron",
        minute=0,
        id="reminder_planner",
        max_instances=1,
        coalesce=True,  # a backlog collapses to one run rather than a burst
    )
    scheduler.add_job(
        send_due_messages,
        trigger="interval",
        seconds=settings.sender_poll_seconds,
        id="message_sender",
        max_instances=1,
        coalesce=True,
    )
    scheduler.add_job(
        requeue_stuck_jobs,
        trigger="interval",
        minutes=5,
        id="requeue_stuck",
        max_instances=1,
        coalesce=True,
    )
    scheduler.add_job(
        cleanup_rate_limits,
        trigger="cron",
        hour=3,
        id="rate_limit_cleanup",
        max_instances=1,
        coalesce=True,
    )
    scheduler.add_job(
        alert_on_failed_waves,
        trigger="interval",
        minutes=15,
        id="failure_alert",
        max_instances=1,
        coalesce=True,
    )
    return scheduler


async def run() -> None:
    logging.basicConfig(level=settings.log_level)
    scheduler = build_scheduler()
    scheduler.start()
    logger.info(
        "worker started (env=%s, dry_run=%s, poll=%ss, tz=%s)",
        settings.environment,
        settings.dry_run,
        settings.sender_poll_seconds,
        settings.timezone,
    )

    # Plan once at startup so a fresh deploy does not wait up to an hour for the first tick.
    await plan_reminders()

    stop = asyncio.Event()

    def _request_stop(*_: Any) -> None:
        logger.info("shutdown signal received")
        stop.set()

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        try:
            loop.add_signal_handler(sig, _request_stop)
        except NotImplementedError:
            signal.signal(sig, _request_stop)

    await stop.wait()
    # wait=True lets an in-flight send finish rather than leaving a job stranded.
    scheduler.shutdown(wait=True)
    logger.info("worker stopped")


if __name__ == "__main__":
    asyncio.run(run())