Samir Boulahtit
0984ff7d17
Merchants can now register domains (e.g., myloyaltyprogram.lu) that all their stores inherit. Individual stores can override with their own custom domain. Resolution priority: StoreDomain > MerchantDomain > subdomain. - Add MerchantDomain model, schema, service, and admin API endpoints - Add merchant domain fallback in platform and store context middleware - Add Merchant.primary_domain and Store.effective_domain properties - Add Alembic migration for merchant_domains table - Update loyalty user journey docs with subscription & domain setup flow - Add unit tests (50 passing) and integration tests (15 passing) Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
603 lines
22 KiB
Python
603 lines
22 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
|
|
from app.modules.tenancy.models import 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_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_api_request(request: Request) -> bool:
|
|
"""Check if request is for API endpoints."""
|
|
return FrontendDetector.is_api_request(request.url.path)
|
|
|
|
@staticmethod
|
|
def is_shop_api_request(request: Request) -> bool:
|
|
"""Check if request is for shop API endpoints."""
|
|
return request.url.path.startswith("/api/v1/shop/")
|
|
|
|
@staticmethod
|
|
def extract_store_from_referer(request: Request) -> dict | None:
|
|
"""
|
|
Extract store context from Referer header.
|
|
|
|
Used for shop API requests where store context comes from the page
|
|
that made the API call (e.g., JavaScript on /stores/wizamart/shop/products
|
|
calling /api/v1/shop/products).
|
|
|
|
Extracts store from Referer URL patterns:
|
|
- http://localhost:8000/stores/wizamart/shop/... → wizamart
|
|
- http://wizamart.platform.com/shop/... → wizamart (subdomain) # noqa
|
|
- http://custom-domain.com/shop/... → 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 shop 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/wizamart/shop/products → wizamart
|
|
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
|
|
# wizamart.platform.com → wizamart
|
|
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 (
|
|
StoreContextManager.is_admin_request(request)
|
|
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)
|
|
|
|
# Handle shop API routes specially - extract store from Referer header
|
|
if StoreContextManager.is_shop_api_request(request):
|
|
logger.debug(
|
|
f"[STORE] Shop API request detected: {request.url.path}",
|
|
extra={
|
|
"path": request.url.path,
|
|
"referer": request.headers.get("referer", ""),
|
|
},
|
|
)
|
|
|
|
store_context = StoreContextManager.extract_store_from_referer(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 = request.url.path
|
|
|
|
logger.debug(
|
|
"[STORE_CONTEXT] Store detected from Referer for shop API",
|
|
extra={
|
|
"store_id": store.id,
|
|
"store_name": store.name,
|
|
"store_subdomain": store.subdomain,
|
|
"detection_method": store_context.get(
|
|
"detection_method"
|
|
),
|
|
"api_path": request.url.path,
|
|
"referer": store_context.get("referer", ""),
|
|
},
|
|
)
|
|
else:
|
|
logger.warning(
|
|
"[WARNING] Store context from Referer but store not found",
|
|
extra={
|
|
"context": store_context,
|
|
"detection_method": store_context.get(
|
|
"detection_method"
|
|
),
|
|
"api_path": request.url.path,
|
|
},
|
|
)
|
|
request.state.store = None
|
|
request.state.store_context = store_context
|
|
request.state.clean_path = request.url.path
|
|
finally:
|
|
db.close()
|
|
else:
|
|
logger.warning(
|
|
"[STORE] Shop API request without Referer header",
|
|
extra={"path": request.url.path},
|
|
)
|
|
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 other 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-shop 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
|