han-app/VM1_app/codebase/backend/sms-service/app/domain.py

from __future__ import annotations

import hashlib
import hmac
import json
import re
import string
from dataclasses import dataclass
from datetime import datetime
from typing import Any

import phonenumbers

from app.db import DeliveryStatus, SendStatus

GSM_BASIC = (
    "@£$¥èéùìòÇ\nØø\rÅåΔ_ΦΓΛΩΠΨΣΘΞ "
    "!\"#¤%&'()*+,-./0123456789:;<=>?"
    "¡ABCDEFGHIJKLMNOPQRSTUVWXYZÄÖÑܧ¿abcdefghijklmnopqrstuvwxyzäöñüà"
)
GSM_EXTENDED = "^{}\\[~]|€"
PHONE_RE = re.compile(r"^\+[1-9]\d{7,14}$")


class DomainError(Exception):
    def __init__(
        self, code: str, status: int, message: str, details: dict[str, Any] | None = None
    ) -> None:
        self.code = code
        self.status = status
        self.message = message
        self.details = details or {}
        super().__init__(message)


def normalize_phone(value: str) -> tuple[str, str, str]:
    if not PHONE_RE.fullmatch(value):
        raise DomainError("sms_request_invalid", 422, "phone_e164 must be valid E.164")
    try:
        parsed = phonenumbers.parse(value, None)
    except phonenumbers.NumberParseException:
        raise DomainError("sms_request_invalid", 422, "phone_e164 must be valid E.164") from None
    if not phonenumbers.is_valid_number(parsed):
        raise DomainError("sms_request_invalid", 422, "phone_e164 must be valid E.164")
    normalized = phonenumbers.format_number(parsed, phonenumbers.PhoneNumberFormat.E164)
    if normalized != value:
        raise DomainError("sms_request_invalid", 422, "phone_e164 must be canonical E.164")
    digits = normalized[1:]
    masked = f"+{digits[:1]}{'*' * max(0, len(digits) - 5)}{digits[-4:]}"
    return normalized, digits, masked


def request_fingerprint(payload: dict[str, Any]) -> str:
    canonical = json.dumps(
        payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False
    )
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


def destination_hmac(phone_e164: str, key: bytes) -> str:
    return hmac.new(key, phone_e164.encode(), hashlib.sha256).hexdigest()


def sms_parts(body: str) -> int:
    if not body or "\ufeff" in body or "\x00" in body:
        raise DomainError("sms_request_invalid", 422, "Rendered message contains invalid text")
    gsm_units = 0
    for char in body:
        if char in GSM_BASIC:
            gsm_units += 1
        elif char in GSM_EXTENDED:
            gsm_units += 2
        else:
            total = len(body.encode("utf-16-be")) // 2
            return 1 if total <= 70 else (total + 66) // 67
    return 1 if gsm_units <= 160 else (gsm_units + 152) // 153


def render_template(
    body_template: str,
    placeholders: list[str],
    substitutions: dict[str, Any],
    max_parts: int,
) -> str:
    expected = set(placeholders)
    supplied = set(substitutions)
    if expected != supplied:
        raise DomainError(
            "sms_request_invalid",
            422,
            "Substitutions do not match template placeholders",
            {"missing": sorted(expected - supplied), "unknown": sorted(supplied - expected)},
        )
    parsed = {
        field_name
        for _, field_name, format_spec, conversion in string.Formatter().parse(body_template)
        if field_name is not None
        and not format_spec
        and not conversion
        and field_name.isidentifier()
    }
    if parsed != expected or any(
        format_spec or conversion
        for _, field_name, format_spec, conversion in string.Formatter().parse(body_template)
        if field_name is not None
    ):
        raise DomainError("sms_request_invalid", 422, "Template placeholder contract is invalid")
    body = body_template.format_map({key: str(value) for key, value in substitutions.items()})
    if len(body.encode("utf-8")) > 2048 or sms_parts(body) > max_parts:
        raise DomainError("sms_request_invalid", 422, "Rendered message exceeds template limit")
    return body


@dataclass(frozen=True)
class ProviderResult:
    send_status: SendStatus
    http_status: int | None = None
    message_uuid: str | None = None
    external_id: str | None = None
    error_code: str | None = None
    error_message: str | None = None
    retry_safe: bool = False
    contract_violation: bool = False


DELIVERY_RANK = {
    DeliveryStatus.UNKNOWN: 0,
    DeliveryStatus.SENT: 1,
    DeliveryStatus.DELIVERED: 2,
    DeliveryStatus.UNDELIVERED: 2,
    DeliveryStatus.UNSENT: 2,
}


def delivery_transition(current: DeliveryStatus, incoming: str) -> DeliveryStatus | None:
    try:
        target = DeliveryStatus(incoming.lower())
    except ValueError:
        return None
    if DELIVERY_RANK[target] < DELIVERY_RANK[current]:
        return current
    if DELIVERY_RANK[target] == DELIVERY_RANK[current] and target != current:
        return current
    return target


def parse_status_time(value: str) -> datetime:
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        raise DomainError("callback_invalid", 422, "Invalid callback status_time") from None
    if parsed.tzinfo is None:
        raise DomainError("callback_invalid", 422, "Callback status_time requires timezone")
    return parsed