refactor(pipeline): migrate ws egress to Redis Pub/Sub and optimize A…

app/core/config.py

@@ -119,10 +119,10 @@ class Settings(BaseSettings):
     FRONTEND_BASE_URL: str = "http://localhost:4200"
 
     # CORS
-    CORS_ORIGINS: list[str] = [
-        "http://localhost:4200",
-        "https://spoken-frontend.onrender.com",
-    ]
+    @property
+    def CORS_ORIGINS(self) -> list[str]:
+        origins = [self.FRONTEND_BASE_URL.rstrip("/")] if self.FRONTEND_BASE_URL else []
+        return [o for o in origins if o]
 
     model_config = SettingsConfigDict(
         env_file=".env", case_sensitive=True, extra="ignore"

app/db/session.py

@@ -70,7 +70,10 @@ def get_engine() -> Engine:
         # pool_recycle handles server-side idle timeouts (Supabase/DigitalOcean).
         connect_args = {}
         if DATABASE_URL.startswith("postgresql"):
-            connect_args["sslmode"] = "require"
+            if "localhost" in DATABASE_URL:
+                connect_args["sslmode"] = "disable"
+            else:
+                connect_args["sslmode"] = "require"
 
         try:
             cached_engine = create_engine(

app/external_services/deepgram/service.py

@@ -29,6 +29,13 @@ class DeepgramSTTService:
 
     def __init__(self, timeout: float = 10.0) -> None:
         self._timeout = timeout
+        self._client: httpx.AsyncClient | None = None
+
+    @property
+    def client(self) -> httpx.AsyncClient:
+        if self._client is None or self._client.is_closed:
+            self._client = httpx.AsyncClient(timeout=self._timeout)
+        return self._client
 
     async def transcribe(
         self,
@@ -63,14 +70,13 @@ async def transcribe(
         }
 
         start = time.monotonic()
-        async with httpx.AsyncClient(timeout=self._timeout) as client:
-            response = await client.post(
-                settings.DEEPGRAM_API_URL,
-                headers=headers,
-                params=params,
-                content=audio_bytes,
-            )
-            response.raise_for_status()
+        response = await self.client.post(
+            settings.DEEPGRAM_API_URL,
+            headers=headers,
+            params=params,
+            content=audio_bytes,
+        )
+        response.raise_for_status()
 
         elapsed_ms = (time.monotonic() - start) * 1000
         logger.debug("Deepgram STT completed in %.1fms", elapsed_ms)

app/external_services/deepl/service.py

@@ -55,6 +55,13 @@ class DeepLTranslationService:
 
     def __init__(self, timeout: float = 10.0) -> None:
         self._timeout = timeout
+        self._client: httpx.AsyncClient | None = None
+
+    @property
+    def client(self) -> httpx.AsyncClient:
+        if self._client is None or self._client.is_closed:
+            self._client = httpx.AsyncClient(timeout=self._timeout)
+        return self._client
 
     async def translate(
         self,
@@ -88,13 +95,12 @@ async def translate(
             payload["source_lang"] = deepl_source
 
         start = time.monotonic()
-        async with httpx.AsyncClient(timeout=self._timeout) as client:
-            response = await client.post(
-                settings.DEEPL_API_URL,
-                headers=headers,
-                json=payload,
-            )
-            response.raise_for_status()
+        response = await self.client.post(
+            settings.DEEPL_API_URL,
+            headers=headers,
+            json=payload,
+        )
+        response.raise_for_status()
 
         elapsed_ms = (time.monotonic() - start) * 1000
         logger.debug("DeepL translation completed in %.1fms", elapsed_ms)
@@ -131,6 +137,13 @@ class OpenAITranslationFallback:
 
     def __init__(self, timeout: float = 15.0) -> None:
         self._timeout = timeout
+        self._client: httpx.AsyncClient | None = None
+
+    @property
+    def client(self) -> httpx.AsyncClient:
+        if self._client is None or self._client.is_closed:
+            self._client = httpx.AsyncClient(timeout=self._timeout)
+        return self._client
 
     async def translate(
         self,
@@ -176,13 +189,12 @@ async def translate(
         }
 
         start = time.monotonic()
-        async with httpx.AsyncClient(timeout=self._timeout) as client:
-            response = await client.post(
-                "https://api.openai.com/v1/chat/completions",
-                headers=headers,
-                json=payload,
-            )
-            response.raise_for_status()
+        response = await self.client.post(
+            "https://api.openai.com/v1/chat/completions",
+            headers=headers,
+            json=payload,
+        )
+        response.raise_for_status()
 
         elapsed_ms = (time.monotonic() - start) * 1000
         logger.debug("OpenAI translation fallback completed in %.1fms", elapsed_ms)

app/external_services/openai_tts/service.py

@@ -33,6 +33,13 @@ class OpenAITTSService:
 
     def __init__(self, timeout: float = 15.0) -> None:
         self._timeout = timeout
+        self._client: httpx.AsyncClient | None = None
+
+    @property
+    def client(self) -> httpx.AsyncClient:
+        if self._client is None or self._client.is_closed:
+            self._client = httpx.AsyncClient(timeout=self._timeout)
+        return self._client
 
     async def synthesize(
         self,
@@ -68,13 +75,12 @@ async def synthesize(
         }
 
         start = time.monotonic()
-        async with httpx.AsyncClient(timeout=self._timeout) as client:
-            response = await client.post(
-                settings.OPENAI_TTS_API_URL,
-                headers=headers,
-                json=payload,
-            )
-            response.raise_for_status()
+        response = await self.client.post(
+            settings.OPENAI_TTS_API_URL,
+            headers=headers,
+            json=payload,
+        )
+        response.raise_for_status()
 
         elapsed_ms = (time.monotonic() - start) * 1000
         logger.debug("OpenAI TTS completed in %.1fms", elapsed_ms)

app/external_services/voiceai/service.py

@@ -36,6 +36,13 @@ class VoiceAITTSService:
 
     def __init__(self, timeout: float = 60.0) -> None:
         self._timeout = timeout
+        self._client: httpx.AsyncClient | None = None
+
+    @property
+    def client(self) -> httpx.AsyncClient:
+        if self._client is None or self._client.is_closed:
+            self._client = httpx.AsyncClient(timeout=self._timeout)
+        return self._client
 
     async def synthesize(
         self,
@@ -83,22 +90,20 @@ async def synthesize(
             "temperature": 1,
             "top_p": 0.8,
         }
-        print(f"Voice.ai Audio format: {audio_format}")
+        logger.debug("Voice.ai Audio format: %s", audio_format)
         if voice_id:
             payload["voice_id"] = voice_id
 
         start = time.monotonic()
-        async with httpx.AsyncClient(timeout=self._timeout) as client:
-            response = await client.post(
-                settings.VOICEAI_TTS_API_URL,
-                headers=headers,
-                json=payload,
-            )
-            response.raise_for_status()
+        response = await self.client.post(
+            settings.VOICEAI_TTS_API_URL,
+            headers=headers,
+            json=payload,
+        )
+        response.raise_for_status()
 
         elapsed_ms = (time.monotonic() - start) * 1000
-        print(f"Voice.ai TTS API completed in {elapsed_ms} ms")
-        logger.debug("Voice.ai TTS completed in %.1fms", elapsed_ms)
+        logger.debug("Voice.ai TTS API completed in %.1fms", elapsed_ms)
 
         return {
             "audio_bytes": response.content,

app/kafka/consumer.py

@@ -123,7 +123,7 @@ async def _consume_loop(self) -> None:
                     age_ms = _time.time() * 1000 - msg.timestamp
                     if age_ms > self.max_message_age_ms:
                         topic_safe = sanitize_log_args(self.topic)[0]
-                        logger.debug(
+                        logger.info(
                             "Skipping stale message on '%s' (age=%.0fms > limit=%dms)",
                             topic_safe,
                             age_ms,

app/kafka/manager.py

@@ -93,10 +93,19 @@ async def _init_topics(self) -> None:
         await admin_client.start()
         try:
             # DLQ topics for each required topic + standard topics
+            PIPELINE_TOPIC_SET = {
+                AUDIO_RAW,
+                AUDIO_SYNTHESIZED,
+                TEXT_ORIGINAL,
+                TEXT_TRANSLATED,
+            }
             new_topics = []
             for topic in TOPICS_TO_CREATE:
+                partitions = 3 if topic in PIPELINE_TOPIC_SET else 1
                 new_topics.append(
-                    NewTopic(name=topic, num_partitions=1, replication_factor=1)
+                    NewTopic(
+                        name=topic, num_partitions=partitions, replication_factor=1
+                    )
                 )
                 new_topics.append(
                     NewTopic(

app/modules/auth/constants.py

@@ -15,3 +15,10 @@ class SupportedLanguage(enum.StrEnum):
     SPANISH = "es"
     ITALIAN = "it"
     PORTUGUESE = "pt"
+    DUTCH = "nl"
+    POLISH = "pl"
+    RUSSIAN = "ru"
+    JAPANESE = "ja"
+    CHINESE = "zh"
+    KOREAN = "ko"
+    TURKISH = "tr"

app/modules/meeting/constants.py

@@ -46,6 +46,40 @@ class InvitationStatus(enum.StrEnum):
 MSG_INVITATIONS_SENT = "Meeting invitations sent."
 
 
+# ── Supported Languages ───────────────────────────────────────────────
+SUPPORTED_LANGUAGES: Final[set[str]] = {
+    "en",  # English
+    "de",  # German
+    "fr",  # French
+    "es",  # Spanish
+    "it",  # Italian
+    "pt",  # Portuguese
+    "nl",  # Dutch
+    "pl",  # Polish
+    "ru",  # Russian
+    "ja",  # Japanese
+    "zh",  # Chinese
+    "ko",  # Korean
+    "tr",  # Turkish
+}
+
+LANGUAGE_NAMES: Final[dict[str, str]] = {
+    "en": "English",
+    "de": "German",
+    "fr": "French",
+    "es": "Spanish",
+    "it": "Italian",
+    "pt": "Portuguese",
+    "nl": "Dutch",
+    "pl": "Polish",
+    "ru": "Russian",
+    "ja": "Japanese",
+    "zh": "Chinese",
+    "ko": "Korean",
+    "tr": "Turkish",
+}
+
+
 # ── Redis Key Patterns ────────────────────────────────────────────────
 def key_room_participants(room_code: str) -> str:
     return f"room:{room_code}:participants"

app/modules/meeting/schemas.py

@@ -7,7 +7,9 @@
 import uuid
 from datetime import datetime
 
-from pydantic import BaseModel, ConfigDict, EmailStr, Field
+from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
+
+from app.modules.meeting.constants import SUPPORTED_LANGUAGES
 
 # ── Request schemas ───────────────────────────────────────────────────
 
@@ -62,6 +64,20 @@ class JoinRoomRequest(BaseModel):
         "Used for STT source language selection.",
     )
 
+    @field_validator("listening_language", "speaking_language", mode="before")
+    @classmethod
+    def validate_language_code(cls, v: str | None) -> str | None:
+        """Ensure language codes are supported ISO 639-1 values."""
+        if v is None:
+            return v
+        code = v.strip().lower()
+        if code not in SUPPORTED_LANGUAGES:
+            supported = ", ".join(sorted(SUPPORTED_LANGUAGES))
+            raise ValueError(
+                f"Unsupported language '{code}'. Supported languages: {supported}"
+            )
+        return code
+
 
 # ── Response schemas ──────────────────────────────────────────────────
 

app/modules/meeting/state.py

@@ -1,7 +1,7 @@
 """Meeting ephemeral Redis State Service module.
 
-Generates atomic mapping tracking natively memory limits smoothly defining
-targets natively.
+Manages live meeting state (lobby, participants, active speaker) using Redis
+for high-performance ephemeral storage.
 """
 
 import json
@@ -23,11 +23,10 @@
 
 
 class MeetingStateService:
-    """Manages ephemeral live room state (lobby, participants presence,
-    active speaker) in Redis.
+    """Manages ephemeral live room state (lobby, participants, active speaker)
+    in Redis.
 
-    All operations are asynchronous and hit Redis directly smoothly handling
-    maps natively seamlessly.
+    All operations are asynchronous and interact with Redis directly.
     """
 
     def __init__(self, redis_client: aioredis.Redis | None = None) -> None:
@@ -48,16 +47,13 @@ async def add_participant(
         """Add or update a user's presence in the active room participants hash.
 
         Args:
-            room_code (str): Identity parameter dynamically natively resolving
-                identifiers.
-            user_id (str): User tracker string mapped locally natively limits
-                logically securely bindings natively.
-            language (str): Locale configuration gracefully array mapping.
-            hardware_ready (bool): Configuration map dynamically natively smoothly
-                correctly natively tracking gracefully gracefully locally securely
-                smoothly gracefully tracking natively handled array limit logically
-                seamlessly bounds dynamically safely correctly securely limits
-                correctly dynamically.
+            room_code: The room's unique identifier code.
+            user_id: The participant's unique identifier.
+            language: The participant's listening language (ISO 639-1).
+            speaking_language: The participant's speaking language (ISO 639-1).
+            hardware_ready: Whether the user's media devices are ready.
+            display_name: The participant's display name.
+            role: The participant's role (host, guest, participant).
         """
         state = {
             "status": "connected",
@@ -80,11 +76,8 @@ async def remove_participant(self, room_code: str, user_id: str) -> None:
         """Remove a user from the active participants hash.
 
         Args:
-            room_code (str): Identity string naturally resolving natively
-                gracefully limits seamlessly dynamically correctly safely mapping
-                dynamically.
-            user_id (str): Evaluator tracking string string parameter seamlessly
-                mapping efficiently limits.
+            room_code: The room's unique identifier code.
+            user_id: The participant's unique identifier.
         """
         await cast(
             "Awaitable[Any]",