"""Realtime domain — business logic.""" from __future__ import annotations from typing import Optional from uuid import UUID import structlog from src.core.exceptions import ConflictError, NotFoundError from src.domain.realtime.models import ProjectRealtimeChannel, ProjectRealtimeConfig from src.domain.realtime.repository import ( ProjectRealtimeChannelRepository, ProjectRealtimeConfigRepository, ) from src.domain.realtime.schemas import ( BroadcastRequest, ChannelResponse, CreateChannelRequest, RealtimeConfigResponse, RealtimeStatsResponse, UpdateChannelRequest, UpdateRealtimeConfigRequest, ) from src.integrations.cloud_control_client import CloudControlClient logger = structlog.get_logger() # Max custom channels per project (soft limit) _DEFAULT_MAX_CHANNELS = 50 # Concurrent connection limits per plan tier — enforced by Cloud DO, declared here _PLAN_MAX_CONNECTIONS: dict[str, int] = { "free": 10, "starter": 50, "pro": 500, } class RealtimeService: def __init__( self, *, config_repo: ProjectRealtimeConfigRepository, channel_repo: ProjectRealtimeChannelRepository, cloud_client: CloudControlClient, ) -> None: self._config = config_repo self._channels = channel_repo self._cloud = cloud_client # ------------------------------------------------------------------ # Config # ------------------------------------------------------------------ async def get_or_create_config(self, project_id: UUID) -> RealtimeConfigResponse: cfg = await self._config.get_by_project(project_id) if not cfg: cfg = await self._config.create(project_id=project_id) return _config_response(cfg) async def update_config( self, project_id: UUID, body: UpdateRealtimeConfigRequest, plan_slug: Optional[str] = None, ) -> RealtimeConfigResponse: cfg = await self._config.get_by_project(project_id) if not cfg: cfg = await self._config.create(project_id=project_id) updates: dict = {} if body.enabled is not None: updates["enabled"] = body.enabled if body.history_enabled is not None: updates["history_enabled"] = body.history_enabled if body.history_buffer_size is not None: updates["history_buffer_size"] = body.history_buffer_size # Sync max_concurrent_connections from plan when enabling if updates.get("enabled") and plan_slug: updates["max_concurrent_connections"] = _plan_max_connections(plan_slug) if updates: cfg = await self._config.update(cfg, **updates) await self._push_config_to_cloud(project_id, cfg) logger.info("realtime_config_updated", project_id=str(project_id)) return _config_response(cfg) # ------------------------------------------------------------------ # Channels # ------------------------------------------------------------------ async def list_channels(self, project_id: UUID) -> list[ChannelResponse]: rows = await self._channels.list_for_project(project_id) return [_channel_response(r) for r in rows] async def create_channel( self, project_id: UUID, body: CreateChannelRequest ) -> ChannelResponse: existing = await self._channels.get_by_name(project_id, body.name) if existing: raise ConflictError(f"Channel {body.name!r} already exists for this project") count = await self._channels.count_for_project(project_id) if count >= _DEFAULT_MAX_CHANNELS: raise ConflictError(f"Custom channel limit ({_DEFAULT_MAX_CHANNELS}) reached") row = await self._channels.create( project_id=project_id, name=body.name, access_policy=body.access_policy, rls_filter=body.rls_filter, ) await self._push_channel_to_cloud(project_id, row) logger.info("realtime_channel_created", project_id=str(project_id), name=body.name) return _channel_response(row) async def update_channel( self, project_id: UUID, name: str, body: UpdateChannelRequest ) -> ChannelResponse: row = await self._channels.get_by_name(project_id, name) if not row: raise NotFoundError(f"Channel {name!r} not found") updates: dict = {} if body.access_policy is not None: updates["access_policy"] = body.access_policy if body.rls_filter is not None: updates["rls_filter"] = body.rls_filter if updates: row = await self._channels.update(row, **updates) await self._push_channel_to_cloud(project_id, row) logger.info("realtime_channel_updated", project_id=str(project_id), name=name) return _channel_response(row) async def delete_channel(self, project_id: UUID, name: str) -> None: row = await self._channels.get_by_name(project_id, name) if not row: raise NotFoundError(f"Channel {name!r} not found") await self._channels.delete(row) await self._cloud.delete_realtime_channel( project_id=str(project_id), name=name ) logger.info("realtime_channel_deleted", project_id=str(project_id), name=name) # ------------------------------------------------------------------ # Stats # ------------------------------------------------------------------ async def get_stats(self, project_id: UUID) -> RealtimeStatsResponse: cfg = await self._config.get_by_project(project_id) channel_count = await self._channels.count_for_project(project_id) return RealtimeStatsResponse( project_id=project_id, enabled=cfg.enabled if cfg else False, channel_count=channel_count, max_concurrent_connections=cfg.max_concurrent_connections if cfg else 10, ) # ------------------------------------------------------------------ # Broadcast (Core → Cloud) # ------------------------------------------------------------------ async def broadcast(self, project_id: UUID, body: BroadcastRequest) -> None: await self._cloud.realtime_broadcast( project_id=str(project_id), channel=body.channel, event=body.event, data=body.data, ) logger.info( "realtime_broadcast", project_id=str(project_id), channel=body.channel, event=body.event, ) # ------------------------------------------------------------------ # Internal # ------------------------------------------------------------------ async def _push_config_to_cloud( self, project_id: UUID, cfg: ProjectRealtimeConfig ) -> None: await self._cloud.push_realtime_config( project_id=str(project_id), enabled=cfg.enabled, max_concurrent_connections=cfg.max_concurrent_connections, history_enabled=cfg.history_enabled, history_buffer_size=cfg.history_buffer_size, ) async def _push_channel_to_cloud( self, project_id: UUID, row: ProjectRealtimeChannel ) -> None: await self._cloud.upsert_realtime_channel( project_id=str(project_id), name=row.name, access_policy=row.access_policy, rls_filter=row.rls_filter, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _plan_max_connections(plan_slug: str) -> int: slug = plan_slug.lower() if "growth" in slug: return 10000 # effectively unlimited return _PLAN_MAX_CONNECTIONS.get(slug, 10) def _config_response(cfg: ProjectRealtimeConfig) -> RealtimeConfigResponse: return RealtimeConfigResponse( project_id=cfg.project_id, enabled=cfg.enabled, max_concurrent_connections=cfg.max_concurrent_connections, history_enabled=cfg.history_enabled, history_buffer_size=cfg.history_buffer_size, updated_at=cfg.updated_at, ) def _channel_response(row: ProjectRealtimeChannel) -> ChannelResponse: return ChannelResponse( id=row.id, project_id=row.project_id, name=row.name, access_policy=row.access_policy, rls_filter=row.rls_filter, created_at=row.created_at, )