"""Worker Fleet domain — business logic.""" from __future__ import annotations import hashlib import hmac from collections import defaultdict from datetime import datetime, timezone from typing import Optional from uuid import UUID import httpx import structlog from src.config import get_settings from src.core.exceptions import ConflictError, NotFoundError from src.core.security.encryption import decrypt_value, encrypt_value from src.domain.fleet.models import WorkerFleetRegistry, WorkerScalingConfig from src.domain.fleet.repository import ( WorkerFleetEventRepository, WorkerFleetRegistryRepository, WorkerScalingConfigRepository, ) from src.domain.fleet.schemas import ( DeployWorkerRequest, DeploymentCompleteRequest, FleetEventResponse, FleetSummaryResponse, RegisterWorkerRequest, ScaleWorkerRequest, ScalingConfigResponse, UpdateScalingConfigRequest, WorkerGroupResponse, WorkerInstanceResponse, ) logger = structlog.get_logger() _CF_API = "https://api.cloudflare.com/client/v4" class FleetService: def __init__( self, *, registry: WorkerFleetRegistryRepository, events: WorkerFleetEventRepository, config: WorkerScalingConfigRepository, ) -> None: self._registry = registry self._events = events self._config = config # ------------------------------------------------------------------ # Fleet overview # ------------------------------------------------------------------ async def get_fleet(self) -> FleetSummaryResponse: rows = await self._registry.list_all() groups: dict[str, list[WorkerFleetRegistry]] = defaultdict(list) for r in rows: groups[r.worker_name].append(r) worker_groups = [] for name, instances in groups.items(): healthy = [i for i in instances if i.last_health_check_ok is True] worker_groups.append( WorkerGroupResponse( worker_name=name, worker_type=instances[0].worker_type, all_healthy=len(healthy) == len(instances), healthy_count=len(healthy), total_count=len(instances), instances=[_instance_response(i) for i in instances], ) ) return FleetSummaryResponse(workers=worker_groups) async def get_worker(self, worker_name: str) -> WorkerGroupResponse: instances = await self._registry.list_for_worker(worker_name) if not instances: raise NotFoundError(f"Worker {worker_name!r} not registered in fleet") healthy = [i for i in instances if i.last_health_check_ok is True] return WorkerGroupResponse( worker_name=worker_name, worker_type=instances[0].worker_type, all_healthy=len(healthy) == len(instances), healthy_count=len(healthy), total_count=len(instances), instances=[_instance_response(i) for i in instances], ) # ------------------------------------------------------------------ # Worker registration (called by admin / CI setup) # ------------------------------------------------------------------ async def register_worker(self, body: RegisterWorkerRequest) -> WorkerInstanceResponse: existing = await self._registry.get_instance(body.worker_name, body.instance_id) if existing: raise ConflictError( f"Worker {body.worker_name!r} instance {body.instance_id!r} already registered" ) row = await self._registry.create( worker_name=body.worker_name, instance_id=body.instance_id, worker_type=body.worker_type, cf_script_id=body.cf_script_id, cf_queue_consumer_id=body.cf_queue_consumer_id, cf_queue_id=body.cf_queue_id, max_concurrency=body.max_concurrency, min_concurrency=body.min_concurrency, scale_up_threshold=body.scale_up_threshold, scale_down_threshold=body.scale_down_threshold, health_url=body.health_url, auto_deploy_enabled=body.auto_deploy_enabled, ) logger.info( "fleet_worker_registered", worker=body.worker_name, instance=body.instance_id, ) return _instance_response(row) # ------------------------------------------------------------------ # Manual deploy trigger # ------------------------------------------------------------------ async def trigger_deploy( self, worker_name: str, body: DeployWorkerRequest ) -> WorkerInstanceResponse: row = await self._registry.get_instance(worker_name, body.instance_id) if not row: raise NotFoundError(f"Worker {worker_name!r}/{body.instance_id!r} not found") cfg = await self._config.get() if not cfg or not cfg.deployment_webhook_url: raise ConflictError("Deployment webhook not configured") row = await self._registry.update( row, target_version=body.target_version, status="deploying", updated_at=datetime.now(timezone.utc), ) await self._fire_deployment_webhook(cfg, worker_name, body.instance_id, body.target_version) await self._events.create( worker_name=worker_name, instance_id=body.instance_id, event_type="deploy_triggered", triggered_by="admin", from_value=row.current_version, to_value=body.target_version, ) logger.info( "fleet_deploy_triggered", worker=worker_name, version=body.target_version, ) return _instance_response(row) # ------------------------------------------------------------------ # Manual scale override # ------------------------------------------------------------------ async def scale_worker( self, worker_name: str, body: ScaleWorkerRequest ) -> WorkerInstanceResponse: row = await self._registry.get_instance(worker_name, body.instance_id) if not row: raise NotFoundError(f"Worker {worker_name!r}/{body.instance_id!r} not found") if row.worker_type != "cf_queue_consumer": raise ConflictError("Concurrency scaling only applies to CF Queue consumers") old = row.current_concurrency await _set_cf_consumer_concurrency(row, body.concurrency) row = await self._registry.update( row, current_concurrency=body.concurrency, updated_at=datetime.now(timezone.utc), ) await self._events.create( worker_name=worker_name, instance_id=body.instance_id, event_type="scaled_up" if (old or 0) < body.concurrency else "scaled_down", triggered_by="admin", from_value=str(old), to_value=str(body.concurrency), ) logger.info( "fleet_manual_scale", worker=worker_name, from_concurrency=old, to_concurrency=body.concurrency, ) return _instance_response(row) # ------------------------------------------------------------------ # Deployment-complete callback (CI → Core) # ------------------------------------------------------------------ async def record_deployment_complete(self, body: DeploymentCompleteRequest) -> None: row = await self._registry.get_instance(body.worker_name, body.instance_id) if not row: logger.warning( "fleet_deploy_complete_unknown", worker=body.worker_name, instance=body.instance_id, ) return now = body.deployed_at or datetime.now(timezone.utc) if body.status == "success": await self._registry.update( row, current_version=body.version, status="active", last_deployed_at=now, updated_at=now, ) await self._events.create( worker_name=body.worker_name, instance_id=body.instance_id, event_type="deployed", triggered_by="ci_webhook", from_value=row.current_version, to_value=body.version, ) else: await self._registry.update(row, status="degraded", updated_at=now) await self._events.create( worker_name=body.worker_name, instance_id=body.instance_id, event_type="deploy_failed", triggered_by="ci_webhook", to_value=body.version, metadata={"error": body.error}, ) logger.info( "fleet_deploy_complete", worker=body.worker_name, version=body.version, status=body.status, ) # ------------------------------------------------------------------ # Events # ------------------------------------------------------------------ async def list_events( self, worker_name: str, limit: int = 50 ) -> list[FleetEventResponse]: rows = await self._events.list_for_worker(worker_name, limit=limit) return [_event_response(r) for r in rows] # ------------------------------------------------------------------ # Scaling config # ------------------------------------------------------------------ async def get_scaling_config(self) -> ScalingConfigResponse: cfg = await self._config.get_or_create() return _scaling_config_response(cfg) async def update_scaling_config( self, body: UpdateScalingConfigRequest ) -> ScalingConfigResponse: cfg = await self._config.get_or_create() updates: dict = {} if body.monitor_interval_seconds is not None: updates["monitor_interval_seconds"] = body.monitor_interval_seconds if body.auto_scale_enabled is not None: updates["auto_scale_enabled"] = body.auto_scale_enabled if body.deployment_webhook_url is not None: updates["deployment_webhook_url"] = body.deployment_webhook_url if body.deployment_webhook_secret is not None: updates["deployment_webhook_secret_enc"] = encrypt_value(body.deployment_webhook_secret) if updates: cfg = await self._config.update(cfg, **updates) return _scaling_config_response(cfg) # ------------------------------------------------------------------ # Internal # ------------------------------------------------------------------ async def _fire_deployment_webhook( self, cfg: WorkerScalingConfig, worker_name: str, instance_id: str, version: str, ) -> None: url = cfg.deployment_webhook_url if not url: return payload = {"worker": worker_name, "instance_id": instance_id, "version": version} headers: dict[str, str] = {"Content-Type": "application/json"} if cfg.deployment_webhook_secret_enc: secret = decrypt_value(cfg.deployment_webhook_secret_enc) sig = hmac.new(secret.encode(), str(payload).encode(), hashlib.sha256).hexdigest() headers["X-P2HS-Signature"] = f"sha256={sig}" async with httpx.AsyncClient(timeout=10) as client: await client.post(url, headers=headers, json=payload) # --------------------------------------------------------------------------- # CF API helpers # --------------------------------------------------------------------------- def _cf_headers() -> dict[str, str]: return { "Authorization": f"Bearer {get_settings().cloudflare_api_token}", "Content-Type": "application/json", } async def _set_cf_consumer_concurrency( row: WorkerFleetRegistry, concurrency: int ) -> None: if not row.cf_queue_consumer_id or not row.cf_queue_id: logger.warning( "fleet_scale_skip", reason="missing_cf_ids", worker=row.worker_name, ) return account_id = get_settings().cloudflare_account_id async with httpx.AsyncClient(timeout=10) as client: resp = await client.put( f"{_CF_API}/accounts/{account_id}/queues/{row.cf_queue_id}" f"/consumers/{row.cf_queue_consumer_id}", headers=_cf_headers(), json={"settings": {"max_concurrency": concurrency}}, ) resp.raise_for_status() logger.info( "cf_consumer_concurrency_set", worker=row.worker_name, concurrency=concurrency, ) async def get_cf_queue_depth(queue_id: str) -> int: account_id = get_settings().cloudflare_account_id async with httpx.AsyncClient(timeout=10) as client: resp = await client.get( f"{_CF_API}/accounts/{account_id}/queues/{queue_id}", headers=_cf_headers(), ) resp.raise_for_status() data = resp.json() return int(data.get("result", {}).get("consumers_total_messages", 0)) # --------------------------------------------------------------------------- # Response builders # --------------------------------------------------------------------------- def _instance_response(row: WorkerFleetRegistry) -> WorkerInstanceResponse: return WorkerInstanceResponse( id=row.id, worker_name=row.worker_name, instance_id=row.instance_id, worker_type=row.worker_type, current_version=row.current_version, target_version=row.target_version, status=row.status, cf_script_id=row.cf_script_id, cf_queue_consumer_id=row.cf_queue_consumer_id, cf_queue_id=row.cf_queue_id, current_concurrency=row.current_concurrency, max_concurrency=row.max_concurrency, min_concurrency=row.min_concurrency, scale_up_threshold=row.scale_up_threshold, scale_down_threshold=row.scale_down_threshold, auto_deploy_enabled=row.auto_deploy_enabled, health_url=row.health_url, last_health_check_at=row.last_health_check_at, last_health_check_ok=row.last_health_check_ok, last_deployed_at=row.last_deployed_at, updated_at=row.updated_at, ) def _event_response(row) -> FleetEventResponse: return FleetEventResponse( id=row.id, worker_name=row.worker_name, instance_id=row.instance_id, event_type=row.event_type, from_value=row.from_value, to_value=row.to_value, triggered_by=row.triggered_by, occurred_at=row.occurred_at, metadata_json=row.metadata_json, ) def _scaling_config_response(cfg: WorkerScalingConfig) -> ScalingConfigResponse: return ScalingConfigResponse( monitor_interval_seconds=cfg.monitor_interval_seconds, auto_scale_enabled=cfg.auto_scale_enabled, deployment_webhook_url=cfg.deployment_webhook_url, )