Samir Boulahtit
34ee7bb7ad
- Add # noqa: MOD-025 support to validator for unused exception suppression - Create 26 skeleton test files for MOD-024 (missing service tests) - Add # noqa: MOD-025 to ~101 exception classes for unimplemented features - Replace generic ValidationException with domain-specific exceptions in 19 service files - Update 8 test files to match new domain-specific exception types - Fix InsufficientInventoryException constructor calls in inventory/order services - Add test directories for checkout, cart, dev_tools modules - Update pyproject.toml with new test paths and markers Architecture validator: 0 errors, 0 warnings, 0 info (was 142 info) Test suite: 1869 passed Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
453 lines
15 KiB
Python
453 lines
15 KiB
Python
# app/modules/tenancy/services/merchant_domain_service.py
|
|
"""
|
|
Merchant domain service for managing merchant-level custom domain operations.
|
|
|
|
This module provides classes and functions for:
|
|
- Adding and removing merchant domains
|
|
- Domain verification via DNS
|
|
- Domain activation and deactivation
|
|
- Setting primary domains
|
|
- Domain validation and normalization
|
|
|
|
Follows the same pattern as StoreDomainService.
|
|
"""
|
|
|
|
import logging
|
|
import secrets
|
|
from datetime import UTC, datetime
|
|
|
|
import dns.resolver
|
|
from sqlalchemy.exc import SQLAlchemyError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.modules.tenancy.exceptions import (
|
|
DNSVerificationException,
|
|
DomainAlreadyVerifiedException,
|
|
DomainNotVerifiedException,
|
|
DomainVerificationFailedException,
|
|
InvalidDomainFormatException,
|
|
MaxDomainsReachedException,
|
|
MerchantDomainAlreadyExistsException,
|
|
MerchantDomainNotFoundException,
|
|
MerchantNotFoundException,
|
|
MerchantValidationException,
|
|
ReservedDomainException,
|
|
)
|
|
from app.modules.tenancy.models import Merchant
|
|
from app.modules.tenancy.models.merchant_domain import MerchantDomain
|
|
from app.modules.tenancy.models.store_domain import StoreDomain
|
|
from app.modules.tenancy.schemas.merchant_domain import (
|
|
MerchantDomainCreate,
|
|
MerchantDomainUpdate,
|
|
)
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class MerchantDomainService:
|
|
"""Service class for merchant domain operations."""
|
|
|
|
def __init__(self):
|
|
self.max_domains_per_merchant = 5
|
|
self.reserved_subdomains = [
|
|
"www",
|
|
"admin",
|
|
"api",
|
|
"mail",
|
|
"smtp",
|
|
"ftp",
|
|
"cpanel",
|
|
"webmail",
|
|
]
|
|
|
|
def add_domain(
|
|
self, db: Session, merchant_id: int, domain_data: MerchantDomainCreate
|
|
) -> MerchantDomain:
|
|
"""
|
|
Add a custom domain to a merchant.
|
|
|
|
Args:
|
|
db: Database session
|
|
merchant_id: Merchant ID to add domain to
|
|
domain_data: Domain creation data
|
|
|
|
Returns:
|
|
Created MerchantDomain object
|
|
|
|
Raises:
|
|
MerchantNotFoundException: If merchant not found
|
|
MerchantDomainAlreadyExistsException: If domain already registered
|
|
MaxDomainsReachedException: If merchant has reached max domains
|
|
InvalidDomainFormatException: If domain format is invalid
|
|
"""
|
|
try:
|
|
# Verify merchant exists
|
|
self._get_merchant_by_id_or_raise(db, merchant_id)
|
|
|
|
# Check domain limit
|
|
self._check_domain_limit(db, merchant_id)
|
|
|
|
# Normalize domain
|
|
normalized_domain = MerchantDomain.normalize_domain(domain_data.domain)
|
|
|
|
# Validate domain format
|
|
self._validate_domain_format(normalized_domain)
|
|
|
|
# Check if domain already exists (in both StoreDomain and MerchantDomain)
|
|
if self._domain_exists_globally(db, normalized_domain):
|
|
raise MerchantDomainAlreadyExistsException(normalized_domain)
|
|
|
|
# If setting as primary, unset other primary domains
|
|
if domain_data.is_primary:
|
|
self._unset_primary_domains(db, merchant_id)
|
|
|
|
# Resolve platform_id: use provided value, or auto-resolve from merchant's primary StorePlatform
|
|
platform_id = domain_data.platform_id
|
|
if not platform_id:
|
|
from app.modules.tenancy.models import Store, StorePlatform
|
|
# Get platform from merchant's first store's primary StorePlatform
|
|
store_ids = (
|
|
db.query(Store.id)
|
|
.filter(Store.merchant_id == merchant_id)
|
|
.subquery()
|
|
)
|
|
primary_sp = (
|
|
db.query(StorePlatform)
|
|
.filter(
|
|
StorePlatform.store_id.in_(store_ids),
|
|
StorePlatform.is_primary.is_(True),
|
|
)
|
|
.first()
|
|
)
|
|
platform_id = primary_sp.platform_id if primary_sp else None
|
|
|
|
# Create domain record
|
|
new_domain = MerchantDomain(
|
|
merchant_id=merchant_id,
|
|
domain=normalized_domain,
|
|
is_primary=domain_data.is_primary,
|
|
platform_id=platform_id,
|
|
verification_token=secrets.token_urlsafe(32),
|
|
is_verified=False,
|
|
is_active=False,
|
|
ssl_status="pending",
|
|
)
|
|
|
|
db.add(new_domain)
|
|
db.flush()
|
|
db.refresh(new_domain)
|
|
|
|
logger.info(f"Domain {normalized_domain} added to merchant {merchant_id}")
|
|
return new_domain
|
|
|
|
except (
|
|
MerchantNotFoundException,
|
|
MerchantDomainAlreadyExistsException,
|
|
MaxDomainsReachedException,
|
|
InvalidDomainFormatException,
|
|
ReservedDomainException,
|
|
):
|
|
raise
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error adding merchant domain: {str(e)}")
|
|
raise MerchantValidationException("Failed to add merchant domain")
|
|
|
|
def get_merchant_domains(
|
|
self, db: Session, merchant_id: int
|
|
) -> list[MerchantDomain]:
|
|
"""
|
|
Get all domains for a merchant.
|
|
|
|
Args:
|
|
db: Database session
|
|
merchant_id: Merchant ID
|
|
|
|
Returns:
|
|
List of MerchantDomain objects
|
|
"""
|
|
try:
|
|
self._get_merchant_by_id_or_raise(db, merchant_id)
|
|
|
|
domains = (
|
|
db.query(MerchantDomain)
|
|
.filter(MerchantDomain.merchant_id == merchant_id)
|
|
.order_by(
|
|
MerchantDomain.is_primary.desc(),
|
|
MerchantDomain.created_at.desc(),
|
|
)
|
|
.all()
|
|
)
|
|
|
|
return domains
|
|
|
|
except MerchantNotFoundException:
|
|
raise
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error getting merchant domains: {str(e)}")
|
|
raise MerchantValidationException("Failed to retrieve merchant domains")
|
|
|
|
def get_domain_by_id(self, db: Session, domain_id: int) -> MerchantDomain:
|
|
"""
|
|
Get merchant domain by ID.
|
|
|
|
Args:
|
|
db: Database session
|
|
domain_id: Domain ID
|
|
|
|
Returns:
|
|
MerchantDomain object
|
|
|
|
Raises:
|
|
MerchantDomainNotFoundException: If domain not found
|
|
"""
|
|
domain = (
|
|
db.query(MerchantDomain)
|
|
.filter(MerchantDomain.id == domain_id)
|
|
.first()
|
|
)
|
|
if not domain:
|
|
raise MerchantDomainNotFoundException(str(domain_id))
|
|
return domain
|
|
|
|
def update_domain(
|
|
self, db: Session, domain_id: int, domain_update: MerchantDomainUpdate
|
|
) -> MerchantDomain:
|
|
"""
|
|
Update merchant domain settings.
|
|
|
|
Args:
|
|
db: Database session
|
|
domain_id: Domain ID
|
|
domain_update: Update data
|
|
|
|
Returns:
|
|
Updated MerchantDomain object
|
|
|
|
Raises:
|
|
MerchantDomainNotFoundException: If domain not found
|
|
DomainNotVerifiedException: If trying to activate unverified domain
|
|
"""
|
|
try:
|
|
domain = self.get_domain_by_id(db, domain_id)
|
|
|
|
# If setting as primary, unset other primary domains
|
|
if domain_update.is_primary:
|
|
self._unset_primary_domains(
|
|
db, domain.merchant_id, exclude_domain_id=domain_id
|
|
)
|
|
domain.is_primary = True
|
|
|
|
# If activating, check verification
|
|
if domain_update.is_active is True and not domain.is_verified:
|
|
raise DomainNotVerifiedException(domain_id, domain.domain)
|
|
|
|
# Update fields
|
|
if domain_update.is_active is not None:
|
|
domain.is_active = domain_update.is_active
|
|
|
|
db.flush()
|
|
db.refresh(domain)
|
|
|
|
logger.info(f"Merchant domain {domain.domain} updated")
|
|
return domain
|
|
|
|
except (MerchantDomainNotFoundException, DomainNotVerifiedException):
|
|
raise
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error updating merchant domain: {str(e)}")
|
|
raise MerchantValidationException("Failed to update merchant domain")
|
|
|
|
def delete_domain(self, db: Session, domain_id: int) -> str:
|
|
"""
|
|
Delete a merchant domain.
|
|
|
|
Args:
|
|
db: Database session
|
|
domain_id: Domain ID
|
|
|
|
Returns:
|
|
Success message
|
|
"""
|
|
try:
|
|
domain = self.get_domain_by_id(db, domain_id)
|
|
domain_name = domain.domain
|
|
merchant_id = domain.merchant_id
|
|
|
|
db.delete(domain)
|
|
|
|
logger.info(
|
|
f"Domain {domain_name} deleted from merchant {merchant_id}"
|
|
)
|
|
return f"Domain {domain_name} deleted successfully"
|
|
|
|
except MerchantDomainNotFoundException:
|
|
raise
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error deleting merchant domain: {str(e)}")
|
|
raise MerchantValidationException("Failed to delete merchant domain")
|
|
|
|
def verify_domain(
|
|
self, db: Session, domain_id: int
|
|
) -> tuple[MerchantDomain, str]:
|
|
"""
|
|
Verify merchant domain ownership via DNS TXT record.
|
|
|
|
The merchant must add a TXT record:
|
|
Name: _wizamart-verify.{domain}
|
|
Value: {verification_token}
|
|
"""
|
|
try:
|
|
domain = self.get_domain_by_id(db, domain_id)
|
|
|
|
if domain.is_verified:
|
|
raise DomainAlreadyVerifiedException(domain_id, domain.domain)
|
|
|
|
try:
|
|
txt_records = dns.resolver.resolve(
|
|
f"_wizamart-verify.{domain.domain}", "TXT"
|
|
)
|
|
|
|
for txt in txt_records:
|
|
txt_value = txt.to_text().strip('"')
|
|
if txt_value == domain.verification_token:
|
|
domain.is_verified = True
|
|
domain.verified_at = datetime.now(UTC)
|
|
db.flush()
|
|
db.refresh(domain)
|
|
|
|
logger.info(
|
|
f"Merchant domain {domain.domain} verified successfully"
|
|
)
|
|
return (
|
|
domain,
|
|
f"Domain {domain.domain} verified successfully",
|
|
)
|
|
|
|
raise DomainVerificationFailedException(
|
|
domain.domain,
|
|
"Verification token not found in DNS records",
|
|
)
|
|
|
|
except dns.resolver.NXDOMAIN:
|
|
raise DomainVerificationFailedException(
|
|
domain.domain,
|
|
f"DNS record _wizamart-verify.{domain.domain} not found",
|
|
)
|
|
except dns.resolver.NoAnswer:
|
|
raise DomainVerificationFailedException(
|
|
domain.domain, "No TXT records found for verification"
|
|
)
|
|
except DomainVerificationFailedException:
|
|
raise
|
|
except dns.resolver.DNSException as dns_error:
|
|
raise DNSVerificationException(domain.domain, str(dns_error))
|
|
|
|
except (
|
|
MerchantDomainNotFoundException,
|
|
DomainAlreadyVerifiedException,
|
|
DomainVerificationFailedException,
|
|
DNSVerificationException,
|
|
):
|
|
raise
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"Error verifying merchant domain: {str(e)}")
|
|
raise MerchantValidationException("Failed to verify merchant domain")
|
|
|
|
def get_verification_instructions(self, db: Session, domain_id: int) -> dict:
|
|
"""Get DNS verification instructions for a merchant domain."""
|
|
domain = self.get_domain_by_id(db, domain_id)
|
|
|
|
return {
|
|
"domain": domain.domain,
|
|
"verification_token": domain.verification_token,
|
|
"instructions": {
|
|
"step1": "Go to your domain's DNS settings (at your domain registrar)",
|
|
"step2": "Add a new TXT record with the following values:",
|
|
"step3": "Wait for DNS propagation (5-15 minutes)",
|
|
"step4": "Click 'Verify Domain' button in admin panel",
|
|
},
|
|
"txt_record": {
|
|
"type": "TXT",
|
|
"name": "_wizamart-verify",
|
|
"value": domain.verification_token,
|
|
"ttl": 3600,
|
|
},
|
|
"common_registrars": {
|
|
"Cloudflare": "https://dash.cloudflare.com",
|
|
"GoDaddy": "https://dcc.godaddy.com/manage/dns",
|
|
"Namecheap": "https://www.namecheap.com/myaccount/domain-list/",
|
|
"Google Domains": "https://domains.google.com",
|
|
},
|
|
}
|
|
|
|
# Private helper methods
|
|
def _get_merchant_by_id_or_raise(
|
|
self, db: Session, merchant_id: int
|
|
) -> Merchant:
|
|
"""Get merchant by ID or raise exception."""
|
|
merchant = (
|
|
db.query(Merchant).filter(Merchant.id == merchant_id).first()
|
|
)
|
|
if not merchant:
|
|
raise MerchantNotFoundException(merchant_id, identifier_type="id")
|
|
return merchant
|
|
|
|
def _check_domain_limit(self, db: Session, merchant_id: int) -> None:
|
|
"""Check if merchant has reached maximum domain limit."""
|
|
domain_count = (
|
|
db.query(MerchantDomain)
|
|
.filter(MerchantDomain.merchant_id == merchant_id)
|
|
.count()
|
|
)
|
|
|
|
if domain_count >= self.max_domains_per_merchant:
|
|
raise MaxDomainsReachedException(
|
|
merchant_id, self.max_domains_per_merchant
|
|
)
|
|
|
|
def _domain_exists_globally(self, db: Session, domain: str) -> bool:
|
|
"""Check if domain already exists in system (StoreDomain or MerchantDomain)."""
|
|
store_exists = (
|
|
db.query(StoreDomain)
|
|
.filter(StoreDomain.domain == domain)
|
|
.first()
|
|
is not None
|
|
)
|
|
if store_exists:
|
|
return True
|
|
|
|
merchant_exists = (
|
|
db.query(MerchantDomain)
|
|
.filter(MerchantDomain.domain == domain)
|
|
.first()
|
|
is not None
|
|
)
|
|
return merchant_exists
|
|
|
|
def _validate_domain_format(self, domain: str) -> None:
|
|
"""Validate domain format and check for reserved subdomains."""
|
|
first_part = domain.split(".")[0]
|
|
if first_part in self.reserved_subdomains:
|
|
raise ReservedDomainException(domain, first_part)
|
|
|
|
def _unset_primary_domains(
|
|
self,
|
|
db: Session,
|
|
merchant_id: int,
|
|
exclude_domain_id: int | None = None,
|
|
) -> None:
|
|
"""Unset all primary domains for merchant."""
|
|
query = db.query(MerchantDomain).filter(
|
|
MerchantDomain.merchant_id == merchant_id,
|
|
MerchantDomain.is_primary == True,
|
|
)
|
|
|
|
if exclude_domain_id:
|
|
query = query.filter(MerchantDomain.id != exclude_domain_id)
|
|
|
|
query.update({"is_primary": False})
|
|
|
|
|
|
# Create service instance
|
|
merchant_domain_service = MerchantDomainService()
|