o2o-negosium-original/negodata/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, UserStatus, UserRole
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, Res_CreateAccount, 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),
        )

    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 create_account(
        self, login_id: str, password: str, company_id: str, name: str, email: str, contact_number: str, role: int
    ) -> Res_CreateAccount:
        LOG.i(f"CREATE : id={login_id}, company_id={company_id}")
        res = Res_CreateAccount()

        # 1) 중복 ID 확인 (Read DB)
        err_type = await DB_SESSION_MNG.execute_lambda(
            users.DBType(),
            DBWRType.DB_READ.value,
            lambda s: self.user_crud.is_user(s, login_id),
        )
        if err_type == ErrorType.DB_ALREADY_SAME_KEY:
            res.result.SetResult(ErrorType.ACCOUNT_ALREADY_EXIST)
            return res
        if err_type != ErrorType.SUCCESS:
            res.result.SetResult(err_type)
            return res

        # 2) 계정 생성 (비밀번호는 bcrypt 해시로 저장)
        user = users(
            company_id=uuid.UUID(company_id),
            id=login_id,
            password=await GetHashedPW(password),
            name=name or None,
            email=email or None,
            contact_number=contact_number or None,
            role=role or UserRole.USER.value,
        )
        err_type = await DB_SESSION_MNG.execute_lambda_run(
            [users.DBType()],
            [lambda s: self.user_crud.add_user(s, user)],
        )
        if err_type != ErrorType.SUCCESS:
            # 사전 검사와 INSERT 사이의 경쟁 조건에서 unique 위반이 나면 동일 코드로 매핑.
            if err_type == ErrorType.DB_ALREADY_SAME_KEY:
                res.result.SetResult(ErrorType.ACCOUNT_ALREADY_EXIST)
            else:
                res.result.SetResult(err_type)
            return res

        res.user_id = str(user.user_id)
        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 = user.role
        res.company = company
        return res

    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