"""Privacy-bounded transactional licensing email with durable delivery state.""" from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timedelta, timezone from email.headerregistry import Address from email.message import EmailMessage from enum import Enum import hashlib import hmac import json import smtplib import ssl from typing import Optional, Protocol from urllib.parse import urlparse from sqlalchemy import or_, select from sqlalchemy.orm import Session from licensing_shared.catalog import LicensingCatalog from licensing_shared.constants import validate_identifier from .constants import NOTIFICATION_PEPPER_MINIMUM_BYTES from .models import ( AuditEvent, License, Membership, NotificationDelivery, NotificationFeedbackEvent, NotificationSuppression, OutboxEvent, Subscription, SubscriptionItem, User, ) from .security import new_identifier CUSTOMER_NOTIFICATION_OUTBOX_EVENT_TYPE = "customer.notification.send" CUSTOMER_NOTIFICATION_SCHEMA = "apolon.licensing.customer-notification" CUSTOMER_NOTIFICATION_SCHEMA_VERSION = 1 DEFAULT_NOTIFICATION_MAXIMUM_AGE_DAYS = 7 DEFAULT_NOTIFICATION_CLAIM_TIMEOUT_MINUTES = 15 DEFAULT_NOTIFICATION_FEEDBACK_RETENTION_DAYS = 365 DEFAULT_SMTP_TIMEOUT_SECONDS = 15.0 MAXIMUM_NOTIFICATION_EVENTS = 1000 MAXIMUM_NOTIFICATION_AGE_DAYS = 30 MAXIMUM_NOTIFICATION_RETRY_ATTEMPTS = 100 MAXIMUM_NOTIFICATION_FEEDBACK_RETENTION_DAYS = 3650 MAXIMUM_MESSAGE_CHARACTERS = 32 * 1024 MAXIMUM_PROVIDER_MESSAGE_ID_CHARACTERS = 256 NOTIFICATION_FEEDBACK_FUTURE_TOLERANCE_MINUTES = 5 NOTIFICATION_RECIPIENT_DIGEST_DOMAIN = b"APOLON-NOTIFICATION-RECIPIENT-V1\x00" SMTP_TLS_MODES = frozenset(("implicit", "starttls")) NOTIFICATION_SUPPRESSION_CLEAR_REASONS = frozenset( ( "address_corrected", "customer_request", "provider_correction", "support_correction", ) ) class CustomerNotificationKind(str, Enum): SUBSCRIPTION_STARTED = "subscription_started" PAYMENT_FAILED = "payment_failed" ACCESS_SUSPENDED = "access_suspended" PAYMENT_RECOVERED = "payment_recovered" CANCELLATION_SCHEDULED = "cancellation_scheduled" CANCELLATION_REVERSED = "cancellation_reversed" SUBSCRIPTION_CANCELED = "subscription_canceled" DISPUTE_OPEN = "dispute_open" DISPUTE_LOST = "dispute_lost" DISPUTE_RESOLVED = "dispute_resolved" REFUNDED = "refunded" class NotificationFeedbackType(str, Enum): SOFT_BOUNCE = "soft_bounce" HARD_BOUNCE = "hard_bounce" COMPLAINT = "complaint" @dataclass(frozen=True) class CustomerNotificationMessage: recipient: str subject: str body: str idempotency_key: str notification_kind: CustomerNotificationKind def __post_init__(self) -> None: object.__setattr__(self, "recipient", _email_address(self.recipient, "recipient")) object.__setattr__(self, "subject", _header_text(self.subject, "subject", 256)) if not isinstance(self.body, str) or not self.body.strip(): raise ValueError("body must be a non-empty string") if len(self.body) > MAXIMUM_MESSAGE_CHARACTERS: raise ValueError("body is too long") object.__setattr__(self, "body", self.body.strip() + "\n") object.__setattr__( self, "idempotency_key", validate_identifier(self.idempotency_key, "idempotency_key"), ) if not isinstance(self.notification_kind, CustomerNotificationKind): raise TypeError("notification_kind must be CustomerNotificationKind") @dataclass(frozen=True) class NotificationSendResult: provider_message_id: str def __post_init__(self) -> None: object.__setattr__( self, "provider_message_id", _header_text( self.provider_message_id, "provider_message_id", MAXIMUM_PROVIDER_MESSAGE_ID_CHARACTERS, ), ) class NotificationProviderFailure(RuntimeError): def __init__(self, error_code: str) -> None: self.error_code = validate_identifier(error_code, "error_code") super().__init__(self.error_code) class CustomerNotificationProvider(Protocol): def send(self, message: CustomerNotificationMessage) -> NotificationSendResult: ... class SmtpCustomerNotificationProvider: def __init__( self, host: str, port: int, tls_mode: str, username: str, password: str, from_address: str, timeout_seconds: float = DEFAULT_SMTP_TIMEOUT_SECONDS, ) -> None: self._host = _smtp_host(host) if isinstance(port, bool) or not isinstance(port, int): raise TypeError("port must be an integer") if port < 1 or port > 65535: raise ValueError("port must be between one and 65535") self._port = port if not isinstance(tls_mode, str) or tls_mode.strip().lower() not in SMTP_TLS_MODES: raise ValueError("tls_mode must be implicit or starttls") self._tls_mode = tls_mode.strip().lower() self._username = _bounded_text(username, "username", 4096) self._password = _bounded_text(password, "password", 4096) self._from_address = _email_address(from_address, "from_address") if isinstance(timeout_seconds, bool) or not isinstance(timeout_seconds, (int, float)): raise TypeError("timeout_seconds must be a number") if timeout_seconds <= 0.0 or timeout_seconds > 120.0: raise ValueError("timeout_seconds must be between zero and 120") self._timeout_seconds = float(timeout_seconds) def send(self, message: CustomerNotificationMessage) -> NotificationSendResult: if not isinstance(message, CustomerNotificationMessage): raise TypeError("message must be CustomerNotificationMessage") mail = EmailMessage() sender = Address(display_name="Apolon Licensing", addr_spec=self._from_address) message_digest = hashlib.sha256( message.idempotency_key.encode("utf-8") ).hexdigest() sender_domain = self._from_address.rsplit("@", 1)[1] provider_message_id = f"" mail["From"] = str(sender) mail["To"] = message.recipient mail["Subject"] = message.subject mail["Message-ID"] = provider_message_id mail["Auto-Submitted"] = "auto-generated" mail["X-Apolon-Notification"] = message.notification_kind.value mail["X-Apolon-Delivery-ID"] = message.idempotency_key mail.set_content(message.body) context = ssl.create_default_context() try: if self._tls_mode == "implicit": with smtplib.SMTP_SSL( self._host, self._port, timeout=self._timeout_seconds, context=context, ) as client: client.login(self._username, self._password) refused = client.send_message(mail) else: with smtplib.SMTP( self._host, self._port, timeout=self._timeout_seconds, ) as client: client.ehlo() client.starttls(context=context) client.ehlo() client.login(self._username, self._password) refused = client.send_message(mail) except (OSError, smtplib.SMTPException) as exc: raise NotificationProviderFailure("smtp_delivery_failed") from exc if refused: raise NotificationProviderFailure("smtp_recipient_refused") return NotificationSendResult(provider_message_id=provider_message_id) def enqueue_subscription_notification( session: Session, subscription_id: str, kind: CustomerNotificationKind, provider_event_id: str, *, now: datetime, ) -> bool: if not isinstance(session, Session): raise TypeError("session must be Session") normalized_subscription = validate_identifier(subscription_id, "subscription_id") if not isinstance(kind, CustomerNotificationKind): raise TypeError("kind must be CustomerNotificationKind") normalized_event = validate_identifier(provider_event_id, "provider_event_id") normalized_now = _utc(now) digest = hashlib.sha256( f"{normalized_subscription}\n{kind.value}\n{normalized_event}".encode("utf-8") ).hexdigest() outbox_id = f"notification_{digest}" if session.get(OutboxEvent, outbox_id) is not None: return False session.add( OutboxEvent( id=outbox_id, event_type=CUSTOMER_NOTIFICATION_OUTBOX_EVENT_TYPE, aggregate_type="subscription_notification", aggregate_id=normalized_subscription, payload_json={ "schema": CUSTOMER_NOTIFICATION_SCHEMA, "schemaVersion": CUSTOMER_NOTIFICATION_SCHEMA_VERSION, "notificationKind": kind.value, "subscriptionId": normalized_subscription, "providerEventId": normalized_event, "occurredAt": normalized_now.isoformat(), }, status="pending", available_at=normalized_now, attempts=0, processed_at=None, last_error_code=None, ) ) return True def subscription_transition_notification_kinds( previous_status: Optional[str], current_status: str, previous_cancel_at_period_end: bool, current_cancel_at_period_end: bool, ) -> tuple[CustomerNotificationKind, ...]: if previous_status is not None and not isinstance(previous_status, str): raise TypeError("previous_status must be a string or None") normalized_current = _bounded_text(current_status, "current_status", 32) for field_name, value in ( ("previous_cancel_at_period_end", previous_cancel_at_period_end), ("current_cancel_at_period_end", current_cancel_at_period_end), ): if not isinstance(value, bool): raise TypeError(f"{field_name} must be a Boolean") values: list[CustomerNotificationKind] = [] if previous_status is None and normalized_current in ("active", "trialing"): values.append(CustomerNotificationKind.SUBSCRIPTION_STARTED) elif previous_status != normalized_current: if normalized_current == "past_due": values.append(CustomerNotificationKind.PAYMENT_FAILED) elif normalized_current in ("paused", "unpaid", "incomplete_expired"): values.append(CustomerNotificationKind.ACCESS_SUSPENDED) elif normalized_current == "canceled": values.append(CustomerNotificationKind.SUBSCRIPTION_CANCELED) elif normalized_current in ("active", "trialing") and previous_status in ( "past_due", "paused", "unpaid", "incomplete", ): values.append(CustomerNotificationKind.PAYMENT_RECOVERED) if previous_cancel_at_period_end != current_cancel_at_period_end: values.append( CustomerNotificationKind.CANCELLATION_SCHEDULED if current_cancel_at_period_end else CustomerNotificationKind.CANCELLATION_REVERSED ) return tuple(values) def billing_hold_notification_kind( previous_hold: Optional[str], current_hold: Optional[str], ) -> Optional[CustomerNotificationKind]: allowed = (None, "dispute_open", "dispute_lost", "refunded") if previous_hold not in allowed or current_hold not in allowed: raise ValueError("billing hold is invalid") if previous_hold == current_hold: return None if current_hold == "dispute_open": return CustomerNotificationKind.DISPUTE_OPEN if current_hold == "dispute_lost": return CustomerNotificationKind.DISPUTE_LOST if current_hold == "refunded": return CustomerNotificationKind.REFUNDED if previous_hold in ("dispute_open", "dispute_lost"): return CustomerNotificationKind.DISPUTE_RESOLVED return None def process_customer_notification_outbox( session: Session, catalog: LicensingCatalog, provider: CustomerNotificationProvider, account_portal_url: str, *, now: datetime, maximum_events: int = 100, maximum_age_days: int = DEFAULT_NOTIFICATION_MAXIMUM_AGE_DAYS, notification_pepper: Optional[bytes] = None, ) -> int: if not isinstance(session, Session): raise TypeError("session must be Session") if not isinstance(catalog, LicensingCatalog): raise TypeError("catalog must be LicensingCatalog") if not callable(getattr(provider, "send", None)): raise TypeError("provider must expose send") normalized_url = _https_url(account_portal_url, "account_portal_url") normalized_now = _utc(now) limit = _bounded_integer(maximum_events, "maximum_events", MAXIMUM_NOTIFICATION_EVENTS) age_days = _bounded_integer( maximum_age_days, "maximum_age_days", MAXIMUM_NOTIFICATION_AGE_DAYS, ) normalized_pepper = _notification_pepper(notification_pepper) processed = 0 for _index in range(limit): with session.begin(): outbox = session.scalar( select(OutboxEvent) .where( OutboxEvent.event_type == CUSTOMER_NOTIFICATION_OUTBOX_EVENT_TYPE, OutboxEvent.status == "pending", OutboxEvent.available_at <= normalized_now, ) .order_by(OutboxEvent.created_at, OutboxEvent.id) .limit(1) .with_for_update(skip_locked=True) ) if outbox is None: break outbox.status = "processing" outbox.attempts += 1 outbox.last_error_code = None if _utc(outbox.created_at) < normalized_now - timedelta(days=age_days): outbox.status = "processed" outbox.processed_at = normalized_now outbox.last_error_code = "notification_expired" _audit_notification( session, outbox, "notification.skipped", {"errorCode": "notification_expired", "recipientCount": 0}, ) processed += 1 continue try: subscription = _subscription_from_outbox(session, outbox) except (TypeError, ValueError): outbox.status = "processed" outbox.processed_at = normalized_now outbox.last_error_code = "notification_payload_invalid" _audit_notification( session, outbox, "notification.skipped", { "errorCode": "notification_payload_invalid", "recipientCount": 0, }, ) processed += 1 continue recipients = _recipient_users(session, subscription.license_id, normalized_now) existing_user_ids = set( session.scalars( select(NotificationDelivery.user_id).where( NotificationDelivery.outbox_event_id == outbox.id ) ) ) for user in recipients: if user.id in existing_user_ids: continue session.add( NotificationDelivery( id=new_identifier("notification_delivery"), outbox_event_id=outbox.id, user_id=user.id, status="pending", attempts=0, provider_message_id=None, recipient_address_digest=None, last_error_code=None, delivered_at=None, feedback_status=None, feedback_at=None, ) ) session.flush() delivery_ids = tuple( session.scalars( select(NotificationDelivery.id) .where(NotificationDelivery.outbox_event_id == outbox.id) .order_by(NotificationDelivery.id) ) ) outbox_id = outbox.id if not delivery_ids: outbox.status = "processed" outbox.processed_at = normalized_now outbox.last_error_code = "notification_no_recipient" _audit_notification( session, outbox, "notification.skipped", {"errorCode": "notification_no_recipient", "recipientCount": 0}, ) processed += 1 continue for delivery_id in delivery_ids: _send_delivery( session, catalog, provider, normalized_url, delivery_id, normalized_now, normalized_pepper, ) with session.begin(): outbox = session.get(OutboxEvent, outbox_id) if outbox is None: continue deliveries = session.scalars( select(NotificationDelivery).where( NotificationDelivery.outbox_event_id == outbox.id ) ).all() failed = tuple(value for value in deliveries if value.status == "failed") delivered = tuple( value for value in deliveries if value.status == "processed" ) skipped = tuple(value for value in deliveries if value.status == "skipped") incomplete = tuple( value for value in deliveries if value.status in ("pending", "sending") ) if failed or incomplete: outbox.status = "failed" outbox.processed_at = normalized_now outbox.last_error_code = ( failed[0].last_error_code if failed else "notification_incomplete" ) _audit_notification( session, outbox, "notification.delivery_failed", { "deliveryCount": len(deliveries), "errorCode": outbox.last_error_code, }, ) continue outbox.status = "processed" outbox.processed_at = normalized_now outbox.last_error_code = ( None if delivered else "notification_all_recipients_skipped" ) _audit_notification( session, outbox, "notification.delivered" if delivered else "notification.skipped", { "deliveredCount": len(delivered), "deliveryCount": len(deliveries), "recipientCount": len(deliveries), "skippedCount": len(skipped), }, ) processed += 1 return processed def record_notification_feedback( session: Session, provider: str, provider_event_id: str, feedback_type: NotificationFeedbackType, *, occurred_at: datetime, now: datetime, delivery_id: Optional[str] = None, provider_message_id: Optional[str] = None, ) -> bool: """Record one already-authenticated provider feedback event. Provider-specific signature verification stays at the adapter boundary. This function deliberately accepts no email address or raw provider payload. """ if not isinstance(session, Session): raise TypeError("session must be Session") normalized_provider = validate_identifier(provider, "provider") normalized_event = validate_identifier(provider_event_id, "provider_event_id") if not isinstance(feedback_type, NotificationFeedbackType): raise TypeError("feedback_type must be NotificationFeedbackType") normalized_occurred = _utc(occurred_at) normalized_now = _utc(now) future_tolerance = timedelta( minutes=NOTIFICATION_FEEDBACK_FUTURE_TOLERANCE_MINUTES ) if normalized_occurred > normalized_now + future_tolerance: raise ValueError("feedback occurred_at is in the future") normalized_delivery = ( None if delivery_id is None else validate_identifier(delivery_id, "delivery_id") ) normalized_message = ( None if provider_message_id is None else _header_text( provider_message_id, "provider_message_id", MAXIMUM_PROVIDER_MESSAGE_ID_CHARACTERS, ) ) if normalized_delivery is None and normalized_message is None: raise ValueError("delivery_id or provider_message_id is required") with session.begin(): existing = session.scalar( select(NotificationFeedbackEvent).where( NotificationFeedbackEvent.provider == normalized_provider, NotificationFeedbackEvent.provider_event_id == normalized_event, ) ) if existing is not None: _validate_feedback_replay( existing, feedback_type, normalized_occurred, normalized_delivery, normalized_message, ) return False delivery = _feedback_delivery( session, normalized_delivery, normalized_message, ) if delivery.recipient_address_digest is None: raise ValueError("notification delivery has no recipient digest") if len(delivery.recipient_address_digest) != hashlib.sha256().digest_size: raise ValueError("notification delivery recipient digest is invalid") if normalized_occurred + future_tolerance < _utc(delivery.created_at): raise ValueError("feedback occurred before the notification delivery") locked_user_id = session.scalar( select(User.id) .where(User.id == delivery.user_id) .with_for_update() ) if locked_user_id is None: raise ValueError("notification delivery user was not found") effective_message = normalized_message or delivery.provider_message_id if effective_message is not None and delivery.provider_message_id is None: delivery.provider_message_id = effective_message feedback_digest = _feedback_payload_digest( normalized_provider, normalized_event, feedback_type, normalized_occurred, delivery.id, effective_message, ) feedback = NotificationFeedbackEvent( id=new_identifier("notification_feedback"), provider=normalized_provider, provider_event_id=normalized_event, delivery_id=delivery.id, delivery_reference_id=delivery.id, provider_message_id=effective_message, feedback_type=feedback_type.value, occurred_at=normalized_occurred, payload_digest=feedback_digest, processed_at=normalized_now, ) session.add(feedback) session.flush() current_priority = _feedback_priority(delivery.feedback_status) new_priority = _feedback_priority(feedback_type.value) if new_priority > current_priority or ( new_priority == current_priority and ( delivery.feedback_at is None or normalized_occurred >= _utc(delivery.feedback_at) ) ): delivery.feedback_status = feedback_type.value delivery.feedback_at = normalized_occurred suppression_active = feedback_type in ( NotificationFeedbackType.HARD_BOUNCE, NotificationFeedbackType.COMPLAINT, ) if suppression_active: suppression = session.scalar( select(NotificationSuppression).where( NotificationSuppression.user_id == delivery.user_id, NotificationSuppression.recipient_address_digest == delivery.recipient_address_digest, ) ) if suppression is None: session.add( NotificationSuppression( id=new_identifier("notification_suppression"), user_id=delivery.user_id, recipient_address_digest=delivery.recipient_address_digest, reason=feedback_type.value, source_feedback_id=feedback.id, active=True, suppressed_at=normalized_occurred, cleared_at=None, ) ) else: was_active = suppression.active suppression.active = True suppression.cleared_at = None if ( not was_active or feedback_type is NotificationFeedbackType.COMPLAINT or suppression.reason != NotificationFeedbackType.COMPLAINT.value ): suppression.reason = feedback_type.value suppression.source_feedback_id = feedback.id if not was_active: suppression.suppressed_at = normalized_occurred elif _utc(suppression.suppressed_at) > normalized_occurred: suppression.suppressed_at = normalized_occurred session.add( AuditEvent( id=new_identifier("audit"), actor_type="system", actor_id=None, action="notification.feedback_recorded", target_type="notification_feedback", target_id=feedback.id, reason=None, correlation_id=feedback.id, source_address_digest=None, metadata_json={ "feedbackType": feedback_type.value, "provider": normalized_provider, "suppressionActive": suppression_active, }, ) ) return True def clear_notification_suppression( session: Session, user_id: str, admin_user_id: str, reason: str, notification_pepper: bytes, *, now: datetime, ) -> bool: if not isinstance(session, Session): raise TypeError("session must be Session") normalized_user = validate_identifier(user_id, "user_id") normalized_admin = validate_identifier(admin_user_id, "admin_user_id") normalized_reason = validate_identifier(reason, "reason") if normalized_reason not in NOTIFICATION_SUPPRESSION_CLEAR_REASONS: raise ValueError("notification suppression clear reason is unsupported") normalized_pepper = _notification_pepper(notification_pepper) normalized_now = _utc(now) with session.begin(): admin = session.get(User, normalized_admin) if admin is None or admin.status != "active" or not admin.is_server_admin: raise PermissionError("an active server administrator is required") user = session.get(User, normalized_user) if user is None or user.verified_email is None: raise ValueError("user has no verified email address") recipient_digest = _recipient_address_digest( _email_address(user.verified_email, "verified_email"), normalized_pepper, ) suppression = session.scalar( select(NotificationSuppression).where( NotificationSuppression.user_id == user.id, NotificationSuppression.recipient_address_digest == recipient_digest, ) ) if suppression is None or not suppression.active: return False suppression.active = False suppression.cleared_at = normalized_now session.add( AuditEvent( id=new_identifier("audit"), actor_type="user", actor_id=admin.id, action="notification.suppression_cleared", target_type="notification_suppression", target_id=suppression.id, reason=normalized_reason, correlation_id=new_identifier("correlation"), source_address_digest=None, metadata_json={"userId": user.id}, ) ) return True def purge_notification_feedback_events( session: Session, *, now: datetime, retention_days: int = DEFAULT_NOTIFICATION_FEEDBACK_RETENTION_DAYS, maximum_events: int = 1000, ) -> int: if not isinstance(session, Session): raise TypeError("session must be Session") normalized_now = _utc(now) normalized_retention = _bounded_integer( retention_days, "retention_days", MAXIMUM_NOTIFICATION_FEEDBACK_RETENTION_DAYS, ) limit = _bounded_integer( maximum_events, "maximum_events", MAXIMUM_NOTIFICATION_EVENTS, ) cutoff = normalized_now - timedelta(days=normalized_retention) with session.begin(): feedback_events = session.scalars( select(NotificationFeedbackEvent) .where(NotificationFeedbackEvent.occurred_at < cutoff) .order_by( NotificationFeedbackEvent.occurred_at, NotificationFeedbackEvent.id, ) .limit(limit) .with_for_update(skip_locked=True) ).all() feedback_ids = tuple(value.id for value in feedback_events) if feedback_ids: suppressions = session.scalars( select(NotificationSuppression).where( NotificationSuppression.source_feedback_id.in_(feedback_ids) ) ).all() for suppression in suppressions: suppression.source_feedback_id = None for feedback in feedback_events: session.delete(feedback) session.add( AuditEvent( id=new_identifier("audit"), actor_type="system", actor_id=None, action="notification.feedback_purged", target_type="notification_feedback", target_id=None, reason=None, correlation_id=new_identifier("correlation"), source_address_digest=None, metadata_json={ "deletedCount": len(feedback_events), "retentionDays": normalized_retention, }, ) ) return len(feedback_events) def requeue_failed_customer_notifications( session: Session, *, now: datetime, maximum_events: int = 100, maximum_attempts: int = 10, ) -> int: if not isinstance(session, Session): raise TypeError("session must be Session") normalized_now = _utc(now) limit = _bounded_integer(maximum_events, "maximum_events", MAXIMUM_NOTIFICATION_EVENTS) attempts = _bounded_integer( maximum_attempts, "maximum_attempts", MAXIMUM_NOTIFICATION_RETRY_ATTEMPTS, ) with session.begin(): outboxes = session.scalars( select(OutboxEvent) .where( OutboxEvent.event_type == CUSTOMER_NOTIFICATION_OUTBOX_EVENT_TYPE, OutboxEvent.status == "failed", OutboxEvent.attempts < attempts, ) .order_by(OutboxEvent.updated_at, OutboxEvent.id) .limit(limit) .with_for_update(skip_locked=True) ).all() for outbox in outboxes: outbox.status = "pending" outbox.available_at = normalized_now outbox.processed_at = None outbox.last_error_code = None deliveries = session.scalars( select(NotificationDelivery).where( NotificationDelivery.outbox_event_id == outbox.id, NotificationDelivery.status == "failed", ) ).all() for delivery in deliveries: delivery.status = "pending" delivery.last_error_code = None return len(outboxes) def recover_stale_customer_notification_claims( session: Session, *, now: datetime, claim_timeout_minutes: int = DEFAULT_NOTIFICATION_CLAIM_TIMEOUT_MINUTES, maximum_events: int = 100, ) -> int: if not isinstance(session, Session): raise TypeError("session must be Session") normalized_now = _utc(now) timeout = _bounded_integer(claim_timeout_minutes, "claim_timeout_minutes", 1440) limit = _bounded_integer(maximum_events, "maximum_events", MAXIMUM_NOTIFICATION_EVENTS) cutoff = normalized_now - timedelta(minutes=timeout) with session.begin(): outboxes = session.scalars( select(OutboxEvent) .where( OutboxEvent.event_type == CUSTOMER_NOTIFICATION_OUTBOX_EVENT_TYPE, OutboxEvent.status == "processing", OutboxEvent.updated_at <= cutoff, ) .order_by(OutboxEvent.updated_at, OutboxEvent.id) .limit(limit) .with_for_update(skip_locked=True) ).all() for outbox in outboxes: outbox.status = "pending" outbox.available_at = normalized_now outbox.processed_at = None outbox.last_error_code = "notification_claim_recovered" deliveries = session.scalars( select(NotificationDelivery).where( NotificationDelivery.outbox_event_id == outbox.id, NotificationDelivery.status == "sending", ) ).all() for delivery in deliveries: delivery.status = "pending" delivery.last_error_code = "notification_claim_recovered" return len(outboxes) def _send_delivery( session: Session, catalog: LicensingCatalog, provider: CustomerNotificationProvider, account_portal_url: str, delivery_id: str, now: datetime, notification_pepper: bytes, ) -> None: normalized_delivery = validate_identifier(delivery_id, "delivery_id") with session.begin(): delivery = session.get(NotificationDelivery, normalized_delivery) if delivery is None or delivery.status in ("processed", "skipped"): return outbox = session.get(OutboxEvent, delivery.outbox_event_id) user = session.get(User, delivery.user_id) if outbox is None or user is None: delivery.status = "skipped" delivery.last_error_code = "notification_subject_missing" return if user.status != "active" or user.verified_email is None: delivery.status = "skipped" delivery.last_error_code = "notification_recipient_unavailable" return try: subscription = _subscription_from_outbox(session, outbox) authorized_user_ids = { candidate.id for candidate in _recipient_users( session, subscription.license_id, now, ) } if user.id not in authorized_user_ids: delivery.status = "skipped" delivery.last_error_code = "notification_recipient_unavailable" return recipient = _email_address(user.verified_email, "verified_email") recipient_digest = _recipient_address_digest( recipient, notification_pepper, ) suppression = session.scalar( select(NotificationSuppression).where( NotificationSuppression.user_id == user.id, NotificationSuppression.recipient_address_digest == recipient_digest, NotificationSuppression.active.is_(True), ) ) if suppression is not None: delivery.recipient_address_digest = recipient_digest delivery.status = "skipped" delivery.last_error_code = "notification_recipient_suppressed" return message = _message_for_outbox( session, catalog, outbox, recipient, account_portal_url, delivery.id, ) except (TypeError, ValueError): delivery.status = "skipped" delivery.last_error_code = "notification_payload_invalid" return delivery.status = "sending" delivery.attempts += 1 delivery.recipient_address_digest = recipient_digest delivery.last_error_code = None try: result = provider.send(message) if not isinstance(result, NotificationSendResult): raise NotificationProviderFailure("notification_provider_contract_invalid") except NotificationProviderFailure as exc: with session.begin(): delivery = session.get(NotificationDelivery, normalized_delivery) if delivery is not None: delivery.status = "failed" delivery.last_error_code = exc.error_code return with session.begin(): delivery = session.get(NotificationDelivery, normalized_delivery) if delivery is None: return delivery.status = "processed" delivery.provider_message_id = result.provider_message_id delivery.last_error_code = None delivery.delivered_at = now def _message_for_outbox( session: Session, catalog: LicensingCatalog, outbox: OutboxEvent, recipient: str, account_portal_url: str, delivery_id: str, ) -> CustomerNotificationMessage: if not isinstance(outbox, OutboxEvent): raise TypeError("outbox must be OutboxEvent") subscription = _subscription_from_outbox(session, outbox) raw_kind = outbox.payload_json.get("notificationKind") try: kind = CustomerNotificationKind(str(raw_kind)) except ValueError as exc: raise ValueError("notification kind is invalid") from exc sku_ids = tuple( session.scalars( select(SubscriptionItem.sku_id) .where(SubscriptionItem.subscription_id == subscription.id) .order_by(SubscriptionItem.sku_id) ) ) labels = tuple( catalog.skus[sku_id].label if sku_id in catalog.skus else sku_id for sku_id in sku_ids ) subject, introduction = _template(kind) subscription_label = ", ".join(labels) if labels else "your subscription" body = ( f"{introduction}\n\n" f"Affected subscription: {subscription_label}\n" f"Manage billing and review status: {account_portal_url}\n\n" "Your perpetual edition and independently purchased add-ons are not " "removed by a subscription billing event.\n\n" "This is a transactional service message about your Apolon license, " "not a marketing email." ) return CustomerNotificationMessage( recipient=recipient, subject=subject, body=body, idempotency_key=delivery_id, notification_kind=kind, ) def _template(kind: CustomerNotificationKind) -> tuple[str, str]: values = { CustomerNotificationKind.SUBSCRIPTION_STARTED: ( "Your Apolon subscription is active", "Your subscription features are ready to use.", ), CustomerNotificationKind.PAYMENT_FAILED: ( "Action needed for your Apolon subscription", "The latest subscription payment needs attention. Access remains available " "during the configured payment grace period.", ), CustomerNotificationKind.ACCESS_SUSPENDED: ( "Apolon subscription access is paused", "Subscription features are paused because the provider reports that payment " "is incomplete or unpaid.", ), CustomerNotificationKind.PAYMENT_RECOVERED: ( "Apolon subscription access is restored", "The payment issue is resolved and subscription features are available again.", ), CustomerNotificationKind.CANCELLATION_SCHEDULED: ( "Apolon subscription cancellation scheduled", "Your subscription is scheduled to end at the close of its current period.", ), CustomerNotificationKind.CANCELLATION_REVERSED: ( "Apolon subscription cancellation reversed", "The scheduled cancellation was removed and the subscription will continue.", ), CustomerNotificationKind.SUBSCRIPTION_CANCELED: ( "Apolon subscription ended", "The provider reports that your subscription has ended.", ), CustomerNotificationKind.DISPUTE_OPEN: ( "Apolon subscription access paused during dispute review", "Subscription features are temporarily paused while the payment dispute is " "reviewed.", ), CustomerNotificationKind.DISPUTE_LOST: ( "Apolon subscription payment dispute was lost", "Subscription features remain paused because the provider reports that the " "payment dispute was lost.", ), CustomerNotificationKind.DISPUTE_RESOLVED: ( "Apolon subscription payment dispute resolved", "The provider reports that the dispute hold is resolved. Subscription access " "now follows the current provider subscription state.", ), CustomerNotificationKind.REFUNDED: ( "Apolon subscription payment refunded", "The subscription payment was fully refunded and subscription features are " "paused.", ), } return values[kind] def _subscription_from_outbox(session: Session, outbox: OutboxEvent) -> Subscription: payload = outbox.payload_json if ( not isinstance(payload, dict) or payload.get("schema") != CUSTOMER_NOTIFICATION_SCHEMA or payload.get("schemaVersion") != CUSTOMER_NOTIFICATION_SCHEMA_VERSION ): raise ValueError("notification outbox payload is invalid") subscription_id = validate_identifier( payload.get("subscriptionId"), "subscription_id", ) subscription = session.get(Subscription, subscription_id) if subscription is None: raise ValueError("notification subscription is missing") return subscription def _recipient_users( session: Session, license_id: str, now: datetime, ) -> tuple[User, ...]: license_row = session.get(License, validate_identifier(license_id, "license_id")) if license_row is None: return () if license_row.owner_user_id is not None: user = session.get(User, license_row.owner_user_id) return ( (user,) if user is not None and user.status == "active" and user.verified_email is not None else () ) if license_row.owner_organization_id is None: return () return tuple( session.scalars( select(User) .join(Membership, Membership.user_id == User.id) .where( Membership.organization_id == license_row.owner_organization_id, Membership.status == "active", Membership.role.in_(("owner", "admin")), or_(Membership.valid_until.is_(None), Membership.valid_until > now), User.status == "active", User.verified_email.is_not(None), ) .order_by(User.id) ) ) def _audit_notification( session: Session, outbox: OutboxEvent, action: str, metadata: dict[str, object], ) -> None: try: kind = CustomerNotificationKind( str(outbox.payload_json.get("notificationKind")) ).value except (AttributeError, ValueError): kind = "invalid" try: correlation_id = validate_identifier( outbox.payload_json.get("providerEventId"), "provider_event_id", ) except (AttributeError, TypeError, ValueError): correlation_id = validate_identifier(outbox.id, "outbox_id") session.add( AuditEvent( id=new_identifier("audit"), actor_type="system", actor_id=None, action=validate_identifier(action, "audit action"), target_type="notification", target_id=outbox.id, reason=None, correlation_id=correlation_id, source_address_digest=None, metadata_json={"notificationKind": kind, **metadata}, ) ) def _validate_feedback_replay( existing: NotificationFeedbackEvent, feedback_type: NotificationFeedbackType, occurred_at: datetime, delivery_id: Optional[str], provider_message_id: Optional[str], ) -> None: if not isinstance(existing, NotificationFeedbackEvent): raise TypeError("existing must be NotificationFeedbackEvent") try: stored_type = NotificationFeedbackType(existing.feedback_type) except ValueError as exc: raise ValueError("stored notification feedback type is invalid") from exc stored_message = ( None if existing.provider_message_id is None else _header_text( existing.provider_message_id, "stored provider_message_id", MAXIMUM_PROVIDER_MESSAGE_ID_CHARACTERS, ) ) expected_digest = _feedback_payload_digest( existing.provider, existing.provider_event_id, stored_type, _utc(existing.occurred_at), validate_identifier( existing.delivery_reference_id, "delivery_reference_id", ), stored_message, ) if not hmac.compare_digest(existing.payload_digest, expected_digest): raise ValueError("stored provider feedback digest is invalid") if ( stored_type is not feedback_type or _utc(existing.occurred_at) != occurred_at or (delivery_id is not None and delivery_id != existing.delivery_reference_id) or ( provider_message_id is not None and provider_message_id != stored_message ) ): raise ValueError("provider feedback event was replayed with new content") def _feedback_delivery( session: Session, delivery_id: Optional[str], provider_message_id: Optional[str], ) -> NotificationDelivery: delivery = None if delivery_id is None else session.get( NotificationDelivery, delivery_id, ) if delivery_id is not None and delivery is None: raise ValueError("notification delivery was not found") message_delivery = None if provider_message_id is not None: matches = session.scalars( select(NotificationDelivery) .where(NotificationDelivery.provider_message_id == provider_message_id) .order_by(NotificationDelivery.id) .limit(2) ).all() if len(matches) > 1: raise ValueError("provider_message_id matches multiple deliveries") if matches: message_delivery = matches[0] elif delivery is None: raise ValueError("notification delivery was not found") if delivery is not None and message_delivery is not None: if delivery.id != message_delivery.id: raise ValueError("notification delivery locators disagree") resolved = delivery or message_delivery if resolved is None: raise ValueError("notification delivery was not found") if ( provider_message_id is not None and resolved.provider_message_id is not None and resolved.provider_message_id != provider_message_id ): raise ValueError("provider_message_id does not match the delivery") return resolved def _feedback_payload_digest( provider: str, provider_event_id: str, feedback_type: NotificationFeedbackType, occurred_at: datetime, delivery_id: str, provider_message_id: Optional[str], ) -> bytes: document = { "deliveryId": delivery_id, "feedbackType": feedback_type.value, "occurredAt": occurred_at.isoformat().replace("+00:00", "Z"), "provider": provider, "providerEventId": provider_event_id, "providerMessageId": provider_message_id, } return hashlib.sha256( json.dumps( document, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("utf-8") ).digest() def _feedback_priority(value: Optional[str]) -> int: priorities = { None: 0, NotificationFeedbackType.SOFT_BOUNCE.value: 1, NotificationFeedbackType.HARD_BOUNCE.value: 2, NotificationFeedbackType.COMPLAINT.value: 3, } if value not in priorities: raise ValueError("stored notification feedback status is invalid") return priorities[value] def _notification_pepper(value: object) -> bytes: if not isinstance(value, bytes): raise TypeError("notification_pepper must be bytes") if len(value) < NOTIFICATION_PEPPER_MINIMUM_BYTES: raise ValueError("notification_pepper is too short") return value def _recipient_address_digest(recipient: str, notification_pepper: bytes) -> bytes: normalized_recipient = _email_address(recipient, "recipient") normalized_pepper = _notification_pepper(notification_pepper) return hmac.new( normalized_pepper, NOTIFICATION_RECIPIENT_DIGEST_DOMAIN + normalized_recipient.encode("utf-8"), hashlib.sha256, ).digest() def _bounded_integer(value: object, field_name: str, maximum: int) -> int: if isinstance(value, bool) or not isinstance(value, int): raise TypeError(f"{field_name} must be an integer") if value < 1 or value > maximum: raise ValueError(f"{field_name} must be between one and {maximum}") return value def _utc(value: datetime) -> datetime: if not isinstance(value, datetime): raise TypeError("value must be a datetime") if value.tzinfo is None or value.utcoffset() is None: return value.replace(tzinfo=timezone.utc) return value.astimezone(timezone.utc) def _bounded_text(value: object, field_name: str, maximum: int) -> str: if not isinstance(value, str) or not value.strip(): raise ValueError(f"{field_name} must be a non-empty string") normalized = value.strip() if len(normalized) > maximum: raise ValueError(f"{field_name} is too long") return normalized def _header_text(value: object, field_name: str, maximum: int) -> str: normalized = _bounded_text(value, field_name, maximum) if "\r" in normalized or "\n" in normalized: raise ValueError(f"{field_name} contains a line break") return normalized def _email_address(value: object, field_name: str) -> str: normalized = _header_text(value, field_name, 256) try: address = Address(addr_spec=normalized) except (TypeError, ValueError) as exc: raise ValueError(f"{field_name} is invalid") from exc if not address.username or not address.domain or address.addr_spec != normalized: raise ValueError(f"{field_name} must be a canonical email address") return address.addr_spec def _smtp_host(value: object) -> str: normalized = _bounded_text(value, "host", 253) if ( any(character.isspace() for character in normalized) or any(character in normalized for character in ("/", "@", "?", "#")) ): raise ValueError("host is invalid") return normalized def _https_url(value: object, field_name: str) -> str: normalized = _header_text(value, field_name, 4096) parsed = urlparse(normalized) if ( parsed.scheme != "https" or not parsed.netloc or parsed.username is not None or parsed.password is not None or parsed.fragment ): raise ValueError(f"{field_name} must be an HTTPS URL without credentials") return normalized __all__ = [ "CUSTOMER_NOTIFICATION_OUTBOX_EVENT_TYPE", "DEFAULT_NOTIFICATION_FEEDBACK_RETENTION_DAYS", "CustomerNotificationKind", "CustomerNotificationMessage", "NotificationFeedbackType", "NotificationProviderFailure", "NotificationSendResult", "SmtpCustomerNotificationProvider", "billing_hold_notification_kind", "clear_notification_suppression", "enqueue_subscription_notification", "process_customer_notification_outbox", "purge_notification_feedback_events", "record_notification_feedback", "recover_stale_customer_notification_claims", "requeue_failed_customer_notifications", "subscription_transition_notification_kinds", ]