"""User service — business logic for user management and SPBU role assignments.""" import secrets import string from sqlalchemy import select from sqlalchemy.exc import IntegrityError, SQLAlchemyError from sqlalchemy.ext.asyncio import AsyncSession from app.core.security import hash_password from app.models.role import UserSpbuAssignment from app.models.user import User from app.repositories import role_repository, spbu_repository, user_repository from app.schemas.user import UserCreate, UserUpdate def _random_password(length: int = 24) -> str: alphabet = string.ascii_letters + string.digits return "".join(secrets.choice(alphabet) for _ in range(length)) async def list_users(db: AsyncSession, actor: User, skip: int = 0, limit: int = 50): """Superadmin sees all users. SPBU Admin sees only users in their own SPBUs (no superadmins).""" if actor.is_superadmin: return await user_repository.get_all(db, skip, limit) result = await db.execute( select(UserSpbuAssignment.spbu_id).where(UserSpbuAssignment.user_id == actor.id) ) actor_spbu_ids = [row[0] for row in result.all()] if not actor_spbu_ids: return [], 0 return await user_repository.get_all_scoped(db, actor_spbu_ids, skip, limit) async def create_user(db: AsyncSession, data: UserCreate, actor: User) -> User: """Create a new user. Password is optional — a random one is generated if omitted.""" if not actor.is_superadmin: raise PermissionError("Hanya Super Admin yang dapat membuat akun pengguna") existing = await user_repository.get_by_email(db, data.email) if existing: raise ValueError("Email sudah terdaftar") if data.username: existing_username = await user_repository.get_by_username(db, data.username) if existing_username: raise ValueError("Username sudah digunakan") password = data.password if data.password else _random_password() try: result = await user_repository.create( db, name=data.name, email=data.email, username=data.username or None, password_hash=hash_password(password), is_superadmin=data.is_superadmin, ) await db.commit() return result except IntegrityError: await db.rollback() raise ValueError("Data konflik atau sudah ada") except SQLAlchemyError: await db.rollback() raise async def get_user(db: AsyncSession, user_id: int) -> User: """Fetch a user by ID, raising ValueError if not found.""" user = await user_repository.get_by_id(db, user_id) if not user: raise ValueError("User tidak ditemukan") return user async def update_user(db: AsyncSession, user_id: int, data: UserUpdate, actor: User) -> User: """Update scalar fields on a user. Only superadmins can edit user profiles.""" if not actor.is_superadmin: raise PermissionError("Hanya Super Admin yang dapat mengubah akun pengguna") user = await get_user(db, user_id) update_data = data.model_dump(exclude_none=True) if "password" in update_data: update_data["password_hash"] = hash_password(update_data.pop("password")) try: result = await user_repository.update(db, user, **update_data) await db.commit() return result except IntegrityError: await db.rollback() raise ValueError("Data konflik atau sudah ada") except SQLAlchemyError: await db.rollback() raise async def deactivate_user(db: AsyncSession, user_id: int, actor: User) -> User: """Deactivate a user account. Non-superadmins cannot deactivate superadmin accounts.""" if not actor.is_superadmin: raise PermissionError("Hanya Super Admin yang dapat menonaktifkan akun pengguna") user = await get_user(db, user_id) if user.id == actor.id: raise ValueError("Tidak bisa menonaktifkan akun sendiri") if user.is_superadmin and not actor.is_superadmin: raise PermissionError("Tidak bisa menonaktifkan akun Super Admin") try: result = await user_repository.update(db, user, is_active=False) await db.commit() return result except SQLAlchemyError: await db.rollback() raise async def delete_user(db: AsyncSession, user_id: int, actor: User) -> None: """Delete a user. Hard delete in dev mode, soft delete in production. Only superadmins can delete users. Cannot delete yourself.""" if not actor.is_superadmin: raise PermissionError("Hanya Super Admin yang dapat menghapus akun pengguna") user = await get_user(db, user_id) if user.id == actor.id: raise ValueError("Tidak bisa menghapus akun sendiri") try: await user_repository.delete_user(db, user, hard=False) await db.commit() except IntegrityError: await db.rollback() raise ValueError("Tidak dapat menghapus user: masih ada data terkait") except SQLAlchemyError: await db.rollback() raise async def assign_to_spbu( db: AsyncSession, user_id: int, spbu_id: int, role_id: int, actor: User ) -> UserSpbuAssignment: """Assign a user to an SPBU with a given role. Updates the role if an assignment already exists. Non-superadmins can only assign to SPBUs they themselves belong to.""" await get_user(db, user_id) spbu = await spbu_repository.get_by_id(db, spbu_id) if not spbu: raise ValueError("SPBU tidak ditemukan") if not actor.is_superadmin: actor_spbu_ids = {a.spbu_id for a in actor.assignments} if spbu_id not in actor_spbu_ids: raise PermissionError("Tidak punya akses ke SPBU ini") role = await role_repository.get_by_id(db, role_id) if not role: raise ValueError("Role tidak ditemukan") existing = await user_repository.get_assignment(db, user_id, spbu_id) try: if existing: result = await user_repository.update_assignment(db, existing, role_id=role_id) else: result = await user_repository.create_assignment(db, user_id, spbu_id, role_id) await db.commit() return result except IntegrityError: await db.rollback() raise ValueError("Data konflik atau sudah ada") except SQLAlchemyError: await db.rollback() raise async def remove_assignment( db: AsyncSession, user_id: int, spbu_id: int, actor: User ) -> None: if not actor.is_superadmin: actor_spbu_ids = {a.spbu_id for a in (actor.assignments or [])} if spbu_id not in actor_spbu_ids: raise PermissionError("Tidak punya akses ke SPBU ini") try: deleted = await user_repository.delete_assignment(db, user_id, spbu_id) if not deleted: raise ValueError("Assignment tidak ditemukan") await db.commit() except ValueError: raise except IntegrityError: await db.rollback() raise ValueError("Tidak dapat menghapus assignment: masih ada data terkait") except SQLAlchemyError: await db.rollback() raise async def get_assignments( db: AsyncSession, user_id: int ) -> list[UserSpbuAssignment]: """Return all SPBU assignments for a given user.""" await get_user(db, user_id) return await user_repository.get_assignments(db, user_id)