diff --git a/CLAUDE.md b/CLAUDE.md index f46569a..9278599 100644 --- a/CLAUDE.md +++ b/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 "" [--event-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 "" [--event-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//sync-status` (poll), `GET|POST ''` (list / create broadcast — draft by default, `send:true`/`scheduled_at` to send), `GET `, `POST /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 `) + `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//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/` 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. diff --git a/api/__init__.py b/api/__init__.py index 1652e7f..e3848d3 100644 --- a/api/__init__.py +++ b/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 diff --git a/api/broadcasts/__init__.py b/api/broadcasts/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/broadcasts/broadcasts_views.py b/api/broadcasts/broadcasts_views.py new file mode 100644 index 0000000..ffb5e58 --- /dev/null +++ b/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//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/", 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//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 diff --git a/api/broadcasts/tests/__init__.py b/api/broadcasts/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/api/broadcasts/tests/test_broadcasts_service.py b/api/broadcasts/tests/test_broadcasts_service.py new file mode 100644 index 0000000..87b2758 --- /dev/null +++ b/api/broadcasts/tests/test_broadcasts_service.py @@ -0,0 +1,365 @@ +"""Tests for the Resend segment sync / broadcast / batch-send service. + +Runs under ENVIRONMENT=test (see conftest / pytest env), so +_notifications_disabled() is True: nothing here may touch the network. +Resend APIs are mocked throughout. +""" +import os +import time +from unittest.mock import patch + +import pytest +from mockfirestore import MockFirestore + +import services.broadcasts_service as svc + + +ACTOR = {"propel_user_id": "test-user", "email": "greg@ohack.org"} + + +@pytest.fixture(autouse=True) +def test_env(monkeypatch): + monkeypatch.setenv("ENVIRONMENT", "test") + monkeypatch.setenv("RESEND_API_KEY", "re_test_full") + monkeypatch.setenv("RESEND_WELCOME_EMAIL_KEY", "re_test_send") + monkeypatch.delenv("RESEND_MARKETING_CONTACT_LIMIT", raising=False) + + +@pytest.fixture +def db(): + mock_db = MockFirestore() + mock_db.collection("users").add({"email_address": "alice@example.com", "name": "Alice A"}) + mock_db.collection("users").add({"email_address": "BOB@example.com", "name": "Bob B"}) + mock_db.collection("users").add({"name": "No Email"}) + mock_db.collection("leads").add({"email": "lead@example.com", "name": "Lead One"}) + mock_db.collection("leads").add({"email": "alice@example.com", "name": "Alice Lead"}) + mock_db.collection("volunteers").add({ + "email": "mentor@example.com", "name": "Mia Mentor", + "volunteer_type": "mentor", "event_id": "2026_fall", "isSelected": True, + }) + mock_db.collection("volunteers").add({ + "email": "pending@example.com", "name": "Pat Pending", + "volunteer_type": "mentor", "event_id": "2026_fall", "isSelected": False, + }) + mock_db.collection("contact_submissions").add({ + "email": "recruiter@corp.com", "firstName": "Rae", "lastName": "Recruiter", + "inquiryType": "recruit", "receiveUpdates": True, + }) + mock_db.collection("contact_submissions").add({ + "email": "npo@example.org", "firstName": "Nia", "lastName": "Npo", + "inquiryType": "nonprofit", "receiveUpdates": False, + }) + mock_db.collection("contact_submissions").add({ + "email": "RECRUITER2@corp.com", "name": "Rex Recruiter", + "inquiryType": "Recruit", "receiveUpdates": False, + }) + with patch.object(svc, "get_db", return_value=mock_db): + yield mock_db + + +class TestNormEmail: + def test_valid_lowercased(self): + assert svc._norm_email(" Foo@Bar.COM ") == "foo@bar.com" + + def test_invalid(self): + assert svc._norm_email("not-an-email") is None + assert svc._norm_email("") is None + assert svc._norm_email(None) is None + + +class TestCollectContacts: + def test_profiles_dedupe_and_case(self, db): + contacts, stats = svc.collect_contacts([{"type": "profiles"}]) + assert set(contacts) == {"alice@example.com", "bob@example.com"} + assert stats["per_source"]["profiles"] == 2 + assert stats["union_total"] == 2 + + def test_union_dedupes_across_sources(self, db): + contacts, stats = svc.collect_contacts([ + {"type": "profiles"}, + {"type": "leads"}, + ]) + # alice appears in both — union removes the overlap + assert stats["union_total"] == 3 + assert stats["overlap_removed"] == 1 + # first source wins the record; names are enriched not replaced + assert contacts["alice@example.com"]["first_name"] == "Alice" + + def test_volunteers_selected_only(self, db): + contacts, _ = svc.collect_contacts([ + {"type": "volunteers", "volunteer_type": "mentor", + "event_id": "2026_fall", "selected_only": True}, + ]) + assert set(contacts) == {"mentor@example.com"} + + def test_contact_submissions_filtered_by_inquiry_type(self, db): + contacts, stats = svc.collect_contacts([ + {"type": "contact_submissions", "inquiry_types": ["recruit"]}, + ]) + # case-insensitive on both the filter and the stored inquiryType + assert set(contacts) == {"recruiter@corp.com", "recruiter2@corp.com"} + assert stats["per_source"]["contact:recruit"] == 2 + + def test_contact_submissions_all_types(self, db): + contacts, _ = svc.collect_contacts([{"type": "contact_submissions"}]) + assert len(contacts) == 3 + + def test_contact_submissions_opt_in_only(self, db): + contacts, stats = svc.collect_contacts([ + {"type": "contact_submissions", "updates_opt_in_only": True}, + ]) + assert set(contacts) == {"recruiter@corp.com"} + assert stats["per_source"]["contact:all:opted-in"] == 1 + + def test_contact_submissions_bad_inquiry_types_shape(self, db): + with pytest.raises(ValueError): + svc.collect_contacts([ + {"type": "contact_submissions", "inquiry_types": "recruit"}, + ]) + + def test_custom_emails_validated(self, db): + contacts, stats = svc.collect_contacts([], ["good@example.com", "bad-email"]) + assert set(contacts) == {"good@example.com"} + assert stats["custom_valid"] == 1 + assert stats["custom_invalid"] == ["bad-email"] + + def test_unknown_source_raises(self, db): + with pytest.raises(ValueError): + svc.collect_contacts([{"type": "nope"}]) + + def test_over_limit_flag(self, db, monkeypatch): + monkeypatch.setenv("RESEND_MARKETING_CONTACT_LIMIT", "1") + _, stats = svc.collect_contacts([{"type": "profiles"}]) + assert stats["over_limit"] is True + assert stats["contact_limit"] == 1 + + +class TestPreviewSources: + def test_preview_returns_stats(self, db): + msg, status = svc.preview_sources({"sources": [{"type": "leads"}]}) + assert status == 200 + assert msg.stats["per_source"]["leads"] == 2 + + def test_preview_bad_source_400(self, db): + msg, status = svc.preview_sources({"sources": [{"type": "bogus"}]}) + assert status == 400 + + +class TestSegmentSync: + def test_sync_simulated_in_test_env(self, db): + with patch.object(svc, "_get_or_create_segment", return_value="seg_1"), \ + patch.object(svc, "_existing_segment_emails", return_value={"alice@example.com"}), \ + patch.object(svc, "send_slack_audit"): + msg, status = svc.start_segment_sync( + {"segment_name": "Test Segment", "sources": [{"type": "profiles"}]}, + ACTOR, + ) + assert status == 202 + assert msg.status == "started" + assert msg.collected == 2 + + # daemon thread — give it a beat to finish the simulated path + for _ in range(50): + status_msg, _ = svc.get_sync_status("seg_1") + if status_msg.status.get("state") == "done": + break + time.sleep(0.05) + final = status_msg.status + assert final["state"] == "done" + assert final["simulated"] is True + assert final["already_in_segment"] == 1 + assert final["to_add"] == 1 + assert final["added"] == 0 # simulated: no writes + + def test_sync_conflict_when_lock_held(self, db): + svc.set_cached(svc._sync_lock_key("seg_locked"), True, ttl=60) + try: + msg, status = svc.start_segment_sync( + {"segment_id": "seg_locked", "sources": [{"type": "profiles"}]}, + ACTOR, + ) + assert status == 409 + assert msg.status == "already_running" + finally: + svc.delete_cached(svc._sync_lock_key("seg_locked")) + + def test_sync_requires_segment(self, db): + _, status = svc.start_segment_sync({"sources": [{"type": "profiles"}]}, ACTOR) + assert status == 400 + + def test_stalled_detection(self, db): + svc.set_cached(svc._sync_status_key("seg_stale"), { + "state": "running", + "updated_at": "2020-01-01T00:00:00+00:00", + }, ttl=600) + try: + msg, _ = svc.get_sync_status("seg_stale") + assert msg.status["state"] == "stalled" + finally: + svc.delete_cached(svc._sync_status_key("seg_stale")) + + def test_status_none_when_unknown(self, db): + msg, status = svc.get_sync_status("seg_unknown") + assert status == 200 + assert msg.status["state"] == "none" + + +class TestBroadcastHtml: + def test_unsubscribe_footer_appended(self): + html = svc.render_broadcast_html("Hello **world**") + assert svc.UNSUBSCRIBE_PLACEHOLDER in html + assert "world" in html + + def test_existing_unsubscribe_not_duplicated(self): + html = svc.render_broadcast_html( + f"Bye [unsubscribe]({svc.UNSUBSCRIBE_PLACEHOLDER})") + assert html.count(svc.UNSUBSCRIBE_PLACEHOLDER) == 1 + + +class TestFromAddress: + def test_default_allowed(self): + addr, err = svc._resolve_from_address(None) + assert err is None + assert "notify.ohack.dev" in addr + + def test_disallowed_domain_rejected(self): + addr, err = svc._resolve_from_address("Evil ") + assert addr is None + assert "not in the allowed list" in err + + def test_friendly_name_parsed(self): + addr, err = svc._resolve_from_address("OHack ") + assert err is None + assert addr == "OHack " + + +class TestCreateBroadcast: + def test_requires_fields(self): + _, status = svc.create_broadcast({}, ACTOR) + assert status == 400 + + def test_simulated_in_test_env(self): + msg, status = svc.create_broadcast({ + "segment_id": "seg_1", "subject": "Hi", "body_markdown": "Hello", + }, ACTOR) + assert status == 200 + assert msg.simulated is True + assert msg.broadcast["status"] == "simulated" + + def test_bad_from_rejected_before_simulation(self): + msg, status = svc.create_broadcast({ + "segment_id": "seg_1", "subject": "Hi", "body_markdown": "Hello", + "from_address": "x@unverified.example.com", + }, ACTOR) + assert status == 400 + + +FAKE_CONTACTS = [ + {"id": "c1", "email": "a@example.com", "first_name": "A", "last_name": "", + "unsubscribed": False, "created_at": "2026-01-01"}, + {"id": "c2", "email": "b@example.com", "first_name": "B", "last_name": "", + "unsubscribed": True, "created_at": "2026-01-02"}, + {"id": "c3", "email": "c@example.com", "first_name": "C", "last_name": "", + "unsubscribed": True, "created_at": "2026-01-03"}, +] + + +class TestContacts: + def _clean(self): + svc.delete_cached(svc._CONTACTS_CACHE_KEY) + svc.delete_cached(svc._PRUNE_LOCK_KEY) + svc.delete_cached(svc._PRUNE_STATUS_KEY) + + def test_list_contacts_counts(self, monkeypatch): + self._clean() + monkeypatch.setenv("RESEND_MARKETING_CONTACT_LIMIT", "2") + with patch.object(svc, "_crawl_all_contacts", return_value=FAKE_CONTACTS): + msg, status = svc.list_contacts(force=True) + self._clean() + assert status == 200 + assert msg.total == 3 + assert msg.unsubscribed_count == 2 + assert msg.over_limit is True + + def test_prune_unsubscribed_targets(self): + self._clean() + with patch.object(svc, "_crawl_all_contacts", return_value=FAKE_CONTACTS), \ + patch.object(svc, "send_slack_audit"): + msg, status = svc.start_contact_prune({"mode": "unsubscribed"}, ACTOR) + assert status == 202 + assert msg.total_targets == 2 + for _ in range(50): + status_msg, _ = svc.get_prune_status() + if status_msg.status.get("state") == "done": + break + time.sleep(0.05) + assert status_msg.status["simulated"] is True + self._clean() + + def test_prune_emails_intersects_known(self): + self._clean() + with patch.object(svc, "_crawl_all_contacts", return_value=FAKE_CONTACTS): + msg, status = svc.start_contact_prune( + {"mode": "emails", "emails": ["A@example.com", "nobody@x.com"]}, ACTOR) + assert status == 202 + assert msg.total_targets == 1 + for _ in range(50): + status_msg, _ = svc.get_prune_status() + if status_msg.status.get("state") == "done": + break + time.sleep(0.05) + self._clean() + + def test_prune_bad_mode(self): + self._clean() + _, status = svc.start_contact_prune({"mode": "everything"}, ACTOR) + assert status == 400 + + def test_prune_conflict_when_running(self): + self._clean() + svc.set_cached(svc._PRUNE_LOCK_KEY, True, ttl=60) + try: + msg, status = svc.start_contact_prune({"mode": "all"}, ACTOR) + assert status == 409 + finally: + self._clean() + + def test_prune_empty_targets(self): + self._clean() + no_unsub = [c for c in FAKE_CONTACTS if not c["unsubscribed"]] + with patch.object(svc, "_crawl_all_contacts", return_value=no_unsub): + msg, status = svc.start_contact_prune({"mode": "unsubscribed"}, ACTOR) + assert status == 200 + assert msg.status == "empty" + self._clean() + + +class TestBatchSend: + def test_simulated_results_in_order(self): + with patch.object(svc, "send_slack_audit"): + msg, status = svc.batch_send_emails({ + "subject": "Hello", + "recipients": [ + {"email": "a@example.com", "name": "A", "message": "hi a"}, + {"email": "not-an-email", "name": "?", "message": "hi"}, + {"email": "b@example.com", "name": "B", + "message": "scan [QRCode:https://ohack.dev]"}, + ], + }, ACTOR) + assert status == 200 + results = msg.results + assert results[0]["success"] is True and results[0].get("simulated") is True + assert results[1]["success"] is False and "invalid email" in results[1]["error"] + assert results[2]["success"] is False and "QR-code" in results[2]["error"] + assert msg.summary == {"total": 3, "successful": 1, "failed": 2, "simulated": True} + + def test_requires_subject_and_recipients(self): + _, status = svc.batch_send_emails({"recipients": []}, ACTOR) + assert status == 400 + _, status = svc.batch_send_emails({"subject": "x"}, ACTOR) + assert status == 400 + + def test_caps_request_size(self): + recipients = [{"email": f"u{i}@example.com", "message": "hi"} for i in range(501)] + _, status = svc.batch_send_emails({"subject": "x", "recipients": recipients}, ACTOR) + assert status == 400 diff --git a/api/slack/slack_service.py b/api/slack/slack_service.py index 2515180..c196ebb 100644 --- a/api/slack/slack_service.py +++ b/api/slack/slack_service.py @@ -144,9 +144,10 @@ def clear_slack_cache() -> Dict[str, Any]: # Clear all Slack caches active_users_cleared = clear_pattern("slack:active_users:*") user_details_cleared = clear_pattern("slack:user_details:*") - + userlist_cleared = clear_pattern("slack:userlist:*") + return { - "success": active_users_cleared and user_details_cleared, + "success": active_users_cleared and user_details_cleared and userlist_cleared, "message": "All Slack caches cleared successfully" } diff --git a/api/slack/tests/test_slack_service.py b/api/slack/tests/test_slack_service.py index 4597503..27b1843 100644 --- a/api/slack/tests/test_slack_service.py +++ b/api/slack/tests/test_slack_service.py @@ -201,9 +201,10 @@ def test_clear_slack_cache(mock_clear_pattern): assert result["message"] == "All Slack caches cleared successfully" # Verify mocks - assert mock_clear_pattern.call_count == 2 + assert mock_clear_pattern.call_count == 3 mock_clear_pattern.assert_any_call("slack:active_users:*") mock_clear_pattern.assert_any_call("slack:user_details:*") + mock_clear_pattern.assert_any_call("slack:userlist:*") # --- Tests for sync_slack_users_to_firestore --- diff --git a/common/utils/slack.py b/common/utils/slack.py index 0d07c4b..e99fa90 100644 --- a/common/utils/slack.py +++ b/common/utils/slack.py @@ -10,6 +10,7 @@ from cachetools import TTLCache, cached from ratelimit import limits, sleep_and_retry import threading +from common.utils.redis_cache import redis_cached load_dotenv() @@ -57,6 +58,11 @@ def presence(user_id=None): client = get_client() return client.users_getPresence(user=user_id) +# Cache the full member crawl so callers that filter it differently (e.g. +# get_active_users with varying active_days) don't each pay a full paginated +# users.list crawl behind the blocking rate limiter. Cache hits skip the +# limiter entirely because redis_cached is the outer decorator. +@redis_cached(prefix="slack:userlist", ttl=600) @RateLimiter(max_calls=20, period=60) def userlist(): client = get_client() diff --git a/services/broadcasts_service.py b/services/broadcasts_service.py new file mode 100644 index 0000000..4597bfd --- /dev/null +++ b/services/broadcasts_service.py @@ -0,0 +1,1146 @@ +"""Resend segment sync + broadcast + batch-send service for the admin Email tab. + +Ports scripts/sync_resend_audience.py into the app: collect contacts from +Firestore (users/volunteers/leads) and Slack, sync them into a Resend segment +(create-only, idempotent), and send one broadcast to the segment instead of +N per-recipient emails. Also hosts the transactional Batch send used by the +frontend's personalized bulk path (100 emails per Resend call). + +API keys: segment/contact/broadcast operations REQUIRE the full-access +RESEND_API_KEY (RESEND_WELCOME_EMAIL_KEY is send-only and 401s on them — never +fall back to it here). Batch sends are Emails-scope and use the welcome key. + +Hazard: resend.api_key is a module-level global shared with every other email +path across gunicorn threads. Each entry point here sets it immediately before +its Resend calls and keeps the call section short; a long-running sync thread +re-sets it defensively per write. +""" + +import os +import re +import threading +import time +from datetime import datetime, timezone +from typing import Dict, Iterable, List, Optional, Set, Tuple + +import markdown +import resend + +from api.messages.message import Message +from common.log import get_logger +from common.utils.firebase import get_db +from common.utils.redis_cache import delete_cached, get_cached, set_cached +from common.utils.slack import send_slack_audit + +logger = get_logger(__name__) + +EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") +QR_MARKER_RE = re.compile(r"\[QRCode:", re.IGNORECASE) +UNSUBSCRIBE_PLACEHOLDER = "{{{RESEND_UNSUBSCRIBE_URL}}}" + +MAX_BATCH_RECIPIENTS_PER_REQUEST = 500 +RESEND_BATCH_CHUNK = 100 # Resend /emails/batch hard limit + +_SYNC_STATUS_TTL = 24 * 3600 +_SYNC_LOCK_TTL = 30 * 60 # self-heals if the worker dies mid-sync +_SYNC_STALL_SECONDS = 120 +_CONTACT_WRITE_SLEEP = 0.05 # proven value from scripts/sync_resend_audience.py + +_CONTACTS_CACHE_KEY = "broadcasts:contacts:index" +_CONTACTS_CACHE_TTL = 60 +_PRUNE_STATUS_KEY = "broadcasts:contacts:prune:status" +_PRUNE_LOCK_KEY = "broadcasts:contacts:prune:lock" +_PRUNE_LOCK_TTL = 30 * 60 +PRUNE_MODES = ("unsubscribed", "emails", "all") + +VOLUNTEER_TYPES = ("mentor", "judge", "sponsor", "volunteer", "hacker") + +DEFAULT_BROADCAST_FROM = "Opportunity Hack " +DEFAULT_BROADCAST_FROM_DOMAINS = "notify.ohack.dev,apply.ohack.dev" +DEFAULT_REPLY_TO = "Opportunity Hack Questions " + + +def _sync_status_key(segment_id: str) -> str: + return f"broadcasts:sync:{segment_id}:status" + + +def _sync_lock_key(segment_id: str) -> str: + return f"broadcasts:sync:{segment_id}:lock" + + +def _notifications_disabled() -> bool: + """Mirror of the ENVIRONMENT=test gate used across services — unit tests + (MockFirestore) must never write to Resend or send email.""" + return os.environ.get("ENVIRONMENT") == "test" + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat() + + +def _contact_limit() -> int: + """Marketing-tier contact cap used for over-limit warnings (Resend free + marketing tier = 1,000 contacts). Raise via env after upgrading the tier.""" + try: + return int(os.environ.get("RESEND_MARKETING_CONTACT_LIMIT", "1000")) + except ValueError: + return 1000 + + +def _resend_full_key() -> str: + key = os.environ.get("RESEND_API_KEY") + if not key: + raise RuntimeError( + "RESEND_API_KEY (full-access) is not configured — required for " + "segment/contact/broadcast operations" + ) + return key + + +def _resend_send_key() -> str: + key = os.environ.get("RESEND_WELCOME_EMAIL_KEY") or os.environ.get("RESEND_API_KEY") + if not key: + raise RuntimeError("RESEND_WELCOME_EMAIL_KEY is not configured") + return key + + +def _segments_api(): + """Prefer the Segments API; older SDKs only expose the Audiences alias + (same server-side objects and ids).""" + return getattr(resend, "Segments", None) or resend.Audiences + + +def _resp_data(resp) -> list: + if isinstance(resp, dict): + return resp.get("data", []) or [] + return getattr(resp, "data", []) or [] + + +def _field(obj, key): + if isinstance(obj, dict): + return obj.get(key) + return getattr(obj, key, None) + + +# --------------------------------------------------------------------------- +# Contact collection (ported from scripts/sync_resend_audience.py) +# --------------------------------------------------------------------------- + +def _norm_email(raw) -> Optional[str]: + if not raw: + return None + e = str(raw).strip().lower() + return e if EMAIL_RE.match(e) else None + + +def _split_name(full: str) -> Tuple[str, str]: + if not full: + return "", "" + parts = full.strip().split() + if len(parts) == 1: + return parts[0], "" + return parts[0], " ".join(parts[1:]) + + +def _add(contacts: Dict[str, dict], email: Optional[str], first: str, last: str, src: str): + """Insert or upgrade a contact entry. Prefers entries with a real name.""" + if not email: + return + first = (first or "").strip() + last = (last or "").strip() + existing = contacts.get(email) + if existing is None: + contacts[email] = { + "email": email, + "first_name": first, + "last_name": last, + "source": src, + } + return + if not existing["first_name"] and first: + existing["first_name"] = first + if not existing["last_name"] and last: + existing["last_name"] = last + + +def load_profiles() -> Dict[str, dict]: + """Registered ohack.dev users (users collection).""" + db = get_db() + out: Dict[str, dict] = {} + for doc in db.collection("users").stream(): + d = doc.to_dict() or {} + email = _norm_email(d.get("email_address")) + if not email: + continue + first, last = _split_name(d.get("name", "") or d.get("nickname", "")) + _add(out, email, first, last, "profiles") + return out + + +def load_volunteers(volunteer_type: Optional[str], event_id: Optional[str], + selected_only: bool) -> Dict[str, dict]: + db = get_db() + query = db.collection("volunteers") + if volunteer_type: + query = query.where("volunteer_type", "==", volunteer_type) + if event_id: + query = query.where("event_id", "==", event_id) + # isSelected filtered in Python — a firestore where would need a composite index. + out: Dict[str, dict] = {} + for doc in query.stream(): + d = doc.to_dict() or {} + if selected_only and not d.get("isSelected"): + continue + email = _norm_email(d.get("email")) + if not email: + continue + first = d.get("first_name", "") or "" + last = d.get("last_name", "") or "" + if not first and not last: + first, last = _split_name(d.get("name", "")) + _add(out, email, first, last, f"volunteers:{volunteer_type or 'all'}") + return out + + +def load_leads() -> Dict[str, dict]: + """Newsletter signups (leads collection, fed by the ohack.dev LeadForm).""" + db = get_db() + out: Dict[str, dict] = {} + for doc in db.collection("leads").stream(): + d = doc.to_dict() or {} + email = _norm_email(d.get("email")) + if not email: + continue + first, last = _split_name(d.get("name", "")) + _add(out, email, first, last, "leads") + return out + + +def load_contact_submissions(inquiry_types: Optional[List[str]] = None, + updates_opt_in_only: bool = False) -> Dict[str, dict]: + """Contact-form submitters (contact_submissions collection, /contact page). + + inquiry_types: keep only these `inquiryType` values (case-insensitive); + empty/None = all. updates_opt_in_only: keep only submitters who checked + the form's `receiveUpdates` box.""" + wanted = {t.strip().lower() for t in (inquiry_types or []) if t and t.strip()} + db = get_db() + out: Dict[str, dict] = {} + for doc in db.collection("contact_submissions").stream(): + d = doc.to_dict() or {} + if wanted and (d.get("inquiryType") or "").strip().lower() not in wanted: + continue + if updates_opt_in_only and not d.get("receiveUpdates"): + continue + email = _norm_email(d.get("email")) + if not email: + continue + first = (d.get("firstName") or "").strip() + last = (d.get("lastName") or "").strip() + if not first and not last: + first, last = _split_name(d.get("name", "")) + _add(out, email, first, last, "contact_submissions") + return out + + +def load_slack_members(active_days: int = 365) -> Dict[str, dict]: + """Slack workspace members with emails. Deleted/disabled/bot/restricted + accounts are already excluded inside get_active_users.""" + from api.slack.slack_service import get_active_users # lazy: avoid import at module load + + out: Dict[str, dict] = {} + for u in get_active_users(days=active_days, admin=True): + email = _norm_email(u.get("email")) + if not email: + continue + first, last = _split_name(u.get("real_name") or u.get("name") or "") + _add(out, email, first, last, "slack") + return out + + +def collect_contacts(sources: List[dict], custom_emails: Optional[List[str]] = None + ) -> Tuple[Dict[str, dict], dict]: + """Resolve a sources spec into a deduped contact dict keyed by lowercase + email, plus per-source stats for the preview UI. + + Source spec entries: + {"type": "profiles"} + {"type": "leads"} + {"type": "volunteers", "volunteer_type"?: str, "event_id"?: str, "selected_only"?: bool} + {"type": "slack", "active_days"?: int} + {"type": "contact_submissions", "inquiry_types"?: [str], "updates_opt_in_only"?: bool} + """ + combined: Dict[str, dict] = {} + per_source: Dict[str, int] = {} + raw_total = 0 + + for spec in sources or []: + stype = (spec or {}).get("type") + if stype == "profiles": + chunk = load_profiles() + label = "profiles" + elif stype == "leads": + chunk = load_leads() + label = "leads" + elif stype == "volunteers": + vtype = spec.get("volunteer_type") or None + if vtype and vtype not in VOLUNTEER_TYPES: + raise ValueError(f"unknown volunteer_type: {vtype}") + chunk = load_volunteers(vtype, spec.get("event_id") or None, + bool(spec.get("selected_only"))) + label = f"volunteers:{vtype or 'all'}" + if spec.get("event_id"): + label += f":{spec['event_id']}" + elif stype == "contact_submissions": + inquiry_types = spec.get("inquiry_types") or [] + if not isinstance(inquiry_types, list): + raise ValueError("inquiry_types must be a list") + chunk = load_contact_submissions( + inquiry_types, bool(spec.get("updates_opt_in_only"))) + label = "contact:" + ( + ",".join(sorted(t.strip().lower() for t in inquiry_types if t.strip())) + or "all") + if spec.get("updates_opt_in_only"): + label += ":opted-in" + elif stype == "slack": + try: + active_days = int(spec.get("active_days", 365)) + except (TypeError, ValueError): + active_days = 365 + active_days = min(max(active_days, 1), 10000) + chunk = load_slack_members(active_days) + label = f"slack:{active_days}d" + else: + raise ValueError(f"unknown source type: {stype}") + + per_source[label] = len(chunk) + raw_total += len(chunk) + for email, rec in chunk.items(): + _add(combined, email, rec["first_name"], rec["last_name"], rec["source"]) + + custom_valid = 0 + custom_invalid: List[str] = [] + for raw in custom_emails or []: + email = _norm_email(raw) + if not email: + custom_invalid.append(str(raw)) + continue + custom_valid += 1 + raw_total += 1 + _add(combined, email, "", "", "custom") + + stats = { + "per_source": per_source, + "custom_valid": custom_valid, + "custom_invalid": custom_invalid, + "union_total": len(combined), + "overlap_removed": raw_total - len(combined), + "contact_limit": _contact_limit(), + "over_limit": len(combined) > _contact_limit(), + } + return combined, stats + + +def preview_sources(payload: dict) -> Tuple[Message, int]: + """Dry-run: counts only, no Resend reads or writes.""" + payload = payload or {} + try: + _, stats = collect_contacts(payload.get("sources"), payload.get("custom_emails")) + except ValueError as e: + return Message(str(e)), 400 + msg = Message("Preview computed") + msg.stats = stats + return msg, 200 + + +# --------------------------------------------------------------------------- +# Segments +# --------------------------------------------------------------------------- + +def list_segments() -> Tuple[Message, int]: + resend.api_key = _resend_full_key() + listed = _segments_api().list() + segments = [ + { + "id": _field(a, "id"), + "name": _field(a, "name"), + "created_at": _field(a, "created_at"), + } + for a in _resp_data(listed) + ] + msg = Message("ok") + msg.segments = segments + return msg, 200 + + +def _get_or_create_segment(name: str) -> str: + """List-then-create by name; reuses standing segments across sends.""" + api = _segments_api() + for a in _resp_data(api.list()): + if _field(a, "name") == name: + return _field(a, "id") + created = api.create({"name": name}) + segment_id = _field(created, "id") + logger.info("created Resend segment '%s' id=%s", name, segment_id) + return segment_id + + +def _existing_segment_emails(segment_id: str) -> Set[str]: + """Full paginated email set already in the segment → idempotent re-syncs.""" + out: Set[str] = set() + after = None + while True: + params: dict = {"limit": 100} + if after: + params["after"] = after + # audience_id is the compat kwarg — segments and audiences share ids. + resp = resend.Contacts.list(audience_id=segment_id, params=params) + data = _resp_data(resp) + if not data: + break + last_id = None + for c in data: + email = _field(c, "email") + if email: + out.add(email.strip().lower()) + last_id = _field(c, "id") + if len(data) < 100 or not last_id: + break + after = last_id + return out + + +# --------------------------------------------------------------------------- +# Contact management (quota lives on GLOBAL contacts — deleting a contact +# account-wide is what frees marketing-tier quota; unsubscribed contacts +# can't receive broadcasts but STILL count against the limit) +# --------------------------------------------------------------------------- + +def _serialize_contact(c) -> dict: + return { + "id": _field(c, "id"), + "email": (_field(c, "email") or "").strip().lower(), + "first_name": _field(c, "first_name") or "", + "last_name": _field(c, "last_name") or "", + "unsubscribed": bool(_field(c, "unsubscribed")), + "created_at": _field(c, "created_at"), + } + + +def _crawl_contacts_page(after: Optional[str]): + params: dict = {"limit": 100} + if after: + params["after"] = after + try: + # Account-level (global) contacts — the set the quota counts. + return resend.Contacts.list(params=params) + except TypeError: + # Very old SDKs require audience_id; fall back to the union across + # segments (may miss contacts in no segment, but better than nothing). + return None + + +def _crawl_all_contacts() -> List[dict]: + out: Dict[str, dict] = {} + after = None + while True: + resp = _crawl_contacts_page(after) + if resp is None: + # Fallback: union of every segment's contacts. + for seg in _resp_data(_segments_api().list()): + seg_id = _field(seg, "id") + seg_after = None + while True: + params = {"limit": 100} + if seg_after: + params["after"] = seg_after + seg_resp = resend.Contacts.list(audience_id=seg_id, params=params) + data = _resp_data(seg_resp) + if not data: + break + for c in data: + rec = _serialize_contact(c) + if rec["email"]: + out.setdefault(rec["email"], rec) + if len(data) < 100: + break + seg_after = _field(data[-1], "id") + break + data = _resp_data(resp) + if not data: + break + for c in data: + rec = _serialize_contact(c) + if rec["email"]: + out.setdefault(rec["email"], rec) + if len(data) < 100: + break + after = _field(data[-1], "id") + return sorted(out.values(), key=lambda r: r["email"]) + + +def list_contacts(force: bool = False) -> Tuple[Message, int]: + """Full contact inventory (cached 60s) + quota picture for the admin UI.""" + contacts = None if force else get_cached(_CONTACTS_CACHE_KEY) + if contacts is None: + resend.api_key = _resend_full_key() + contacts = _crawl_all_contacts() + set_cached(_CONTACTS_CACHE_KEY, contacts, ttl=_CONTACTS_CACHE_TTL) + + unsubscribed = sum(1 for c in contacts if c.get("unsubscribed")) + msg = Message("ok") + msg.contacts = contacts + msg.total = len(contacts) + msg.unsubscribed_count = unsubscribed + msg.contact_limit = _contact_limit() + msg.over_limit = len(contacts) > _contact_limit() + return msg, 200 + + +def start_contact_prune(payload: dict, actor: Optional[dict]) -> Tuple[Message, int]: + """Delete GLOBAL contacts to reclaim marketing-tier quota. Modes: + - "unsubscribed": every unsubscribed contact (safe — they can't receive + broadcasts anyway, but they count against the limit) + - "emails": an explicit list (admin-selected in the UI) + - "all": everything (danger — the UI requires typed confirmation) + Runs in a daemon thread (one global job at a time) with polled status.""" + payload = payload or {} + mode = (payload.get("mode") or "").strip() + if mode not in PRUNE_MODES: + return Message(f"mode must be one of {', '.join(PRUNE_MODES)}"), 400 + + if get_cached(_PRUNE_LOCK_KEY) is not None: + msg = Message("A contact prune is already running") + msg.status = "already_running" + return msg, 409 + + resend.api_key = _resend_full_key() + contacts = _crawl_all_contacts() + + if mode == "unsubscribed": + targets = [c["email"] for c in contacts if c.get("unsubscribed")] + elif mode == "all": + targets = [c["email"] for c in contacts] + else: + requested = {e for e in ((_norm_email(x) for x in payload.get("emails") or [])) if e} + if not requested: + return Message("emails is required for mode=emails"), 400 + known = {c["email"] for c in contacts} + targets = sorted(requested & known) + + if not targets: + msg = Message("Nothing to delete for the requested mode") + msg.status = "empty" + msg.total_targets = 0 + return msg, 200 + + set_cached(_PRUNE_LOCK_KEY, True, ttl=_PRUNE_LOCK_TTL) + status = { + "state": "running", + "mode": mode, + "total_targets": len(targets), + "deleted": 0, + "failed": 0, + "started_at": _now_iso(), + "updated_at": _now_iso(), + "finished_at": None, + "error": None, + "requested_by": (actor or {}).get("email"), + "simulated": _notifications_disabled(), + } + set_cached(_PRUNE_STATUS_KEY, status, ttl=_SYNC_STATUS_TTL) + + threading.Thread( + target=_run_contact_prune, + args=(mode, targets, actor), + daemon=True, + ).start() + + msg = Message("Contact prune started") + msg.status = "started" + msg.mode = mode + msg.total_targets = len(targets) + return msg, 202 + + +def _update_prune_status(patch: dict) -> dict: + status = get_cached(_PRUNE_STATUS_KEY) or {} + status.update(patch) + status["updated_at"] = _now_iso() + set_cached(_PRUNE_STATUS_KEY, status, ttl=_SYNC_STATUS_TTL) + return status + + +def _run_contact_prune(mode: str, targets: List[str], actor: Optional[dict]) -> None: + try: + if _notifications_disabled(): + _update_prune_status({"state": "done", "finished_at": _now_iso(), + "simulated": True}) + return + + deleted, failed = 0, 0 + for email in targets: + try: + resend.api_key = _resend_full_key() + # No audience_id → deletes the GLOBAL contact (frees quota). + resend.Contacts.remove(email=email) + deleted += 1 + except Exception as e: + failed += 1 + logger.warning("contact prune: failed to delete %s: %s", email, e) + if (deleted + failed) % 25 == 0: + _update_prune_status({"deleted": deleted, "failed": failed}) + time.sleep(_CONTACT_WRITE_SLEEP) + + _update_prune_status({ + "state": "done", + "deleted": deleted, + "failed": failed, + "finished_at": _now_iso(), + }) + send_slack_audit( + action="broadcast_contact_prune", + message=f"Resend contact prune finished: mode={mode} " + f"deleted={deleted} failed={failed} of {len(targets)}", + payload={"requested_by": (actor or {}).get("email")}, + ) + except Exception as e: + logger.error("contact prune failed: %s", e) + _update_prune_status({"state": "error", "error": str(e), + "finished_at": _now_iso()}) + finally: + delete_cached(_PRUNE_LOCK_KEY) + delete_cached(_CONTACTS_CACHE_KEY) + + +def get_prune_status() -> Tuple[Message, int]: + status = get_cached(_PRUNE_STATUS_KEY) + if not status: + msg = Message("No prune recorded") + msg.status = {"state": "none"} + return msg, 200 + if status.get("state") == "running": + try: + updated = datetime.fromisoformat(status["updated_at"]) + if (datetime.now(timezone.utc) - updated).total_seconds() > _SYNC_STALL_SECONDS: + status = dict(status) + status["state"] = "stalled" + except (KeyError, ValueError): + pass + msg = Message("ok") + msg.status = status + return msg, 200 + + +# --------------------------------------------------------------------------- +# Segment sync (background thread + redis status/lock) +# --------------------------------------------------------------------------- + +def start_segment_sync(payload: dict, actor: Optional[dict]) -> Tuple[Message, int]: + """Resolve the segment, collect contacts inline (fast), then hand the + Resend writes to a daemon thread. Never sync inline: thousands of contact + creates at ~20/s would blow the 120s gunicorn timeout.""" + payload = payload or {} + segment_id = (payload.get("segment_id") or "").strip() + segment_name = (payload.get("segment_name") or "").strip() + if not segment_id and not segment_name: + return Message("segment_id or segment_name is required"), 400 + + resend.api_key = _resend_full_key() + if not segment_id: + segment_id = _get_or_create_segment(segment_name) + if not segment_id: + return Message("Could not resolve Resend segment"), 502 + + lock_key = _sync_lock_key(segment_id) + if get_cached(lock_key) is not None: + msg = Message("A sync for this segment is already running") + msg.status = "already_running" + msg.segment_id = segment_id + return msg, 409 + + try: + contacts, stats = collect_contacts(payload.get("sources"), payload.get("custom_emails")) + except ValueError as e: + return Message(str(e)), 400 + if not contacts: + return Message("No contacts collected from the requested sources"), 400 + + set_cached(lock_key, True, ttl=_SYNC_LOCK_TTL) + status = { + "state": "running", + "segment_id": segment_id, + "collected": len(contacts), + "already_in_segment": None, + "to_add": None, + "added": 0, + "failed": 0, + "stats": stats, + "started_at": _now_iso(), + "updated_at": _now_iso(), + "finished_at": None, + "error": None, + "requested_by": (actor or {}).get("email"), + "simulated": _notifications_disabled(), + } + set_cached(_sync_status_key(segment_id), status, ttl=_SYNC_STATUS_TTL) + + threading.Thread( + target=_run_segment_sync, + args=(segment_id, contacts, actor), + daemon=True, + ).start() + + msg = Message("Segment sync started") + msg.status = "started" + msg.segment_id = segment_id + msg.collected = len(contacts) + msg.stats = stats + return msg, 202 + + +def _update_sync_status(segment_id: str, patch: dict) -> dict: + status = get_cached(_sync_status_key(segment_id)) or {} + status.update(patch) + status["updated_at"] = _now_iso() + set_cached(_sync_status_key(segment_id), status, ttl=_SYNC_STATUS_TTL) + return status + + +def _run_segment_sync(segment_id: str, contacts: Dict[str, dict], actor: Optional[dict]) -> None: + try: + resend.api_key = _resend_full_key() + already = _existing_segment_emails(segment_id) + to_add = [rec for email, rec in sorted(contacts.items()) if email not in already] + _update_sync_status(segment_id, { + "already_in_segment": len(already), + "to_add": len(to_add), + "projected_total": len(already) + len(to_add), + "over_limit": (len(already) + len(to_add)) > _contact_limit(), + }) + + added, failed = 0, 0 + if _notifications_disabled(): + _update_sync_status(segment_id, {"state": "done", "finished_at": _now_iso(), + "simulated": True}) + return + + for rec in to_add: + try: + resend.api_key = _resend_full_key() + # Create-only: existing contacts are never updated, so a + # previously-unsubscribed person is never resubscribed. + resend.Contacts.create({ + "audience_id": segment_id, + "email": rec["email"], + "first_name": rec["first_name"], + "last_name": rec["last_name"], + "unsubscribed": False, + }) + added += 1 + except Exception as e: + failed += 1 + logger.warning("segment sync: failed to add %s: %s", rec["email"], e) + if (added + failed) % 25 == 0: + _update_sync_status(segment_id, {"added": added, "failed": failed}) + time.sleep(_CONTACT_WRITE_SLEEP) + + _update_sync_status(segment_id, { + "state": "done", + "added": added, + "failed": failed, + "finished_at": _now_iso(), + }) + send_slack_audit( + action="broadcast_segment_sync", + message=f"Resend segment sync finished: segment={segment_id} " + f"collected={len(contacts)} added={added} failed={failed}", + payload={"requested_by": (actor or {}).get("email")}, + ) + except Exception as e: + logger.error("segment sync failed for %s: %s", segment_id, e) + _update_sync_status(segment_id, { + "state": "error", + "error": str(e), + "finished_at": _now_iso(), + }) + finally: + delete_cached(_sync_lock_key(segment_id)) + delete_cached(_CONTACTS_CACHE_KEY) # sync changes the global contact count + + +def get_sync_status(segment_id: str) -> Tuple[Message, int]: + status = get_cached(_sync_status_key(segment_id)) + if not status: + msg = Message("No sync recorded for this segment") + msg.status = {"state": "none", "segment_id": segment_id} + return msg, 200 + + # A daemon thread dies with its gunicorn worker; the lock TTL self-heals, + # and a stale heartbeat is surfaced so the UI can offer a (safe, + # idempotent) retry. + if status.get("state") == "running": + try: + updated = datetime.fromisoformat(status["updated_at"]) + age = (datetime.now(timezone.utc) - updated).total_seconds() + if age > _SYNC_STALL_SECONDS: + status = dict(status) + status["state"] = "stalled" + except (KeyError, ValueError): + pass + + msg = Message("ok") + msg.status = status + return msg, 200 + + +# --------------------------------------------------------------------------- +# Email HTML rendering (shell adapted from volunteers_service._send_email_to_user) +# --------------------------------------------------------------------------- + +_EMAIL_FOOTER_HTML = """ + +
+

💚 Support Our Mission

+

Just $17 feeds a hacker building solutions for nonprofits!

+ +

Corporate employees: Find us on Benevity • 501(c)(3) tax-deductible

+
+ + +
+

🌟 Stay Connected

+
+ Instagram | + LinkedIn | + Slack | + GitHub | + Threads | + Facebook +
+

Help us reach more people - share our mission! 🚀

+
+""" + +_UNSUBSCRIBE_FOOTER = f""" +
+

+ You are receiving this because you are part of the Opportunity Hack community. + Unsubscribe +

+""" + +_GREETING_MAP = { + "mentor": "Dear Mentor", + "sponsor": "Dear Sponsor", + "judge": "Dear Judge", + "hacker": "Dear Participant", + "volunteer": "Dear Volunteer", + "community": "Hello", +} + + +def _markdown_to_html(message: str) -> str: + import html as html_lib + try: + return markdown.markdown(message, extensions=["nl2br", "fenced_code"]) + except Exception as markdown_error: + logger.warning("Failed to convert markdown, falling back to basic formatting: %s", + markdown_error) + return html_lib.escape(message).replace("\n", "
") + + +def render_message_html(message: str, name: str, recipient_type: str) -> str: + """Per-recipient shell for batch sends — mirrors _send_email_to_user's look.""" + import html as html_lib + greeting = _GREETING_MAP.get((recipient_type or "").lower(), "Hello") + formatted_message = _markdown_to_html(message) + return f""" +

{greeting} {html_lib.escape(name or '')},

+

You have received a message from the Opportunity Hack team:

+
+

{formatted_message}

+
+

Best regards,
The Opportunity Hack Team

+ {_EMAIL_FOOTER_HTML} + """ + + +def render_broadcast_html(body_markdown: str) -> str: + """Broadcast shell — no per-recipient greeting; unsubscribe link is + mandatory (Resend rejects broadcasts without it).""" + html = f""" + {_markdown_to_html(body_markdown)} +

Best regards,
The Opportunity Hack Team

+ {_EMAIL_FOOTER_HTML} + """ + if UNSUBSCRIBE_PLACEHOLDER not in html: + html += _UNSUBSCRIBE_FOOTER + return html + + +# --------------------------------------------------------------------------- +# Broadcasts +# --------------------------------------------------------------------------- + +def _broadcast_from_domains() -> List[str]: + raw = os.environ.get("RESEND_BROADCAST_FROM_DOMAINS", DEFAULT_BROADCAST_FROM_DOMAINS) + return [d.strip().lower() for d in raw.split(",") if d.strip()] + + +def _from_address_domain(address: str) -> Optional[str]: + match = re.search(r"@([A-Za-z0-9.-]+)>?\s*$", address or "") + return match.group(1).lower() if match else None + + +def _resolve_from_address(requested: Optional[str]) -> Tuple[Optional[str], Optional[str]]: + """Returns (from_address, error). Requested addresses must be on a + verified domain from the allowlist (notifs.ohack.org is deliberately NOT + allowlisted — its Resend verification is partially failed).""" + from_address = (requested or "").strip() or os.environ.get( + "RESEND_BROADCAST_FROM", DEFAULT_BROADCAST_FROM) + domain = _from_address_domain(from_address) + if not domain: + return None, f"Could not parse a domain from from_address: {from_address}" + if domain not in _broadcast_from_domains(): + return None, (f"from_address domain '{domain}' is not in the allowed list " + f"({', '.join(_broadcast_from_domains())})") + return from_address, None + + +def _broadcasts_api(): + api = getattr(resend, "Broadcasts", None) + if api is None: + raise RuntimeError( + "The installed resend SDK has no Broadcasts API — upgrade to the " + "version pinned in requirements.txt (pip install -U resend)" + ) + return api + + +def _serialize_broadcast(b) -> dict: + return { + "id": _field(b, "id"), + "name": _field(b, "name"), + "subject": _field(b, "subject"), + "status": _field(b, "status"), + "segment_id": _field(b, "segment_id") or _field(b, "audience_id"), + "created_at": _field(b, "created_at"), + "scheduled_at": _field(b, "scheduled_at"), + "sent_at": _field(b, "sent_at"), + } + + +def create_broadcast(payload: dict, actor: Optional[dict]) -> Tuple[Message, int]: + payload = payload or {} + segment_id = (payload.get("segment_id") or "").strip() + subject = (payload.get("subject") or "").strip() + body_markdown = (payload.get("body_markdown") or "").strip() + if not segment_id: + return Message("segment_id is required"), 400 + if not subject: + return Message("subject is required"), 400 + if not body_markdown: + return Message("body_markdown is required"), 400 + + from_address, from_error = _resolve_from_address(payload.get("from_address")) + if from_error: + return Message(from_error), 400 + + html = render_broadcast_html(body_markdown) + send_now = bool(payload.get("send")) + scheduled_at = (payload.get("scheduled_at") or "").strip() or None + + if _notifications_disabled(): + msg = Message("Broadcast simulated (notifications disabled)") + msg.broadcast = {"id": None, "status": "simulated", "segment_id": segment_id, + "subject": subject} + msg.simulated = True + return msg, 200 + + resend.api_key = _resend_full_key() + params = { + "from": from_address, + "segment_id": segment_id, + "subject": subject, + "html": html, + "name": (payload.get("name") or subject)[:120], + "reply_to": payload.get("reply_to") or DEFAULT_REPLY_TO, + } + if send_now: + params["send"] = True + if scheduled_at: + params["scheduled_at"] = scheduled_at + + api = _broadcasts_api() + try: + created = api.create(params) + except Exception as first_error: + # Older SDK/API combos only accept the deprecated audience_id name. + params.pop("segment_id", None) + params["audience_id"] = segment_id + try: + created = api.create(params) + except Exception: + raise first_error + + broadcast_id = _field(created, "id") + send_slack_audit( + action="broadcast_created", + message=f"Resend broadcast {'sent' if send_now else 'drafted'}: " + f"subject='{subject}' segment={segment_id} id={broadcast_id}", + payload={"requested_by": (actor or {}).get("email"), "from": from_address}, + ) + + msg = Message("Broadcast sent" if send_now else "Broadcast draft created") + msg.broadcast = {"id": broadcast_id, "status": "sent" if send_now else "draft", + "segment_id": segment_id, "subject": subject, "from": from_address} + return msg, 201 + + +def send_broadcast(broadcast_id: str, payload: dict, actor: Optional[dict]) -> Tuple[Message, int]: + if not broadcast_id: + return Message("broadcast_id is required"), 400 + + if _notifications_disabled(): + msg = Message("Broadcast send simulated (notifications disabled)") + msg.simulated = True + return msg, 200 + + resend.api_key = _resend_full_key() + params = {"broadcast_id": broadcast_id} + scheduled_at = ((payload or {}).get("scheduled_at") or "").strip() + if scheduled_at: + params["scheduled_at"] = scheduled_at + result = _broadcasts_api().send(params) + + send_slack_audit( + action="broadcast_sent", + message=f"Resend broadcast sent: id={broadcast_id}", + payload={"requested_by": (actor or {}).get("email")}, + ) + msg = Message("Broadcast sent") + msg.broadcast = {"id": _field(result, "id") or broadcast_id} + return msg, 200 + + +def list_broadcasts() -> Tuple[Message, int]: + resend.api_key = _resend_full_key() + listed = _broadcasts_api().list() + msg = Message("ok") + msg.broadcasts = [_serialize_broadcast(b) for b in _resp_data(listed)] + return msg, 200 + + +def get_broadcast(broadcast_id: str) -> Tuple[Message, int]: + resend.api_key = _resend_full_key() + b = _broadcasts_api().get(broadcast_id) + msg = Message("ok") + msg.broadcast = _serialize_broadcast(b) + return msg, 200 + + +# --------------------------------------------------------------------------- +# Transactional batch send (Resend /emails/batch — 100 per call) +# --------------------------------------------------------------------------- + +def batch_send_emails(payload: dict, actor: Optional[dict]) -> Tuple[Message, int]: + """Send pre-personalized messages to email-only recipients in chunks of + 100 through the Resend Batch API (transactional quota). Recipients: + [{email, name, message}]. Messages with [QRCode:...] markers are rejected + per-recipient — the Batch API doesn't support attachments; the frontend + routes those down the per-recipient path.""" + payload = payload or {} + subject = (payload.get("subject") or "").strip() + recipients = payload.get("recipients") or [] + recipient_type = (payload.get("recipient_type") or "community").strip() + + if not subject: + return Message("subject is required"), 400 + if not isinstance(recipients, list) or not recipients: + return Message("recipients is required"), 400 + if len(recipients) > MAX_BATCH_RECIPIENTS_PER_REQUEST: + return Message( + f"Too many recipients in one request (max {MAX_BATCH_RECIPIENTS_PER_REQUEST}); " + "send in chunks"), 400 + + email_subject = f"{subject} - Message from Opportunity Hack Team" + results: List[dict] = [] + entries: List[Optional[dict]] = [] # params per recipient; None = pre-failed + + for r in recipients: + r = r or {} + email = _norm_email(r.get("email")) + message = r.get("message") or "" + if not email: + results.append({"email": r.get("email"), "success": False, + "error": "invalid email"}) + entries.append(None) + continue + if QR_MARKER_RE.search(message): + results.append({"email": email, "success": False, + "error": "QR-code messages are not supported by batch send — " + "use the per-recipient path"}) + entries.append(None) + continue + results.append({"email": email, "success": False, "error": None}) + entries.append({ + "from": "Opportunity Hack ", + "to": [email], + "reply_to": DEFAULT_REPLY_TO, + "subject": email_subject, + "html": render_message_html(message, r.get("name") or email, recipient_type), + }) + + simulated = _notifications_disabled() + if not simulated: + resend.api_key = _resend_send_key() + batch_api = getattr(resend, "Batch", None) + + pending = [(i, e) for i, e in enumerate(entries) if e is not None] + for offset in range(0, len(pending), RESEND_BATCH_CHUNK): + chunk = pending[offset:offset + RESEND_BATCH_CHUNK] + try: + if batch_api is not None: + resp = batch_api.send([e for _, e in chunk]) + ids = [_field(item, "id") for item in _resp_data(resp)] + for (i, _), rid in zip(chunk, ids + [None] * len(chunk)): + results[i].update({"success": True, "resend_id": rid, "error": None}) + else: + # Dev fallback for SDKs predating Batch: sequential sends. + for i, entry in chunk: + sent = resend.Emails.send(entry) + results[i].update({"success": True, + "resend_id": _field(sent, "id"), "error": None}) + except Exception as e: + logger.error("batch send chunk failed: %s", e) + for i, _ in chunk: + results[i].update({"success": False, "error": str(e)}) + else: + for i, e in enumerate(entries): + if e is not None: + results[i].update({"success": True, "simulated": True, "error": None}) + + successful = sum(1 for r in results if r["success"]) + summary = { + "total": len(results), + "successful": successful, + "failed": len(results) - successful, + "simulated": simulated, + } + send_slack_audit( + action="admin_batch_email_send", + message=f"Batch email send: subject='{subject}' " + f"{successful}/{len(results)} successful (simulated={simulated})", + payload={"requested_by": (actor or {}).get("email"), + "recipient_type": recipient_type}, + ) + + msg = Message("Batch send complete") + msg.results = results + msg.summary = summary + return msg, 200