import logging from collections.abc import Callable from time import time from typing import cast from anydi import singleton from cachetools import TTLCache from fansifter_common.adapters.twilio import TwilioClient from fansifter_common.adapters.twilio.models import SendMessageRequest from fansifter_common.encrypter import Encrypter from fansifter_common.identifiers.utils import encrypt_profile_token from ows_text_campaigns.adapters.dynamodb import DynamoDBClient from ows_text_campaigns.adapters.exceptions import PreferenceCenterError from ows_text_campaigns.adapters.kafka import KafkaClient from ows_text_campaigns.adapters.ows_preference_center import ( OwsPreferenceCenterClient, ) from ows_text_campaigns.adapters.ows_url_shortener import OwsUrlShortenerClient from ows_text_campaigns.api.schemas import TwilioInboundMessageBody from ows_text_campaigns.artist.handlers import ( GetArtistSettingsByPhoneNumberHandler, GetArtistSettingsByPhoneNumberRequest, ) from ows_text_campaigns.artist.handlers.get_artist_settings_by_phone_number import ( GetArtistSettingsByPhoneNumberResponse as ArtistSettings, ) from ows_text_campaigns.container import ( PreferenceCenterProfileUrlTemplate, UrlShortenerDomain, ) from ows_text_campaigns.twilio.services import TwilioAccountService logger = logging.getLogger(__name__) @singleton class TwilioInboundMessageHandler: def __init__( self, service: TwilioAccountService, twilio_client: TwilioClient, kafka_client: KafkaClient, dynamo_client: DynamoDBClient, preference_center_client: OwsPreferenceCenterClient, preference_center_encrypter: Encrypter, preference_center_profile_url_template: PreferenceCenterProfileUrlTemplate, ows_url_shortener_client: OwsUrlShortenerClient, url_shortener_domain: UrlShortenerDomain, artist_settings_handler: GetArtistSettingsByPhoneNumberHandler, artist_settings_ttl: int = 3600, path_length: int = 12, ) -> None: self.twilio_client = twilio_client self.twilio_auth: dict[str, str] = {} self.service = service self.kafka_client = kafka_client self.dynamo_client = dynamo_client self.preference_center_client = preference_center_client self.preference_center_encrypter = preference_center_encrypter self.preference_center_profile_url_template = ( preference_center_profile_url_template ) self.url_shortener_client = ows_url_shortener_client self.url_shortener_domain = url_shortener_domain self.artist_settings_handler = artist_settings_handler self.artist_settings: dict[str, dict[str, int | ArtistSettings]] = {} self.artist_settings_cache = cast( TTLCache[str, ArtistSettings], TTLCache(maxsize=1024, ttl=artist_settings_ttl), ) self.path_length = path_length def handle(self, body: TwilioInboundMessageBody) -> None: timestamp = round(time()) if body.account_sid not in self.twilio_auth: auth_token = self.service.get_auth_token(body.account_sid) self.twilio_auth[body.account_sid] = auth_token fan_state = self.dynamo_client.get_fan_state(body.fan_key) artist_settings = self.get_artist_settings(body.artist_phone_number) state_handler = self.state_handlers_mapping[fan_state] new_fan_state = state_handler(body, artist_settings) if new_fan_state and (new_fan_state != fan_state): self.dynamo_client.set_fan_state(body.fan_key, new_fan_state) self.kafka_client.send_fan_state( { "fanPhoneNumber": body.fan_phone_number, "artistPhoneNumber": body.artist_phone_number, "source": body.source, "channel": body.channel, "fanCountry": body.fan_country_code, "fanState": body.fan_state, "fanAreaCode": body.fan_area_code, "subscriptionStatus": new_fan_state, "createdAt": timestamp, }, ) self.kafka_client.send_event_message( data={"body": body.model_dump(), "timestamp": timestamp} ) def get_artist_settings(self, artist_phone_number: str) -> ArtistSettings: if artist_phone_number not in self.artist_settings_cache: artist_settings = self.artist_settings_handler.handler( GetArtistSettingsByPhoneNumberRequest( phone_number=artist_phone_number, ) ) self.artist_settings_cache[artist_phone_number] = artist_settings return self.artist_settings_cache[artist_phone_number] @property def state_handlers_mapping( self, ) -> dict[ str, Callable[ [TwilioInboundMessageBody, ArtistSettings], str | None, ], ]: return { "greetings": self.handle_greetings_state, "opt_in_asked": self.handle_opt_in_asked_state, "opted_in": self.handle_opted_in_state, "opted_out": self.handle_opted_out_state, } def handle_greetings_state( self, body: TwilioInboundMessageBody, artist_settings: ArtistSettings, ) -> str | None: if body.message.lower() == "start": response_message = artist_settings.opt_in_message_terms else: response_message = f"{artist_settings.opt_in_message}\n{artist_settings.opt_in_message_terms}" self.send_response(body, response_message) return "opt_in_asked" def handle_opt_in_asked_state( self, body: TwilioInboundMessageBody, artist_settings: ArtistSettings, ) -> str | None: if body.message.lower() == "yes": response_message = f"{artist_settings.welcome_message}\n{artist_settings.welcome_message_instructions}" response_media_url = artist_settings.vcard_url self.send_response(body, response_message, response_media_url) return "opted_in" def handle_opted_in_state( self, body: TwilioInboundMessageBody, artist_settings: ArtistSettings, ) -> str | None: if body.message.lower() == "info": try: fan_profile = self.preference_center_client.get_fan_profile( body.fan_phone_number ) encrypted_profile_token = encrypt_profile_token( self.preference_center_encrypter, identifier=fan_profile.identifier ) fan_profile_url = self.preference_center_profile_url_template.format( fan_token=encrypted_profile_token ) short_url = self.url_shortener_client.shorten_url( fan_profile_url, self.url_shortener_domain, path_length=self.path_length, ) info_message = artist_settings.info_message.format( preference_center_url=short_url.short_url, artist_name=artist_settings.artist_name, ) self.send_response(body, info_message) except PreferenceCenterError: logger.warning("Failed to fetch fan profile", exc_info=True) # Temporary text fallback_message = "Your profile is still being created. Please check back in a few minutes." self.send_response(body, fallback_message) elif body.message.lower() == "stop": return "opted_out" def handle_opted_out_state( self, body: TwilioInboundMessageBody, artist_settings: ArtistSettings, ) -> str | None: if body.message.lower() == "start": response_message = artist_settings.reopt_in_message.format( artist_name=artist_settings.artist_name ) self.send_response(body, response_message) return "opted_in" def send_response( self, body: TwilioInboundMessageBody, response_message: str, response_media_url: str | None = None, ) -> None: self.twilio_client.send_message( request=SendMessageRequest( account_sid=body.account_sid, to=body.fan_phone_number, from_=body.artist_phone_number, body=response_message, media_url=response_media_url, ), auth=(body.account_sid, self.twilio_auth[body.account_sid]), )