Resend broadcasts: segment sync, one-call mass email, batch sends, contact pruning

CLAUDE.md

@@ -154,7 +154,19 @@ Two scripts live in `scripts/` for diagnosing and backfilling team rosters on `/
 - **Confirmation email tracking**: `send_volunteer_confirmation_email` now accepts `volunteer_id` and appends a `sent_emails` record with `recipient_type: 'application_confirmation'` after sending.
 
 ## Resend audience sync
-`scripts/sync_resend_audience.py --source {all|profiles|volunteers|mentors|judges|sponsors|helpers|leads} --audience "<name>" [--event-id <id>] [--selected-only] [--apply]` — pulls emails from Firestore (`users.email_address`, `volunteers.email` filtered by `volunteer_type`, `leads.email`) and upserts contacts into a Resend audience (creates if missing). Dry-run by default. Re-runnable: lists existing audience contacts first and only POSTs new emails. Needs `RESEND_API_KEY` with audiences scope — the existing `RESEND_WELCOME_EMAIL_KEY` is send-only and will 401. Uses the deprecated `resend.Audiences` SDK class (now an alias for Segments) — fine for now, but if it breaks switch to `resend.Segments`.
+`scripts/sync_resend_audience.py --source {all|profiles|volunteers|mentors|judges|sponsors|helpers|leads} --audience "<name>" [--event-id <id>] [--selected-only] [--apply]` — pulls emails from Firestore (`users.email_address`, `volunteers.email` filtered by `volunteer_type`, `leads.email`) and upserts contacts into a Resend audience (creates if missing). Dry-run by default. Re-runnable: lists existing audience contacts first and only POSTs new emails. Needs `RESEND_API_KEY` with audiences scope — the existing `RESEND_WELCOME_EMAIL_KEY` is send-only and will 401. Uses the deprecated `resend.Audiences` SDK class (now an alias for Segments) — fine for now, but if it breaks switch to `resend.Segments`. **This logic is now ALSO ported into `services/broadcasts_service.py` (below) for the admin UI — keep loader/dedupe changes in sync or (better) treat the script as the ad-hoc CLI and the service as the source of truth.**
+
+## Broadcasts blueprint (`api/broadcasts/` + `services/broadcasts_service.py`, Aug 2026)
+Powers the `/admin/communication?tab=email` Broadcast mode + the personalized bulk path. All routes `volunteer.admin`-gated, email_templates-style thin views. Routes under `/api/admin/broadcasts`: `GET segments` (list Resend segments), `POST preview` (dry-run per-source counts, no Resend writes), `POST segments/sync` (start background contact sync; 202 or 409 `already_running`), `GET segments/<id>/sync-status` (poll), `GET|POST ''` (list / create broadcast — draft by default, `send:true`/`scheduled_at` to send), `GET <id>`, `POST <id>/send`, `POST batch-send` (transactional Resend Batch, see below). Load-bearing details:
+- **Sources spec** consumed by preview+sync: `{"sources":[{"type":"profiles"|"leads"|"volunteers"|"slack"|"contact_submissions", ...}], "custom_emails":[...]}` — volunteers takes `volunteer_type`/`event_id`/`selected_only`; slack takes `active_days` (365 default / 10000 = everyone) and reuses `get_active_users(days, admin=True)` (deleted/bot/restricted already excluded there); contact_submissions takes `inquiry_types` (list, case-insensitive match on the doc's `inquiryType`; empty = all) + `updates_opt_in_only` (the form's `receiveUpdates` box). `collect_contacts` dedupes by lowercase email (first source wins the record; later sources fill blank names) and returns stats `{per_source, custom_valid, custom_invalid, union_total, overlap_removed, contact_limit, over_limit}`.
+- **Segment sync is a daemon thread + redis** (never inline — thousands of `Contacts.create` at ~20/s vs the 120s gunicorn timeout). Keys `broadcasts:sync:{segment_id}:status` (TTL 24h) / `:lock` (TTL 30min — self-heals worker death). Heartbeat every 25 contacts; `get_sync_status` reports `state:"stalled"` when a "running" status hasn't been touched for 120s (retry is safe — the sync diffs against `_existing_segment_emails` first, so it's idempotent). Contacts are **CREATE-only** — never update existing, never re-subscribe an unsubscribed contact.
+- **API keys:** segment/contact/broadcast ops use `_resend_full_key()` = `RESEND_API_KEY` with NO welcome-key fallback (it 401s). Batch send uses `RESEND_WELCOME_EMAIL_KEY` (Emails scope). `resend.api_key` is a module-global shared across threads — set it immediately before each call section.
+- **From-address allowlist:** `RESEND_BROADCAST_FROM` (default `Opportunity Hack <updates@notify.ohack.dev>`) + `RESEND_BROADCAST_FROM_DOMAINS` (default `notify.ohack.dev,apply.ohack.dev`). `notifs.ohack.org` is deliberately NOT allowlisted — its Resend domain verification is `partially_failed` (transactional sends still hardcode it; fix the DNS or migrate separately).
+- **Contact-cap guardrail:** `RESEND_MARKETING_CONTACT_LIMIT` (default 1000 = the free marketing tier OHack is on as of Aug 2026; Resend bills marketing by CONTACT COUNT, not sends — 5k=$40/mo, 10k=$80/mo). Preview/sync report `over_limit`; sync proceeds and captures per-contact failures. Broadcast HTML gets a `{{{RESEND_UNSUBSCRIBE_URL}}}` footer appended server-side if missing (Resend rejects broadcasts without it).
+- **`POST batch-send`** (`{subject, recipient_type?, recipients:[{email,name,message}]}`, ≤500/request): renders each pre-personalized message through the same HTML shell as `_send_email_to_user` and sends via `resend.Batch.send` in chunks of 100 (transactional quota — this replaced the frontend's one-request-per-recipient loop for email-only recipients). `[QRCode:...]` messages are rejected per-recipient (Batch has no attachments); registered-user sends stay on `/api/admin/<id>/message` (Slack DM side effect). Falls back to sequential `Emails.send` when the SDK predates Batch (local env note: requirements pins resend 2.22.0 but the conda env had 2.3.0 — `pip install -U resend==2.22.0`).
+- **Contact management (quota reclaim):** `GET /admin/broadcasts/contacts` (full account-level crawl, 60s redis cache `broadcasts:contacts:index`, `?force=true`; returns `contacts/total/unsubscribed_count/contact_limit/over_limit`), `POST /admin/broadcasts/contacts/prune` (modes `unsubscribed|emails|all`; ONE global background job, lock `broadcasts:contacts:prune:lock` + status `...:prune:status`, same stall/heartbeat semantics as sync), `GET .../prune-status`. Deletes use `resend.Contacts.remove(email=...)` with NO `audience_id` → removes the **GLOBAL** contact, which is what frees marketing quota (unsubscribed contacts still count against it). Sync + prune both bust the contacts cache on completion.
+- All sends + sync completions are gated by a local `_notifications_disabled()` (ENVIRONMENT=test → `simulated:true`) and audited via `send_slack_audit`. Tests: `api/broadcasts/tests/test_broadcasts_service.py`.
+- **`userlist()` in `common/utils/slack.py` is now redis-cached** (`slack:userlist`, TTL 600s, decorator ABOVE the RateLimiter so cache hits skip the blocking limiter) — one crawl serves every `active_days` filter; `clear_slack_cache()` clears the new prefix too. All userlist consumers now see up-to-10-min-stale member data (fine — the "activity" field only changes on profile updates).
 
 The frontend `/hack/<event_id>` page's "Team Members:" list is `teams.users[]` (DocumentReferences). The bug pattern that motivated this: a team's `users[]` only contains the user who created the team on ohack.dev; everyone else registered via Devpost/JotForm and was never linked. Use `audit` first to confirm, then `import ... --csv-type roster` (or `projects` for old Devpost exports) to backfill.
 

api/__init__.py

@@ -191,6 +191,7 @@ def add_headers(response):
     from api.feedback import feedback_views
     from api.praisebot import praisebot_views
     from api.jobs import jobs_views
+    from api.broadcasts import broadcasts_views
 
     app.register_blueprint(messages_views.bp)
     app.register_blueprint(exception_views.bp)
@@ -217,5 +218,6 @@ def add_headers(response):
     app.register_blueprint(feedback_views.bp)
     app.register_blueprint(praisebot_views.bp)
     app.register_blueprint(jobs_views.bp)
+    app.register_blueprint(broadcasts_views.bp)
 
     return app

api/broadcasts/broadcasts_views.py

@@ -0,0 +1,149 @@
+"""Admin Resend segment/broadcast/batch-send endpoints.
+
+Powers the /admin/communication Email tab: preview recipient sources
+(profiles/volunteers/leads/slack/custom), sync them into a Resend segment
+(background thread + polled status), create/send broadcasts, and the
+transactional batch-send used by the personalized bulk path.
+
+All routes are volunteer.admin-gated. Logic lives in
+services/broadcasts_service.py.
+"""
+
+from flask import Blueprint, request
+
+from common.log import get_logger
+from common.auth import auth, auth_user, getOrgId
+from services.broadcasts_service import (
+    batch_send_emails,
+    create_broadcast,
+    get_broadcast,
+    get_prune_status,
+    get_sync_status,
+    list_broadcasts,
+    list_contacts,
+    list_segments,
+    preview_sources,
+    send_broadcast,
+    start_contact_prune,
+    start_segment_sync,
+)
+
+logger = get_logger(__name__)
+
+bp = Blueprint("broadcasts", __name__, url_prefix="/api")
+
+
+def _actor_from_request():
+    try:
+        return {
+            "propel_user_id": auth_user.user_id if auth_user else None,
+            "email": getattr(auth_user, "email", None) if auth_user else None,
+        }
+    except Exception:
+        return None
+
+
+@bp.route("/admin/broadcasts/segments", methods=["GET"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_list_segments():
+    logger.info("GET /admin/broadcasts/segments called")
+    msg, status_code = list_segments()
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts/preview", methods=["POST"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_preview_sources():
+    logger.info("POST /admin/broadcasts/preview called")
+    msg, status_code = preview_sources(request.get_json())
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts/segments/sync", methods=["POST"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_start_segment_sync():
+    logger.info("POST /admin/broadcasts/segments/sync called")
+    msg, status_code = start_segment_sync(request.get_json(), _actor_from_request())
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts/segments/<segment_id>/sync-status", methods=["GET"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_get_sync_status(segment_id):
+    msg, status_code = get_sync_status(segment_id)
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts", methods=["GET"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_list_broadcasts():
+    logger.info("GET /admin/broadcasts called")
+    msg, status_code = list_broadcasts()
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts", methods=["POST"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_create_broadcast():
+    logger.info("POST /admin/broadcasts called")
+    msg, status_code = create_broadcast(request.get_json(), _actor_from_request())
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts/<broadcast_id>", methods=["GET"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_get_broadcast(broadcast_id):
+    msg, status_code = get_broadcast(broadcast_id)
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts/<broadcast_id>/send", methods=["POST"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_send_broadcast(broadcast_id):
+    logger.info(f"POST /admin/broadcasts/{broadcast_id}/send called")
+    msg, status_code = send_broadcast(broadcast_id, request.get_json(silent=True), _actor_from_request())
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts/batch-send", methods=["POST"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_batch_send():
+    logger.info("POST /admin/broadcasts/batch-send called")
+    msg, status_code = batch_send_emails(request.get_json(), _actor_from_request())
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts/contacts", methods=["GET"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_list_contacts():
+    logger.info("GET /admin/broadcasts/contacts called")
+    force = request.args.get("force", "false").lower() == "true"
+    msg, status_code = list_contacts(force=force)
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts/contacts/prune", methods=["POST"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_start_contact_prune():
+    logger.info("POST /admin/broadcasts/contacts/prune called")
+    msg, status_code = start_contact_prune(request.get_json(), _actor_from_request())
+    return vars(msg), status_code
+
+
+@bp.route("/admin/broadcasts/contacts/prune-status", methods=["GET"])
+@auth.require_user
+@auth.require_org_member_with_permission("volunteer.admin", req_to_org_id=getOrgId)
+def admin_get_prune_status():
+    msg, status_code = get_prune_status()
+    return vars(msg), status_code