# DYNAMIC NODE.JS DESIGNER - AI-powered Express.js architecture based on actual features # Uses Claude AI to generate Node.js/Express backend based on functional requirements from typing import Dict, Any from loguru import logger from designers.base_designer import BaseBackendDesigner from prompts.backend.nodejs_prompts import NodejsPrompts class NodejsDesigner(BaseBackendDesigner): """Dynamic Node.js specialist - Generates Express.js architecture based on actual project features""" def __init__(self): super().__init__() self.prompts = NodejsPrompts() logger.info("⚙️ Dynamic Node.js Designer initialized - AI-powered feature-based API design") def get_technology_name(self) -> str: return "Node.js" async def design_architecture(self, context: Dict[str, Any]) -> Dict[str, Any]: """Design Node.js/Express architecture dynamically based on actual features and tech stack""" try: logger.info("⚙️ Node.js Designer analyzing project features...") # Extract real project data functional_reqs = context['functional_requirements'] tech_stack = context['technology_stack'] business_context = context['business_context'] logger.info(f" Feature: {functional_reqs['feature_name']}") logger.info(f" Technical Requirements: {len(functional_reqs['technical_requirements'])} items") logger.info(f" Business Rules: {len(functional_reqs['business_logic_rules'])} rules") # Generate AI prompt based on actual project requirements prompt = self.prompts.create_dynamic_nodejs_prompt( feature_name=functional_reqs['feature_name'], feature_description=functional_reqs['description'], technical_requirements=functional_reqs['technical_requirements'], business_logic_rules=functional_reqs['business_logic_rules'], complexity_level=functional_reqs['complexity_level'], tech_stack=tech_stack, all_features=functional_reqs['all_features'] ) # Get AI-generated Node.js architecture logger.info("🤖 Generating Node.js architecture with Claude AI...") response = await self.claude_client.generate_architecture(prompt) if response.get('success'): nodejs_architecture = response['data'] # Enhance with Node.js-specific patterns based on tech stack enhanced_architecture = self._enhance_with_tech_stack( nodejs_architecture, tech_stack, functional_reqs ) logger.info("✅ Dynamic Node.js architecture generated successfully") return { "success": True, "architecture": enhanced_architecture, "specialist": "Node.js", "framework": "Express.js", "generated_for_feature": functional_reqs['feature_name'], "authentication": self._extract_auth_method(tech_stack), "database_integration": self._extract_database_config(tech_stack), "patterns_used": self._extract_nodejs_patterns(tech_stack, functional_reqs), "ai_generated": True, "feature_specific": True } else: logger.warning("Claude AI generation failed, creating feature-based fallback") return self._create_feature_based_fallback(functional_reqs, tech_stack) except Exception as e: logger.error(f"❌ Node.js architecture design failed: {e}") return self._create_feature_based_fallback(functional_reqs, tech_stack) async def design_api_endpoints(self, context: Dict[str, Any]) -> Dict[str, Any]: """Design Express.js API endpoints based on actual features""" # Will implement specific API design if needed pass async def design_middleware(self, context: Dict[str, Any]) -> Dict[str, Any]: """Design Express.js middleware chain based on requirements""" # Will implement specific middleware design if needed pass async def design_services(self, context: Dict[str, Any]) -> Dict[str, Any]: """Design service layer based on business logic rules""" # Will implement specific service design if needed pass def _enhance_with_tech_stack(self, architecture: Dict, tech_stack: Dict, functional_reqs: Dict) -> Dict: """Enhance AI-generated architecture with specific tech stack choices""" # Extract tech stack details backend_config = tech_stack.get('backend', {}) database_config = tech_stack.get('database', {}) security_config = tech_stack.get('security', {}) # Enhance folder structure based on complexity and features if 'folder_structure' not in architecture: architecture['folder_structure'] = {} # Add tech-stack-specific configuration architecture['folder_structure'].update({ "package_json_dependencies": self._generate_dependencies(tech_stack), "database_configuration": self._generate_db_config(database_config), "authentication_setup": self._generate_auth_setup(security_config), "middleware_configuration": self._generate_middleware_config(functional_reqs) }) # Add API endpoints based on features architecture['api_endpoints'] = self._generate_feature_endpoints(functional_reqs, tech_stack) # Add environment configuration architecture['environment_configuration'] = { "environment_variables": self._generate_env_vars(tech_stack), "docker_configuration": self._generate_docker_config(tech_stack), "deployment_setup": self._generate_deployment_config(tech_stack) } return architecture def _extract_auth_method(self, tech_stack: Dict) -> str: """Extract authentication method from tech stack""" security_config = tech_stack.get('security', {}) auth_method = security_config.get('authentication', 'JWT') return auth_method def _extract_database_config(self, tech_stack: Dict) -> Dict: """Extract database configuration from tech stack""" database_config = tech_stack.get('database', {}) return { "primary_database": database_config.get('primary', 'PostgreSQL'), "secondary_databases": database_config.get('secondary', []), "orm_choice": self._determine_orm(database_config), "connection_pooling": True } def _determine_orm(self, database_config: Dict) -> str: """Determine ORM based on database choice""" primary_db = database_config.get('primary', '').lower() if 'postgresql' in primary_db or 'mysql' in primary_db: return 'Prisma' # Modern choice for SQL databases elif 'mongodb' in primary_db: return 'Mongoose' else: return 'Prisma' # Default def _extract_nodejs_patterns(self, tech_stack: Dict, functional_reqs: Dict) -> list: """Extract Node.js patterns based on tech stack and requirements""" patterns = [ "Express.js Framework", "Async/Await Pattern", "Error Handling Middleware", "Input Validation with Joi", "CORS Configuration", "Security Headers with Helmet" ] # Add patterns based on authentication auth_method = self._extract_auth_method(tech_stack) if auth_method == 'JWT': patterns.extend([ "JWT Token Authentication", "Refresh Token Pattern", "Protected Route Middleware" ]) # Add patterns based on complexity complexity = functional_reqs.get('complexity_level', 'medium') if complexity == 'high': patterns.extend([ "Service Layer Pattern", "Repository Pattern", "Dependency Injection", "Request Rate Limiting" ]) # Add patterns based on business rules business_rules = functional_reqs.get('business_logic_rules', []) if business_rules: patterns.append("Business Logic Validation") patterns.append("Role-Based Access Control") return patterns def _generate_dependencies(self, tech_stack: Dict) -> Dict: """Generate package.json dependencies based on tech stack""" backend_config = tech_stack.get('backend', {}) database_config = tech_stack.get('database', {}) dependencies = { "express": "^4.18.0", "cors": "^2.8.5", "helmet": "^6.0.0", "morgan": "^1.10.0", "joi": "^17.7.0", "bcryptjs": "^2.4.3", "dotenv": "^16.0.3" } # Add authentication dependencies auth_method = self._extract_auth_method(tech_stack) if auth_method == 'JWT': dependencies["jsonwebtoken"] = "^9.0.0" # Add database dependencies orm_choice = self._determine_orm(database_config) if orm_choice == 'Prisma': dependencies.update({ "prisma": "^4.8.0", "@prisma/client": "^4.8.0" }) elif orm_choice == 'Mongoose': dependencies["mongoose"] = "^6.8.0" # Add PostgreSQL specific primary_db = database_config.get('primary', '').lower() if 'postgresql' in primary_db: dependencies["pg"] = "^8.8.0" dependencies["@types/pg"] = "^8.6.0" return dependencies def _generate_db_config(self, database_config: Dict) -> Dict: """Generate database configuration""" primary_db = database_config.get('primary', 'PostgreSQL') orm_choice = self._determine_orm(database_config) config = { "database_system": primary_db, "orm": orm_choice, "connection_string": "Process.env.DATABASE_URL", "connection_pooling": { "min": 2, "max": 10, "idle_timeout": 30000 } } if orm_choice == 'Prisma': config["schema_file"] = "prisma/schema.prisma" config["migration_command"] = "npx prisma migrate dev" return config def _generate_auth_setup(self, security_config: Dict) -> Dict: """Generate authentication setup""" auth_method = security_config.get('authentication', 'JWT') if auth_method == 'JWT': return { "token_type": "JWT", "secret_env_var": "JWT_SECRET", "token_expiry": "24h", "refresh_token": True, "middleware": "authenticateToken middleware", "password_hashing": "bcryptjs with salt rounds 12" } return {"method": auth_method} def _generate_middleware_config(self, functional_reqs: Dict) -> Dict: """Generate middleware configuration based on requirements""" middleware = { "global_middleware": [ "helmet() - Security headers", "cors() - CORS configuration", "express.json() - JSON body parser", "morgan('combined') - Request logging" ], "authentication_middleware": "JWT verification and user extraction", "validation_middleware": "Joi schema validation for requests", "error_middleware": "Global error handler" } # Add business rule middleware business_rules = functional_reqs.get('business_logic_rules', []) if business_rules: middleware["business_rule_middleware"] = [ f"Middleware for: {rule}" for rule in business_rules[:3] # First 3 rules ] return middleware def _generate_feature_endpoints(self, functional_reqs: Dict, tech_stack: Dict) -> Dict: """Generate API endpoints based on actual features""" feature_name = functional_reqs.get('feature_name', 'Item') feature_lower = feature_name.lower().replace(' ', '') # Base authentication endpoints endpoints = { "authentication": { f"POST /api/v1/auth/register": { "description": "User registration", "middleware": ["validateRegistration"], "request_body": { "email": "string (required)", "password": "string (required, min 8 chars)", "name": "string (required)" }, "response": "User object with JWT token" }, f"POST /api/v1/auth/login": { "description": "User login", "middleware": ["validateLogin", "rateLimiter"], "request_body": { "email": "string (required)", "password": "string (required)" }, "response": "User object with JWT token" } } } # Feature-specific endpoints feature_endpoints = { f"GET /api/v1/{feature_lower}": { "description": f"Get all {feature_name.lower()} items", "middleware": ["authenticateToken"], "query_params": { "page": "number (optional)", "limit": "number (optional)", "search": "string (optional)" }, "response": f"Array of {feature_name.lower()} items with pagination" }, f"POST /api/v1/{feature_lower}": { "description": f"Create new {feature_name.lower()}", "middleware": ["authenticateToken", f"validate{feature_name}"], "request_body": f"Dynamic based on {feature_name} requirements", "response": f"Created {feature_name.lower()} object" }, f"GET /api/v1/{feature_lower}/:id": { "description": f"Get specific {feature_name.lower()} by ID", "middleware": ["authenticateToken", f"authorize{feature_name}Access"], "response": f"{feature_name} object or 404" }, f"PUT /api/v1/{feature_lower}/:id": { "description": f"Update {feature_name.lower()}", "middleware": ["authenticateToken", f"authorize{feature_name}Edit", f"validate{feature_name}Update"], "response": f"Updated {feature_name.lower()} object" }, f"DELETE /api/v1/{feature_lower}/:id": { "description": f"Delete {feature_name.lower()}", "middleware": ["authenticateToken", f"authorize{feature_name}Delete"], "response": "Success confirmation" } } endpoints[feature_lower] = feature_endpoints # Add business rule endpoints if needed business_rules = functional_reqs.get('business_logic_rules', []) if business_rules: endpoints["business_operations"] = { f"POST /api/v1/{feature_lower}/validate": { "description": f"Validate {feature_name.lower()} against business rules", "middleware": ["authenticateToken"], "business_rules_applied": business_rules } } return endpoints def _generate_env_vars(self, tech_stack: Dict) -> list: """Generate environment variables based on tech stack""" env_vars = [ "NODE_ENV", "PORT", "DATABASE_URL", "JWT_SECRET", "JWT_EXPIRES_IN" ] # Add database-specific variables database_config = tech_stack.get('database', {}) primary_db = database_config.get('primary', '').lower() if 'postgresql' in primary_db: env_vars.extend([ "POSTGRES_HOST", "POSTGRES_PORT", "POSTGRES_DB", "POSTGRES_USER", "POSTGRES_PASSWORD" ]) # Add Redis if mentioned secondary_dbs = database_config.get('secondary', []) if any('redis' in db.lower() for db in secondary_dbs): env_vars.extend([ "REDIS_URL", "REDIS_PASSWORD" ]) return env_vars def _generate_docker_config(self, tech_stack: Dict) -> Dict: """Generate Docker configuration""" return { "dockerfile": "Multi-stage Node.js Dockerfile", "base_image": "node:18-alpine", "working_directory": "/app", "exposed_port": 8001, "health_check": "GET /health endpoint", "optimization": "Node modules caching, multi-stage build" } def _generate_deployment_config(self, tech_stack: Dict) -> Dict: """Generate deployment configuration""" cloud_provider = tech_stack.get('infrastructure', {}).get('cloud_provider', 'AWS') return { "cloud_provider": cloud_provider, "containerization": "Docker with Express.js", "scaling": "Horizontal scaling with load balancer", "monitoring": "Health checks and performance metrics", "logging": "Structured logging with correlation IDs" } def _create_feature_based_fallback(self, functional_reqs: Dict, tech_stack: Dict) -> Dict: """Create fallback Node.js architecture based on actual features""" logger.warning("Creating feature-based Node.js fallback architecture") feature_name = functional_reqs.get('feature_name', 'Application') return { "success": True, "architecture": { "folder_structure": { "src/controllers": f"Controllers for {feature_name} operations", "src/services": f"Business logic services for {feature_name}", "src/models": f"Data models for {feature_name}", "src/routes": f"API routes for {feature_name}", "src/middleware": "Authentication and validation middleware", "src/config": "Database and application configuration" }, "api_endpoints": { "authentication": "POST /auth/login, POST /auth/register", f"{feature_name.lower()}": f"CRUD operations for {feature_name}" }, "database_integration": self._extract_database_config(tech_stack), "authentication": self._extract_auth_method(tech_stack), "dependencies": self._generate_dependencies(tech_stack) }, "specialist": "Node.js", "framework": "Express.js", "fallback": True, "feature_based": True, "generated_for": feature_name }