from copy import deepcopy from typing import Any, List, Optional from fastapi import APIRouter, Depends, HTTPException, Query from loguru import logger from pydantic import BaseModel from sqlalchemy.exc import IntegrityError from api.constants import ( DEFAULT_CAMPAIGN_RETRY_CONFIG, DEFAULT_ORG_CONCURRENCY_LIMIT, DEPLOYMENT_MODE, ) from api.db import db_client from api.db.models import UserModel from api.db.telephony_configuration_client import TelephonyConfigurationInUseError from api.enums import OrganizationConfigurationKey, PostHogEvent from api.schemas.ai_model_configuration import ( DOGRAH_DEFAULT_LANGUAGE, DOGRAH_DEFAULT_VOICE, DOGRAH_SPEED_MAX, DOGRAH_SPEED_MIN, DOGRAH_SPEED_OPTIONS, DOGRAH_SPEED_STEP, OrganizationAIModelConfigurationResponse, OrganizationAIModelConfigurationV2, ) from api.schemas.organization_preferences import OrganizationPreferences from api.schemas.telephony_config import ( TelephonyConfigRequest, TelephonyConfigurationCreateRequest, TelephonyConfigurationDetail, TelephonyConfigurationListItem, TelephonyConfigurationListResponse, TelephonyConfigurationResponse, TelephonyConfigurationUpdateRequest, ) from api.schemas.telephony_phone_number import ( PhoneNumberCreateRequest, PhoneNumberListResponse, PhoneNumberResponse, PhoneNumberUpdateRequest, ProviderSyncStatus, ) from api.services.auth.depends import ( get_user, get_user_with_selected_organization, ) from api.services.configuration.ai_model_configuration import ( check_for_masked_keys_in_ai_model_configuration_v2, compile_ai_model_configuration_v2, convert_legacy_ai_model_configuration_to_v2, get_organization_ai_model_configuration_v2, get_resolved_ai_model_configuration, mask_ai_model_configuration_v2, merge_ai_model_configuration_v2_secrets, migrate_workflow_model_configurations_to_v2, upsert_organization_ai_model_configuration_v2, ) from api.services.configuration.check_validity import UserConfigurationValidator from api.services.configuration.defaults import DEFAULT_SERVICE_PROVIDERS from api.services.configuration.masking import is_mask_of, mask_key, mask_user_config from api.services.configuration.registry import ( DOGRAH_MULTILINGUAL_AUTODETECT_LANGUAGES, DOGRAH_STT_LANGUAGES, REGISTRY, DograhTTSService, ServiceProviders, ServiceType, ) from api.services.mps_billing import ensure_hosted_mps_billing_account_v2 from api.services.mps_service_key_client import mps_service_key_client from api.services.organization_context import ( OrganizationContextResponse, get_organization_context, ) from api.services.organization_preferences import ( external_pbx_integrations_enabled, get_organization_preferences, upsert_organization_preferences, ) from api.services.posthog_client import capture_event from api.services.telephony import registry as telephony_registry from api.services.telephony.factory import get_telephony_provider_by_id from api.services.worker_sync.manager import get_worker_sync_manager from api.services.worker_sync.protocol import WorkerSyncEventType from api.utils.common import get_backend_endpoints router = APIRouter(prefix="/organizations", tags=["organizations"]) def _sensitive_fields(provider_name: str) -> List[str]: """Field names that should be masked when displaying stored config. Sourced from ProviderUIField.sensitive in the registry — the same source of truth that drives the form-rendering UI. """ spec = telephony_registry.get_optional(provider_name) if spec is None or spec.ui_metadata is None: return [] return [f.name for f in spec.ui_metadata.fields if f.sensitive] def _mask_sensitive(provider_name: str, value: dict) -> dict: """Return a copy of ``value`` with sensitive fields masked for display.""" out = deepcopy(value) for field_name in _sensitive_fields(provider_name): v = _get_nested_field(out, field_name) if v: _set_nested_field(out, field_name, mask_key(str(v))) return out class TelephonyProviderUIOption(BaseModel): value: str label: str class TelephonyProviderUICondition(BaseModel): field: str equals: Any class TelephonyProviderUIField(BaseModel): """One form field on a telephony provider's configuration UI.""" name: str label: str type: str required: bool sensitive: bool description: Optional[str] = None placeholder: Optional[str] = None options: Optional[List[TelephonyProviderUIOption]] = None visible_when: Optional[TelephonyProviderUICondition] = None section: Optional[str] = None class TelephonyProviderMetadata(BaseModel): """UI form metadata for a single telephony provider.""" provider: str display_name: str fields: List[TelephonyProviderUIField] docs_url: Optional[str] = None class TelephonyProvidersMetadataResponse(BaseModel): """List of UI form definitions used by the telephony-config screen.""" providers: List[TelephonyProviderMetadata] class TelephonyConfigWarningsResponse(BaseModel): """Aggregated telephony-configuration warning counts for the user's org. Drives the page banner and nav badge that nudge customers to finish optional-but-recommended configuration steps. Shape is a flat dict so new warning types can be added without breaking the client. """ telnyx_missing_webhook_public_key_count: int vonage_missing_signature_secret_count: int class ModelConfigurationMetricPrice(BaseModel): metric_code: str display_name: str unit: str price_per_minute: float currency: str rounding_policy: str class ModelConfigurationPricingResponse(BaseModel): """MPS-owned effective prices relevant to model configuration choices.""" platform_usage: ModelConfigurationMetricPrice | None = None dograh_model: ModelConfigurationMetricPrice | None = None @router.get("/context", response_model=OrganizationContextResponse) async def get_current_organization_context(user: UserModel = Depends(get_user)): """Return organization-scoped configuration signals owned by Dograh.""" return await get_organization_context(user) @router.get( "/telephony-providers/metadata", response_model=TelephonyProvidersMetadataResponse, ) async def get_telephony_providers_metadata(user: UserModel = Depends(get_user)): """Return the list of available telephony providers and their form schemas. The UI uses this to render the configuration form generically instead of hard-coding fields per provider. Adding a new provider only requires declaring its ui_metadata in providers//__init__.py. """ if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") external_pbx_enabled = await external_pbx_integrations_enabled( user.selected_organization_id ) providers = [] for spec in telephony_registry.all_specs(): if spec.ui_metadata is None: continue providers.append( TelephonyProviderMetadata( provider=spec.name, display_name=spec.ui_metadata.display_name, fields=[ TelephonyProviderUIField( name=f.name, label=f.label, type=f.type, required=f.required, sensitive=f.sensitive, description=f.description, placeholder=f.placeholder, options=( [ {"value": option.value, "label": option.label} for option in f.options ] if f.options else None ), visible_when=( { "field": f.visible_when.field, "equals": f.visible_when.equals, } if f.visible_when else None ), section=f.section, ) for f in spec.ui_metadata.fields if not f.feature_gate or ( f.feature_gate == "external_pbx_integrations" and external_pbx_enabled ) ], docs_url=spec.ui_metadata.docs_url, ) ) return TelephonyProvidersMetadataResponse(providers=providers) @router.get( "/telephony-config-warnings", response_model=TelephonyConfigWarningsResponse, ) async def get_telephony_config_warnings(user: UserModel = Depends(get_user)): """Return aggregated warning counts for the current org's telephony configs. Surfaces provider configs missing webhook-verification credentials. """ if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") telnyx_missing = await db_client.count_telnyx_configs_missing_webhook_public_key( user.selected_organization_id ) vonage_missing = await db_client.count_vonage_configs_missing_signature_secret( user.selected_organization_id ) return TelephonyConfigWarningsResponse( telnyx_missing_webhook_public_key_count=telnyx_missing, vonage_missing_signature_secret_count=vonage_missing, ) # --------------------------------------------------------------------------- # AI model configurations v2 # --------------------------------------------------------------------------- def _dograh_allows_custom_voice() -> bool: extra = DograhTTSService.model_fields["voice"].json_schema_extra if isinstance(extra, dict): return bool(extra.get("allow_custom_input", False)) return False def _byok_provider_schemas(service_type: ServiceType) -> dict[str, dict]: return { provider: model_cls.model_json_schema() for provider, model_cls in REGISTRY[service_type].items() if provider != ServiceProviders.DOGRAH.value } async def _model_configuration_v2_response( *, user: UserModel, configuration: OrganizationAIModelConfigurationV2 | None = None, ) -> OrganizationAIModelConfigurationResponse: resolved = await get_resolved_ai_model_configuration( organization_id=user.selected_organization_id, ) raw_configuration = ( configuration if configuration is not None else resolved.organization_configuration ) return OrganizationAIModelConfigurationResponse( configuration=mask_ai_model_configuration_v2(raw_configuration), effective_configuration=mask_user_config(resolved.effective), source=resolved.source, ) @router.get("/model-configurations/v2/defaults") async def get_model_configuration_v2_defaults( user: UserModel = Depends(get_user_with_selected_organization), ): byok_default_providers = { service: provider for service, provider in DEFAULT_SERVICE_PROVIDERS.items() if provider != ServiceProviders.DOGRAH.value } return { "dograh": { "voices": [DOGRAH_DEFAULT_VOICE], "allow_custom_input": _dograh_allows_custom_voice(), "speeds": list(DOGRAH_SPEED_OPTIONS), "speed_range": { "min": DOGRAH_SPEED_MIN, "max": DOGRAH_SPEED_MAX, "step": DOGRAH_SPEED_STEP, }, "languages": DOGRAH_STT_LANGUAGES, "multilingual_languages": DOGRAH_MULTILINGUAL_AUTODETECT_LANGUAGES, "defaults": { "voice": DOGRAH_DEFAULT_VOICE, "speed": 1.0, "language": DOGRAH_DEFAULT_LANGUAGE, }, }, "byok": { "pipeline": { "llm": _byok_provider_schemas(ServiceType.LLM), "tts": _byok_provider_schemas(ServiceType.TTS), "stt": _byok_provider_schemas(ServiceType.STT), "embeddings": _byok_provider_schemas(ServiceType.EMBEDDINGS), "default_providers": byok_default_providers, }, "realtime": { "realtime": _byok_provider_schemas(ServiceType.REALTIME), "llm": _byok_provider_schemas(ServiceType.LLM), "embeddings": _byok_provider_schemas(ServiceType.EMBEDDINGS), "default_providers": byok_default_providers, }, }, } @router.get( "/model-configurations/v2", response_model=OrganizationAIModelConfigurationResponse, ) async def get_model_configuration_v2( user: UserModel = Depends(get_user_with_selected_organization), ): return await _model_configuration_v2_response(user=user) @router.get( "/model-configurations/v2/pricing", response_model=ModelConfigurationPricingResponse, ) async def get_model_configuration_pricing( user: UserModel = Depends(get_user_with_selected_organization), ) -> ModelConfigurationPricingResponse: """Return the hosted organization prices shown in Model Configurations.""" if DEPLOYMENT_MODE == "oss": return ModelConfigurationPricingResponse() try: pricing = await mps_service_key_client.get_billing_pricing( user.selected_organization_id, ) return ModelConfigurationPricingResponse.model_validate(pricing) except Exception as exc: logger.error( "Failed to get MPS model-configuration pricing for organization {}: {}", user.selected_organization_id, exc, ) raise HTTPException( status_code=502, detail="Failed to retrieve model configuration pricing", ) from exc @router.put( "/model-configurations/v2", response_model=OrganizationAIModelConfigurationResponse, ) async def save_model_configuration_v2( request: OrganizationAIModelConfigurationV2, user: UserModel = Depends(get_user_with_selected_organization), ): organization_id = user.selected_organization_id existing = await get_organization_ai_model_configuration_v2(organization_id) configuration = merge_ai_model_configuration_v2_secrets(request, existing) try: check_for_masked_keys_in_ai_model_configuration_v2(configuration) effective = compile_ai_model_configuration_v2(configuration) await UserConfigurationValidator().validate( effective, organization_id=organization_id, created_by=user.provider_id, ) except ValueError as exc: raise HTTPException(status_code=422, detail=exc.args[0]) await upsert_organization_ai_model_configuration_v2( organization_id, configuration, ) return await _model_configuration_v2_response( user=user, configuration=configuration, ) @router.get("/model-configurations/v2/migration-preview") async def preview_model_configuration_v2_migration( user: UserModel = Depends(get_user_with_selected_organization), ): legacy = await db_client.get_user_configurations(user.id) try: configuration = convert_legacy_ai_model_configuration_to_v2(legacy) except ValueError as exc: raise HTTPException(status_code=422, detail=str(exc)) return { "configuration": mask_ai_model_configuration_v2(configuration), "effective_configuration": mask_user_config( compile_ai_model_configuration_v2(configuration) ), } @router.post( "/model-configurations/v2/migrate", response_model=OrganizationAIModelConfigurationResponse, ) async def migrate_model_configuration_v2( force: bool = Query(default=False), user: UserModel = Depends(get_user_with_selected_organization), ): organization_id = user.selected_organization_id existing = await get_organization_ai_model_configuration_v2(organization_id) if existing is not None and not force: raise HTTPException( status_code=409, detail="Organization already has a v2 model configuration", ) legacy = await db_client.get_user_configurations(user.id) try: configuration = convert_legacy_ai_model_configuration_to_v2(legacy) effective = compile_ai_model_configuration_v2(configuration) await UserConfigurationValidator().validate( effective, organization_id=organization_id, created_by=user.provider_id, ) except ValueError as exc: raise HTTPException(status_code=422, detail=exc.args[0]) if DEPLOYMENT_MODE != "oss": try: await ensure_hosted_mps_billing_account_v2( organization_id, created_by=str(user.provider_id), ) except Exception as exc: logger.error( "Failed to initialize MPS billing account for organization {}: {}", organization_id, exc, ) raise HTTPException( status_code=502, detail="Failed to initialize MPS billing account", ) await upsert_organization_ai_model_configuration_v2( organization_id, configuration, ) await migrate_workflow_model_configurations_to_v2( organization_id=organization_id, fallback_user_config=legacy, ) return await _model_configuration_v2_response( user=user, configuration=configuration, ) @router.get("/preferences", response_model=OrganizationPreferences) async def get_preferences( user: UserModel = Depends(get_user_with_selected_organization), ): organization_id = user.selected_organization_id return await get_organization_preferences(organization_id) @router.put("/preferences", response_model=OrganizationPreferences) async def save_preferences( request: OrganizationPreferences, user: UserModel = Depends(get_user_with_selected_organization), ): organization_id = user.selected_organization_id return await upsert_organization_preferences( organization_id, request, ) @router.get( "/model-configurations/preferences", response_model=OrganizationPreferences, include_in_schema=False, ) async def get_model_configuration_preferences_legacy( user: UserModel = Depends(get_user_with_selected_organization), ): return await get_preferences(user=user) def preserve_masked_fields(provider: str, request_dict: dict, existing: dict): """If the client re-submitted a masked sensitive field, restore the original.""" for field_name in _sensitive_fields(provider): v = _get_nested_field(request_dict, field_name) existing_value = _get_nested_field(existing, field_name) if v and is_mask_of(v, existing_value or ""): _set_nested_field(request_dict, field_name, existing_value) def _get_nested_field(value: dict, dotted_path: str): current = value for part in dotted_path.split("."): if not isinstance(current, dict): return None current = current.get(part) return current def _set_nested_field(value: dict, dotted_path: str, field_value) -> None: current = value parts = dotted_path.split(".") for part in parts[:-1]: child = current.get(part) if not isinstance(child, dict): child = {} current[part] = child current = child current[parts[-1]] = field_value def _credentials_from_payload(config: TelephonyConfigRequest) -> dict: """Provider credentials only — strip provider/from_numbers from the payload.""" payload = config.model_dump() payload.pop("provider", None) payload.pop("from_numbers", None) return payload async def _run_preprocess_hook(provider: str, credentials: dict) -> dict: """Invoke the provider's optional credentials preprocessor before save.""" spec = telephony_registry.get_optional(provider) if spec and spec.preprocess_credentials_on_save: return await spec.preprocess_credentials_on_save(credentials) return credentials def _phone_number_to_response( row, inbound_workflow_name: Optional[str] = None ) -> PhoneNumberResponse: response = PhoneNumberResponse.model_validate(row) response.inbound_workflow_name = inbound_workflow_name return response async def _sync_inbound_for_phone_number( config_id: int, organization_id: int, address: str ) -> ProviderSyncStatus: """Push inbound webhook configuration to the provider. ``attach=True``: ask the provider to route this number's inbound calls to our workflow-agnostic dispatcher (``/api/v1/telephony/inbound/run``). ``attach=False``: ask the provider to detach. The dispatcher resolves the workflow from the called number's ``inbound_workflow_id``, so the webhook URL is the same for every assignment — providers only need to bind/unbind the number, not rewrite per-workflow URLs. """ try: provider = await get_telephony_provider_by_id(config_id, organization_id) except Exception as e: logger.error(f"Failed to load telephony provider for config {config_id}: {e}") return ProviderSyncStatus(ok=False, message=f"Provider load failed: {e}") backend_endpoint, _ = await get_backend_endpoints() webhook_url = f"{backend_endpoint}/api/v1/telephony/inbound/run" try: result = await provider.configure_inbound(address, webhook_url) except Exception as e: logger.error( f"Provider configure_inbound raised for config {config_id} " f"address {address}: {e}" ) return ProviderSyncStatus(ok=False, message=f"Provider sync failed: {e}") return ProviderSyncStatus(ok=result.ok, message=result.message) # --------------------------------------------------------------------------- # Multi-config CRUD # --------------------------------------------------------------------------- @router.get("/telephony-configs", response_model=TelephonyConfigurationListResponse) async def list_telephony_configurations(user: UserModel = Depends(get_user)): """List the org's telephony configurations with phone-number counts.""" if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") rows = await db_client.list_telephony_configurations(user.selected_organization_id) items: List[TelephonyConfigurationListItem] = [] for row in rows: numbers = await db_client.list_phone_numbers_for_config(row.id) items.append( TelephonyConfigurationListItem( id=row.id, name=row.name, provider=row.provider, is_default_outbound=row.is_default_outbound, phone_number_count=len([n for n in numbers if n.is_active]), created_at=row.created_at, updated_at=row.updated_at, ) ) return TelephonyConfigurationListResponse(configurations=items) @router.post("/telephony-configs", response_model=TelephonyConfigurationDetail) async def create_telephony_configuration( request: TelephonyConfigurationCreateRequest, user: UserModel = Depends(get_user), ): """Create a new telephony configuration for the org.""" if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") credentials = _credentials_from_payload(request.config) credentials = await _run_preprocess_hook(request.config.provider, credentials) try: row = await db_client.create_telephony_configuration( organization_id=user.selected_organization_id, name=request.name, provider=request.config.provider, credentials=credentials, is_default_outbound=request.is_default_outbound, ) except IntegrityError as e: if "uq_telephony_configurations_org_name" in str(e): raise HTTPException( status_code=409, detail=( f"A telephony configuration named '{request.name}' already " f"exists in this organization. Pick a different name." ), ) raise HTTPException( status_code=409, detail="Telephony configuration violates a uniqueness constraint.", ) capture_event( distinct_id=str(user.provider_id), event=PostHogEvent.TELEPHONY_CONFIGURED, properties={ "provider": request.config.provider, "organization_id": user.selected_organization_id, "config_id": row.id, }, ) return _detail_response(row) @router.get( "/telephony-configs/{config_id}", response_model=TelephonyConfigurationDetail ) async def get_telephony_configuration_by_id( config_id: int, user: UserModel = Depends(get_user) ): if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") row = await db_client.get_telephony_configuration_for_org( config_id, user.selected_organization_id ) if not row: raise HTTPException(status_code=404, detail="Telephony configuration not found") return _detail_response(row) @router.put( "/telephony-configs/{config_id}", response_model=TelephonyConfigurationDetail ) async def update_telephony_configuration( config_id: int, request: TelephonyConfigurationUpdateRequest, user: UserModel = Depends(get_user), ): if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") existing = await db_client.get_telephony_configuration_for_org( config_id, user.selected_organization_id ) if not existing: raise HTTPException(status_code=404, detail="Telephony configuration not found") credentials = None if request.config is not None: if request.config.provider != existing.provider: raise HTTPException( status_code=400, detail="Provider cannot be changed; create a new configuration instead.", ) credentials = _credentials_from_payload(request.config) preserve_masked_fields( existing.provider, credentials, existing.credentials or {} ) credentials = await _run_preprocess_hook(existing.provider, credentials) row = await db_client.update_telephony_configuration( config_id=config_id, organization_id=user.selected_organization_id, name=request.name, credentials=credentials, ) return _detail_response(row) @router.post( "/telephony-configs/{config_id}/set-default-outbound", response_model=TelephonyConfigurationDetail, ) async def set_default_outbound(config_id: int, user: UserModel = Depends(get_user)): if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") row = await db_client.set_default_telephony_configuration( config_id, user.selected_organization_id ) if not row: raise HTTPException(status_code=404, detail="Telephony configuration not found") return _detail_response(row) @router.delete("/telephony-configs/{config_id}") async def delete_telephony_configuration( config_id: int, user: UserModel = Depends(get_user) ): if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") try: deleted = await db_client.delete_telephony_configuration( config_id, user.selected_organization_id ) except TelephonyConfigurationInUseError as e: raise HTTPException(status_code=409, detail=str(e)) if not deleted: raise HTTPException(status_code=404, detail="Telephony configuration not found") return {"message": "Telephony configuration deleted"} def _detail_response(row) -> TelephonyConfigurationDetail: masked = _mask_sensitive(row.provider, row.credentials or {}) return TelephonyConfigurationDetail( id=row.id, name=row.name, provider=row.provider, is_default_outbound=row.is_default_outbound, credentials=masked, created_at=row.created_at, updated_at=row.updated_at, ) # --------------------------------------------------------------------------- # Phone numbers (nested under a config) # --------------------------------------------------------------------------- async def _ensure_config_belongs_to_org(config_id: int, organization_id: int): cfg = await db_client.get_telephony_configuration_for_org( config_id, organization_id ) if not cfg: raise HTTPException(status_code=404, detail="Telephony configuration not found") return cfg async def _ensure_workflow_belongs_to_org(workflow_id: int, organization_id: int): workflow = await db_client.get_workflow( workflow_id, organization_id=organization_id ) if not workflow: raise HTTPException(status_code=404, detail="Workflow not found") return workflow @router.get( "/telephony-configs/{config_id}/phone-numbers", response_model=PhoneNumberListResponse, ) async def list_phone_numbers(config_id: int, user: UserModel = Depends(get_user)): if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") await _ensure_config_belongs_to_org(config_id, user.selected_organization_id) rows = await db_client.list_phone_numbers_with_workflow_name_for_config(config_id) return PhoneNumberListResponse( phone_numbers=[_phone_number_to_response(r, name) for r, name in rows] ) @router.post( "/telephony-configs/{config_id}/phone-numbers", response_model=PhoneNumberResponse, ) async def create_phone_number( config_id: int, request: PhoneNumberCreateRequest, user: UserModel = Depends(get_user), ): if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") cfg = await _ensure_config_belongs_to_org(config_id, user.selected_organization_id) if request.inbound_workflow_id is not None: await _ensure_workflow_belongs_to_org( request.inbound_workflow_id, user.selected_organization_id ) # Inbound dispatch (find_inbound_route_by_account) keys on (provider, # credentials[account_id_field], address_normalized) without the org, so # that tuple has to be globally unique. Reject up front if another config — # in this org or any other — already owns the same combination. spec = telephony_registry.get_optional(cfg.provider) account_field = spec.account_id_credential_field if spec else "" account_id = (cfg.credentials or {}).get(account_field) if account_field else None if account_id: try: conflict = await db_client.find_inbound_routing_conflict( provider=cfg.provider, account_id_field=account_field, account_id=account_id, address=request.address, country_hint=request.country_code, ) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) if conflict: existing_cfg, existing_phone = conflict same_org = existing_cfg.organization_id == user.selected_organization_id scope = ( f"telephony configuration '{existing_cfg.name}'" if same_org else "another organization using the same provider account" ) raise HTTPException( status_code=409, detail=( f"Phone number {existing_phone.address} is already registered " f"under {scope}. Inbound calls cannot be uniquely routed when " f"the same number is configured against the same provider " f"account in more than one place." ), ) try: row = await db_client.create_phone_number( organization_id=user.selected_organization_id, telephony_configuration_id=config_id, address=request.address, country_code=request.country_code, label=request.label, inbound_workflow_id=request.inbound_workflow_id, is_active=request.is_active, is_default_caller_id=request.is_default_caller_id, extra_metadata=request.extra_metadata, ) except IntegrityError: raise HTTPException( status_code=409, detail="A phone number with this address already exists in the org.", ) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) response = _phone_number_to_response(row) if request.inbound_workflow_id is not None: response.provider_sync = await _sync_inbound_for_phone_number( config_id, user.selected_organization_id, row.address ) return response @router.get( "/telephony-configs/{config_id}/phone-numbers/{phone_number_id}", response_model=PhoneNumberResponse, ) async def get_phone_number( config_id: int, phone_number_id: int, user: UserModel = Depends(get_user), ): if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") await _ensure_config_belongs_to_org(config_id, user.selected_organization_id) row = await db_client.get_phone_number_for_config(phone_number_id, config_id) if not row: raise HTTPException(status_code=404, detail="Phone number not found") return _phone_number_to_response(row) @router.put( "/telephony-configs/{config_id}/phone-numbers/{phone_number_id}", response_model=PhoneNumberResponse, ) async def update_phone_number( config_id: int, phone_number_id: int, request: PhoneNumberUpdateRequest, user: UserModel = Depends(get_user), ): if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") await _ensure_config_belongs_to_org(config_id, user.selected_organization_id) existing = await db_client.get_phone_number_for_config(phone_number_id, config_id) if not existing: raise HTTPException(status_code=404, detail="Phone number not found") if request.inbound_workflow_id is not None: await _ensure_workflow_belongs_to_org( request.inbound_workflow_id, user.selected_organization_id ) row = await db_client.update_phone_number( phone_number_id=phone_number_id, telephony_configuration_id=config_id, label=request.label, inbound_workflow_id=request.inbound_workflow_id, is_active=request.is_active, country_code=request.country_code, extra_metadata=request.extra_metadata, clear_inbound_workflow=request.clear_inbound_workflow, ) if not row: raise HTTPException(status_code=404, detail="Phone number not found") response = _phone_number_to_response(row) # Sync the provider application or address with the inbound # calling webhook address response.provider_sync = await _sync_inbound_for_phone_number( config_id, user.selected_organization_id, row.address ) return response @router.post( "/telephony-configs/{config_id}/phone-numbers/{phone_number_id}/set-default-caller", response_model=PhoneNumberResponse, ) async def set_default_caller_id( config_id: int, phone_number_id: int, user: UserModel = Depends(get_user), ): if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") await _ensure_config_belongs_to_org(config_id, user.selected_organization_id) row = await db_client.set_default_caller_id(phone_number_id, config_id) if not row: raise HTTPException(status_code=404, detail="Phone number not found") return _phone_number_to_response(row) @router.delete("/telephony-configs/{config_id}/phone-numbers/{phone_number_id}") async def delete_phone_number( config_id: int, phone_number_id: int, user: UserModel = Depends(get_user), ): if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") await _ensure_config_belongs_to_org(config_id, user.selected_organization_id) existing = await db_client.get_phone_number_for_config(phone_number_id, config_id) if not existing: raise HTTPException(status_code=404, detail="Phone number not found") deleted = await db_client.delete_phone_number(phone_number_id, config_id) if not deleted: raise HTTPException(status_code=404, detail="Phone number not found") return {"message": "Phone number deleted"} # --------------------------------------------------------------------------- # Legacy single-config shim # --------------------------------------------------------------------------- @router.get("/telephony-config", response_model=TelephonyConfigurationResponse) async def get_telephony_configuration(user: UserModel = Depends(get_user)): """Legacy: returns the org's default config in the original per-provider response shape so the existing single-form UI keeps working. Prefer the multi-config endpoints (``/telephony-configs``) for new clients. """ if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") cfg = await db_client.get_default_telephony_configuration( user.selected_organization_id ) if not cfg: return TelephonyConfigurationResponse() spec = telephony_registry.get_optional(cfg.provider) if spec is None: return TelephonyConfigurationResponse() addresses = await db_client.list_active_normalized_addresses_for_config(cfg.id) masked = _mask_sensitive(cfg.provider, cfg.credentials or {}) payload = {**masked, "provider": cfg.provider, "from_numbers": addresses} response_obj = spec.config_response_cls.model_validate(payload) return TelephonyConfigurationResponse(**{cfg.provider: response_obj}) @router.post("/telephony-config") async def save_telephony_configuration( request: TelephonyConfigRequest, user: UserModel = Depends(get_user), ): """Legacy: upserts the org's default config (and its phone numbers) in the original payload shape so existing UI clients keep working. Prefer the multi-config + phone-number endpoints for new clients. """ if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") payload = request.model_dump() new_addresses = payload.pop("from_numbers", []) or [] payload.pop("provider", None) default = await db_client.get_default_telephony_configuration( user.selected_organization_id ) if default and default.provider == request.provider: preserve_masked_fields(request.provider, payload, default.credentials or {}) row = await db_client.update_telephony_configuration( config_id=default.id, organization_id=user.selected_organization_id, credentials=payload, ) else: row = await db_client.create_telephony_configuration( organization_id=user.selected_organization_id, name=f"{request.provider.title()} Default", provider=request.provider, credentials=payload, is_default_outbound=True, ) # Replace the phone-number set with the inline payload. existing_numbers = await db_client.list_phone_numbers_for_config(row.id) existing_by_address = {n.address: n for n in existing_numbers} incoming_set = set(new_addresses) for addr in new_addresses: if addr in existing_by_address: continue try: await db_client.create_phone_number( organization_id=user.selected_organization_id, telephony_configuration_id=row.id, address=addr, ) except IntegrityError: logger.warning( f"Skipping duplicate phone number {addr!r} for config {row.id}" ) except ValueError as e: logger.warning(f"Skipping invalid phone number {addr!r}: {e}") for n in existing_numbers: if n.address not in incoming_set: await db_client.delete_phone_number(n.id, row.id) capture_event( distinct_id=str(user.provider_id), event=PostHogEvent.TELEPHONY_CONFIGURED, properties={ "provider": request.provider, "phone_number_count": len(new_addresses), "organization_id": user.selected_organization_id, }, ) return {"message": "Telephony configuration saved successfully"} class LangfuseCredentialsRequest(BaseModel): host: str public_key: str secret_key: str class LangfuseCredentialsResponse(BaseModel): host: str = "" public_key: str = "" secret_key: str = "" configured: bool = False @router.get("/langfuse-credentials", response_model=LangfuseCredentialsResponse) async def get_langfuse_credentials(user: UserModel = Depends(get_user)): """Get Langfuse credentials for the user's organization with masked sensitive fields.""" if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") config = await db_client.get_configuration( user.selected_organization_id, OrganizationConfigurationKey.LANGFUSE_CREDENTIALS.value, ) if not config or not config.value: return LangfuseCredentialsResponse() return LangfuseCredentialsResponse( host=config.value.get("host", ""), public_key=mask_key(config.value.get("public_key", "")), secret_key=mask_key(config.value.get("secret_key", "")), configured=True, ) @router.post("/langfuse-credentials") async def save_langfuse_credentials( request: LangfuseCredentialsRequest, user: UserModel = Depends(get_user), ): """Save Langfuse credentials for the user's organization.""" if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") existing_config = await db_client.get_configuration( user.selected_organization_id, OrganizationConfigurationKey.LANGFUSE_CREDENTIALS.value, ) config_value = { "host": request.host, "public_key": request.public_key, "secret_key": request.secret_key, } # Preserve masked fields if existing_config and existing_config.value: if is_mask_of(request.public_key, existing_config.value.get("public_key", "")): config_value["public_key"] = existing_config.value["public_key"] if is_mask_of(request.secret_key, existing_config.value.get("secret_key", "")): config_value["secret_key"] = existing_config.value["secret_key"] await db_client.upsert_configuration( user.selected_organization_id, OrganizationConfigurationKey.LANGFUSE_CREDENTIALS.value, config_value, ) # Broadcast to all workers so every process updates its in-memory exporter await get_worker_sync_manager().broadcast( WorkerSyncEventType.LANGFUSE_CREDENTIALS, action="update", org_id=user.selected_organization_id, ) return {"message": "Langfuse credentials saved successfully"} @router.delete("/langfuse-credentials") async def delete_langfuse_credentials(user: UserModel = Depends(get_user)): """Delete Langfuse credentials for the user's organization.""" if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") deleted = await db_client.delete_configuration( user.selected_organization_id, OrganizationConfigurationKey.LANGFUSE_CREDENTIALS.value, ) if not deleted: raise HTTPException(status_code=404, detail="No Langfuse credentials found") # Broadcast to all workers so every process removes its in-memory exporter await get_worker_sync_manager().broadcast( WorkerSyncEventType.LANGFUSE_CREDENTIALS, action="delete", org_id=user.selected_organization_id, ) return {"message": "Langfuse credentials deleted successfully"} class RetryConfigResponse(BaseModel): enabled: bool max_retries: int retry_delay_seconds: int retry_on_busy: bool retry_on_no_answer: bool retry_on_voicemail: bool class TimeSlotResponse(BaseModel): day_of_week: int start_time: str end_time: str class ScheduleConfigResponse(BaseModel): enabled: bool timezone: str slots: List[TimeSlotResponse] class CircuitBreakerConfigResponse(BaseModel): enabled: bool = False failure_threshold: float = 0.5 window_seconds: int = 120 min_calls_in_window: int = 5 class LastCampaignSettingsResponse(BaseModel): retry_config: Optional[RetryConfigResponse] = None max_concurrency: Optional[int] = None schedule_config: Optional[ScheduleConfigResponse] = None circuit_breaker: Optional[CircuitBreakerConfigResponse] = None class CampaignDefaultsResponse(BaseModel): concurrent_call_limit: int from_numbers_count: int default_retry_config: RetryConfigResponse last_campaign_settings: Optional[LastCampaignSettingsResponse] = None @router.get("/campaign-defaults", response_model=CampaignDefaultsResponse) async def get_campaign_defaults(user: UserModel = Depends(get_user)): """Get campaign limits for the user's organization. Returns the organization's concurrent call limit and default retry configuration. """ if not user.selected_organization_id: raise HTTPException(status_code=400, detail="No organization selected") # Get concurrent call limit concurrent_limit = DEFAULT_ORG_CONCURRENCY_LIMIT try: config = await db_client.get_configuration( user.selected_organization_id, OrganizationConfigurationKey.CONCURRENT_CALL_LIMIT.value, ) if config and config.value: concurrent_limit = int( config.value.get("value", DEFAULT_ORG_CONCURRENCY_LIMIT) ) except Exception: pass # Phone-number count from the org's default telephony config (used by the # campaign UI to validate max_concurrency against caller-id supply). from_numbers_count = 0 try: default_cfg = await db_client.get_default_telephony_configuration( user.selected_organization_id ) if default_cfg: addresses = await db_client.list_active_normalized_addresses_for_config( default_cfg.id ) from_numbers_count = len(addresses) except Exception: pass # Get last campaign settings for pre-population last_campaign_settings = None try: last_campaign = await db_client.get_latest_campaign( user.selected_organization_id ) if last_campaign: retry = None if last_campaign.retry_config: retry = RetryConfigResponse(**last_campaign.retry_config) max_conc = None sched = None cb = CircuitBreakerConfigResponse() if last_campaign.orchestrator_metadata: max_conc = last_campaign.orchestrator_metadata.get("max_concurrency") sc = last_campaign.orchestrator_metadata.get("schedule_config") if sc: sched = ScheduleConfigResponse( enabled=sc.get("enabled", False), timezone=sc.get("timezone", "UTC"), slots=[ TimeSlotResponse(**slot) for slot in sc.get("slots", []) ], ) cb_data = last_campaign.orchestrator_metadata.get("circuit_breaker") if cb_data: cb = CircuitBreakerConfigResponse(**cb_data) else: cb = CircuitBreakerConfigResponse() last_campaign_settings = LastCampaignSettingsResponse( retry_config=retry, max_concurrency=max_conc, schedule_config=sched, circuit_breaker=cb, ) except Exception: pass return CampaignDefaultsResponse( concurrent_call_limit=concurrent_limit, from_numbers_count=from_numbers_count, default_retry_config=RetryConfigResponse(**DEFAULT_CAMPAIGN_RETRY_CONFIG), last_campaign_settings=last_campaign_settings, )