import logging from collections import defaultdict from collections.abc import Sequence from concurrent.futures import ThreadPoolExecutor from datetime import datetime from fansifter_common.utils import timezone from app import lock from app.adapters.aws import lambda_client from app.adapters.db import db from app.adapters.ows_dmp import ows_dmp_client from app.adapters.ows_text_campaigns import ows_text_campaigns_client from app.config import settings from app.enums import CampaignCancelReason, CampaignStatus from app.models import AudienceTextFan, BatchRecipient, Campaign, CampaignBatch from app.types import BatchGate, BatchSend, CampaignSendCountry, CampaignSendTimezone MAX_CONCURRENT_QUIET_HOURS_LOOKUPS = 10 logger = logging.getLogger(__name__) def handle() -> Sequence[BatchSend]: dispatch_time = timezone.now() with db.session_factory(): campaigns = Campaign.query.find_ready_to_dispatch() if not campaigns: logger.info("No campaigns are ready to dispatch at this time.") return [] logger.info( "Found %d campaign(s) ready to dispatch at %s.", len(campaigns), dispatch_time.isoformat(), ) for campaign in campaigns: try: _prepare_campaign(campaign) except Exception as exc: logger.error( "Failed to prepare campaign `%s` for dispatch.", campaign.name, exc_info=exc, extra={"campaign_id": campaign.id}, ) with db.session_factory(): batch_sends = _plan_batch_sends(dispatch_time=dispatch_time) logger.info("Planned %d batch send(s) for dispatch.", len(batch_sends)) for batch_send in batch_sends: try: _dispatch_batch(batch_send) except Exception as exc: logger.error( "Failed to dispatch batch `%s`.", batch_send.batch_id, exc_info=exc, extra={ "campaign_id": batch_send.campaign_id, "batch_id": batch_send.batch_id, }, ) return batch_sends def _prepare_campaign(campaign: Campaign) -> None: """Split a campaign's audience into batches grouped by (country, state).""" if campaign.is_prepared: return if campaign.audience_id is None: logger.error( "Campaign `%s` is missing an audience; cannot proceed.", campaign.name, extra={"campaign_id": campaign.id}, ) return try: fan_count = ows_dmp_client.upsert_audience_fans( audience_id=campaign.audience_id ) except Exception as exc: logger.error( "Failed to refresh audience fans for campaign `%s`; will retry next tick.", campaign.name, exc_info=exc, extra={"campaign_id": campaign.id}, ) return if fan_count <= 0: with db.transaction(): _cancel_with_no_recipients(campaign, reason=CampaignCancelReason.NO_FANS) return channel = campaign.channel.as_literal() with db.session_factory(): groups = AudienceTextFan.query.count_by_group( audience_id=campaign.audience_id, channel=channel ) if not groups: with db.transaction(): _cancel_with_no_recipients(campaign, reason=CampaignCancelReason.NO_FANS) return # Defense-in-depth: drop prohibited channel/country groups before they enter # the pipeline. The authoritative rule is enforced upstream. blocked_countries = settings.channel_country_blocklist.get(channel, frozenset()) if blocked_countries: blocked = [g for g in groups if g.country_code in blocked_countries] if blocked: logger.warning( "Excluding %d recipient(s) across %d group(s) from campaign `%s`: " "channel `%s` is not allowed in %s.", sum(g.recipient_count for g in blocked), len(blocked), campaign.name, channel, sorted({g.country_code for g in blocked}), extra={"campaign_id": campaign.id, "channel": channel}, ) groups = [g for g in groups if g.country_code not in blocked_countries] if not groups: with db.transaction(): _cancel_with_no_recipients( campaign, reason=CampaignCancelReason.CHANNEL_NOT_ALLOWED ) return with db.transaction(): recipient_count = 0 for group in groups: batch = CampaignBatch( campaign_id=campaign.id, country_code=group.country_code, state_province=group.state_province, batch_size=group.recipient_count, ) batch.save(flush=True) inserted = BatchRecipient.query.bulk_create_from_audience( batch_id=batch.id, audience_id=campaign.audience_id, channel=channel, country_code=group.country_code, state_province=group.state_province, ) batch.batch_size = inserted recipient_count += inserted campaign.status = CampaignStatus.IN_PROGRESS campaign.recipients_count = recipient_count campaign.prepared_at = timezone.now() campaign.save() logger.info( "Campaign `%s` prepared for sending to %d recipient(s) across %d group(s).", campaign.name, recipient_count, len(groups), extra={ "campaign_id": campaign.id, "recipient_count": recipient_count, "group_count": len(groups), }, ) def _cancel_with_no_recipients( campaign: Campaign, *, reason: CampaignCancelReason ) -> None: logger.error( "Cancelling campaign `%s`: no sendable recipients (%s).", campaign.name, reason, extra={"campaign_id": campaign.id}, ) campaign.cancel(reason) campaign.save() def _plan_batch_sends(*, dispatch_time: datetime) -> list[BatchSend]: """Gate active batches by quiet hours, then cap how much each may send now.""" active_batches = CampaignBatch.query.active().all() if not active_batches: logger.info("No active campaign batches are available for dispatch.") return [] batches_by_campaign: dict[str, list[CampaignBatch]] = defaultdict(list) for batch in active_batches: batches_by_campaign[batch.campaign_id].append(batch) countries_by_campaign = _fetch_send_timezones( campaign_ids=list(batches_by_campaign), dispatch_time=dispatch_time ) batch_sends: list[BatchSend] = [] for campaign_id, batches in batches_by_campaign.items(): countries = countries_by_campaign.get(campaign_id) if countries is None: # Fail closed: couldn't confirm quiet hours, so don't send # (already logged in `_fetch_send_timezones`). continue timezones_by_country: dict[str, list[CampaignSendTimezone]] = defaultdict(list) for country in countries: timezones_by_country[country.country_code].extend(country.timezones) for batch in batches: gate = _resolve_batch_gate( batch, timezones_by_country.get(batch.country_code, []) ) if not gate.is_sendable: continue messages_to_send = min( batch.batch_size - batch.batch_offset, settings.max_recipients_per_batch_send, ) if messages_to_send <= 0: continue batch_sends.append( BatchSend( batch_id=batch.id, campaign_id=campaign_id, country_code=batch.country_code, state_province=batch.state_province, messages_to_send=messages_to_send, safe_until=gate.safe_until, ) ) logger.info( "Batch send planning complete: %d total batch send(s) generated.", len(batch_sends), ) return batch_sends def _fetch_send_timezones( *, campaign_ids: list[str], dispatch_time: datetime ) -> dict[str, list[CampaignSendCountry]]: """Fetch quiet-hours status for all campaigns concurrently.""" def _fetch_one(campaign_id: str) -> tuple[str, list[CampaignSendCountry] | None]: try: return campaign_id, ows_text_campaigns_client.get_send_timezones( campaign_id=campaign_id, send_at=dispatch_time ) except Exception as exc: logger.error( "Failed to fetch quiet-hours status for campaign `%s`; " "skipping its batches this tick.", campaign_id, exc_info=exc, extra={"campaign_id": campaign_id}, ) return campaign_id, None max_workers = min(len(campaign_ids), MAX_CONCURRENT_QUIET_HOURS_LOOKUPS) with ThreadPoolExecutor(max_workers=max_workers) as executor: results = dict(executor.map(_fetch_one, campaign_ids)) return { campaign_id: countries for campaign_id, countries in results.items() if countries is not None } def _resolve_batch_gate( batch: CampaignBatch, timezones: list[CampaignSendTimezone] ) -> BatchGate: if not timezones: # No compliance rule resolved for this country -- fail closed. logger.warning( "No compliance rule resolved for country `%s`; skipping batch `%s`.", batch.country_code, batch.id, extra={"batch_id": batch.id, "country_code": batch.country_code}, ) return BatchGate(is_sendable=False, safe_until=None) matching = [ tz for tz in timezones if batch.state_province is None or tz.state_provinces is None or batch.state_province in tz.state_provinces ] if not matching: # State not covered by any resolved rule -- require the whole country safe. matching = timezones if not all(tz.is_sendable for tz in matching): return BatchGate(is_sendable=False, safe_until=None) # Earliest close across matching groups, so a sender never overruns any of them. closes = [close for tz in matching if (close := tz.safe_until) is not None] return BatchGate(is_sendable=True, safe_until=min(closes) if closes else None) def _dispatch_batch(batch_send: BatchSend) -> None: if not lock.acquire(batch_send.batch_id): logger.warning( "Batch `%s` is already being processed by another lambda; skipping.", batch_send.batch_id, extra={ "campaign_id": batch_send.campaign_id, "batch_id": batch_send.batch_id, }, ) return safe_until = batch_send.safe_until.isoformat() if batch_send.safe_until else None invoked = lambda_client.invoke( function_name=settings.sender_lambda_function_name, invocation_type="Event", data={ "batch_id": batch_send.batch_id, "campaign_id": batch_send.campaign_id, "messages_to_send": batch_send.messages_to_send, "safe_until": safe_until, }, ) log_extra = { "campaign_id": batch_send.campaign_id, "batch_id": batch_send.batch_id, "messages_to_send": batch_send.messages_to_send, "safe_until": safe_until, } if not invoked: # Nothing is actually processing this batch -- release so the next # tick can retry immediately instead of waiting out the full TTL. lock.release(batch_send.batch_id) logger.error( "Failed to invoke sender lambda for batch `%s`.", batch_send.batch_id, extra=log_extra, ) else: logger.info( "Successfully invoked sender lambda for batch `%s` (%d messages).", batch_send.batch_id, batch_send.messages_to_send, extra=log_extra, )