"""Guest CRUD, search, filters and manual RSVP override (tasks 3.6, 3.7). Deletion is soft (spec guest-management): the guest disappears from counts, sends and exports, but their history survives for the audit trail. Manual accept/decline exists because elderly guests will phone the host rather than use the link, and those answers still have to reach the caterer. """ import uuid from datetime import UTC, datetime from typing import Literal from zoneinfo import ZoneInfo from fastapi import ( APIRouter, BackgroundTasks, Depends, HTTPException, Query, Request, Response, status, ) from pydantic import BaseModel, EmailStr, Field, field_validator from sqlalchemy import Select, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.config import get_settings from app.db import get_session from app.models import Event, Guest, Invitation, MessageJob from app.models.card import EventCardDesign from app.models.enums import ( CardDesignStatus, EventType, GuestSide, GuestSource, HistoryActor, InvitationStatus, InvitationType, Locale, PreferredChannel, ) from app.services import audit, link_preview, messaging, rsvp_service, scope from app.services.auth import CurrentAdmin, require from app.services.phone import InvalidPhoneNumberError, normalize_email, normalize_phone from app.services.policy import Action from app.services.tokens import allocate_short_code, allocate_token router = APIRouter(tags=["admin"]) MAX_PAGE_SIZE = 200 def _client_ip(request: Request) -> str | None: fwd = request.headers.get("x-forwarded-for") return fwd.split(",")[0].strip() if fwd else (request.client.host if request.client else None) class GuestWrite(BaseModel): full_name: str = Field(min_length=1, max_length=200) email: EmailStr | None = None phone: str | None = None whatsapp_phone: str | None = None side: GuestSide = GuestSide.COMMON group_tag: list[str] = Field(default_factory=list) preferred_locale: Locale = Locale.EN preferred_channel: PreferredChannel = PreferredChannel.AUTO max_guests: int = Field(default=4, ge=1, le=20) #: Chooses which greeting sentence their invitation renders. Defaults to `single`, which #: is also how every record created before this field existed behaves. invitation_type: InvitationType = InvitationType.SINGLE # No `events` list any more: a guest belongs to exactly one event (design D11) and that # event comes from the route, so it cannot be contradicted by the body. @field_validator("phone", "whatsapp_phone") @classmethod def _norm(cls, v: str | None) -> str | None: if not v or not v.strip(): return None try: return normalize_phone(v) except InvalidPhoneNumberError as exc: raise ValueError(str(exc)) from exc class InvitationSummary(BaseModel): id: uuid.UUID event_type: EventType event_title_en: str status: InvitationStatus party_size: int | None max_guests: int token: str short_code: str opened_at: datetime | None responded_at: datetime | None #: Whether the event is published. A draft event's invitation page answers 404 to every #: guest, so a link handed out now is dead — the drawer says so beside the link rather #: than letting a host paste it into WhatsApp and find out from the guest. event_published: bool class GuestRead(BaseModel): id: uuid.UUID event_id: uuid.UUID full_name: str email: str | None phone_e164: str | None whatsapp_phone_e164: str | None side: GuestSide group_tag: list[str] preferred_locale: Locale preferred_channel: PreferredChannel invitation_type: InvitationType source: GuestSource do_not_contact: bool email_invalid: bool created_at: datetime invitations: list[InvitationSummary] class GuestPage(BaseModel): items: list[GuestRead] total: int page: int page_size: int def _to_read(guest: Guest) -> GuestRead: return GuestRead( id=guest.id, event_id=guest.event_id, full_name=guest.full_name, email=guest.email, phone_e164=guest.phone_e164, whatsapp_phone_e164=guest.whatsapp_phone_e164, side=guest.side, group_tag=list(guest.group_tag or []), preferred_locale=guest.preferred_locale, preferred_channel=guest.preferred_channel, invitation_type=guest.invitation_type, source=guest.source, do_not_contact=guest.do_not_contact, email_invalid=guest.email_invalid, created_at=guest.created_at, invitations=[ InvitationSummary( id=inv.id, event_type=inv.event.type, event_title_en=inv.event.title_en, status=inv.status, party_size=inv.rsvp.party_size if inv.rsvp else None, max_guests=inv.max_guests, token=inv.token, short_code=inv.short_code, opened_at=inv.opened_at, responded_at=inv.responded_at, event_published=inv.event.is_published, ) for inv in guest.invitations ], ) @router.get("/admin/events/{event_id}/guests", response_model=GuestPage) async def list_event_guests( event_id: uuid.UUID, search: str | None = None, invitation_status: InvitationStatus | None = None, side: GuestSide | None = None, tag: str | None = None, page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=MAX_PAGE_SIZE), admin: CurrentAdmin = Depends(require(Action.VIEW_DASHBOARD)), session: AsyncSession = Depends(get_session), ) -> GuestPage: """One event's guest list — the list a host actually works from (spec event-management).""" await scope.require_event_access(session, admin, event_id) return await _guest_page( session, admin=admin, event_id=event_id, search=search, invitation_status=invitation_status, side=side, tag=tag, page=page, page_size=page_size, ) @router.get("/admin/guests", response_model=GuestPage) async def list_guests( search: str | None = None, event_id: uuid.UUID | None = None, invitation_status: InvitationStatus | None = None, side: GuestSide | None = None, tag: str | None = None, page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=MAX_PAGE_SIZE), admin: CurrentAdmin = Depends(require(Action.VIEW_DASHBOARD)), session: AsyncSession = Depends(get_session), ) -> GuestPage: """Search across every event's list (FR-4.5). Kept alongside the event-scoped route because export and global search still need it. The count it reports is of guest *records*: a person attending three ceremonies appears three times, which is what event-scoped guests mean (design D11). """ return await _guest_page( session, admin=admin, event_id=event_id, search=search, invitation_status=invitation_status, side=side, tag=tag, page=page, page_size=page_size, ) def _filtered_guests( *, admin: CurrentAdmin, event_id: uuid.UUID | None, search: str | None, invitation_status: InvitationStatus | None, side: GuestSide | None, tag: str | None, ) -> Select[tuple[Guest]]: """The guest query these filters describe, before ordering or paging. Extracted so the list and the select-all-matching id endpoint cannot drift (add-bulk-invitation-send D2). They must describe the same set: an admin who reads "412 matching" and presses select-all has to get those 412 and nobody else, and two copies of this filter chain is exactly how that stops being true. `admin` is the first parameter for the same reason: scope belongs in the one place both of those callers already share, so neither can be scoped without the other. It is applied before every filter below, so a filter can only ever narrow the caller's own set — asking for another owner's `event_id` returns nothing rather than widening it. """ stmt = select(Guest).where(Guest.is_deleted.is_(False), scope.guest_in_scope(admin)) if event_id: stmt = stmt.where(Guest.event_id == event_id) if search: like = f"%{search.strip()}%" stmt = stmt.where( or_(Guest.full_name.ilike(like), Guest.email.ilike(like), Guest.phone_e164.ilike(like)) ) if side: stmt = stmt.where(Guest.side == side) if tag: stmt = stmt.where(Guest.group_tag.contains([tag])) if invitation_status: stmt = ( stmt.join(Invitation, Invitation.guest_id == Guest.id) .where(Invitation.status == invitation_status) .distinct() ) return stmt async def _guest_page( session: AsyncSession, *, admin: CurrentAdmin, event_id: uuid.UUID | None, search: str | None, invitation_status: InvitationStatus | None, side: GuestSide | None, tag: str | None, page: int, page_size: int, ) -> GuestPage: """Shared query behind both the scoped and the global list.""" stmt = _filtered_guests( admin=admin, event_id=event_id, search=search, invitation_status=invitation_status, side=side, tag=tag, ) total = await session.scalar(select(func.count()).select_from(stmt.subquery())) or 0 rows = await session.scalars( stmt.options( selectinload(Guest.invitations).selectinload(Invitation.event), selectinload(Guest.invitations).selectinload(Invitation.rsvp), ) .order_by(Guest.full_name) .offset((page - 1) * page_size) .limit(page_size) ) return GuestPage( items=[_to_read(g) for g in rows.unique()], total=int(total), page=page, page_size=page_size, ) async def _load_guest(session: AsyncSession, guest_id: uuid.UUID, admin: CurrentAdmin) -> Guest: guest = await session.scalar( select(Guest) .where( Guest.id == guest_id, Guest.is_deleted.is_(False), scope.guest_in_scope(admin), ) .options( selectinload(Guest.invitations).selectinload(Invitation.event), selectinload(Guest.invitations).selectinload(Invitation.rsvp), ) # The callers that matter have just modified this guest in this same session, and # without populate_existing SQLAlchemy hands back the identity-map instance with # whatever collections it already had — so a guest whose invitations were created a # line ago comes back with the empty collection it was constructed with, and the # response tells the caller it has no invitations while the database disagrees. .execution_options(populate_existing=True) ) if guest is None: # Same answer whether the guest does not exist or belongs to another owner's event. raise scope.not_found() return guest @router.get("/admin/guests/{guest_id}", response_model=GuestRead) async def read_guest( guest_id: uuid.UUID, admin: CurrentAdmin = Depends(require(Action.VIEW_DASHBOARD)), session: AsyncSession = Depends(get_session), ) -> GuestRead: return _to_read(await _load_guest(session, guest_id, admin)) async def _ensure_invitation(session: AsyncSession, guest: Guest, max_guests: int) -> None: """Give the guest their one invitation, to their own event. A guest belongs to exactly one event (design D11), so there is one invitation and it is created with the guest. Never replaces an existing one — revoking a link that has already been sent is a different, deliberate action. """ if guest.invitations: return session.add( Invitation( event_id=guest.event_id, guest_id=guest.id, token=await allocate_token(session), short_code=await allocate_short_code(session), max_guests=max_guests, ) ) @router.post( "/admin/events/{event_id}/guests", response_model=GuestRead, status_code=status.HTTP_201_CREATED, ) async def create_guest( event_id: uuid.UUID, payload: GuestWrite, request: Request, admin: CurrentAdmin = Depends(require(Action.ADD_EDIT_GUESTS)), session: AsyncSession = Depends(get_session), ) -> GuestRead: """Add a guest to one event's list, with their invitation and token.""" if not payload.phone and not payload.email: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="A guest needs at least a phone number or an email address", ) event = await scope.scoped_event(session, admin, event_id) email = normalize_email(payload.email) # Scoped to this event: the same phone under a different event is a different record, # not a duplicate (design D11). duplicate = await rsvp_service.find_existing_guest( session, event_id, phone=payload.phone, email=email ) if duplicate is not None: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail=( f"{duplicate.full_name} is already on this event's guest list with that " "phone or email." ), ) guest = Guest( wedding_id=event.wedding_id, event_id=event_id, full_name=payload.full_name.strip(), email=email, phone_e164=payload.phone, whatsapp_phone_e164=payload.whatsapp_phone or payload.phone, side=payload.side, group_tag=payload.group_tag, preferred_locale=payload.preferred_locale, preferred_channel=payload.preferred_channel, invitation_type=payload.invitation_type, source=GuestSource.MANUAL, # Initialised in the constructor rather than assigned after the flush. # `guest.invitations = []` looks equivalent and is not: assigning to a relationship # makes SQLAlchemy load the current collection first so it can work out what # changed, and that lazy SELECT is illegal under asyncio. The assignment triggered # exactly the load it existed to prevent, and every guest created from the admin # died on it with MissingGreenlet. A brand-new guest has no invitations, so setting # the collection empty here is both true and load-free. invitations=[], ) session.add(guest) await session.flush() await _ensure_invitation(session, guest, payload.max_guests) await session.flush() audit.record( session, action=audit.Actions.GUEST_CREATE, admin_user_id=admin.id, actor_email=admin.email, entity_type="guest", entity_id=guest.id, after={"full_name": guest.full_name, "event": event.slug}, ip=_client_ip(request), ) return _to_read(await _load_guest(session, guest.id, admin)) @router.patch("/admin/guests/{guest_id}", response_model=GuestRead) async def update_guest( guest_id: uuid.UUID, payload: GuestWrite, request: Request, admin: CurrentAdmin = Depends(require(Action.ADD_EDIT_GUESTS)), session: AsyncSession = Depends(get_session), ) -> GuestRead: guest = await _load_guest(session, guest_id, admin) before = { "full_name": guest.full_name, "email": guest.email, "phone_e164": guest.phone_e164, "side": str(guest.side), } guest.full_name = payload.full_name.strip() guest.email = normalize_email(payload.email) guest.phone_e164 = payload.phone guest.whatsapp_phone_e164 = payload.whatsapp_phone or payload.phone guest.side = payload.side guest.group_tag = payload.group_tag guest.preferred_locale = payload.preferred_locale guest.preferred_channel = payload.preferred_channel # Changing this changes only which sentence renders next time — not their token, their # RSVP, or the headcount (spec invitation-presentation). guest.invitation_type = payload.invitation_type # The event is not editable: moving a guest between events would orphan the invitation # link already in their hand. Removing them from one list and adding them to the other # is the honest operation, and it is two deliberate steps. await _ensure_invitation(session, guest, payload.max_guests) await session.flush() after = { "full_name": guest.full_name, "email": guest.email, "phone_e164": guest.phone_e164, "side": str(guest.side), } changed_before, changed_after = audit.diff(before, after) audit.record( session, action=audit.Actions.GUEST_UPDATE, admin_user_id=admin.id, actor_email=admin.email, entity_type="guest", entity_id=guest.id, before=changed_before, after=changed_after, ip=_client_ip(request), ) return _to_read(await _load_guest(session, guest_id, admin)) @router.delete("/admin/guests/{guest_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_guest( guest_id: uuid.UUID, request: Request, admin: CurrentAdmin = Depends(require(Action.DELETE_GUESTS)), session: AsyncSession = Depends(get_session), ) -> None: """Soft delete — Super Admin only. History survives for the audit trail.""" guest = await _load_guest(session, guest_id, admin) guest.is_deleted = True audit.record( session, action=audit.Actions.GUEST_DELETE, admin_user_id=admin.id, actor_email=admin.email, entity_type="guest", entity_id=guest.id, before={"full_name": guest.full_name}, ip=_client_ip(request), ) class ManualRsvpRequest(BaseModel): """A phone RSVP recorded on the guest's behalf (FR-4.6).""" response: Literal["accepted", "declined"] party_size: int = Field(default=1, ge=1) note: str | None = None @router.post("/admin/invitations/{invitation_id}/rsvp", response_model=InvitationSummary) async def override_rsvp( invitation_id: uuid.UUID, payload: ManualRsvpRequest, request: Request, admin: CurrentAdmin = Depends(require(Action.ADD_EDIT_GUESTS)), session: AsyncSession = Depends(get_session), ) -> InvitationSummary: invitation = await session.scalar( select(Invitation) .where(Invitation.id == invitation_id, scope.invitation_in_scope(admin)) .options(selectinload(Invitation.guest), selectinload(Invitation.event)) ) if invitation is None: raise scope.not_found() guest = invitation.guest if payload.response == "accepted": rsvp = await rsvp_service.accept( session, invitation, name=guest.full_name if guest else "Guest", phone=guest.phone_e164 if guest else None, email=guest.email if guest else None, party_size=payload.party_size, dietary_notes=payload.note, actor=HistoryActor.ADMIN, actor_admin_id=admin.id, source="admin_manual", ) else: rsvp = await rsvp_service.decline( session, invitation, reason=payload.note, actor=HistoryActor.ADMIN, actor_admin_id=admin.id, ) audit.record( session, action=audit.Actions.RSVP_OVERRIDE, admin_user_id=admin.id, actor_email=admin.email, entity_type="invitation", entity_id=invitation.id, after={"response": payload.response, "party_size": payload.party_size}, ip=_client_ip(request), ) await session.flush() return InvitationSummary( id=invitation.id, event_type=invitation.event.type, event_title_en=invitation.event.title_en, status=invitation.status, party_size=rsvp.party_size if rsvp else None, max_guests=invitation.max_guests, token=invitation.token, short_code=invitation.short_code, opened_at=invitation.opened_at, responded_at=invitation.responded_at, event_published=invitation.event.is_published, ) # --------------------------------------- manual invitation send (add-guest-invitation-send) # # Inviting one named guest is the most frequent thing a host does, and until now it happened # outside the product: copy the link, paste it into WhatsApp, and nothing recorded that it # went. These two routes are that action — compose on the server (D4), send through the same # queue every wave uses (D1). class AdminPreviewRead(BaseModel): """The link preview as this one guest will see it, for the compose panel (design D5). A separate model from the guest routes' `LinkPreviewRead` rather than that one with a name bolted on. `LinkPreviewRead` exists to be guest-free — it is what ends up in meta tags a crawler caches — and a nullable `guest_name` on it would be one careless assignment away from putting a guest's name there. Two types make the mistake unrepresentable instead of merely discouraged (design D3). """ title: str description: str guest_name: str image_url: str | None #: None when the event has no published card, or has one published without a picture. #: The panel then shows the text-only preview and says what to do about it. has_image: bool class ComposedMessageRead(BaseModel): """The default message, in the parts the admin edits.""" subject: str header: str body: str footer: str invite_url: str #: What the block in the guest's email — and any chat application the link is pasted #: into — will show. Rendered from the same model both of those use. preview: AdminPreviewRead to_email: str | None event_title: str #: When this invitation was last emailed, so a duplicate is visible before it is sent (D8). last_sent_at: datetime | None #: Why the send control must be disabled, in words an admin can act on. None means send. blocked_reason: str | None #: Whether sending right now would land inside quiet hours, and the local clock time it #: would land at. The panel warns on this; the server checks again at send time, because #: this answer goes stale and the browser's clock is not Asia/Dhaka. in_quiet_hours: bool local_time: str #: False while the event is a draft. Every link in this message would answer 404, so the #: panel says so before the admin writes anything and the send route refuses outright. event_published: bool def _require_published(event: Event) -> None: """Refuse to send invitations to a draft event. `/i/{token}` and `/e/{slug}` both 404 while `is_published` is false — deliberately, so a half-filled event cannot be walked into. The consequence nothing used to state is that a perfectly valid invitation link is *dead* until the event is published, and a send is the one action that puts that link somewhere it cannot be taken back from. So this is a refusal rather than a warning, and it lives here rather than in the browser: the compose panels carry `event_published` to explain it up front, but an event unpublished between opening the composer and pressing Send has to be caught by the server (D1). """ if event.is_published: return raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={ "code": "event_unpublished", "message": ( f"{event.title_en} is still a draft, so its invitation links answer " f"'page not found'. Publish the event, then send." ), }, ) async def _load_invitation_for_send( session: AsyncSession, invitation_id: uuid.UUID, admin: CurrentAdmin ) -> Invitation: invitation = await session.scalar( select(Invitation) .where(Invitation.id == invitation_id, scope.invitation_in_scope(admin)) .options( selectinload(Invitation.guest), selectinload(Invitation.event).selectinload(Event.wedding), ) ) if invitation is None: raise scope.not_found() return invitation def _quiet_hours_now() -> tuple[bool, str]: """Whether now is inside the quiet window, and the local clock time.""" settings = get_settings() now = datetime.now(tz=UTC) inside = messaging.in_quiet_hours( now, settings.timezone, settings.quiet_hours_start, settings.quiet_hours_end ) return inside, now.astimezone(ZoneInfo(settings.timezone)).strftime("%H:%M") @router.get("/admin/invitations/{invitation_id}/message", response_model=ComposedMessageRead) async def compose_invitation_message( invitation_id: uuid.UUID, admin: CurrentAdmin = Depends(require(Action.SEND_MESSAGES)), session: AsyncSession = Depends(get_session), ) -> ComposedMessageRead: """The message this guest would receive, ready to be edited (D4).""" invitation = await _load_invitation_for_send(session, invitation_id, admin) composed = messaging.compose_manual(invitation) inside, local_time = _quiet_hours_now() design = await session.scalar( select(EventCardDesign).where( EventCardDesign.event_id == invitation.event_id, EventCardDesign.status == CardDesignStatus.PUBLISHED, ) ) built = link_preview.for_guest( invitation.event, invitation.event.wedding, guest_name=(invitation.guest.full_name if invitation.guest else ""), base_url=get_settings().app_base_url, card_config=design.config if design is not None else None, ) return ComposedMessageRead( subject=composed.subject, header=composed.header, body=composed.body, footer=composed.footer, invite_url=composed.invite_url, preview=AdminPreviewRead( title=built.title, description=built.description, guest_name=built.guest_name or "", image_url=built.image.url if built.image else None, has_image=built.image is not None, ), to_email=composed.to_email, event_title=composed.event_title, last_sent_at=await messaging.last_manual_send_at(session, invitation.id), blocked_reason=messaging.block_reason(invitation.guest), in_quiet_hours=inside, local_time=local_time, event_published=invitation.event.is_published, ) @router.post( "/admin/invitations/{invitation_id}/link-copied", status_code=status.HTTP_204_NO_CONTENT, ) async def record_link_copy( invitation_id: uuid.UUID, request: Request, admin: CurrentAdmin = Depends(require(Action.SEND_MESSAGES)), session: AsyncSession = Depends(get_session), ) -> Response: """Record that an admin took this invitation's link to deliver by hand (design D8). Writes an audit entry and nothing else: no `message_job`, no `message_log`, no `last_sent_at`. Copying a link is not a send, and recording it as one would put a delivery in the host's history that never happened. Worth being honest about what this does and does not prove. The compose panel already received `invite_url` when it loaded, so the token was disclosed before this fires — what is recorded here is the admin's *intent* to hand the link out, not the moment of disclosure. That is still the useful question afterwards ("who gave this out, and when"), but it is not a containment boundary and should not be mistaken for one. Deliberately permitted for a suppressed guest. Suppression governs what the system sends, not what a host may hand someone in person; the panel states the suppression above the control so the admin decides knowingly. """ invitation = await _load_invitation_for_send(session, invitation_id, admin) audit.record( session, action=audit.Actions.INVITATION_LINK_COPY, admin_user_id=admin.id, actor_email=admin.email, entity_type="invitation", entity_id=invitation.id, # No token, in full or in part: this row is read by people, and the log is not a # place a bearer secret belongs (design D11). after={ "event_id": str(invitation.event_id), "suppressed": messaging.block_reason(invitation.guest) is not None, }, ip=_client_ip(request), ) await session.commit() return Response(status_code=status.HTTP_204_NO_CONTENT) class SendInvitationRequest(BaseModel): subject: str header: str body: str footer: str #: Set by the client only after the admin acknowledged the quiet-hours warning (D5). confirmed_quiet_hours: bool = False class SendInvitationResult(BaseModel): job_id: uuid.UUID #: The job's terminal state — `sent`, `failed` or `skipped`. Never `queued`: an admin #: watching a button press needs the outcome, not a receipt (D7). status: str sent_at: datetime | None error_message: str | None skip_reason: str | None #: True when nothing left the building. Reported plainly rather than dressed as a send. dry_run: bool @router.post("/admin/invitations/{invitation_id}/send", response_model=SendInvitationResult) async def send_invitation_email( invitation_id: uuid.UUID, payload: SendInvitationRequest, request: Request, admin: CurrentAdmin = Depends(require(Action.SEND_MESSAGES)), session: AsyncSession = Depends(get_session), ) -> SendInvitationResult: """Send one hand-edited invitation email, and report what actually happened.""" settings = get_settings() invitation = await _load_invitation_for_send(session, invitation_id, admin) subject = payload.subject.strip() body = messaging.assemble(payload.header, payload.body, payload.footer) if not subject or not body: missing = "a subject" if not subject else "a body" if not subject and not body: missing = "both a subject and a body" raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={"code": "empty_message", "message": f"A message needs {missing}."}, ) # Consent first, and no confirmation can get past it (D6). Checked here so the admin gets # a reason rather than a job that is silently skipped forty seconds later; `process_job` # checks again, which is what catches an unsubscribe landing mid-compose. blocked = messaging.block_reason(invitation.guest) if blocked: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={"code": "suppressed", "message": blocked}, ) # Every link in this message would land on a 404 until the event is published. _require_published(invitation.event) # Nothing is configured yet, and a stack trace from a vendor is not an answer to a button # press. Checked before inserting so a misconfigured deployment does not accumulate jobs. if not settings.dry_run and not settings.email_provider_api_key: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail={ "code": "not_configured", "message": "Email sending is not configured: EMAIL_PROVIDER_API_KEY is not set.", }, ) inside, local_time = _quiet_hours_now() if inside and not payload.confirmed_quiet_hours: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={ "code": "quiet_hours", "message": ( f"It is {local_time} for your guests, inside quiet hours " f"({settings.quiet_hours_start}-{settings.quiet_hours_end}). Send anyway?" ), "local_time": local_time, }, ) job_id = await messaging.enqueue( session, invitation_id=invitation.id, template_id=None, schedule_id=None, scheduled_for=datetime.now(tz=UTC), subject=subject, body_text=body, override_quiet_hours=inside, sent_by_admin_id=admin.id, key=messaging.manual_idempotency_key(invitation.id), ) if job_id is None: # pragma: no cover — the key carries 8 random bytes raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={"code": "duplicate", "message": "That send was already recorded."}, ) # Committed before the provider call so a crash mid-send leaves a claimable row rather # than nothing — the worker's stuck-job sweep is the safety net, exactly as for a wave. await session.commit() job = await session.get(MessageJob, job_id) if job is None: # pragma: no cover — it was committed one statement ago raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="The send was not recorded" ) # `audience_accepted=False`: an invitation is being sent, so the guest has not answered # yet — requiring `accepted` here would skip every real invitation. outcome = await messaging.process_job(session, job, audience_accepted=False) audit.record( session, action=audit.Actions.MESSAGE_SEND, admin_user_id=admin.id, actor_email=admin.email, entity_type="invitation", entity_id=invitation.id, after={ "manual": True, "event": invitation.event.title_en, "outcome": outcome, "job_id": str(job_id), }, ip=_client_ip(request), ) await session.flush() return SendInvitationResult( job_id=job.id, status=str(job.status), sent_at=job.sent_at, error_message=job.error_message, skip_reason=job.skip_reason, dry_run=settings.dry_run, ) # ------------------------------------------------------------------ batched invitation send # # The same act as the panel above, for a chosen set of guests instead of one # (add-bulk-invitation-send). What is different, and what these four routes exist to hold # correct: the message is authored once per invitation type and personalised per recipient # (D3), the selection is explicit and confined to one event (D1, D2), the batch id is the # duplicate boundary (D5), and recording is decoupled from delivery so 1,500 guests do not # have to fit inside one request (D6). #: A whole guest list is ~1,500. The cap is above any real list and below the point where a #: request stops being interactive; it refuses rather than truncating, because a silently #: short selection sends to some of the people the admin asked for and reports success. MAX_BATCH_GUESTS = 2000 class GuestIdsRead(BaseModel): """Every guest matching a set of filters, for select-all-matching (D2).""" guest_ids: list[uuid.UUID] total: int @router.get("/admin/events/{event_id}/guest-ids", response_model=GuestIdsRead) async def list_event_guest_ids( event_id: uuid.UUID, search: str | None = None, invitation_status: InvitationStatus | None = None, side: GuestSide | None = None, tag: str | None = None, admin: CurrentAdmin = Depends(require(Action.SEND_MESSAGES)), session: AsyncSession = Depends(get_session), ) -> GuestIdsRead: """Resolve "select all N matching" into the ids it means (D2). The selection is resolved here, once, rather than re-evaluated at send time: a filter-shaped selection can grow between the count on the button and the messages going out — an import finishing in another tab is enough — and the set the admin approved is the set that must be sent. """ await scope.require_event_access(session, admin, event_id) stmt = _filtered_guests( admin=admin, event_id=event_id, search=search, invitation_status=invitation_status, side=side, tag=tag, ) total = await session.scalar(select(func.count()).select_from(stmt.subquery())) or 0 if total > MAX_BATCH_GUESTS: raise HTTPException( status_code=status.HTTP_413_REQUEST_ENTITY_TOO_LARGE, detail={ "code": "selection_too_large", "message": ( f"That filter matches {total} guests, more than the {MAX_BATCH_GUESTS} " f"a single send can cover. Narrow it and send in more than one go." ), }, ) rows = await session.scalars(stmt.with_only_columns(Guest.id).order_by(None)) ids = list(rows) return GuestIdsRead(guest_ids=ids, total=len(ids)) class ExclusionRead(BaseModel): """One reason some of the selection will not be emailed, and how many (D10).""" code: str reason: str count: int class ComposedPaneRead(BaseModel): """One pane: the message for one invitation type in one locale, with placeholders intact.""" locale: Locale invitation_type: InvitationType subject: str header: str body: str footer: str #: How many of the selected guests this pane will be sent to. Zero is a real answer, and #: the pane is still returned so the composer can show it inert rather than hide it. recipient_count: int class BulkComposeRequest(BaseModel): guest_ids: list[uuid.UUID] = Field(min_length=1, max_length=MAX_BATCH_GUESTS) class BulkComposeRead(BaseModel): event_title: str #: One pair per locale present in the selection (D11) — both invitation types, always. panes: list[ComposedPaneRead] selected: int deliverable: int exclusions: list[ExclusionRead] #: Selected guests whose record no longer resolves — deleted since the list was drawn. missing: int #: The placeholders the composed text may carry, so the composer can explain them (D4). placeholders: list[str] in_quiet_hours: bool local_time: str #: False while the event is a draft. Every `{invitation_link}` this batch substitutes #: would answer 404, so the composer says so before the admin writes to 300 people. event_published: bool async def _load_event_for_send( session: AsyncSession, event_id: uuid.UUID, admin: CurrentAdmin ) -> Event: event = await session.scalar( select(Event) .where(Event.id == event_id, scope.visible_events(admin)) .options(selectinload(Event.wedding)) ) if event is None: raise scope.not_found() return event async def _resolve_selection( session: AsyncSession, event: Event, guest_ids: list[uuid.UUID] ) -> tuple[list[Invitation], int]: """The invitations a selection names, and how many of it no longer exists (D1). Two failures that look alike from the outside are told apart here, because the right answer differs. A guest belonging to *another* event is a contradiction — the route says which event this is, and honouring the body instead would mail one event's wording to another event's list — so the whole request is refused. A guest who simply no longer resolves was deleted between drawing the list and pressing Send; refusing there would block a send the admin can do nothing to fix, so they are counted and reported instead. """ known = { row.id: row for row in await session.execute( select(Guest.id, Guest.event_id, Guest.is_deleted).where(Guest.id.in_(guest_ids)) ) } foreign = [gid for gid, row in known.items() if row.event_id != event.id] if foreign: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={ "code": "foreign_guest", "message": ( f"{len(foreign)} of the selected guests belong to a different event. " f"A guest belongs to exactly one event, and the message text belongs " f"to this one." ), }, ) live = [gid for gid in guest_ids if gid in known and not known[gid].is_deleted] missing = len(guest_ids) - len(live) if not live: return [], missing rows = await session.scalars( select(Invitation) .where(Invitation.event_id == event.id, Invitation.guest_id.in_(live)) .options( selectinload(Invitation.guest), selectinload(Invitation.event).selectinload(Event.wedding), ) ) invitations = list(rows) # A live guest with no invitation is not a state `create_guest` can produce, but counting # it beats reporting a send to more people than were mailed. return invitations, missing + (len(live) - len(invitations)) def _exclusions(counts: dict[str, int]) -> list[ExclusionRead]: """Suppression counts in a stable order, each with the sentence that explains it.""" wording = { messaging.BLOCK_CODE_NO_EMAIL: messaging.BLOCK_NO_EMAIL, messaging.BLOCK_CODE_OPTED_OUT: messaging.BLOCK_OPTED_OUT, messaging.BLOCK_CODE_EMAIL_INVALID: messaging.BLOCK_EMAIL_INVALID, } return [ ExclusionRead(code=code, reason=reason, count=counts[code]) for code, reason in wording.items() if counts.get(code) ] @router.post("/admin/events/{event_id}/invitations/compose", response_model=BulkComposeRead) async def compose_bulk_invitation( event_id: uuid.UUID, payload: BulkComposeRequest, admin: CurrentAdmin = Depends(require(Action.SEND_MESSAGES)), session: AsyncSession = Depends(get_session), ) -> BulkComposeRead: """The two messages this selection would receive, ready to be edited (D1, D10, D11).""" event = await _load_event_for_send(session, event_id, admin) invitations, missing = await _resolve_selection(session, event, payload.guest_ids) locales: list[Locale] = [] recipients: dict[tuple[Locale, InvitationType], int] = {} excluded: dict[str, int] = {} for invitation in invitations: guest = invitation.guest blocked = messaging.suppression(guest) if blocked is not None or guest is None: code = blocked.code if blocked else messaging.BLOCK_CODE_NO_EMAIL excluded[code] = excluded.get(code, 0) + 1 continue if guest.preferred_locale not in locales: locales.append(guest.preferred_locale) key = (guest.preferred_locale, guest.invitation_type) recipients[key] = recipients.get(key, 0) + 1 # Every selected guest suppressed still needs a composer to open — it has to be able to # say who was excluded and why. English is the fallback pair in that case, and it sends # to nobody either way. if not locales: locales = [Locale.EN] panes = [ ComposedPaneRead( locale=locale, invitation_type=invitation_type, recipient_count=recipients.get((locale, invitation_type), 0), **vars( messaging.compose_bulk( event, event.wedding, locale=locale, invitation_type=invitation_type ) ), ) for locale in locales for invitation_type in (InvitationType.SINGLE, InvitationType.FAMILY) ] inside, local_time = _quiet_hours_now() return BulkComposeRead( event_title=event.title_en, panes=panes, selected=len(payload.guest_ids), deliverable=sum(recipients.values()), exclusions=_exclusions(excluded), missing=missing, placeholders=sorted(messaging.PLACEHOLDERS), in_quiet_hours=inside, local_time=local_time, event_published=event.is_published, ) class BulkPaneWrite(BaseModel): """One pane as the admin edited it.""" locale: Locale invitation_type: InvitationType subject: str header: str body: str footer: str class SendBatchRequest(BaseModel): #: Generated by the browser when the composer opens, and reused for every attempt from #: it — which is what makes a double-clicked Send insert nothing twice (D5). batch_id: uuid.UUID guest_ids: list[uuid.UUID] = Field(min_length=1, max_length=MAX_BATCH_GUESTS) panes: list[BulkPaneWrite] #: Set only after the admin acknowledged the quiet-hours warning (D9). confirmed_quiet_hours: bool = False class SendBatchResult(BaseModel): batch_id: uuid.UUID queued: int #: True when this batch id was already recorded — a repeated request, not a second send. already_recorded: bool exclusions: list[ExclusionRead] missing: int #: True when nothing will leave the building. Reported plainly rather than as a send. dry_run: bool @router.post("/admin/events/{event_id}/invitations/send-batch", response_model=SendBatchResult) async def send_invitation_batch( event_id: uuid.UUID, payload: SendBatchRequest, request: Request, background: BackgroundTasks, admin: CurrentAdmin = Depends(require(Action.SEND_MESSAGES)), session: AsyncSession = Depends(get_session), ) -> SendBatchResult: """Record one batched send and start it, without waiting for it to finish (D6).""" settings = get_settings() event = await _load_event_for_send(session, event_id, admin) # Before anything else, including the quiet-hours question: a resubmitted request must # report the batch that already exists rather than being refused for a decision the admin # already made (D5). existing = ( await session.scalar( select(func.count(MessageJob.id)).where(MessageJob.batch_id == payload.batch_id) ) ) or 0 if existing: return SendBatchResult( batch_id=payload.batch_id, queued=int(existing), already_recorded=True, exclusions=[], missing=0, dry_run=settings.dry_run, ) # After the idempotency check and before anything is composed or counted: a batch already # recorded still reports itself even if the event has since been unpublished, but a new # one cannot be started while every link it would carry is dead. _require_published(event) invitations, missing = await _resolve_selection(session, event, payload.guest_ids) # Which panes actually matter. A pane with no recipients is ignored entirely — refusing a # send because the unused family pane has a typo in it would be obstructive, and its text # reaches nobody. needed: set[tuple[Locale, InvitationType]] = set() for invitation in invitations: guest = invitation.guest if guest is not None and messaging.suppression(guest) is None: needed.add((guest.preferred_locale, guest.invitation_type)) if not needed: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={ "code": "nobody_to_send_to", "message": "None of the selected guests can be emailed, so nothing was sent.", }, ) templates: dict[tuple[Locale, InvitationType], messaging.ComposedTemplate] = {} for pane in payload.panes: key = (pane.locale, pane.invitation_type) if key not in needed: continue subject, body = pane.subject.strip(), pane.body.strip() if not subject or not body: missing_part = "a subject" if not subject else "a body" if not subject and not body: missing_part = "both a subject and a body" raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={ "code": "empty_message", "message": f"The {pane.invitation_type} message needs {missing_part}.", "locale": str(pane.locale), "invitation_type": str(pane.invitation_type), }, ) template = messaging.ComposedTemplate( subject=subject, header=pane.header, body=pane.body, footer=pane.footer ) try: messaging.validate_template(template) except messaging.TemplateError as exc: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={ "code": "bad_placeholder", "message": str(exc), "locale": str(pane.locale), "invitation_type": str(pane.invitation_type), }, ) from exc templates[key] = template absent = needed - set(templates) if absent: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail={ "code": "missing_pane", "message": ( "Some selected guests have no message composed for their language and " "invitation type." ), }, ) # Nothing is configured yet, and a vendor stack trace is not an answer to a button press. # Checked before inserting so a misconfigured deployment does not accumulate 300 jobs. if not settings.dry_run and not settings.email_provider_api_key: raise HTTPException( status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail={ "code": "not_configured", "message": "Email sending is not configured: EMAIL_PROVIDER_API_KEY is not set.", }, ) # One decision for the whole batch (D9). Asked once, and stamped on every row so a job the # worker picks up at 23:50 still honours what the admin was told. inside, local_time = _quiet_hours_now() if inside and not payload.confirmed_quiet_hours: raise HTTPException( status_code=status.HTTP_409_CONFLICT, detail={ "code": "quiet_hours", "message": ( f"It is {local_time} for your guests, inside quiet hours " f"({settings.quiet_hours_start}-{settings.quiet_hours_end}). " f"Send to all of them anyway?" ), "local_time": local_time, }, ) result = await messaging.enqueue_batch( session, invitations=invitations, templates=templates, batch_id=payload.batch_id, override_quiet_hours=inside, sent_by_admin_id=admin.id, ) # One row for the whole send, not one per recipient: 300 audit entries for one click is # noise that buries the entry someone will actually go looking for. audit.record( session, action=audit.Actions.MESSAGE_SEND, admin_user_id=admin.id, actor_email=admin.email, entity_type="event", entity_id=event.id, after={ "batch": True, "batch_id": str(payload.batch_id), "event": event.title_en, "selected": len(payload.guest_ids), "queued": result.queued, "excluded": result.excluded, "override_quiet_hours": inside, }, ip=_client_ip(request), ) # Committed before the sending starts, so a crash mid-batch leaves claimable rows rather # than nothing — the worker's poll is the safety net, exactly as it is for a wave. await session.commit() background.add_task(messaging.dispatch_batch, payload.batch_id) return SendBatchResult( batch_id=payload.batch_id, queued=result.queued, already_recorded=False, exclusions=_exclusions(result.excluded), missing=missing, dry_run=settings.dry_run, ) class BatchProblemRead(BaseModel): guest_name: str status: str reason: str class BatchProgressRead(BaseModel): batch_id: uuid.UUID total: int waiting: int sent: int failed: int skipped: int finished: bool problems: list[BatchProblemRead] #: True when there are more failures than `problems` lists. The counts are always exact. problems_capped: bool dry_run: bool @router.get("/admin/send-batches/{batch_id}", response_model=BatchProgressRead) async def read_batch_progress( batch_id: uuid.UUID, admin: CurrentAdmin = Depends(require(Action.SEND_MESSAGES)), session: AsyncSession = Depends(get_session), ) -> BatchProgressRead: """How a batch is getting on (D7). Polled while anything is still waiting. An unknown batch reads as an empty, finished one rather than a 404: a batch is a set of jobs, not a row, so "no jobs" is a truthful answer and the composer has nothing useful to do with an error here. A batch id comes from the browser, so one host could poll another's — and `problems` carries guest *names*. Scoped for that reason: a batch with no jobs the caller may see reads exactly like a batch id that was never used, which is the same answer this endpoint already gives for one that does not exist. """ reachable = await session.scalar( select(MessageJob.id) .join(Invitation, Invitation.id == MessageJob.invitation_id) .where(MessageJob.batch_id == batch_id, scope.invitation_in_scope(admin)) .limit(1) ) if reachable is None: return BatchProgressRead( batch_id=batch_id, total=0, waiting=0, sent=0, failed=0, skipped=0, finished=True, problems=[], problems_capped=False, dry_run=get_settings().dry_run, ) progress = await messaging.batch_progress(session, batch_id) return BatchProgressRead( batch_id=batch_id, total=progress.total, waiting=progress.waiting, sent=progress.sent, failed=progress.failed, skipped=progress.skipped, finished=progress.finished, problems=[ BatchProblemRead(guest_name=p.guest_name, status=p.status, reason=p.reason) for p in progress.problems ], problems_capped=len(progress.problems) >= messaging.BATCH_PROBLEM_LIMIT, dry_run=get_settings().dry_run, )