ss-tools/backend/src/services/profile_preference_service.py

# #region profile_preference_service [C:4] [TYPE Module] [SEMANTICS profile,preference,crud,persistence]
# @BRIEF Profile preference persistence — read, update, and normalize user dashboard filter preferences
#        with validation, encryption of git tokens, and deterministic default construction.
# @LAYER Domain
# @RELATION DEPENDS_ON -> [AuthRepository]
# @RELATION DEPENDS_ON -> [UserDashboardPreference]
# @RELATION DEPENDS_ON -> [profile_utils]
# @RELATION DEPENDS_ON -> [EncryptionManager]
# @RATIONALE Extracted from ProfileService to satisfy INV_7. Preference CRUD is the core profile
#            operation with DB persistence, token encryption, and cross-field validation — a
#            cohesive unit that isolates cleanly from security badges and Superset lookups.
# @REJECTED Keeping preference logic inside ProfileService was rejected because it forces all
#            three domains (preferences, security, lookup) into one class exceeding 150 lines,
#            and couples the validation logic to unrelated dependencies.
# @PRE db session is active.
# @POST Preference rows are created/updated with normalized values and encrypted tokens.
# @SIDE_EFFECT Writes preference records via AuthRepository; encrypts git tokens.

from datetime import datetime
from typing import Any

from sqlalchemy.orm import Session

from ..core.auth.repository import AuthRepository
from ..core.logger import belief_scope, logger
from ..models.auth import User
from ..models.profile import UserDashboardPreference
from ..schemas.profile import (
    ProfilePreference,
    ProfilePreferenceResponse,
    ProfilePreferenceUpdateRequest,
)
from .llm_provider import EncryptionManager
from .profile_utils import (
    ProfileAuthorizationError,
    ProfileValidationError,
    build_default_preference,
    mask_secret_value,
    normalize_density,
    normalize_start_page,
    normalize_username,
    sanitize_secret,
    sanitize_text,
    sanitize_username,
    validate_update_payload,
)
from .security_badge_service import SecurityBadgeService


# #region ProfilePreferenceService [C:4] [TYPE Class] [SEMANTICS preference,crud,validation,encryption]
# @BRIEF Handles profile preference persistence, validation, and token encryption.
# @RELATION DEPENDS_ON -> [AuthRepository]
# @RELATION DEPENDS_ON -> [EncryptionManager]
# @RELATION DEPENDS_ON -> [UserDashboardPreference]
# @PRE db session is active.
# @POST Preference CRUD operations are user-scoped and return normalized ProfilePreferenceResponse.
class ProfilePreferenceService:
    """Profile preference persistence and validation."""

    # #region __init__ [TYPE Function]
    # @BRIEF Initialize with DB session and references to shared services.
    # @PRE db session is active.
    # @POST Service is ready for preference persistence operations.
    def __init__(self, db: Session, config_manager: Any, plugin_loader: Any = None):
        self.db = db
        self.config_manager = config_manager
        self.plugin_loader = plugin_loader
        self.auth_repository = AuthRepository(db)
        self.encryption = EncryptionManager()
        self.security_badge_service = SecurityBadgeService(plugin_loader)
    # #endregion __init__

    # #region get_my_preference [TYPE Function]
    # @BRIEF Return current user's persisted preference or default non-configured view.
    # @PRE current_user is authenticated.
    # @POST Returned payload belongs to current_user only.
    def get_my_preference(self, current_user: User) -> ProfilePreferenceResponse:
        with belief_scope(
            "ProfilePreferenceService.get_my_preference", f"user_id={current_user.id}"
        ):
            logger.reflect("Loading current user's dashboard preference")
            preference = self._get_preference_row(current_user.id)
            security_summary = self.security_badge_service.build_security_summary(current_user)

            if preference is None:
                return ProfilePreferenceResponse(
                    status="success",
                    message="Preference not configured yet",
                    preference=build_default_preference(current_user.id),
                    security=security_summary,
                )
            return ProfilePreferenceResponse(
                status="success",
                message="Preference loaded",
                preference=self._to_preference_payload(
                    preference, str(current_user.id)
                ),
                security=security_summary,
            )
    # #endregion get_my_preference

    # #region get_dashboard_filter_binding [TYPE Function]
    # @BRIEF Return only dashboard-filter fields required by dashboards listing hot path.
    # @PRE current_user is authenticated.
    # @POST Returns normalized username and profile-default filter toggles without security summary expansion.
    def get_dashboard_filter_binding(self, current_user: User) -> dict:
        with belief_scope(
            "ProfilePreferenceService.get_dashboard_filter_binding", f"user_id={current_user.id}"
        ):
            preference = self._get_preference_row(current_user.id)
            if preference is None:
                return {
                    "superset_username": None,
                    "superset_username_normalized": None,
                    "show_only_my_dashboards": False,
                    "show_only_slug_dashboards": True,
                }

            return {
                "superset_username": sanitize_username(
                    preference.superset_username
                ),
                "superset_username_normalized": normalize_username(
                    preference.superset_username
                ),
                "show_only_my_dashboards": bool(preference.show_only_my_dashboards),
                "show_only_slug_dashboards": bool(
                    preference.show_only_slug_dashboards
                    if preference.show_only_slug_dashboards is not None
                    else True
                ),
            }
    # #endregion get_dashboard_filter_binding

    # #region update_my_preference [TYPE Function]
    # @BRIEF Validate and persist current user's profile preference in self-scoped mode.
    # @PRE current_user is authenticated and payload is provided.
    # @POST Preference row for current_user is created/updated when validation passes.
    def update_my_preference(
        self,
        current_user: User,
        payload: ProfilePreferenceUpdateRequest,
        target_user_id: str | None = None,
    ) -> ProfilePreferenceResponse:
        with belief_scope(
            "ProfilePreferenceService.update_my_preference", f"user_id={current_user.id}"
        ):
            logger.reason("Evaluating self-scope guard before preference mutation")
            requested_user_id = str(target_user_id or current_user.id)
            if requested_user_id != str(current_user.id):
                logger.explore("Cross-user mutation attempt blocked")
                raise ProfileAuthorizationError(
                    "Cross-user preference mutation is forbidden"
                )

            preference = self._get_or_create_preference_row(current_user.id)
            provided_fields = set(getattr(payload, "model_fields_set", set()))

            effective_superset_username = sanitize_username(
                preference.superset_username
            )
            if "superset_username" in provided_fields:
                effective_superset_username = sanitize_username(
                    payload.superset_username
                )

            effective_show_only = bool(preference.show_only_my_dashboards)
            if "show_only_my_dashboards" in provided_fields:
                effective_show_only = bool(payload.show_only_my_dashboards)

            effective_show_only_slug = (
                bool(preference.show_only_slug_dashboards)
                if preference.show_only_slug_dashboards is not None
                else True
            )
            if "show_only_slug_dashboards" in provided_fields:
                effective_show_only_slug = bool(payload.show_only_slug_dashboards)

            effective_git_username = sanitize_text(preference.git_username)
            if "git_username" in provided_fields:
                effective_git_username = sanitize_text(payload.git_username)

            effective_git_email = sanitize_text(preference.git_email)
            if "git_email" in provided_fields:
                effective_git_email = sanitize_text(payload.git_email)

            effective_start_page = normalize_start_page(preference.start_page)
            if "start_page" in provided_fields:
                effective_start_page = normalize_start_page(payload.start_page)

            effective_auto_open_task_drawer = (
                bool(preference.auto_open_task_drawer)
                if preference.auto_open_task_drawer is not None
                else True
            )
            if "auto_open_task_drawer" in provided_fields:
                effective_auto_open_task_drawer = bool(payload.auto_open_task_drawer)

            effective_dashboards_table_density = normalize_density(
                preference.dashboards_table_density
            )
            if "dashboards_table_density" in provided_fields:
                effective_dashboards_table_density = normalize_density(
                    payload.dashboards_table_density
                )

            effective_telegram_id = sanitize_text(preference.telegram_id)
            if "telegram_id" in provided_fields:
                effective_telegram_id = sanitize_text(payload.telegram_id)

            effective_email_address = sanitize_text(preference.email_address)
            if "email_address" in provided_fields:
                effective_email_address = sanitize_text(payload.email_address)

            effective_notify_on_fail = (
                bool(preference.notify_on_fail)
                if preference.notify_on_fail is not None
                else True
            )
            if "notify_on_fail" in provided_fields:
                effective_notify_on_fail = bool(payload.notify_on_fail)

            validation_errors = validate_update_payload(
                superset_username=effective_superset_username,
                show_only_my_dashboards=effective_show_only,
                git_email=effective_git_email,
                start_page=effective_start_page,
                dashboards_table_density=effective_dashboards_table_density,
                email_address=effective_email_address,
            )
            if validation_errors:
                logger.reflect("Validation failed; mutation is denied")
                raise ProfileValidationError(validation_errors)

            preference.superset_username = effective_superset_username
            preference.superset_username_normalized = normalize_username(
                effective_superset_username
            )
            preference.show_only_my_dashboards = effective_show_only
            preference.show_only_slug_dashboards = effective_show_only_slug

            preference.git_username = effective_git_username
            preference.git_email = effective_git_email

            if "git_personal_access_token" in provided_fields:
                sanitized_token = sanitize_secret(
                    payload.git_personal_access_token
                )
                if sanitized_token is None:
                    preference.git_personal_access_token_encrypted = None
                else:
                    preference.git_personal_access_token_encrypted = (
                        self.encryption.encrypt(sanitized_token)
                    )

            preference.start_page = effective_start_page
            preference.auto_open_task_drawer = effective_auto_open_task_drawer
            preference.dashboards_table_density = effective_dashboards_table_density
            preference.telegram_id = effective_telegram_id
            preference.email_address = effective_email_address
            preference.notify_on_fail = effective_notify_on_fail
            preference.updated_at = datetime.utcnow()

            persisted_preference = self.auth_repository.save_user_dashboard_preference(
                preference
            )

            logger.reason("Preference persisted successfully")
            return ProfilePreferenceResponse(
                status="success",
                message="Preference saved",
                preference=self._to_preference_payload(
                    persisted_preference,
                    str(current_user.id),
                ),
                security=self.security_badge_service.build_security_summary(current_user),
            )
    # #endregion update_my_preference

    # #region _to_preference_payload [TYPE Function]
    # @BRIEF Map ORM preference row to API DTO with token metadata.
    # @PRE preference row can contain nullable optional fields.
    # @POST Returns normalized ProfilePreference object.
    def _to_preference_payload(
        self,
        preference: UserDashboardPreference,
        user_id: str,
    ) -> ProfilePreference:
        encrypted_token = sanitize_text(
            preference.git_personal_access_token_encrypted
        )
        token_masked = None
        if encrypted_token:
            try:
                decrypted_token = self.encryption.decrypt(encrypted_token)
                token_masked = mask_secret_value(decrypted_token)
            except Exception:
                token_masked = "***"

        created_at = getattr(preference, "created_at", None) or datetime.utcnow()
        updated_at = getattr(preference, "updated_at", None) or created_at

        return ProfilePreference(
            user_id=str(user_id),
            superset_username=sanitize_username(preference.superset_username),
            superset_username_normalized=normalize_username(
                preference.superset_username_normalized
            ),
            show_only_my_dashboards=bool(preference.show_only_my_dashboards),
            show_only_slug_dashboards=(
                bool(preference.show_only_slug_dashboards)
                if preference.show_only_slug_dashboards is not None
                else True
            ),
            git_username=sanitize_text(preference.git_username),
            git_email=sanitize_text(preference.git_email),
            has_git_personal_access_token=bool(encrypted_token),
            git_personal_access_token_masked=token_masked,
            start_page=normalize_start_page(preference.start_page),
            auto_open_task_drawer=(
                bool(preference.auto_open_task_drawer)
                if preference.auto_open_task_drawer is not None
                else True
            ),
            dashboards_table_density=normalize_density(
                preference.dashboards_table_density
            ),
            telegram_id=sanitize_text(preference.telegram_id),
            email_address=sanitize_text(preference.email_address),
            notify_on_fail=bool(preference.notify_on_fail)
            if preference.notify_on_fail is not None
            else True,
            created_at=created_at,
            updated_at=updated_at,
        )
    # #endregion _to_preference_payload

    # #region _build_default_preference [C:1] [TYPE Function]
    # @BRIEF Delegate to profile_utils.build_default_preference.
    def _build_default_preference(self, user_id: str) -> ProfilePreference:
        return build_default_preference(user_id)
    # #endregion _build_default_preference

    # #region _get_preference_row [C:2] [TYPE Function]
    # @BRIEF Return persisted preference row for user or None.
    def _get_preference_row(self, user_id: str) -> UserDashboardPreference | None:
        return self.auth_repository.get_user_dashboard_preference(str(user_id))
    # #endregion _get_preference_row

    # #region _get_or_create_preference_row [C:2] [TYPE Function]
    # @BRIEF Return existing preference row or create new unsaved row.
    def _get_or_create_preference_row(self, user_id: str) -> UserDashboardPreference:
        existing = self._get_preference_row(user_id)
        if existing is not None:
            return existing
        return UserDashboardPreference(user_id=str(user_id))
    # #endregion _get_or_create_preference_row
# #endregion ProfilePreferenceService
# #endregion profile_preference_service