RuoYi-Vue3-FastAPI/ruoyi-fastapi-backend/cli/runtime/config/service.py

from typing import Any, Literal

from cli.exit_codes import ARGUMENT_ERROR, DATABASE_ERROR, REDIS_ERROR

from .gateway import ConfigInfrastructureGateway
from .support import ConfigDomainSupport


class ConfigRuntimeService:
    """
    参数配置运行时服务。

    该服务作为参数配置运行时 facade，对外统一暴露参数配置列表、详情、
    写入、缓存刷新与一致性诊断入口。

    :param infrastructure_gateway: 参数配置基础设施网关
    :param domain_support: 参数配置领域支持对象
    """

    def __init__(
        self,
        *,
        infrastructure_gateway: ConfigInfrastructureGateway | None = None,
        domain_support: ConfigDomainSupport | None = None,
    ) -> None:
        """
        初始化参数配置运行时服务。

        :param infrastructure_gateway: 参数配置基础设施网关
        :param domain_support: 参数配置领域支持对象
        :return: None
        """
        self.infrastructure_gateway = infrastructure_gateway or ConfigInfrastructureGateway()
        self.domain_support = domain_support or ConfigDomainSupport(self.infrastructure_gateway)

    @staticmethod
    def build_cli_config_model(config_vo_module: Any, config_record: Any) -> Any | None:
        """
        将 ORM 配置记录显式转换为 CLI 可用的配置模型。

        `ConfigModel` 当前未统一开启 `populate_by_name`，因此 CLI 侧在处理
        ORM 蛇形字段对象时，需要主动映射为驼峰 alias 输入，避免详情查询结果
        被静默序列化为空。

        :param config_vo_module: 配置 VO 模块
        :param config_record: ORM 配置记录
        :return: CLI 可用配置模型
        """
        if config_record is None:
            return None
        return config_vo_module.ConfigModel(
            configId=getattr(config_record, 'config_id', None),
            configName=getattr(config_record, 'config_name', None),
            configKey=getattr(config_record, 'config_key', None),
            configValue=getattr(config_record, 'config_value', None),
            configType=getattr(config_record, 'config_type', None),
            createBy=getattr(config_record, 'create_by', None),
            createTime=getattr(config_record, 'create_time', None),
            updateBy=getattr(config_record, 'update_by', None),
            updateTime=getattr(config_record, 'update_time', None),
            remark=getattr(config_record, 'remark', None),
        )

    async def load_config_from_database(
        self,
        config_key: str,
    ) -> tuple[Any | None, dict[str, Any] | None] | dict[str, Any]:
        """
        从数据库加载参数配置。

        :param config_key: 参数键名
        :return: 成功时返回配置模型与序列化结果，失败时返回错误结果字典
        """
        async_session_local = self.infrastructure_gateway.get_async_session_local()
        config_dao = self.infrastructure_gateway.get_config_dao()
        config_vo_module = self.infrastructure_gateway.get_config_vo_module()
        try:
            async with async_session_local() as session:
                config_record = await config_dao.get_config_detail_by_info(
                    session,
                    config_vo_module.ConfigModel(configKey=config_key),
                )
        except Exception as exc:
            return {'ok': False, 'message': '读取参数配置失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}

        database_model = self.build_cli_config_model(config_vo_module, config_record)
        return database_model, self.domain_support.serialize_config_model(database_model)

    async def load_config_from_cache(
        self,
        config_key: str,
    ) -> tuple[str | None, dict[str, Any] | None] | dict[str, Any]:
        """
        从缓存加载参数配置。

        :param config_key: 参数键名
        :return: 成功时返回原始缓存值与序列化结果，失败时返回错误结果字典
        """
        redis = None
        redis_util = self.infrastructure_gateway.get_redis_util()
        config_service = self.infrastructure_gateway.get_config_service()
        redis_error = self.infrastructure_gateway.get_redis_error_class()
        try:
            redis = await redis_util.create_redis_pool(log_enabled=False)
            cache_value = await config_service.query_config_list_from_cache_services(redis, config_key)
            return cache_value, self.domain_support.serialize_cache_payload(config_key, cache_value)
        except redis_error as exc:
            return {'ok': False, 'message': '读取参数缓存失败', 'error': str(exc), 'exit_code': REDIS_ERROR}
        finally:
            if redis is not None:
                await redis.close()

    async def list_configs(
        self,
        config_name: str = '',
        config_key: str = '',
        config_type: Literal['Y', 'N'] | None = None,
        begin_date: str = '',
        end_date: str = '',
        *,
        paged: bool = False,
        page_num: int = 1,
        page_size: int = 20,
    ) -> dict[str, Any]:
        """
        查询参数配置列表。

        :param config_name: 参数名称过滤条件
        :param config_key: 参数键名过滤条件
        :param config_type: 参数类型过滤条件
        :param begin_date: 开始日期
        :param end_date: 结束日期
        :param paged: 是否开启分页
        :param page_num: 页码
        :param page_size: 每页数量
        :return: CLI 标准结果字典
        """
        async_session_local = self.infrastructure_gateway.get_async_session_local()
        config_service = self.infrastructure_gateway.get_config_service()
        query_model = self.domain_support.build_config_query(
            config_name,
            config_key,
            config_type,
            begin_date,
            end_date,
            page_num,
            page_size,
        )
        try:
            async with async_session_local() as session:
                result = await config_service.get_config_list_services(session, query_model, is_page=paged)
        except Exception as exc:
            return {'ok': False, 'message': '读取参数配置列表失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}

        filters = self.domain_support.build_list_filters(
            config_name=config_name,
            config_key=config_key,
            config_type=config_type,
            begin_date=begin_date,
            end_date=end_date,
            paged=paged,
            page_num=page_num,
            page_size=page_size,
        )
        if paged and isinstance(result, self.infrastructure_gateway.get_page_model()):
            page_payload = result.model_dump(by_alias=True)
            page_payload['rows'] = [
                self.domain_support.sanitize_config_mapping(item) for item in page_payload.get('rows', [])
            ]
            return {'ok': True, 'filters': filters, 'page': page_payload}

        items = [self.domain_support.sanitize_config_mapping(item) for item in result]
        return {'ok': True, 'filters': filters, 'count': len(items), 'items': items}

    async def get_config(
        self,
        config_key: str,
        *,
        source: Literal['db', 'cache', 'both'] = 'both',
    ) -> dict[str, Any]:
        """
        按参数键读取配置详情。

        :param config_key: 参数键名
        :param source: 读取来源
        :return: CLI 标准结果字典
        """
        database_model: Any | None = None
        database_payload: dict[str, Any] | None = None
        cache_value: str | None = None
        cache_payload: dict[str, Any] | None = None

        if source in {'db', 'both'}:
            database_result = await self.load_config_from_database(config_key)
            if isinstance(database_result, dict):
                return database_result
            database_model, database_payload = database_result

        if source in {'cache', 'both'}:
            cache_result = await self.load_config_from_cache(config_key)
            if isinstance(cache_result, dict):
                error_payload = {**cache_result, 'source': source}
                if database_payload is not None:
                    error_payload['database'] = database_payload
                return error_payload
            cache_value, cache_payload = cache_result

        if source == 'db' and database_payload is None:
            return self.domain_support.build_missing_config_result(config_key, source)
        if source == 'cache' and cache_payload is None:
            return self.domain_support.build_missing_config_result(config_key, source)
        if source == 'both' and database_payload is None and cache_payload is None:
            return self.domain_support.build_missing_config_result(config_key, source)

        payload = {'ok': True, 'key': config_key, 'source': source}
        if source in {'db', 'both'}:
            payload['database'] = database_payload
        if source in {'cache', 'both'}:
            payload['cache'] = cache_payload
        if source == 'both':
            payload['inSync'] = (database_model.config_value if database_model else None) == cache_value
        return payload

    async def set_config(
        self,
        config_key: str,
        config_value: str,
        *,
        config_name: str | None = None,
        config_type: Literal['Y', 'N'] | None = None,
        remark: str | None = None,
        dry_run: bool = False,
    ) -> dict[str, Any]:
        """
        新增或更新参数配置，并同步缓存。

        :param config_key: 参数键名
        :param config_value: 参数键值
        :param config_name: 参数名称
        :param config_type: 参数类型
        :param remark: 备注
        :param dry_run: 是否仅执行演练
        :return: CLI 标准结果字典
        """
        async_session_local = self.infrastructure_gateway.get_async_session_local()
        config_dao = self.infrastructure_gateway.get_config_dao()
        config_vo_module = self.infrastructure_gateway.get_config_vo_module()
        redis_util = self.infrastructure_gateway.get_redis_util()
        redis_init_key_config = self.infrastructure_gateway.get_redis_init_key_config()
        redis_error = self.infrastructure_gateway.get_redis_error_class()
        existing_config = None
        target_config = None
        try:
            async with async_session_local() as session:
                config_record = await config_dao.get_config_detail_by_info(
                    session,
                    config_vo_module.ConfigModel(configKey=config_key),
                )
                existing_config = self.build_cli_config_model(config_vo_module, config_record)
                if existing_config is None and not config_name:
                    return {
                        'ok': False,
                        'message': '新增参数配置时必须传入 --name',
                        'exit_code': ARGUMENT_ERROR,
                    }

                target_config = self.domain_support.build_target_config_model(
                    config_key,
                    config_value,
                    config_name,
                    config_type,
                    remark,
                    existing_config,
                )
                if dry_run:
                    return {
                        'ok': True,
                        'message': '参数配置演练完成，未执行实际写入',
                        'dryRun': True,
                        'action': 'update' if existing_config else 'create',
                        'config': self.domain_support.serialize_config_model(target_config),
                    }

                if existing_config:
                    await config_dao.edit_config_dao(session, target_config.model_dump(exclude_none=True))
                else:
                    await config_dao.add_config_dao(session, target_config)
                await session.commit()
        except Exception as exc:
            return {'ok': False, 'message': '写入参数配置失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}

        redis = None
        try:
            redis = await redis_util.create_redis_pool(log_enabled=False)
            await redis.set(f'{redis_init_key_config.SYS_CONFIG.key}:{config_key}', config_value)
        except redis_error as exc:
            return {
                'ok': False,
                'message': '参数配置已写入数据库，但同步缓存失败',
                'error': str(exc),
                'config': self.domain_support.serialize_config_model(target_config),
                'databaseCommitted': True,
                'exit_code': REDIS_ERROR,
            }
        finally:
            if redis is not None:
                await redis.close()

        return {
            'ok': True,
            'message': '参数配置已同步到数据库和缓存',
            'action': 'update' if existing_config else 'create',
            'config': self.domain_support.serialize_config_model(target_config),
        }

    async def sync_config_cache(self) -> dict[str, Any]:
        """
        刷新参数配置缓存。

        :return: CLI 标准结果字典
        """
        redis = None
        async_session_local = self.infrastructure_gateway.get_async_session_local()
        config_service = self.infrastructure_gateway.get_config_service()
        config_vo_module = self.infrastructure_gateway.get_config_vo_module()
        redis_util = self.infrastructure_gateway.get_redis_util()
        redis_error = self.infrastructure_gateway.get_redis_error_class()
        try:
            async with async_session_local() as session:
                config_list = await config_service.get_config_list_services(
                    session,
                    config_vo_module.ConfigPageQueryModel(),
                    is_page=False,
                )
                redis = await redis_util.create_redis_pool(log_enabled=False)
                await config_service.init_cache_sys_config_services(session, redis)
        except redis_error as exc:
            return {'ok': False, 'message': '刷新参数缓存失败', 'error': str(exc), 'exit_code': REDIS_ERROR}
        except Exception as exc:
            return {'ok': False, 'message': '刷新参数缓存失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}
        finally:
            if redis is not None:
                await redis.close()

        return {'ok': True, 'message': '参数缓存刷新成功', 'count': len(config_list)}

    async def diagnose_config(self, *, sample_limit: int = 10) -> dict[str, Any]:
        """
        诊断数据库参数配置与 Redis 缓存的一致性状态。

        :param sample_limit: 示例键名输出上限
        :return: CLI 标准结果字典
        """
        redis = None
        async_session_local = self.infrastructure_gateway.get_async_session_local()
        config_service = self.infrastructure_gateway.get_config_service()
        config_vo_module = self.infrastructure_gateway.get_config_vo_module()
        redis_util = self.infrastructure_gateway.get_redis_util()
        redis_init_key_config = self.infrastructure_gateway.get_redis_init_key_config()
        redis_error = self.infrastructure_gateway.get_redis_error_class()
        try:
            async with async_session_local() as session:
                config_list = await config_service.get_config_list_services(
                    session,
                    config_vo_module.ConfigPageQueryModel(),
                    is_page=False,
                )
            redis = await redis_util.create_redis_pool(log_enabled=False)
            cache_prefix = f'{redis_init_key_config.SYS_CONFIG.key}:'
            cache_keys = sorted(await redis.keys(f'{cache_prefix}*'))
            cache_values = await redis.mget(*cache_keys) if cache_keys else []
        except redis_error as exc:
            return {'ok': False, 'message': '读取参数缓存诊断信息失败', 'error': str(exc), 'exit_code': REDIS_ERROR}
        except Exception as exc:
            return {'ok': False, 'message': '读取参数配置诊断信息失败', 'error': str(exc), 'exit_code': DATABASE_ERROR}
        finally:
            if redis is not None:
                await redis.close()

        database_map = {str(item.get('configKey')): str(item.get('configValue', '')) for item in config_list}
        cache_map = {
            cache_key.replace(cache_prefix, '', 1): '' if cache_value is None else str(cache_value)
            for cache_key, cache_value in zip(cache_keys, cache_values, strict=False)
        }
        missing_in_cache = sorted(key for key in database_map if key not in cache_map)
        orphan_in_cache = sorted(key for key in cache_map if key not in database_map)
        mismatch_keys = sorted(
            key for key, database_value in database_map.items() if key in cache_map and database_value != cache_map[key]
        )
        in_sync = not missing_in_cache and not orphan_in_cache and not mismatch_keys
        return {
            'ok': in_sync,
            'message': '参数配置数据库与缓存一致' if in_sync else '参数配置存在数据库与缓存不一致项',
            'databaseCount': len(database_map),
            'cacheCount': len(cache_map),
            'missingInCacheCount': len(missing_in_cache),
            'orphanInCacheCount': len(orphan_in_cache),
            'mismatchCount': len(mismatch_keys),
            'sampleLimit': sample_limit,
            'missingInCache': missing_in_cache[:sample_limit],
            'orphanInCache': orphan_in_cache[:sample_limit],
            'mismatchKeys': mismatch_keys[:sample_limit],
        }


CONFIG_RUNTIME = ConfigRuntimeService()