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 ]