"""Messaging orchestration: policy, idempotency, rendering and delivery.""" from __future__ import annotations import re from datetime import datetime, timedelta from email.utils import parseaddr from bridge_platform.logging.audit_logger import AuditLogger from bridge_platform.logging.platform_logger import get_platform_logger, log_structured from bridge_platform.quotas.service import QuotaExceeded, consume from bridge_platform.secrets.secrets_manager import SecretsManager from .drivers import default_registry from .repository import MessagingRepository from .templates import TemplateError, render_template LOGGER = get_platform_logger("messaging_service") E164 = re.compile(r"^\+[1-9][0-9]{7,14}$") class MessagingError(Exception): def __init__(self, code: str, message: str, *, status: int = 400, retryable: bool = False) -> None: self.code, self.status, self.retryable = code, status, retryable super().__init__(message) def normalize_phone(value: str) -> str: compact = re.sub(r"[\s().-]+", "", str(value or "")) if compact.startswith("0") and len(compact) == 10: compact = "+61" + compact[1:] elif compact.startswith("61"): compact = "+" + compact if not E164.fullmatch(compact): raise MessagingError("invalid_recipient", "Phone number must be a valid E.164 number.") return compact def normalize_email(value: str) -> str: _, address = parseaddr(str(value or "").strip()) if address != str(value or "").strip() or address.count("@") != 1 or len(address) > 320: raise MessagingError("invalid_recipient", "Email address is invalid.") local, domain = address.rsplit("@", 1) if not local or "." not in domain or any(c.isspace() for c in address): raise MessagingError("invalid_recipient", "Email address is invalid.") return f"{local}@{domain.lower()}" def mask_recipient(value: str, channel: str) -> str: if channel == "sms": return value[:5] + "*" * max(6, len(value) - 7) + value[-2:] local, domain = value.split("@", 1) return local[:1] + "***@" + domain class MessagingService: def __init__(self, repository=None, registry=None, secrets=None) -> None: self.repository = repository or MessagingRepository() self.registry = registry or default_registry() self.secrets = secrets or SecretsManager() def send_template(self, context: dict, payload: dict) -> dict: tenant_id = _tenant_id(context) requesting_app = _requesting_app(context) channel = str(payload.get("channel") or "") key = str(payload.get("idempotency_key") or "").strip() template_key = str(payload.get("template_key") or "").strip() purpose = str(payload.get("purpose") or "").strip() if channel not in {"sms", "email"} or not key or not template_key or not purpose: raise MessagingError("invalid_input", "channel, template_key, purpose and idempotency_key are required.") if not 8 <= len(key) <= 128: raise MessagingError("invalid_input", "idempotency_key must contain 8 to 128 characters.") permission = self.repository.permission(tenant_id, requesting_app) if not permission or channel not in (permission.channels_json or []) or template_key not in (permission.templates_json or []) or purpose not in (permission.purposes_json or []): raise MessagingError("permission_denied", "The requesting application is not allowed to send this template.", status=403) recipient_input = payload.get("recipient") or {} recipient = normalize_phone(recipient_input.get("phone")) if channel == "sms" else normalize_email(recipient_input.get("email")) template = self.repository.template(tenant_id, template_key, channel) if not template: raise MessagingError("template_not_found", "Message template is unavailable.", status=404) try: rendered = render_template(template, payload.get("template_data") or {}) except TemplateError as exc: raise MessagingError("template_invalid", str(exc)) from exc existing = self.repository.existing(tenant_id, requesting_app, "send_template_v1", key) if existing: return _delivery_data(existing, replay=True) self._check_recipient_limits(tenant_id, recipient, purpose) try: consume(tenant_id=tenant_id, app_id="messaging_service", metric=f"messages_{channel}", metadata={"template_key": template_key, "purpose": purpose}) except QuotaExceeded as exc: raise MessagingError("rate_limit_exceeded", "Tenant messaging limit exceeded.", status=429) from exc providers = self.repository.providers(tenant_id, channel) if not providers: raise MessagingError("provider_unavailable", "No healthy provider is configured for this channel.", status=503, retryable=True) metadata = payload.get("metadata") if isinstance(payload.get("metadata"), dict) else {} delivery, created = self.repository.create_delivery( tenant_id=tenant_id, requesting_app=requesting_app, requested_by=str(context.get("user_id") or "") or None, operation="send_template_v1", idempotency_key=key, channel=channel, recipient_normalized=recipient, recipient_masked=mask_recipient(recipient, channel), template_key=template.key, template_version=template.version, purpose=purpose, correlation_id=str(metadata.get("correlation_id") or context.get("request_id") or ""), metadata_json={k: metadata[k] for k in ("resource_type", "resource_id") if k in metadata}, ) if not created: return _delivery_data(delivery, replay=True) result_delivery = self._deliver(delivery, providers, rendered, permission.allow_fallback) log_structured(LOGGER, "message_delivery_result", tenant_id=tenant_id, app_id=requesting_app, delivery_id=delivery.delivery_id, channel=channel, template_key=template_key, purpose=purpose, recipient=delivery.recipient_masked, status=result_delivery.status, correlation_id=delivery.correlation_id) AuditLogger().log(tenant_id=tenant_id, actor=requesting_app, action="messaging.send_template", status=result_delivery.status, delivery_id=delivery.delivery_id, recipient=delivery.recipient_masked, template_key=template_key) return _delivery_data(result_delivery) def _deliver(self, delivery, providers, rendered, allow_fallback): attempt = 0 selected = providers if allow_fallback else providers[:1] last_delivery = delivery for provider in selected: now = datetime.utcnow() if provider.per_minute_limit is not None and self.repository.count_since( tenant_id=delivery.tenant_id, provider_id=provider.provider_id, since=now - timedelta(minutes=1), ) >= provider.per_minute_limit: continue if provider.daily_limit is not None and self.repository.count_since( tenant_id=delivery.tenant_id, provider_id=provider.provider_id, since=now - timedelta(days=1), ) >= provider.daily_limit: continue secret = self.secrets.get_secret(tenant_id=delivery.tenant_id, app_id="messaging_service", secret_name=provider.secret_ref) if provider.secret_ref else None driver = self.registry.get(provider.driver) tries = max(1, int(provider.max_retries) + 1) for _ in range(tries): attempt += 1 result = driver.send(provider.config_json or {}, secret, { "recipient": delivery.recipient_normalized, "subject": rendered["subject"], "text": rendered["text"], "html": rendered["html"], }) last_delivery = self.repository.complete_attempt(delivery.delivery_id, provider.provider_id, attempt, result) if result.accepted or not result.retryable: break if last_delivery.status == "accepted": return last_delivery return last_delivery def _check_recipient_limits(self, tenant_id: str, recipient: str, purpose: str) -> None: now = datetime.utcnow() if self.repository.count_since(tenant_id=tenant_id, recipient=recipient, since=now - timedelta(minutes=1)) >= 3: raise MessagingError("rate_limit_exceeded", "Recipient rate limit exceeded.", status=429) if self.repository.count_since(tenant_id=tenant_id, purpose=purpose, since=now - timedelta(minutes=1)) >= 20: raise MessagingError("rate_limit_exceeded", "Purpose rate limit exceeded.", status=429) def health_check(self, context: dict, provider_id: str) -> dict: provider = self.repository.provider(_tenant_id(context), provider_id) if not provider: raise MessagingError("provider_not_found", "Provider was not found.", status=404) secret = self.secrets.get_secret(tenant_id=provider.tenant_id, app_id="messaging_service", secret_name=provider.secret_ref) if provider.secret_ref else None result = self.registry.get(provider.driver).health_check(provider.config_json or {}, secret) self.repository.update_health(provider.provider_id, result) return {"provider_id": provider.provider_id, "provider": provider.name, "status": result.status, "connected": result.accepted, "latency_ms": result.latency_ms, "checked_at": datetime.utcnow().isoformat() + "Z"} def _tenant_id(context: dict) -> str: tenant_id = str(((context.get("tenant") or {}).get("tenant_id")) or "") if not tenant_id: raise MessagingError("tenant_context_missing", "Trusted tenant context is required.") return tenant_id def _requesting_app(context: dict) -> str: trusted = str(context.get("requesting_app") or "").strip() if not trusted: raise MessagingError("caller_context_missing", "Trusted caller application context is required.", status=403) return trusted def _delivery_data(delivery, replay=False) -> dict: return {"delivery_id": delivery.delivery_id, "channel": delivery.channel, "provider_id": delivery.provider_id, "status": delivery.status, "created_at": delivery.created_at.isoformat() + "Z", "correlation_id": delivery.correlation_id, "idempotent_replay": replay}