백엔드 프레임 설계

backend/.dockerignore

@@ -0,0 +1,7 @@
+__pycache__/
+*.pyc
+.pytest_cache/
+.git/
+tests/
+loadtest/
+*.md

backend/Dockerfile

@@ -0,0 +1,16 @@
+FROM python:3.12-slim
+
+WORKDIR /app
+
+# 의존성 먼저 설치 (레이어 캐시 활용)
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+# docker-compose 에서 APP_ENV=docker 로 config.docker.toml 을 읽는다.
+ENV APP_ENV=docker
+
+EXPOSE 9300
+
+CMD ["python", "web_main.py"]

backend/README.md

@@ -0,0 +1,62 @@
+# Negosium Backend
+
+DerbyMasters_Server 아키텍처를 이식한 FastAPI 골격. 기능은 **JWT id/pw 로그인**만 예시 구현.
+실행·테스트·벤치마크 종합은 [레포 최상위 README](../README.md) 참고.
+
+## 디렉토리 구조
+
+```
+backend/
+├── web_main.py                  # 엔트리포인트 (uvicorn)
+├── config/                      # 환경설정 (APP_ENV 별 toml 로드)
+├── common/
+│   ├── enums.py                 # ErrorType / DBType / DBWRType / EXCEPTION_*
+│   ├── models/gmodel.py         # 프로토콜 베이스 (WebPacketProtocol 등)
+│   └── database/
+│       ├── db_session_manager.py# ★ DB Read/Write + 람다 실행 핵심
+│       └── model/models.py      # ORM 모델 (tbl_account)
+├── crud/user_crud.py            # DB 접근 (I*CRUD 인터페이스 + 구현)
+├── services/auth_service.py     # 비즈니스 로직
+└── router/
+    ├── router.py                # FastAPI app
+    └── v1/
+        ├── auth/{account,protocol}.py   # 엔드포인트 / Req_·Res_
+        └── validator/dependencies.py    # ★ JWT 발급·검증, 해시, RemoveNoneResponse
+```
+
+## 핵심 패턴
+- **MVC**: router(컨트롤러) → service(로직) → crud(쿼리). crud 는 인터페이스/구현 분리.
+- **Depends 주입**: service 가 `IUserCRUD = Depends(UserCRUD)`, router 가 `Depends(AuthService)`, 인증은 `Depends(IsValidAccessToken)`.
+- **DB Read/Write 분리**: `DBType` × `DBWRType` 조합마다 별도 async 엔진(조회=Read, 변경=Write).
+- **람다 DB 실행**: service 는 세션을 직접 열지 않고 람다를 매니저에 넘긴다. 세션/트랜잭션은 매니저가 책임.
+  ```python
+  err, acc = await DB_SESSION_MNG.execute_lambda(           # 단일 조회
+      tbl_account.DBType(), DBWRType.DB_READ.value, lambda s: crud.get_account_by_id(s, id))
+  err = await DB_SESSION_MNG.execute_lambda_run(            # 여러 변경 = 1 트랜잭션
+      [tbl_account.DBType()], [lambda s: crud.update_last_login(s, uid)])
+  ```
+- **비동기**: 전 계층 async/await, SQLAlchemy async + asyncpg, task 단위 scoped session.
+- **Protocol 규약**: 모든 패킷 `WebPacketProtocol` 상속, `Req_*`/`Res_*`(응답엔 `result: ErrorInfo`), 라우터별 `protocol.py`.
+- **ResponseNone**: 응답의 `None` 필드 재귀 제거(`RemoveNoneResponse`).
+- **bcrypt 비차단**: `GetHashedPW`/`VerifyPW` 를 `asyncio.to_thread` 로 오프로드(이벤트 루프 비차단). → [벤치마크](../README.md#성능--벤치마크)
+
+## 엔드포인트
+| Method | Path | 설명 |
+|---|---|---|
+| POST | `/v1/auth/create` | 계정 생성 (pw bcrypt 해시) |
+| POST | `/v1/auth/login` | 로그인, access/refresh 토큰 발급 |
+| POST | `/v1/auth/refresh_token` | access 토큰 재발급 (refresh 필요) |
+| GET | `/v1/auth/me` | 내 정보 (access 토큰 필요) |
+
+## 실행 / 테스트
+```bash
+# 레포 최상위에서 docker compose up -d (postgres + 서버). 상세는 루트 README.
+cd backend
+pip install -r requirements.txt        # 실행
+python web_main.py                      # APP_ENV 기본 local
+pip install -r requirements-dev.txt     # 테스트
+python -m pytest
+```
+- 서버: http://localhost:9300/docs
+- 환경: `config.{local,test,docker}.toml` (`APP_ENV` 로 선택, docker 는 DB 호스트=`host.docker.internal`)
+

backend/common/database/db_session_manager.py

@@ -0,0 +1,198 @@
+from asyncio import current_task
+
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_scoped_session
+from sqlalchemy.util._collections import immutabledict
+
+from common.database.model.models import MAIN_BASE
+from common.enums import DBType, DBWRType, ErrorType
+from common.logger import LOG
+from common.singleton import Singleton
+from config.server_configs import main_db_config
+
+
+class DBSessionManager(Singleton):
+    """DB 세션/엔진 관리자 (싱글톤).
+
+    핵심 패턴
+    - DBType(논리 DB) x DBWRType(Read/Write) 조합마다 별도 async 엔진을 둔다.
+      => 조회는 Read 복제본, 변경은 Write 주 DB 로 자연스럽게 분리된다.
+    - 비즈니스 로직(service)은 직접 세션을 열지 않고 "람다"를 넘긴다.
+      execute_lambda        : 단일 쿼리 (주로 조회)
+      execute_lambda_run    : 동일 DB 의 여러 변경 쿼리를 한 트랜잭션으로 commit
+      세션 open/close 와 commit/rollback 은 매니저가 책임진다.
+    """
+
+    def __init__(self):
+        if DBSessionManager.is_init():
+            LOG.e_no_callstack("already init DBSessionManager")
+            return
+        DBSessionManager.set_init()
+
+        self.__DB_URL_MAP = {"postgresql": "postgresql+asyncpg"}
+        # 종료 시 dispose 하기 위해 생성한 엔진을 모아둔다.
+        self.__engines = []
+        # 논리 DB -> config. DB 가 늘어나면 여기에 추가만 하면 된다.
+        self.__db_type_map = {
+            DBType.MAIN.value: main_db_config,
+        }
+
+        # Write 엔진 맵
+        self.__write_session = {
+            DBType.MAIN.value: self.create_engine(DBType.MAIN.value, DBWRType.DB_WRITE.value),
+        }
+        # Read 엔진 맵
+        self.__read_session = {
+            DBType.MAIN.value: self.create_engine(DBType.MAIN.value, DBWRType.DB_READ.value),
+        }
+
+    def create_engine(self, db_type: int, db_wr_type: int):
+        db_config = self.__db_type_map.get(db_type)
+        if not db_config:
+            raise ValueError("Invalid database type")
+
+        if db_wr_type == DBWRType.DB_READ.value:
+            pw = (":" + db_config.read_pw) if len(db_config.read_pw) > 0 else ""
+            db_url = f"{self.__DB_URL_MAP[db_config.db_type]}://{db_config.read_id}{pw}@{db_config.read_host}:{db_config.read_port}/{db_config.name}"
+            LOG.i(f"Read DB create engine url : {db_url}")
+        else:
+            pw = (":" + db_config.write_pw) if len(db_config.write_pw) > 0 else ""
+            db_url = f"{self.__DB_URL_MAP[db_config.db_type]}://{db_config.write_id}{pw}@{db_config.write_host}:{db_config.write_port}/{db_config.name}"
+            LOG.i(f"Write DB create engine url : {db_url}")
+
+        engine = create_async_engine(
+            db_url,
+            echo=db_config.show_log,
+            pool_size=db_config.pool_size,
+            max_overflow=db_config.max_overflow,
+            pool_pre_ping=True,
+            pool_recycle=600,
+        )
+        self.__engines.append(engine)
+        scoped_session = async_scoped_session(
+            sessionmaker(engine, class_=AsyncSession, expire_on_commit=False, autocommit=False, autoflush=False),
+            scopefunc=current_task,
+        )
+        return scoped_session
+
+    async def dispose_all(self):
+        """모든 엔진의 커넥션 풀을 정리한다. 앱 종료/테스트 종료 시 호출한다.
+        호출하지 않으면 풀 커넥션이 이벤트 루프 종료 후 GC 되며 경고를 남긴다.
+        """
+        for engine in self.__engines:
+            await engine.dispose()
+
+    # ---- 세션 lifecycle -------------------------------------------------
+    async def start_session(self, db_type: int, db_wr_type: int) -> AsyncSession:
+        if db_wr_type == DBWRType.DB_WRITE.value:
+            return self.__write_session[db_type]()
+        return self.__read_session[db_type]()
+
+    async def end_session(self, db_type: int, db_wr_type: int):
+        if db_wr_type == DBWRType.DB_WRITE.value:
+            await self.__write_session[db_type].remove()
+        else:
+            await self.__read_session[db_type].remove()
+
+    # ---- 저수준 DB 연산 (crud 에서 호출) --------------------------------
+    async def run(self, db: AsyncSession, err_msg="DB Run Failed", raise_error=True) -> ErrorType:
+        try:
+            await db.commit()
+            return ErrorType.SUCCESS
+        except IntegrityError as ex:
+            await db.rollback()
+            LOG.e_no_callstack(f"duplicated. {ex}")
+            return ErrorType.DB_ALREADY_SAME_KEY
+        except Exception as ex:
+            await db.rollback()
+            err_type = ErrorType.DB_RUN_FAILED
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            if raise_error:
+                raise RuntimeError(err_type.name, err_msg)
+            return err_type
+
+    async def insert(self, db: AsyncSession, obj, err_msg="DB Failed", raise_error=True) -> ErrorType:
+        try:
+            if isinstance(obj, MAIN_BASE):
+                db.add(obj)
+            elif isinstance(obj, list):
+                db.add_all(obj)
+            else:
+                raise RuntimeError("DO NOT USE QUERY IN DBJOB")
+            return ErrorType.SUCCESS
+        except Exception as ex:
+            await db.rollback()
+            err_type = ErrorType.DB_RUN_FAILED
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            if raise_error:
+                raise RuntimeError(err_type.name, err_msg)
+            return err_type
+
+    async def add(self, db: AsyncSession, query, err_msg="DB Operation Failed", raise_error=True) -> ErrorType:
+        """update/delete 등 비-select 쿼리 실행."""
+        try:
+            if hasattr(query, "column_descriptions"):
+                raise RuntimeError("DO NOT USE SELECT QUERY IN DBJOB")
+            await db.execute(query, execution_options=immutabledict({"synchronize_session": "fetch"}))
+            return ErrorType.SUCCESS
+        except IntegrityError as ex:
+            await db.rollback()
+            err_type = ErrorType.DB_ALREADY_SAME_KEY
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            return err_type
+        except Exception as ex:
+            await db.rollback()
+            err_type = ErrorType.DB_RUN_FAILED
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            if raise_error:
+                raise RuntimeError(err_type.name, err_msg)
+            return err_type
+
+    async def execute(self, db: AsyncSession, query, err_msg="DB Query Execution Failed", raise_error=True) -> tuple[ErrorType, list]:
+        """select 쿼리 실행 후 결과 리스트 반환."""
+        try:
+            if not hasattr(query, "column_descriptions"):
+                raise RuntimeError("DO NOT USE NON-SELECT QUERY IN DBJOB")
+            res = await db.execute(query, execution_options=immutabledict({"synchronize_session": "fetch"}))
+            return ErrorType.SUCCESS, res.scalars().fetchall() if 1 == len(query.column_descriptions) else res.all()
+        except Exception as ex:
+            err_type = ErrorType.DB_RUN_FAILED
+            LOG.e_no_callstack(f"[{err_type.name}] {err_msg=}, {ex=}")
+            if raise_error:
+                raise RuntimeError(err_type.name, err_msg)
+            return err_type, []
+
+    # ---- 람다 실행 진입점 (service 에서 호출) ---------------------------
+    async def execute_lambda(self, db_type: int, db_wr_type: int, func):
+        """단일 쿼리 호출. func(session) 한 개를 실행하고 결과를 그대로 반환."""
+        s = await self.start_session(db_type, db_wr_type)
+        try:
+            return await func(s)
+        finally:
+            await self.end_session(db_type, db_wr_type)
+
+    async def execute_lambda_run(self, db_type_list: list[int], func_list: list):
+        """동일 DB 의 변경 쿼리 여러 개를 한 트랜잭션으로 실행 후 commit.
+        하나라도 SUCCESS 가 아니면 즉시 중단(rollback)된다.
+        """
+        temp_list = list(set(db_type_list))
+        if len(temp_list) != 1:
+            return ErrorType.DB_INVALID_TYPE
+
+        db_type = temp_list[0]
+        s = await self.start_session(db_type, DBWRType.DB_WRITE.value)
+        try:
+            for func in func_list:
+                err_type = await func(s)
+                if err_type != ErrorType.SUCCESS:
+                    return err_type
+            return await self.run(s)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED
+        finally:
+            await self.end_session(db_type, DBWRType.DB_WRITE.value)
+
+
+DB_SESSION_MNG = DBSessionManager()

backend/common/database/model/models.py

@@ -0,0 +1,26 @@
+from sqlalchemy.orm import declarative_base
+from sqlalchemy import Column, Integer, String, Boolean, DateTime
+from sqlalchemy.sql import text
+
+from common.enums import DBType
+
+# 모든 ORM 모델의 베이스. insert 시 isinstance 체크에도 사용된다.
+MAIN_BASE = declarative_base()
+
+
+class tbl_account(MAIN_BASE):
+    # 모델이 자신이 속한 논리 DB 를 알려준다 (람다 실행 시 DBType 으로 세션 선택).
+    @staticmethod
+    def DBType():
+        return DBType.MAIN.value
+
+    __tablename__ = "tbl_account"
+
+    uid = Column(Integer, primary_key=True, autoincrement=True)
+    id = Column(String(45), nullable=False, unique=True)  # 로그인 ID. 중복 가입 방지 위해 unique.
+    pw = Column(String(255), nullable=False, default="")  # bcrypt 해시 저장
+    nickname = Column(String(45), nullable=False, default="")
+    is_blocked = Column(Boolean, nullable=False, default=False)
+    # PostgreSQL UTC now: now() 는 timestamptz 이므로 utc 로 변환해 timestamp 로 저장.
+    last_login_at = Column(DateTime, nullable=False, server_default=text("(now() AT TIME ZONE 'utc')"))
+    create_at = Column(DateTime, server_default=text("(now() AT TIME ZONE 'utc')"))

backend/common/enums.py

@@ -0,0 +1,61 @@
+from enum import Enum, auto
+
+from fastapi import HTTPException
+
+
+class ErrorType(Enum):
+    """서버 전역 결과 코드. Res_WebPacketProtocol.result 에 담겨 클라이언트로 전달된다.
+    HTTP status 와 겹치지 않도록 구간을 분리해서 관리한다.
+    """
+
+    SUCCESS = 0
+    FAIL = 1
+
+    # DB / Redis 에러
+    DB_RUN_FAILED = 10
+    DB_ALREADY_SAME_KEY = auto()
+    DB_INVALID_KEY = auto()
+    DB_EMPTY_DATA = auto()
+    DB_INVALID_TYPE = auto()
+
+    # 요청/직렬화 에러
+    JSON_PARSE_ERROR = 100
+    INVALID_REQUEST_DATA = auto()
+    INTERNAL_EXCEPTION = auto()
+
+    # http 에러 코드와 겹치지 않게 설정 - router 전용 예외 발생 옵션
+    HTTP_INVALID_CLIENT_REQUEST = 419
+    HTTP_TO_MANY_REQUEST = 429
+    HTTP_INVALID_CLIENT_ACCESS = 433
+    HTTP_ACCESS_TOKEN_EXPIRED = 434
+    HTTP_REFRESH_TOKEN_EXPIRED = 435
+    HTTP_INVALID_TOKEN_ACCESS = 436
+
+    # 계정 관련 에러
+    ACCOUNT_INVALID_INFO = 1200
+    ACCOUNT_ALREADY_EXIST = auto()
+    ACCOUNT_BLOCKED_USER = auto()
+
+
+# ErrorType 의 HTTP_* 값과 status_code 를 맞춰 router 단에서 raise 한다.
+EXCEPTION_INVALID_CLIENT_REQUEST = HTTPException(status_code=ErrorType.HTTP_INVALID_CLIENT_REQUEST.value, detail=ErrorType.HTTP_INVALID_CLIENT_REQUEST.name)
+EXCEPTION_TO_MANY_REQUEST = HTTPException(status_code=ErrorType.HTTP_TO_MANY_REQUEST.value, detail=ErrorType.HTTP_TO_MANY_REQUEST.name)
+EXCEPTION_INVALID_CLIENT_ACCESS = HTTPException(status_code=ErrorType.HTTP_INVALID_CLIENT_ACCESS.value, detail=ErrorType.HTTP_INVALID_CLIENT_ACCESS.name)
+EXCEPTION_ACCESS_TOKEN_EXPIRED = HTTPException(status_code=ErrorType.HTTP_ACCESS_TOKEN_EXPIRED.value, detail=ErrorType.HTTP_ACCESS_TOKEN_EXPIRED.name)
+EXCEPTION_REFRESH_TOKEN_EXPIRED = HTTPException(status_code=ErrorType.HTTP_REFRESH_TOKEN_EXPIRED.value, detail=ErrorType.HTTP_REFRESH_TOKEN_EXPIRED.name)
+EXCEPTION_HTTP_INVALID_TOKEN_ACCESS = HTTPException(status_code=ErrorType.HTTP_INVALID_TOKEN_ACCESS.value, detail=ErrorType.HTTP_INVALID_TOKEN_ACCESS.name)
+
+
+class DBType(Enum):
+    """논리 DB 구분. 모델마다 DBType() 으로 자신이 속한 DB 를 반환한다.
+    DB 가 늘어나면 여기에 추가하고 db_session_manager 의 맵에 등록만 하면 된다.
+    """
+
+    MAIN = 1
+
+
+class DBWRType(Enum):
+    """Read / Write 접속 구분. 조회는 DB_READ, 변경은 DB_WRITE 엔진을 사용한다."""
+
+    DB_READ = 1
+    DB_WRITE = 2

backend/common/logger.py

@@ -0,0 +1,43 @@
+import sys
+import traceback
+from datetime import datetime, timezone
+
+
+class _Logger:
+    """원본 DerbyServer LOG 인터페이스를 간소화한 버전.
+    LOG.i / LOG.d / LOG.w / LOG.e_no_callstack / LOG.SetPrefix 를 제공한다.
+    """
+
+    def __init__(self):
+        self._prefix = ""
+
+    def SetPrefix(self, prefix: str):
+        self._prefix = prefix
+
+    def _now(self) -> str:
+        return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
+
+    def _write(self, level: str, msg):
+        head = f"{self._now()} [{level}]"
+        if self._prefix:
+            head += f"[{self._prefix}]"
+        print(f"{head} {msg}", file=sys.stderr if level in ("WARN", "ERROR") else sys.stdout)
+
+    def d(self, msg):
+        self._write("DEBUG", msg)
+
+    def i(self, msg):
+        self._write("INFO", msg)
+
+    def w(self, msg):
+        self._write("WARN", msg)
+
+    def e(self, msg):
+        self._write("ERROR", msg)
+        traceback.print_stack()
+
+    def e_no_callstack(self, msg):
+        self._write("ERROR", msg)
+
+
+LOG = _Logger()

backend/common/models/gmodel.py

@@ -0,0 +1,64 @@
+import json
+from typing import Optional
+
+from pydantic import BaseModel, Field
+
+from common.enums import ErrorType
+
+
+class StructModel:
+    """프로토콜/구조체 식별용 마커 클래스."""
+
+    pass
+
+
+class ErrorInfo(BaseModel, StructModel):
+    """모든 응답에 공통으로 실리는 결과 정보. result.success / code / desc 로 내려간다."""
+
+    success: Optional[bool] = True
+    code: Optional[int] = ErrorType.SUCCESS.value
+    desc: Optional[str] = ErrorType.SUCCESS.name
+
+    def SetResult(self, enum: ErrorType):
+        if enum is not None:
+            self.success = ErrorType.SUCCESS.value == enum.value
+            self.code = enum.value
+            self.desc = enum.name
+
+
+# ---- Protocol 규약 -------------------------------------------------------
+# 모든 통신 패킷은 WebPacketProtocol 을 상속한다.
+# 요청   : Req_xxx (WebPacketProtocol)
+# 응답   : Res_xxx (Res_WebPacketProtocol) - 항상 result 필드를 가진다.
+# 각 라우터 폴더의 protocol.py 에 Req_/Res_ 를 정의한다.
+class WebPacketProtocol(BaseModel, StructModel):
+    pass
+
+
+class Req_WebPacketProtocol(WebPacketProtocol):
+    pass
+
+
+class Res_WebPacketProtocol(WebPacketProtocol):
+    # default_factory 로 인스턴스마다 새 ErrorInfo 를 생성한다 (mutable default 공유 방지).
+    result: ErrorInfo = Field(default_factory=ErrorInfo)
+    msg: Optional[str] = None
+
+
+class UserInfo(StructModel):
+    """JWT subject 로 인코딩되는 유저 식별 정보."""
+
+    uid: int
+    id: str
+    nickname: str
+
+    def __init__(self, *args, **kwargs) -> None:
+        super().__init__()
+        for dictionary in args:
+            for key in dictionary:
+                setattr(self, key, dictionary[key])
+        for key in kwargs:
+            setattr(self, key, kwargs[key])
+
+    def to_json(self) -> str:
+        return json.dumps(self.__dict__)

backend/common/singleton.py

@@ -0,0 +1,16 @@
+class Singleton:
+    _init = False
+
+    def __new__(cls, *args, **kwargs):
+        if not hasattr(cls, "instance"):
+            cls.instance = super(Singleton, cls).__new__(cls)
+
+        return cls.instance
+
+    @classmethod
+    def is_init(cls):
+        return cls._init
+
+    @classmethod
+    def set_init(cls):
+        cls._init = True

backend/common/utils/gtime.py

@@ -0,0 +1,21 @@
+from datetime import datetime, timezone, timedelta
+
+
+class GTime:
+    """서버 전역에서 UTC 기준 시간을 사용하기 위한 유틸. (원본 DerbyServer 패턴 축약)"""
+
+    @staticmethod
+    def UTC() -> datetime:
+        return datetime.now(timezone.utc).replace(tzinfo=None)
+
+    @staticmethod
+    def UTCStr(fmt: str = "%Y-%m-%d %H:%M:%S") -> str:
+        return GTime.UTC().strftime(fmt)
+
+    @staticmethod
+    def AddMinutes(minutes: int) -> datetime:
+        return GTime.UTC() + timedelta(minutes=minutes)
+
+    @staticmethod
+    def AddDays(days: int) -> datetime:
+        return GTime.UTC() + timedelta(days=days)

backend/config/config.docker.toml

@@ -0,0 +1,33 @@
+[WebServerConfig]
+server_name = "NegosiumServer"
+port = 9300
+process_count = 4
+is_ssl = false
+is_test = true
+
+[LogConfig]
+print_console = true
+log_level = "debug"
+
+# docker-compose 네트워크에서는 DB 가 compose 밖에 있으므로 호스트의 DB 에 host.docker.internal 로 접근한다.
+[MainDBConfig]
+db_type = "postgresql"
+name = "negosium_db"
+write_host = "host.docker.internal"
+write_port = 5432
+write_id = "postgres"
+write_pw = "password"
+read_host = "host.docker.internal"
+read_port = 5432
+read_id = "postgres"
+read_pw = "password"
+show_log = false
+# 워커 4개 기준: (pool_size + max_overflow) x 2(R/W) x 4 = 320 < max_connections(500)
+pool_size = 20
+max_overflow = 20
+
+[JwtToken]
+access_key = "docker_access_secret_key"
+refresh_key = "docker_refresh_secret_key"
+access_expire_min = 30
+refresh_expire_day = 7

backend/config/config.local.toml

@@ -0,0 +1,32 @@
+[WebServerConfig]
+server_name = "NegosiumServer"
+port = 9300
+process_count = 1
+is_ssl = false
+is_test = true
+
+[LogConfig]
+print_console = true
+log_level = "debug"
+
+# DB Read/Write 분리. 단일 DB 환경에서는 read/write 동일 호스트로 설정.
+[MainDBConfig]
+db_type = "postgresql"
+name = "negosium_db"
+write_host = "127.0.0.1"
+write_port = 5432
+write_id = "postgres"
+write_pw = "password"
+read_host = "127.0.0.1"
+read_port = 5432
+read_id = "postgres"
+read_pw = "password"
+show_log = false
+pool_size = 100
+max_overflow = 200
+
+[JwtToken]
+access_key = "CHANGE_ME_ACCESS_SECRET_KEY"
+refresh_key = "CHANGE_ME_REFRESH_SECRET_KEY"
+access_expire_min = 30
+refresh_expire_day = 7

backend/config/config.test.toml

@@ -0,0 +1,33 @@
+[WebServerConfig]
+server_name = "NegosiumServer-Test"
+port = 9301
+process_count = 1
+is_ssl = false
+is_test = true
+
+[LogConfig]
+print_console = true
+log_level = "debug"
+
+# 테스트는 단일 postgres 의 negosium_db 를 사용한다.
+# pytest 가 tbl_account 를 TRUNCATE 로 비워 격리하므로, 운영 데이터가 있다면 주의.
+[MainDBConfig]
+db_type = "postgresql"
+name = "negosium_db"
+write_host = "127.0.0.1"
+write_port = 5432
+write_id = "postgres"
+write_pw = "password"
+read_host = "127.0.0.1"
+read_port = 5432
+read_id = "postgres"
+read_pw = "password"
+show_log = false
+pool_size = 100
+max_overflow = 200
+
+[JwtToken]
+access_key = "test_access_secret_key"
+refresh_key = "test_refresh_secret_key"
+access_expire_min = 30
+refresh_expire_day = 7

backend/config/config_loader.py

@@ -0,0 +1,37 @@
+import tomllib
+from typing import Optional, Type, Dict, TypeVar
+from pydantic import BaseModel
+
+
+class ConfigModel(BaseModel):
+    pass
+
+
+# APP_ENV
+# local : 로컬 환경(개인 pc)
+# dev   : 개발환경 (사내 pc)
+# prod  : 서비스 환경 (클라우드 서버)
+#
+# 실행시 환경변수 설정
+# linux  : export APP_ENV=dev
+# window : set APP_ENV=dev
+class Configs:
+    ConfigType = TypeVar("ConfigType", bound=ConfigModel)
+
+    def __init__(self, file_path: str):
+        self._settings: Dict[Type["Configs.ConfigType"], "Configs.ConfigType"] = self._load_settings_from_toml(file_path)
+
+    def _load_settings_from_toml(self, file_path: str) -> Dict[Type[ConfigType], ConfigType]:
+        with open(file_path, "rb") as f:
+            toml_content = tomllib.load(f)
+
+        config_subclasses = ConfigModel.__subclasses__()
+        configs = {
+            config_class: config_class.model_validate(toml_content[config_class.__name__])
+            for config_class in config_subclasses
+            if config_class.__name__ in toml_content
+        }
+        return configs
+
+    def get(self, config_class: Type[ConfigType]) -> Optional[ConfigType]:
+        return self._settings.get(config_class)

backend/config/config_models.py

@@ -0,0 +1,41 @@
+from config.config_loader import ConfigModel
+
+
+class WebServerConfig(ConfigModel):
+    server_name: str = ""
+    port: int = 0
+    process_count: int = 1
+    is_ssl: bool = False
+    is_test: bool = False
+
+
+class LogConfig(ConfigModel):
+    print_console: bool = True
+    log_level: str = "debug"
+
+
+# DB Read/Write 분리 설정.
+# 하나의 논리 DB 에 대해 write(주) / read(복제) 접속 정보를 각각 가진다.
+class MainDBConfig(ConfigModel):
+    db_type: str = "postgresql"
+    name: str = ""
+    write_host: str = ""
+    write_port: int = 5432
+    write_id: str = ""
+    write_pw: str = ""
+    read_host: str = ""
+    read_port: int = 5432
+    read_id: str = ""
+    read_pw: str = ""
+    show_log: bool = False
+    # 커넥션 풀 사이징. 실제 동시 커넥션 상한 = (pool_size + max_overflow) x 엔진수(R/W=2) x 워커수.
+    # PostgreSQL max_connections 를 넘지 않도록 설정해야 한다. (예: 10+20=30 x 2 x 5워커 = 300)
+    pool_size: int = 10
+    max_overflow: int = 20
+
+
+class JwtToken(ConfigModel):
+    access_key: str = ""
+    refresh_key: str = ""
+    access_expire_min: int = 30
+    refresh_expire_day: int = 7

backend/config/server_configs.py

@@ -0,0 +1,20 @@
+import os
+
+from config.config_loader import Configs
+from config.config_models import WebServerConfig, LogConfig, MainDBConfig, JwtToken
+
+# 실행 환경 결정 (기본 local). 환경변수 APP_ENV 로 변경.
+APP_ENV = os.environ.get("APP_ENV", "local")
+
+_config_dir = os.path.dirname(__file__)
+_config_file = os.path.join(_config_dir, f"config.{APP_ENV}.toml")
+
+if not os.path.exists(_config_file):
+    raise FileNotFoundError(f"설정 파일이 없습니다: {_config_file} (APP_ENV={APP_ENV}). config.<env>.toml 을 생성하세요.")
+
+configs = Configs(_config_file)
+
+web_server_config: WebServerConfig = configs.get(WebServerConfig)
+log_config: LogConfig = configs.get(LogConfig)
+main_db_config: MainDBConfig = configs.get(MainDBConfig)
+jwt_token_config: JwtToken = configs.get(JwtToken)

backend/conftest.py

@@ -0,0 +1,54 @@
+# pytest 진입 시점에 가장 먼저 APP_ENV=test 를 설정해야 한다.
+# (config.server_configs 가 import 되는 순간 config.<APP_ENV>.toml 을 읽기 때문)
+import os
+
+os.environ.setdefault("APP_ENV", "test")
+
+import pytest_asyncio
+from httpx import ASGITransport, AsyncClient
+from sqlalchemy import text
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from common.database.model.models import MAIN_BASE
+from config.server_configs import main_db_config
+
+
+def _write_url(cfg) -> str:
+    pw = f":{cfg.write_pw}" if cfg.write_pw else ""
+    return f"postgresql+asyncpg://{cfg.write_id}{pw}@{cfg.write_host}:{cfg.write_port}/{cfg.name}"
+
+
+@pytest_asyncio.fixture
+async def db_engine():
+    """테스트용 스키마를 보장하고, 매 테스트 시작 시 테이블을 비워 격리한다.
+
+    앱(DB_SESSION_MNG)은 자체 엔진으로 같은 DB(config.test.toml)에 접속하므로,
+    여기서 만든 스키마를 그대로 공유한다.
+    """
+    engine = create_async_engine(_write_url(main_db_config))
+    async with engine.begin() as conn:
+        await conn.run_sync(MAIN_BASE.metadata.create_all)  # 이미 있으면 skip
+        await conn.execute(text("TRUNCATE TABLE tbl_account"))
+    yield engine
+    await engine.dispose()
+
+
+@pytest_asyncio.fixture(scope="session", autouse=True)
+async def _dispose_app_engines():
+    """테스트 세션이 끝날 때 앱 싱글톤 엔진을 정리한다.
+    (이벤트 루프 종료 후 커넥션이 GC 되며 나오는 'Event loop is closed' 경고 제거)
+    """
+    yield
+    from common.database.db_session_manager import DB_SESSION_MNG
+
+    await DB_SESSION_MNG.dispose_all()
+
+
+@pytest_asyncio.fixture
+async def client(db_engine):
+    """앱을 실제 네트워크 없이 호출하는 httpx 클라이언트 (ASGITransport)."""
+    from router.router import app
+
+    transport = ASGITransport(app=app)
+    async with AsyncClient(transport=transport, base_url="http://test") as ac:
+        yield ac

backend/crud/user_crud.py

@@ -0,0 +1,75 @@
+from abc import ABC, abstractmethod
+from typing import Tuple
+
+from sqlalchemy import select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.database.model.models import tbl_account
+from common.enums import ErrorType
+from common.logger import LOG
+from common.utils.gtime import GTime
+
+
+# CRUD 는 인터페이스(I*) 와 구현(*) 으로 분리한다.
+# - service 는 인터페이스 타입에 의존하고 Depends 로 구현을 주입받는다 (테스트/교체 용이).
+# - 모든 메서드는 (session, ...) 을 받는다. session 은 람다 호출 시 매니저가 넘겨준다.
+class IUserCRUD(ABC):
+    @abstractmethod
+    async def get_account_by_id(self, cdb: AsyncSession, user_id: str) -> Tuple[ErrorType, tbl_account]:
+        pass
+
+    @abstractmethod
+    async def is_account(self, cdb: AsyncSession, user_id: str) -> ErrorType:
+        pass
+
+    @abstractmethod
+    async def add_account(self, cdb: AsyncSession, account: tbl_account) -> ErrorType:
+        pass
+
+    @abstractmethod
+    async def update_last_login(self, cdb: AsyncSession, user_uid: int) -> ErrorType:
+        pass
+
+
+class UserCRUD(IUserCRUD):
+    async def get_account_by_id(self, cdb: AsyncSession, user_id: str) -> Tuple[ErrorType, tbl_account]:
+        try:
+            query = select(tbl_account).where(tbl_account.id == user_id).limit(1)
+            err_type, row_list = await DB_SESSION_MNG.execute(cdb, query, f"get_account_by_id(ID:{user_id}) failed.")
+            if err_type != ErrorType.SUCCESS:
+                return err_type, None
+            if len(row_list) != 1:
+                return ErrorType.DB_INVALID_KEY, None
+            return ErrorType.SUCCESS, row_list[0]
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED, None
+
+    async def is_account(self, cdb: AsyncSession, user_id: str) -> ErrorType:
+        try:
+            query = select(tbl_account).where(tbl_account.id == user_id).limit(1)
+            err_type, row_list = await DB_SESSION_MNG.execute(cdb, query)
+            if err_type != ErrorType.SUCCESS:
+                return err_type
+            if row_list:
+                return ErrorType.DB_ALREADY_SAME_KEY
+            return ErrorType.SUCCESS
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED
+
+    async def add_account(self, cdb: AsyncSession, account: tbl_account) -> ErrorType:
+        try:
+            return await DB_SESSION_MNG.insert(cdb, account)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED
+
+    async def update_last_login(self, cdb: AsyncSession, user_uid: int) -> ErrorType:
+        try:
+            query = update(tbl_account).where(tbl_account.uid == user_uid).values(last_login_at=GTime.UTC())
+            return await DB_SESSION_MNG.add(cdb, query)
+        except Exception as ex:
+            LOG.e_no_callstack(ex)
+            return ErrorType.DB_RUN_FAILED

backend/loadtest/locustfile.py

@@ -0,0 +1,80 @@
+"""Negosium 인증 서버 부하 테스트.
+
+실행:
+    pip install locust
+    locust -f loadtest/locustfile.py --host http://localhost:9300
+    # 웹 UI: http://localhost:8089 에서 사용자 수/spawn rate 입력
+
+    # 헤드리스(CI) 예시 - 100 VU, 10/s 증가, 2분:
+    locust -f loadtest/locustfile.py --host http://localhost:9300 \
+           --headless -u 100 -r 10 -t 2m
+
+주의:
+- /login 은 bcrypt 검증(CPU 바운드)이 들어가 가장 무겁다. RPS 가 낮으면
+  거의 확실히 bcrypt cost 가 병목이다 (DB 아님).
+- 부하를 올리며 서버 측에서 PostgreSQL 커넥션 수를 함께 모니터링하라:
+      SELECT count(*) FROM pg_stat_activity;
+  (pool_size + max_overflow) x 2(R/W) x 워커수 가 max_connections 를 넘으면
+  max_connections 초과 시 실패한다.
+"""
+
+import random
+
+from locust import HttpUser, between, events, task
+
+
+# 각 시뮬레이션 유저는 고유 계정을 만들어 로그인 흐름을 반복한다.
+class AuthUser(HttpUser):
+    wait_time = between(0.5, 2.0)
+
+    def on_start(self):
+        # 유저별 고유 계정 생성 후 1회 로그인하여 토큰 확보.
+        self.user_id = f"load_{random.randint(0, 1_000_000_000)}"
+        self.password = "pw1234"
+        self.token = None
+
+        self.client.post(
+            "/v1/auth/create",
+            json={"id": self.user_id, "pw": self.password, "nickname": "load"},
+            name="POST /v1/auth/create",
+        )
+        self._login()
+
+    def _login(self):
+        with self.client.post(
+            "/v1/auth/login",
+            json={"id": self.user_id, "pw": self.password},
+            name="POST /v1/auth/login",
+            catch_response=True,
+        ) as resp:
+            if resp.status_code == 200 and resp.json().get("result", {}).get("success"):
+                self.token = resp.json().get("access_token")
+                resp.success()
+            else:
+                resp.failure(f"login failed: {resp.status_code} {resp.text[:120]}")
+
+    @task(5)
+    def me(self):
+        # JWT 검증만 하는 경량 경로 (DB 無). bcrypt 경로와 처리량 비교용.
+        if not self.token:
+            return
+        self.client.get(
+            "/v1/auth/me",
+            headers={"Authorization": f"Bearer {self.token}"},
+            name="GET /v1/auth/me",
+        )
+
+    @task(2)
+    def login(self):
+        # bcrypt + DB write 가 포함된 무거운 경로.
+        self._login()
+
+    @task(1)
+    def healthz(self):
+        # 베이스라인 (앱 오버헤드 측정).
+        self.client.get("/healthz", name="GET /healthz")
+
+
+@events.test_start.add_listener
+def _on_start(environment, **kwargs):
+    print("부하 테스트 시작 - PostgreSQL 커넥션 수 모니터링 권장 (pg_stat_activity)")

backend/pytest.ini

@@ -0,0 +1,9 @@
+[pytest]
+asyncio_mode = auto
+# DB_SESSION_MNG(싱글톤)의 커넥션 풀이 첫 이벤트 루프에 묶이므로,
+# 모든 테스트/픽스처가 단일 session 루프를 공유하게 한다.
+asyncio_default_fixture_loop_scope = session
+asyncio_default_test_loop_scope = session
+testpaths = tests
+filterwarnings =
+    ignore::pytest.PytestUnraisableExceptionWarning

backend/requirements-dev.txt

@@ -0,0 +1,9 @@
+-r requirements.txt
+
+# 테스트
+pytest
+pytest-asyncio
+httpx
+
+# 부하 테스트
+locust

backend/requirements.txt

@@ -0,0 +1,8 @@
+fastapi
+uvicorn[standard]
+sqlalchemy>=2.0
+asyncpg
+python-jose[cryptography]
+bcrypt
+orjson
+pydantic>=2.0

backend/router/router.py

@@ -0,0 +1,44 @@
+import time
+from contextlib import asynccontextmanager
+
+from fastapi import FastAPI, Request
+from fastapi.middleware.gzip import GZipMiddleware
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.logger import LOG
+from common.utils.gtime import GTime
+import router.v1.auth.account
+
+API_SERVER_START_TIME = GTime.UTCStr()
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # startup
+    yield
+    # shutdown: DB 엔진 커넥션 풀 정리
+    await DB_SESSION_MNG.dispose_all()
+
+
+app = FastAPI(title="Negosium Api Server", lifespan=lifespan)
+
+# Accept-Encoding: gzip 요청에 대해 1000 bytes 이상 응답을 압축.
+app.add_middleware(GZipMiddleware, minimum_size=1000)
+
+
+@app.middleware("http")
+async def log_time(request: Request, call_next):
+    start_time = time.time()
+    response = await call_next(request)
+    elapsed = time.time() - start_time
+    LOG.d(f"took: {elapsed:.4f} - {request.url.path}")
+    return response
+
+
+@app.get(path="/healthz", responses={404: {"description": "Not found"}})
+async def healthz():
+    return API_SERVER_START_TIME
+
+
+# 각 도메인 라우터를 등록한다. 새 기능 추가 시 router.v1.<domain>.<file> 를 import 후 include.
+app.include_router(router.v1.auth.account.router)

backend/router/v1/auth/account.py

@@ -0,0 +1,42 @@
+from fastapi import APIRouter, Depends, Request
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+from common.models.gmodel import UserInfo
+from router.v1.validator.dependencies import IsValidAccessToken, IsValidRefreshToken, RemoveNoneResponse
+from services.auth_service import AuthService
+from .protocol import Req_CreateAccount, Req_Login, Res_CreateAccount, Res_Login, Res_RefreshToken
+
+security = HTTPBearer()
+
+# 라우터(MVC 의 컨트롤러). 요청 검증 -> service 호출 -> RemoveNoneResponse 반환만 담당.
+router = APIRouter(prefix="/v1/auth", tags=["Auth"], responses={404: {"description": "Not found"}})
+
+
+@router.post(path="/login", response_model=Res_Login, summary="로그인", description="id/pw 로 로그인하고 JWT 토큰을 발급한다.")
+async def login(request: Request, req: Req_Login, service: AuthService = Depends()):
+    return RemoveNoneResponse(await service.attempt_login(req.id, req.pw, request.client.host))
+
+
+@router.post(path="/create", response_model=Res_CreateAccount, summary="계정 생성", description="새 계정을 생성한다.")
+async def create_account(request: Request, req: Req_CreateAccount, service: AuthService = Depends()):
+    return RemoveNoneResponse(await service.create_account(req.id, req.pw, req.nickname, request.client.host))
+
+
+@router.post(
+    path="/refresh_token",
+    dependencies=[Depends(IsValidRefreshToken)],
+    response_model=Res_RefreshToken,
+    summary="액세스 토큰 갱신",
+    description="refresh 토큰으로 access 토큰을 재발급한다.",
+)
+async def refresh_token(service: AuthService = Depends(), credentials: HTTPAuthorizationCredentials = Depends(security)):
+    return RemoveNoneResponse(await service.refresh_token(credentials.credentials))
+
+
+@router.get(
+    path="/me",
+    summary="내 정보 (보호된 엔드포인트 예시)",
+    description="유효한 access 토큰이 있어야 호출 가능. 토큰 검증 결과 UserInfo 를 주입받는다.",
+)
+async def me(user_info: UserInfo = Depends(IsValidAccessToken)):
+    return {"uid": user_info.uid, "id": user_info.id, "nickname": user_info.nickname}

backend/router/v1/auth/protocol.py

@@ -0,0 +1,34 @@
+from pydantic import Field
+
+from common.models.gmodel import Res_WebPacketProtocol, WebPacketProtocol
+
+
+# 라우터 폴더마다 protocol.py 를 두고 Req_/Res_ 를 정의한다 (protocol 규약).
+class AuthProtocol(WebPacketProtocol):
+    pass
+
+
+class Req_Login(AuthProtocol):
+    id: str = ""
+    pw: str = ""
+
+
+class Res_Login(Res_WebPacketProtocol):
+    uid: int = Field(0, description="user uid", json_schema_extra={"format": "int64"})
+    nickname: str = ""
+    access_token: str = ""
+    refresh_token: str = ""
+
+
+class Req_CreateAccount(AuthProtocol):
+    id: str = ""
+    pw: str = ""
+    nickname: str = ""
+
+
+class Res_CreateAccount(Res_WebPacketProtocol):
+    uid: int = Field(0, description="생성된 user uid", json_schema_extra={"format": "int64"})
+
+
+class Res_RefreshToken(Res_WebPacketProtocol):
+    access_token: str = ""

backend/router/v1/validator/dependencies.py

@@ -0,0 +1,113 @@
+import asyncio
+import json
+from typing import Any, Union
+
+from fastapi import Depends
+from fastapi.responses import ORJSONResponse
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+import bcrypt
+from jose import jwt, JWTError, ExpiredSignatureError
+
+from common.enums import (
+    EXCEPTION_ACCESS_TOKEN_EXPIRED,
+    EXCEPTION_INVALID_CLIENT_ACCESS,
+    EXCEPTION_REFRESH_TOKEN_EXPIRED,
+)
+from common.logger import LOG
+from common.models.gmodel import UserInfo
+from common.utils.gtime import GTime
+from config.server_configs import jwt_token_config
+
+security = HTTPBearer()
+
+
+# ---- 비밀번호 해시 (bcrypt) ------------------------------------------------
+# bcrypt 는 CPU 바운드 동기 작업이라 그대로 호출하면 asyncio 이벤트 루프를 막아
+# 같은 워커의 다른 요청(healthz 등)까지 멈춘다. 스레드풀(asyncio.to_thread)로 보낸다.
+# bcrypt 는 해싱 중 GIL 을 해제하므로 스레드들이 여러 코어에서 실제 병렬로 돈다.
+# 입력은 최대 72 bytes 까지만 사용하므로 사전에 잘라준다.
+def _hash_pw(pw: str) -> str:
+    return bcrypt.hashpw(pw.encode("utf-8")[:72], bcrypt.gensalt()).decode("utf-8")
+
+
+def _verify_pw(pw: str, hashed_pw: str) -> bool:
+    try:
+        return bcrypt.checkpw(pw.encode("utf-8")[:72], hashed_pw.encode("utf-8"))
+    except (ValueError, TypeError):
+        return False
+
+
+async def GetHashedPW(pw: str) -> str:
+    return await asyncio.to_thread(_hash_pw, pw)
+
+
+async def VerifyPW(pw: str, hashed_pw: str) -> bool:
+    return await asyncio.to_thread(_verify_pw, pw, hashed_pw)
+
+
+# ---- JWT 토큰 발급/검증 ----------------------------------------------------
+JWT_ALGORITHM = "HS256"
+JWT_ACCESS_SECRET = jwt_token_config.access_key
+JWT_REFRESH_SECRET = jwt_token_config.refresh_key
+ACCESS_TOKEN_EXPIRE_MIN = jwt_token_config.access_expire_min
+REFRESH_TOKEN_EXPIRE_MIN = 60 * 24 * jwt_token_config.refresh_expire_day
+
+
+def __create_token(subject: Union[str, Any], secret_key: str, expire_min: int) -> str:
+    to_encode = {
+        "sub": str(subject),
+        "exp": GTime.AddMinutes(expire_min),  # jose 가 exp 클레임을 자동 검증
+    }
+    return jwt.encode(to_encode, secret_key, JWT_ALGORITHM)
+
+
+def CreateAccessToken(subject: UserInfo) -> str:
+    return __create_token(subject.to_json(), JWT_ACCESS_SECRET, ACCESS_TOKEN_EXPIRE_MIN)
+
+
+def CreateRefreshToken(subject: UserInfo) -> str:
+    return __create_token(subject.to_json(), JWT_REFRESH_SECRET, REFRESH_TOKEN_EXPIRE_MIN)
+
+
+def __decode_token(jwt_token: str, secret_key: str, expired_exception) -> UserInfo:
+    try:
+        decoded = jwt.decode(jwt_token, secret_key, algorithms=[JWT_ALGORITHM])
+        return UserInfo(**json.loads(decoded.get("sub")))
+    except ExpiredSignatureError:
+        raise expired_exception
+    except JWTError as ex:
+        LOG.e_no_callstack(ex)
+        raise EXCEPTION_INVALID_CLIENT_ACCESS
+
+
+def DecodeAccessToken(jwt_token: str) -> UserInfo:
+    return __decode_token(jwt_token, JWT_ACCESS_SECRET, EXCEPTION_ACCESS_TOKEN_EXPIRED)
+
+
+def DecodeRefreshToken(jwt_token: str) -> UserInfo:
+    return __decode_token(jwt_token, JWT_REFRESH_SECRET, EXCEPTION_REFRESH_TOKEN_EXPIRED)
+
+
+# ---- Depends 용 토큰 검증기 ------------------------------------------------
+# 보호된 엔드포인트에서 dependencies=[Depends(IsValidAccessToken)] 로 사용.
+async def IsValidAccessToken(credentials: HTTPAuthorizationCredentials = Depends(security)) -> UserInfo:
+    return DecodeAccessToken(credentials.credentials)
+
+
+async def IsValidRefreshToken(credentials: HTTPAuthorizationCredentials = Depends(security)) -> UserInfo:
+    return DecodeRefreshToken(credentials.credentials)
+
+
+# ---- ResponseNone 처리 -----------------------------------------------------
+# 응답 객체에서 값이 None 인 필드를 재귀적으로 제거하여 페이로드를 줄인다.
+# 모든 라우터는 return RemoveNoneResponse(await service....) 형태로 반환한다.
+def RemoveNoneValues(obj: Any) -> Any:
+    if isinstance(obj, dict):
+        return {k: RemoveNoneValues(v) for k, v in obj.items() if v is not None}
+    if isinstance(obj, list):
+        return [RemoveNoneValues(v) for v in obj]
+    return obj
+
+
+def RemoveNoneResponse(obj) -> ORJSONResponse:
+    return ORJSONResponse(content=RemoveNoneValues(obj.model_dump()))

backend/services/auth_service.py

@@ -0,0 +1,115 @@
+from fastapi import Depends
+
+from common.database.db_session_manager import DB_SESSION_MNG
+from common.database.model.models import tbl_account
+from common.enums import DBWRType, ErrorType
+from common.logger import LOG
+from common.models.gmodel import UserInfo
+from crud.user_crud import IUserCRUD, UserCRUD
+from router.v1.auth.protocol import Res_CreateAccount, Res_Login, 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
+
+    async def attempt_login(self, id: str, pw: str, connect_ip: str) -> Res_Login:
+        LOG.i(f"LOGIN : {id=}")
+        res = Res_Login()
+
+        # 1) 계정 조회 (Read DB)
+        err_type, account = await DB_SESSION_MNG.execute_lambda(
+            tbl_account.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.user_crud.get_account_by_id(s, id),
+        )
+        if err_type != ErrorType.SUCCESS:
+            # 계정 없음/조회 실패 모두 로그인 실패로 일반화
+            res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO)
+            return res
+        account: tbl_account
+
+        # 2) 비밀번호 검증
+        if not await VerifyPW(pw, account.pw):
+            res.result.SetResult(ErrorType.ACCOUNT_INVALID_INFO)
+            return res
+
+        # 3) 차단 여부
+        if account.is_blocked:
+            res.result.SetResult(ErrorType.ACCOUNT_BLOCKED_USER)
+            return res
+
+        # 4) 토큰 발급
+        user_info = UserInfo(uid=account.uid, id=account.id, nickname=account.nickname)
+        res.access_token = CreateAccessToken(user_info)
+        res.refresh_token = CreateRefreshToken(user_info)
+
+        # 5) 마지막 로그인 시간 갱신 (Write DB, 트랜잭션)
+        err_type = await DB_SESSION_MNG.execute_lambda_run(
+            [tbl_account.DBType()],
+            [lambda s: self.user_crud.update_last_login(s, account.uid)],
+        )
+        if err_type != ErrorType.SUCCESS:
+            res.result.SetResult(err_type)
+            return res
+
+        res.uid = account.uid
+        res.nickname = account.nickname
+        return res
+
+    async def create_account(self, id: str, pw: str, nickname: str, connect_ip: str) -> Res_CreateAccount:
+        LOG.i(f"CREATE : {id=}, {nickname=}")
+        res = Res_CreateAccount()
+
+        # 1) 중복 ID 확인 (Read DB)
+        err_type = await DB_SESSION_MNG.execute_lambda(
+            tbl_account.DBType(),
+            DBWRType.DB_READ.value,
+            lambda s: self.user_crud.is_account(s, 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 해시로 저장)
+        account = tbl_account(id=id, pw=await GetHashedPW(pw), nickname=nickname or id)
+        err_type = await DB_SESSION_MNG.execute_lambda_run(
+            [tbl_account.DBType()],
+            [lambda s: self.user_crud.add_account(s, account)],
+        )
+        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.uid = account.uid
+        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

backend/tests/test_auth.py

@@ -0,0 +1,63 @@
+"""auth 도메인 e2e 테스트.
+
+실행 전제: docker-compose 로 PostgreSQL 이 떠 있어야 한다 (negosium_db 사용).
+    docker compose up -d  # 또는 로컬 postgres
+    cd backend && python -m pytest
+"""
+
+
+async def test_create_and_login_flow(client):
+    # 1) 계정 생성
+    r = await client.post("/v1/auth/create", json={"id": "user1", "pw": "pw1234", "nickname": "닉네임"})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["result"]["success"] is True
+    assert body["uid"] > 0
+
+    # 2) 로그인 -> 토큰 발급
+    r = await client.post("/v1/auth/login", json={"id": "user1", "pw": "pw1234"})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["result"]["success"] is True
+    assert body["access_token"]
+    assert body["refresh_token"]
+    assert body["nickname"] == "닉네임"
+    access_token = body["access_token"]
+
+    # 3) 보호된 엔드포인트 호출
+    r = await client.get("/v1/auth/me", headers={"Authorization": f"Bearer {access_token}"})
+    assert r.status_code == 200
+    assert r.json()["id"] == "user1"
+
+
+async def test_login_with_wrong_password(client):
+    await client.post("/v1/auth/create", json={"id": "user2", "pw": "correct", "nickname": "n"})
+
+    r = await client.post("/v1/auth/login", json={"id": "user2", "pw": "wrong"})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["result"]["success"] is False
+    # 자격증명 오류는 ACCOUNT_INVALID_INFO(1200)
+    assert body["result"]["code"] == 1200
+    assert body.get("access_token", "") == ""  # 실패 시 토큰은 빈 문자열
+
+
+async def test_login_nonexistent_account(client):
+    r = await client.post("/v1/auth/login", json={"id": "ghost", "pw": "whatever"})
+    assert r.json()["result"]["success"] is False
+
+
+async def test_duplicate_account_create(client):
+    r1 = await client.post("/v1/auth/create", json={"id": "dup", "pw": "pw1234", "nickname": "n"})
+    assert r1.json()["result"]["success"] is True
+
+    r2 = await client.post("/v1/auth/create", json={"id": "dup", "pw": "pw5678", "nickname": "n2"})
+    body = r2.json()
+    assert body["result"]["success"] is False
+    # ACCOUNT_ALREADY_EXIST(1201)
+    assert body["result"]["code"] == 1201
+
+
+async def test_me_without_token_is_rejected(client):
+    r = await client.get("/v1/auth/me")
+    assert r.status_code in (401, 403)  # HTTPBearer 가 자격증명 없음을 거부

backend/web_main.py

@@ -0,0 +1,41 @@
+# 실행 방법
+#   pip install -r requirements.txt
+#   python web_main.py                # 기본 local 환경
+#   APP_ENV=dev python web_main.py    # 환경 지정
+#
+# 또는 uvicorn 직접 실행:
+#   uvicorn router.router:app --reload --host=0.0.0.0 --port=9300
+
+import uvicorn
+
+from common.logger import LOG
+from config.server_configs import web_server_config
+
+LOG.SetPrefix(web_server_config.server_name)
+
+# import 시점에 app 및 DB 세션 매니저(싱글톤)가 초기화된다.
+import router.router
+
+if __name__ == "__main__":
+    LOG.i(f"Server Name : {web_server_config.server_name}")
+    LOG.i(f"Server Port : {web_server_config.port}")
+    LOG.i(f"API Server start time : {router.router.API_SERVER_START_TIME}")
+
+    if web_server_config.is_ssl:
+        uvicorn.run(
+            "router.router:app",
+            host="0.0.0.0",
+            port=web_server_config.port,
+            access_log=False,
+            workers=web_server_config.process_count,
+            ssl_keyfile="./SSL/key.pem",
+            ssl_certfile="./SSL/cert.pem",
+        )
+    else:
+        uvicorn.run(
+            "router.router:app",
+            host="0.0.0.0",
+            port=web_server_config.port,
+            access_log=False,
+            workers=web_server_config.process_count,
+        )