""" apply_service: programmatic invocation of the cluster apply pipeline. Used by: - routers/site_wizard.py (atomic create flow with apply_immediately=true) - routers/letsencrypt.py _complete_certificate (post-completion auto-apply) Design (Section 4.5 of v1.5.0 plan): - Reuses the existing /api/clusters/{id}/apply-changes route handler so that the response shape (`latest_version`, `consolidated_version_id`, `sync_results`, `applied_count`, `agents_notified`) and transaction boundary are byte-identical to UI-driven applies. - M23/M46 (R65): when user_id is None (e.g. ACME completion auto-apply with legacy created_by NULL), falls back to a short-lived JWT minted for the first active admin user (`is_admin=TRUE AND is_active=TRUE`). - This file deliberately delegates rather than duplicating the ~800 LOC apply pipeline; that keeps drift impossible. The wrapper only injects auth. """ import logging from datetime import timedelta from typing import Any, Dict, Optional from database.connection import close_database_connection, get_database_connection from utils.auth import create_access_token logger = logging.getLogger(__name__) async def _resolve_user_id(user_id: Optional[int]) -> Optional[int]: """If user_id is provided AND still valid (active), return it; else return the first active admin user's id. Returns None if no admin user exists (extreme edge case). M46/R65: schema accuracy — `users.is_super_admin` does NOT exist; the correct columns are `is_admin` (BOOLEAN) and `is_active` (BOOLEAN). Bulgu #27 fix: when a wizard-staged ACME order's `created_by` user has been deleted/deactivated by the time the post-completion auto-apply fires (could be 24h+ later), do NOT mint a JWT for that ghost user — we'd just produce a 401 from get_current_user_from_token. Fall through to the admin fallback instead. """ conn = await get_database_connection() try: if user_id is not None: valid = await conn.fetchval( "SELECT id FROM users WHERE id = $1 AND is_active = TRUE", user_id, ) if valid: return valid logger.warning( "apply_service: requested user_id=%s no longer exists or is inactive — " "falling back to admin", user_id, ) admin_id = await conn.fetchval( """ SELECT id FROM users WHERE is_admin = TRUE AND is_active = TRUE ORDER BY id ASC LIMIT 1 """ ) if not admin_id: logger.error( "apply_service: no active admin user found (is_admin=TRUE AND is_active=TRUE)" ) return admin_id finally: await close_database_connection(conn) def _mint_internal_jwt(user_id: int) -> str: """Mint a short-lived JWT for an internal apply call. Token uses the standard claim shape (`sub`/`user_id`) accepted by auth_middleware.get_current_user_from_token. """ return create_access_token( {"sub": str(user_id), "user_id": user_id}, expires_delta=timedelta(minutes=5), ) async def apply_cluster_pending( cluster_id: int, *, user_id: Optional[int] = None, apply_request: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """Programmatic equivalent of POST /api/clusters/{cluster_id}/apply-changes. Returns the same response dict the HTTP endpoint returns: { "message": str, "applied_count": int, "latest_version": str, "consolidated_version_id": int, "sync_results": list, "agents_notified": int, ...optionally global_ssl_applied... } """ # Local import to avoid cluster.py <-> apply_service circular import at # module load (cluster.py uses apply_service indirectly via routers/__init__). from routers.cluster import apply_pending_changes # noqa: WPS433 (intentional) resolved = await _resolve_user_id(user_id) if resolved is None: raise RuntimeError( "apply_service.apply_cluster_pending: no admin user available for system context apply" ) auth_header = f"Bearer {_mint_internal_jwt(resolved)}" return await apply_pending_changes( cluster_id=cluster_id, apply_request=apply_request or {}, authorization=auth_header, )