Samir Boulahtit
e9253fbd84
Replace all ~1,086 occurrences of Wizamart/wizamart/WIZAMART/WizaMart with Orion/orion/ORION across 184 files. This includes database identifiers, email addresses, domain references, R2 bucket names, DNS prefixes, encryption salt, Celery app name, config defaults, Docker configs, CI configs, documentation, seed data, and templates. Renames homepage-wizamart.html template to homepage-orion.html. Fixes duplicate file_pattern key in api.yaml architecture rule. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
520 lines
19 KiB
Python
520 lines
19 KiB
Python
# middleware/store_context.py
|
|
"""
|
|
Store Context Middleware (Class-Based)
|
|
|
|
Detects store from host/domain/path and injects into request.state.
|
|
Handles three routing modes:
|
|
1. Custom domains (customdomain1.com → Store 1)
|
|
2. Subdomains (store1.platform.com → Store 1)
|
|
3. Path-based (/store/store1/ or /stores/store1/ → Store 1)
|
|
|
|
Also extracts clean_path for nested routing patterns.
|
|
|
|
IMPORTANT: This middleware runs AFTER PlatformContextMiddleware.
|
|
Uses request.state.platform_clean_path when available (set by PlatformContextMiddleware).
|
|
"""
|
|
|
|
import logging
|
|
|
|
from fastapi import Request
|
|
from sqlalchemy import func
|
|
from sqlalchemy.orm import Session
|
|
from starlette.middleware.base import BaseHTTPMiddleware
|
|
|
|
from app.core.config import settings
|
|
from app.core.database import get_db
|
|
from app.core.frontend_detector import FrontendDetector
|
|
from app.modules.tenancy.models import Store, StoreDomain
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class StoreContextManager:
|
|
"""Manages store context detection for multi-tenant routing."""
|
|
|
|
@staticmethod
|
|
def detect_store_context(request: Request) -> dict | None:
|
|
"""
|
|
Detect store context from request.
|
|
|
|
Priority order:
|
|
1. Custom domain (customdomain1.com)
|
|
2. Subdomain (store1.platform.com)
|
|
3. Path-based (/store/store1/ or /stores/store1/)
|
|
|
|
Uses platform_clean_path from PlatformContextMiddleware when available.
|
|
This path has the platform prefix stripped (e.g., /oms/stores/foo → /stores/foo).
|
|
|
|
Returns dict with store info or None if not found.
|
|
"""
|
|
host = request.headers.get("host", "")
|
|
# Use platform_clean_path if available (set by PlatformContextMiddleware)
|
|
path = getattr(request.state, "platform_clean_path", None) or request.url.path
|
|
|
|
# Remove port from host if present (e.g., localhost:8000 -> localhost)
|
|
if ":" in host:
|
|
host = host.split(":")[0]
|
|
|
|
# Method 1: Custom domain detection (HIGHEST PRIORITY)
|
|
# Check if this is a custom domain (not platform.com and not localhost)
|
|
platform_domain = getattr(settings, "platform_domain", "platform.com")
|
|
|
|
is_custom_domain = (
|
|
host
|
|
and not host.endswith(f".{platform_domain}")
|
|
and host != platform_domain
|
|
and host
|
|
not in ["localhost", "127.0.0.1", "admin.localhost", "admin.127.0.0.1"]
|
|
and not host.startswith("admin.")
|
|
)
|
|
|
|
if is_custom_domain:
|
|
normalized_domain = StoreDomain.normalize_domain(host)
|
|
return {
|
|
"domain": normalized_domain,
|
|
"detection_method": "custom_domain",
|
|
"host": host,
|
|
"original_host": request.headers.get("host", ""),
|
|
}
|
|
|
|
# Method 2: Subdomain detection (store1.platform.com)
|
|
if "." in host:
|
|
parts = host.split(".")
|
|
# Check if it's a valid subdomain (not www, admin, api)
|
|
if len(parts) >= 2 and parts[0] not in ["www", "admin", "api"]:
|
|
subdomain = parts[0]
|
|
return {
|
|
"subdomain": subdomain,
|
|
"detection_method": "subdomain",
|
|
"host": host,
|
|
}
|
|
|
|
# Method 3: Path-based detection (/store/storename/ or /stores/storename/)
|
|
# Support BOTH patterns for flexibility
|
|
if path.startswith(("/store/", "/stores/")):
|
|
# Determine which pattern
|
|
if path.startswith("/stores/"):
|
|
prefix_len = len("/stores/")
|
|
else:
|
|
prefix_len = len("/store/")
|
|
|
|
path_parts = path[prefix_len:].split("/")
|
|
if len(path_parts) >= 1 and path_parts[0]:
|
|
store_code = path_parts[0]
|
|
return {
|
|
"subdomain": store_code,
|
|
"detection_method": "path",
|
|
"path_prefix": path[: prefix_len + len(store_code)],
|
|
"full_prefix": path[:prefix_len], # /store/ or /stores/
|
|
"host": host,
|
|
}
|
|
|
|
return None
|
|
|
|
@staticmethod
|
|
def get_store_from_context(db: Session, context: dict) -> Store | None:
|
|
"""
|
|
Get store from database using context information.
|
|
|
|
Supports three methods:
|
|
1. Custom domain lookup (StoreDomain table)
|
|
2. Subdomain lookup (Store.subdomain)
|
|
3. Path-based lookup (Store.subdomain)
|
|
"""
|
|
if not context:
|
|
return None
|
|
|
|
store = None
|
|
|
|
# Method 1: Custom domain lookup
|
|
if context.get("detection_method") == "custom_domain":
|
|
domain = context.get("domain")
|
|
if domain:
|
|
store_domain = (
|
|
db.query(StoreDomain)
|
|
.filter(StoreDomain.domain == domain)
|
|
.filter(StoreDomain.is_active.is_(True))
|
|
.filter(StoreDomain.is_verified.is_(True))
|
|
.first()
|
|
)
|
|
|
|
if store_domain:
|
|
store = store_domain.store
|
|
if not store or not store.is_active:
|
|
logger.warning(f"Store for domain {domain} is not active")
|
|
return None
|
|
|
|
logger.info(
|
|
f"[OK] Store found via custom domain: {domain} → {store.name}"
|
|
)
|
|
return store
|
|
|
|
# Fallback: Try merchant-level domain
|
|
from app.modules.tenancy.models.merchant_domain import MerchantDomain
|
|
merchant_domain = (
|
|
db.query(MerchantDomain)
|
|
.filter(
|
|
MerchantDomain.domain == domain,
|
|
MerchantDomain.is_active.is_(True),
|
|
MerchantDomain.is_verified.is_(True),
|
|
)
|
|
.first()
|
|
)
|
|
if merchant_domain:
|
|
store = (
|
|
db.query(Store)
|
|
.filter(
|
|
Store.merchant_id == merchant_domain.merchant_id,
|
|
Store.is_active.is_(True),
|
|
)
|
|
.order_by(Store.id)
|
|
.first()
|
|
)
|
|
if store:
|
|
context["merchant_domain"] = True
|
|
context["merchant_id"] = merchant_domain.merchant_id
|
|
logger.info(
|
|
f"[OK] Store found via merchant domain: {domain} → {store.name}"
|
|
)
|
|
return store
|
|
|
|
logger.warning(f"No active store found for custom domain: {domain}")
|
|
return None
|
|
|
|
# Method 2 & 3: Subdomain or path-based lookup
|
|
if "subdomain" in context:
|
|
subdomain = context["subdomain"]
|
|
store = (
|
|
db.query(Store)
|
|
.filter(func.lower(Store.subdomain) == subdomain.lower())
|
|
.filter(Store.is_active.is_(True))
|
|
.first()
|
|
)
|
|
|
|
if store:
|
|
method = context.get("detection_method", "unknown")
|
|
logger.info(
|
|
f"[OK] Store found via {method}: {subdomain} → {store.name}"
|
|
)
|
|
else:
|
|
logger.warning(f"No active store found for subdomain: {subdomain}")
|
|
|
|
return store
|
|
|
|
@staticmethod
|
|
def extract_clean_path(request: Request, store_context: dict | None) -> str:
|
|
"""
|
|
Extract clean path without store prefix for routing.
|
|
|
|
Supports both /store/ and /stores/ prefixes.
|
|
"""
|
|
if not store_context:
|
|
return request.url.path
|
|
|
|
# Only strip path prefix for path-based detection
|
|
if store_context.get("detection_method") == "path":
|
|
path = request.url.path
|
|
path_prefix = store_context.get("path_prefix", "")
|
|
|
|
if path.startswith(path_prefix):
|
|
clean_path = path[len(path_prefix) :]
|
|
return clean_path if clean_path else "/"
|
|
|
|
return request.url.path
|
|
|
|
@staticmethod
|
|
def is_api_request(request: Request) -> bool:
|
|
"""Check if request is for API endpoints."""
|
|
return FrontendDetector.is_api_request(request.url.path)
|
|
|
|
@staticmethod
|
|
def extract_store_from_referer(request: Request) -> dict | None:
|
|
"""
|
|
Extract store context from Referer header.
|
|
|
|
Used for storefront API requests where store context comes from the page
|
|
that made the API call (e.g., JavaScript on /stores/orion/storefront/products
|
|
calling /api/v1/storefront/products).
|
|
|
|
Extracts store from Referer URL patterns:
|
|
- http://localhost:8000/stores/orion/storefront/... → orion
|
|
- http://orion.platform.com/storefront/... → orion (subdomain) # noqa
|
|
- http://custom-domain.com/storefront/... → custom-domain.com # noqa
|
|
|
|
Returns store context dict or None if unable to extract.
|
|
"""
|
|
referer = request.headers.get("referer") or request.headers.get("origin")
|
|
|
|
if not referer:
|
|
logger.debug("[STORE] No Referer/Origin header for storefront API request")
|
|
return None
|
|
|
|
try:
|
|
from urllib.parse import urlparse
|
|
|
|
parsed = urlparse(referer)
|
|
referer_host = parsed.hostname or ""
|
|
referer_path = parsed.path or ""
|
|
|
|
# Remove port from host
|
|
if ":" in referer_host:
|
|
referer_host = referer_host.split(":")[0]
|
|
|
|
logger.debug(
|
|
"[STORE] Extracting store from Referer",
|
|
extra={
|
|
"referer": referer,
|
|
"referer_host": referer_host,
|
|
"referer_path": referer_path,
|
|
},
|
|
)
|
|
|
|
# Method 1: Path-based detection from referer path
|
|
# /stores/orion/storefront/products → orion
|
|
if referer_path.startswith(("/stores/", "/store/")):
|
|
prefix = (
|
|
"/stores/" if referer_path.startswith("/stores/") else "/store/"
|
|
)
|
|
path_parts = referer_path[len(prefix) :].split("/")
|
|
if len(path_parts) >= 1 and path_parts[0]:
|
|
store_code = path_parts[0]
|
|
prefix_len = len(prefix)
|
|
logger.debug(
|
|
f"[STORE] Extracted store from Referer path: {store_code}",
|
|
extra={"store_code": store_code, "method": "referer_path"},
|
|
)
|
|
# Use "path" as detection_method to be consistent with direct path detection
|
|
# This allows cookie path logic to work the same way
|
|
return {
|
|
"subdomain": store_code,
|
|
"detection_method": "path", # Consistent with direct path detection
|
|
"path_prefix": referer_path[
|
|
: prefix_len + len(store_code)
|
|
], # /store/store1
|
|
"full_prefix": prefix, # /store/ or /stores/
|
|
"host": referer_host,
|
|
"referer": referer,
|
|
}
|
|
|
|
# Method 2: Subdomain detection from referer host
|
|
# orion.platform.com → orion
|
|
platform_domain = getattr(settings, "platform_domain", "platform.com")
|
|
if "." in referer_host:
|
|
parts = referer_host.split(".")
|
|
if len(parts) >= 2 and parts[0] not in ["www", "admin", "api"]:
|
|
# Check if it's a subdomain of platform domain
|
|
if referer_host.endswith(f".{platform_domain}"):
|
|
subdomain = parts[0]
|
|
logger.debug(
|
|
f"[STORE] Extracted store from Referer subdomain: {subdomain}",
|
|
extra={
|
|
"subdomain": subdomain,
|
|
"method": "referer_subdomain",
|
|
},
|
|
)
|
|
return {
|
|
"subdomain": subdomain,
|
|
"detection_method": "referer_subdomain",
|
|
"host": referer_host,
|
|
"referer": referer,
|
|
}
|
|
|
|
# Method 3: Custom domain detection from referer host
|
|
# custom-shop.com → custom-shop.com
|
|
is_custom_domain = (
|
|
referer_host
|
|
and not referer_host.endswith(f".{platform_domain}")
|
|
and referer_host != platform_domain
|
|
and referer_host not in ["localhost", "127.0.0.1"]
|
|
and not referer_host.startswith("admin.")
|
|
)
|
|
|
|
if is_custom_domain:
|
|
from app.modules.tenancy.models import StoreDomain
|
|
|
|
normalized_domain = StoreDomain.normalize_domain(referer_host)
|
|
logger.debug(
|
|
f"[STORE] Extracted store from Referer custom domain: {normalized_domain}",
|
|
extra={
|
|
"domain": normalized_domain,
|
|
"method": "referer_custom_domain",
|
|
},
|
|
)
|
|
return {
|
|
"domain": normalized_domain,
|
|
"detection_method": "referer_custom_domain",
|
|
"host": referer_host,
|
|
"referer": referer,
|
|
}
|
|
|
|
except Exception as e:
|
|
logger.warning(
|
|
f"[STORE] Failed to extract store from Referer: {e}",
|
|
extra={"referer": referer, "error": str(e)},
|
|
)
|
|
|
|
return None
|
|
|
|
@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 StoreContextMiddleware(BaseHTTPMiddleware):
|
|
"""
|
|
Middleware to inject store context into request state.
|
|
|
|
Class-based middleware provides:
|
|
- Better state management
|
|
- Easier testing
|
|
- More organized code
|
|
- Standard ASGI pattern
|
|
|
|
Runs AFTER PlatformContextMiddleware in the request chain.
|
|
Uses request.state.platform_clean_path for path-based store detection.
|
|
|
|
Sets:
|
|
request.state.store: Store object
|
|
request.state.store_context: Detection metadata
|
|
request.state.clean_path: Path without store prefix
|
|
"""
|
|
|
|
async def dispatch(self, request: Request, call_next):
|
|
"""
|
|
Detect and inject store context.
|
|
"""
|
|
# Skip store detection for admin, static files, and system requests
|
|
if (
|
|
FrontendDetector.is_admin(request.headers.get("host", ""), request.url.path)
|
|
or StoreContextManager.is_static_file_request(request)
|
|
or request.url.path in ["/", "/health", "/docs", "/redoc", "/openapi.json"]
|
|
):
|
|
logger.debug(
|
|
f"[STORE] Skipping store detection: {request.url.path}",
|
|
extra={"path": request.url.path, "reason": "admin/static/system"},
|
|
)
|
|
request.state.store = None
|
|
request.state.store_context = None
|
|
request.state.clean_path = request.url.path
|
|
return await call_next(request)
|
|
|
|
# Skip store detection for API routes (admin API, store API have store_id in URL)
|
|
if StoreContextManager.is_api_request(request):
|
|
logger.debug(
|
|
f"[STORE] Skipping store detection for non-storefront API: {request.url.path}",
|
|
extra={"path": request.url.path, "reason": "api"},
|
|
)
|
|
request.state.store = None
|
|
request.state.store_context = None
|
|
request.state.clean_path = request.url.path
|
|
return await call_next(request)
|
|
|
|
# Detect store context
|
|
store_context = StoreContextManager.detect_store_context(request)
|
|
|
|
if store_context:
|
|
db_gen = get_db()
|
|
db = next(db_gen)
|
|
try:
|
|
store = StoreContextManager.get_store_from_context(
|
|
db, store_context
|
|
)
|
|
|
|
if store:
|
|
request.state.store = store
|
|
request.state.store_context = store_context
|
|
request.state.clean_path = StoreContextManager.extract_clean_path(
|
|
request, store_context
|
|
)
|
|
|
|
logger.debug(
|
|
"[STORE_CONTEXT] Store detected",
|
|
extra={
|
|
"store_id": store.id,
|
|
"store_name": store.name,
|
|
"store_subdomain": store.subdomain,
|
|
"detection_method": store_context.get("detection_method"),
|
|
"original_path": request.url.path,
|
|
"clean_path": request.state.clean_path,
|
|
},
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"[WARNING] Store context detected but store not found",
|
|
extra={
|
|
"context": store_context,
|
|
"detection_method": store_context.get("detection_method"),
|
|
},
|
|
)
|
|
request.state.store = None
|
|
request.state.store_context = store_context
|
|
request.state.clean_path = request.url.path
|
|
finally:
|
|
db.close()
|
|
else:
|
|
logger.debug(
|
|
"[STORE] No store context detected",
|
|
extra={
|
|
"path": request.url.path,
|
|
"host": request.headers.get("host", ""),
|
|
},
|
|
)
|
|
request.state.store = None
|
|
request.state.store_context = None
|
|
request.state.clean_path = request.url.path
|
|
|
|
# Continue to next middleware
|
|
return await call_next(request)
|
|
|
|
|
|
def get_current_store(request: Request) -> Store | None:
|
|
"""Helper function to get current store from request state."""
|
|
return getattr(request.state, "store", None)
|
|
|
|
|
|
def require_store_context():
|
|
"""Dependency to require store context in endpoints."""
|
|
|
|
def dependency(request: Request):
|
|
store = get_current_store(request)
|
|
if not store:
|
|
from app.modules.tenancy.exceptions import StoreNotFoundException
|
|
|
|
raise StoreNotFoundException("unknown", identifier_type="context")
|
|
return store
|
|
|
|
return dependency
|