diff --git a/.gitignore b/.gitignore index fb92aff..6e270a6 100644 --- a/.gitignore +++ b/.gitignore @@ -12,8 +12,8 @@ var/ sdist/ develop-eggs/ .installed.cfg -lib/ -lib64/ +/lib/ +/lib64/ wheels/ .pytest_cache/ .coverage diff --git a/backend/.env.example b/backend/.env.example index 866ee91..b9a4d52 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -3,10 +3,25 @@ APP_ENV=development DEBUG=true HOST=0.0.0.0 PORT=8000 + +# CORS +CORS_ORIGINS=["http://localhost:3000","http://localhost:3001"] + +# Ollama (local LLM) OLLAMA_BASE_URL=http://localhost:11434 QWEN_MODEL=qwen3.5:latest LLM_TEMPERATURE=0.2 LLM_TIMEOUT_SECONDS=90 + +# Anthropic (Terraform generation) +ANTHROPIC_API_KEY=sk-ant-... +ANTHROPIC_MODEL=claude-sonnet-4-20250514 + +# Deploy +DEPLOY_WORKSPACE_ROOT=/tmp/cloudforge +DEPLOY_DRY_RUN=true + +# Agents ENABLE_WEB_SEARCH=true MAX_CLARIFICATION_ROUNDS=3 MAX_RESEARCH_ROUNDS=3 diff --git a/backend/app/config.py b/backend/app/config.py index 0001c49..4c5a7e2 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -7,10 +7,25 @@ class Settings(BaseSettings): debug: bool = True host: str = "0.0.0.0" port: int = 8000 + + # CORS + cors_origins: list[str] = ["http://localhost:3000", "http://localhost:3001"] + + # LLM — Ollama (local) ollama_base_url: str = "http://localhost:11434" qwen_model: str = "qwen3.5:latest" llm_temperature: float = 0.2 llm_timeout_seconds: int = 90 + + # LLM — Anthropic (for Terraform generation) + anthropic_api_key: str = "" + anthropic_model: str = "claude-sonnet-4-20250514" + + # Deploy settings + deploy_workspace_root: str = "/tmp/cloudforge" + deploy_dry_run: bool = True # When True, simulates terraform commands + + # Agents enable_web_search: bool = True max_clarification_rounds: int = 6 max_research_rounds: int = 3 diff --git a/backend/app/main.py b/backend/app/main.py index 2b7f544..6c0f9e4 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,14 +1,25 @@ from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware from app.config import settings -from app.routers import health, workflows +from app.routers import health, workflows, deploy app = FastAPI( title=settings.app_name, debug=settings.debug, ) +# CORS — allow the Next.js frontend to reach the API +app.add_middleware( + CORSMiddleware, + allow_origins=settings.cors_origins, + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + app.include_router(health.router) app.include_router(workflows.router) +app.include_router(deploy.router) @app.get("/") diff --git a/backend/app/routers/deploy.py b/backend/app/routers/deploy.py new file mode 100644 index 0000000..044bd69 --- /dev/null +++ b/backend/app/routers/deploy.py @@ -0,0 +1,167 @@ +""" +Deploy router — SSE streaming deployment endpoints. + +Endpoints: + POST /deploy/start → Start a new deployment, returns deployment_id + GET /deploy/{id}/stream → SSE stream of deployment events + GET /deploy/{id}/status → Current deployment status (JSON) + POST /deploy/{id}/rollback → Rollback a deployment + POST /deploy/{id}/cancel → Cancel a running deployment + GET /deploy/list → List all deployments +""" + +from __future__ import annotations + +import logging + +from fastapi import APIRouter, HTTPException +from fastapi.responses import StreamingResponse + +from app.schemas.deploy import ( + DeployListItem, + DeployStartResponse, + DeployStatusResponse, + RollbackRequest, + RollbackResponse, + StartDeployRequest, +) +from app.services.deploy_orchestrator import DeploymentOrchestrator + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/deploy", tags=["deploy"]) + +# Shared orchestrator instance +_orchestrator = DeploymentOrchestrator() + + +@router.post("/start", response_model=DeployStartResponse) +async def start_deployment(payload: StartDeployRequest) -> DeployStartResponse: + """ + Start a new infrastructure deployment. + + Accepts the architecture data (nodes + edges) from the forge pipeline + and kicks off the Terraform generation → init → plan → apply pipeline. + """ + # Convert Pydantic models to dicts for the orchestrator + arch_data = { + "nodes": [node.model_dump(by_alias=True) for node in payload.architecture_data.nodes], + "edges": [edge.model_dump(by_alias=True) for edge in payload.architecture_data.edges], + } + + deployment_id = await _orchestrator.start_deployment( + architecture_data=arch_data, + project_name=payload.project_name, + region=payload.region, + environment=payload.environment, + aws_credentials=payload.aws_credentials, + ) + + return DeployStartResponse( + deployment_id=deployment_id, + status="accepted", + message=f"Deployment {deployment_id} started — stream events at /deploy/{deployment_id}/stream", + ) + + +@router.get("/{deployment_id}/stream") +async def stream_deployment(deployment_id: str) -> StreamingResponse: + """ + SSE stream of deployment events. + + Connect via EventSource or fetch with streaming. + Events: log, node_status, stage_change, terraform_output, error, complete. + """ + deployment = await _orchestrator.get_deployment_status(deployment_id) + if not deployment: + raise HTTPException(status_code=404, detail="Deployment not found") + + async def event_generator(): + try: + async for event in _orchestrator.stream_events(deployment_id): + yield event.to_sse() + except Exception as e: + logger.error("SSE stream error for %s: %s", deployment_id, str(e)) + yield f"data: {{\"type\": \"error\", \"message\": \"{str(e)}\"}}\n\n" + + return StreamingResponse( + event_generator(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", + }, + ) + + +@router.get("/{deployment_id}/status", response_model=DeployStatusResponse) +async def get_deployment_status(deployment_id: str) -> DeployStatusResponse: + """Get the current status of a deployment.""" + deployment = await _orchestrator.get_deployment_status(deployment_id) + if not deployment: + raise HTTPException(status_code=404, detail="Deployment not found") + + return DeployStatusResponse( + deployment_id=deployment["deployment_id"], + project_name=deployment["project_name"], + status=deployment["status"].value if hasattr(deployment["status"], "value") else deployment["status"], + region=deployment["region"], + environment=deployment["environment"], + node_statuses=deployment.get("node_statuses", {}), + outputs=deployment.get("outputs", {}), + is_rollback=deployment.get("is_rollback", False), + created_at=deployment["created_at"], + updated_at=deployment["updated_at"], + ) + + +@router.post("/{deployment_id}/rollback", response_model=RollbackResponse) +async def rollback_deployment( + deployment_id: str, payload: RollbackRequest +) -> RollbackResponse: + """Rollback a completed or failed deployment using terraform destroy.""" + if not payload.confirm: + raise HTTPException( + status_code=400, detail="Rollback not confirmed — set confirm=true" + ) + + try: + rollback_id = await _orchestrator.rollback_deployment(deployment_id) + return RollbackResponse( + rollback_id=rollback_id, + status="accepted", + message=f"Rollback {rollback_id} started for deployment {deployment_id}", + ) + except ValueError as e: + raise HTTPException(status_code=404, detail=str(e)) + + +@router.post("/{deployment_id}/cancel") +async def cancel_deployment(deployment_id: str) -> dict: + """Cancel a running deployment.""" + cancelled = await _orchestrator.cancel_deployment(deployment_id) + if not cancelled: + raise HTTPException( + status_code=409, + detail="Deployment is not running or already completed", + ) + return {"deployment_id": deployment_id, "status": "cancelled"} + + +@router.get("/list", response_model=list[DeployListItem]) +async def list_deployments() -> list[DeployListItem]: + """List all deployments.""" + deployments = await _orchestrator.state_manager.list_deployments() + return [ + DeployListItem( + deployment_id=d["deployment_id"], + project_name=d["project_name"], + status=d["status"].value if hasattr(d["status"], "value") else d["status"], + region=d["region"], + environment=d["environment"], + is_rollback=d.get("is_rollback", False), + created_at=d["created_at"], + ) + for d in deployments + ] diff --git a/backend/app/schemas/deploy.py b/backend/app/schemas/deploy.py new file mode 100644 index 0000000..1a1945d --- /dev/null +++ b/backend/app/schemas/deploy.py @@ -0,0 +1,98 @@ +"""Pydantic schemas for the deploy API.""" + +from __future__ import annotations + +from typing import Any, Literal + +from pydantic import BaseModel, Field + + +# ── Request schemas ────────────────────────────────────────────────── + +class ArchNodeSchema(BaseModel): + """A single node from the CloudForge architecture.""" + id: str + label: str + sublabel: str = "" + type: Literal["compute", "storage", "cache", "gateway", "queue", "auth"] + x: float = 0 + y: float = 0 + terraformResource: str = "" + estimatedCost: str = "" + config: dict[str, str] = Field(default_factory=dict) + whyChosen: str = "" + validates: list[str] = Field(default_factory=list) + blocks: list[str] = Field(default_factory=list) + deployStatus: str = "queued" + + +class ArchEdgeSchema(BaseModel): + """An edge between two architecture nodes.""" + source: str = Field(alias="from", default="") + target: str = Field(alias="to", default="") + + model_config = {"populate_by_name": True} + + +class ArchitectureDataSchema(BaseModel): + """Full architecture spec from the frontend.""" + nodes: list[ArchNodeSchema] = Field(default_factory=list) + edges: list[ArchEdgeSchema] = Field(default_factory=list) + + +class StartDeployRequest(BaseModel): + """Request body for POST /deploy/start.""" + architecture_data: ArchitectureDataSchema + project_name: str = "cloudforge-project" + region: str = "us-east-1" + environment: str = "prod" + aws_credentials: dict[str, str] | None = Field( + default=None, + description="Optional AWS credentials (access_key_id, secret_access_key, session_token)", + ) + + +class RollbackRequest(BaseModel): + """Request body for POST /deploy/{deployment_id}/rollback.""" + confirm: bool = True + + +# ── Response schemas ───────────────────────────────────────────────── + +class DeployStartResponse(BaseModel): + """Response for POST /deploy/start.""" + deployment_id: str + status: str = "accepted" + message: str = "Deployment started" + + +class DeployStatusResponse(BaseModel): + """Response for GET /deploy/{deployment_id}/status.""" + deployment_id: str + project_name: str + status: str + region: str + environment: str + node_statuses: dict[str, str] = Field(default_factory=dict) + outputs: dict[str, Any] = Field(default_factory=dict) + is_rollback: bool = False + created_at: str + updated_at: str + + +class DeployListItem(BaseModel): + """Summary item for GET /deploy/list.""" + deployment_id: str + project_name: str + status: str + region: str + environment: str + is_rollback: bool = False + created_at: str + + +class RollbackResponse(BaseModel): + """Response for POST /deploy/{deployment_id}/rollback.""" + rollback_id: str + status: str = "accepted" + message: str = "Rollback started" diff --git a/backend/app/services/deploy_orchestrator.py b/backend/app/services/deploy_orchestrator.py new file mode 100644 index 0000000..f1b386f --- /dev/null +++ b/backend/app/services/deploy_orchestrator.py @@ -0,0 +1,558 @@ +""" +Deployment orchestrator service. + +Manages the full lifecycle of an infrastructure deployment: +1. Generate Terraform from architecture spec +2. Write files to a workspace directory +3. Run terraform init → plan → apply +4. Stream progress events back to the caller +5. Track resource provisioning status +""" + +import asyncio +import json +import logging +import os +import shutil +import subprocess +import tempfile +import uuid +from datetime import datetime, timezone +from enum import Enum +from pathlib import Path +from typing import Any, AsyncGenerator + +from app.services.terraform_generator import TerraformGenerator +from app.services.deploy_state import DeploymentStateManager, DeploymentStatus + +logger = logging.getLogger(__name__) + + +class DeployEventType(str, Enum): + """Types of events emitted during deployment.""" + LOG = "log" + NODE_STATUS = "node_status" + STAGE_CHANGE = "stage_change" + TERRAFORM_OUTPUT = "terraform_output" + ERROR = "error" + COMPLETE = "complete" + + +class DeployEvent: + """A single event in the deployment stream.""" + + def __init__( + self, + event_type: DeployEventType, + message: str, + data: dict[str, Any] | None = None, + ): + self.event_type = event_type + self.message = message + self.data = data or {} + self.timestamp = datetime.now(timezone.utc).isoformat() + + def to_sse(self) -> str: + """Format as Server-Sent Event.""" + payload = { + "type": self.event_type.value, + "message": self.message, + "data": self.data, + "timestamp": self.timestamp, + } + return f"data: {json.dumps(payload)}\n\n" + + +class DeploymentOrchestrator: + """ + Orchestrates the full deployment pipeline. + + Usage: + orchestrator = DeploymentOrchestrator() + deployment_id = await orchestrator.start_deployment(arch_data, project_name) + async for event in orchestrator.stream_events(deployment_id): + # Send to client via SSE + yield event.to_sse() + """ + + def __init__(self): + self.terraform_generator = TerraformGenerator() + self.state_manager = DeploymentStateManager() + self._active_deployments: dict[str, asyncio.Task] = {} + + async def start_deployment( + self, + architecture_data: dict[str, Any], + project_name: str = "cloudforge-project", + region: str = "us-east-1", + environment: str = "prod", + aws_credentials: dict[str, str] | None = None, + ) -> str: + """ + Initialize and start a new deployment. + + Returns the deployment_id for tracking. + """ + deployment_id = f"dep_{uuid.uuid4().hex[:12]}" + + # Create deployment record + await self.state_manager.create_deployment( + deployment_id=deployment_id, + project_name=project_name, + region=region, + environment=environment, + architecture_data=architecture_data, + ) + + # Extract node IDs for status tracking + nodes = architecture_data.get("nodes", []) + for node in nodes: + await self.state_manager.set_node_status( + deployment_id, node["id"], "queued" + ) + + # Start the deployment pipeline as a background task + task = asyncio.create_task( + self._run_pipeline( + deployment_id, + architecture_data, + project_name, + region, + environment, + aws_credentials, + ) + ) + self._active_deployments[deployment_id] = task + + return deployment_id + + async def stream_events( + self, deployment_id: str + ) -> AsyncGenerator[DeployEvent, None]: + """Stream deployment events as they occur.""" + async for event in self.state_manager.subscribe(deployment_id): + yield event + + async def get_deployment_status(self, deployment_id: str) -> dict[str, Any]: + """Get current deployment status.""" + return await self.state_manager.get_deployment(deployment_id) + + async def cancel_deployment(self, deployment_id: str) -> bool: + """Cancel a running deployment.""" + task = self._active_deployments.get(deployment_id) + if task and not task.done(): + task.cancel() + await self.state_manager.update_status( + deployment_id, DeploymentStatus.CANCELLED + ) + await self.state_manager.emit( + deployment_id, + DeployEvent( + DeployEventType.LOG, + "Deployment cancelled by user", + ), + ) + return True + return False + + async def rollback_deployment(self, deployment_id: str) -> str: + """ + Rollback a completed or failed deployment using terraform destroy. + Returns a new deployment_id for the rollback operation. + """ + deployment = await self.state_manager.get_deployment(deployment_id) + if not deployment: + raise ValueError(f"Deployment {deployment_id} not found") + + rollback_id = f"rb_{uuid.uuid4().hex[:12]}" + workspace = deployment.get("workspace_dir") + + if not workspace or not Path(workspace).exists(): + raise ValueError("Workspace not found — cannot rollback") + + await self.state_manager.create_deployment( + deployment_id=rollback_id, + project_name=deployment["project_name"], + region=deployment["region"], + environment=deployment["environment"], + architecture_data=deployment["architecture_data"], + is_rollback=True, + parent_deployment_id=deployment_id, + ) + + task = asyncio.create_task( + self._run_rollback(rollback_id, workspace, deployment_id) + ) + self._active_deployments[rollback_id] = task + + return rollback_id + + # ── Pipeline stages ─────────────────────────────────────────────── + + async def _run_pipeline( + self, + deployment_id: str, + architecture_data: dict[str, Any], + project_name: str, + region: str, + environment: str, + aws_credentials: dict[str, str] | None, + ) -> None: + """Execute the full deployment pipeline.""" + workspace = None + try: + # Stage 1: Generate Terraform + await self._emit(deployment_id, DeployEventType.STAGE_CHANGE, "Generating Terraform", {"stage": "generate"}) + await self._emit(deployment_id, DeployEventType.LOG, "Generating Terraform HCL from architecture spec...") + await self.state_manager.update_status(deployment_id, DeploymentStatus.GENERATING) + + tf_result = await self.terraform_generator.generate( + architecture_data, project_name, region, environment + ) + + if not tf_result.get("files"): + await self._emit(deployment_id, DeployEventType.ERROR, "No Terraform files generated") + await self.state_manager.update_status(deployment_id, DeploymentStatus.FAILED) + return + + for warning in tf_result.get("warnings", []): + await self._emit(deployment_id, DeployEventType.LOG, f"Warning: {warning}") + + await self._emit( + deployment_id, + DeployEventType.LOG, + f"Generated {len(tf_result['files'])} Terraform files ({tf_result.get('estimated_resources', 0)} resources)", + ) + + # Stage 2: Write to workspace + await self._emit(deployment_id, DeployEventType.STAGE_CHANGE, "Preparing workspace", {"stage": "workspace"}) + workspace = self._create_workspace(deployment_id, tf_result["files"]) + await self.state_manager.set_workspace(deployment_id, workspace) + await self._emit(deployment_id, DeployEventType.LOG, f"Workspace ready: {Path(workspace).name}") + + # Store generated terraform for reference + await self.state_manager.store_terraform_files(deployment_id, tf_result["files"]) + + # Stage 3: Terraform init + await self._emit(deployment_id, DeployEventType.STAGE_CHANGE, "Initializing Terraform", {"stage": "init"}) + await self.state_manager.update_status(deployment_id, DeploymentStatus.INITIALIZING) + await self._emit(deployment_id, DeployEventType.LOG, "Running terraform init...") + + init_ok = await self._run_terraform_command( + deployment_id, workspace, ["init", "-no-color", "-input=false"], + aws_credentials, + ) + if not init_ok: + await self.state_manager.update_status(deployment_id, DeploymentStatus.FAILED) + return + + await self._emit(deployment_id, DeployEventType.LOG, "Terraform initialized successfully") + + # Stage 4: Terraform plan + await self._emit(deployment_id, DeployEventType.STAGE_CHANGE, "Planning infrastructure", {"stage": "plan"}) + await self.state_manager.update_status(deployment_id, DeploymentStatus.PLANNING) + await self._emit(deployment_id, DeployEventType.LOG, "Running terraform plan...") + + plan_ok = await self._run_terraform_command( + deployment_id, + workspace, + ["plan", "-no-color", "-input=false", "-out=tfplan"], + aws_credentials, + ) + if not plan_ok: + await self.state_manager.update_status(deployment_id, DeploymentStatus.FAILED) + return + + await self._emit(deployment_id, DeployEventType.LOG, "Terraform plan complete — reviewing changes") + + # Stage 5: Terraform apply + await self._emit(deployment_id, DeployEventType.STAGE_CHANGE, "Provisioning infrastructure", {"stage": "apply"}) + await self.state_manager.update_status(deployment_id, DeploymentStatus.APPLYING) + + nodes = architecture_data.get("nodes", []) + + # Update node statuses to provisioning as apply starts + for node in nodes: + await self._emit( + deployment_id, + DeployEventType.NODE_STATUS, + f"Provisioning {node['label']}...", + {"nodeId": node["id"], "status": "provisioning"}, + ) + await self.state_manager.set_node_status( + deployment_id, node["id"], "provisioning" + ) + await self._emit( + deployment_id, + DeployEventType.LOG, + f"Provisioning {node['label']} ({node.get('terraformResource', 'unknown')})...", + ) + # Small delay between nodes for realistic streaming + await asyncio.sleep(0.3) + + apply_ok = await self._run_terraform_command( + deployment_id, + workspace, + ["apply", "-no-color", "-input=false", "-auto-approve", "tfplan"], + aws_credentials, + ) + + if apply_ok: + # Mark all nodes as live + for node in nodes: + await self._emit( + deployment_id, + DeployEventType.NODE_STATUS, + f"{node['label']} is live", + {"nodeId": node["id"], "status": "live"}, + ) + await self.state_manager.set_node_status( + deployment_id, node["id"], "live" + ) + await asyncio.sleep(0.2) + + # Get outputs + outputs = await self._get_terraform_outputs(workspace, aws_credentials) + await self.state_manager.store_outputs(deployment_id, outputs) + + await self._emit(deployment_id, DeployEventType.LOG, "All resources provisioned successfully") + await self._emit(deployment_id, DeployEventType.LOG, f"Terraform state stored in workspace") + await self.state_manager.update_status(deployment_id, DeploymentStatus.COMPLETE) + await self._emit( + deployment_id, + DeployEventType.COMPLETE, + "Deployment complete", + {"outputs": outputs}, + ) + else: + # Partial failure — check which resources were created + await self.state_manager.update_status(deployment_id, DeploymentStatus.FAILED) + await self._emit( + deployment_id, + DeployEventType.ERROR, + "Terraform apply failed. Some resources may have been created. Check logs and consider rollback.", + ) + + except asyncio.CancelledError: + logger.info("Deployment %s cancelled", deployment_id) + await self.state_manager.update_status(deployment_id, DeploymentStatus.CANCELLED) + raise + except Exception as e: + logger.exception("Deployment %s failed: %s", deployment_id, str(e)) + await self._emit(deployment_id, DeployEventType.ERROR, f"Deployment failed: {str(e)}") + await self.state_manager.update_status(deployment_id, DeploymentStatus.FAILED) + finally: + self._active_deployments.pop(deployment_id, None) + + async def _run_rollback( + self, rollback_id: str, workspace: str, original_deployment_id: str + ) -> None: + """Run terraform destroy for rollback.""" + try: + await self.state_manager.update_status(rollback_id, DeploymentStatus.APPLYING) + await self._emit(rollback_id, DeployEventType.LOG, f"Rolling back deployment {original_deployment_id}...") + await self._emit(rollback_id, DeployEventType.LOG, "Running terraform destroy...") + + ok = await self._run_terraform_command( + rollback_id, + workspace, + ["destroy", "-no-color", "-input=false", "-auto-approve"], + None, + ) + + if ok: + await self.state_manager.update_status(rollback_id, DeploymentStatus.COMPLETE) + await self.state_manager.update_status(original_deployment_id, DeploymentStatus.ROLLED_BACK) + await self._emit(rollback_id, DeployEventType.COMPLETE, "Rollback complete — all resources destroyed") + else: + await self.state_manager.update_status(rollback_id, DeploymentStatus.FAILED) + await self._emit(rollback_id, DeployEventType.ERROR, "Rollback failed — manual cleanup may be needed") + + except Exception as e: + logger.exception("Rollback %s failed: %s", rollback_id, str(e)) + await self._emit(rollback_id, DeployEventType.ERROR, f"Rollback error: {str(e)}") + await self.state_manager.update_status(rollback_id, DeploymentStatus.FAILED) + finally: + self._active_deployments.pop(rollback_id, None) + + # ── Helpers ─────────────────────────────────────────────────────── + + def _create_workspace(self, deployment_id: str, files: list[dict]) -> str: + """Create a temporary workspace directory and write Terraform files.""" + workspace = os.path.join(tempfile.gettempdir(), "cloudforge", deployment_id) + infra_dir = os.path.join(workspace, "infra") + os.makedirs(infra_dir, exist_ok=True) + + for file_info in files: + file_path = os.path.join(workspace, file_info["path"]) + os.makedirs(os.path.dirname(file_path), exist_ok=True) + with open(file_path, "w") as f: + f.write(file_info["content"]) + + return workspace + + async def _run_terraform_command( + self, + deployment_id: str, + workspace: str, + args: list[str], + aws_credentials: dict[str, str] | None, + ) -> bool: + """Run a terraform command and stream output.""" + infra_dir = os.path.join(workspace, "infra") + + env = os.environ.copy() + if aws_credentials: + env["AWS_ACCESS_KEY_ID"] = aws_credentials.get("access_key_id", "") + env["AWS_SECRET_ACCESS_KEY"] = aws_credentials.get("secret_access_key", "") + if aws_credentials.get("session_token"): + env["AWS_SESSION_TOKEN"] = aws_credentials["session_token"] + if aws_credentials.get("region"): + env["AWS_DEFAULT_REGION"] = aws_credentials["region"] + + # Check if terraform is available + terraform_bin = shutil.which("terraform") or shutil.which("tofu") + if not terraform_bin: + await self._emit( + deployment_id, + DeployEventType.LOG, + "Terraform binary not found — running in dry-run mode", + ) + # Simulate terraform execution for demo/development + return await self._simulate_terraform(deployment_id, args) + + cmd = [terraform_bin] + args + try: + process = await asyncio.create_subprocess_exec( + *cmd, + cwd=infra_dir, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT, + env=env, + ) + + async for line in process.stdout: + text = line.decode("utf-8", errors="replace").rstrip() + if text: + await self._emit( + deployment_id, + DeployEventType.TERRAFORM_OUTPUT, + text, + ) + + await process.wait() + return process.returncode == 0 + + except FileNotFoundError: + await self._emit(deployment_id, DeployEventType.ERROR, "Terraform command not found") + return False + except Exception as e: + await self._emit(deployment_id, DeployEventType.ERROR, f"Command error: {str(e)}") + return False + + async def _simulate_terraform(self, deployment_id: str, args: list[str]) -> bool: + """Simulate terraform commands for development/demo without real terraform.""" + command = args[0] if args else "unknown" + + simulations = { + "init": [ + "Initializing the backend...", + "Initializing provider plugins...", + "- Finding hashicorp/aws versions matching \"~> 5.0\"...", + "- Installing hashicorp/aws v5.82.2...", + "- Installed hashicorp/aws v5.82.2 (signed by HashiCorp)", + "Terraform has been successfully initialized!", + ], + "plan": [ + "Terraform used the selected providers to generate the following execution plan.", + "Resource actions are indicated with the following symbols:", + " + create", + "", + "Plan: 5 to add, 0 to change, 0 to destroy.", + ], + "apply": [ + "aws_apigatewayv2_api.apigw: Creating...", + "aws_apigatewayv2_api.apigw: Creation complete after 2s [id=abc123]", + "aws_lambda_function.lambda: Creating...", + "aws_lambda_function.lambda: Creation complete after 8s [id=auth-function]", + "aws_elasticache_cluster.redis: Creating...", + "aws_elasticache_cluster.redis: Still creating... [10s elapsed]", + "aws_elasticache_cluster.redis: Creation complete after 15s", + "aws_db_instance.rds: Creating...", + "aws_db_instance.rds: Still creating... [20s elapsed]", + "aws_db_instance.rds: Creation complete after 25s", + "aws_secretsmanager_secret.secrets: Creating...", + "aws_secretsmanager_secret.secrets: Creation complete after 1s", + "", + "Apply complete! Resources: 5 added, 0 changed, 0 destroyed.", + ], + "destroy": [ + "aws_apigatewayv2_api.apigw: Destroying...", + "aws_apigatewayv2_api.apigw: Destruction complete after 1s", + "aws_lambda_function.lambda: Destroying...", + "aws_lambda_function.lambda: Destruction complete after 3s", + "aws_elasticache_cluster.redis: Destroying...", + "aws_elasticache_cluster.redis: Destruction complete after 5s", + "aws_db_instance.rds: Destroying...", + "aws_db_instance.rds: Destruction complete after 10s", + "aws_secretsmanager_secret.secrets: Destroying...", + "aws_secretsmanager_secret.secrets: Destruction complete after 1s", + "", + "Destroy complete! Resources: 5 destroyed.", + ], + } + + lines = simulations.get(command, [f"Simulating terraform {command}..."]) + for line in lines: + await self._emit(deployment_id, DeployEventType.TERRAFORM_OUTPUT, line) + await asyncio.sleep(0.15 + (0.3 if "Creating" in line or "Destroying" in line else 0)) + + return True + + async def _get_terraform_outputs( + self, workspace: str, aws_credentials: dict[str, str] | None + ) -> dict[str, Any]: + """Retrieve terraform outputs after successful apply.""" + infra_dir = os.path.join(workspace, "infra") + terraform_bin = shutil.which("terraform") or shutil.which("tofu") + + if not terraform_bin: + # Return simulated outputs + return { + "api_endpoint": {"value": "https://abc123.execute-api.us-east-1.amazonaws.com"}, + "lambda_arn": {"value": "arn:aws:lambda:us-east-1:123456789:function:cloudforge-auth"}, + "rds_endpoint": {"value": "cloudforge-db.cluster-xyz.us-east-1.rds.amazonaws.com:5432"}, + "redis_endpoint": {"value": "cloudforge-cache.abc123.use1.cache.amazonaws.com:6379"}, + } + + env = os.environ.copy() + if aws_credentials: + env["AWS_ACCESS_KEY_ID"] = aws_credentials.get("access_key_id", "") + env["AWS_SECRET_ACCESS_KEY"] = aws_credentials.get("secret_access_key", "") + + try: + result = await asyncio.create_subprocess_exec( + terraform_bin, "output", "-json", "-no-color", + cwd=infra_dir, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + env=env, + ) + stdout, _ = await result.communicate() + if result.returncode == 0: + return json.loads(stdout.decode()) + except Exception as e: + logger.warning("Failed to get terraform outputs: %s", str(e)) + + return {} + + async def _emit( + self, + deployment_id: str, + event_type: DeployEventType, + message: str, + data: dict[str, Any] | None = None, + ) -> None: + """Emit a deployment event.""" + event = DeployEvent(event_type, message, data) + await self.state_manager.emit(deployment_id, event) diff --git a/backend/app/services/deploy_state.py b/backend/app/services/deploy_state.py new file mode 100644 index 0000000..05bada0 --- /dev/null +++ b/backend/app/services/deploy_state.py @@ -0,0 +1,225 @@ +""" +Deployment state manager. + +In-memory state tracking for deployments with async event streaming. +In production, swap with Redis or a database backend. +""" + +import asyncio +import logging +from datetime import datetime, timezone +from enum import Enum +from typing import Any, AsyncGenerator + +logger = logging.getLogger(__name__) + + +class DeploymentStatus(str, Enum): + """Deployment lifecycle states.""" + PENDING = "pending" + GENERATING = "generating" + INITIALIZING = "initializing" + PLANNING = "planning" + APPLYING = "applying" + COMPLETE = "complete" + FAILED = "failed" + CANCELLED = "cancelled" + ROLLED_BACK = "rolled_back" + + +class DeploymentStateManager: + """ + In-memory deployment state manager with event streaming. + + Each deployment has: + - Status and metadata + - Node-level provisioning status + - An asyncio.Queue for event streaming + - Generated terraform files + - Terraform outputs (after apply) + """ + + def __init__(self): + self._deployments: dict[str, dict[str, Any]] = {} + self._queues: dict[str, list[asyncio.Queue]] = {} + self._events: dict[str, list[Any]] = {} + + async def create_deployment( + self, + deployment_id: str, + project_name: str, + region: str, + environment: str, + architecture_data: dict[str, Any], + is_rollback: bool = False, + parent_deployment_id: str | None = None, + ) -> dict[str, Any]: + """Create a new deployment record.""" + deployment = { + "deployment_id": deployment_id, + "project_name": project_name, + "region": region, + "environment": environment, + "architecture_data": architecture_data, + "status": DeploymentStatus.PENDING, + "node_statuses": {}, + "workspace_dir": None, + "terraform_files": [], + "outputs": {}, + "is_rollback": is_rollback, + "parent_deployment_id": parent_deployment_id, + "created_at": datetime.now(timezone.utc).isoformat(), + "updated_at": datetime.now(timezone.utc).isoformat(), + } + self._deployments[deployment_id] = deployment + self._queues[deployment_id] = [] + self._events[deployment_id] = [] + return deployment + + async def get_deployment(self, deployment_id: str) -> dict[str, Any] | None: + """Get deployment by ID.""" + return self._deployments.get(deployment_id) + + async def list_deployments(self) -> list[dict[str, Any]]: + """List all deployments (summary view).""" + return [ + { + "deployment_id": d["deployment_id"], + "project_name": d["project_name"], + "status": d["status"], + "region": d["region"], + "environment": d["environment"], + "is_rollback": d["is_rollback"], + "created_at": d["created_at"], + } + for d in self._deployments.values() + ] + + async def update_status( + self, deployment_id: str, status: DeploymentStatus + ) -> None: + """Update deployment status.""" + if deployment_id in self._deployments: + self._deployments[deployment_id]["status"] = status + self._deployments[deployment_id]["updated_at"] = ( + datetime.now(timezone.utc).isoformat() + ) + + async def set_node_status( + self, deployment_id: str, node_id: str, status: str + ) -> None: + """Update a specific node's deployment status.""" + if deployment_id in self._deployments: + self._deployments[deployment_id]["node_statuses"][node_id] = status + + async def set_workspace(self, deployment_id: str, workspace_dir: str) -> None: + """Set the workspace directory for a deployment.""" + if deployment_id in self._deployments: + self._deployments[deployment_id]["workspace_dir"] = workspace_dir + + async def store_terraform_files( + self, deployment_id: str, files: list[dict] + ) -> None: + """Store generated Terraform files.""" + if deployment_id in self._deployments: + self._deployments[deployment_id]["terraform_files"] = files + + async def store_outputs( + self, deployment_id: str, outputs: dict[str, Any] + ) -> None: + """Store Terraform outputs after apply.""" + if deployment_id in self._deployments: + self._deployments[deployment_id]["outputs"] = outputs + + async def emit(self, deployment_id: str, event: Any) -> None: + """Emit an event to all subscribers of a deployment.""" + # Store event for replay + if deployment_id in self._events: + self._events[deployment_id].append(event) + + # Push to all active subscriber queues + queues = self._queues.get(deployment_id, []) + for queue in queues: + try: + queue.put_nowait(event) + except asyncio.QueueFull: + logger.warning( + "Event queue full for deployment %s, dropping event", + deployment_id, + ) + + async def subscribe( + self, deployment_id: str, replay: bool = True + ) -> AsyncGenerator[Any, None]: + """ + Subscribe to deployment events. + + If replay=True, first yields all past events, then streams new ones. + Yields a sentinel None when the deployment is terminal. + """ + queue: asyncio.Queue = asyncio.Queue(maxsize=1000) + + # Register subscriber + if deployment_id not in self._queues: + self._queues[deployment_id] = [] + self._queues[deployment_id].append(queue) + + try: + # Replay past events + if replay and deployment_id in self._events: + for event in self._events[deployment_id]: + yield event + + # Stream new events + while True: + try: + event = await asyncio.wait_for(queue.get(), timeout=30.0) + yield event + + # Check if this is a terminal event + if hasattr(event, "event_type"): + from app.services.deploy_orchestrator import DeployEventType + if event.event_type in ( + DeployEventType.COMPLETE, + DeployEventType.ERROR, + ): + # Check if the deployment is in a terminal state + deployment = self._deployments.get(deployment_id) + if deployment and deployment["status"] in ( + DeploymentStatus.COMPLETE, + DeploymentStatus.FAILED, + DeploymentStatus.CANCELLED, + DeploymentStatus.ROLLED_BACK, + ): + return + + except asyncio.TimeoutError: + # Send keepalive — check if deployment is still active + deployment = self._deployments.get(deployment_id) + if not deployment: + return + if deployment["status"] in ( + DeploymentStatus.COMPLETE, + DeploymentStatus.FAILED, + DeploymentStatus.CANCELLED, + DeploymentStatus.ROLLED_BACK, + ): + return + # Yield a keepalive comment (SSE comment) + continue + + finally: + # Unsubscribe + queues = self._queues.get(deployment_id, []) + if queue in queues: + queues.remove(queue) + + async def get_events(self, deployment_id: str) -> list[Any]: + """Get all stored events for a deployment.""" + return self._events.get(deployment_id, []) + + async def cleanup(self, deployment_id: str) -> None: + """Clean up deployment state (call after TTL or manual cleanup).""" + self._deployments.pop(deployment_id, None) + self._queues.pop(deployment_id, None) + self._events.pop(deployment_id, None) diff --git a/backend/app/services/terraform_generator.py b/backend/app/services/terraform_generator.py new file mode 100644 index 0000000..bfbc84e --- /dev/null +++ b/backend/app/services/terraform_generator.py @@ -0,0 +1,488 @@ +""" +Terraform HCL generation service. + +Takes CloudForge architecture data and generates production-ready Terraform code +using Claude (Anthropic) as the LLM backbone. +""" + +import json +import logging +from typing import Any + +from langchain_anthropic import ChatAnthropic +from langchain_core.messages import HumanMessage, SystemMessage + +from app.config import settings + +logger = logging.getLogger(__name__) + +# System prompt for Terraform generation +TERRAFORM_SYSTEM_PROMPT = """You are an expert AWS infrastructure engineer and Terraform specialist. +Given a CloudForge architecture specification, generate production-ready Terraform HCL code. + +Rules: +1. Use Terraform >= 1.6 with AWS provider ~> 5.0 +2. Use an S3 backend for state storage +3. Parameterize with variables (region, project_name, environment) +4. Add proper tags to all resources: Project, Environment, ManagedBy=CloudForge +5. Follow AWS security best practices: + - Least privilege IAM policies + - Encryption at rest and in transit + - VPC isolation where applicable + - Security groups with minimal ingress +6. Include outputs for resource ARNs and endpoints +7. Use consistent naming: ${var.project_name}-${var.environment}- +8. Add lifecycle rules where appropriate (prevent_destroy for databases) + +Output Format: +Return a JSON object with this structure: +{ + "files": [ + { + "name": "main.tf", + "path": "infra/main.tf", + "content": "... HCL content ..." + }, + ... + ], + "plan_summary": "Brief description of what will be created", + "estimated_resources": 5, + "warnings": ["any warnings about the config"] +} + +Generate separate files for: main.tf (provider/backend), variables.tf, outputs.tf, +and one file per major resource group (e.g., lambda.tf, rds.tf, etc.).""" + + +class TerraformGenerator: + """Generates Terraform HCL from CloudForge architecture specifications.""" + + def __init__(self): + self.llm = self._init_llm() + + def _init_llm(self) -> ChatAnthropic: + """Initialize the Claude LLM for Terraform generation.""" + return ChatAnthropic( + model=getattr(settings, "anthropic_model", "claude-sonnet-4-20250514"), + temperature=0.1, + max_tokens=8192, + anthropic_api_key=getattr(settings, "anthropic_api_key", None), + ) + + async def generate( + self, + architecture_data: dict[str, Any], + project_name: str = "cloudforge-project", + region: str = "us-east-1", + environment: str = "prod", + ) -> dict[str, Any]: + """ + Generate Terraform files from architecture specification. + + Args: + architecture_data: Dict with 'nodes' and 'edges' from the forge pipeline + project_name: Name of the project for resource naming + region: AWS region to deploy to + environment: Deployment environment (dev/staging/prod) + + Returns: + Dict with 'files', 'plan_summary', 'estimated_resources', 'warnings' + """ + nodes = architecture_data.get("nodes", []) + edges = architecture_data.get("edges", []) + + if not nodes: + return { + "files": [], + "plan_summary": "No resources to generate", + "estimated_resources": 0, + "warnings": ["No architecture nodes provided"], + } + + prompt = self._build_prompt(nodes, edges, project_name, region, environment) + + logger.info( + "Generating Terraform for %d nodes, %d edges", len(nodes), len(edges) + ) + + try: + response = await self.llm.ainvoke( + [ + SystemMessage(content=TERRAFORM_SYSTEM_PROMPT), + HumanMessage(content=prompt), + ] + ) + + result = self._parse_response(response.content) + logger.info( + "Generated %d Terraform files", len(result.get("files", [])) + ) + return result + + except Exception as e: + logger.error("Terraform generation failed: %s", str(e)) + # Fallback: generate basic Terraform from templates + return self._generate_fallback(nodes, edges, project_name, region, environment) + + def _build_prompt( + self, + nodes: list[dict], + edges: list[dict], + project_name: str, + region: str, + environment: str, + ) -> str: + """Build the user prompt with architecture details.""" + arch_spec = { + "project_name": project_name, + "region": region, + "environment": environment, + "resources": [], + "connections": [], + } + + for node in nodes: + arch_spec["resources"].append({ + "id": node.get("id"), + "name": node.get("label"), + "description": node.get("sublabel", ""), + "type": node.get("type"), + "terraform_resource": node.get("terraformResource"), + "estimated_cost": node.get("estimatedCost"), + "config": node.get("config", {}), + "requirements": node.get("validates", []), + }) + + for edge in edges: + arch_spec["connections"].append({ + "from": edge.get("from"), + "to": edge.get("to"), + }) + + return f"""Generate Terraform HCL for the following CloudForge architecture: + +```json +{json.dumps(arch_spec, indent=2)} +``` + +Requirements: +- {len(nodes)} AWS resources to provision +- All resources in {region} +- Project name: {project_name} +- Environment: {environment} +- Include proper IAM roles and security groups +- Wire service connections based on the edges (e.g., Lambda → RDS means Lambda needs DB endpoint env var) +""" + + def _parse_response(self, content: str) -> dict[str, Any]: + """Parse the LLM response into structured Terraform output.""" + if isinstance(content, list): + # Handle Anthropic's content block format + text = "" + for block in content: + if hasattr(block, "text"): + text += block.text + elif isinstance(block, dict) and "text" in block: + text += block["text"] + elif isinstance(block, str): + text += block + content = text + + # Try to extract JSON from the response + try: + # Look for JSON block in markdown + if "```json" in content: + json_str = content.split("```json")[1].split("```")[0].strip() + elif "```" in content: + json_str = content.split("```")[1].split("```")[0].strip() + else: + json_str = content.strip() + + result = json.loads(json_str) + + # Validate structure + if "files" not in result: + result = {"files": [], "plan_summary": "Parse error", "estimated_resources": 0, "warnings": ["Could not parse LLM response"]} + + return result + + except (json.JSONDecodeError, IndexError) as e: + logger.warning("Failed to parse LLM JSON response: %s", str(e)) + return { + "files": [], + "plan_summary": "Failed to parse response", + "estimated_resources": 0, + "warnings": [f"JSON parse error: {str(e)}"], + } + + def _generate_fallback( + self, + nodes: list[dict], + edges: list[dict], + project_name: str, + region: str, + environment: str, + ) -> dict[str, Any]: + """Generate basic Terraform from templates when LLM fails.""" + files = [] + + # main.tf + main_tf = f'''terraform {{ + required_version = ">= 1.6" + required_providers {{ + aws = {{ + source = "hashicorp/aws" + version = "~> 5.0" + }} + }} + backend "s3" {{ + bucket = "{project_name}-tf-state" + key = "{project_name}/terraform.tfstate" + region = "{region}" + }} +}} + +provider "aws" {{ + region = var.aws_region + + default_tags {{ + tags = {{ + Project = var.project_name + Environment = var.environment + ManagedBy = "CloudForge" + }} + }} +}} +''' + files.append({"name": "main.tf", "path": "infra/main.tf", "content": main_tf}) + + # variables.tf + variables_tf = f'''variable "project_name" {{ + type = string + default = "{project_name}" +}} + +variable "environment" {{ + type = string + default = "{environment}" +}} + +variable "aws_region" {{ + type = string + default = "{region}" +}} +''' + files.append({"name": "variables.tf", "path": "infra/variables.tf", "content": variables_tf}) + + # Generate resource files from nodes + resource_map = { + "aws_lambda_function": self._gen_lambda_tf, + "aws_db_instance": self._gen_rds_tf, + "aws_elasticache_cluster": self._gen_elasticache_tf, + "aws_apigatewayv2_api": self._gen_apigateway_tf, + "aws_secretsmanager_secret": self._gen_secrets_tf, + } + + outputs = [] + for node in nodes: + tf_resource = node.get("terraformResource", "") + generator = resource_map.get(tf_resource) + if generator: + file_info, output_lines = generator(node, project_name, environment) + files.append(file_info) + outputs.extend(output_lines) + + # outputs.tf + if outputs: + outputs_content = "\n\n".join(outputs) + files.append({"name": "outputs.tf", "path": "infra/outputs.tf", "content": outputs_content}) + + return { + "files": files, + "plan_summary": f"Fallback generation: {len(nodes)} resources across {len(files)} files", + "estimated_resources": len(nodes), + "warnings": ["Generated from templates (LLM unavailable). Review before applying."], + } + + @staticmethod + def _gen_lambda_tf(node: dict, project_name: str, environment: str) -> tuple[dict, list[str]]: + config = node.get("config", {}) + resource_name = node.get("id", "function") + content = f'''resource "aws_lambda_function" "{resource_name}" {{ + function_name = "${{var.project_name}}-${{var.environment}}-{node.get("label", "function").lower().replace(" ", "-")}" + runtime = "{config.get("runtime", "nodejs20.x")}" + handler = "index.handler" + memory_size = {config.get("memory", "512").replace(" MB", "")} + timeout = {config.get("timeout", "10").replace("s", "")} + architectures = ["{config.get("architecture", "arm64")}"] + + role = aws_iam_role.{resource_name}_role.arn + + environment {{ + variables = {{ + NODE_ENV = var.environment + }} + }} + + tags = {{ + Name = "${{var.project_name}}-${{var.environment}}-{resource_name}" + }} +}} + +resource "aws_iam_role" "{resource_name}_role" {{ + name = "${{var.project_name}}-${{var.environment}}-{resource_name}-role" + + assume_role_policy = jsonencode({{ + Version = "2012-10-17" + Statement = [{{ + Action = "sts:AssumeRole" + Effect = "Allow" + Principal = {{ + Service = "lambda.amazonaws.com" + }} + }}] + }}) +}} + +resource "aws_iam_role_policy_attachment" "{resource_name}_basic" {{ + role = aws_iam_role.{resource_name}_role.name + policy_arn = "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole" +}} +''' + outputs = [ + f'output "{resource_name}_arn" {{\n value = aws_lambda_function.{resource_name}.arn\n}}', + f'output "{resource_name}_function_name" {{\n value = aws_lambda_function.{resource_name}.function_name\n}}', + ] + return {"name": f"{resource_name}.tf", "path": f"infra/{resource_name}.tf", "content": content}, outputs + + @staticmethod + def _gen_rds_tf(node: dict, project_name: str, environment: str) -> tuple[dict, list[str]]: + config = node.get("config", {}) + resource_name = node.get("id", "database") + content = f'''resource "aws_db_instance" "{resource_name}" {{ + identifier = "${{var.project_name}}-${{var.environment}}-db" + engine = "{config.get("engine", "postgres")}" + engine_version = "{config.get("version", "15")}" + instance_class = "{config.get("instance", "db.t3.micro")}" + allocated_storage = {config.get("storage", "20").replace(" GB", "")} + + db_name = replace("${{var.project_name}}_${{var.environment}}", "-", "_") + username = var.db_username + password = var.db_password + + skip_final_snapshot = var.environment != "prod" + backup_retention_period = {config.get("backup_retention", "7").replace(" days", "")} + deletion_protection = var.environment == "prod" + storage_encrypted = true + + tags = {{ + Name = "${{var.project_name}}-${{var.environment}}-db" + }} + + lifecycle {{ + prevent_destroy = false + }} +}} + +variable "db_username" {{ + type = string + default = "cloudforge_admin" + sensitive = true +}} + +variable "db_password" {{ + type = string + sensitive = true +}} +''' + outputs = [ + f'output "{resource_name}_endpoint" {{\n value = aws_db_instance.{resource_name}.endpoint\n}}', + f'output "{resource_name}_address" {{\n value = aws_db_instance.{resource_name}.address\n}}', + ] + return {"name": f"{resource_name}.tf", "path": f"infra/{resource_name}.tf", "content": content}, outputs + + @staticmethod + def _gen_elasticache_tf(node: dict, project_name: str, environment: str) -> tuple[dict, list[str]]: + config = node.get("config", {}) + resource_name = node.get("id", "cache") + content = f'''resource "aws_elasticache_cluster" "{resource_name}" {{ + cluster_id = "${{var.project_name}}-${{var.environment}}-cache" + engine = "{config.get("engine", "redis")}" + engine_version = "{config.get("version", "7.0")}" + node_type = "{config.get("instance", "cache.t3.micro")}" + num_cache_nodes = 1 + port = 6379 + + parameter_group_name = "default.redis7" + + tags = {{ + Name = "${{var.project_name}}-${{var.environment}}-cache" + }} +}} +''' + outputs = [ + f'output "{resource_name}_endpoint" {{\n value = aws_elasticache_cluster.{resource_name}.cache_nodes[0].address\n}}', + ] + return {"name": f"{resource_name}.tf", "path": f"infra/{resource_name}.tf", "content": content}, outputs + + @staticmethod + def _gen_apigateway_tf(node: dict, project_name: str, environment: str) -> tuple[dict, list[str]]: + config = node.get("config", {}) + resource_name = node.get("id", "api") + content = f'''resource "aws_apigatewayv2_api" "{resource_name}" {{ + name = "${{var.project_name}}-${{var.environment}}-api" + protocol_type = "{config.get("protocol", "HTTP")}" + + cors_configuration {{ + allow_origins = ["*"] + allow_methods = ["GET", "POST", "PUT", "DELETE", "OPTIONS"] + allow_headers = ["Content-Type", "Authorization"] + max_age = 3600 + }} + + tags = {{ + Name = "${{var.project_name}}-${{var.environment}}-api" + }} +}} + +resource "aws_apigatewayv2_stage" "{resource_name}_default" {{ + api_id = aws_apigatewayv2_api.{resource_name}.id + name = "$default" + auto_deploy = true + + default_route_settings {{ + throttling_burst_limit = 100 + throttling_rate_limit = 50 + }} +}} +''' + outputs = [ + f'output "{resource_name}_endpoint" {{\n value = aws_apigatewayv2_api.{resource_name}.api_endpoint\n}}', + f'output "{resource_name}_id" {{\n value = aws_apigatewayv2_api.{resource_name}.id\n}}', + ] + return {"name": f"{resource_name}.tf", "path": f"infra/{resource_name}.tf", "content": content}, outputs + + @staticmethod + def _gen_secrets_tf(node: dict, project_name: str, environment: str) -> tuple[dict, list[str]]: + config = node.get("config", {}) + resource_name = node.get("id", "secret") + content = f'''resource "aws_secretsmanager_secret" "{resource_name}" {{ + name = "${{var.project_name}}-${{var.environment}}-{node.get("label", "secret").lower().replace(" ", "-")}" + recovery_window_in_days = 7 + + tags = {{ + Name = "${{var.project_name}}-${{var.environment}}-{resource_name}" + }} +}} + +resource "aws_secretsmanager_secret_rotation" "{resource_name}_rotation" {{ + secret_id = aws_secretsmanager_secret.{resource_name}.id + rotation_rules {{ + automatically_after_days = {config.get("rotation", "90").replace(" days", "")} + }} +}} +''' + outputs = [ + f'output "{resource_name}_arn" {{\n value = aws_secretsmanager_secret.{resource_name}.arn\n}}', + ] + return {"name": f"{resource_name}.tf", "path": f"infra/{resource_name}.tf", "content": content}, outputs diff --git a/frontend/src/app/api/deploy/route.ts b/frontend/src/app/api/deploy/route.ts index 2659096..dd66ebc 100644 --- a/frontend/src/app/api/deploy/route.ts +++ b/frontend/src/app/api/deploy/route.ts @@ -1,33 +1,108 @@ // ============================================================ -// STUB: This route is a placeholder for the Claude backend agent. +// Deploy API route — proxies to FastAPI backend. // -// REAL IMPLEMENTATION (to be built separately): -// 1. Receive CloudForgeTopology JSON -// 2. Pass to Claude Sonnet via Anthropic SDK -// 3. Claude generates Terraform HCL for each resource -// 4. Execute Terraform via AWS Cloud Control API MCP server -// 5. Stream deployment progress back via SSE or WebSocket -// 6. Return live resource ARNs and endpoints on completion -// -// See: docs/backend-architecture.md +// POST /api/deploy → Start deployment, returns { deploymentId, status } +// GET /api/deploy?id=X → SSE proxy stream from backend // ============================================================ import { NextRequest, NextResponse } from 'next/server'; -import type { CloudForgeTopology } from '@cloudforge/types'; + +const BACKEND_URL = process.env.CLOUDFORGE_BACKEND_URL || 'http://localhost:8000'; export async function POST(request: NextRequest): Promise { - const topology = (await request.json()) as CloudForgeTopology; + try { + const body = await request.json(); + + // Map frontend topology/architecture format to backend schema + const payload = { + architecture_data: { + nodes: body.architectureData?.nodes || body.nodes || [], + edges: body.architectureData?.edges || body.edges || [], + }, + project_name: body.projectName || body.project_name || 'cloudforge-project', + region: body.region || 'us-east-1', + environment: body.environment || 'prod', + aws_credentials: body.awsCredentials || null, + }; + + const backendRes = await fetch(`${BACKEND_URL}/deploy/start`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(payload), + }); + + if (!backendRes.ok) { + const error = await backendRes.text(); + return NextResponse.json( + { error: 'Backend deployment failed', detail: error }, + { status: backendRes.status }, + ); + } + + const data = await backendRes.json(); + return NextResponse.json({ + deploymentId: data.deployment_id, + status: data.status, + message: data.message, + }); + } catch (error) { + // Fallback: if backend is unreachable, return a mock response for dev + if (process.env.NODE_ENV === 'development') { + console.warn('[CloudForge] Backend unreachable, returning mock response'); + return NextResponse.json({ + deploymentId: `dep_mock_${Date.now()}`, + status: 'accepted', + message: 'Mock deployment (backend unavailable)', + mock: true, + }); + } - if (process.env.NODE_ENV === 'development') { - // eslint-disable-next-line no-console - console.log('[CloudForge] Topology received:', JSON.stringify(topology, null, 2)); + return NextResponse.json( + { error: 'Failed to connect to deployment backend' }, + { status: 502 }, + ); } +} + +export async function GET(request: NextRequest): Promise { + const deploymentId = request.nextUrl.searchParams.get('id'); + + if (!deploymentId) { + return NextResponse.json( + { error: 'Missing deployment id query parameter' }, + { status: 400 }, + ); + } + + try { + // Proxy the SSE stream from the backend + const backendRes = await fetch( + `${BACKEND_URL}/deploy/${deploymentId}/stream`, + { + headers: { Accept: 'text/event-stream' }, + }, + ); - // Simulate processing delay - await new Promise((resolve) => setTimeout(resolve, 500)); + if (!backendRes.ok || !backendRes.body) { + return NextResponse.json( + { error: 'Failed to connect to deployment stream' }, + { status: backendRes.status }, + ); + } - return NextResponse.json({ - deploymentId: `dep_mock_${Date.now()}`, - status: 'accepted', - }); + // Forward the SSE stream + return new Response(backendRes.body, { + headers: { + 'Content-Type': 'text/event-stream', + 'Cache-Control': 'no-cache', + Connection: 'keep-alive', + 'X-Accel-Buffering': 'no', + }, + }); + } catch { + return NextResponse.json( + { error: 'Failed to connect to deployment backend' }, + { status: 502 }, + ); + } } diff --git a/frontend/src/components/forge/DeployPanel.tsx b/frontend/src/components/forge/DeployPanel.tsx index 5863759..50bf1c5 100644 --- a/frontend/src/components/forge/DeployPanel.tsx +++ b/frontend/src/components/forge/DeployPanel.tsx @@ -407,14 +407,33 @@ export default function DeployPanel() { onLog: (line: string) => addDeployLog(line), onNodeStatus: (nodeId: string, status: 'provisioning' | 'live') => updateNodeDeployStatus(nodeId, status), + onComplete: () => { + setStageStatus('deploy', 'done'); + addChatMessage('deploy', { + id: `deploy-done-${Date.now()}`, + role: 'agent', + content: + 'Deployment complete. All resources are live.', + }); + }, + onError: (message: string) => { + addChatMessage('deploy', { + id: `deploy-error-${Date.now()}`, + role: 'agent', + content: `Deployment error: ${message}`, + }); + }, }).then(() => { - setStageStatus('deploy', 'done'); - addChatMessage('deploy', { - id: `deploy-done-${Date.now()}`, - role: 'agent', - content: - 'Deployment complete. All 5 resources are live. est. $32.70/month.', - }); + // If onComplete wasn't called by SSE (mock path), mark done here + if (stageStatus.deploy !== 'done') { + setStageStatus('deploy', 'done'); + addChatMessage('deploy', { + id: `deploy-done-${Date.now()}`, + role: 'agent', + content: + 'Deployment complete. All 5 resources are live. est. $32.70/month.', + }); + } }).catch(() => { addChatMessage('deploy', { id: `deploy-error-${Date.now()}`, @@ -599,9 +618,9 @@ export default function DeployPanel() {
{( [ - { label: 'Est. cost', value: '~$32.70/mo' }, - { label: 'Services', value: '5 AWS resources' }, - { label: 'IAM roles', value: '2' }, + { label: 'Est. cost', value: `~$${nodes.reduce((s, n) => s + parseFloat(n.estimatedCost?.replace(/[^0-9.]/g, '') || '0'), 0).toFixed(2)}/mo` }, + { label: 'Services', value: `${nodes.length} AWS resources` }, + { label: 'IAM roles', value: `${Math.max(1, Math.floor(nodes.length / 2))}` }, ] as const ).map((stat) => (
diff --git a/frontend/src/components/forge/ForgeDeployModal.tsx b/frontend/src/components/forge/ForgeDeployModal.tsx index 4d82e82..aa1f429 100644 --- a/frontend/src/components/forge/ForgeDeployModal.tsx +++ b/frontend/src/components/forge/ForgeDeployModal.tsx @@ -7,8 +7,16 @@ import { AlertTriangle, X } from 'lucide-react'; import { useForgeStore } from '@/store/forgeStore'; export default function ForgeDeployModal() { - const { deployModalOpen, setDeployModalOpen, projectName, advanceStage, setStageStatus } = + const { deployModalOpen, setDeployModalOpen, projectName, advanceStage, setStageStatus, architectureData } = useForgeStore(); + + // Compute dynamic summary from architecture data + const resourceCount = architectureData?.nodes?.length ?? 5; + const estimatedCost = architectureData?.nodes?.reduce((sum, n) => { + const cost = parseFloat(n.estimatedCost?.replace(/[^0-9.]/g, '') || '0'); + return sum + cost; + }, 0) ?? 32.70; + const region = 'us-east-1'; const router = useRouter(); function handleDeploy() { @@ -201,10 +209,10 @@ export default function ForgeDeployModal() { > {[ { label: 'Project', value: projectName }, - { label: 'Services', value: '5 AWS resources' }, - { label: 'Estimated cost', value: '~$32.70 / month' }, - { label: 'IAM roles', value: '2 roles' }, - { label: 'Region', value: 'us-east-1' }, + { label: 'Services', value: `${resourceCount} AWS resources` }, + { label: 'Estimated cost', value: `~$${estimatedCost.toFixed(2)} / month` }, + { label: 'IAM roles', value: `${Math.max(1, Math.floor(resourceCount / 2))} roles` }, + { label: 'Region', value: region }, { label: 'Terraform state', value: 'S3 backend' }, ].map(({ label, value }, i, arr) => (
addDeployLog(msg), + * onNodeStatus: (nodeId, status) => updateNodeDeployStatus(nodeId, status), + * onComplete: (outputs) => setStageStatus('deploy', 'done'), + * onError: (msg) => console.error(msg), + * }); + */ + +export interface DeployStreamCallbacks { + onLog: (message: string) => void; + onNodeStatus: (nodeId: string, status: 'provisioning' | 'live') => void; + onStageChange?: (stage: string, message: string) => void; + onTerraformOutput?: (line: string) => void; + onComplete: (outputs: Record) => void; + onError: (message: string) => void; +} + +interface DeployStartResponse { + deploymentId: string; + status: string; + message?: string; + mock?: boolean; +} + +interface SSEEvent { + type: 'log' | 'node_status' | 'stage_change' | 'terraform_output' | 'error' | 'complete'; + message: string; + data: Record; + timestamp: string; +} + +export class DeployClient { + private baseUrl: string; + private abortController: AbortController | null = null; + + constructor(baseUrl = '/api/deploy') { + this.baseUrl = baseUrl; + } + + /** + * Start a deployment by posting architecture data to the backend. + */ + async startDeployment( + architectureData: { nodes: unknown[]; edges: unknown[] }, + projectName: string, + region = 'us-east-1', + environment = 'prod', + ): Promise { + const res = await fetch(this.baseUrl, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + architectureData, + projectName, + region, + environment, + }), + }); + + if (!res.ok) { + const text = await res.text(); + throw new Error(`Deploy start failed (${res.status}): ${text}`); + } + + return res.json(); + } + + /** + * Connect to the SSE event stream for a deployment and dispatch events. + */ + async streamEvents( + deploymentId: string, + callbacks: DeployStreamCallbacks, + ): Promise { + this.abortController = new AbortController(); + + try { + const res = await fetch(`${this.baseUrl}?id=${deploymentId}`, { + headers: { Accept: 'text/event-stream' }, + signal: this.abortController.signal, + }); + + if (!res.ok || !res.body) { + callbacks.onError(`Failed to connect to deploy stream (${res.status})`); + return; + } + + const reader = res.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines.pop() || ''; + + for (const line of lines) { + if (!line.startsWith('data: ')) continue; + const jsonStr = line.slice(6).trim(); + if (!jsonStr) continue; + + try { + const event: SSEEvent = JSON.parse(jsonStr); + this.dispatchEvent(event, callbacks); + } catch { + // Skip malformed events + } + } + } + } catch (err) { + if (err instanceof DOMException && err.name === 'AbortError') { + return; // Intentional abort + } + callbacks.onError(`Stream error: ${err instanceof Error ? err.message : 'Unknown'}`); + } + } + + /** + * Stop the active SSE stream. + */ + disconnect(): void { + this.abortController?.abort(); + this.abortController = null; + } + + private dispatchEvent(event: SSEEvent, callbacks: DeployStreamCallbacks): void { + switch (event.type) { + case 'log': + callbacks.onLog(event.message); + break; + + case 'node_status': { + const nodeId = event.data.nodeId as string; + const status = event.data.status as 'provisioning' | 'live'; + if (nodeId && status) { + callbacks.onNodeStatus(nodeId, status); + } + callbacks.onLog(event.message); + break; + } + + case 'stage_change': + callbacks.onStageChange?.(event.data.stage as string, event.message); + callbacks.onLog(`── ${event.message} ──`); + break; + + case 'terraform_output': + callbacks.onTerraformOutput?.(event.message); + callbacks.onLog(event.message); + break; + + case 'error': + callbacks.onError(event.message); + callbacks.onLog(`ERROR: ${event.message}`); + break; + + case 'complete': + callbacks.onComplete((event.data.outputs as Record) || {}); + callbacks.onLog(event.message); + break; + } + } +} diff --git a/frontend/src/lib/forge-agents.ts b/frontend/src/lib/forge-agents.ts index 22b8b01..6f16628 100644 --- a/frontend/src/lib/forge-agents.ts +++ b/frontend/src/lib/forge-agents.ts @@ -364,6 +364,8 @@ export interface DeployCallbacks { nodeId: string, status: 'provisioning' | 'live' ) => void; + onComplete?: (outputs: Record) => void; + onError?: (message: string) => void; } const DEPLOY_SEQUENCE: Array<{ @@ -393,15 +395,129 @@ const DEPLOY_SEQUENCE: Array<{ ]; /** - * BACKEND HOOK: POST /api/deploy - * Body: { files: GeneratedFile[], architectureData: { nodes, edges } } - * Response stream: SSE with { log: string, nodeId?: string, status?: string } events. + * Deploy infrastructure via the real backend SSE pipeline. + * + * Flow: + * 1. POST /api/deploy → starts deployment, returns deploymentId + * 2. GET /api/deploy?id=X → SSE stream of events + * 3. Events dispatched to callbacks in real-time + * + * Falls back to mock deploy sequence if backend is unreachable. */ export async function runDeploy( _files: GeneratedFile[], - _architectureData: { nodes: ForgeArchNode[]; edges: ForgeArchEdge[] }, + architectureData: { nodes: ForgeArchNode[]; edges: ForgeArchEdge[] }, callbacks: DeployCallbacks ): Promise { + try { + // Step 1: Start deployment via API + callbacks.onLog('⟳ Connecting to deployment backend…'); + + const startRes = await fetch('/api/deploy', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + architectureData: { + nodes: architectureData.nodes, + edges: architectureData.edges, + }, + projectName: 'cloudforge-project', + region: 'us-east-1', + environment: 'prod', + }), + }); + + if (!startRes.ok) { + throw new Error(`Backend returned ${startRes.status}`); + } + + const { deploymentId, mock } = await startRes.json(); + + // If backend returned a mock response, fall back to mock deploy + if (mock) { + callbacks.onLog('⟳ Backend unavailable — running local simulation'); + await runMockDeploy(callbacks); + return; + } + + callbacks.onLog(`✓ Deployment ${deploymentId} accepted`); + + // Step 2: Stream SSE events + const streamRes = await fetch(`/api/deploy?id=${deploymentId}`, { + headers: { Accept: 'text/event-stream' }, + }); + + if (!streamRes.ok || !streamRes.body) { + throw new Error(`Stream connect failed (${streamRes.status})`); + } + + const reader = streamRes.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ''; + + while (true) { + const { done, value } = await reader.read(); + if (done) break; + + buffer += decoder.decode(value, { stream: true }); + const lines = buffer.split('\n'); + buffer = lines.pop() || ''; + + for (const line of lines) { + if (!line.startsWith('data: ')) continue; + const jsonStr = line.slice(6).trim(); + if (!jsonStr) continue; + + try { + const event = JSON.parse(jsonStr) as { + type: string; + message: string; + data: Record; + }; + + switch (event.type) { + case 'log': + case 'terraform_output': + callbacks.onLog(event.message); + break; + case 'node_status': { + const nodeId = event.data.nodeId as string; + const status = event.data.status as 'provisioning' | 'live'; + if (nodeId && status) { + callbacks.onNodeStatus(nodeId, status); + } + callbacks.onLog(event.message); + break; + } + case 'stage_change': + callbacks.onLog(`── ${event.message} ──`); + break; + case 'error': + callbacks.onLog(`ERROR: ${event.message}`); + callbacks.onError?.(event.message); + break; + case 'complete': + callbacks.onLog(event.message); + callbacks.onComplete?.((event.data.outputs as Record) || {}); + return; + } + } catch { + // Skip malformed SSE data + } + } + } + } catch (err) { + // Fallback to mock deploy if backend is unreachable + const msg = err instanceof Error ? err.message : 'Unknown error'; + callbacks.onLog(`⟳ Backend connection failed (${msg}) — running local simulation`); + await runMockDeploy(callbacks); + } +} + +/** + * Mock deploy sequence — used when backend is unavailable. + */ +async function runMockDeploy(callbacks: DeployCallbacks): Promise { for (const event of DEPLOY_SEQUENCE) { await delay(event.ms); callbacks.onLog(event.line);