feat: add migrate api

app/admin/router.py

@@ -1,13 +1,26 @@
 import subprocess
 import sys
 
-from fastapi import Depends, HTTPException
+from fastapi import Depends, HTTPException, Query
 from fastapi.security import APIKeyHeader
+from sqlalchemy import func, select, text
+from ulid import ULID
 
 from app.core.router import create_router
+from app.database.deps import SessionDep
+from app.posts.models import Post
+from app.schemas import APISchema
 
 from .config import admin_settings
 
+
+class FillPublicIdsResponse(APISchema):
+    success: bool
+    message: str
+    processed: int
+    remaining: int
+
+
 api_key_header = APIKeyHeader(name="X-Admin-Token", scheme_name="Admin Auth")
 
 
@@ -25,3 +38,65 @@ async def migrate_db_schema():
     subprocess.run([sys.executable, "-m", "alembic", "upgrade", "head"], check=True)
 
     return {"success": True}
+
+
+@router.post("/fill-public-ids")
+async def fill_public_ids(
+    session: SessionDep,
+    batch_size: int = Query(default=100, ge=1, le=1000),
+) -> FillPublicIdsResponse:
+    """
+    public_id가 NULL인 Post에 ULID를 채웁니다.
+    created_at 기반으로 ULID를 생성합니다.
+
+    - batch_size: 한 번에 처리할 개수 (기본값: 100, 최대: 1000)
+    - 여러 번 호출해서 점진적으로 처리할 수 있습니다.
+    """
+    # NULL인 Post 개수 확인
+    total_remaining = await session.scalar(
+        select(func.count()).select_from(Post).where(Post.public_id.is_(None))
+    )
+
+    if not total_remaining:
+        return FillPublicIdsResponse(
+            success=True,
+            message="모든 Post에 public_id가 이미 있습니다.",
+            processed=0,
+            remaining=0,
+        )
+
+    # 배치 처리: SELECT로 id, created_at 조회
+    posts = await session.execute(
+        select(Post.id, Post.created_at)
+        .where(Post.public_id.is_(None))
+        .order_by(Post.id)
+        .limit(batch_size)
+    )
+    posts_list = list(posts)
+
+    if not posts_list:
+        return FillPublicIdsResponse(
+            success=True,
+            message="모든 Post에 public_id가 이미 있습니다.",
+            processed=0,
+            remaining=0,
+        )
+
+    # Raw SQL로 업데이트 (onupdate 트리거 우회)
+    for post in posts_list:
+        ulid_value = str(ULID.from_datetime(post.created_at))
+        await session.execute(
+            text("UPDATE post SET public_id = :ulid WHERE id = :id"),
+            {"ulid": ulid_value, "id": post.id},
+        )
+    await session.commit()
+
+    processed = len(posts_list)
+    remaining = total_remaining - processed
+
+    return FillPublicIdsResponse(
+        success=True,
+        message=f"{processed}개 처리 완료",
+        processed=processed,
+        remaining=remaining,
+    )