orion/app/modules/tenancy/services/store_subdomain_service.py

# app/modules/tenancy/services/store_subdomain_service.py
"""
Service for managing StorePlatform custom subdomains.

Handles validation (format, uniqueness) and CRUD operations on
the custom_subdomain field of StorePlatform entries.
"""

import logging
import re

from sqlalchemy import func
from sqlalchemy.orm import Session

from app.exceptions.base import (
    ConflictException,
    ResourceNotFoundException,
    ValidationException,
)
from app.modules.tenancy.models import Platform, Store
from app.modules.tenancy.models.store_platform import StorePlatform

logger = logging.getLogger(__name__)

# Subdomain rules: lowercase alphanumeric + hyphens, 3-63 chars, no leading/trailing hyphen
_SUBDOMAIN_RE = re.compile(r"^[a-z0-9]([a-z0-9-]{1,61}[a-z0-9])?$")


class StoreSubdomainService:
    """Manage custom subdomains on StorePlatform entries."""

    def get_custom_subdomains(self, db: Session, store_id: int) -> list[dict]:
        """
        List all platform memberships for a store with their custom subdomains.

        Returns a list of dicts with platform info and custom_subdomain (may be None).
        """
        store = db.query(Store).filter(Store.id == store_id).first()
        if not store:
            raise ResourceNotFoundException("Store", str(store_id))

        memberships = (
            db.query(StorePlatform)
            .filter(
                StorePlatform.store_id == store_id,
                StorePlatform.is_active.is_(True),
            )
            .all()
        )

        results = []
        for sp in memberships:
            platform = db.query(Platform).filter(Platform.id == sp.platform_id).first()
            if not platform:
                continue
            results.append({
                "store_platform_id": sp.id,
                "platform_id": sp.platform_id,
                "platform_code": platform.code,
                "platform_name": platform.name,
                "platform_domain": platform.domain,
                "custom_subdomain": sp.custom_subdomain,
                "default_subdomain": store.subdomain,
                "full_url": (
                    f"{sp.custom_subdomain}.{platform.domain}"
                    if sp.custom_subdomain and platform.domain
                    else None
                ),
                "default_url": (
                    f"{store.subdomain}.{platform.domain}"
                    if store.subdomain and platform.domain
                    else None
                ),
            })

        return results

    def set_custom_subdomain(
        self, db: Session, store_id: int, platform_id: int, subdomain: str
    ) -> StorePlatform:
        """
        Set or update the custom_subdomain for a store on a specific platform.

        Validates:
        - Subdomain format (lowercase, alphanumeric + hyphens)
        - Uniqueness on the platform (no other store claims it)
        - StorePlatform membership exists and is active
        """
        subdomain = subdomain.strip().lower()

        # Validate format
        if not _SUBDOMAIN_RE.match(subdomain):
            raise ValidationException(
                "Must be 3-63 characters, lowercase alphanumeric and hyphens, "
                "cannot start or end with a hyphen.",
                field="custom_subdomain",
            )

        # Find the membership
        sp = (
            db.query(StorePlatform)
            .filter(
                StorePlatform.store_id == store_id,
                StorePlatform.platform_id == platform_id,
                StorePlatform.is_active.is_(True),
            )
            .first()
        )
        if not sp:
            raise ResourceNotFoundException(
                "StorePlatform",
                f"store_id={store_id}, platform_id={platform_id}",
            )

        # Check uniqueness on this platform (exclude current entry)
        existing = (
            db.query(StorePlatform)
            .filter(
                func.lower(StorePlatform.custom_subdomain) == subdomain,
                StorePlatform.platform_id == platform_id,
                StorePlatform.id != sp.id,
            )
            .first()
        )
        if existing:
            raise ConflictException(
                f"Subdomain '{subdomain}' is already claimed by another store "
                f"on this platform."
            )

        sp.custom_subdomain = subdomain
        db.flush()

        logger.info(
            f"Set custom_subdomain='{subdomain}' for store_id={store_id} "
            f"on platform_id={platform_id}"
        )
        return sp

    def clear_custom_subdomain(
        self, db: Session, store_id: int, platform_id: int
    ) -> StorePlatform:
        """Clear the custom_subdomain for a store on a specific platform."""
        sp = (
            db.query(StorePlatform)
            .filter(
                StorePlatform.store_id == store_id,
                StorePlatform.platform_id == platform_id,
                StorePlatform.is_active.is_(True),
            )
            .first()
        )
        if not sp:
            raise ResourceNotFoundException(
                "StorePlatform",
                f"store_id={store_id}, platform_id={platform_id}",
            )

        old_value = sp.custom_subdomain
        sp.custom_subdomain = None
        db.flush()

        logger.info(
            f"Cleared custom_subdomain (was '{old_value}') for store_id={store_id} "
            f"on platform_id={platform_id}"
        )
        return sp


store_subdomain_service = StoreSubdomainService()