harness_zbt/backend/app/infrastructure/mysql/repositories.py

from sqlalchemy import select
from sqlalchemy.orm import Session, sessionmaker

from app.core.identifiers import new_ulid
from app.domains.catalog.models import Plan, Product, ProductChangeLog, ProductVersion
from app.domains.catalog.repository import CatalogRepository
from app.domains.identity.models import AdminRole, AdminUser, AuthSession, H5User
from app.domains.identity.repository import IdentityRepository
from app.infrastructure.mysql.core_models import (
    AdminUserRecord,
    AuthSessionRecord,
    H5UserRecord,
    PermissionRecord,
    PlanRecord,
    ProductChangeLogRecord,
    ProductRecord,
    ProductVersionRecord,
    RoleRecord,
)

SessionFactory = sessionmaker[Session]


class SqlAlchemyIdentityRepository(IdentityRepository):
    def __init__(self, session_factory: SessionFactory) -> None:
        self._session_factory = session_factory

    def get_h5_user_by_mobile(self, mobile: str) -> H5User | None:
        with self._session_factory() as session:
            record = session.scalar(select(H5UserRecord).where(H5UserRecord.mobile == mobile))
            return self._to_h5_user(record) if record else None

    def get_h5_user(self, user_id: str) -> H5User | None:
        with self._session_factory() as session:
            record = session.get(H5UserRecord, user_id)
            return self._to_h5_user(record) if record else None

    def list_h5_users(self) -> list[H5User]:
        with self._session_factory() as session:
            records = session.scalars(
                select(H5UserRecord).order_by(H5UserRecord.created_at.desc())
            ).all()
            return [self._to_h5_user(record) for record in records]

    def save_h5_user(self, user: H5User) -> None:
        with self._session_factory() as session:
            session.merge(
                H5UserRecord(
                    id=user.id,
                    mobile=user.mobile,
                    mobile_masked=user.mobile_masked,
                    display_name=user.display_name,
                    status=user.status,
                    created_at=user.created_at,
                )
            )
            session.commit()

    def get_admin_user_by_username(self, username: str) -> AdminUser | None:
        with self._session_factory() as session:
            record = session.scalar(
                select(AdminUserRecord).where(AdminUserRecord.username == username)
            )
            return self._to_admin_user(record) if record else None

    def get_admin_user(self, user_id: str) -> AdminUser | None:
        with self._session_factory() as session:
            record = session.get(AdminUserRecord, user_id)
            return self._to_admin_user(record) if record else None

    def list_admin_users(self) -> list[AdminUser]:
        with self._session_factory() as session:
            records = session.scalars(
                select(AdminUserRecord).order_by(AdminUserRecord.username)
            ).all()
            return [self._to_admin_user(record) for record in records]

    def save_admin_user(self, user: AdminUser) -> None:
        with self._session_factory() as session:
            roles: list[RoleRecord] = []
            for role_code in user.roles:
                role = session.scalar(select(RoleRecord).where(RoleRecord.code == role_code))
                if role is None:
                    role = RoleRecord(
                        id=new_ulid(),
                        code=role_code,
                        name=role_code,
                        data_scope=user.data_scope,
                        status="ACTIVE",
                    )
                    session.add(role)
                permissions: list[PermissionRecord] = []
                for permission_code in user.permissions:
                    permission = session.scalar(
                        select(PermissionRecord).where(PermissionRecord.code == permission_code)
                    )
                    if permission is None:
                        resource, _, action = permission_code.partition(":")
                        permission = PermissionRecord(
                            id=new_ulid(),
                            code=permission_code,
                            name=permission_code,
                            resource=resource,
                            action=action or "read",
                        )
                        session.add(permission)
                    permissions.append(permission)
                role.permissions = permissions
                roles.append(role)
            record = session.get(AdminUserRecord, user.id)
            if record is None:
                record = AdminUserRecord(
                    id=user.id,
                    username=user.username,
                    password_hash=user.password_hash,
                    display_name=user.display_name,
                    status=user.status,
                )
                session.add(record)
            else:
                record.username = user.username
                record.password_hash = user.password_hash
                record.display_name = user.display_name
                record.status = user.status
            record.roles = roles
            session.commit()

    def list_roles(self) -> list[AdminRole]:
        with self._session_factory() as session:
            records = session.scalars(select(RoleRecord).order_by(RoleRecord.code)).all()
            return [
                AdminRole(
                    code=record.code,
                    name=record.name,
                    data_scope=record.data_scope,
                    status=record.status,
                    permissions=tuple(sorted(permission.code for permission in record.permissions)),
                )
                for record in records
            ]

    def save_role(self, role: AdminRole) -> None:
        with self._session_factory() as session:
            record = session.scalar(select(RoleRecord).where(RoleRecord.code == role.code))
            if record is None:
                raise KeyError(role.code)
            permissions: list[PermissionRecord] = []
            for permission_code in role.permissions:
                permission = session.scalar(
                    select(PermissionRecord).where(PermissionRecord.code == permission_code)
                )
                if permission is None:
                    resource, _, action = permission_code.partition(":")
                    permission = PermissionRecord(
                        id=new_ulid(),
                        code=permission_code,
                        name=permission_code,
                        resource=resource,
                        action=action or "read",
                    )
                    session.add(permission)
                permissions.append(permission)
            record.name = role.name
            record.data_scope = role.data_scope
            record.status = role.status
            record.permissions = permissions
            session.commit()

    def get_session(self, session_id: str) -> AuthSession | None:
        with self._session_factory() as session:
            record = session.get(AuthSessionRecord, session_id)
            if record is None:
                return None
            return AuthSession(
                id=record.id,
                subject_type=record.subject_type,
                subject_id=record.subject_id,
                refresh_jti_hash=record.refresh_jti_hash,
                expires_at=record.expires_at,
                revoked_at=record.revoked_at,
            )

    def save_session(self, auth_session: AuthSession) -> None:
        with self._session_factory() as session:
            session.merge(
                AuthSessionRecord(
                    id=auth_session.id,
                    subject_type=auth_session.subject_type,
                    subject_id=auth_session.subject_id,
                    refresh_jti_hash=auth_session.refresh_jti_hash,
                    expires_at=auth_session.expires_at,
                    revoked_at=auth_session.revoked_at,
                )
            )
            session.commit()

    @staticmethod
    def _to_h5_user(record: H5UserRecord) -> H5User:
        return H5User(
            id=record.id,
            mobile=record.mobile,
            mobile_masked=record.mobile_masked,
            display_name=record.display_name,
            status=record.status,
            created_at=record.created_at,
        )

    @staticmethod
    def _to_admin_user(record: AdminUserRecord) -> AdminUser:
        permissions = {permission.code for role in record.roles for permission in role.permissions}
        scopes = {role.data_scope for role in record.roles}
        data_scope = "ALL" if "ALL" in scopes else sorted(scopes)[0]
        return AdminUser(
            id=record.id,
            username=record.username,
            password_hash=record.password_hash,
            display_name=record.display_name,
            status=record.status,
            roles=tuple(sorted(role.code for role in record.roles)),
            permissions=tuple(sorted(permissions)),
            data_scope=data_scope,
        )


class SqlAlchemyCatalogRepository(CatalogRepository):
    def __init__(self, session_factory: SessionFactory) -> None:
        self._session_factory = session_factory

    def save_product(self, product: Product) -> None:
        with self._session_factory() as session:
            session.merge(
                ProductRecord(
                    id=product.id,
                    product_code=product.product_code,
                    name=product.name,
                    category=product.category,
                    summary=product.summary,
                    status=product.status,
                )
            )
            session.commit()

    def save_version(self, version: ProductVersion) -> None:
        with self._session_factory() as session:
            record = session.get(ProductVersionRecord, version.id)
            if record is None:
                record = ProductVersionRecord(
                    id=version.id,
                    product_id=version.product_id,
                    version_no=version.version_no,
                )
                session.add(record)
            record.version_no = version.version_no
            record.status = version.status
            record.effective_from = version.effective_from
            record.effective_to = version.effective_to
            record.rule_version = version.rule_version
            record.rate_version = version.rate_version
            record.terms_summary = version.terms_summary
            record.plans = [
                PlanRecord(
                    id=plan.id,
                    plan_code=plan.code,
                    name=plan.name,
                    summary=plan.summary,
                    status=plan.status,
                    premium_cents=plan.premium_cents,
                    coverage_amount_cents=plan.coverage_amount_cents,
                    min_age=plan.min_age,
                    max_age=plan.max_age,
                )
                for plan in version.plans
            ]
            session.commit()

    def delete_version(self, version_id: str) -> None:
        with self._session_factory() as session:
            record = session.get(ProductVersionRecord, version_id)
            if record is not None:
                session.delete(record)
                session.commit()

    def save_change_log(self, log: ProductChangeLog) -> None:
        with self._session_factory() as session:
            session.merge(
                ProductChangeLogRecord(
                    id=log.id,
                    product_id=log.product_id,
                    version_id=log.version_id,
                    action=log.action,
                    actor_id=log.actor_id,
                    actor_name=log.actor_name,
                    detail_json=log.detail,
                    created_at=log.created_at,
                )
            )
            session.commit()

    def list_change_logs(self, product_id: str) -> list[ProductChangeLog]:
        with self._session_factory() as session:
            records = session.scalars(
                select(ProductChangeLogRecord)
                .where(ProductChangeLogRecord.product_id == product_id)
                .order_by(ProductChangeLogRecord.created_at.desc())
            ).all()
            return [
                ProductChangeLog(
                    id=record.id,
                    product_id=record.product_id,
                    version_id=record.version_id,
                    action=record.action,
                    actor_id=record.actor_id,
                    actor_name=record.actor_name,
                    detail=dict(record.detail_json),
                    created_at=record.created_at,
                )
                for record in records
            ]

    def list_products(self) -> list[Product]:
        with self._session_factory() as session:
            records = session.scalars(select(ProductRecord)).all()
            return [
                Product(
                    id=record.id,
                    product_code=record.product_code,
                    name=record.name,
                    category=record.category,
                    summary=record.summary,
                    status=record.status,
                )
                for record in records
            ]

    def list_versions(self, product_id: str) -> list[ProductVersion]:
        with self._session_factory() as session:
            records = session.scalars(
                select(ProductVersionRecord).where(ProductVersionRecord.product_id == product_id)
            ).all()
            return [
                ProductVersion(
                    id=record.id,
                    product_id=record.product_id,
                    version_no=record.version_no,
                    status=record.status,
                    effective_from=record.effective_from,
                    effective_to=record.effective_to,
                    rule_version=record.rule_version,
                    rate_version=record.rate_version,
                    terms_summary=record.terms_summary,
                    plans=tuple(
                        Plan(
                            id=plan.id,
                            code=plan.plan_code,
                            name=plan.name,
                            summary=plan.summary,
                            status=plan.status,
                            premium_cents=plan.premium_cents,
                            coverage_amount_cents=plan.coverage_amount_cents,
                            min_age=plan.min_age,
                            max_age=plan.max_age,
                        )
                        for plan in record.plans
                    ),
                )
                for record in records
            ]