DataOps-platform/app/core/data_source/service.py

"""Secure cross-store lifecycle orchestration for external data sources."""

from dataclasses import replace

from app.core.common.identifiers import new_governance_uid
from app.core.data_source.definitions import DataSourceFilters
from app.core.data_source.errors import (
    DataSourceConfigurationInvalid,
    DataSourceError,
    DataSourceNotFound,
)
from app.core.data_source.models import (
    DataSourceCredential,
    DataSourceDefinition,
)


TYPE_ALIASES = {
    "postgres": "postgresql",
    "postgresql": "postgresql",
    "mysql": "mysql",
}
POOL_INVALIDATION_REASONS = {
    "admin_reset",
    "configuration_changed",
    "credential_rotated",
}


class DataSourceService:
    def __init__(
        self,
        *,
        definitions,
        credentials,
        platform_session,
        adapter_resolver,
        connection_manager,
        settings_resolver,
        outbox_enqueuer,
    ):
        self.definitions = definitions
        self.credentials = credentials
        self.platform_session = platform_session
        self.adapter_resolver = adapter_resolver
        self.connection_manager = connection_manager
        self.settings_resolver = settings_resolver
        self.outbox_enqueuer = outbox_enqueuer

    @staticmethod
    def _database_type(payload):
        requested = str(payload.get("type") or "").strip().lower()
        database_type = TYPE_ALIASES.get(requested)
        if database_type is None:
            raise DataSourceConfigurationInvalid(
                "only PostgreSQL and MySQL data sources are supported"
            )
        return database_type

    @staticmethod
    def _required_text(payload, name):
        value = str(payload.get(name) or "").strip()
        if not value:
            raise DataSourceConfigurationInvalid(f"{name} is required")
        return value

    @classmethod
    def _definition(
        cls,
        payload,
        *,
        uid,
        credential_version,
        existing=None,
    ):
        database_type = cls._database_type(payload)
        try:
            port = int(payload.get("port"))
        except (TypeError, ValueError) as exc:
            raise DataSourceConfigurationInvalid("port is invalid") from exc
        if port < 1 or port > 65535:
            raise DataSourceConfigurationInvalid("port is invalid")

        tls_options = payload.get("tls_options") or {}
        if not isinstance(tls_options, dict):
            raise DataSourceConfigurationInvalid("tls_options is invalid")
        return DataSourceDefinition(
            uid=uid,
            name_en=cls._required_text(payload, "name_en"),
            name_zh=payload.get("name_zh"),
            database_type=database_type,
            host=cls._required_text(payload, "host"),
            port=port,
            database=cls._required_text(payload, "database"),
            schema=payload.get("schema"),
            credential_ref=uid,
            credential_version=credential_version,
            pool_size=payload.get("pool_size"),
            max_overflow=payload.get("max_overflow"),
            tls_options=tls_options,
            status=bool(payload.get("status", True)),
            description=payload.get("desc"),
            extra_properties=(
                dict(existing.extra_properties) if existing else {}
            ),
        )

    def _effective_credential(self, payload, existing):
        username = payload.get("username")
        password = payload.get("password")
        current = None
        if existing is not None and (not username or not password):
            current = self.credentials.get_active(
                self.platform_session(),
                existing.uid,
                existing.credential_version,
            )
        username = str(username or (current.username if current else "")).strip()
        password = str(password or (current.password if current else ""))
        if not username or not password:
            raise DataSourceConfigurationInvalid(
                "username and password are required"
            )
        options = payload.get("credential_options") or (
            dict(current.options) if current else {}
        )
        if not isinstance(options, dict):
            raise DataSourceConfigurationInvalid(
                "credential options are invalid"
            )
        return DataSourceCredential(username, password, options)

    def save(self, payload, *, actor_uid):
        if not isinstance(payload, dict):
            raise DataSourceConfigurationInvalid()
        requested_uid = str(payload.get("uid") or "").strip() or None
        existing = (
            self.definitions.get(requested_uid)
            if requested_uid is not None
            else None
        )
        if requested_uid is not None and existing is None:
            raise DataSourceNotFound()
        uid = requested_uid or new_governance_uid()
        credential = self._effective_credential(payload, existing)

        candidate = self._definition(
            payload,
            uid=uid,
            credential_version=(
                existing.credential_version if existing else 1
            ),
            existing=existing,
        )
        adapter = self.adapter_resolver(candidate.database_type)
        adapter.validate_options(candidate.tls_options)
        settings = self.settings_resolver(candidate.pool_overrides())
        adapter.test_connection(
            candidate,
            credential,
            query_timeout=settings["query_timeout"],
        )

        session = self.platform_session()
        previous_version = (
            existing.credential_version if existing is not None else None
        )
        try:
            sealed = self.credentials.create_version(
                session,
                data_source_uid=uid,
                credential=credential,
                actor_uid=actor_uid,
            )
            self.outbox_enqueuer(
                session,
                aggregate_type="datasource",
                aggregate_id=uid,
                event_type="datasource.credential_version_created",
                payload={
                    "uid": uid,
                    "database_type": candidate.database_type,
                    "credential_version": sealed.credential_version,
                },
            )
            session.commit()
        except Exception:
            session.rollback()
            raise

        candidate = replace(
            candidate,
            credential_version=sealed.credential_version,
        )
        try:
            saved = self.definitions.save(candidate)
        except Exception as graph_error:
            try:
                self.credentials.compensate_failed_activation(
                    session,
                    data_source_uid=uid,
                    failed_version=sealed.credential_version,
                    restore_version=previous_version,
                    actor_uid=actor_uid,
                )
                session.commit()
            except Exception as compensation_error:
                session.rollback()
                raise DataSourceError(
                    "data source save requires reconciliation"
                ) from compensation_error
            raise DataSourceError(
                "data source definition could not be saved"
            ) from graph_error

        if existing is not None:
            self.connection_manager.invalidate(
                uid,
                "configuration_changed",
            )
        return saved, existing is None

    def list(self, payload):
        payload = payload if isinstance(payload, dict) else {}

        def optional_text(name):
            value = payload.get(name)
            if value is None:
                return None
            normalized = str(value).strip()
            return normalized or None

        filters = DataSourceFilters(
            uid=optional_text("uid"),
            name_en=optional_text("name_en"),
            name_zh=optional_text("name_zh"),
            database_type=optional_text("type"),
            status=payload.get("status"),
        )
        return self.definitions.list(filters)

    def test_connection(self, payload):
        if not isinstance(payload, dict):
            raise DataSourceConfigurationInvalid()
        uid = str(payload.get("uid") or "").strip() or None
        existing = self.definitions.get(uid) if uid else None
        if uid and existing is None:
            raise DataSourceNotFound()
        effective_uid = uid or new_governance_uid()
        credential = self._effective_credential(payload, existing)
        candidate = self._definition(
            payload,
            uid=effective_uid,
            credential_version=(
                existing.credential_version if existing else 1
            ),
            existing=existing,
        )
        adapter = self.adapter_resolver(candidate.database_type)
        settings = self.settings_resolver(candidate.pool_overrides())
        adapter.test_connection(
            candidate,
            credential,
            query_timeout=settings["query_timeout"],
        )
        return {"connected": True, "message": "连接测试成功"}

    def delete(self, uid, *, actor_uid):
        uid = str(uid or "").strip()
        existing = self.definitions.get(uid)
        if existing is None:
            raise DataSourceNotFound()
        self.connection_manager.invalidate(uid, "datasource_deleted")
        if not self.definitions.delete(uid):
            raise DataSourceNotFound()

        session = self.platform_session()
        try:
            revoked = self.credentials.revoke_all(
                session,
                data_source_uid=uid,
                actor_uid=actor_uid,
            )
            self.outbox_enqueuer(
                session,
                aggregate_type="datasource",
                aggregate_id=uid,
                event_type="datasource.deleted",
                payload={"uid": uid},
            )
            session.commit()
        except Exception as exc:
            session.rollback()
            raise DataSourceError(
                "data source deletion requires reconciliation"
            ) from exc
        return {"uid": uid, "revoked_credential_versions": revoked}

    def pool_statuses(self):
        return self.connection_manager.snapshot()

    def pool_status(self, uid):
        uid = str(uid or "").strip()
        for status in self.pool_statuses():
            if status.data_source_uid == uid:
                return status
        raise DataSourceNotFound()

    def invalidate_pool(self, uid, *, reason, actor_uid):
        uid = str(uid or "").strip()
        if reason not in POOL_INVALIDATION_REASONS:
            raise DataSourceConfigurationInvalid(
                "pool invalidation reason is invalid"
            )
        if self.definitions.get(uid) is None:
            raise DataSourceNotFound()
        self.connection_manager.invalidate(uid, reason)
        session = self.platform_session()
        try:
            self.credentials.record_pool_event(
                session,
                data_source_uid=uid,
                actor_uid=actor_uid,
                event_type="pool_invalidated",
                safe_detail=f"pool invalidated: {reason}",
            )
            session.commit()
        except Exception:
            session.rollback()
            raise
        return {"uid": uid, "invalidated": True, "reason": reason}

    @staticmethod
    def serialize(definition):
        return {
            "uid": definition.uid,
            "name_en": definition.name_en,
            "name_zh": definition.name_zh,
            "type": definition.database_type,
            "host": definition.host,
            "port": definition.port,
            "database": definition.database,
            "schema": definition.schema,
            "pool_size": definition.pool_size,
            "max_overflow": definition.max_overflow,
            "status": definition.status,
            "desc": definition.description,
            "credential_configured": bool(
                definition.credential_version
            ),
        }

    @staticmethod
    def serialize_pool_status(status):
        return {
            "data_source_uid": status.data_source_uid,
            "credential_version": status.credential_version,
            "pool_state": status.pool_state,
            "pool_size": status.pool_size,
            "checked_out": status.checked_out,
            "checked_out_peak": status.checked_out_peak,
            "checked_in": status.checked_in,
            "overflow": status.overflow,
            "leases": status.leases,
            "checkout_wait_ms": status.checkout_wait_ms,
            "last_used_at": status.last_used_at,
            "connection_created_total": status.connection_created_total,
            "connection_failed_total": status.connection_failed_total,
            "pool_timeout_total": status.pool_timeout_total,
            "invalidated_total": status.invalidated_total,
            "query_total": status.query_total,
            "last_query_duration_ms": status.last_query_duration_ms,
            "consecutive_failures": status.consecutive_failures,
            "circuit_open_until": status.circuit_open_until,
        }


def build_data_source_service(manager):
    from app.core.events.outbox import enqueue_outbox

    return DataSourceService(
        definitions=manager.definitions,
        credentials=manager.credentials,
        platform_session=manager.platform_session,
        adapter_resolver=manager.adapter_resolver,
        connection_manager=manager,
        settings_resolver=manager.settings_resolver,
        outbox_enqueuer=enqueue_outbox,
    )