# middleware/platform_context.py """ Platform Context Middleware Detects platform from host/domain/path and injects into request.state. This middleware runs BEFORE StoreContextMiddleware to establish platform context. Handles two routing modes: 1. Production: Domain-based (oms.lu, loyalty.lu → Platform detection) 2. Development: Path-based (localhost:9999/platforms/oms/*, localhost:9999/platforms/loyalty/*) URL Structure: - Main marketing site: localhost:9999/ (no platform prefix) → uses 'main' platform - Platform sites: localhost:9999/platforms/{code}/ → uses specific platform Also provides platform_clean_path for downstream middleware to use. """ import logging from fastapi import Request from sqlalchemy.orm import Session from app.core.database import get_db from app.core.frontend_detector import FrontendDetector from app.modules.enums import FrontendType from app.modules.tenancy.models import Platform # Note: We use pure ASGI middleware (not BaseHTTPMiddleware) to enable path rewriting logger = logging.getLogger(__name__) # Default platform code for main marketing site DEFAULT_PLATFORM_CODE = "main" class PlatformContextManager: """Manages platform context detection for multi-platform routing.""" @staticmethod def detect_platform_context(request: Request) -> dict | None: """ Detect platform context from request. Priority order: 1. Domain-based (production): oms.lu → platform code "oms" 2. Path-based (development): localhost:9999/platforms/oms/* → platform code "oms" 3. Default: localhost without /platforms/ prefix → 'main' platform (marketing site) URL Structure: - / → Main marketing site ('main' platform) - /about → Main marketing site about page - /platforms/oms/ → OMS platform homepage - /platforms/oms/pricing → OMS platform pricing - /platforms/loyalty/ → Loyalty platform homepage Returns dict with platform info or None if not detected. """ host = request.headers.get("host", "") path = request.url.path # Remove port from host if present (e.g., localhost:9999 -> localhost) host_without_port = host.split(":")[0] if ":" in host else host # Skip platform detection for admin routes - admin is global if FrontendDetector.is_admin(host, path): return None # Method 1: Domain-based detection (production) # Check if the host matches a known platform domain # This will be resolved in get_platform_from_context by DB lookup if host_without_port and host_without_port not in ["localhost", "127.0.0.1"]: # Could be a platform domain or a store subdomain/custom domain # Check if it's a known platform domain pattern # For now, assume non-localhost hosts that aren't subdomains are platform domains if "." in host_without_port: # This could be: # - Platform domain: oms.lu, loyalty.lu # - Store subdomain: store.oms.lu # - Custom domain: shop.mymerchant.com # We detect platform domain vs subdomain by checking if it's a root domain parts = host_without_port.split(".") if len(parts) == 2: # e.g., oms.lu (root domain) return { "domain": host_without_port, "detection_method": "domain", "host": host, "original_path": path, } # Method 2: Path-based detection (development) - ONLY for /platforms/ prefix # Check for path prefix like /platforms/oms/, /platforms/loyalty/ if path.startswith("/platforms/"): # Extract: /platforms/oms/pricing → code="oms", clean_path="/pricing" path_after_platforms = path[11:] # Remove "/platforms/" parts = path_after_platforms.split("/", 1) platform_code = parts[0].lower() if platform_code: clean_path = "/" + parts[1] if len(parts) > 1 and parts[1] else "/" return { "path_prefix": platform_code, "detection_method": "path", "host": host, "original_path": path, "clean_path": clean_path, } # Method 3: Default platform for localhost without /platforms/ prefix # This serves the main marketing site # Store routes require explicit platform via /platforms/{code}/store/... if host_without_port in ["localhost", "127.0.0.1"]: if path.startswith(("/store/", "/stores/")): return None # No platform — handlers will show appropriate error return { "path_prefix": DEFAULT_PLATFORM_CODE, "detection_method": "default", "host": host, "original_path": path, "clean_path": path, # No path rewrite for main site } return None @staticmethod def get_platform_from_context(db: Session, context: dict) -> Platform | None: """ Get platform from database using context information. Supports: 1. Domain-based lookup (Platform.domain) 2. Path-prefix lookup (Platform.path_prefix) 3. Default lookup (Platform.code) """ if not context: return None platform = None # Method 1: Domain-based lookup if context.get("detection_method") == "domain": domain = context.get("domain") if domain: # Try Platform.domain first platform = ( db.query(Platform) .filter(Platform.domain == domain) .filter(Platform.is_active.is_(True)) .first() ) if platform: logger.debug( f"[PLATFORM] Platform found via domain: {domain} → {platform.name}" ) return platform # Fallback: Check StoreDomain for custom store domains from app.modules.tenancy.models import StoreDomain store_domain = ( db.query(StoreDomain) .filter(StoreDomain.domain == domain, StoreDomain.is_active.is_(True)) .first() ) if store_domain and store_domain.platform_id: platform = ( db.query(Platform) .filter(Platform.id == store_domain.platform_id, Platform.is_active.is_(True)) .first() ) if platform: logger.debug( f"[PLATFORM] Platform found via store domain: {domain} → {platform.name}" ) return platform logger.debug(f"[PLATFORM] No platform found for domain: {domain}") # Method 2: Path-prefix lookup if context.get("detection_method") == "path": path_prefix = context.get("path_prefix") if path_prefix: # First try path_prefix, then try code platform = ( db.query(Platform) .filter( (Platform.path_prefix == path_prefix) | (Platform.code == path_prefix) ) .filter(Platform.is_active.is_(True)) .first() ) if platform: logger.debug( f"[PLATFORM] Platform found via path prefix: {path_prefix} → {platform.name}" ) return platform logger.debug(f"[PLATFORM] No platform found for path prefix: {path_prefix}") # Method 3: Default lookup if context.get("detection_method") == "default": platform = ( db.query(Platform) .filter(Platform.code == DEFAULT_PLATFORM_CODE) .filter(Platform.is_active.is_(True)) .first() ) if platform: logger.debug( f"[PLATFORM] Default platform found: {platform.name}" ) return platform return None @staticmethod def extract_clean_path(request: Request, platform_context: dict | None) -> str: """ Extract clean path without platform prefix for routing. Downstream middleware (like StoreContextMiddleware) should use this clean path for their detection logic. """ if not platform_context: return request.url.path # For path-based detection, use the pre-computed clean path if platform_context.get("detection_method") == "path": return platform_context.get("clean_path", request.url.path) # For domain-based or default, path remains unchanged return request.url.path @staticmethod def is_admin_request(request: Request) -> bool: """ Check if request is for admin interface. DEPRECATED: Use FrontendDetector.is_admin() instead. Kept for backwards compatibility. """ host = request.headers.get("host", "") path = request.url.path return FrontendDetector.is_admin(host, path) @staticmethod def is_static_file_request(request: Request) -> bool: """Check if request is for static files.""" path = request.url.path.lower() static_extensions = ( ".ico", ".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".woff", ".woff2", ".ttf", ".eot", ".webp", ".map", ".json", ".xml", ".txt", ".pdf", ".webmanifest", ) static_paths = ("/static/", "/media/", "/assets/", "/.well-known/") if path.endswith(static_extensions): return True if any(path.startswith(static_path) for static_path in static_paths): return True return "favicon.ico" in path class PlatformContextMiddleware: """ ASGI Middleware to inject platform context and rewrite URL paths. This middleware: 1. Detects platform from domain (production) or path prefix (development) 2. Rewrites the URL path to remove platform prefix for routing 3. Stores platform info in request state for handlers Runs BEFORE StoreContextMiddleware to establish platform context. Sets in scope['state']: platform: Platform object platform_context: Detection metadata platform_clean_path: Path without platform prefix platform_original_path: Original path before rewrite """ def __init__(self, app): self.app = app async def __call__(self, scope, receive, send): """ASGI interface - allows path rewriting before routing.""" if scope["type"] != "http": await self.app(scope, receive, send) return # Initialize state dict if not present if "state" not in scope: scope["state"] = {} path = scope["path"] host = "" for header_name, header_value in scope.get("headers", []): if header_name == b"host": host = header_value.decode("utf-8") break # Skip for static files if self._is_static_file(path): scope["state"]["platform"] = None scope["state"]["platform_context"] = None scope["state"]["platform_clean_path"] = path scope["state"]["platform_original_path"] = path await self.app(scope, receive, send) return # Skip for system endpoints if path in ["/health", "/docs", "/redoc", "/openapi.json"]: scope["state"]["platform"] = None scope["state"]["platform_context"] = None scope["state"]["platform_clean_path"] = path scope["state"]["platform_original_path"] = path await self.app(scope, receive, send) return # Skip for admin requests if FrontendDetector.is_admin(host, path): scope["state"]["platform"] = None scope["state"]["platform_context"] = None scope["state"]["platform_clean_path"] = path scope["state"]["platform_original_path"] = path await self.app(scope, receive, send) return # Detect platform context platform_context = self._detect_platform_context(path, host) if platform_context: db_gen = get_db() db = next(db_gen) try: platform = PlatformContextManager.get_platform_from_context( db, platform_context ) if platform: clean_path = platform_context.get("clean_path", path) # Store in scope state scope["state"]["platform"] = platform scope["state"]["platform_context"] = platform_context scope["state"]["platform_clean_path"] = clean_path scope["state"]["platform_original_path"] = path # REWRITE THE PATH for routing # This is the key: FastAPI will route based on this rewritten path if platform_context.get("detection_method") == "path": scope["path"] = clean_path # Also update raw_path if present if "raw_path" in scope: scope["raw_path"] = clean_path.encode("utf-8") logger.debug( f"[PLATFORM] Detected: {platform.code}, " f"original={path}, routed={scope['path']}" ) else: # Platform code not found in database scope["state"]["platform"] = None scope["state"]["platform_context"] = None scope["state"]["platform_clean_path"] = path scope["state"]["platform_original_path"] = path finally: db.close() else: scope["state"]["platform"] = None scope["state"]["platform_context"] = None scope["state"]["platform_clean_path"] = path scope["state"]["platform_original_path"] = path await self.app(scope, receive, send) def _detect_platform_context(self, path: str, host: str) -> dict | None: """ Detect platform from path or host. URL Structure: - / → Main marketing site ('main' platform) - /about → Main marketing site about page - /platforms/oms/ → OMS platform homepage - /platforms/oms/pricing → OMS platform pricing - /platforms/loyalty/ → Loyalty platform homepage """ host_without_port = host.split(":")[0] if ":" in host else host # Method 1: Domain-based (production) if host_without_port and host_without_port not in ["localhost", "127.0.0.1"]: if "." in host_without_port: parts = host_without_port.split(".") if len(parts) == 2: # Root domain like oms.lu return { "domain": host_without_port, "detection_method": "domain", "host": host, "original_path": path, "clean_path": path, # No path rewrite for domain-based } # Method 2: Path-based (development) - ONLY for /platforms/ prefix if path.startswith("/platforms/"): # Extract: /platforms/oms/pricing → code="oms", clean_path="/pricing" path_after_platforms = path[11:] # Remove "/platforms/" parts = path_after_platforms.split("/", 1) platform_code = parts[0].lower() if platform_code: clean_path = "/" + parts[1] if len(parts) > 1 and parts[1] else "/" return { "path_prefix": platform_code, "detection_method": "path", "host": host, "original_path": path, "clean_path": clean_path, } # Method 3: Default for localhost - serves main marketing site # Store routes require explicit platform via /platforms/{code}/store/... if host_without_port in ["localhost", "127.0.0.1"]: if path.startswith(("/store/", "/stores/")): return None # No platform — handlers will show appropriate error return { "path_prefix": DEFAULT_PLATFORM_CODE, "detection_method": "default", "host": host, "original_path": path, "clean_path": path, # No path rewrite for main site } return None def _is_static_file(self, path: str) -> bool: """Check if path is for static files.""" path_lower = path.lower() static_extensions = ( ".ico", ".css", ".js", ".png", ".jpg", ".jpeg", ".gif", ".svg", ".woff", ".woff2", ".ttf", ".eot", ".webp", ".map", ".json", ".xml", ".txt", ".pdf", ".webmanifest", ) static_paths = ("/static/", "/media/", "/assets/", "/.well-known/") if path_lower.endswith(static_extensions): return True if any(path_lower.startswith(p) for p in static_paths): return True return "favicon.ico" in path_lower def _is_admin_request(self, path: str, host: str) -> bool: """ Check if request is for admin interface. DEPRECATED: Use FrontendDetector.is_admin() instead. Kept for backwards compatibility. """ return FrontendDetector.is_admin(host, path) def get_current_platform(request: Request) -> Platform | None: """Helper function to get current platform from request state.""" return getattr(request.state, "platform", None) def require_platform_context(): """Dependency to require platform context in endpoints.""" def dependency(request: Request): platform = get_current_platform(request) if not platform: from fastapi import HTTPException raise HTTPException( status_code=404, detail="Platform not found" ) return platform return dependency