o2o-site-AEO/solution/backend/services/auth_service.py

import uuid

from fastapi import Depends

from common.database.db_session_manager import DB_SESSION_MNG
from common.database.model.models import users
from common.enums import DBWRType, ErrorType, UserRole, UserStatus
from common.logger import LOG
from common.models.gmodel import UserInfo
from crud.user_crud import IUserCRUD, UserCRUD
from router.v1.auth.protocol import CompanyData, Req_UpdateMe, Res_Login, Res_Me, Res_RefreshToken
from router.v1.validator.dependencies import (
    CreateAccessToken,
    CreateRefreshToken,
    DecodeRefreshToken,
    GetHashedPW,
    VerifyPW,
)


class AuthService:
    """비즈니스 로직 계층 (MVC 의 컨트롤러-서비스 분리에서 서비스).

    - CRUD 는 Depends 로 인터페이스 타입으로 주입받는다.
    - DB 접근은 DB_SESSION_MNG 의 람다 실행으로만 한다.
      조회 = execute_lambda(..., DB_READ, lambda s: crud.xxx(s, ...))
      변경 = execute_lambda_run([DBType], [lambda s: crud.xxx(s, ...)])
    - 모든 메서드는 Res_* 를 만들어 result 에 ErrorType 을 세팅해 반환한다.
    """

    def __init__(self, user_crud: IUserCRUD = Depends(UserCRUD)):
        self.user_crud = user_crud

    @staticmethod
    def _user_info(user: users) -> UserInfo:
        # uuid → str (JWT json 직렬화 위해). 기능 라우터는 company_id 로 스코프한다.
        return UserInfo(
            user_id=str(user.user_id),
            id=user.id,
            company_id=str(user.company_id),
            role=user.role,
        )

    async def attempt_login(self, login_id: str, password: str, connect_ip: str) -> Res_Login:
        LOG.i(f"LOGIN : id={login_id}")
        res = Res_Login()

        # 1) 계정 조회 (Read DB)
        err_type, user = await DB_SESSION_MNG.execute_lambda(
            users.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.user_crud.get_user_by_login_id(s, login_id),
        )
        if err_type != ErrorType.SUCCESS:
            # 계정 없음/조회 실패 모두 로그인 실패로 일반화
            res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO)
            return res
        user: users

        # 2) 비밀번호 검증
        if not await VerifyPW(password, user.password):
            res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO)
            return res

        # 3) 상태 확인 (활성 아니면 차단)
        if user.status != UserStatus.ACTIVE.value:
            res.result.SetResult(ErrorType.ACCOUNT_BLOCKED_USER)
            return res

        # 4) 토큰 발급
        user_info = self._user_info(user)
        res.access_token = CreateAccessToken(user_info)
        res.refresh_token = CreateRefreshToken(user_info)

        # 5) 마지막 접속 시간 갱신 (Write DB, 트랜잭션)
        err_type = await DB_SESSION_MNG.execute_lambda_run(
            [users.DBType()],
            [lambda s: self.user_crud.update_last_accessed(s, user.user_id)],
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        return res

    async def get_me(self, user_info: UserInfo) -> Res_Me:
        res = Res_Me()

        # 1) 유저 조회
        err_type, user = await DB_SESSION_MNG.execute_lambda(
            users.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.user_crud.get_user_by_login_id(s, user_info.id),
        )
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO)
            return res
        user: users

        # 2) 소속사 조회 (없어도 치명적 아님)
        company = None
        c_err, company_row = await DB_SESSION_MNG.execute_lambda(
            users.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.user_crud.get_company(s, user.company_id),
        )
        if c_err == ErrorType.SUCCESS and company_row is not None:
            company = CompanyData(company_id=str(company_row.company_id), name=company_row.name)

        res.user_id = str(user.user_id)
        res.id = user.id
        res.name = user.name
        res.email = user.email
        res.contact_number = user.contact_number
        res.role = UserRole(user.role)
        res.company = company
        return res

    async def update_me(self, user_info: UserInfo, req: Req_UpdateMe) -> Res_Me:
        res = Res_Me()
        data = req.model_dump(exclude_unset=True)

        # 비밀번호: 값 있으면 해시 교체, 비었으면 변경 안 함.
        if data.get("password"):
            data["password"] = await GetHashedPW(data["password"])
        else:
            data.pop("password", None)
        # 빈 문자열은 NULL 로 저장(미입력 = 값 비움).
        for k in ("name", "email", "contact_number"):
            if k in data and data[k] == "":
                data[k] = None

        if data:
            err_type = await DB_SESSION_MNG.execute_lambda_run(
                [users.DBType()],
                [lambda s: self.user_crud.update_user(s, uuid.UUID(user_info.user_id), data)],
            )
            if err_type != ErrorType.SUCCESS:
                res.result.SetResult(err_type)
                return res

        # 갱신 후 최신 정보로 응답(프론트가 스토어 갱신에 사용).
        return await self.get_me(user_info)

    async def refresh_token(self, refresh_token: str) -> Res_RefreshToken:
        res = Res_RefreshToken()
        # refresh 토큰 검증은 라우터 Depends(IsValidRefreshToken) 에서 1차 수행됨.
        user_info = DecodeRefreshToken(refresh_token)
        res.access_token = CreateAccessToken(user_info)
        return res