"""The message pipeline (task 4.4, design D3/D5, spec messaging, reminders).

`message_job` is the queue — there is no broker. The sender claims due rows with
`FOR UPDATE SKIP LOCKED`, which lets concurrent workers take disjoint batches without
blocking, and transitions state inside the same transaction.

Two invariants carry the system:

* `idempotency_key` is unique. Without it, a duplicated planner run or a restart mid-batch
  would message every guest twice — the cheapest insurance in the whole design.
* The sender re-reads the invitation's live status before sending. That recheck is what
  makes a cancellation take effect even for a job that was already queued.
"""

import asyncio
import logging
import re
import secrets
import time
import uuid
from collections.abc import Sequence
from dataclasses import dataclass
from datetime import UTC, datetime, timedelta
from datetime import time as dt_time
from zoneinfo import ZoneInfo

from sqlalchemy import func, select, update
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.config import get_settings
from app.db import get_session_factory
from app.models import Event, Guest, Invitation, MessageJob, MessageTemplate, Wedding
from app.models.card import EventCardDesign
from app.models.enums import (
    CardDesignStatus,
    Channel,
    InvitationStatus,
    InvitationType,
    Locale,
    MessageJobStatus,
    TemplatePurpose,
)
from app.services import greeting, link_preview, providers, rsvp_service, templating

logger = logging.getLogger(__name__)

#: Backoff schedule from FR-6.10. Three attempts, then the job is failed and surfaced.
RETRY_DELAYS = (timedelta(minutes=1), timedelta(minutes=10), timedelta(hours=1))
MAX_ATTEMPTS = len(RETRY_DELAYS)

SKIP_NO_EMAIL = "guest has no email address (v1 is email-only)"
# Suppression lives on the guest record, and a record belongs to one event (design D11), so
# both of these describe this event alone — the same person may still be mailed about another.
SKIP_DO_NOT_CONTACT = "guest opted out of this event"
SKIP_EMAIL_INVALID = "address hard-bounced previously for this event"
SKIP_NOT_ACCEPTED = "invitation is no longer accepted"


def idempotency_key(
    invitation_id: uuid.UUID, schedule_id: uuid.UUID | None, channel: Channel
) -> str:
    """invitation:schedule:channel (PRD §7.3). `direct` covers non-scheduled sends."""
    return f"{invitation_id}:{schedule_id or 'direct'}:{channel}"


def manual_idempotency_key(invitation_id: uuid.UUID, channel: Channel = Channel.EMAIL) -> str:
    """A key unique to one manual send (add-guest-invitation-send D2).

    The scheduled key deduplicates on purpose: a planner that runs twice must not mail anyone
    twice. A manual send is the opposite case — an admin resending an invitation the guest
    says never arrived means it, and the shared `direct` key would make that second click an
    `ON CONFLICT DO NOTHING` no-op reported to them as success. Silent is the part that
    matters; nothing would have gone out.

    The nonce still rides to the provider as its idempotency header, so a network retry
    *within* one send collapses vendor-side. Duplicate protection for waves is untouched:
    they keep using `idempotency_key` above.
    """
    return f"{invitation_id}:manual-{secrets.token_hex(8)}:{channel}"


def batch_idempotency_key(
    invitation_id: uuid.UUID, batch_id: uuid.UUID, channel: Channel = Channel.EMAIL
) -> str:
    """A key unique to one invitation within one batch (add-bulk-invitation-send D5).

    Both behaviours the batched send needs come out of this one form. *Within* a batch the key
    repeats exactly, so a double-clicked Send, a retried request or a refreshed browser hits
    `ON CONFLICT DO NOTHING` and inserts nothing twice. *Between* batches the id differs, so a
    host deliberately sending again is delivered — the same reasoning as the manual nonce
    above, moved up from one send to one action.

    That is also why the batch id comes from the client: it has to exist before the request
    that might be duplicated. Generating it here would make the second click a second batch.
    """
    return f"{invitation_id}:batch-{batch_id}:{channel}"


class TokenBucket:
    """In-process rate limiter for provider calls (design D10).

    In-process is correct here because exactly one worker sends. If that ever changes, this
    is the piece that must move to Postgres alongside the queue.
    """

    def __init__(self, rate_per_second: float, capacity: int) -> None:
        self._rate = rate_per_second
        self._capacity = capacity
        self._tokens = float(capacity)
        self._updated = time.monotonic()
        self._lock = asyncio.Lock()

    async def acquire(self) -> None:
        async with self._lock:
            now = time.monotonic()
            self._tokens = min(self._capacity, self._tokens + (now - self._updated) * self._rate)
            self._updated = now
            if self._tokens < 1:
                wait = (1 - self._tokens) / self._rate
                await asyncio.sleep(wait)
                self._tokens = 0.0
                self._updated = time.monotonic()
            else:
                self._tokens -= 1


_bucket = TokenBucket(rate_per_second=10, capacity=20)


def in_quiet_hours(moment: datetime, timezone: str, start: str, end: str) -> bool:
    """FR-6.9: nothing goes out between 22:00 and 08:00 local."""
    local = moment.astimezone(ZoneInfo(timezone)).time()
    start_t = dt_time.fromisoformat(start)
    end_t = dt_time.fromisoformat(end)
    if start_t <= end_t:
        return start_t <= local < end_t
    # The window wraps midnight, which is the normal case for 22:00-08:00.
    return local >= start_t or local < end_t


def defer_past_quiet_hours(moment: datetime, timezone: str, start: str, end: str) -> datetime:
    """Move a send to the next 08:00 local if it lands in the quiet window."""
    if not in_quiet_hours(moment, timezone, start, end):
        return moment
    tz = ZoneInfo(timezone)
    local = moment.astimezone(tz)
    end_t = dt_time.fromisoformat(end)
    target = local.replace(hour=end_t.hour, minute=end_t.minute, second=0, microsecond=0)
    if local.time() >= dt_time.fromisoformat(start):
        target += timedelta(days=1)  # after 22:00 -> tomorrow morning
    return target.astimezone(UTC)


async def enqueue(
    session: AsyncSession,
    *,
    invitation_id: uuid.UUID,
    template_id: uuid.UUID | None,
    schedule_id: uuid.UUID | None,
    scheduled_for: datetime,
    channel: Channel = Channel.EMAIL,
    subject: str | None = None,
    body_text: str | None = None,
    override_quiet_hours: bool = False,
    sent_by_admin_id: uuid.UUID | None = None,
    batch_id: uuid.UUID | None = None,
    key: str | None = None,
) -> uuid.UUID | None:
    """Insert a job if one does not already exist. Returns its id, or None on conflict.

    ON CONFLICT DO NOTHING against the unique idempotency key is the entire duplicate
    defence — it holds no matter how many planners run concurrently.

    Every keyword after `channel` exists for the manual and batched paths and defaults to what
    the scheduled and bulk callers already did: no stored content, quiet hours respected, no
    admin recorded, no batch, and the derived `invitation:schedule:channel` key.
    """
    result = await session.execute(
        insert(MessageJob)
        .values(
            invitation_id=invitation_id,
            template_id=template_id,
            schedule_id=schedule_id,
            channel=channel,
            scheduled_for=scheduled_for,
            status=MessageJobStatus.QUEUED,
            idempotency_key=key or idempotency_key(invitation_id, schedule_id, channel),
            subject=subject,
            body_text=body_text,
            override_quiet_hours=override_quiet_hours,
            sent_by_admin_id=sent_by_admin_id,
            batch_id=batch_id,
        )
        .on_conflict_do_nothing(index_elements=["idempotency_key"])
        .returning(MessageJob.id)
    )
    inserted: uuid.UUID | None = result.scalar_one_or_none()
    return inserted


async def claim_due_jobs(
    session: AsyncSession, limit: int, *, batch_id: uuid.UUID | None = None
) -> list[MessageJob]:
    """Take a batch of due jobs, marking them `sending` so nobody else picks them up.

    `SKIP LOCKED` is what makes this safe with more than one worker: a second worker steps
    over the locked rows instead of waiting on them. It is also what lets the API dispatch a
    batch it has just recorded while the worker's poll is running — the two take disjoint rows
    rather than sending anything twice.

    `batch_id` narrows the claim to one batch. The API passes it so pressing Send does not
    quietly turn the web process into a second sender for every queued reminder in the system;
    the worker passes nothing and keeps draining the whole queue as before.
    """
    now = datetime.now(UTC)
    filters = [
        MessageJob.status == MessageJobStatus.QUEUED,
        MessageJob.scheduled_for <= now,
    ]
    if batch_id is not None:
        filters.append(MessageJob.batch_id == batch_id)
    subquery = (
        select(MessageJob.id)
        .where(*filters)
        .order_by(MessageJob.scheduled_for)
        .limit(limit)
        .with_for_update(skip_locked=True)
        .scalar_subquery()
    )
    rows = await session.execute(
        update(MessageJob)
        .where(MessageJob.id.in_(subquery))
        .values(status=MessageJobStatus.SENDING)
        .returning(MessageJob.id)
    )
    ids = [r[0] for r in rows]
    if not ids:
        return []

    # MessageJob has no ORM relationships by design: the sender loads exactly what it
    # needs per job, which keeps the claim query narrow and lock windows short.
    jobs = await session.scalars(select(MessageJob).where(MessageJob.id.in_(ids)))
    return list(jobs)


async def requeue_stuck(session: AsyncSession, older_than: timedelta) -> int:
    """Return jobs abandoned in `sending` to the queue, and report how many.

    Claiming marks a job `sending` before the send is attempted; if the process dies in
    between, the row would sit there forever and that guest silently gets nothing. This is
    the only path that recovers it.

    `older_than` is a grace window, not a formality: requeueing a job whose sender is merely
    slow hands the same email to a second sender, and the guest receives it twice.
    """
    cutoff = datetime.now(UTC) - older_than
    result = await session.execute(
        update(MessageJob)
        .where(MessageJob.status == MessageJobStatus.SENDING, MessageJob.updated_at < cutoff)
        .values(status=MessageJobStatus.QUEUED)
        .returning(MessageJob.id)
    )
    return len(result.all())


@dataclass
class SendOutcome:
    sent: int = 0
    skipped: int = 0
    failed: int = 0
    deferred: int = 0


def _skip_reason(
    invitation: Invitation, guest: Guest | None, audience_accepted: bool
) -> str | None:
    """Decide whether this job should still go out, at send time rather than plan time."""
    if audience_accepted and invitation.status is not InvitationStatus.ACCEPTED:
        return SKIP_NOT_ACCEPTED
    if guest is None:
        return SKIP_NO_EMAIL
    if guest.do_not_contact:
        return SKIP_DO_NOT_CONTACT
    if not guest.email:
        return SKIP_NO_EMAIL
    if guest.email_invalid:
        return SKIP_EMAIL_INVALID
    return None


async def _load_template(
    session: AsyncSession, purpose: TemplatePurpose, locale: Locale
) -> MessageTemplate | None:
    """Fall back to English if a locale variant is missing — sending the wrong language
    beats sending nothing."""
    template = await session.scalar(
        select(MessageTemplate).where(
            MessageTemplate.channel == Channel.EMAIL,
            MessageTemplate.purpose == purpose,
            MessageTemplate.locale == locale,
            MessageTemplate.is_active.is_(True),
        )
    )
    if template is not None:
        return template
    fallback: MessageTemplate | None = await session.scalar(
        select(MessageTemplate).where(
            MessageTemplate.channel == Channel.EMAIL,
            MessageTemplate.purpose == purpose,
            MessageTemplate.locale == Locale.EN,
            MessageTemplate.is_active.is_(True),
        )
    )
    return fallback


async def preview_block_for(session: AsyncSession, invitation: Invitation) -> str:
    """The link-preview block for this guest's invitation email (design D4).

    Built at send time and never stored. It carries the guest's name and their token URL,
    so writing it into `message_log` would put both somewhere the PII rule keeps them out of.

    `for_guest` and not `for_event`: this is rendered by us, for one inbox, and no crawler
    ever sees it — the guest-free rule that governs the meta tags is about a different
    audience (design D3).

    Returns an empty string when the event has no published design, or has one published
    without a picture. The block still renders in that case; it simply has no image in it,
    which is the degradation the spec asks for.
    """
    settings = get_settings()
    guest = invitation.guest
    event = invitation.event

    design = await session.scalar(
        select(EventCardDesign).where(
            EventCardDesign.event_id == event.id,
            EventCardDesign.status == CardDesignStatus.PUBLISHED,
        )
    )
    built = link_preview.for_guest(
        event,
        event.wedding,
        guest_name=(guest.full_name if guest else ""),
        base_url=settings.app_base_url,
        card_config=design.config if design is not None else None,
    )
    return link_preview.email_block(
        built, invite_url=link_preview.invitation_url(settings.app_base_url, invitation.token)
    )


async def render_for_invitation(
    session: AsyncSession,
    invitation: Invitation,
    purpose: TemplatePurpose,
) -> templating.RenderedMessage | None:
    """Render one message. Returns None when no template exists for the purpose."""
    settings = get_settings()
    guest = invitation.guest
    event = invitation.event
    wedding = event.wedding
    locale = guest.preferred_locale if guest else Locale.EN

    template = await _load_template(session, purpose, locale)
    if template is None:
        logger.warning("no active template for purpose=%s locale=%s", purpose, locale)
        return None

    invite_url, cancel_url = rsvp_service.build_urls(settings.app_base_url, invitation.token)
    unsubscribe_url = rsvp_service.build_unsubscribe_url(settings.app_base_url, invitation.token)
    event_title = event.title_bn if locale is Locale.BN else event.title_en
    variables = templating.build_variables(
        guest_name=(guest.full_name if guest else "Guest"),
        event_title=event_title,
        starts_at=event.starts_at,
        venue=event.venue_name,
        invite_url=invite_url,
        cancel_url=cancel_url,
        map_url=event.map_url,
        couple_names=f"{wedding.bride_name} & {wedding.groom_name}",
        unsubscribe_url=unsubscribe_url,
        timezone=wedding.timezone or settings.timezone,
        locale=locale,
    )

    subject, missing_subject = templating.render(template.subject or "", variables)
    body, missing_body = templating.render(template.body_text or template.body, variables)
    # Every email carries an opt-out (spec messaging FR-6.11), and it is added here rather
    # than left to each template so a host editing their copy cannot remove it.
    body = templating.append_unsubscribe(
        body, event_title=event_title, url=unsubscribe_url, locale=locale
    )
    # Invitations only. A reminder is a nudge to someone who already has the invitation
    # in their inbox, and repeating the card on every one turns it into wallpaper.
    block = (
        await preview_block_for(session, invitation) if purpose is TemplatePurpose.INVITE else ""
    )

    return templating.RenderedMessage(
        subject=subject,
        html=templating.to_html(body, preview_block=block),
        text=body,
        missing_placeholders=sorted(set(missing_subject) | set(missing_body)),
    )


# ------------------------------------------------------- manual, per-guest composition
#
# One admin, one guest, one message they can edit before it goes (D1-D4). Composition lives
# here rather than in the frontend because the sentence a guest reads resolves
# event -> wedding -> built-in default by locale *and* invitation type, and deliberately does
# not fall back across locales. A second implementation of that in TypeScript is how the
# admin's preview and the guest's invitation page start disagreeing about the same sentence.

#: The wording around the host's message. Not template rows: these are structural, the admin
#: edits them per send, and a missing template must never be able to leave a message headerless.
MANUAL_CHROME: dict[Locale, dict[str, str]] = {
    Locale.EN: {
        "header": "Dear {guest_name}",
        "footer": "Regards\n{host_name}",
        "subject": "You're invited — {event_title}",
    },
    Locale.BN: {
        "header": "প্রিয় {guest_name}",
        "footer": "শুভেচ্ছান্তে\n{host_name}",
        "subject": "আপনার আমন্ত্রণ — {event_title}",
    },
}

#: Why a manual send is refused, in words an admin can act on. These are not the `SKIP_*`
#: reasons above: those explain a job the worker declined, these explain a button that is
#: disabled before anything is queued.
BLOCK_NO_EMAIL = "This guest has no email address."
BLOCK_OPTED_OUT = "This guest unsubscribed from this event."
BLOCK_EMAIL_INVALID = "This address hard-bounced, so email to it is disabled."


@dataclass(frozen=True)
class ComposedMessage:
    """The default message for one invitation, in the parts the admin edits."""

    subject: str
    header: str
    body: str
    footer: str
    invite_url: str
    to_email: str | None
    locale: Locale
    event_title: str


#: Stable codes for the same three states, so a batch can report "3 opted out, 1 bounced"
#: without the interface matching on English sentences.
BLOCK_CODE_NO_EMAIL = "no_email"
BLOCK_CODE_OPTED_OUT = "opted_out"
BLOCK_CODE_EMAIL_INVALID = "email_invalid"


@dataclass(frozen=True)
class Suppression:
    """Why one guest cannot be emailed: a code to count by, a sentence to show."""

    code: str
    reason: str


def suppression(guest: Guest | None) -> Suppression | None:
    """Why this guest cannot be emailed, or None (add-guest-invitation-send D6).

    `do_not_contact` is a consent record and is checked first: it is the one state that no
    correction elsewhere in the interface should appear to resolve.

    One function rather than a code check beside a reason check — the batch counts by code and
    the compose panel shows the sentence, and two implementations of "can this guest be
    emailed" is exactly the pair that drifts.
    """
    if guest is None:
        return Suppression(BLOCK_CODE_NO_EMAIL, BLOCK_NO_EMAIL)
    if guest.do_not_contact:
        return Suppression(BLOCK_CODE_OPTED_OUT, BLOCK_OPTED_OUT)
    if not guest.email:
        return Suppression(BLOCK_CODE_NO_EMAIL, BLOCK_NO_EMAIL)
    if guest.email_invalid:
        return Suppression(BLOCK_CODE_EMAIL_INVALID, BLOCK_EMAIL_INVALID)
    return None


def block_reason(guest: Guest | None) -> str | None:
    """The sentence form of `suppression`, for the single-guest compose panel."""
    found = suppression(guest)
    return found.reason if found else None


def assemble(header: str, body: str, footer: str) -> str:
    """The plain-text message that is stored and sent.

    One function so the compose endpoint and the send endpoint cannot disagree about how the
    three parts join — if they did, an admin would approve one message and a guest would
    receive another.
    """
    parts = [part.strip() for part in (header, body, footer)]
    return "\n\n".join(part for part in parts if part)


def compose_manual(invitation: Invitation) -> ComposedMessage:
    """The default message for one invitation (D4).

    Requires `invitation.guest` and `invitation.event.wedding` to be loaded — the caller is
    already fetching them to authorize the request.
    """
    settings = get_settings()
    guest = invitation.guest
    event = invitation.event
    wedding = event.wedding
    locale = guest.preferred_locale if guest else Locale.EN
    chrome = MANUAL_CHROME.get(locale, MANUAL_CHROME[Locale.EN])

    invite_url, _cancel_url = rsvp_service.build_urls(settings.app_base_url, invitation.token)
    event_title = event.title_bn if locale is Locale.BN else event.title_en
    host_name = f"{wedding.bride_name} & {wedding.groom_name}"

    # The host's own sentence, resolved exactly as the invitation page resolves it, then the
    # link — which is the whole point of the email, so it is part of the body an admin can
    # move rather than something bolted on after they have finished editing.
    sentence = greeting.resolve(
        event.invitation_messages,
        wedding.invitation_messages,
        locale=locale,
        invitation_type=guest.invitation_type if guest else InvitationType.SINGLE,
    )

    return ComposedMessage(
        subject=chrome["subject"].format(event_title=event_title),
        header=chrome["header"].format(guest_name=guest.full_name if guest else ""),
        body=f"{sentence}\n\n{invite_url}",
        footer=chrome["footer"].format(host_name=host_name),
        invite_url=invite_url,
        to_email=guest.email if guest else None,
        locale=locale,
        event_title=event_title,
    )


# ------------------------------------------------------- batched, per-type composition
#
# One admin, one event, two messages — one per invitation type — sent to a chosen set of
# guests (add-bulk-invitation-send D3, D4, D11). The difference from the manual path above is
# that no guest is involved in composing: the guest-specific parts stay as placeholders and
# are substituted per recipient at enqueue time, so one authored text becomes 300 personal
# messages carrying 300 different links.

#: The only placeholders a batched composition may carry (D4).
PLACEHOLDER_GUEST_NAME = "guest_name"
PLACEHOLDER_INVITATION_LINK = "invitation_link"
PLACEHOLDERS = frozenset({PLACEHOLDER_GUEST_NAME, PLACEHOLDER_INVITATION_LINK})

#: Anything in braces. Deliberately matches unknown names too — finding them is the point.
_PLACEHOLDER_RE = re.compile(r"\{[^{}]*\}")


class TemplateError(ValueError):
    """A composed template that must not be sent. Carries wording an admin can act on."""


class UnknownPlaceholderError(TemplateError):
    """Something in braces that is not one of ours.

    Refused rather than passed through literally: a mistyped `{guset_name}` delivered to 300
    inboxes cannot be recalled, and the admin who typed it is the only person who can fix it.
    That makes rejecting the rare literal brace in invitation copy the cheaper mistake.
    """

    def __init__(self, found: str) -> None:
        super().__init__(
            f"{found} is not a placeholder this message understands. "
            f"Use {{{PLACEHOLDER_GUEST_NAME}}} or {{{PLACEHOLDER_INVITATION_LINK}}}, "
            f"or remove the braces."
        )
        self.found = found


class MissingLinkPlaceholderError(TemplateError):
    """The body no longer carries the invitation link."""

    def __init__(self) -> None:
        super().__init__(
            f"The message needs {{{PLACEHOLDER_INVITATION_LINK}}} in its body — "
            f"without it the guest receives an invitation with no way to answer."
        )


@dataclass(frozen=True)
class ComposedTemplate:
    """One pane: the message for one invitation type, before any guest is substituted in."""

    subject: str
    header: str
    body: str
    footer: str


@dataclass(frozen=True)
class RenderedParts:
    """One recipient's finished message, ready to be stored on their job."""

    subject: str
    body: str


def compose_bulk(
    event: Event, wedding: Wedding, *, locale: Locale, invitation_type: InvitationType
) -> ComposedTemplate:
    """The default message for one invitation type of one event (D3, D11).

    Takes no guest, because it describes all of them. The parts that differ per recipient are
    left as placeholders; everything constant for the event — the title in the subject, the
    couple's names in the footer — is resolved now, so the admin edits words rather than
    template syntax.

    The invitation sentence comes from the same `greeting.resolve` chain the invitation page
    and `compose_manual` use, and deliberately does not fall back across locales.
    """
    chrome = MANUAL_CHROME.get(locale, MANUAL_CHROME[Locale.EN])
    event_title = event.title_bn if locale is Locale.BN else event.title_en
    host_name = f"{wedding.bride_name} & {wedding.groom_name}"

    sentence = greeting.resolve(
        event.invitation_messages,
        wedding.invitation_messages,
        locale=locale,
        invitation_type=invitation_type,
    )

    return ComposedTemplate(
        subject=chrome["subject"].format(event_title=event_title),
        # Used verbatim: the chrome already reads `Dear {guest_name}`, and that placeholder is
        # exactly what has to survive composition.
        header=chrome["header"],
        body=f"{sentence}\n\n{{{PLACEHOLDER_INVITATION_LINK}}}",
        footer=chrome["footer"].format(host_name=host_name),
    )


def validate_template(template: ComposedTemplate) -> None:
    """Refuse a template that must not reach 300 inboxes. Raises `TemplateError` (D4).

    Checked across all four parts, before the first row is inserted, so a batch is
    all-or-nothing on this rather than half-sent and half-refused.
    """
    for part in (template.subject, template.header, template.body, template.footer):
        for found in _PLACEHOLDER_RE.findall(part):
            if found[1:-1] not in PLACEHOLDERS:
                raise UnknownPlaceholderError(found)

    if f"{{{PLACEHOLDER_INVITATION_LINK}}}" not in template.body:
        raise MissingLinkPlaceholderError()


def render_for_guest(
    template: ComposedTemplate, *, guest_name: str, invite_url: str
) -> RenderedParts:
    """One recipient's copy of a composed template (D3).

    Substitution is `str.replace`, not `str.format`. The text is written by a person, and
    `format` would give every remaining brace in it meaning — an unmatched `{` becomes an
    exception thrown at send time, in a loop over 300 guests, long after the admin could have
    been told. `validate_template` has already refused the braces that matter; the rest are
    left alone as the literal characters they look like.

    `assemble` joins the three parts, so a batched message and a single manual one are the
    same message put together the same way.
    """

    def fill(part: str) -> str:
        return part.replace(f"{{{PLACEHOLDER_GUEST_NAME}}}", guest_name).replace(
            f"{{{PLACEHOLDER_INVITATION_LINK}}}", invite_url
        )

    return RenderedParts(
        subject=fill(template.subject),
        body=assemble(fill(template.header), fill(template.body), fill(template.footer)),
    )


#: How many failed or skipped recipients a progress report names. The counts above it are
#: always exact; this bounds the list a provider outage could otherwise make enormous, polled
#: every couple of seconds.
BATCH_PROBLEM_LIMIT = 200


@dataclass(frozen=True)
class BatchEnqueueResult:
    """What one batch insert actually did."""

    queued: int
    #: Rows that already existed under this batch id — a repeated request, not a new send.
    duplicate: int
    #: Suppressed guests, counted by `Suppression.code`. They get no job at all (D10).
    excluded: dict[str, int]


async def enqueue_batch(
    session: AsyncSession,
    *,
    invitations: Sequence[Invitation],
    templates: dict[tuple[Locale, InvitationType], ComposedTemplate],
    batch_id: uuid.UUID,
    override_quiet_hours: bool,
    sent_by_admin_id: uuid.UUID | None,
    channel: Channel = Channel.EMAIL,
) -> BatchEnqueueResult:
    """Record one batched send: one job per sendable guest, in one statement (D3, D5, D10).

    Each job carries its own finished subject and body, so the send path needs no new branch —
    `process_job` already prefers stored content over a template. The cost is that the batch
    stores every recipient's name and tokenised URL in `message_job.body_text`, which is why
    that column must never be rendered into the message log.

    Suppressed guests are filtered here rather than skipped later: a job that exists only to
    be refused is noise in the log, and the admin needs the count *before* they press Send.
    `process_job` still re-checks, which is what catches an unsubscribe that lands between
    recording and sending.

    Requires `invitation.guest` to be loaded — the caller has already fetched them to
    authorize the request and to count the panes.
    """
    settings = get_settings()
    now = datetime.now(UTC)

    rows: list[dict[str, object]] = []
    excluded: dict[str, int] = {}

    for invitation in invitations:
        guest = invitation.guest
        blocked = suppression(guest)
        if blocked is not None or guest is None:
            code = blocked.code if blocked else BLOCK_CODE_NO_EMAIL
            excluded[code] = excluded.get(code, 0) + 1
            continue

        template = templates.get((guest.preferred_locale, guest.invitation_type))
        if template is None:  # pragma: no cover — the caller derives the pairs from these rows
            raise TemplateError(
                f"No message was composed for {guest.preferred_locale}/{guest.invitation_type}."
            )

        invite_url, _cancel_url = rsvp_service.build_urls(settings.app_base_url, invitation.token)
        parts = render_for_guest(template, guest_name=guest.full_name, invite_url=invite_url)
        rows.append(
            {
                "invitation_id": invitation.id,
                "template_id": None,
                "schedule_id": None,
                "channel": channel,
                "scheduled_for": now,
                "status": MessageJobStatus.QUEUED,
                "idempotency_key": batch_idempotency_key(invitation.id, batch_id, channel),
                "subject": parts.subject,
                "body_text": parts.body,
                "override_quiet_hours": override_quiet_hours,
                "sent_by_admin_id": sent_by_admin_id,
                "batch_id": batch_id,
            }
        )

    if not rows:
        return BatchEnqueueResult(queued=0, duplicate=0, excluded=excluded)

    # One statement rather than one per guest: at 1,500 recipients the round trips are the
    # difference between an interactive request and a timeout.
    result = await session.execute(
        insert(MessageJob)
        .values(rows)
        .on_conflict_do_nothing(index_elements=["idempotency_key"])
        .returning(MessageJob.id)
    )
    queued = len(result.all())
    return BatchEnqueueResult(queued=queued, duplicate=len(rows) - queued, excluded=excluded)


@dataclass(frozen=True)
class BatchProblem:
    """One recipient a batch could not deliver to, in the words the admin needs."""

    guest_name: str
    status: str
    reason: str


@dataclass(frozen=True)
class BatchProgress:
    """How a batch is getting on (D7). Counts are exact; `problems` is capped."""

    total: int
    #: Still to go: queued, waiting on a retry, or in flight.
    waiting: int
    sent: int
    failed: int
    skipped: int
    problems: list[BatchProblem]

    @property
    def finished(self) -> bool:
        return self.waiting == 0


async def batch_progress(session: AsyncSession, batch_id: uuid.UUID) -> BatchProgress:
    """Counts by status for one batch, plus who did not receive it and why (D7).

    Two queries rather than one pass over the rows: a poll every couple of seconds should not
    drag 1,500 bodies across the wire to add up four numbers.
    """
    counts: dict[MessageJobStatus, int] = {
        row_status: int(row_count)
        for row_status, row_count in (
            await session.execute(
                select(MessageJob.status, func.count())
                .where(MessageJob.batch_id == batch_id)
                .group_by(MessageJob.status)
            )
        ).all()
    }

    def tally(*statuses: MessageJobStatus) -> int:
        return sum(counts.get(s, 0) for s in statuses)

    problems = [
        BatchProblem(
            guest_name=name or "(removed guest)",
            status=str(status),
            # A skip explains itself; a failure carries the provider's words. Neither is
            # guaranteed to be set, and "no reason recorded" beats an empty cell.
            reason=skip_reason or error_message or "no reason recorded",
        )
        for status, skip_reason, error_message, name in (
            await session.execute(
                select(
                    MessageJob.status,
                    MessageJob.skip_reason,
                    MessageJob.error_message,
                    Guest.full_name,
                )
                .join(Invitation, Invitation.id == MessageJob.invitation_id)
                .outerjoin(Guest, Guest.id == Invitation.guest_id)
                .where(
                    MessageJob.batch_id == batch_id,
                    MessageJob.status.in_([MessageJobStatus.FAILED, MessageJobStatus.SKIPPED]),
                )
                .order_by(Guest.full_name)
                .limit(BATCH_PROBLEM_LIMIT)
            )
        ).all()
    ]

    return BatchProgress(
        total=sum(counts.values()),
        waiting=tally(MessageJobStatus.QUEUED, MessageJobStatus.SENDING),
        # Delivered and read are sent plus a receipt; for "did it go out?" all three are the
        # same answer, and a batch that lost them would report a shrinking sent count as
        # webhooks arrived.
        sent=tally(MessageJobStatus.SENT, MessageJobStatus.DELIVERED, MessageJobStatus.READ),
        failed=tally(MessageJobStatus.FAILED),
        skipped=tally(MessageJobStatus.SKIPPED),
        problems=problems,
    )


def render_stored(
    job: MessageJob, invitation: Invitation, preview_block: str = ""
) -> templating.RenderedMessage:
    """Turn a job's own stored text into a sendable message (D3, D12).

    The same two finishing steps the template path applies: the opt-out footer, and the
    plain-text-to-HTML wrapper. An admin who deleted the unsubscribe wording while editing
    gets it back — a host editing their own copy is exactly the case that footer defends
    against, and the guest's right to unsubscribe does not depend on who typed the body.
    """
    settings = get_settings()
    guest = invitation.guest
    locale = guest.preferred_locale if guest else Locale.EN
    event = invitation.event
    event_title = event.title_bn if locale is Locale.BN else event.title_en

    unsubscribe_url = rsvp_service.build_unsubscribe_url(settings.app_base_url, invitation.token)
    body = templating.append_unsubscribe(
        job.body_text or "", event_title=event_title, url=unsubscribe_url, locale=locale
    )
    return templating.RenderedMessage(
        subject=job.subject or "",
        html=templating.to_html(body, preview_block=preview_block),
        text=body,
        missing_placeholders=[],
    )


async def last_manual_send_at(session: AsyncSession, invitation_id: uuid.UUID) -> datetime | None:
    """When this invitation was last emailed, for the compose panel to state (D8).

    Read from the jobs rather than denormalised onto the invitation: the data is already
    there, and a cached column would need maintaining in the webhook path too.
    """
    return await session.scalar(
        select(func.max(MessageJob.sent_at)).where(
            MessageJob.invitation_id == invitation_id,
            MessageJob.status.in_([MessageJobStatus.SENT, MessageJobStatus.DELIVERED]),
        )
    )


async def process_job(
    session: AsyncSession, job: MessageJob, *, audience_accepted: bool = True
) -> str:
    """Send one claimed job. Returns the resulting status name.

    Every exit path writes a terminal or retry state — a job must never be left `sending`,
    because nothing would ever pick it up again.
    """
    settings = get_settings()

    invitation = await session.scalar(
        select(Invitation)
        .where(Invitation.id == job.invitation_id)
        .options(
            selectinload(Invitation.guest),
            selectinload(Invitation.event).selectinload(Event.wedding),
        )
    )
    if invitation is None:
        job.status = MessageJobStatus.SKIPPED
        job.skip_reason = "invitation no longer exists"
        return job.status

    # A job carrying its own content is an invitation an admin composed and sent, so it never
    # requires an accepted invitation — by definition it goes to someone who has not answered
    # (add-bulk-invitation-send D8). The rule lives on the job rather than on the caller
    # because the worker claims these too, on retry and after a restart, and it passes the
    # default `True`; deciding it there would skip every batched invitation as "no longer
    # accepted". Reminders carry no stored body, so their behaviour is unchanged.
    requires_accepted = audience_accepted and job.body_text is None

    # Live-status recheck. This is what makes cancellation retroactive.
    reason = _skip_reason(invitation, invitation.guest, requires_accepted)
    if reason:
        job.status = MessageJobStatus.SKIPPED
        job.skip_reason = reason
        return job.status

    now = datetime.now(UTC)
    # An admin who confirmed a late-night send has already made this call, and deferring it
    # to 08:00 while telling them it went out would be a lie (D5). Nothing else sets the
    # flag, so every planned wave still defers exactly as FR-6.9 requires.
    if not job.override_quiet_hours:
        deferred = defer_past_quiet_hours(
            now, settings.timezone, settings.quiet_hours_start, settings.quiet_hours_end
        )
        if deferred > now:
            job.status = MessageJobStatus.QUEUED
            job.scheduled_for = deferred
            return "deferred"

    if job.body_text:
        # A hand-edited message: there is no template that could reproduce it, so the job
        # carries its own copy (D3). The unsubscribe footer is still appended here rather
        # than trusted to the admin's editing — it is the one line a host may not remove.
        rendered = render_stored(job, invitation, await preview_block_for(session, invitation))
    else:
        purpose = TemplatePurpose.INVITE
        if job.schedule_id is not None:
            purpose = TemplatePurpose.REMINDER_2  # refined by the planner via template_id

        maybe_rendered = await render_for_invitation(session, invitation, purpose)
        if maybe_rendered is None:
            job.status = MessageJobStatus.FAILED
            job.error_code = "no_template"
            job.error_message = f"no active template for {purpose}"
            return job.status
        rendered = maybe_rendered

    # _skip_reason already guaranteed this, but an assert would vanish under -O and the
    # failure mode would be a confusing provider error instead of a clear skip.
    guest_email = invitation.guest.email if invitation.guest else None
    if not guest_email:
        job.status = MessageJobStatus.SKIPPED
        job.skip_reason = SKIP_NO_EMAIL
        return job.status

    job.attempts += 1
    try:
        await _bucket.acquire()
        provider = providers.get_provider(Channel.EMAIL)
        result = await provider.send(
            providers.SendRequest(
                to=guest_email,
                subject=rendered.subject,
                html=rendered.html,
                text=rendered.text,
                idempotency_key=job.idempotency_key,
                # Every send, not manual ones only: a guest replying to a reminder should
                # reach the couple too, and dropping that reply is the same failure (D9).
                reply_to=invitation.event.wedding.host_email,
            )
        )
    except Exception as exc:
        if job.attempts < MAX_ATTEMPTS:
            job.status = MessageJobStatus.QUEUED
            job.scheduled_for = now + RETRY_DELAYS[job.attempts - 1]
            job.error_message = str(exc)[:500]
            logger.warning("send attempt %d failed, retrying: %s", job.attempts, exc)
            return "retry"
        job.status = MessageJobStatus.FAILED
        job.error_code = type(exc).__name__
        job.error_message = str(exc)[:500]
        logger.error("send failed permanently after %d attempts", job.attempts)
        return job.status

    job.status = MessageJobStatus.SENT
    job.provider_message_id = result.provider_message_id
    job.sent_at = now
    job.error_code = None
    job.error_message = None
    return job.status


async def dispatch_batch(batch_id: uuid.UUID) -> None:
    """Send one batch's jobs now, from the API process, after the request has returned (D6).

    Runs as a FastAPI `BackgroundTask`, so it opens its own session — the request's is closed
    by the time this starts. It claims only this batch (`claim_due_jobs(batch_id=…)`), because
    pressing Send must not turn the web process into a second sender for every queued reminder
    in the system.

    This is an accelerator, not the delivery guarantee. The rows are committed before it
    starts, so if it never runs — a restart, a crash, an unhandled provider hang — the worker's
    30-second poll picks the same jobs up and finishes them. `SKIP LOCKED` is what makes both
    working at once safe, and D8 is what stops the worker skipping them as "not accepted".

    Bounded by the number of rows the batch could contain: each pass claims and drains a
    slice, and the loop ends when a pass finds nothing left. A job that fails into a retry has
    a future `scheduled_for`, so it is not due and does not spin here — the worker collects it
    later.
    """
    settings = get_settings()
    async with get_session_factory()() as session:
        while True:
            try:
                jobs = await claim_due_jobs(session, settings.sender_batch_size, batch_id=batch_id)
                if not jobs:
                    return
                for job in jobs:
                    # `audience_accepted=False` is belt and braces: these jobs carry stored
                    # content, so `process_job` would reach the same conclusion on its own.
                    await process_job(session, job, audience_accepted=False)
                await session.commit()
            except Exception:
                await session.rollback()
                # Deliberately swallowed after logging: this is a background task with nobody
                # to return an error to, and the jobs it did not finish are still queued for
                # the worker. Re-raising would lose that distinction in a stack trace.
                logger.exception(
                    "batch dispatch failed for %s; leaving the rest to the worker", batch_id
                )
                return