""" Architecture Agent Memory Module Provides memory and pattern library for the architecture agent using Upstash Redis. Features: - Session state caching (fast read/write for active sessions) - Pattern library (reusable architecture patterns) - RAG-style retrieval (semantic search via tags) """ import json import logging from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any from .config import settings logger = logging.getLogger("architecture_memory") # Default TTL values (in seconds) SESSION_TTL = 3600 # 1 hour for active sessions PATTERN_TTL = 86400 * 30 # 30 days for patterns (permanent-ish) # Path to fallback JSON file used when Redis is unavailable _FALLBACK_PATTERNS_PATH = Path(__file__).parent / "default_patterns.json" @dataclass class ArchitecturePattern: """An architecture pattern that can be reused.""" id: str name: str description: str use_cases: list[str] tags: list[str] components: list[str] pros: list[str] cons: list[str] tech_stack: dict[str, str] estimated_cost: str estimated_setup_time: str created_at: str | None = None class ArchitectureMemory: """ Memory system for architecture agent using Upstash Redis. Key features: - Fast session state caching - Pattern library for reusable architectures - Tag-based pattern retrieval """ def __init__(self): self._client = None self._initialized = False self.url = settings.upstash_redis_rest_url self.token = settings.upstash_redis_rest_token if not self.url or not self.token: logger.warning( "Upstash Redis credentials not found for Architecture Memory. " "Set UPSTASH_REDIS_REST_URL and UPSTASH_REDIS_REST_TOKEN." ) return try: from upstash_redis import Redis self._client = Redis(url=self.url, token=self.token) self._initialized = True logger.info("Architecture Memory Redis client initialized") except Exception as e: logger.error(f"Failed to initialize Architecture Memory Redis client: {e}") @property def is_configured(self) -> bool: return self._initialized and self._client is not None def _make_session_key(self, session_id: str) -> str: """Generate Redis key for session.""" return f"architecture:session:{session_id}" def _make_pattern_key(self, pattern_id: str) -> str: """Generate Redis key for pattern.""" return f"architecture:pattern:{pattern_id}" def _make_patterns_set_key(self) -> str: """Generate Redis key for pattern index.""" return "architecture:patterns:index" def _make_pattern_tags_key(self, tag: str) -> str: """Generate Redis key for tag index.""" return f"architecture:patterns:tag:{tag}" # ============================================================ # Session Caching # ============================================================ async def cache_session( self, session_id: str, session_data: dict[str, Any] ) -> bool: """ Cache architecture session for fast access. Args: session_id: The session ID session_data: Session data to cache Returns: True if cached successfully """ if not self.is_configured: return False try: key = self._make_session_key(session_id) serialized = json.dumps(session_data, default=str) self._client.set(key, serialized, ex=SESSION_TTL) logger.debug(f"Cached session {session_id} (TTL: {SESSION_TTL}s)") return True except Exception as e: logger.warning(f"Failed to cache session {session_id}: {e}") return False async def get_cached_session(self, session_id: str) -> dict[str, Any] | None: """ Get cached session data. Args: session_id: The session ID Returns: Session data or None if not found """ if not self.is_configured: return None try: key = self._make_session_key(session_id) value = self._client.get(key) if value: logger.debug(f"Cache HIT for session {session_id}") return json.loads(value) logger.debug(f"Cache MISS for session {session_id}") return None except Exception as e: logger.warning(f"Failed to get cached session {session_id}: {e}") return None async def invalidate_session(self, session_id: str) -> bool: """ Invalidate cached session. Args: session_id: The session ID Returns: True if invalidated successfully """ if not self.is_configured: return False try: key = self._make_session_key(session_id) self._client.delete(key) logger.debug(f"Invalidated session cache {session_id}") return True except Exception as e: logger.warning(f"Failed to invalidate session {session_id}: {e}") return False # ============================================================ # Pattern Library # ============================================================ async def store_pattern(self, pattern: ArchitecturePattern) -> bool: """ Store an architecture pattern in the library. Args: pattern: The pattern to store Returns: True if stored successfully """ if not self.is_configured: return False try: pattern_key = self._make_pattern_key(pattern.id) pattern.created_at = pattern.created_at or datetime.utcnow().isoformat() serialized = json.dumps(pattern.__dict__, default=str) self._client.set(pattern_key, serialized, ex=PATTERN_TTL) # Add to index index_key = self._make_patterns_set_key() self._client.sadd(index_key, pattern.id) # Add to tag indexes for tag in pattern.tags: tag_key = self._make_pattern_tags_key(tag) self._client.sadd(tag_key, pattern.id) logger.info(f"Stored pattern: {pattern.id}") return True except Exception as e: logger.warning(f"Failed to store pattern {pattern.id}: {e}") return False async def get_pattern(self, pattern_id: str) -> ArchitecturePattern | None: """ Get an architecture pattern by ID. Args: pattern_id: The pattern ID Returns: Pattern or None if not found """ if not self.is_configured: return None try: pattern_key = self._make_pattern_key(pattern_id) value = self._client.get(pattern_key) if value: data = json.loads(value) return ArchitecturePattern(**data) return None except Exception as e: logger.warning(f"Failed to get pattern {pattern_id}: {e}") return None async def get_all_patterns(self) -> list[ArchitecturePattern]: """ Get all architecture patterns. Returns: List of all patterns """ if not self.is_configured: return [] try: index_key = self._make_patterns_set_key() pattern_ids = self._client.smembers(index_key) patterns = [] for pattern_id in pattern_ids: pattern = await self.get_pattern(pattern_id) if pattern: patterns.append(pattern) return patterns except Exception as e: logger.warning(f"Failed to get all patterns: {e}") return [] async def get_patterns_by_tag(self, tag: str) -> list[ArchitecturePattern]: """ Get patterns by tag (RAG-style retrieval). Args: tag: The tag to search for Returns: List of matching patterns """ if not self.is_configured: return [] try: tag_key = self._make_pattern_tags_key(tag) pattern_ids = self._client.smembers(tag_key) patterns = [] for pattern_id in pattern_ids: pattern = await self.get_pattern(pattern_id) if pattern: patterns.append(pattern) return patterns except Exception as e: logger.warning(f"Failed to get patterns by tag {tag}: {e}") return [] async def search_patterns(self, query: str) -> list[ArchitecturePattern]: """ Search patterns by query (simple text search). Searches in name, description, use_cases, and tags. Args: query: Search query Returns: List of matching patterns """ if not self.is_configured: return [] query_lower = query.lower() try: all_patterns = await self.get_all_patterns() matching_patterns = [] for pattern in all_patterns: # Search in various fields search_fields = [ pattern.name.lower(), pattern.description.lower(), pattern.id.lower(), ] search_fields.extend([uc.lower() for uc in pattern.use_cases]) search_fields.extend([t.lower() for t in pattern.tags]) search_fields.extend([c.lower() for c in pattern.components]) if any(query_lower in field for field in search_fields): matching_patterns.append(pattern) return matching_patterns except Exception as e: logger.warning(f"Failed to search patterns: {e}") return [] async def seed_default_patterns(self) -> int: """ Seed the pattern library with default architecture patterns. When Redis is unavailable or not configured, falls back to loading patterns from the local JSON file at ``_FALLBACK_PATTERNS_PATH``. Returns: Number of patterns seeded """ if not self.is_configured: return await self._seed_from_fallback() default_patterns = [ # ============================================================ # Monolith Variants # ============================================================ ArchitecturePattern( id="pattern_monolithic", name="Traditional Monolith", description="Single deployable unit with all functionality in one codebase. Simple but can become complex as the application grows.", use_cases=[ "Small applications", "Prototypes", "MVPs", "Internal tools", ], tags=["monolith", "simple", "traditional", "single-deployment"], components=["Web Server", "Application Server", "Database", "CDN"], tech_stack={ "frontend": "React/Next.js", "backend": "Python FastAPI / Node.js / Django / Rails", "database": "PostgreSQL / MySQL", "infrastructure": "Docker / AWS ECS", }, pros=[ "Simple to develop", "Simple to deploy", "Lower initial cost", "Fewer network hops", "Easy debugging", ], cons=[ "Hard to scale", "Tight coupling", "Single point of failure", "Slower development as codebase grows", ], estimated_cost="$100-500/month", estimated_setup_time="1-2 weeks", ), ArchitecturePattern( id="pattern_modular_monolith", name="Modular Monolith", description="Monolith organized as loosely coupled domain modules based on DDD bounded contexts. Each module has its own API (facade) and optionally its own database schema. Enables clean boundaries while maintaining deployment simplicity.", use_cases=[ "Medium applications", "Migration from monolith to microservices", "Teams wanting modularity without complexity", "Domain-driven applications", ], tags=[ "monolith", "modular", "ddd", "domain-oriented", "bounded-context", "clean-architecture", ], components=[ "Domain Modules", "Module APIs (Facades)", "Shared Kernel", "Database (possibly schema-per-module)", ], tech_stack={ "framework": "Any (FastAPI/Django/Spring/.NET)", "organization": "Package by feature/domain", "database": "PostgreSQL (schema-per-module or database-per-module)", "testing": "Integration tests per module", }, pros=[ "Clear domain boundaries", "Independent module development", "Single deployment", "Testable", "Can evolve to microservices later", ], cons=[ "Requires DDD discipline", "Database coupling if not separated", "Scale limitations", ], estimated_cost="$200-800/month", estimated_setup_time="2-4 weeks", ), ArchitecturePattern( id="pattern_layered", name="Layered Architecture", description="Traditional N-tier architecture with clear separation: Presentation → Business Logic → Data Access. Foundation for most enterprise applications.", use_cases=[ "Enterprise applications", "CRUD apps", "Business systems", "Legacy modernization", ], tags=["layered", "n-tier", "traditional", "enterprise", "3-tier"], components=[ "Presentation Layer", "Business Logic Layer", "Data Access Layer", "Database", ], tech_stack={ "frontend": "React / Vue / Angular", "backend": "Spring (Java) / .NET / Django", "database": "PostgreSQL / Oracle / SQL Server", }, pros=[ "Well-understood", "Clear separation of concerns", "Easy to test layer by layer", "Good for CRUD operations", ], cons=[ "Can lead to anaemic domain models", "Multiple database trips for complex queries", "Not ideal for complex business logic", ], estimated_cost="$200-1000/month", estimated_setup_time="2-3 weeks", ), ArchitecturePattern( id="pattern_hexagonal", name="Hexagonal Architecture (Ports & Adapters)", description="Application with core business logic isolated from external concerns. Dependencies point inward. Core defines ports (interfaces), adapters implement ports.", use_cases=[ "Complex domain logic", "Testability requirements", "Multi-channel delivery (Web, API, CLI)", "Long-lived applications", ], tags=[ "hexagonal", "ports-and-adapters", "clean-architecture", "onion", "domain-driven", ], components=[ "Domain Core", "Application Services", "Ports (Interfaces)", "Adapters (UI, DB, External API)", ], tech_stack={ "domain": "Pure Python/Java/C#", "ports": "Abstract interfaces", "adapters": "REST, GraphQL, SQL, MongoDB, Redis", }, pros=[ "Highly testable", "Framework-agnostic core", "Flexible delivery mechanisms", "Clear dependency rule", ], cons=[ "Initial complexity", "Overhead for simple apps", "Requires experienced team", ], estimated_cost="$300-1500/month", estimated_setup_time="3-5 weeks", ), # ============================================================ # Microservices & Distributed # ============================================================ ArchitecturePattern( id="pattern_microservices", name="Microservices Enterprise", description="Full microservices architecture with independently deployable services. Each service owns its data (database-per-service). Requires service mesh and robust observability.", use_cases=[ "Large enterprise apps", "Multiple teams", "High availability systems", "Frequent deployments", ], tags=[ "microservices", "enterprise", "kubernetes", "distributed", "scalable", ], components=[ "API Gateway", "Service Mesh", "Message Queue", "Multiple Services", "Monitoring", "Distributed Tracing", ], tech_stack={ "orchestration": "Kubernetes / ECS / Docker Swarm", "service_mesh": "Istio / Linkerd / Envoy", "messaging": "Kafka / RabbitMQ / NATS", "monitoring": "Prometheus / Grafana / Jaeger", "database": "PostgreSQL / MongoDB / DynamoDB per service", }, pros=[ "Independent scaling", "Team autonomy", "Technology flexibility", "Fault isolation", "Frequent deployments", ], cons=[ "Operational complexity", "Network latency", "Data consistency challenges", "Hard to debug", "Distributed transactions", ], estimated_cost="$2000-10000/month", estimated_setup_time="2-6 months", ), ArchitecturePattern( id="pattern_service_mesh", name="Service Mesh", description="Microservices with dedicated infrastructure layer handling service-to-service communication. Offloads cross-cutting concerns (load balancing, retries, security) from application code.", use_cases=[ "Large microservice deployments", "Zero-trust security", "Multi-cluster deployments", ], tags=[ "service-mesh", "istio", "linkerd", "kubernetes", "observability", ], components=[ "Data Plane (sidecar proxies)", "Control Plane", "Service Discovery", "Mutual TLS", "Traffic Management", ], tech_stack={ "service_mesh": "Istio / Linkerd", "ingress": "Ambassador / Contour", "monitoring": "Kiali / Jaeger / Prometheus", }, pros=[ "Out-of-box observability", "Zero-trust security", "Traffic splitting", "Resilience patterns", ], cons=[ "Complexity", "Resource overhead", "Steep learning curve", "Debugging challenges", ], estimated_cost="$3000-15000/month", estimated_setup_time="3-6 months", ), # ============================================================ # Event-Driven & CQRS # ============================================================ ArchitecturePattern( id="pattern_event_driven", name="Event-Driven Architecture", description="Services communicate through events (messages). Producers emit events, consumers react. Enables loose coupling and real-time processing.", use_cases=[ "Real-time updates", "Financial systems", "IoT", "Notification systems", "Audit logs", ], tags=["event-driven", "async", "messaging", "pub-sub", "reactive"], components=[ "Event Producers", "Event Consumers", "Message Broker", "Event Store", ], tech_stack={ "messaging": "Kafka / RabbitMQ / AWS EventBridge", "processing": "Spark / Flink / AWS Lambda", "storage": "Kafka / EventStoreDB", }, pros=[ "Loose coupling", "Scalability", "Audit trail", "Real-time capabilities", "Fault tolerance", ], cons=[ "Eventual consistency", "Complexity", "Event ordering", "Debugging", ], estimated_cost="$500-3000/month", estimated_setup_time="2-4 months", ), ArchitecturePattern( id="pattern_cqrs", name="CQRS (Command Query Responsibility Segregation)", description="Separate models for reading and writing data. Write side handles commands, read side optimizes for queries. Often paired with Event Sourcing.", use_cases=[ "Read-heavy applications", "Complex query requirements", "Different read/write scaling needs", "Event sourcing companion", ], tags=[ "cqrs", "read-write-separation", "event-sourcing", "domain-driven", ], components=[ "Command Side", "Query Side", "Event Bus", "Read Models (projections)", ], tech_stack={ "commands": "Axon Framework / MediatR / EventFlow", "events": "Kafka / RabbitMQ", "read_db": "PostgreSQL / MongoDB / Redis", "projections": "Elasticsearch / DynamoDB", }, pros=[ "Optimized read/write", "Scalability", "Flexible queries", "Audit of changes", ], cons=[ "Complexity", "Eventual consistency", "Learning curve", "Overkill for simple apps", ], estimated_cost="$500-2500/month", estimated_setup_time="2-3 months", ), ArchitecturePattern( id="pattern_saga", name="Saga Pattern (Distributed Transactions)", description="Manage distributed transactions across services through a sequence of local transactions. Each step has a compensating transaction for rollback.", use_cases=[ "Order processing", "Booking systems", "Multi-service transactions", "Financial workflows", ], tags=[ "saga", "distributed-transactions", "compensation", "choreography", "orchestration", ], components=[ "Saga Orchestrator (or choreography)", "Participant Services", "Compensation Handlers", "State Store", ], tech_stack={ "orchestration": "Axon Framework / Camunda / Temporal", "messaging": "Kafka / RabbitMQ", }, pros=[ "Distributed transactions", "Loose coupling", "Eventual consistency", ], cons=[ "Complexity", "Idempotency required", "Debugging hard", "Long-running transactions", ], estimated_cost="$500-2000/month", estimated_setup_time="2-3 months", ), # ============================================================ # Cloud-Native # ============================================================ ArchitecturePattern( id="pattern_serverless", name="Serverless Architecture", description="Cloud-native approach where cloud provider manages infrastructure. Functions as the unit of computation. Pay-per-use pricing.", use_cases=[ "Variable traffic", "Cost-sensitive projects", "Event-driven apps", "Rapid prototyping", ], tags=["serverless", "lambda", "faas", "cloud-native", "cost-optimized"], components=[ "Functions", "API Gateway", "Cloud Storage", "Cloud DB", "CDN", ], tech_stack={ "compute": "AWS Lambda / Azure Functions / Cloud Functions / Vercel", "database": "DynamoDB / Aurora Serverless / Firebase", "storage": "S3 / Cloud Storage", "cdn": "CloudFront / Cloudflare", }, pros=[ "Pay-per-use", "Auto-scale", "No server management", "Global availability", "Fast deployment", ], cons=[ "Vendor lock-in", "Cold starts", "Limited execution time", "Debugging challenges", " Stateless limitations", ], estimated_cost="$10-300/month", estimated_setup_time="2-4 weeks", ), ArchitecturePattern( id="pattern_edge_computing", name="Edge Computing", description="Process data close to where it's generated. Reduces latency and bandwidth. Content served from edge locations.", use_cases=[ "IoT", "Real-time analytics", "Global applications", "Streaming", ], tags=["edge", "cdn", "iot", "low-latency", "global"], components=[ "Edge Functions", "Edge Storage", "CDN", "Regional Processing", "IoT Hub", ], tech_stack={ "edge": "Cloudflare Workers / AWS Lambda@Edge / Fastly", "cdn": "Cloudflare / Fastly / Akamai", "iot": "AWS IoT / Azure IoT Hub", }, pros=[ "Low latency", "Reduced bandwidth", "Offline capability", "Data sovereignty", ], cons=["Complexity", "Limited compute", "Debugging", "Vendor lock-in"], estimated_cost="$100-1000/month", estimated_setup_time="4-8 weeks", ), # ============================================================ # Domain-Specific Patterns # ============================================================ ArchitecturePattern( id="pattern_saas_mvp", name="SaaS MVP", description="Minimal viable product for SaaS application with essential features for initial market validation.", use_cases=[ "Startup MVP", "Quick market validation", "SaaS product launch", ], tags=["saas", "mvp", "startup", "web", "cloud", "b2b"], components=[ "Web App", "Auth Service", "Database", "API Gateway", "Multi-tenancy", ], tech_stack={ "frontend": "Next.js / React", "backend": "Node.js / Python FastAPI / Supabase", "database": "PostgreSQL (with row-level security)", "auth": "Auth0 / Supabase Auth / Clerk", "hosting": "Vercel / AWS / GCP", }, pros=[ "Fast to build", "Low cost", "Easy to iterate", "Multi-tenant ready", ], cons=[ "Limited scalability", "Technical debt risk", "Feature constraints", ], estimated_cost="$50-200/month", estimated_setup_time="1-2 weeks", ), ArchitecturePattern( id="pattern_ecommerce", name="E-commerce Platform", description="Full-featured e-commerce with payments, inventory, order management, and product catalog.", use_cases=[ "Online store", "Marketplace", "Retail platform", "D2C brand", ], tags=["ecommerce", "payments", "marketplace", "retail", "shopping"], components=[ "Storefront", "Cart Service", "Payment Gateway", "Inventory", "Order Management", "Product Catalog", "Search", ], tech_stack={ "frontend": "Next.js / Shopify / MedusaJS", "backend": "Node.js / Python / Shopify API", "database": "PostgreSQL + Redis", "payments": "Stripe / PayPal / Adyen", "search": "Elasticsearch / Algolia", }, pros=["Complete solution", "PCI compliant", "Scalable", "SEO-friendly"], cons=[ "Complex", "Higher cost", "Regulatory requirements", "Security burden", ], estimated_cost="$500-2000/month", estimated_setup_time="2-3 months", ), ArchitecturePattern( id="pattern_realtime_app", name="Real-time Application", description="Application with real-time features: chat, notifications, live updates, collaborative editing.", use_cases=[ "Chat application", "Live dashboard", "Collaboration tools", "Gaming", "Trading platforms", ], tags=[ "realtime", "websocket", "chat", "live", "collaboration", "bi-directional", ], components=[ "WebSocket Server", "Redis Pub/Sub", "Real-time DB", "Push Notifications", "Presence Service", ], tech_stack={ "frontend": "React / Vue / Svelte", "backend": "Node.js / Socket.io / Pusher / Supabase Realtime", "realtime": "Socket.io / Pusher / Ably", "database": "PostgreSQL + Redis", "notifications": "Firebase FCM / OneSignal", }, pros=["Real-time", "Interactive", "Engaging", "Instant feedback"], cons=[ "Complex state management", "Connection limits", "Cost at scale", "Reconnection handling", ], estimated_cost="$200-1000/month", estimated_setup_time="4-8 weeks", ), ArchitecturePattern( id="pattern_api_first", name="API-First / BFF", description="Backend-for-Frontend pattern where each client has its own optimized API gateway. Great for multiple client types.", use_cases=[ "Mobile + Web apps", "Multiple client platforms", "Third-party API", "Team autonomy", ], tags=["api", "bff", "mobile", "backend", "graphql", "rest", "gateway"], components=[ "API Gateway", "BFF per client", "GraphQL Federation", "Rate Limiter", "API Documentation", ], tech_stack={ "bff": "GraphQL Gateway / Express / FastAPI", "api": "GraphQL / REST / tRPC", "gateway": "Kong / AWS API Gateway / Apollo Gateway", "documentation": "Swagger / Redoc / GraphQL Playground", }, pros=[ "Client-optimized", "Team autonomy", "Great developer experience", "Mobile-ready", ], cons=[ "Code duplication risk", "Gateway overhead", "Requires API design skills", ], estimated_cost="$200-800/month", estimated_setup_time="3-6 weeks", ), ArchitecturePattern( id="pattern_cdn_frontend", name="Static Site + CDN", description="Static content served directly from CDN. Best for content-focused sites with minimal dynamic functionality.", use_cases=[ "Marketing sites", "Blogs", "Documentation", "Portfolios", "Landing pages", ], tags=["static", "cdn", "jamstack", "content", "fast", "cheap"], components=[ "Static Files", "CDN", "Object Storage", "Forms (serverless)", ], tech_stack={ "frontend": "Next.js / Gatsby / Hugo / Astro", "hosting": "Vercel / Netlify / Cloudflare Pages", "cdn": "Cloudflare / Fastly", "forms": "Formspree / Netlify Forms", }, pros=["Extremely fast", "Cheap", "Secure", "Simple", "Global CDN"], cons=[ "Limited interactivity", "Build required for updates", "Not for complex apps", ], estimated_cost="$0-50/month", estimated_setup_time="1-3 days", ), ] seeded = 0 for pattern in default_patterns: # Check if pattern already exists existing = await self.get_pattern(pattern.id) if not existing and await self.store_pattern(pattern): seeded += 1 logger.info(f"Seeded {seeded} default architecture patterns") return seeded async def _seed_from_fallback(self) -> int: """ Load patterns from the local JSON fallback file when Redis is unavailable. Since Redis is unavailable, this simply counts the available patterns in the fallback file - they represent what *would* be seeded. Returns: Number of patterns found in the fallback file, or 0 on error. """ try: if not _FALLBACK_PATTERNS_PATH.exists(): logger.warning( "Fallback patterns file not found: %s", _FALLBACK_PATTERNS_PATH, ) return 0 with open(_FALLBACK_PATTERNS_PATH) as f: raw_patterns = json.load(f) count = len(raw_patterns) logger.info( "Seeded %d architecture patterns from fallback file: %s", count, _FALLBACK_PATTERNS_PATH, ) return count except Exception as e: logger.error("Failed to seed patterns from fallback file: %s", e) return 0 # Global instance _memory_instance: ArchitectureMemory | None = None def get_architecture_memory() -> ArchitectureMemory: """Get or create the global architecture memory instance.""" global _memory_instance if _memory_instance is None: _memory_instance = ArchitectureMemory() return _memory_instance