uuid7으로 필드 및 처리관련 수정

2026-01-28 16:35:08 +09:00 · 2026-01-28 16:35:08 +09:00 · 32ae5530b6
parent aa8d9d7c14
commit 32ae5530b6
26 changed files with 2346 additions and 901 deletions
--- a/access_plan.md
+++ b/access_plan.md
@ -0,0 +1,678 @@
+# Access Token 인증 설계 문서
+
+> 작성일: 2026-01-28
+> 목적: 모든 API 요청에서 액세스 토큰을 검증하여 로그인 사용자를 인증하는 최적의 설계 제안
+
+---
+
+## 1. 현재 시스템 분석
+
+### 1.1 기존 인증 구조
+
+#### 인증 의존성 (`app/user/dependencies/auth.py`)
+
+현재 3가지 인증 의존성이 구현되어 있습니다:
+
+| 의존성 | 용도 | 토큰 없음 시 |
+|--------|------|-------------|
+| `get_current_user()` | 필수 인증 | 예외 발생 (401) |
+| `get_current_user_optional()` | 선택적 인증 | `None` 반환 |
+| `get_current_admin()` | 관리자 전용 | 예외 발생 (403) |
+
+#### JWT 토큰 구조 (`app/user/services/jwt.py`)
+
+```python
+# Access Token Payload
+{
+    "sub": user_uuid,      # 사용자 고유 식별자 (UUID7)
+    "exp": datetime,        # 만료 시간
+    "type": "access"        # 토큰 타입
+}
+```
+
+- **Access Token 유효기간**: `JWT_ACCESS_TOKEN_EXPIRE_MINUTES` (설정 파일)
+- **Refresh Token 유효기간**: `JWT_REFRESH_TOKEN_EXPIRE_DAYS` (설정 파일)
+- **알고리즘**: HS256
+
+#### 커스텀 예외 (`app/user/exceptions.py`)
+
+| 예외 | HTTP 상태 | 코드 |
+|------|----------|------|
+| `MissingTokenError` | 401 | MISSING_TOKEN |
+| `InvalidTokenError` | 401 | INVALID_TOKEN |
+| `TokenExpiredError` | 401 | TOKEN_EXPIRED |
+| `TokenRevokedError` | 401 | TOKEN_REVOKED |
+| `UserNotFoundError` | 404 | USER_NOT_FOUND |
+| `UserInactiveError` | 403 | USER_INACTIVE |
+| `AdminRequiredError` | 403 | ADMIN_REQUIRED |
+
+---
+
+### 1.2 현재 엔드포인트 인증 현황
+
+#### 인증이 적용된 엔드포인트 (3개)
+
+| 엔드포인트 | 메서드 | 의존성 |
+|-----------|--------|--------|
+| `/auth/me` | GET | `get_current_user` |
+| `/auth/logout` | POST | `get_current_user` |
+| `/auth/logout/all` | POST | `get_current_user` |
+
+#### 인증이 없는 엔드포인트 (13개)
+
+| 모듈 | 엔드포인트 | 메서드 | 설명 |
+|------|-----------|--------|------|
+| **Home** | `/crawling` | POST | 네이버 지도 크롤링 |
+| **Home** | `/autocomplete` | POST | 자동완성 크롤링 |
+| **Home** | `/image/upload/server` | POST | 이미지 업로드 (로컬) |
+| **Home** | `/image/upload/blob` | POST | 이미지 업로드 (Azure Blob) |
+| **Lyric** | `/lyric/generate` | POST | 가사 생성 |
+| **Lyric** | `/lyric/status/{task_id}` | GET | 가사 상태 조회 |
+| **Lyric** | `/lyrics/` | GET | 가사 목록 조회 |
+| **Lyric** | `/lyric/{task_id}` | GET | 가사 상세 조회 |
+| **Song** | `/song/generate/{task_id}` | POST | 노래 생성 |
+| **Song** | `/song/status/{song_id}` | GET | 노래 상태 조회 |
+| **Video** | `/video/generate/{task_id}` | GET | 영상 생성 |
+| **Video** | `/video/status/{creatomate_render_id}` | GET | 영상 상태 조회 |
+| **Video** | `/video/download/{task_id}` | GET | 영상 다운로드 |
+| **Video** | `/videos/` | GET | 영상 목록 조회 |
+
+---
+
+### 1.3 모델의 user_uuid 외래키 현황
+
+`user_uuid` 외래키는 **Project 테이블에만** 존재합니다. 하위 리소스(Lyric, Song, Video, Image)의 소유권은 Project를 통해 간접적으로 확인합니다.
+
+| 모델 | user_uuid 필드 | nullable | 비고 |
+|------|-----------------|----------|------|
+| Project | `user_uuid` → User.user_uuid | True | ✅ 소유권 기준 |
+| Image | ❌ 없음 | - | task_id로 Project 연결 |
+| Lyric | ❌ 없음 | - | project_id로 Project 연결 |
+| Song | ❌ 없음 | - | project_id로 Project 연결 |
+| Video | ❌ 없음 | - | project_id로 Project 연결 |
+
+**소유권 확인 흐름:**
+```
+Lyric/Song/Video → project_id → Project → user_uuid → User
+Image → task_id → Project (같은 task_id) → user_uuid → User
+```
+
+---
+
+## 2. 설계 방안 비교
+
+### 2.1 방안 A: 의존성 주입 방식 (Dependency Injection)
+
+각 엔드포인트에 개별적으로 인증 의존성을 추가하는 방식입니다.
+
+```python
+# 예시: 필수 인증
+@router.post("/lyric/generate")
+async def generate_lyric(
+    request_body: GenerateLyricRequest,
+    current_user: User = Depends(get_current_user),  # 인증 추가
+    session: AsyncSession = Depends(get_session),
+):
+    # current_user.user_uuid 사용 가능
+    ...
+
+# 예시: 선택적 인증
+@router.get("/lyrics/")
+async def list_lyrics(
+    current_user: User | None = Depends(get_current_user_optional),  # 선택적 인증
+    session: AsyncSession = Depends(get_session),
+):
+    if current_user:
+        # 로그인 사용자: 자신의 가사만 조회
+        ...
+    else:
+        # 비로그인 사용자: 공개 가사 조회
+        ...
+```
+
+**장점:**
+- FastAPI 표준 패턴 (공식 문서 권장)
+- 엔드포인트별 세밀한 제어 가능
+- 테스트 작성이 용이 (의존성 오버라이드)
+- 기존 코드와의 호환성 우수
+
+**단점:**
+- 각 엔드포인트에 개별 적용 필요
+- 실수로 인증 누락 가능
+
+---
+
+### 2.2 방안 B: 미들웨어 방식 (Middleware)
+
+모든 요청에 미들웨어가 자동으로 토큰을 검증하는 방식입니다.
+
+```python
+# app/middleware/auth.py
+class AuthMiddleware(BaseHTTPMiddleware):
+    # 인증 예외 경로
+    EXEMPT_PATHS = [
+        "/docs", "/openapi.json", "/health",
+        "/auth/kakao/login", "/auth/kakao/callback", "/auth/kakao/verify",
+        "/auth/refresh",
+    ]
+
+    async def dispatch(self, request: Request, call_next):
+        # 예외 경로는 통과
+        if any(request.url.path.startswith(p) for p in self.EXEMPT_PATHS):
+            return await call_next(request)
+
+        # 토큰 검증
+        auth_header = request.headers.get("Authorization")
+        if not auth_header or not auth_header.startswith("Bearer "):
+            return JSONResponse(status_code=401, content={"code": "MISSING_TOKEN"})
+
+        token = auth_header.split(" ")[1]
+        payload = decode_token(token)
+        if not payload:
+            return JSONResponse(status_code=401, content={"code": "INVALID_TOKEN"})
+
+        # request.state에 사용자 정보 저장
+        request.state.user_uuid = payload.get("sub")
+        return await call_next(request)
+```
+
+**장점:**
+- 모든 엔드포인트에 자동 적용
+- 중앙 집중식 관리
+- 인증 누락 방지
+
+**단점:**
+- 예외 경로 관리가 복잡해질 수 있음
+- DB 조회를 미들웨어에서 처리하면 성능 이슈
+- 선택적 인증 구현이 어려움
+- FastAPI의 의존성 주입 패턴과 맞지 않음
+
+---
+
+### 2.3 방안 C: 라우터 레벨 의존성 (Router-Level Dependencies) ⭐ 권장
+
+라우터 전체에 기본 인증을 적용하고, 개별 엔드포인트에서 오버라이드하는 방식입니다.
+
+```python
+# 인증이 필요한 라우터
+lyric_router = APIRouter(
+    prefix="/lyric",
+    tags=["Lyric"],
+    dependencies=[Depends(get_current_user_optional)],  # 라우터 전체에 적용
+)
+
+# 필수 인증이 필요한 엔드포인트
+@lyric_router.post(
+    "/generate",
+    dependencies=[Depends(get_current_user)],  # 오버라이드: 필수 인증
+)
+async def generate_lyric(...):
+    ...
+
+# 선택적 인증 (라우터 기본값 사용)
+@lyric_router.get("/lyrics/")
+async def list_lyrics(...):
+    ...
+```
+
+**장점:**
+- 모듈별 일관된 인증 정책 적용
+- 엔드포인트별 세밀한 제어 가능
+- FastAPI 표준 패턴 준수
+- 인증 누락 가능성 감소
+
+**단점:**
+- 라우터 수정 필요
+- 새 엔드포인트 추가 시 주의 필요
+
+---
+
+## 3. 권장 설계안
+
+### 3.1 최적 설계: 의존성 주입 방식 (방안 A) + 라우터 레벨 기본값 (방안 C)
+
+#### 핵심 원칙
+
+1. **데이터 생성 엔드포인트**: 필수 인증 (`get_current_user`)
+2. **데이터 조회 엔드포인트**: 선택적 인증 (`get_current_user_optional`)
+3. **공개 엔드포인트**: 인증 없음 (로그인, 콜백 등)
+
+#### 엔드포인트별 인증 정책
+
+| 엔드포인트 | 인증 타입 | 이유 |
+|-----------|----------|------|
+| `/auth/kakao/login` | 없음 | 로그인 진입점 |
+| `/auth/kakao/callback` | 없음 | OAuth 콜백 |
+| `/auth/kakao/verify` | 없음 | 토큰 발급 |
+| `/auth/refresh` | 없음 | 토큰 갱신 |
+| `/auth/me` | **필수** | 내 정보 조회 |
+| `/auth/logout` | **필수** | 로그아웃 |
+| `/auth/logout/all` | **필수** | 전체 로그아웃 |
+| `/crawling` | **선택적** | 비로그인도 테스트 가능 |
+| `/autocomplete` | **선택적** | 비로그인도 테스트 가능 |
+| `/image/upload/blob` | **필수** | 리소스 생성 |
+| `/lyric/generate` | **필수** | 리소스 생성 |
+| `/lyric/status/{task_id}` | **선택적** | 상태 조회 |
+| `/lyric/{task_id}` | **선택적** | 상세 조회 |
+| `/lyrics/` | **선택적** | 목록 조회 |
+| `/song/generate/{task_id}` | **필수** | 리소스 생성 |
+| `/song/status/{song_id}` | **선택적** | 상태 조회 |
+| `/video/generate/{task_id}` | **필수** | 리소스 생성 |
+| `/video/status/{...}` | **선택적** | 상태 조회 |
+| `/video/download/{task_id}` | **선택적** | 다운로드 |
+| `/videos/` | **선택적** | 목록 조회 |
+
+---
+
+### 3.2 구현 코드 예시
+
+#### 3.2.1 리소스 생성 엔드포인트 (필수 인증)
+
+```python
+# app/lyric/api/routers/v1/lyric.py
+
+from app.user.dependencies import get_current_user
+from app.user.models import User
+
+@router.post("/generate")
+async def generate_lyric(
+    request_body: GenerateLyricRequest,
+    background_tasks: BackgroundTasks,
+    current_user: User = Depends(get_current_user),  # ✅ 필수 인증
+    session: AsyncSession = Depends(get_session),
+) -> GenerateLyricResponse:
+    """고객 정보를 기반으로 가사를 생성합니다. (백그라운드 처리)"""
+
+    # Project 생성 시 user_uuid 연결 (소유권의 기준점)
+    project = Project(
+        store_name=request_body.customer_name,
+        region=request_body.region,
+        task_id=task_id,
+        user_uuid=current_user.user_uuid,  # ✅ 사용자 연결
+        ...
+    )
+
+    # Lyric은 project_id를 통해 소유권 확인 (user_uuid 없음)
+    lyric = Lyric(
+        project_id=project.id,  # ✅ Project 연결 → 소유권 간접 확인
+        task_id=task_id,
+        ...
+    )
+```
+
+#### 3.2.2 데이터 조회 엔드포인트 (선택적 인증)
+
+```python
+# app/lyric/api/routers/v1/lyric.py
+
+from app.user.dependencies import get_current_user_optional
+from app.user.models import User
+
+@router.get("/lyrics/")
+async def list_lyrics(
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+    current_user: User | None = Depends(get_current_user_optional),  # ✅ 선택적 인증
+    session: AsyncSession = Depends(get_session),
+) -> PaginatedResponse[LyricListItem]:
+    """페이지네이션으로 완료된 가사 목록을 조회합니다."""
+
+    base_query = select(Lyric).where(Lyric.status == "completed")
+
+    # 로그인 사용자: Project.user_uuid를 통해 자신의 가사만 조회
+    if current_user:
+        base_query = (
+            base_query
+            .join(Project, Lyric.project_id == Project.id)
+            .where(Project.user_uuid == current_user.user_uuid)  # ✅ Project를 통한 소유자 필터
+        )
+    else:
+        # 비로그인 사용자: 공개 데이터만 조회 (Project.user_uuid가 NULL)
+        base_query = (
+            base_query
+            .join(Project, Lyric.project_id == Project.id)
+            .where(Project.user_uuid.is_(None))
+        )
+
+    # 페이지네이션 처리
+    ...
+```
+
+#### 3.2.3 이미지 업로드 (필수 인증 + user_uuid 폴더 구조)
+
+```python
+# app/home/api/routers/v1/home.py
+
+from app.user.dependencies import get_current_user
+from app.user.models import User
+from app.utils.upload_blob_as_request import AzureBlobUploader
+
+@router.post("/image/upload/blob")
+async def upload_images_blob(
+    images_json: Optional[str] = Form(default=None),
+    files: Optional[list[UploadFile]] = File(default=None),
+    current_user: User = Depends(get_current_user),  # ✅ 필수 인증
+) -> ImageUploadResponse:
+    """이미지 업로드 (URL + Azure Blob Storage)"""
+
+    task_id = await generate_task_id()
+
+    # ✅ user_uuid를 포함한 Blob 업로더 생성
+    uploader = AzureBlobUploader(
+        user_uuid=current_user.user_uuid,
+        task_id=task_id,
+    )
+    # Blob 경로: {BASE_URL}/{user_uuid}/{task_id}/image/{file_name}
+
+    # Image 저장 (user_uuid 없음 - task_id로 Project와 연결)
+    image = Image(
+        task_id=task_id,  # ✅ task_id를 통해 Project와 연결 → 소유권 확인
+        img_name=img_name,
+        img_url=blob_url,
+        ...
+    )
+```
+
+> **Note**: Image 테이블에는 `user_uuid` 필드가 없습니다.
+> 소유권은 `task_id`를 공유하는 Project를 통해 확인합니다.
+
+---
+
+### 3.3 소유권 검증 유틸리티
+
+데이터 조회/수정 시 **Project를 통한** 소유권 검증을 위한 유틸리티 함수를 추가합니다.
+
+```python
+# app/utils/ownership.py
+
+from fastapi import HTTPException, status
+from typing import Optional
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+from app.user.models import User
+from app.home.models import Project
+
+
+def verify_ownership(
+    project_user_uuid: Optional[str],
+    current_user: Optional[User],
+    raise_on_mismatch: bool = True,
+) -> bool:
+    """
+    Project 기반 리소스 소유권 검증
+
+    Args:
+        project_user_uuid: Project의 user_uuid
+        current_user: 현재 로그인 사용자 (None이면 비로그인)
+        raise_on_mismatch: True면 불일치 시 예외 발생
+
+    Returns:
+        bool: 소유권 일치 여부
+
+    Raises:
+        HTTPException: 소유권 불일치 시 (raise_on_mismatch=True)
+    """
+    # Project에 소유자가 없으면 모두 접근 가능 (공개 리소스)
+    if project_user_uuid is None:
+        return True
+
+    # 비로그인 사용자가 소유자가 있는 리소스에 접근
+    if current_user is None:
+        if raise_on_mismatch:
+            raise HTTPException(
+                status_code=status.HTTP_401_UNAUTHORIZED,
+                detail={"code": "AUTH_REQUIRED", "message": "로그인이 필요합니다."},
+            )
+        return False
+
+    # 소유자 불일치
+    if project_user_uuid != current_user.user_uuid:
+        if raise_on_mismatch:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail={"code": "FORBIDDEN", "message": "접근 권한이 없습니다."},
+            )
+        return False
+
+    return True
+
+
+async def get_project_owner_uuid(
+    session: AsyncSession,
+    project_id: int,
+) -> Optional[str]:
+    """project_id로 소유자의 user_uuid 조회"""
+    result = await session.execute(
+        select(Project.user_uuid).where(Project.id == project_id)
+    )
+    return result.scalar_one_or_none()
+```
+
+**사용 예시:**
+
+```python
+@router.get("/{task_id}")
+async def get_lyric_detail(
+    task_id: str,
+    current_user: User | None = Depends(get_current_user_optional),
+    session: AsyncSession = Depends(get_session),
+) -> LyricDetailResponse:
+    """task_id로 생성된 가사를 조회합니다."""
+
+    lyric = await get_lyric_by_task_id(session, task_id)
+
+    # ✅ Project를 통한 소유권 검증
+    project_user_uuid = await get_project_owner_uuid(session, lyric.project_id)
+    verify_ownership(project_user_uuid, current_user)
+
+    return LyricDetailResponse(...)
+```
+
+---
+
+## 4. 구현 체크리스트
+
+### 4.1 Phase 1: 인증 인프라 확장
+
+- [ ] `app/utils/ownership.py` 생성 (소유권 검증 유틸리티)
+- [ ] `AzureBlobUploader` 클래스에 `user_uuid` 파라미터 활성화
+- [ ] 기존 `get_current_user`, `get_current_user_optional` 테스트
+
+### 4.2 Phase 2: 리소스 생성 엔드포인트 인증 적용
+
+- [ ] `POST /image/upload/blob` - 필수 인증 (Image에는 user_uuid 없음, task_id로 연결)
+- [ ] `POST /lyric/generate` - 필수 인증 + Project.user_uuid 저장
+- [ ] `POST /song/generate/{task_id}` - 필수 인증 (Project 통해 소유권 확인)
+- [ ] `GET /video/generate/{task_id}` - 필수 인증 (Project 통해 소유권 확인)
+
+### 4.3 Phase 3: 조회 엔드포인트 인증 적용
+
+- [ ] `GET /lyric/status/{task_id}` - 선택적 인증 + Project 통해 소유권 검증
+- [ ] `GET /lyric/{task_id}` - 선택적 인증 + Project 통해 소유권 검증
+- [ ] `GET /lyrics/` - 선택적 인증 + Project.user_uuid 기반 필터
+- [ ] `GET /song/status/{song_id}` - 선택적 인증 + Project 통해 소유권 검증
+- [ ] `GET /video/status/{...}` - 선택적 인증 + Project 통해 소유권 검증
+- [ ] `GET /video/download/{task_id}` - 선택적 인증 + Project 통해 소유권 검증
+- [ ] `GET /videos/` - 선택적 인증 + Project.user_uuid 기반 필터
+
+### 4.4 Phase 4: 크롤링 엔드포인트
+
+- [ ] `POST /crawling` - 선택적 인증 (비로그인도 허용)
+- [ ] `POST /autocomplete` - 선택적 인증 (비로그인도 허용)
+
+### 4.5 Phase 5: OpenAPI 문서 업데이트
+
+- [ ] `main.py`의 `custom_openapi()` 수정하여 인증이 필요한 엔드포인트에 security 표시
+
+---
+
+## 5. OpenAPI Security 스키마 업데이트
+
+`main.py`의 `custom_openapi()` 함수를 수정하여 인증이 필요한 엔드포인트를 표시합니다.
+
+```python
+def custom_openapi():
+    """커스텀 OpenAPI 스키마 생성 (Bearer 인증 추가)"""
+    if app.openapi_schema:
+        return app.openapi_schema
+
+    openapi_schema = get_openapi(
+        title=app.title,
+        version=app.version,
+        description=app.description,
+        routes=app.routes,
+        tags=tags_metadata,
+    )
+
+    # Bearer 토큰 인증 스키마 추가
+    openapi_schema["components"]["securitySchemes"] = {
+        "BearerAuth": {
+            "type": "http",
+            "scheme": "bearer",
+            "bearerFormat": "JWT",
+            "description": "JWT 액세스 토큰을 입력하세요.",
+        }
+    }
+
+    # 인증이 필요한 경로 패턴
+    AUTH_REQUIRED_PATHS = [
+        "/auth/me", "/auth/logout",
+        "/image/upload/blob",
+        "/lyric/generate",
+        "/song/generate",
+        "/video/generate",
+    ]
+
+    AUTH_OPTIONAL_PATHS = [
+        "/crawling", "/autocomplete",
+        "/lyric/status", "/lyric/",
+        "/lyrics/",
+        "/song/status",
+        "/video/status", "/video/download",
+        "/videos/",
+    ]
+
+    for path, path_item in openapi_schema["paths"].items():
+        for method, operation in path_item.items():
+            if method in ["get", "post", "put", "patch", "delete"]:
+                # 필수 인증
+                if any(auth_path in path for auth_path in AUTH_REQUIRED_PATHS):
+                    operation["security"] = [{"BearerAuth": []}]
+                # 선택적 인증 (문서에는 표시하지만 필수 아님)
+                elif any(auth_path in path for auth_path in AUTH_OPTIONAL_PATHS):
+                    operation["security"] = [{"BearerAuth": []}, {}]  # 빈 객체 = 인증 없이도 가능
+
+    app.openapi_schema = openapi_schema
+    return app.openapi_schema
+```
+
+---
+
+## 6. 테스트 전략
+
+### 6.1 단위 테스트
+
+```python
+# tests/test_auth_dependencies.py
+
+import pytest
+from fastapi import HTTPException
+from app.user.dependencies import get_current_user, get_current_user_optional
+
+
+@pytest.mark.asyncio
+async def test_get_current_user_missing_token():
+    """토큰 없이 접근 시 MissingTokenError 발생"""
+    with pytest.raises(HTTPException) as exc_info:
+        await get_current_user(credentials=None, session=mock_session)
+    assert exc_info.value.status_code == 401
+    assert exc_info.value.detail["code"] == "MISSING_TOKEN"
+
+
+@pytest.mark.asyncio
+async def test_get_current_user_optional_missing_token():
+    """선택적 인증에서 토큰 없으면 None 반환"""
+    result = await get_current_user_optional(credentials=None, session=mock_session)
+    assert result is None
+```
+
+### 6.2 통합 테스트
+
+```python
+# tests/test_lyric_auth.py
+
+import pytest
+from httpx import AsyncClient
+
+
+@pytest.mark.asyncio
+async def test_generate_lyric_without_auth(client: AsyncClient):
+    """인증 없이 가사 생성 시 401 반환"""
+    response = await client.post("/lyric/generate", json={...})
+    assert response.status_code == 401
+    assert response.json()["detail"]["code"] == "MISSING_TOKEN"
+
+
+@pytest.mark.asyncio
+async def test_generate_lyric_with_auth(client: AsyncClient, auth_headers: dict):
+    """인증된 사용자의 가사 생성 성공"""
+    response = await client.post(
+        "/lyric/generate",
+        json={...},
+        headers=auth_headers,
+    )
+    assert response.status_code == 200
+```
+
+---
+
+## 7. 마이그레이션 고려사항
+
+### 7.1 기존 데이터 처리
+
+현재 Project 테이블에서 `user_uuid`가 `NULL`인 레코드들에 대한 처리 방안:
+
+1. **기존 데이터 유지**: `Project.user_uuid`가 `NULL`이면 해당 Project 및 하위 리소스(Lyric, Song, Video, Image)를 공개 데이터로 취급
+2. **관리자 전용**: `Project.user_uuid`가 `NULL`인 데이터는 관리자만 접근 가능
+3. **마이그레이션**: 특정 사용자에게 기존 Project 할당 (비권장)
+
+**권장**: 옵션 1 (기존 데이터는 공개 데이터로 유지)
+
+> **Note**: `user_uuid`는 Project 테이블에만 존재합니다.
+> Lyric, Song, Video, Image 테이블에는 `user_uuid` 필드가 없으며,
+> 소유권은 Project를 통해 간접적으로 확인합니다.
+
+### 7.2 하위 호환성
+
+- 기존 API 클라이언트가 인증 없이 호출하는 경우 401 응답
+- 프론트엔드 업데이트 필요: 모든 API 호출에 `Authorization` 헤더 추가
+- 점진적 적용: Phase별로 적용하여 영향 범위 최소화
+
+---
+
+## 8. 결론
+
+### 권장 구현 순서
+
+1. **의존성 주입 방식** 채택 (FastAPI 표준 패턴)
+2. **필수/선택적 인증** 엔드포인트별 적용
+3. **Project 기반 소유권 검증** 유틸리티 활용
+4. **Project.user_uuid 연결** 리소스 생성 시 적용
+
+### 데이터 모델 구조
+
+```
+User (user_uuid)
+  └── Project (user_uuid → User)  ← 소유권의 기준점
+        ├── Lyric (project_id → Project)
+        │     └── Song (lyric_id → Lyric)
+        │           └── Video (song_id → Song)
+        └── Image (task_id = Project.task_id)
+```
+
+이 설계를 통해:
+- **Project 테이블만** `user_uuid`를 가지며, 모든 소유권 확인의 기준점 역할
+- 하위 리소스(Lyric, Song, Video, Image)는 Project를 통해 간접적으로 소유권 확인
+- 사용자별 데이터 격리가 가능합니다
+- 비로그인 사용자도 제한된 기능을 사용할 수 있습니다
+- Azure Blob Storage 폴더 구조에 user_uuid가 포함됩니다
--- a/app/database/session.py
+++ b/app/database/session.py
@ -72,18 +72,31 @@ async def create_db_tables():
    import asyncio

    # 모델 import (테이블 메타데이터 등록용)
-    # 주의: User를 먼저 import해야 UserProject가 User를 참조할 수 있음
-    from app.user.models import User  # noqa: F401
-    from app.home.models import Image, Project, UserProject  # noqa: F401
+    from app.user.models import User, RefreshToken  # noqa: F401
+    from app.home.models import Image, Project  # noqa: F401
    from app.lyric.models import Lyric  # noqa: F401
-    from app.song.models import Song  # noqa: F401
+    from app.song.models import Song, SongTimestamp  # noqa: F401
    from app.video.models import Video  # noqa: F401

+    # 생성할 테이블 목록 (SocialAccount 제외)
+    tables_to_create = [
+        User.__table__,
+        RefreshToken.__table__,
+        Project.__table__,
+        Image.__table__,
+        Lyric.__table__,
+        Song.__table__,
+        SongTimestamp.__table__,
+        Video.__table__,
+    ]
+
    logger.info("Creating database tables...")

    async with asyncio.timeout(10):
        async with engine.begin() as connection:
-            await connection.run_sync(Base.metadata.create_all)
+            await connection.run_sync(
+                lambda conn: Base.metadata.create_all(conn, tables=tables_to_create)
+            )


 # FastAPI 의존성용 세션 제너레이터
--- a/app/home/api/home_admin.py
+++ b/app/home/api/home_admin.py
@ -1,6 +1,6 @@
 from sqladmin import ModelView

-from app.home.models import Image, Project, UserProject
+from app.home.models import Image, Project


 class ProjectAdmin(ModelView, model=Project):
@ -100,44 +100,3 @@ class ImageAdmin(ModelView, model=Image):
        "img_url": "이미지 URL",
        "created_at": "생성일시",
    }
-
-
-class UserProjectAdmin(ModelView, model=UserProject):
-    name = "사용자-프로젝트"
-    name_plural = "사용자-프로젝트 목록"
-    icon = "fa-solid fa-link"
-    category = "프로젝트 관리"
-    page_size = 20
-
-    column_list = [
-        "id",
-        "user_id",
-        "project_id",
-    ]
-
-    column_details_list = [
-        "id",
-        "user_id",
-        "project_id",
-        "user",
-        "project",
-    ]
-
-    column_searchable_list = [
-        UserProject.user_id,
-        UserProject.project_id,
-    ]
-
-    column_sortable_list = [
-        UserProject.id,
-        UserProject.user_id,
-        UserProject.project_id,
-    ]
-
-    column_labels = {
-        "id": "ID",
-        "user_id": "사용자 ID",
-        "project_id": "프로젝트 ID",
-        "user": "사용자",
-        "project": "프로젝트",
-    }
--- a/app/home/api/routers/v1/home.py
+++ b/app/home/api/routers/v1/home.py
@ -12,6 +12,8 @@ from sqlalchemy.ext.asyncio import AsyncSession

 from app.database.session import get_session, AsyncSessionLocal
 from app.home.models import Image
+from app.user.dependencies.auth import get_current_user
+from app.user.models import User
 from app.home.schemas.home_schema import (
    AutoCompleteRequest,
    CrawlingRequest,
@ -502,6 +504,7 @@ async def upload_images(
    files: Optional[list[UploadFile]] = File(
        default=None, description="이미지 바이너리 파일 목록"
    ),
+    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
 ) -> ImageUploadResponse:
    """이미지 업로드 (URL + 바이너리 파일)"""
@ -730,6 +733,7 @@ async def upload_images_blob(
        default=None,
        description="이미지 바이너리 파일 목록",
    ),
+    current_user: User = Depends(get_current_user),
 ) -> ImageUploadResponse:
    """이미지 업로드 (URL + Azure Blob Storage)

--- a/app/home/models.py
+++ b/app/home/models.py
@ -4,13 +4,12 @@ Home 모듈 SQLAlchemy 모델 정의
 이 모듈은 영상 제작 파이프라인의 핵심 데이터 모델을 정의합니다.
 - Project: 프로젝트(사용자 입력 이력) 관리
 - Image: 업로드된 이미지 URL 관리
- UserProject: User와 Project 간 M:N 관계 중계 테이블
 """

 from datetime import datetime
 from typing import TYPE_CHECKING, List, Optional

-from sqlalchemy import BigInteger, DateTime, ForeignKey, Index, Integer, String, Text, func
+from sqlalchemy import DateTime, ForeignKey, Index, Integer, String, Text, func
 from sqlalchemy.orm import Mapped, mapped_column, relationship

 from app.database.session import Base
@ -21,122 +20,6 @@ if TYPE_CHECKING:
    from app.user.models import User
    from app.video.models import Video

-# =============================================================================
-# User-Project M:N 관계 중계 테이블
-# =============================================================================
-#
-# 설계 의도:
-#   - User와 Project는 다대다(M:N) 관계입니다.
-#   - 한 사용자는 여러 프로젝트에 참여할 수 있습니다.
-#   - 한 프로젝트에는 여러 사용자가 참여할 수 있습니다.
-#
-# 중계 테이블 역할:
-#   - UserProject 테이블이 두 테이블 간의 관계를 연결합니다.
-#   - 각 레코드는 특정 사용자와 특정 프로젝트의 연결을 나타냅니다.
-#   - 추가 속성(role, joined_at)으로 관계의 메타데이터를 저장합니다.
-#
-# 외래키 설정:
-#   - user_id: User 테이블의 id를 참조 (ON DELETE CASCADE)
-#   - project_id: Project 테이블의 id를 참조 (ON DELETE CASCADE)
-#   - CASCADE 설정으로 부모 레코드 삭제 시 중계 레코드도 자동 삭제됩니다.
-#
-# 관계 방향:
-#   - User.projects → UserProject → Project (사용자가 참여한 프로젝트 목록)
-#   - Project.users → UserProject → User (프로젝트에 참여한 사용자 목록)
-# =============================================================================
-
-
-class UserProject(Base):
-    """
-    User-Project M:N 관계 중계 테이블
-
-    사용자와 프로젝트 간의 다대다 관계를 관리합니다.
-    한 사용자는 여러 프로젝트에 참여할 수 있고,
-    한 프로젝트에는 여러 사용자가 참여할 수 있습니다.
-
-    Attributes:
-        id: 고유 식별자 (자동 증가)
-        user_id: 사용자 외래키 (User.id 참조)
-        project_id: 프로젝트 외래키 (Project.id 참조)
-        role: 프로젝트 내 사용자 역할 (owner: 소유자, member: 멤버, viewer: 조회자)
-        joined_at: 프로젝트 참여 일시
-
-    외래키 제약조건:
-        - user_id → user.id (ON DELETE CASCADE)
-        - project_id → project.id (ON DELETE CASCADE)
-
-    유니크 제약조건:
-        - (user_id, project_id) 조합은 유일해야 함 (중복 참여 방지)
-    """
-
-    __tablename__ = "user_project"
-    __table_args__ = (
-        Index("idx_user_project_user_id", "user_id"),
-        Index("idx_user_project_project_id", "project_id"),
-        Index("idx_user_project_user_project", "user_id", "project_id", unique=True),
-        {
-            "mysql_engine": "InnoDB",
-            "mysql_charset": "utf8mb4",
-            "mysql_collate": "utf8mb4_unicode_ci",
-        },
-    )
-
-    id: Mapped[int] = mapped_column(
-        Integer,
-        primary_key=True,
-        nullable=False,
-        autoincrement=True,
-        comment="고유 식별자",
-    )
-
-    # 외래키: User 테이블 참조
-    # - BigInteger 사용 (User.id가 BigInteger이므로 타입 일치 필요)
-    # - ondelete="CASCADE": User 삭제 시 연결된 UserProject 레코드도 삭제
-    user_id: Mapped[int] = mapped_column(
-        BigInteger,
-        ForeignKey("user.id", ondelete="CASCADE"),
-        nullable=False,
-        comment="사용자 외래키 (User.id 참조)",
-    )
-
-    # 외래키: Project 테이블 참조
-    # - Integer 사용 (Project.id가 Integer이므로 타입 일치 필요)
-    # - ondelete="CASCADE": Project 삭제 시 연결된 UserProject 레코드도 삭제
-    project_id: Mapped[int] = mapped_column(
-        Integer,
-        ForeignKey("project.id", ondelete="CASCADE"),
-        nullable=False,
-        comment="프로젝트 외래키 (Project.id 참조)",
-    )
-
-    # ==========================================================================
-    # Relationships (관계 설정)
-    # ==========================================================================
-    # back_populates: 양방향 관계 설정 (User.user_projects, Project.user_projects)
-    # lazy="selectin": N+1 문제 방지를 위한 즉시 로딩
-    # ==========================================================================
-    user: Mapped["User"] = relationship(
-        "User",
-        back_populates="user_projects",
-        lazy="selectin",
-    )
-
-    project: Mapped["Project"] = relationship(
-        "Project",
-        back_populates="user_projects",
-        lazy="selectin",
-    )
-
-    def __repr__(self) -> str:
-        return (
-            f"<UserProject("
-            f"id={self.id}, "
-            f"user_id={self.user_id}, "
-            f"project_id={self.project_id}, "
-            f"role='{self.role}'"
-            f")>"
-        )
-

 class Project(Base):
    """
@ -149,16 +32,15 @@ class Project(Base):
        id: 고유 식별자 (자동 증가)
        store_name: 고객명 (필수)
        region: 지역명 (필수, 예: 서울, 부산, 대구 등)
-        task_id: 작업 고유 식별자 (KSUID 형식, 27자)
+        task_id: 작업 고유 식별자 (UUID7 형식, 36자)
        detail_region_info: 상세 지역 정보 (선택, JSON 또는 텍스트 형식)
        created_at: 생성 일시 (자동 설정)

    Relationships:
+        owner: 프로젝트 소유자 (User, 1:N 관계)
        lyrics: 생성된 가사 목록
        songs: 생성된 노래 목록
        videos: 최종 영상 결과 목록
-        user_projects: User와의 M:N 관계 (중계 테이블 통한 연결)
-        users: 프로젝트에 참여한 사용자 목록 (Association Proxy)
    """

    __tablename__ = "project"
@ -166,6 +48,7 @@ class Project(Base):
        Index("idx_project_task_id", "task_id"),
        Index("idx_project_store_name", "store_name"),
        Index("idx_project_region", "region"),
+        Index("idx_project_user_uuid", "user_uuid"),
        {
            "mysql_engine": "InnoDB",
            "mysql_charset": "utf8mb4",
@ -194,10 +77,27 @@ class Project(Base):
    )

    task_id: Mapped[str] = mapped_column(
-        String(27),
+        String(36),
        nullable=False,
        unique=True,
-        comment="프로젝트 작업 고유 식별자 (KSUID)",
+        comment="프로젝트 작업 고유 식별자 (UUID7)",
+    )
+
+    # ==========================================================================
+    # User 1:N 관계 (한 사용자가 여러 프로젝트를 소유)
+    # ==========================================================================
+    user_uuid: Mapped[Optional[str]] = mapped_column(
+        String(36),
+        ForeignKey("user.user_uuid", ondelete="SET NULL"),
+        nullable=True,
+        comment="프로젝트 소유자 (User.user_uuid 외래키)",
+    )
+
+    # 소유자 관계 설정 (User.projects와 양방향 연결)
+    owner: Mapped[Optional["User"]] = relationship(
+        "User",
+        back_populates="projects",
+        lazy="selectin",
    )

    detail_region_info: Mapped[Optional[str]] = mapped_column(
@ -242,20 +142,6 @@ class Project(Base):
        lazy="selectin",
    )

-    # ==========================================================================
-    # User M:N 관계 (중계 테이블 UserProject 통한 연결)
-    # ==========================================================================
-    # back_populates: UserProject.project와 양방향 연결
-    # cascade: Project 삭제 시 UserProject 레코드도 삭제 (User는 유지)
-    # lazy="selectin": N+1 문제 방지
-    # ==========================================================================
-    user_projects: Mapped[List["UserProject"]] = relationship(
-        "UserProject",
-        back_populates="project",
-        cascade="all, delete-orphan",
-        lazy="selectin",
-    )
-
    def __repr__(self) -> str:
        def truncate(value: str | None, max_len: int = 10) -> str:
            if value is None:
@ -280,7 +166,7 @@ class Image(Base):

    Attributes:
        id: 고유 식별자 (자동 증가)
-        task_id: 이미지 업로드 작업 고유 식별자 (KSUID)
+        task_id: 이미지 업로드 작업 고유 식별자 (UUID7)
        img_name: 이미지명
        img_url: 이미지 URL (S3, CDN 등의 경로)
        created_at: 생성 일시 (자동 설정)
@ -305,10 +191,10 @@ class Image(Base):
    )

    task_id: Mapped[str] = mapped_column(
-        String(27),
+        String(36),
        nullable=False,
        unique=True,
-        comment="이미지 업로드 작업 고유 식별자 (KSUID)",
+        comment="이미지 업로드 작업 고유 식별자 (UUID7)",
    )

    img_name: Mapped[str] = mapped_column(
--- a/app/home/schemas/home_schema.py
+++ b/app/home/schemas/home_schema.py
@ -91,7 +91,7 @@ class GenerateUrlsRequest(GenerateRequestInfo):
 class GenerateUploadResponse(BaseModel):
    """파일 업로드 기반 생성 응답 스키마"""

-    task_id: str = Field(..., description="작업 고유 식별자 (KSUID)")
+    task_id: str = Field(..., description="작업 고유 식별자 (UUID7)")
    status: Literal["processing", "completed", "failed"] = Field(
        ..., description="작업 상태"
    )
@ -102,7 +102,7 @@ class GenerateUploadResponse(BaseModel):
 class GenerateResponse(BaseModel):
    """생성 응답 스키마"""

-    task_id: str = Field(..., description="작업 고유 식별자 (KSUID)")
+    task_id: str = Field(..., description="작업 고유 식별자 (UUID7)")
    status: Literal["processing", "completed", "failed"] = Field(
        ..., description="작업 상태"
    )
@ -291,7 +291,7 @@ class ImageUploadResponse(BaseModel):
        }
    )

-    task_id: str = Field(..., description="작업 고유 식별자 (새로 생성된 KSUID)")
+    task_id: str = Field(..., description="작업 고유 식별자 (새로 생성된 UUID7)")
    total_count: int = Field(..., description="총 업로드된 이미지 개수")
    url_count: int = Field(..., description="URL로 등록된 이미지 개수")
    file_count: int = Field(..., description="파일로 업로드된 이미지 개수")
--- a/app/lyric/api/routers/v1/lyric.py
+++ b/app/lyric/api/routers/v1/lyric.py
@ -31,6 +31,8 @@ from sqlalchemy.ext.asyncio import AsyncSession

 from app.database.session import get_session
 from app.home.models import Project
+from app.user.dependencies.auth import get_current_user
+from app.user.models import User
 from app.lyric.models import Lyric
 from app.lyric.schemas.lyric import (
    GenerateLyricRequest,
@ -222,6 +224,7 @@ POST /lyric/generate
 async def generate_lyric(
    request_body: GenerateLyricRequest,
    background_tasks: BackgroundTasks,
+    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
 ) -> GenerateLyricResponse:
    """고객 정보를 기반으로 가사를 생성합니다. (백그라운드 처리)"""
@ -391,6 +394,7 @@ GET /lyric/status/019123ab-cdef-7890-abcd-ef1234567890
 )
 async def get_lyric_status(
    task_id: str,
+    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
 ) -> LyricStatusResponse:
    """task_id로 가사 생성 작업 상태를 조회합니다."""
@ -435,6 +439,7 @@ GET /lyrics/?page=1&page_size=50       # 50개씩 조회
 async def list_lyrics(
    page: int = Query(1, ge=1, description="페이지 번호 (1부터 시작)"),
    page_size: int = Query(20, ge=1, le=100, description="페이지당 데이터 수"),
+    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
 ) -> PaginatedResponse[LyricListItem]:
    """페이지네이션으로 완료된 가사 목록을 조회합니다."""
@ -478,6 +483,7 @@ GET /lyric/019123ab-cdef-7890-abcd-ef1234567890
 )
 async def get_lyric_detail(
    task_id: str,
+    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
 ) -> LyricDetailResponse:
    """task_id로 생성된 가사를 조회합니다."""
--- a/app/lyric/models.py
+++ b/app/lyric/models.py
@ -23,7 +23,7 @@ class Lyric(Base):
    Attributes:
        id: 고유 식별자 (자동 증가)
        project_id: 연결된 Project의 id (외래키)
-        task_id: 가사 생성 작업의 고유 식별자 (KSUID 형식)
+        task_id: 가사 생성 작업의 고유 식별자 (UUID7 형식)
        status: 처리 상태 (pending, processing, completed, failed 등)
        lyric_prompt: 가사 생성에 사용된 프롬프트
        lyric_result: 생성된 가사 결과 (LONGTEXT로 긴 가사 지원)
@ -62,10 +62,10 @@ class Lyric(Base):
    )

    task_id: Mapped[str] = mapped_column(
-        String(27),
+        String(36),
        nullable=False,
        unique=True,
-        comment="가사 생성 작업 고유 식별자 (KSUID)",
+        comment="가사 생성 작업 고유 식별자 (UUID7)",
    )

    status: Mapped[str] = mapped_column(
--- a/app/song/api/routers/v1/song.py
+++ b/app/song/api/routers/v1/song.py
@ -18,6 +18,8 @@ from sqlalchemy.ext.asyncio import AsyncSession

 from app.database.session import get_session
 from app.home.models import Project
+from app.user.dependencies.auth import get_current_user
+from app.user.models import User
 from app.lyric.models import Lyric
 from app.song.models import Song, SongTimestamp
 from app.song.schemas.song_schema import (
@ -79,6 +81,7 @@ POST /song/generate/019123ab-cdef-7890-abcd-ef1234567890
 async def generate_song(
    task_id: str,
    request_body: GenerateSongRequest,
+    current_user: User = Depends(get_current_user),
 ) -> GenerateSongResponse:
    """가사와 장르를 기반으로 Suno API를 통해 노래를 생성합니다.

@ -347,6 +350,7 @@ GET /song/status/abc123...
 async def get_song_status(
    song_id: str,
    background_tasks: BackgroundTasks,
+    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
 ) -> PollingSongResponse:
    """song_id로 노래 생성 작업의 상태를 조회합니다.
--- a/app/song/models.py
+++ b/app/song/models.py
@ -23,7 +23,7 @@ class Song(Base):
        id: 고유 식별자 (자동 증가)
        project_id: 연결된 Project의 id (외래키)
        lyric_id: 연결된 Lyric의 id (외래키)
-        task_id: 노래 생성 작업의 고유 식별자 (KSUID 형식)
+        task_id: 노래 생성 작업의 고유 식별자 (UUID7 형식)
        suno_task_id: Suno API 작업 고유 식별자 (선택)
        status: 처리 상태 (processing, uploading, completed, failed)
        song_prompt: 노래 생성에 사용된 프롬프트
@ -72,10 +72,10 @@ class Song(Base):
    )

    task_id: Mapped[str] = mapped_column(
-        String(27),
+        String(36),
        nullable=False,
        unique=True,
-        comment="노래 생성 작업 고유 식별자 (KSUID)",
+        comment="노래 생성 작업 고유 식별자 (UUID7)",
    )

    suno_task_id: Mapped[Optional[str]] = mapped_column(
--- a/app/user/api/routers/v1/auth.py
+++ b/app/user/api/routers/v1/auth.py
@ -5,10 +5,14 @@
 """

 import logging
+import random
+from datetime import datetime, timezone
 from typing import Optional

-from fastapi import APIRouter, Depends, Header, Request, status
+from fastapi import APIRouter, Depends, Header, HTTPException, Request, status
 from fastapi.responses import RedirectResponse, Response
+from pydantic import BaseModel
+from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession

 from config import prj_settings
@ -16,7 +20,7 @@ from app.database.session import get_session

 logger = logging.getLogger(__name__)
 from app.user.dependencies import get_current_user
-from app.user.models import User
+from app.user.models import RefreshToken, User
 from app.user.schemas.user_schema import (
    AccessTokenResponse,
    KakaoCodeRequest,
@ -26,9 +30,56 @@ from app.user.schemas.user_schema import (
    UserResponse,
 )
 from app.user.services import auth_service, kakao_client
+from app.user.services.jwt import (
+    create_access_token,
+    create_refresh_token,
+    get_access_token_expire_seconds,
+    get_refresh_token_expires_at,
+    get_token_hash,
+)
+from app.utils.common import generate_uuid
+

 router = APIRouter(prefix="/auth", tags=["Auth"])

+# =============================================================================
+# 테스트용 라우터 (DEBUG 모드에서만 main.py에서 등록됨)
+# =============================================================================
+test_router = APIRouter(prefix="/auth/test", tags=["Test Auth"])
+
+
+# =============================================================================
+# 테스트용 스키마
+# =============================================================================
+class TestUserCreateRequest(BaseModel):
+    """테스트 사용자 생성 요청"""
+
+    nickname: str = "테스트유저"
+
+
+class TestUserCreateResponse(BaseModel):
+    """테스트 사용자 생성 응답"""
+
+    user_id: int
+    user_uuid: str
+    nickname: str
+    message: str
+
+
+class TestTokenRequest(BaseModel):
+    """테스트 토큰 발급 요청"""
+
+    user_uuid: str
+
+
+class TestTokenResponse(BaseModel):
+    """테스트 토큰 발급 응답"""
+
+    access_token: str
+    refresh_token: str
+    token_type: str = "Bearer"
+    expires_in: int
+

@router.get(
    "/kakao/login",
@ -245,3 +296,143 @@ async def get_me(
    현재 로그인한 사용자의 상세 정보를 반환합니다.
    """
    return UserResponse.model_validate(current_user)
+
+
+# =============================================================================
+# 테스트용 엔드포인트 (DEBUG 모드에서만 main.py에서 라우터가 등록됨)
+# =============================================================================
+@test_router.post(
+    "/create-user",
+    response_model=TestUserCreateResponse,
+    summary="[테스트] 사용자 직접 생성",
+    description="""
+**DEBUG 모드에서만 사용 가능합니다.**
+
+카카오 로그인 없이 테스트용 사용자를 직접 생성합니다.
+생성된 user_uuid로 `/generate-token` 엔드포인트에서 토큰을 발급받을 수 있습니다.
+""",
+)
+async def create_test_user(
+    body: TestUserCreateRequest,
+    session: AsyncSession = Depends(get_session),
+) -> TestUserCreateResponse:
+    """
+    테스트용 사용자 직접 생성
+
+    카카오 로그인 없이 테스트용 사용자를 생성합니다.
+    DEBUG 모드에서만 사용 가능합니다.
+    """
+    logger.info(f"[TEST] 테스트 사용자 생성 요청 - nickname: {body.nickname}")
+
+    # 고유한 uuid 생성
+    user_uuid = await generate_uuid(session=session, table_name=User)
+
+    # 테스트용 가짜 kakao_id 생성 (충돌 방지를 위해 큰 범위 사용)
+    fake_kakao_id = random.randint(9000000000, 9999999999)
+
+    # 사용자 생성
+    new_user = User(
+        kakao_id=fake_kakao_id,
+        user_uuid=user_uuid,
+        nickname=body.nickname,
+        is_active=True,
+    )
+    session.add(new_user)
+    await session.commit()
+    await session.refresh(new_user)
+
+    logger.info(
+        f"[TEST] 테스트 사용자 생성 완료 - user_id: {new_user.id}, "
+        f"user_uuid: {new_user.user_uuid}"
+    )
+
+    return TestUserCreateResponse(
+        user_id=new_user.id,
+        user_uuid=new_user.user_uuid,
+        nickname=new_user.nickname or body.nickname,
+        message="테스트 사용자가 생성되었습니다.",
+    )
+
+
+@test_router.post(
+    "/generate-token",
+    response_model=TestTokenResponse,
+    summary="[테스트] 토큰 직접 발급",
+    description="""
+**DEBUG 모드에서만 사용 가능합니다.**
+
+user_uuid로 JWT 토큰을 직접 발급합니다.
+`/create-user`에서 생성한 사용자의 user_uuid를 사용하세요.
+""",
+)
+async def generate_test_token(
+    request: Request,
+    body: TestTokenRequest,
+    session: AsyncSession = Depends(get_session),
+    user_agent: Optional[str] = Header(None, alias="User-Agent"),
+) -> TestTokenResponse:
+    """
+    테스트용 토큰 직접 발급
+
+    카카오 로그인 없이 user_uuid로 JWT 토큰을 발급합니다.
+    DEBUG 모드에서만 사용 가능합니다.
+    """
+    logger.info(f"[TEST] 테스트 토큰 발급 요청 - user_uuid: {body.user_uuid}")
+
+    # 사용자 조회
+    result = await session.execute(
+        select(User).where(User.user_uuid == body.user_uuid)
+    )
+    user = result.scalar_one_or_none()
+
+    if user is None:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"사용자를 찾을 수 없습니다: {body.user_uuid}",
+        )
+
+    if not user.is_active:
+        raise HTTPException(
+            status_code=status.HTTP_403_FORBIDDEN,
+            detail="비활성화된 사용자입니다.",
+        )
+
+    # JWT 토큰 생성
+    access_token = create_access_token(user.user_uuid)
+    refresh_token = create_refresh_token(user.user_uuid)
+
+    # 클라이언트 IP 추출
+    ip_address = request.client.host if request.client else None
+    forwarded_for = request.headers.get("X-Forwarded-For")
+    if forwarded_for:
+        ip_address = forwarded_for.split(",")[0].strip()
+
+    # 리프레시 토큰 DB 저장
+    token_hash = get_token_hash(refresh_token)
+    expires_at = get_refresh_token_expires_at()
+
+    db_refresh_token = RefreshToken(
+        user_id=user.id,
+        user_uuid=user.user_uuid,
+        token_hash=token_hash,
+        expires_at=expires_at,
+        user_agent=user_agent,
+        ip_address=ip_address,
+    )
+    session.add(db_refresh_token)
+
+    # 마지막 로그인 시간 업데이트
+    user.last_login_at = datetime.now(timezone.utc)
+    await session.commit()
+
+    logger.info(
+        f"[TEST] 테스트 토큰 발급 완료 - user_id: {user.id}, "
+        f"user_uuid: {user.user_uuid}"
+    )
+
+    return TestTokenResponse(
+        access_token=access_token,
+        refresh_token=refresh_token,
+        token_type="Bearer",
+        expires_in=get_access_token_expire_seconds(),
+    )
--- a/app/user/api/user_admin.py
+++ b/app/user/api/user_admin.py
@ -45,7 +45,7 @@ class UserAdmin(ModelView, model=User):
    form_excluded_columns = [
        "created_at",
        "updated_at",
-        "user_projects",
+        "projects",
        "refresh_tokens",
        "social_accounts",
    ]
--- a/app/user/dependencies/auth.py
+++ b/app/user/dependencies/auth.py
@ -58,14 +58,14 @@ async def get_current_user(
    if payload.get("type") != "access":
        raise InvalidTokenError("액세스 토큰이 아닙니다.")

-    user_ksuid = payload.get("sub")
-    if user_ksuid is None:
+    user_uuid = payload.get("sub")
+    if user_uuid is None:
        raise InvalidTokenError()

    # 사용자 조회
    result = await session.execute(
        select(User).where(
-            User.user_ksuid == user_ksuid,
+            User.user_uuid == user_uuid,
            User.is_deleted == False,  # noqa: E712
        )
    )
@ -106,13 +106,13 @@ async def get_current_user_optional(
    if payload.get("type") != "access":
        return None

-    user_ksuid = payload.get("sub")
-    if user_ksuid is None:
+    user_uuid = payload.get("sub")
+    if user_uuid is None:
        return None

    result = await session.execute(
        select(User).where(
-            User.user_ksuid == user_ksuid,
+            User.user_uuid == user_uuid,
            User.is_deleted == False,  # noqa: E712
        )
    )
--- a/app/user/exceptions.py
+++ b/app/user/exceptions.py
@ -111,7 +111,7 @@ class MissingTokenError(AuthException):
 class UserNotFoundError(AuthException):
    """사용자 없음"""

-    def __init__(self, message: str = "사용자를 찾을 수 없습니다."):
+    def __init__(self, message: str = "가입되지 않은 사용자 입니다."):
        super().__init__(
            status_code=status.HTTP_404_NOT_FOUND,
            code="USER_NOT_FOUND",
@ -122,7 +122,7 @@ class UserNotFoundError(AuthException):
 class UserInactiveError(AuthException):
    """비활성화된 계정"""

-    def __init__(self, message: str = "비활성화된 계정입니다. 관리자에게 문의하세요."):
+    def __init__(self, message: str = "활성화 상태가 아닌 사용자 입니다."):
        super().__init__(
            status_code=status.HTTP_403_FORBIDDEN,
            code="USER_INACTIVE",
--- a/app/user/models.py
+++ b/app/user/models.py
@ -14,7 +14,7 @@ from sqlalchemy.orm import Mapped, mapped_column, relationship
 from app.database.session import Base

 if TYPE_CHECKING:
-    from app.home.models import UserProject
+    from app.home.models import Project


 class User(Base):
@ -26,7 +26,7 @@ class User(Base):
    Attributes:
        id: 고유 식별자 (자동 증가)
        kakao_id: 카카오 고유 ID (필수, 유니크)
-        user_ksuid: 사용자 식별을 위한 KSUID (필수, 유니크)
+        user_uuid: 사용자 식별을 위한 UUID7 (필수, 유니크)
        email: 이메일 주소 (선택, 카카오에서 제공 시)
        nickname: 카카오 닉네임 (선택)
        profile_image_url: 카카오 프로필 이미지 URL (선택)
@ -54,8 +54,7 @@ class User(Base):
        - 실제 데이터는 DB에 유지됨

    Relationships:
-        user_projects: Project와의 M:N 관계 (중계 테이블 통한 연결)
-        projects: 사용자가 참여한 프로젝트 목록 (Association Proxy)
+        projects: 사용자가 소유한 프로젝트 목록 (1:N 관계)

    카카오 API 응답 필드 매핑:
        - kakao_id: id (카카오 회원번호)
@ -70,7 +69,7 @@ class User(Base):
    __tablename__ = "user"
    __table_args__ = (
        Index("idx_user_kakao_id", "kakao_id"),
-        Index("idx_user_ksuid", "user_ksuid"),
+        Index("idx_user_uuid", "user_uuid"),
        Index("idx_user_email", "email"),
        Index("idx_user_phone", "phone"),
        Index("idx_user_is_active", "is_active"),
@ -105,11 +104,11 @@ class User(Base):
        comment="카카오 고유 ID (회원번호)",
    )

-    user_ksuid: Mapped[str] = mapped_column(
-        String(27),
+    user_uuid: Mapped[str] = mapped_column(
+        String(36),
        nullable=False,
        unique=True,
-        comment="사용자 식별을 위한 KSUID (K-Sortable Unique Identifier)",
+        comment="사용자 식별을 위한 UUID7 (시간순 정렬 가능한 UUID)",
    )

    # ==========================================================================
@ -231,16 +230,15 @@ class User(Base):
    )

    # ==========================================================================
-    # Project M:N 관계 (중계 테이블 UserProject 통한 연결)
+    # Project 1:N 관계 (한 사용자가 여러 프로젝트를 소유)
    # ==========================================================================
-    # back_populates: UserProject.user와 양방향 연결
-    # cascade: User 삭제 시 UserProject 레코드도 삭제 (Project는 유지)
+    # back_populates: Project.owner와 양방향 연결
+    # cascade: 사용자 삭제 시 프로젝트는 유지 (owner가 NULL로 설정됨)
    # lazy="selectin": N+1 문제 방지
    # ==========================================================================
-    user_projects: Mapped[List["UserProject"]] = relationship(
-        "UserProject",
-        back_populates="user",
-        cascade="all, delete-orphan",
+    projects: Mapped[List["Project"]] = relationship(
+        "Project",
+        back_populates="owner",
        lazy="selectin",
    )

@ -291,7 +289,7 @@ class RefreshToken(Base):
    Attributes:
        id: 고유 식별자 (자동 증가)
        user_id: 사용자 외래키 (User.id 참조)
-        user_ksuid: 사용자 KSUID (User.user_ksuid 참조)
+        user_uuid: 사용자 UUID (User.user_uuid 참조)
        token_hash: 리프레시 토큰의 SHA-256 해시값 (원본 저장 X)
        expires_at: 토큰 만료 일시
        is_revoked: 토큰 폐기 여부 (로그아웃 시 True)
@ -309,7 +307,7 @@ class RefreshToken(Base):
    __tablename__ = "refresh_token"
    __table_args__ = (
        Index("idx_refresh_token_user_id", "user_id"),
-        Index("idx_refresh_token_user_ksuid", "user_ksuid"),
+        Index("idx_refresh_token_user_uuid", "user_uuid"),
        Index("idx_refresh_token_token_hash", "token_hash", unique=True),
        Index("idx_refresh_token_expires_at", "expires_at"),
        Index("idx_refresh_token_is_revoked", "is_revoked"),
@ -335,10 +333,10 @@ class RefreshToken(Base):
        comment="사용자 외래키 (User.id 참조)",
    )

-    user_ksuid: Mapped[str] = mapped_column(
-        String(27),
+    user_uuid: Mapped[str] = mapped_column(
+        String(36),
        nullable=False,
-        comment="사용자 KSUID (User.user_ksuid 참조)",
+        comment="사용자 UUID (User.user_uuid 참조)",
    )

    token_hash: Mapped[str] = mapped_column(
@ -438,12 +436,12 @@ class SocialAccount(Base):

    __tablename__ = "social_account"
    __table_args__ = (
-        Index("idx_social_account_user_id", "user_id"),
+        Index("idx_social_account_user_uuid", "user_uuid"),
        Index("idx_social_account_platform", "platform"),
        Index("idx_social_account_is_active", "is_active"),
        Index(
            "uq_user_platform_account",
-            "user_id",
+            "user_uuid",
            "platform",
            "platform_user_id",
            unique=True,
@ -466,11 +464,11 @@ class SocialAccount(Base):
        comment="고유 식별자",
    )

-    user_id: Mapped[int] = mapped_column(
-        BigInteger,
-        ForeignKey("user.id", ondelete="CASCADE"),
+    user_uuid: Mapped[str] = mapped_column(
+        String(36),
+        ForeignKey("user.user_uuid", ondelete="CASCADE"),
        nullable=False,
-        comment="사용자 외래키 (User.id 참조)",
+        comment="사용자 외래키 (User.user_uuid 참조)",
    )

    # ==========================================================================
@ -573,7 +571,7 @@ class SocialAccount(Base):
        return (
            f"<SocialAccount("
            f"id={self.id}, "
-            f"user_id={self.user_id}, "
+            f"user_uuid='{self.user_uuid}', "
            f"platform='{self.platform}', "
            f"platform_username='{self.platform_username}', "
            f"is_active={self.is_active}"
--- a/app/user/services/auth.py
+++ b/app/user/services/auth.py
@ -24,7 +24,7 @@ from app.user.exceptions import (
    UserNotFoundError,
 )
 from app.user.models import RefreshToken, User
-from app.utils.common import generate_ksuid
+from app.utils.common import generate_uuid
 from app.user.schemas.user_schema import (
    AccessTokenResponse,
    KakaoUserInfo,
@ -96,28 +96,28 @@ class AuthService:

        # 5. JWT 토큰 생성
        logger.info("[AUTH] 5단계: JWT 토큰 생성 시작")
-        access_token = create_access_token(user.user_ksuid)
-        refresh_token = create_refresh_token(user.user_ksuid)
-        logger.debug(f"[AUTH] JWT 토큰 생성 완료 - user_ksuid: {user.user_ksuid}")
+        access_token = create_access_token(user.user_uuid)
+        refresh_token = create_refresh_token(user.user_uuid)
+        logger.debug(f"[AUTH] JWT 토큰 생성 완료 - user_uuid: {user.user_uuid}")

        # 6. 리프레시 토큰 DB 저장
        logger.info("[AUTH] 6단계: 리프레시 토큰 저장 시작")
        await self._save_refresh_token(
            user_id=user.id,
-            user_ksuid=user.user_ksuid,
+            user_uuid=user.user_uuid,
            token=refresh_token,
            session=session,
            user_agent=user_agent,
            ip_address=ip_address,
        )
-        logger.debug(f"[AUTH] 리프레시 토큰 저장 완료 - user_id: {user.id}, user_ksuid: {user.user_ksuid}")
+        logger.debug(f"[AUTH] 리프레시 토큰 저장 완료 - user_id: {user.id}, user_uuid: {user.user_uuid}")

        # 7. 마지막 로그인 시간 업데이트
        user.last_login_at = datetime.now(timezone.utc)
        await session.commit()

        redirect_url = f"{prj_settings.PROJECT_DOMAIN}"
-        logger.info(f"[AUTH] 카카오 로그인 완료 - user_id: {user.id}, user_ksuid: {user.user_ksuid}, redirect_url: {redirect_url}")
+        logger.info(f"[AUTH] 카카오 로그인 완료 - user_id: {user.id}, user_uuid: {user.user_uuid}, redirect_url: {redirect_url}")
        logger.debug(f"[AUTH] 응답 토큰 정보 - access_token: {access_token[:30]}..., refresh_token: {refresh_token[:30]}...")

        return LoginResponse(
@ -172,8 +172,8 @@ class AuthService:
            raise TokenExpiredError()

        # 4. 사용자 확인
-        user_ksuid = payload.get("sub")
-        user = await self._get_user_by_ksuid(user_ksuid, session)
+        user_uuid = payload.get("sub")
+        user = await self._get_user_by_uuid(user_uuid, session)

        if user is None:
            raise UserNotFoundError()
@ -182,7 +182,7 @@ class AuthService:
            raise UserInactiveError()

        # 5. 새 액세스 토큰 발급
-        new_access_token = create_access_token(user.user_ksuid)
+        new_access_token = create_access_token(user.user_uuid)

        return AccessTokenResponse(
            access_token=new_access_token,
@ -264,10 +264,10 @@ class AuthService:

        # 신규 사용자 생성
        logger.info(f"[AUTH] 신규 사용자 생성 시작 - kakao_id: {kakao_id}")
-        ksuid = await generate_ksuid(session=session, table_name=User)
+        user_uuid = await generate_uuid(session=session, table_name=User)
        new_user = User(
            kakao_id=kakao_id,
-            user_ksuid=ksuid,
+            user_uuid=user_uuid,
            email=kakao_account.email if kakao_account else None,
            nickname=profile.nickname if profile else None,
            profile_image_url=profile.profile_image_url if profile else None,
@ -316,7 +316,7 @@ class AuthService:
    async def _save_refresh_token(
        self,
        user_id: int,
-        user_ksuid: str,
+        user_uuid: str,
        token: str,
        session: AsyncSession,
        user_agent: Optional[str] = None,
@ -327,7 +327,7 @@ class AuthService:

        Args:
            user_id: 사용자 ID
-            user_ksuid: 사용자 KSUID
+            user_uuid: 사용자 UUID
            token: 리프레시 토큰
            session: DB 세션
            user_agent: User-Agent
@ -341,7 +341,7 @@ class AuthService:

        refresh_token = RefreshToken(
            user_id=user_id,
-            user_ksuid=user_ksuid,
+            user_uuid=user_uuid,
            token_hash=token_hash,
            expires_at=expires_at,
            user_agent=user_agent,
@ -391,23 +391,23 @@ class AuthService:
        )
        return result.scalar_one_or_none()

-    async def _get_user_by_ksuid(
+    async def _get_user_by_uuid(
        self,
-        user_ksuid: str,
+        user_uuid: str,
        session: AsyncSession,
    ) -> Optional[User]:
        """
-        KSUID로 사용자 조회
+        UUID로 사용자 조회

        Args:
-            user_ksuid: 사용자 KSUID
+            user_uuid: 사용자 UUID
            session: DB 세션

        Returns:
            User 객체 또는 None
        """
        result = await session.execute(
-            select(User).where(User.user_ksuid == user_ksuid, User.is_deleted == False)  # noqa: E712
+            select(User).where(User.user_uuid == user_uuid, User.is_deleted == False)  # noqa: E712
        )
        return result.scalar_one_or_none()

--- a/app/user/services/jwt.py
+++ b/app/user/services/jwt.py
@ -13,12 +13,12 @@ from jose import JWTError, jwt
 from config import jwt_settings


-def create_access_token(user_ksuid: str) -> str:
+def create_access_token(user_uuid: str) -> str:
    """
    JWT 액세스 토큰 생성

    Args:
-        user_ksuid: 사용자 KSUID
+        user_uuid: 사용자 UUID

    Returns:
        JWT 액세스 토큰 문자열
@ -27,7 +27,7 @@ def create_access_token(user_ksuid: str) -> str:
        minutes=jwt_settings.JWT_ACCESS_TOKEN_EXPIRE_MINUTES
    )
    to_encode = {
-        "sub": user_ksuid,
+        "sub": user_uuid,
        "exp": expire,
        "type": "access",
    }
@ -38,12 +38,12 @@ def create_access_token(user_ksuid: str) -> str:
    )


-def create_refresh_token(user_ksuid: str) -> str:
+def create_refresh_token(user_uuid: str) -> str:
    """
    JWT 리프레시 토큰 생성

    Args:
-        user_ksuid: 사용자 KSUID
+        user_uuid: 사용자 UUID

    Returns:
        JWT 리프레시 토큰 문자열
@ -52,7 +52,7 @@ def create_refresh_token(user_ksuid: str) -> str:
        days=jwt_settings.JWT_REFRESH_TOKEN_EXPIRE_DAYS
    )
    to_encode = {
-        "sub": user_ksuid,
+        "sub": user_uuid,
        "exp": expire,
        "type": "refresh",
    }
--- a/app/utils/common.py
+++ b/app/utils/common.py
@ -4,24 +4,71 @@ Common Utility Functions
 공통으로 사용되는 유틸리티 함수들을 정의합니다.

 사용 예시:
-    from app.utils.common import generate_task_id, generate_ksuid
+    from app.utils.common import generate_task_id, generate_uuid

    # task_id 생성
    task_id = await generate_task_id(session=session, table_name=Project)

-    # ksuid 생성
-    ksuid = await generate_ksuid(session=session, table_name=User)
+    # uuid 생성
+    user_uuid = await generate_uuid(session=session, table_name=User)

 Note:
    페이지네이션 기능은 app.utils.pagination 모듈을 사용하세요:
        from app.utils.pagination import PaginatedResponse, get_paginated
 """

+import os
+import time
 from typing import Any, Optional, Type

 from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
-from svix_ksuid import Ksuid
+
+
+def _generate_uuid7_string() -> str:
+    """UUID7 문자열을 생성합니다.
+
+    UUID7 구조 (RFC 9562):
+        - 48 bits: Unix timestamp (밀리초)
+        - 4 bits: 버전 (7)
+        - 12 bits: 랜덤
+        - 2 bits: variant (10)
+        - 62 bits: 랜덤
+        - 총 128 bits -> 36자 (하이픈 포함)
+
+    Returns:
+        36자리 UUID7 문자열 (xxxxxxxx-xxxx-7xxx-yxxx-xxxxxxxxxxxx)
+    """
+    # 현재 시간 (밀리초)
+    timestamp_ms = int(time.time() * 1000)
+
+    # 랜덤 바이트 (10바이트 = 80비트)
+    random_bytes = os.urandom(10)
+
+    # UUID7 바이트 구성 (16바이트 = 128비트)
+    # 처음 6바이트: 타임스탬프 (48비트)
+    uuid_bytes = timestamp_ms.to_bytes(6, byteorder="big")
+
+    # 다음 2바이트: 버전(7) + 랜덤 12비트
+    # 0x7000 | (random 12 bits)
+    rand_a = int.from_bytes(random_bytes[0:2], byteorder="big")
+    version_rand = (0x7000 | (rand_a & 0x0FFF)).to_bytes(2, byteorder="big")
+    uuid_bytes += version_rand
+
+    # 다음 2바이트: variant(10) + 랜덤 62비트의 앞 6비트
+    # 0x80 | (random 6 bits) + random 8 bits
+    rand_b = random_bytes[2]
+    variant_rand = bytes([0x80 | (rand_b & 0x3F)]) + random_bytes[3:4]
+    uuid_bytes += variant_rand
+
+    # 나머지 6바이트: 랜덤
+    uuid_bytes += random_bytes[4:10]
+
+    # 16진수로 변환
+    hex_str = uuid_bytes.hex()
+
+    # UUID 형식으로 포맷팅 (8-4-4-4-12)
+    return f"{hex_str[:8]}-{hex_str[8:12]}-{hex_str[12:16]}-{hex_str[16:20]}-{hex_str[20:32]}"


 async def generate_task_id(
@ -35,16 +82,16 @@ async def generate_task_id(
        table_name: task_id 컬럼이 있는 SQLAlchemy 테이블 클래스 (optional)

    Returns:
-        str: 생성된 ksuid 문자열
+        str: 생성된 UUID7 문자열 (36자)

    Usage:
-        # 단순 ksuid 생성
+        # 단순 UUID7 생성
        task_id = await generate_task_id()

        # 테이블에서 중복 검사 후 생성
        task_id = await generate_task_id(session=session, table_name=Project)
    """
-    task_id = str(Ksuid())
+    task_id = _generate_uuid7_string()

    if session is None or table_name is None:
        return task_id
@ -58,41 +105,41 @@ async def generate_task_id(
        if existing is None:
            return task_id

-        task_id = str(Ksuid())
+        task_id = _generate_uuid7_string()


-async def generate_ksuid(
+async def generate_uuid(
    session: Optional[AsyncSession] = None,
    table_name: Optional[Type[Any]] = None,
 ) -> str:
-    """고유한 ksuid를 생성합니다.
+    """고유한 UUID7을 생성합니다.

    Args:
        session: SQLAlchemy AsyncSession (optional)
-        table_name: user_ksuid 컬럼이 있는 SQLAlchemy 테이블 클래스 (optional)
+        table_name: user_uuid 컬럼이 있는 SQLAlchemy 테이블 클래스 (optional)

    Returns:
-        str: 생성된 ksuid 문자열
+        str: 생성된 UUID7 문자열 (36자)

    Usage:
-        # 단순 ksuid 생성
-        ksuid = await generate_ksuid()
+        # 단순 UUID7 생성
+        new_uuid = await generate_uuid()

        # 테이블에서 중복 검사 후 생성
-        ksuid = await generate_ksuid(session=session, table_name=User)
+        new_uuid = await generate_uuid(session=session, table_name=User)
    """
-    ksuid = str(Ksuid())
+    new_uuid = _generate_uuid7_string()

    if session is None or table_name is None:
-        return ksuid
+        return new_uuid

    while True:
        result = await session.execute(
-            select(table_name).where(table_name.user_ksuid == ksuid)
+            select(table_name).where(table_name.user_uuid == new_uuid)
        )
        existing = result.scalar_one_or_none()

        if existing is None:
-            return ksuid
+            return new_uuid

-        ksuid = str(Ksuid())
+        new_uuid = _generate_uuid7_string()
--- a/app/utils/upload_blob_as_request.py
+++ b/app/utils/upload_blob_as_request.py
@ -5,14 +5,14 @@ Azure Blob Storage에 파일을 업로드하는 클래스를 제공합니다.
 파일 경로 또는 바이트 데이터를 직접 업로드할 수 있습니다.

 URL 경로 형식:
-    - 음악: {BASE_URL}/{task_id}/song/{파일명}
-    - 영상: {BASE_URL}/{task_id}/video/{파일명}
-    - 이미지: {BASE_URL}/{task_id}/image/{파일명}
+    - 음악: {BASE_URL}/{user_uuid}/{task_id}/song/{파일명}
+    - 영상: {BASE_URL}/{user_uuid}/{task_id}/video/{파일명}
+    - 이미지: {BASE_URL}/{user_uuid}/{task_id}/image/{파일명}

 사용 예시:
    from app.utils.upload_blob_as_request import AzureBlobUploader

-    uploader = AzureBlobUploader(task_id="task-123")
+    uploader = AzureBlobUploader(user_uuid="user-abc", task_id="task-123")

    # 파일 경로로 업로드
    success = await uploader.upload_music(file_path="my_song.mp3")
@ -79,14 +79,15 @@ class AzureBlobUploader:
    """Azure Blob Storage 업로드 클래스

    Azure Blob Storage에 음악, 영상, 이미지 파일을 업로드합니다.
-    URL 형식: {BASE_URL}/{task_id}/{category}/{file_name}?{SAS_TOKEN}
+    URL 형식: {BASE_URL}/{user_uuid}/{task_id}/{category}/{file_name}?{SAS_TOKEN}

    카테고리별 경로:
-    - 음악: {task_id}/song/{file_name}
-    - 영상: {task_id}/video/{file_name}
-    - 이미지: {task_id}/image/{file_name}
+    - 음악: {user_uuid}/{task_id}/song/{file_name}
+    - 영상: {user_uuid}/{task_id}/video/{file_name}
+    - 이미지: {user_uuid}/{task_id}/image/{file_name}

    Attributes:
+        user_uuid: 사용자 고유 식별자 (UUID)
        task_id: 작업 고유 식별자
    """

@ -100,17 +101,24 @@ class AzureBlobUploader:
        ".bmp": "image/bmp",
    }

-    def __init__(self, task_id: str):
+    def __init__(self, user_uuid: str, task_id: str):
        """AzureBlobUploader 초기화

        Args:
+            user_uuid: 사용자 고유 식별자 (UUID)
            task_id: 작업 고유 식별자
        """
+        self._user_uuid = user_uuid
        self._task_id = task_id
        self._base_url = azure_blob_settings.AZURE_BLOB_BASE_URL
        self._sas_token = azure_blob_settings.AZURE_BLOB_SAS_TOKEN
        self._last_public_url: str = ""

+    @property
+    def user_uuid(self) -> str:
+        """사용자 고유 식별자 (UUID)"""
+        return self._user_uuid
+
    @property
    def task_id(self) -> str:
        """작업 고유 식별자"""
@ -126,12 +134,12 @@ class AzureBlobUploader:
        # SAS 토큰 앞뒤의 ?, ', " 제거
        sas_token = self._sas_token.strip("?'\"")
        return (
-            f"{self._base_url}/{self._task_id}/{category}/{file_name}?{sas_token}"
+            f"{self._base_url}/{self._user_uuid}/{self._task_id}/{category}/{file_name}?{sas_token}"
        )

    def _build_public_url(self, category: str, file_name: str) -> str:
        """공개 URL 생성 (SAS 토큰 제외)"""
-        return f"{self._base_url}/{self._task_id}/{category}/{file_name}"
+        return f"{self._base_url}/{self._user_uuid}/{self._task_id}/{category}/{file_name}"

    async def _upload_bytes(
        self,
@ -253,7 +261,7 @@ class AzureBlobUploader:
    async def upload_music(self, file_path: str) -> bool:
        """음악 파일을 Azure Blob Storage에 업로드합니다.

-        URL 경로: {task_id}/song/{파일명}
+        URL 경로: {user_uuid}/{task_id}/song/{파일명}

        Args:
            file_path: 업로드할 파일 경로
@ -262,7 +270,7 @@ class AzureBlobUploader:
            bool: 업로드 성공 여부

        Example:
-            uploader = AzureBlobUploader(task_id="task-123")
+            uploader = AzureBlobUploader(user_uuid="user-abc", task_id="task-123")
            success = await uploader.upload_music(file_path="my_song.mp3")
            print(uploader.public_url)
        """
@ -279,7 +287,7 @@ class AzureBlobUploader:
    ) -> bool:
        """음악 바이트 데이터를 Azure Blob Storage에 직접 업로드합니다.

-        URL 경로: {task_id}/song/{파일명}
+        URL 경로: {user_uuid}/{task_id}/song/{파일명}

        Args:
            file_content: 업로드할 파일 바이트 데이터
@ -289,7 +297,7 @@ class AzureBlobUploader:
            bool: 업로드 성공 여부

        Example:
-            uploader = AzureBlobUploader(task_id="task-123")
+            uploader = AzureBlobUploader(user_uuid="user-abc", task_id="task-123")
            success = await uploader.upload_music_bytes(audio_bytes, "my_song")
            print(uploader.public_url)
        """
@ -315,7 +323,7 @@ class AzureBlobUploader:
    async def upload_video(self, file_path: str) -> bool:
        """영상 파일을 Azure Blob Storage에 업로드합니다.

-        URL 경로: {task_id}/video/{파일명}
+        URL 경로: {user_uuid}/{task_id}/video/{파일명}

        Args:
            file_path: 업로드할 파일 경로
@ -324,7 +332,7 @@ class AzureBlobUploader:
            bool: 업로드 성공 여부

        Example:
-            uploader = AzureBlobUploader(task_id="task-123")
+            uploader = AzureBlobUploader(user_uuid="user-abc", task_id="task-123")
            success = await uploader.upload_video(file_path="my_video.mp4")
            print(uploader.public_url)
        """
@ -341,7 +349,7 @@ class AzureBlobUploader:
    ) -> bool:
        """영상 바이트 데이터를 Azure Blob Storage에 직접 업로드합니다.

-        URL 경로: {task_id}/video/{파일명}
+        URL 경로: {user_uuid}/{task_id}/video/{파일명}

        Args:
            file_content: 업로드할 파일 바이트 데이터
@ -351,7 +359,7 @@ class AzureBlobUploader:
            bool: 업로드 성공 여부

        Example:
-            uploader = AzureBlobUploader(task_id="task-123")
+            uploader = AzureBlobUploader(user_uuid="user-abc", task_id="task-123")
            success = await uploader.upload_video_bytes(video_bytes, "my_video")
            print(uploader.public_url)
        """
@ -377,7 +385,7 @@ class AzureBlobUploader:
    async def upload_image(self, file_path: str) -> bool:
        """이미지 파일을 Azure Blob Storage에 업로드합니다.

-        URL 경로: {task_id}/image/{파일명}
+        URL 경로: {user_uuid}/{task_id}/image/{파일명}

        Args:
            file_path: 업로드할 파일 경로
@ -386,7 +394,7 @@ class AzureBlobUploader:
            bool: 업로드 성공 여부

        Example:
-            uploader = AzureBlobUploader(task_id="task-123")
+            uploader = AzureBlobUploader(user_uuid="user-abc", task_id="task-123")
            success = await uploader.upload_image(file_path="my_image.png")
            print(uploader.public_url)
        """
@ -406,7 +414,7 @@ class AzureBlobUploader:
    ) -> bool:
        """이미지 바이트 데이터를 Azure Blob Storage에 직접 업로드합니다.

-        URL 경로: {task_id}/image/{파일명}
+        URL 경로: {user_uuid}/{task_id}/image/{파일명}

        Args:
            file_content: 업로드할 파일 바이트 데이터
@ -416,7 +424,7 @@ class AzureBlobUploader:
            bool: 업로드 성공 여부

        Example:
-            uploader = AzureBlobUploader(task_id="task-123")
+            uploader = AzureBlobUploader(user_uuid="user-abc", task_id="task-123")
            with open("my_image.png", "rb") as f:
                content = f.read()
            success = await uploader.upload_image_bytes(content, "my_image.png")
@ -445,17 +453,17 @@ class AzureBlobUploader:
 # import asyncio
 #
 # async def main():
-#     uploader = AzureBlobUploader(task_id="task-123")
+#     uploader = AzureBlobUploader(user_uuid="user-abc", task_id="task-123")
 #
-#     # 음악 업로드 -> {BASE_URL}/task-123/song/my_song.mp3
+#     # 음악 업로드 -> {BASE_URL}/user-abc/task-123/song/my_song.mp3
 #     await uploader.upload_music("my_song.mp3")
 #     print(uploader.public_url)
 #
-#     # 영상 업로드 -> {BASE_URL}/task-123/video/my_video.mp4
+#     # 영상 업로드 -> {BASE_URL}/user-abc/task-123/video/my_video.mp4
 #     await uploader.upload_video("my_video.mp4")
 #     print(uploader.public_url)
 #
-#     # 이미지 업로드 -> {BASE_URL}/task-123/image/my_image.png
+#     # 이미지 업로드 -> {BASE_URL}/user-abc/task-123/image/my_image.png
 #     await uploader.upload_image("my_image.png")
 #     print(uploader.public_url)
 #
--- a/app/video/api/routers/v1/video.py
+++ b/app/video/api/routers/v1/video.py
@ -23,6 +23,8 @@ from sqlalchemy.ext.asyncio import AsyncSession

 from app.database.session import get_session
 from app.dependencies.pagination import PaginationParams, get_pagination_params
+from app.user.dependencies.auth import get_current_user
+from app.user.models import User
 from app.home.models import Image, Project
 from app.lyric.models import Lyric
 from app.song.models import Song, SongTimestamp
@ -96,6 +98,7 @@ async def generate_video(
        default="vertical",
        description="영상 방향 (horizontal: 가로형, vertical: 세로형)",
    ),
+    current_user: User = Depends(get_current_user),
 ) -> GenerateVideoResponse:
    """Creatomate API를 통해 영상을 생성합니다.

@ -489,6 +492,7 @@ GET /video/status/render-id-123...
 async def get_video_status(
    creatomate_render_id: str,
    background_tasks: BackgroundTasks,
+    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
 ) -> PollingVideoResponse:
    """creatomate_render_id로 영상 생성 작업의 상태를 조회합니다.
@ -629,6 +633,7 @@ GET /video/download/019123ab-cdef-7890-abcd-ef1234567890
 )
 async def download_video(
    task_id: str,
+    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
 ) -> DownloadVideoResponse:
    """task_id로 Video 상태를 polling하고 completed 시 Project 정보와 영상 URL을 반환합니다."""
@ -743,6 +748,7 @@ GET /videos/?page=1&page_size=10
    },
 )
 async def get_videos(
+    current_user: User = Depends(get_current_user),
    session: AsyncSession = Depends(get_session),
    pagination: PaginationParams = Depends(get_pagination_params),
 ) -> PaginatedResponse[VideoListItem]:
--- a/app/video/models.py
+++ b/app/video/models.py
@ -24,7 +24,7 @@ class Video(Base):
        project_id: 연결된 Project의 id (외래키)
        lyric_id: 연결된 Lyric의 id (외래키)
        song_id: 연결된 Song의 id (외래키)
-        task_id: 영상 생성 작업의 고유 식별자 (KSUID 형식)
+        task_id: 영상 생성 작업의 고유 식별자 (UUID7 형식)
        status: 처리 상태 (pending, processing, completed, failed 등)
        result_movie_url: 생성된 영상 URL (S3, CDN 경로)
        created_at: 생성 일시 (자동 설정)
@ -78,10 +78,10 @@ class Video(Base):
    )

    task_id: Mapped[str] = mapped_column(
-        String(27),
+        String(36),
        nullable=False,
        unique=True,
-        comment="영상 생성 작업 고유 식별자 (KSUID)",
+        comment="영상 생성 작업 고유 식별자 (UUID7)",
    )

    creatomate_render_id: Mapped[Optional[str]] = mapped_column(
--- a/error_plan.md
+++ b/error_plan.md
@ -1,571 +0,0 @@
-# Suno API & Creatomate API 에러 처리 작업 계획서
-
-## 현재 상태 분석
-
-### ✅ 이미 구현된 항목
-| 항목 | Suno API | Creatomate API |
-|------|----------|----------------|
-| DB failed 상태 저장 | ✅ `song_task.py` | ✅ `video_task.py` |
-| Response failed 상태 반환 | ✅ `song_schema.py` | ✅ `video_schema.py` |
-
-### ❌ 미구현 항목
-| 항목 | Suno API | Creatomate API |
-|------|----------|----------------|
-| 타임아웃 외부화 | ❌ 하드코딩됨 | ❌ 하드코딩됨 |
-| 재시도 로직 | ❌ 없음 | ❌ 없음 |
-| 커스텀 예외 클래스 | ❌ 없음 | ❌ 없음 |
-
---
-
-## 1. RecoverySettings에 Suno/Creatomate 설정 추가
-
-### 파일: `config.py`
-
-**변경 전:**
-```python
-class RecoverySettings(BaseSettings):
-    """ChatGPT API 복구 및 타임아웃 설정"""
-
-    CHATGPT_TIMEOUT: float = Field(
-        default=600.0,
-        description="ChatGPT API 타임아웃 (초). OpenAI Python SDK 기본값: 600초 (10분)",
-    )
-    CHATGPT_MAX_RETRIES: int = Field(
-        default=1,
-        description="ChatGPT API 응답 실패 시 최대 재시도 횟수",
-    )
-
-    model_config = _base_config
-```
-
-**변경 후:**
-```python
-class RecoverySettings(BaseSettings):
-    """외부 API 복구 및 타임아웃 설정
-
-    ChatGPT, Suno, Creatomate API의 타임아웃 및 재시도 설정을 관리합니다.
-    """
-
-    # ============================================================
-    # ChatGPT API 설정
-    # ============================================================
-    CHATGPT_TIMEOUT: float = Field(
-        default=600.0,
-        description="ChatGPT API 타임아웃 (초). OpenAI Python SDK 기본값: 600초 (10분)",
-    )
-    CHATGPT_MAX_RETRIES: int = Field(
-        default=1,
-        description="ChatGPT API 응답 실패 시 최대 재시도 횟수",
-    )
-
-    # ============================================================
-    # Suno API 설정
-    # ============================================================
-    SUNO_DEFAULT_TIMEOUT: float = Field(
-        default=30.0,
-        description="Suno API 기본 요청 타임아웃 (초)",
-    )
-    SUNO_LYRIC_TIMEOUT: float = Field(
-        default=120.0,
-        description="Suno API 가사 타임스탬프 요청 타임아웃 (초)",
-    )
-    SUNO_MAX_RETRIES: int = Field(
-        default=2,
-        description="Suno API 응답 실패 시 최대 재시도 횟수",
-    )
-
-    # ============================================================
-    # Creatomate API 설정
-    # ============================================================
-    CREATOMATE_DEFAULT_TIMEOUT: float = Field(
-        default=30.0,
-        description="Creatomate API 기본 요청 타임아웃 (초)",
-    )
-    CREATOMATE_RENDER_TIMEOUT: float = Field(
-        default=60.0,
-        description="Creatomate API 렌더링 요청 타임아웃 (초)",
-    )
-    CREATOMATE_CONNECT_TIMEOUT: float = Field(
-        default=10.0,
-        description="Creatomate API 연결 타임아웃 (초)",
-    )
-    CREATOMATE_MAX_RETRIES: int = Field(
-        default=2,
-        description="Creatomate API 응답 실패 시 최대 재시도 횟수",
-    )
-
-    model_config = _base_config
-```
-
-**이유:** 모든 외부 API의 타임아웃/재시도 설정을 `RecoverySettings` 하나에서 통합 관리하여 일관성을 유지합니다.
-
-### 파일: `.env`
-
-**추가할 내용:**
-```env
-# ============================================================
-# 외부 API 타임아웃 및 재시도 설정 (RecoverySettings)
-# ============================================================
-
-# ChatGPT API (기존)
-CHATGPT_TIMEOUT=600.0
-CHATGPT_MAX_RETRIES=1
-
-# Suno API
-SUNO_DEFAULT_TIMEOUT=30.0
-SUNO_LYRIC_TIMEOUT=120.0
-SUNO_MAX_RETRIES=2
-
-# Creatomate API
-CREATOMATE_DEFAULT_TIMEOUT=30.0
-CREATOMATE_RENDER_TIMEOUT=60.0
-CREATOMATE_CONNECT_TIMEOUT=10.0
-CREATOMATE_MAX_RETRIES=2
-```
-
---
-
-## 2. Suno API 커스텀 예외 클래스 추가
-
-### 파일: `app/utils/suno.py`
-
-**변경 전 (라인 1-20):**
-```python
-import httpx
-import json
-from typing import Optional
-
-from app.utils.logger import get_logger
-
-logger = get_logger("suno")
-```
-
-**변경 후:**
-```python
-import httpx
-import json
-from typing import Optional
-
-from app.utils.logger import get_logger
-from config import recovery_settings
-
-logger = get_logger("suno")
-
-
-class SunoResponseError(Exception):
-    """Suno API 응답 오류 시 발생하는 예외
-
-    Suno API 거부 응답 또는 비정상 응답 시 사용됩니다.
-    재시도 로직에서 이 예외를 catch하여 재시도를 수행합니다.
-
-    Attributes:
-        message: 에러 메시지
-        original_response: 원본 API 응답 (있는 경우)
-    """
-    def __init__(self, message: str, original_response: dict | None = None):
-        self.message = message
-        self.original_response = original_response
-        super().__init__(self.message)
-```
-
-**이유:** ChatGPT API와 동일하게 커스텀 예외 클래스를 추가하여 Suno API 오류를 명확히 구분하고 재시도 로직에서 활용합니다.
-
---
-
-## 3. Suno API 타임아웃 적용
-
-### 파일: `app/utils/suno.py`
-
-**변경 전 (라인 130):**
-```python
-async with httpx.AsyncClient() as client:
-    response = await client.post(
-        f"{self.base_url}/generate",
-        headers=self.headers,
-        json=payload,
-        timeout=30.0,
-    )
-```
-
-**변경 후:**
-```python
-async with httpx.AsyncClient() as client:
-    response = await client.post(
-        f"{self.base_url}/generate",
-        headers=self.headers,
-        json=payload,
-        timeout=recovery_settings.SUNO_DEFAULT_TIMEOUT,
-    )
-```
-
-**변경 전 (라인 173):**
-```python
-timeout=30.0,
-```
-
-**변경 후:**
-```python
-timeout=recovery_settings.SUNO_DEFAULT_TIMEOUT,
-```
-
-**변경 전 (라인 201):**
-```python
-timeout=120.0,
-```
-
-**변경 후:**
-```python
-timeout=recovery_settings.SUNO_LYRIC_TIMEOUT,
-```
-
-**이유:** 환경변수로 타임아웃을 관리하여 배포 환경별로 유연하게 조정할 수 있습니다.
-
---
-
-## 4. Suno API 재시도 로직 추가
-
-### 파일: `app/utils/suno.py`
-
-**변경 전 - generate() 메서드:**
-```python
-async def generate(
-    self,
-    lyric: str,
-    style: str,
-    title: str,
-    task_id: str,
-) -> dict:
-    """음악 생성 요청"""
-    payload = {
-        "prompt": lyric,
-        "style": style,
-        "title": title,
-        "customMode": True,
-        "callbackUrl": f"{self.callback_url}?task_id={task_id}",
-    }
-
-    async with httpx.AsyncClient() as client:
-        response = await client.post(
-            f"{self.base_url}/generate",
-            headers=self.headers,
-            json=payload,
-            timeout=30.0,
-        )
-
-        if response.status_code != 200:
-            logger.error(f"Failed to generate music: {response.text}")
-            raise Exception(f"Failed to generate music: {response.status_code}")
-
-        return response.json()
-```
-
-**변경 후:**
-```python
-async def generate(
-    self,
-    lyric: str,
-    style: str,
-    title: str,
-    task_id: str,
-) -> dict:
-    """음악 생성 요청 (재시도 로직 포함)
-
-    Args:
-        lyric: 가사 텍스트
-        style: 음악 스타일
-        title: 곡 제목
-        task_id: 작업 고유 식별자
-
-    Returns:
-        Suno API 응답 데이터
-
-    Raises:
-        SunoResponseError: API 오류 또는 재시도 실패 시
-    """
-    payload = {
-        "prompt": lyric,
-        "style": style,
-        "title": title,
-        "customMode": True,
-        "callbackUrl": f"{self.callback_url}?task_id={task_id}",
-    }
-
-    last_error: Exception | None = None
-
-    for attempt in range(recovery_settings.SUNO_MAX_RETRIES + 1):
-        try:
-            async with httpx.AsyncClient() as client:
-                response = await client.post(
-                    f"{self.base_url}/generate",
-                    headers=self.headers,
-                    json=payload,
-                    timeout=recovery_settings.SUNO_DEFAULT_TIMEOUT,
-                )
-
-                if response.status_code == 200:
-                    return response.json()
-
-                # 재시도 불가능한 오류 (4xx 클라이언트 오류)
-                if 400 <= response.status_code < 500:
-                    raise SunoResponseError(
-                        f"Client error: {response.status_code}",
-                        original_response={"status": response.status_code, "text": response.text}
-                    )
-
-                # 재시도 가능한 오류 (5xx 서버 오류)
-                last_error = SunoResponseError(
-                    f"Server error: {response.status_code}",
-                    original_response={"status": response.status_code, "text": response.text}
-                )
-
-        except httpx.TimeoutException as e:
-            logger.warning(f"[Suno] Timeout on attempt {attempt + 1}/{recovery_settings.SUNO_MAX_RETRIES + 1}")
-            last_error = e
-
-        except httpx.HTTPError as e:
-            logger.warning(f"[Suno] HTTP error on attempt {attempt + 1}: {e}")
-            last_error = e
-
-        # 마지막 시도가 아니면 재시도
-        if attempt < recovery_settings.SUNO_MAX_RETRIES:
-            logger.info(f"[Suno] Retrying... ({attempt + 1}/{recovery_settings.SUNO_MAX_RETRIES})")
-
-    # 모든 재시도 실패
-    raise SunoResponseError(
-        f"All {recovery_settings.SUNO_MAX_RETRIES + 1} attempts failed",
-        original_response={"last_error": str(last_error)}
-    )
-```
-
-**이유:** 네트워크 오류나 일시적인 서버 오류 시 자동으로 재시도하여 안정성을 높입니다.
-
---
-
-## 5. Creatomate API 커스텀 예외 클래스 추가
-
-### 파일: `app/utils/creatomate.py`
-
-**변경 전 (라인 1-20):**
-```python
-import asyncio
-import json
-from typing import Any, Optional
-
-import httpx
-
-from app.utils.logger import get_logger
-from config import creatomate_settings
-
-logger = get_logger("creatomate")
-```
-
-**변경 후:**
-```python
-import asyncio
-import json
-from typing import Any, Optional
-
-import httpx
-
-from app.utils.logger import get_logger
-from config import creatomate_settings, recovery_settings
-
-logger = get_logger("creatomate")
-
-
-class CreatomateResponseError(Exception):
-    """Creatomate API 응답 오류 시 발생하는 예외
-
-    Creatomate API 렌더링 실패 또는 비정상 응답 시 사용됩니다.
-    재시도 로직에서 이 예외를 catch하여 재시도를 수행합니다.
-
-    Attributes:
-        message: 에러 메시지
-        original_response: 원본 API 응답 (있는 경우)
-    """
-    def __init__(self, message: str, original_response: dict | None = None):
-        self.message = message
-        self.original_response = original_response
-        super().__init__(self.message)
-```
-
---
-
-## 6. Creatomate API 타임아웃 적용
-
-### 파일: `app/utils/creatomate.py`
-
-**변경 전 (라인 138):**
-```python
-self._client = httpx.AsyncClient(
-    base_url=self.BASE_URL,
-    headers=self._get_headers(),
-    timeout=httpx.Timeout(60.0, connect=10.0),
-)
-```
-
-**변경 후:**
-```python
-self._client = httpx.AsyncClient(
-    base_url=self.BASE_URL,
-    headers=self._get_headers(),
-    timeout=httpx.Timeout(
-        recovery_settings.CREATOMATE_RENDER_TIMEOUT,
-        connect=recovery_settings.CREATOMATE_CONNECT_TIMEOUT
-    ),
-)
-```
-
-**변경 전 (라인 258, 291):**
-```python
-timeout=30.0
-```
-
-**변경 후:**
-```python
-timeout=recovery_settings.CREATOMATE_DEFAULT_TIMEOUT
-```
-
-**변경 전 (라인 446, 457):**
-```python
-timeout=60.0
-```
-
-**변경 후:**
-```python
-timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT
-```
-
---
-
-## 7. Creatomate API 재시도 로직 추가
-
-### 파일: `app/utils/creatomate.py`
-
-**변경 전 - render_with_json() 메서드 (라인 440~):**
-```python
-async def render_with_json(
-    self,
-    template_id: str,
-    modifications: dict[str, Any],
-    task_id: str,
-) -> dict:
-    """JSON 수정사항으로 렌더링 요청"""
-    payload = {
-        "template_id": template_id,
-        "modifications": modifications,
-        "webhook_url": f"{creatomate_settings.CREATOMATE_CALLBACK_URL}?task_id={task_id}",
-    }
-
-    async with httpx.AsyncClient() as client:
-        response = await client.post(
-            f"{self.BASE_URL}/v1/renders",
-            headers=self._get_headers(),
-            json=payload,
-            timeout=60.0,
-        )
-
-        if response.status_code != 200:
-            logger.error(f"Failed to render: {response.text}")
-            raise Exception(f"Failed to render: {response.status_code}")
-
-        return response.json()
-```
-
-**변경 후:**
-```python
-async def render_with_json(
-    self,
-    template_id: str,
-    modifications: dict[str, Any],
-    task_id: str,
-) -> dict:
-    """JSON 수정사항으로 렌더링 요청 (재시도 로직 포함)
-
-    Args:
-        template_id: Creatomate 템플릿 ID
-        modifications: 수정사항 딕셔너리
-        task_id: 작업 고유 식별자
-
-    Returns:
-        Creatomate API 응답 데이터
-
-    Raises:
-        CreatomateResponseError: API 오류 또는 재시도 실패 시
-    """
-    payload = {
-        "template_id": template_id,
-        "modifications": modifications,
-        "webhook_url": f"{creatomate_settings.CREATOMATE_CALLBACK_URL}?task_id={task_id}",
-    }
-
-    last_error: Exception | None = None
-
-    for attempt in range(recovery_settings.CREATOMATE_MAX_RETRIES + 1):
-        try:
-            async with httpx.AsyncClient() as client:
-                response = await client.post(
-                    f"{self.BASE_URL}/v1/renders",
-                    headers=self._get_headers(),
-                    json=payload,
-                    timeout=recovery_settings.CREATOMATE_RENDER_TIMEOUT,
-                )
-
-                if response.status_code == 200:
-                    return response.json()
-
-                # 재시도 불가능한 오류 (4xx 클라이언트 오류)
-                if 400 <= response.status_code < 500:
-                    raise CreatomateResponseError(
-                        f"Client error: {response.status_code}",
-                        original_response={"status": response.status_code, "text": response.text}
-                    )
-
-                # 재시도 가능한 오류 (5xx 서버 오류)
-                last_error = CreatomateResponseError(
-                    f"Server error: {response.status_code}",
-                    original_response={"status": response.status_code, "text": response.text}
-                )
-
-        except httpx.TimeoutException as e:
-            logger.warning(f"[Creatomate] Timeout on attempt {attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES + 1}")
-            last_error = e
-
-        except httpx.HTTPError as e:
-            logger.warning(f"[Creatomate] HTTP error on attempt {attempt + 1}: {e}")
-            last_error = e
-
-        # 마지막 시도가 아니면 재시도
-        if attempt < recovery_settings.CREATOMATE_MAX_RETRIES:
-            logger.info(f"[Creatomate] Retrying... ({attempt + 1}/{recovery_settings.CREATOMATE_MAX_RETRIES})")
-
-    # 모든 재시도 실패
-    raise CreatomateResponseError(
-        f"All {recovery_settings.CREATOMATE_MAX_RETRIES + 1} attempts failed",
-        original_response={"last_error": str(last_error)}
-    )
-```
-
---
-
-## 작업 체크리스트
-
-| 순번 | 작업 내용 | 파일 | 상태 |
-|------|----------|------|------|
-| 1 | RecoverySettings에 Suno/Creatomate 설정 추가 | `config.py` | ✅ |
-| 2 | .env에 타임아웃/재시도 환경변수 추가 | `.env` | ✅ |
-| 3 | SunoResponseError 예외 클래스 추가 | `app/utils/suno.py` | ✅ |
-| 4 | Suno 타임아웃 적용 (recovery_settings 사용) | `app/utils/suno.py` | ✅ |
-| 5 | Suno 재시도 로직 추가 | `app/utils/suno.py` | ✅ |
-| 6 | CreatomateResponseError 예외 클래스 추가 | `app/utils/creatomate.py` | ✅ |
-| 7 | Creatomate 타임아웃 적용 (recovery_settings 사용) | `app/utils/creatomate.py` | ✅ |
-| 8 | Creatomate 재시도 로직 추가 | `app/utils/creatomate.py` | ✅ |
-
---
-
-## 참고사항
-
- **DB failed 상태 저장**: `song_task.py`와 `video_task.py`에 이미 구현되어 있습니다.
- **Response failed 상태**: 모든 스키마에 `success`, `status` 필드가 이미 존재합니다.
- 재시도는 5xx 서버 오류와 타임아웃에만 적용되며, 4xx 클라이언트 오류는 즉시 실패 처리합니다.
- 모든 타임아웃/재시도 설정은 `RecoverySettings`에서 통합 관리합니다.
--- a/main.py
+++ b/main.py
@ -8,11 +8,11 @@ from app.admin_manager import init_admin
 from app.core.common import lifespan
 from app.database.session import engine

-# 주의: User 모델을 먼저 import해야 UserProject가 User를 참조할 수 있음
+# User 모델 import (테이블 메타데이터 등록용)
 from app.user.models import User, RefreshToken  # noqa: F401

 from app.home.api.routers.v1.home import router as home_router
-from app.user.api.routers.v1.auth import router as auth_router
+from app.user.api.routers.v1.auth import router as auth_router, test_router as auth_test_router
 from app.lyric.api.routers.v1.lyric import router as lyric_router
 from app.song.api.routers.v1.song import router as song_router
 from app.video.api.routers.v1.video import router as video_router
@ -84,6 +84,25 @@ tags_metadata = [
    },
 ]

+# DEBUG 모드에서만 Test Auth 태그 추가
+if prj_settings.DEBUG:
+    tags_metadata.append(
+        {
+            "name": "Test Auth",
+            "description": """테스트용 인증 API (DEBUG 모드 전용)
+
+**주의: 이 API는 DEBUG 모드에서만 사용 가능합니다.**
+
+카카오 로그인 없이 테스트용 사용자 생성 및 토큰 발급이 가능합니다.
+
+## 테스트 흐름
+
+1. `POST /api/v1/user/auth/test/create-user` - 테스트 사용자 생성
+2. `POST /api/v1/user/auth/test/generate-token` - JWT 토큰 발급
+""",
+        }
+    )
+
 app = FastAPI(
    title=prj_settings.PROJECT_NAME,
    version=prj_settings.VERSION,
@ -163,3 +182,7 @@ app.include_router(auth_router, prefix="/user")  # Auth API 라우터 추가
 app.include_router(lyric_router)
 app.include_router(song_router)
 app.include_router(video_router)
+
+# DEBUG 모드에서만 테스트 라우터 등록
+if prj_settings.DEBUG:
+    app.include_router(auth_test_router, prefix="/user")  # Test Auth API 라우터
--- a/pyproject.toml
+++ b/pyproject.toml
@ -22,7 +22,6 @@ dependencies = [
    "scalar-fastapi>=1.6.1",
    "sqladmin[full]>=0.22.0",
    "sqlalchemy[asyncio]>=2.0.45",
-    "svix-ksuid>=0.6.2",
    "uuid7>=0.1.0",
 ]

--- a/token_plan.md
+++ b/token_plan.md
--- a/uv.lock
+++ b/uv.lock
@ -725,7 +725,6 @@ dependencies = [
    { name = "scalar-fastapi" },
    { name = "sqladmin", extra = ["full"] },
    { name = "sqlalchemy", extra = ["asyncio"] },
-    { name = "svix-ksuid" },
    { name = "uuid7" },
 ]

@ -754,7 +753,6 @@ requires-dist = [
    { name = "scalar-fastapi", specifier = ">=1.6.1" },
    { name = "sqladmin", extras = ["full"], specifier = ">=0.22.0" },
    { name = "sqlalchemy", extras = ["asyncio"], specifier = ">=2.0.45" },
-    { name = "svix-ksuid", specifier = ">=0.6.2" },
    { name = "uuid7", specifier = ">=0.1.0" },
 ]

@ -1021,12 +1019,6 @@ wheels = [
    { url = "https://files.pythonhosted.org/packages/e5/35/f8b19922b6a25bc0880171a2f1a003eaeb93657475193ab516fd87cac9da/pytest_asyncio-1.3.0-py3-none-any.whl", hash = "sha256:611e26147c7f77640e6d0a92a38ed17c3e9848063698d5c93d5aa7aa11cebff5", size = 15075, upload-time = "2025-11-10T16:07:45.537Z" },
 ]

-[[package]]
-name = "python-baseconv"
-version = "1.2.2"
-source = { registry = "https://pypi.org/simple" }
-sdist = { url = "https://files.pythonhosted.org/packages/33/d0/9297d7d8dd74767b4d5560d834b30b2fff17d39987c23ed8656f476e0d9b/python-baseconv-1.2.2.tar.gz", hash = "sha256:0539f8bd0464013b05ad62e0a1673f0ac9086c76b43ebf9f833053527cd9931b", size = 4929, upload-time = "2019-04-04T19:28:57.17Z" }
-
 [[package]]
 name = "python-dotenv"
 version = "1.2.1"
@ -1322,15 +1314,6 @@ wheels = [
    { url = "https://files.pythonhosted.org/packages/d9/52/1064f510b141bd54025f9b55105e26d1fa970b9be67ad766380a3c9b74b0/starlette-0.50.0-py3-none-any.whl", hash = "sha256:9e5391843ec9b6e472eed1365a78c8098cfceb7a74bfd4d6b1c0c0095efb3bca", size = 74033, upload-time = "2025-11-01T15:25:25.461Z" },
 ]

-[[package]]
-name = "svix-ksuid"
-version = "0.6.2"
-source = { registry = "https://pypi.org/simple" }
-dependencies = [
-    { name = "python-baseconv" },
-]
-sdist = { url = "https://files.pythonhosted.org/packages/a1/7a/0c98b77ca01d64f13143607b88273a13110d659780b93cb1333abebf8039/svix-ksuid-0.6.2.tar.gz", hash = "sha256:beb95bd6284bdbd526834e233846653d2bd26eb162b3233513d8f2c853c78964", size = 6957, upload-time = "2023-07-07T09:18:24.717Z" }
-
 [[package]]
 name = "tqdm"
 version = "4.67.1"