import logging from collections import defaultdict from collections.abc import Sequence from datetime import datetime import redis from anydi import singleton from fansifter_common.utils import timezone from app import quotacalc from app.adapters.aws import LambdaClient from app.adapters.db import Database from app.adapters.ows_dmp import OwsDmpClient from app.config import Settings from app.enums import CampaignCancelReason, CampaignStatus from app.models import Campaign, CampaignBatch, DeliveryQuota from app.repositories import ( BatchRecipientRepository, CampaignBatchRepository, CampaignRepository, DeliveryQuotaDefaultRepository, DeliveryQuotaGrowthRateRepository, DeliveryQuotaRepository, ) from app.strategies import QuotaAllocationStrategy from app.types import BatchSend logger = logging.getLogger(__name__) @singleton class DispatchCampaignsHandler: """Handles the complete process of identifying, preparing, planning, and dispatching email campaigns.""" def __init__( self, db: Database, redis_client: redis.Redis, lambda_client: LambdaClient, ows_dmp_client: OwsDmpClient, campaign_repository: CampaignRepository, batch_repository: CampaignBatchRepository, batch_recipient_repository: BatchRecipientRepository, quota_repository: DeliveryQuotaRepository, quota_default_repository: DeliveryQuotaDefaultRepository, quota_growth_rate_repository: DeliveryQuotaGrowthRateRepository, quota_strategy: QuotaAllocationStrategy, settings: Settings, ) -> None: """Initialize the dispatch handler with required dependencies.""" self.db = db self.redis_client = redis_client self.lambda_client = lambda_client self.ows_dmp_client = ows_dmp_client self.campaign_repository = campaign_repository self.batch_repository = batch_repository self.batch_recipient_repository = batch_recipient_repository self.quota_repository = quota_repository self.quota_default_repository = quota_default_repository self.quota_growth_rate_repository = quota_growth_rate_repository self.quota_strategy = quota_strategy self.settings = settings logger.info( "Initialized dispatcher with quota strategy: %s", self.quota_strategy.strategy_type, extra={"quota_strategy": self.quota_strategy.strategy_type}, ) def handle(self) -> Sequence[BatchSend]: # Set current dispatch time dispatch_time = timezone.now() # Step 1: Fetch all campaigns ready for dispatch with self.db.session_factory(): campaigns = self.campaign_repository.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(), ) # Step 2: Prepare each campaign for dispatch (create batches, recipients, etc.) for campaign in campaigns: self.prepare_campaign_for_dispatch(campaign) # Step 3: Determine how much each batch can send based on quotas with self.db.session_factory(): batch_sends = self.plan_batch_sends(dispatch_time=dispatch_time) logger.info("Planned %d batch send(s) for dispatch.", len(batch_sends)) # Step 4: Trigger sending by invoking AWS Lambda for each batch for batch_send in batch_sends: lock_key = self.settings.redis_sender_lock_key.format( batch_id=batch_send.batch_id ) if self.redis_client.exists(lock_key): 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, }, ) continue logger.info( "Ready to send %d emails for provider `%s` (hourly_quota=%d).", batch_send.emails_to_send, batch_send.provider, batch_send.current_hourly_quota, extra={ "batch_id": batch_send.batch_id, "batch_size": batch_send.batch_size, "batch_offset": batch_send.batch_offset, "campaign_id": batch_send.campaign_id, "provider": batch_send.provider, "emails_to_send": batch_send.emails_to_send, "current_hourly_quota": batch_send.current_hourly_quota, }, ) invoked = self.lambda_client.invoke( function_name=self.settings.sender_lambda_function_name, invocation_type="Event", data={ "batch_id": batch_send.batch_id, "campaign_id": batch_send.campaign_id, "emails_to_send": batch_send.emails_to_send, "current_hourly_quota": batch_send.current_hourly_quota, "dispatch_time": dispatch_time.isoformat(), }, ) if not invoked: logger.error( "Failed to invoke sender lambda for provider `%s`.", batch_send.provider, extra={ "batch_id": batch_send.batch_id, "batch_size": batch_send.batch_size, "batch_offset": batch_send.batch_offset, "campaign_id": batch_send.campaign_id, "provider": batch_send.provider, "lambda_function": self.settings.sender_lambda_function_name, }, ) else: logger.info( "Successfully invoked sender lambda for provider `%s`.", batch_send.provider, extra={ "batch_id": batch_send.batch_id, "batch_size": batch_send.batch_size, "batch_offset": batch_send.batch_offset, "campaign_id": batch_send.campaign_id, "provider": batch_send.provider, "lambda_function": self.settings.sender_lambda_function_name, }, ) return batch_sends def prepare_campaign_for_dispatch(self, campaign: Campaign) -> None: """Prepare a campaign for sending.""" if campaign.is_prepared: with self.db.transaction(): self.campaign_repository.update_as_sent(campaign.id) logger.info( "Skipping preparation for campaign `%s`; already prepared.", campaign.name, extra={"campaign_id": campaign.id}, ) return None if campaign.audience_id is None: logger.error( "Campaign `%s` is missing an audience; cannot proceed.", campaign.name, extra={"campaign_id": campaign.id}, ) return None # Attempt to fetch or create audience fans try: fan_count = self.ows_dmp_client.upsert_audience_fans( audience_id=campaign.audience_id ) except Exception as exc: logger.error( "Failed to create target audience fans for campaign `%s`.", campaign.name, extra={"campaign_id": campaign.id}, exc_info=exc, ) return None # Begin the database transaction only after the audience # fans upsert completes successfully with self.db.transaction(): if fan_count <= 0: self.cancel_with_no_recipients(campaign) return None # Create campaign batches self.batch_repository.create_bulk( campaign_id=campaign.id, fansifter_only=self.settings.dispatch_fansifter_only, ) # Create batch recipients recipient_count = self.batch_recipient_repository.create_bulk( campaign_id=campaign.id, fansifter_only=self.settings.dispatch_fansifter_only, ) if recipient_count <= 0: self.cancel_with_no_recipients(campaign) return None campaign.status = CampaignStatus.IN_PROGRESS campaign.recipient_count = recipient_count campaign.prepared_at = timezone.now() self.campaign_repository.save(campaign) logger.info( "Campaign `%s` prepared for sending to %d recipients.", campaign.name, recipient_count, extra={ "campaign_id": campaign.id, "recipient_count": recipient_count, "started_at": campaign.prepared_at, }, ) return None def plan_batch_sends(self, *, dispatch_time: datetime) -> Sequence[BatchSend]: """ Compute how many emails each batch can send based on its domain/provider quotas. """ active_batches = self.batch_repository.find_active() active_batches_for_quota: dict[tuple[str, str], list[CampaignBatch]] = ( defaultdict(list) ) for batch in active_batches: active_batches_for_quota[(batch.domain_id, batch.provider)].append(batch) if not active_batches: logger.info("No active campaign batches are available for dispatch.") return [] logger.info( "Planning batch sends for %d active batch(es) across %d quota group(s).", len(active_batches), len(active_batches_for_quota), ) batch_sends: list[BatchSend] = [] # Group batches by (domain_id, provider) to apply delivery quotas for (domain_id, provider), batches in active_batches_for_quota.items(): delivery_quota = self.get_delivery_quota( domain_id=domain_id, provider=provider ) growth_rate = self.get_growth_rate(domain_id=domain_id, provider=provider) current_hourly_quota = quotacalc.calculate_current_hourly_quota( last_sent_at=delivery_quota.last_sent_at, last_hourly_quota=delivery_quota.last_hourly_quota, current_time=dispatch_time, growth_rate=growth_rate, use_cooldown=self.settings.use_quota_cooldown, ) available_quota = quotacalc.calculate_available_quota( last_sent_at=delivery_quota.last_sent_at, last_hour_sent_emails=delivery_quota.last_hour_sent_emails, current_hourly_quota=current_hourly_quota, current_time=dispatch_time, ) if available_quota <= 0: logger.warning( "Quota exhausted for domain `%s` with provider `%s`; skipping batches.", domain_id, provider, extra={ "domain_id": domain_id, "provider": provider, "available_quota": available_quota, }, ) continue logger.debug( "Allocating %d available emails across %d batch(es) using strategy=%s (domain=%s, provider=%s).", available_quota, len(batches), self.quota_strategy.strategy_type, domain_id, provider, extra={ "domain_id": domain_id, "provider": provider, "available_quota": available_quota, "strategy": self.quota_strategy.strategy_type, }, ) # Delegate allocation to strategy allocations = self.quota_strategy.allocate_quota( batches=batches, available_quota=int(available_quota), current_time=dispatch_time, ) # Convert strategy allocations to BatchSend objects batches_by_id = {batch.id: batch for batch in batches} for allocation in allocations: batch = batches_by_id[allocation.batch_id] batch_sends.append( BatchSend( batch_id=allocation.batch_id, batch_size=batch.batch_size, batch_offset=batch.batch_offset, campaign_id=allocation.campaign_id, provider=provider, emails_to_send=allocation.emails_to_send, current_hourly_quota=current_hourly_quota, ) ) logger.debug( "Strategy allocated %d email(s) for batch_id=%d (campaign=%s).", allocation.emails_to_send, allocation.batch_id, allocation.campaign_id, extra={ "domain_id": domain_id, "provider": provider, "batch_id": allocation.batch_id, "batch_size": batch.batch_size, "batch_offset": batch.batch_offset, "campaign_id": allocation.campaign_id, "emails_to_send": allocation.emails_to_send, }, ) logger.info( "Batch send planning complete: %d total batch send(s) generated.", len(batch_sends), ) return batch_sends def cancel_with_no_recipients(self, campaign: Campaign) -> None: """Cancel a campaign when no valid recipients are found.""" logger.error( "Cancelling campaign `%s`: no recipients found.", campaign.name, extra={"campaign_id": campaign.id}, ) campaign.status = CampaignStatus.CANCELLED campaign.cancelled_at = timezone.now() campaign.cancel_reason = CampaignCancelReason.NO_FANS self.campaign_repository.save(campaign) return None def get_delivery_quota(self, *, domain_id: str, provider: str) -> DeliveryQuota: """ Retrieve or initialize a delivery quota record for a given domain and provider. """ delivery_quota = self.quota_repository.get_by_domain_id_and_provider( domain_id=domain_id, provider=provider ) if delivery_quota is not None: return delivery_quota # Fallback: use default quota values if no record exists delivery_quota = DeliveryQuota(domain_id=domain_id, provider=provider) delivery_quota_default = ( self.quota_default_repository.get_by_domain_id_and_provider( domain_id=domain_id, provider=provider ) ) if delivery_quota_default is not None: delivery_quota.last_sent_at = None delivery_quota.last_hourly_quota = delivery_quota_default.last_hourly_quota delivery_quota.last_hour_sent_emails = 0 return delivery_quota def get_growth_rate(self, *, domain_id: str, provider: str) -> float | None: """Fetch the delivery quota growth rate for a given domain and provider, if available.""" delivery_quota_growth_rate = ( self.quota_growth_rate_repository.get_by_domain_id_and_provider( domain_id=domain_id, provider=provider ) ) return ( delivery_quota_growth_rate.growth_rate if delivery_quota_growth_rate else None )