"""Logs domain — business logic.""" from __future__ import annotations from datetime import datetime, timedelta, timezone from typing import Optional from uuid import UUID import httpx import structlog from src.core.exceptions import ConflictError, NotFoundError from src.core.security.encryption import decrypt_value, encrypt_value from src.domain.logs.models import ProjectLogConfig, ProjectLogDrain from src.domain.logs.repository import ProjectLogConfigRepository, ProjectLogDrainRepository from src.domain.logs.schemas import ( CreateLogDrainRequest, LogConfigResponse, LogDrainResponse, TestDrainResponse, UpdateLogConfigRequest, UpdateLogDrainRequest, _RETENTION_BY_PLAN, ) from src.integrations.cloud_control_client import CloudControlClient logger = structlog.get_logger() _MAX_DRAINS_PER_PROJECT = 5 class LogsService: def __init__( self, *, config_repo: ProjectLogConfigRepository, drain_repo: ProjectLogDrainRepository, cloud_client: CloudControlClient, ) -> None: self._config = config_repo self._drains = drain_repo self._cloud = cloud_client # ------------------------------------------------------------------ # Log config # ------------------------------------------------------------------ async def get_or_create_config( self, project_id: UUID, plan_slug: Optional[str] = None ) -> LogConfigResponse: cfg = await self._config.get_by_project(project_id) if not cfg: retention = _retention_for_plan(plan_slug) cfg = await self._config.create(project_id=project_id, retention_days=retention) return _config_response(cfg) async def update_config( self, project_id: UUID, body: UpdateLogConfigRequest, plan_slug: Optional[str] = None, ) -> LogConfigResponse: cfg = await self._config.get_by_project(project_id) if not cfg: cfg = await self._config.create( project_id=project_id, retention_days=_retention_for_plan(plan_slug), ) updates: dict = {} if body.debug_enabled is not None: updates["debug_enabled"] = body.debug_enabled if body.debug_enabled and body.debug_duration_minutes: duration = min(body.debug_duration_minutes, 120) updates["debug_expires_at"] = datetime.now(timezone.utc) + timedelta( minutes=duration ) elif not body.debug_enabled: updates["debug_expires_at"] = None if body.min_drain_level is not None: updates["min_drain_level"] = body.min_drain_level if updates: cfg = await self._config.update(cfg, **updates) await self._push_log_config(project_id, cfg) logger.info("log_config_updated", project_id=str(project_id)) return _config_response(cfg) # ------------------------------------------------------------------ # Log drains # ------------------------------------------------------------------ async def list_drains(self, project_id: UUID) -> list[LogDrainResponse]: rows = await self._drains.list_for_project(project_id) return [_drain_response(r) for r in rows] async def create_drain( self, project_id: UUID, body: CreateLogDrainRequest ) -> LogDrainResponse: existing = await self._drains.get_by_name(project_id, body.name) if existing: raise ConflictError(f"Log drain {body.name!r} already exists") count = len(await self._drains.list_for_project(project_id)) if count >= _MAX_DRAINS_PER_PROJECT: raise ConflictError(f"Maximum {_MAX_DRAINS_PER_PROJECT} log drains per project") row = await self._drains.create( project_id=project_id, name=body.name, destination_type=body.destination_type, endpoint_url=body.endpoint_url, auth_secret_enc=encrypt_value(body.auth_secret) if body.auth_secret else None, min_level=body.min_level, sources_json=body.sources, enabled=body.enabled, ) await self._push_drain_to_cloud(project_id, row) logger.info("log_drain_created", project_id=str(project_id), name=body.name) return _drain_response(row) async def update_drain( self, project_id: UUID, drain_id: UUID, body: UpdateLogDrainRequest, ) -> LogDrainResponse: row = await self._drains.get_by_id(drain_id) if not row or row.project_id != project_id: raise NotFoundError(f"Log drain {drain_id} not found") updates: dict = {} if body.endpoint_url is not None: updates["endpoint_url"] = body.endpoint_url if body.auth_secret is not None: updates["auth_secret_enc"] = encrypt_value(body.auth_secret) if body.min_level is not None: updates["min_level"] = body.min_level if body.sources is not None: updates["sources_json"] = body.sources if body.enabled is not None: updates["enabled"] = body.enabled if updates: row = await self._drains.update(row, **updates) await self._push_drain_to_cloud(project_id, row) return _drain_response(row) async def delete_drain(self, project_id: UUID, drain_id: UUID) -> None: row = await self._drains.get_by_id(drain_id) if not row or row.project_id != project_id: raise NotFoundError(f"Log drain {drain_id} not found") await self._drains.delete(row) await self._cloud.delete_log_drain(project_id=str(project_id), drain_id=str(drain_id)) logger.info("log_drain_deleted", project_id=str(project_id), drain_id=str(drain_id)) async def test_drain(self, project_id: UUID, drain_id: UUID) -> TestDrainResponse: row = await self._drains.get_by_id(drain_id) if not row or row.project_id != project_id: raise NotFoundError(f"Log drain {drain_id} not found") payload = { "test": True, "project_id": str(project_id), "message": "P2HS log drain test", "level": "info", "source": "gateway", } headers: dict[str, str] = {"Content-Type": "application/json"} if row.auth_secret_enc: secret = decrypt_value(row.auth_secret_enc) headers["Authorization"] = f"Bearer {secret}" try: async with httpx.AsyncClient(timeout=10) as client: resp = await client.post(row.endpoint_url, headers=headers, json=payload) return TestDrainResponse(ok=resp.status_code < 400, status_code=resp.status_code) except Exception as exc: return TestDrainResponse(ok=False, error=str(exc)) # ------------------------------------------------------------------ # Internal # ------------------------------------------------------------------ async def _push_log_config( self, project_id: UUID, cfg: ProjectLogConfig ) -> None: await self._cloud.push_log_config( project_id=str(project_id), debug_enabled=cfg.debug_enabled, debug_expires_at=cfg.debug_expires_at.isoformat() if cfg.debug_expires_at else None, min_drain_level=cfg.min_drain_level, retention_days=cfg.retention_days, ) async def _push_drain_to_cloud( self, project_id: UUID, row: ProjectLogDrain ) -> None: secret_ref = f"log_drain:{row.id}" if row.auth_secret_enc else None await self._cloud.upsert_log_drain( project_id=str(project_id), drain_id=str(row.id), destination_type=row.destination_type, endpoint_url=row.endpoint_url, secret_ref=secret_ref, min_level=row.min_level, sources=row.sources_json, enabled=row.enabled, ) # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _retention_for_plan(plan_slug: Optional[str]) -> int: if not plan_slug: return 1 slug = plan_slug.lower().split("_")[0] return _RETENTION_BY_PLAN.get(slug, 1) def _config_response(cfg: ProjectLogConfig) -> LogConfigResponse: return LogConfigResponse( project_id=cfg.project_id, debug_enabled=cfg.debug_enabled, debug_expires_at=cfg.debug_expires_at, min_drain_level=cfg.min_drain_level, retention_days=cfg.retention_days, updated_at=cfg.updated_at, ) def _drain_response(row: ProjectLogDrain) -> LogDrainResponse: return LogDrainResponse( id=row.id, project_id=row.project_id, name=row.name, destination_type=row.destination_type, endpoint_url=row.endpoint_url, min_level=row.min_level, sources=row.sources_json, enabled=row.enabled, created_at=row.created_at, updated_at=row.updated_at, )