# app/modules/tenancy/services/store_domain_service.py """ Store domain service for managing custom domain operations. This module provides classes and functions for: - Adding and removing custom domains - Domain verification via DNS - Domain activation and deactivation - Setting primary domains - Domain validation and normalization """ import logging import secrets from datetime import UTC, datetime import dns.resolver from sqlalchemy.exc import SQLAlchemyError from sqlalchemy.orm import Session from app.exceptions import ValidationException from app.modules.tenancy.exceptions import ( DNSVerificationException, DomainAlreadyVerifiedException, DomainNotVerifiedException, DomainVerificationFailedException, InvalidDomainFormatException, MaxDomainsReachedException, ReservedDomainException, StoreDomainAlreadyExistsException, StoreDomainNotFoundException, StoreNotFoundException, ) from app.modules.tenancy.models import Store, StoreDomain from app.modules.tenancy.schemas.store_domain import ( StoreDomainCreate, StoreDomainUpdate, ) logger = logging.getLogger(__name__) class StoreDomainService: """Service class for store domain operations.""" def __init__(self): self.max_domains_per_store = 10 # Configure as needed self.reserved_subdomains = [ "www", "admin", "api", "mail", "smtp", "ftp", "cpanel", "webmail", ] def add_domain( self, db: Session, store_id: int, domain_data: StoreDomainCreate ) -> StoreDomain: """ Add a custom domain to store. Args: db: Database session store_id: Store ID to add domain to domain_data: Domain creation data Returns: Created StoreDomain object Raises: StoreNotFoundException: If store not found StoreDomainAlreadyExistsException: If domain already registered MaxDomainsReachedException: If store has reached max domains InvalidDomainFormatException: If domain format is invalid """ try: # Verify store exists self._get_store_by_id_or_raise(db, store_id) # Check domain limit self._check_domain_limit(db, store_id) # Normalize domain normalized_domain = StoreDomain.normalize_domain(domain_data.domain) # Validate domain format self._validate_domain_format(normalized_domain) # Check if domain already exists if self._domain_exists(db, normalized_domain): existing_domain = ( db.query(StoreDomain) .filter(StoreDomain.domain == normalized_domain) .first() ) raise StoreDomainAlreadyExistsException( normalized_domain, existing_domain.store_id if existing_domain else None, ) # If setting as primary, unset other primary domains if domain_data.is_primary: self._unset_primary_domains(db, store_id) # Resolve platform_id: use provided value, or auto-resolve from primary StorePlatform platform_id = domain_data.platform_id if not platform_id: from app.modules.tenancy.models import StorePlatform primary_sp = ( db.query(StorePlatform) .filter(StorePlatform.store_id == store_id, StorePlatform.is_primary.is_(True)) .first() ) platform_id = primary_sp.platform_id if primary_sp else None # Create domain record new_domain = StoreDomain( store_id=store_id, domain=normalized_domain, is_primary=domain_data.is_primary, platform_id=platform_id, verification_token=secrets.token_urlsafe(32), is_verified=False, # Requires DNS verification is_active=False, # Cannot be active until verified ssl_status="pending", ) db.add(new_domain) db.flush() db.refresh(new_domain) logger.info(f"Domain {normalized_domain} added to store {store_id}") return new_domain except ( StoreNotFoundException, StoreDomainAlreadyExistsException, MaxDomainsReachedException, InvalidDomainFormatException, ReservedDomainException, ): raise except SQLAlchemyError as e: logger.error(f"Error adding domain: {str(e)}") raise ValidationException("Failed to add domain") def get_store_domains(self, db: Session, store_id: int) -> list[StoreDomain]: """ Get all domains for a store. Args: db: Database session store_id: Store ID Returns: List of StoreDomain objects Raises: StoreNotFoundException: If store not found """ try: # Verify store exists self._get_store_by_id_or_raise(db, store_id) domains = ( db.query(StoreDomain) .filter(StoreDomain.store_id == store_id) .order_by( StoreDomain.is_primary.desc(), StoreDomain.created_at.desc() ) .all() ) return domains except StoreNotFoundException: raise except SQLAlchemyError as e: logger.error(f"Error getting store domains: {str(e)}") raise ValidationException("Failed to retrieve domains") def get_domain_by_id(self, db: Session, domain_id: int) -> StoreDomain: """ Get domain by ID. Args: db: Database session domain_id: Domain ID Returns: StoreDomain object Raises: StoreDomainNotFoundException: If domain not found """ domain = db.query(StoreDomain).filter(StoreDomain.id == domain_id).first() if not domain: raise StoreDomainNotFoundException(str(domain_id)) return domain def update_domain( self, db: Session, domain_id: int, domain_update: StoreDomainUpdate ) -> StoreDomain: """ Update domain settings. Args: db: Database session domain_id: Domain ID domain_update: Update data Returns: Updated StoreDomain object Raises: StoreDomainNotFoundException: 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.store_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"Domain {domain.domain} updated") return domain except (StoreDomainNotFoundException, DomainNotVerifiedException): raise except SQLAlchemyError as e: logger.error(f"Error updating domain: {str(e)}") raise ValidationException("Failed to update domain") def delete_domain(self, db: Session, domain_id: int) -> str: """ Delete a custom domain. Args: db: Database session domain_id: Domain ID Returns: Success message Raises: StoreDomainNotFoundException: If domain not found """ try: domain = self.get_domain_by_id(db, domain_id) domain_name = domain.domain store_id = domain.store_id db.delete(domain) logger.info(f"Domain {domain_name} deleted from store {store_id}") return f"Domain {domain_name} deleted successfully" except StoreDomainNotFoundException: raise except SQLAlchemyError as e: logger.error(f"Error deleting domain: {str(e)}") raise ValidationException("Failed to delete domain") def verify_domain(self, db: Session, domain_id: int) -> tuple[StoreDomain, str]: """ Verify domain ownership via DNS TXT record. The store must add a TXT record: Name: _wizamart-verify.{domain} Value: {verification_token} Args: db: Database session domain_id: Domain ID Returns: Tuple of (verified_domain, message) Raises: StoreDomainNotFoundException: If domain not found DomainAlreadyVerifiedException: If already verified DomainVerificationFailedException: If verification fails """ try: domain = self.get_domain_by_id(db, domain_id) # Check if already verified if domain.is_verified: raise DomainAlreadyVerifiedException(domain_id, domain.domain) # Query DNS TXT records try: txt_records = dns.resolver.resolve( f"_wizamart-verify.{domain.domain}", "TXT" ) # Check if verification token is present for txt in txt_records: txt_value = txt.to_text().strip('"') if txt_value == domain.verification_token: # Verification successful domain.is_verified = True domain.verified_at = datetime.now(UTC) db.flush() db.refresh(domain) logger.info(f"Domain {domain.domain} verified successfully") return domain, f"Domain {domain.domain} verified successfully" # Token not found 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 ( StoreDomainNotFoundException, DomainAlreadyVerifiedException, DomainVerificationFailedException, DNSVerificationException, ): raise except SQLAlchemyError as e: logger.error(f"Error verifying domain: {str(e)}") raise ValidationException("Failed to verify domain") def get_verification_instructions(self, db: Session, domain_id: int) -> dict: """ Get DNS verification instructions for domain. Args: db: Database session domain_id: Domain ID Returns: Dict with verification instructions Raises: StoreDomainNotFoundException: If domain not found """ 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_store_by_id_or_raise(self, db: Session, store_id: int) -> Store: """Get store by ID or raise exception.""" store = db.query(Store).filter(Store.id == store_id).first() if not store: raise StoreNotFoundException(str(store_id), identifier_type="id") return store def _check_domain_limit(self, db: Session, store_id: int) -> None: """Check if store has reached maximum domain limit.""" domain_count = ( db.query(StoreDomain).filter(StoreDomain.store_id == store_id).count() ) if domain_count >= self.max_domains_per_store: raise MaxDomainsReachedException(store_id, self.max_domains_per_store) def _domain_exists(self, db: Session, domain: str) -> bool: """Check if domain already exists in system.""" return ( db.query(StoreDomain).filter(StoreDomain.domain == domain).first() is not None ) def _validate_domain_format(self, domain: str) -> None: """Validate domain format and check for reserved subdomains.""" # 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, store_id: int, exclude_domain_id: int | None = None ) -> None: """Unset all primary domains for store.""" query = db.query(StoreDomain).filter( StoreDomain.store_id == store_id, StoreDomain.is_primary == True ) if exclude_domain_id: query = query.filter(StoreDomain.id != exclude_domain_id) query.update({"is_primary": False}) # Create service instance store_domain_service = StoreDomainService()