from __future__ import annotations import hashlib import json from uuid import UUID from fastapi import APIRouter, Depends, Response from redis import asyncio as aioredis from sqlalchemy.ext.asyncio import AsyncSession from src.core.database import get_db from src.core.dependencies import require_active_billing_for_project_id from src.core.exceptions import NotFoundError, ValidationError from src.dependencies import get_redis_client from src.domain.auth.models import User from src.domain.auth.router import get_current_user from src.domain.deployments.repository import ( DeploymentTargetRepository, DeployRepository, DomainRepository, ) from src.domain.deployments.service import DeployService from src.domain.integration_events.command_catalog import command_definition from src.domain.integration_events.repository import IntegrationEventsRepository from src.domain.integration_events.schemas import ( ControlPlaneCommandRequest, ControlPlaneCommandResponse, EventRegistryItem, IntegrationSubscriptionResponse, UpsertSubscriptionRequest, ) from src.domain.integration_events.service import IntegrationEventsService from src.domain.projects.repository import ProjectRepository from src.domain.projects.router import get_project_service from src.domain.projects.schemas import ProjectCreate, project_to_response from src.domain.revisions.repository import RevisionRepository from src.domain.workspaces.router import get_workspace_service from src.domain.workspaces.schemas import WorkspaceResponse from src.integrations.cloud_control_client import get_cloud_client_for_workspace from src.workers.enqueue import enqueue_run_project_deploy registry_router = APIRouter(prefix="/api/event-registry", tags=["integration-events"]) project_router = APIRouter( prefix="/api/projects/{project_id}/event-subscriptions", tags=["integration-events"], ) command_router = APIRouter(prefix="/api/integration-commands", tags=["integration-events"]) def _svc(db: AsyncSession, cloud) -> IntegrationEventsService: return IntegrationEventsService(repo=IntegrationEventsRepository(db), cloud=cloud) @command_router.post("", response_model=ControlPlaneCommandResponse) async def execute_core_command( body: ControlPlaneCommandRequest, user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), redis: aioredis.Redis = Depends(get_redis_client), ) -> ControlPlaneCommandResponse: """Execute a user-authorized Core control-plane command. This is intentionally separate from Cloud's integration command endpoint: lifecycle operations can mutate tenancy, billing, repositories, or deploys and therefore must remain behind Core's domain services and permissions. """ definition = command_definition(body.command) if not definition or definition["owner"] != "core": raise ValidationError("Command is not a Core-owned control-plane command") if definition["requires_user_principal"] and not user.id: raise ValidationError("A user principal is required") idem_digest = hashlib.sha256( f"{user.id}:{body.command}:{body.idempotency_key}".encode() ).hexdigest() idem_key = f"p2hs:integration-command:{idem_digest}" cached = await redis.get(idem_key) if cached: cached_value = cached.decode() if isinstance(cached, bytes) else cached return ControlPlaneCommandResponse.model_validate(json.loads(cached_value)) payload = body.input if body.command == "workspace.create": name = str(payload.get("name") or "").strip() plan_slug = str(payload.get("plan_slug") or "free").strip() if not name: raise ValidationError("name is required") ws = await get_workspace_service(db) created = await ws.create_workspace(name, user.id, plan_slug) result = WorkspaceResponse.model_validate(created).model_dump(mode="json") elif body.command == "workspace.member.invite": if body.workspace_id is None: raise ValidationError("workspace_id is required") email = str(payload.get("email") or "").strip() if not email: raise ValidationError("email is required") ws = await get_workspace_service(db) token, email_sent = await ws.invite( body.workspace_id, user.id, email, str(payload.get("role") or "editor"), ) result = {"token": token, "email": email, "email_sent": email_sent, "expires_in_days": 7} elif body.command == "project.create": if body.workspace_id is None: raise ValidationError("workspace_id is required") name = str(payload.get("name") or "").strip() if not name: raise ValidationError("name is required") project_input = ProjectCreate( name=name, type=payload.get("type"), description=payload.get("description"), intent=payload.get("intent"), preferred_stack=payload.get("preferred_stack"), stack_selection_mode=payload.get("stack_selection_mode") or "auto", discovery_requested=bool(payload.get("discovery_requested", False)), template_id=payload.get("template_id"), ) project_service = await get_project_service(db) project = await project_service.create_project( workspace_id=body.workspace_id, name=project_input.name, project_type=project_input.type.value if project_input.type else None, description=project_input.description, user_id=user.id, template_id=project_input.template_id, discovery_requested=project_input.discovery_requested, intent=project_input.intent, preferred_stack=( project_input.preferred_stack.value if project_input.preferred_stack else None ), stack_selection_mode=project_input.stack_selection_mode.value, ) result = project_to_response(project).model_dump(mode="json") elif body.command == "project.archive": if body.project_id is None: raise ValidationError("project_id is required") project = await ProjectRepository(db).get_by_id(body.project_id) if not project or (body.workspace_id and project.workspace_id != body.workspace_id): raise NotFoundError("Project not found in workspace") project_service = await get_project_service(db) await project_service.delete_project(body.project_id, user.id) result = {"project_id": str(body.project_id), "archived": True} elif body.command == "project.deploy": if body.project_id is None: raise ValidationError("project_id is required") project = await ProjectRepository(db).get_by_id(body.project_id) if not project or (body.workspace_id and project.workspace_id != body.workspace_id): raise NotFoundError("Project not found in workspace") from src.domain.workspaces.repository import MembershipRepository if not await MembershipRepository(db).has_permission( project.workspace_id, user.id, "deploy.publish" ): from src.core.exceptions import ForbiddenError raise ForbiddenError() revision = await RevisionRepository(db).get_latest(body.project_id) if not revision: raise ValidationError("Project has no revision to deploy") deploy_service = DeployService( deploy_repo=DeployRepository(db), domain_repo=DomainRepository(db), target_repo=DeploymentTargetRepository(db), project_repo=ProjectRepository(db), ) environment = str(payload.get("environment") or "production") if environment not in {"production", "staging", "preview"}: raise ValidationError("environment must be production, staging, or preview") deploy = await deploy_service.create_deploy( body.project_id, revision.id, environment, user.id, ) await enqueue_run_project_deploy(str(deploy.id)) result = { "deploy_id": str(deploy.id), "project_id": str(body.project_id), "revision_id": str(revision.id), } else: raise ValidationError("Unsupported Core command") response = ControlPlaneCommandResponse( command=body.command, command_id=body.command_id, result=result, ) await redis.setex( idem_key, 86400, json.dumps(response.model_dump(mode="json")), ) return response @registry_router.get("", response_model=list[EventRegistryItem]) async def list_platform_event_registry( _: User = Depends(get_current_user), ) -> list[EventRegistryItem]: from src.domain.integration_events.catalog import STANDARD_EVENTS from src.domain.integration_events.schemas import EventRegistryItem as Item return [Item.model_validate(r) for r in STANDARD_EVENTS] @project_router.get("", response_model=list[IntegrationSubscriptionResponse]) async def list_project_subscriptions( project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> list[IntegrationSubscriptionResponse]: project = await ProjectRepository(db).get_by_id(project_id) if not project: raise NotFoundError("Project not found") cloud = await get_cloud_client_for_workspace(project.workspace_id, db) return await _svc(db, cloud).list_subscriptions(project_id) @project_router.post("", response_model=IntegrationSubscriptionResponse, status_code=201) async def create_project_subscription( body: UpsertSubscriptionRequest, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> IntegrationSubscriptionResponse: project = await ProjectRepository(db).get_by_id(project_id) if not project: raise NotFoundError("Project not found") cloud = await get_cloud_client_for_workspace(project.workspace_id, db) return await _svc(db, cloud).create_subscription( workspace_id=project.workspace_id, project_id=project_id, body=body, ) @project_router.delete("/{subscription_id}", status_code=204, response_class=Response) async def delete_project_subscription( subscription_id: UUID, project_id: UUID = Depends(require_active_billing_for_project_id), _: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ) -> Response: project = await ProjectRepository(db).get_by_id(project_id) if not project: raise NotFoundError("Project not found") cloud = await get_cloud_client_for_workspace(project.workspace_id, db) await _svc(db, cloud).delete_subscription(project_id, subscription_id) return Response(status_code=204)