DOCS/plans/250831_unified_id_system_implementation_roadmap.md

Unified ID System Implementation Roadmap

작성일: 2025-08-31
작성자: 시스템 설계팀
상태: 🟡 구현 대기
목표: 모든 서비스에서 UUID를 Primary Key로 사용하는 통합 ID 체계 구현

---

1. 현재 상황 분석

1.1 문제점

• rb8001: UUID 매핑 API 제공하지만 내부적으로 혼용
• skill-email: slack_id 직접 사용 (UUID 미지원)
• Gateway: JWT에서 UUID 추출 후 전달
• ChromaDB: 일관성 없는 collection 명명 체계 → 서비스별 prefix 방식으로 통일 필요 ({service}_{uuid})

1.2 영향 범위

• 서비스 간 통신 오류 발생
• 데이터 불일치로 인한 기능 장애
• ChromaDB 컬렉션 분리로 검색 실패

---

2. 구현 단계별 로드맵

Phase 1: 데이터베이스 준비 (Day 1-2)

목표: UUID 기반 스키마 확립 및 기존 데이터 마이그레이션

1.1 테이블 스키마 업데이트

-- users 테이블 (이미 UUID 있음, 제약조건 추가)
ALTER TABLE user 
ADD CONSTRAINT users_uuid_unique UNIQUE(id);

-- slack_user_mapping 인덱스 추가
CREATE INDEX idx_slack_user_mapping_user_id ON slack_user_mapping(user_id);
CREATE INDEX idx_slack_user_mapping_slack_user ON slack_user_mapping(slack_user_id, team_id);

-- gmail_passports UUID 참조 추가
ALTER TABLE gmail_passports 
ADD COLUMN user_uuid UUID REFERENCES user(id);

-- 기존 slack_user_id 기반 데이터 마이그레이션
UPDATE gmail_passports gp
SET user_uuid = (
    SELECT sum.user_id 
    FROM slack_user_mapping sum 
    WHERE sum.slack_user_id = gp.slack_user_id 
    AND sum.team_id = gp.team_id
);

-- workspace_member UUID 확인
ALTER TABLE workspace_member 
ADD CONSTRAINT fk_workspace_member_user_id 
FOREIGN KEY (user_id) REFERENCES user(id);

1.2 데이터 검증 스크립트

# /home/heejae/scripts/verify_uuid_migration.py
import psycopg2
from uuid import UUID

def verify_uuid_consistency():
    conn = psycopg2.connect(...)
    cur = conn.cursor()
    
    # 1. 모든 slack_user_mapping이 유효한 UUID 가리키는지 확인
    cur.execute("""
        SELECT sum.*, u.id 
        FROM slack_user_mapping sum
        LEFT JOIN user u ON sum.user_id = u.id
        WHERE u.id IS NULL
    """)
    orphaned = cur.fetchall()
    if orphaned:
        print(f"⚠️ Found {len(orphaned)} orphaned mappings")
    
    # 2. gmail_passports의 UUID 참조 확인
    cur.execute("""
        SELECT gp.id, gp.slack_user_id, gp.user_uuid
        FROM gmail_passports gp
        WHERE gp.user_uuid IS NULL AND gp.slack_user_id IS NOT NULL
    """)
    unmigrated = cur.fetchall()
    if unmigrated:
        print(f"⚠️ Found {len(unmigrated)} unmigrated gmail passports")
    
    return len(orphaned) == 0 and len(unmigrated) == 0

---

Phase 2: skill-email 서비스 수정 (Day 3-4)

목표: UUID를 primary identifier로 받아들이도록 수정

2.1 CredentialsProvider 수정

# /home/heejae/skill-email/services/db_credentials_provider.py

class DatabaseCredentialsProvider(CredentialsProvider):
    async def get_credentials(self, user_identifier: str) -> Optional[Credentials]:
        """
        user_identifier: UUID 또는 slack_user_id
        UUID 우선, 없으면 slack_user_id로 폴백
        """
        try:
            # UUID 형식 검증
            uuid_obj = UUID(user_identifier)
            is_uuid = True
        except ValueError:
            is_uuid = False
        
        if is_uuid:
            # UUID로 직접 조회
            query = """
                SELECT id, encrypted_credentials, slack_user_id, team_id
                FROM gmail_passports
                WHERE user_uuid = %s AND is_equipped = true
            """
            params = (user_identifier,)
        else:
            # 레거시: slack_user_id로 조회 (deprecated)
            logger.warning(f"Using deprecated slack_user_id lookup: {user_identifier}")
            query = """
                SELECT id, encrypted_credentials, slack_user_id, team_id
                FROM gmail_passports
                WHERE slack_user_id = %s AND is_equipped = true
            """
            params = (user_identifier,)

2.2 API 엔드포인트 수정

# /home/heejae/skill-email/main.py

@app.post("/email/list")
async def list_emails(request: EmailRequest):
    # user_id를 UUID로 받음
    user_uuid = request.user_id  # 이제 UUID
    
    # ChromaDB collection 명명 규칙 통일 (서비스별 prefix)
    collection_name = f"skill_email_{user_uuid}"  # 서비스_UUID 형식
    
    credentials = await credentials_provider.get_credentials(user_uuid)
    # ...

---

Phase 3: rb8001 서비스 일관성 확보 (Day 5-6)

목표: 내부적으로 UUID만 사용, Slack ID는 진입점에서만 변환

3.1 메시지 라우터 수정

# /home/heejae/rb8001/services/router.py

async def route_message(self, message: str, user_id: str, channel: str, thread_ts: str = None):
    # user_id가 Slack ID인 경우 UUID로 변환
    if channel.startswith(('D', 'C', 'G')):  # Slack 채널
        user_uuid = await self.get_uuid_from_slack_id(user_id)
        if not user_uuid:
            logger.error(f"No UUID mapping for slack_id: {user_id}")
            return {"error": "User not found"}
    else:  # Frontend 또는 기타
        user_uuid = user_id  # 이미 UUID
    
    # 이후 모든 처리는 user_uuid 사용
    await self.save_conversation_log(
        user_id=user_uuid,  # UUID 저장
        channel=channel,
        message=message,
        thread_ts=thread_ts
    )

3.2 Slack 이벤트 핸들러 수정

# /home/heejae/rb8001/handlers/slack_events.py

@app.post("/slack/events")
async def handle_slack_event(request: Request):
    event = await request.json()
    
    if event.get("type") == "url_verification":
        return {"challenge": event.get("challenge")}
    
    if event.get("type") == "event_callback":
        slack_event = event.get("event", {})
        slack_user_id = slack_event.get("user")
        team_id = slack_event.get("team")
        
        # Slack ID → UUID 변환
        user_uuid = await get_uuid_from_slack_mapping(slack_user_id, team_id)
        
        # 내부 처리는 모두 UUID 사용
        await process_event_with_uuid(slack_event, user_uuid)

---

Phase 4: ChromaDB Collection 정리 (Day 7)

목표: 서비스별 prefix + UUID 기반 일관된 명명 체계 적용

4.1 Collection 마이그레이션 스크립트

# /home/heejae/scripts/migrate_chromadb_collections.py

import chromadb
from chromadb.config import Settings

client = chromadb.Client(Settings(
    chroma_db_impl="duckdb+parquet",
    persist_directory="/path/to/chromadb"
))

async def migrate_collections():
    # 1. 기존 컬렉션 목록 조회
    collections = client.list_collections()
    
    for collection in collections:
        old_name = collection.name
        
        # slack_id 기반 이름 → UUID 기반으로 변환
        if old_name.startswith("U") and "_" in old_name:
            slack_id = old_name.split("_")[0]
            
            # DB에서 UUID 조회
            user_uuid = await get_uuid_from_slack_id(slack_id)
            
            if user_uuid:
                # 서비스별 prefix 방식 (예: rb8001_{uuid}, skill_email_{uuid})
                service_name = old_name.split("_")[0] if "_" in old_name else "rb8001"
                new_name = f"{service_name}_{user_uuid}"
                
                # 컬렉션 복사
                old_collection = client.get_collection(old_name)
                new_collection = client.create_collection(new_name)
                
                # 데이터 마이그레이션
                data = old_collection.get()
                if data['ids']:
                    new_collection.add(
                        ids=data['ids'],
                        documents=data['documents'],
                        metadatas=data['metadatas']
                    )
                
                # 기존 컬렉션 삭제 (백업 후)
                client.delete_collection(old_name)
                print(f"✅ Migrated: {old_name} → {new_name}")

---

Phase 5: Gateway 검증 강화 (Day 8)

목표: JWT에서 UUID 추출 및 검증 강화

5.1 JWT 검증 미들웨어 개선

# /home/heejae/robeing-gateway/app/auth.py

async def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = jwt.decode(token, JWT_SECRET_KEY, algorithms=[ALGORITHM])
        user_id = payload.get("sub")  # UUID
        
        if not user_id:
            raise HTTPException(status_code=401, detail="Invalid token")
        
        # UUID 형식 검증
        try:
            UUID(user_id)
        except ValueError:
            logger.error(f"Invalid UUID in token: {user_id}")
            raise HTTPException(status_code=401, detail="Invalid user ID format")
        
        return {"user_id": user_id, "uuid": user_id}  # 명확히 UUID임을 표시
    except JWTError:
        raise HTTPException(status_code=401, detail="Could not validate credentials")

---

Phase 6: 통합 테스트 (Day 9-10)

목표: 전체 시스템 통합 테스트 및 검증

6.1 End-to-End 테스트 시나리오

# /home/heejae/tests/test_unified_id_system.py

async def test_full_flow():
    # 1. Slack 로그인으로 UUID 생성
    slack_user_id = "U123456"
    team_id = "T789012"
    
    # 2. UUID 매핑 확인
    user_uuid = await create_or_get_uuid_mapping(slack_user_id, team_id)
    assert user_uuid is not None
    
    # 3. Gmail 패스포트 연결 (UUID 사용)
    gmail_passport = await connect_gmail_passport(user_uuid)
    assert gmail_passport.user_uuid == user_uuid
    
    # 4. skill-email 호출 (UUID 전달)
    emails = await fetch_emails(user_uuid)
    assert emails is not None
    
    # 5. ChromaDB 컬렉션 확인 (서비스별 prefix)
    collection_name = f"skill_email_{user_uuid}"
    collection = chromadb_client.get_collection(collection_name)
    assert collection is not None
    
    # 6. rb8001 메시지 처리
    response = await send_message_to_rb8001(
        user_id=user_uuid,
        message="Get my emails",
        channel="frontend"
    )
    assert response.status_code == 200

6.2 회귀 테스트

• 기존 Slack ID 기반 요청이 여전히 작동하는지 확인 (하위 호환성)
• UUID 기반 새 요청이 모든 서비스에서 작동하는지 확인
• 데이터 일관성 검증

---

3. 롤백 계획

3.1 데이터베이스 롤백

-- gmail_passports UUID 컬럼 제거
ALTER TABLE gmail_passports DROP COLUMN user_uuid;

-- 인덱스 제거
DROP INDEX idx_slack_user_mapping_user_id;
DROP INDEX idx_slack_user_mapping_slack_user;

3.2 서비스 롤백

• Docker 이미지 태그를 이전 버전으로 변경
• 환경변수 USE_UUID_SYSTEM=false 설정으로 레거시 모드 활성화

---

4. 모니터링 및 알림

4.1 핵심 메트릭

• UUID 변환 실패율
• 서비스 간 통신 오류율
• ChromaDB 조회 성공률

4.2 알림 설정

# 변환 실패 시 알림
if not uuid_mapping:
    logger.error(f"UUID mapping failed for slack_id: {slack_id}")
    send_alert("UUID_MAPPING_FAILURE", {
        "slack_id": slack_id,
        "service": "rb8001",
        "timestamp": datetime.now()
    })

---

5. 타임라인

단계	기간	담당	상태
Phase 1: DB 준비	Day 1-2	DBA팀	🔴 대기
Phase 2: skill-email	Day 3-4	백엔드팀	🔴 대기
Phase 3: rb8001	Day 5-6	백엔드팀	🔴 대기
Phase 4: ChromaDB	Day 7	데이터팀	🔴 대기
Phase 5: Gateway	Day 8	백엔드팀	🔴 대기
Phase 6: 테스트	Day 9-10	QA팀	🔴 대기

---

6. 위험 요소 및 대응

6.1 높은 위험

• 데이터 손실: 마이그레이션 전 전체 백업 필수
• 서비스 중단: 카나리 배포로 점진적 롤아웃

6.2 중간 위험

• 성능 저하: UUID 변환 캐싱으로 최소화
• 하위 호환성: 레거시 모드 6개월 유지

---

7. 성공 기준

• ✅ 모든 서비스가 UUID를 primary key로 사용
• ✅ Slack ID는 진입점에서만 UUID로 변환
• ✅ ChromaDB 컬렉션명 통일 ({service}_{uuid} 형식)
• ✅ 서비스 간 통신 오류 0%
• ✅ 기존 기능 100% 하위 호환성 유지

---

8. 참고 문서

• 250828_slack_auth_integration_completed.md
• 250828_slack_integration_level3_plan.md
• 250828_conversation_log_channel_구분_개선.md