DOCS/plans/completed/250812_로컬개발자_Slack사용자매핑_구현계획.md

Slack 사용자 매핑 구현 계획 (로컬 개발자)

작성일: 2025년 8월 12일
작성자: 로컬 개발자 (Claude)
관련 문서: 250812_claude_Slack_메시지_처리_아키텍처_분석.md

1. 구현 목표

24서버팀 분석을 기반으로 로컬 개발자가 구현할 수 있는 Slack 사용자 매핑 시스템을 설계합니다.

1.1 핵심 문제 해결

• Slack user_id (U0925SXQFDK)와 시스템 user_id 연결
• 사용자별 ChromaDB 컬렉션 분리
• 멀티 워크스페이스 지원

2. 구현 가능한 코드 변경사항

2.1 rb10508_micro 임시 매핑 구현

파일: rb10508_micro/app/services/slack_service.py

# 임시 매핑 (DB 연동 전까지 사용)
TEMP_USER_MAPPING = {
    "U0925SXQFDK": {
        "username": "happybell80",
        "user_id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
    },
    "U04KJHGLS": {
        "username": "eagle0914", 
        "user_id": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb"
    },
    "UHHYONG91": {
        "username": "hhyong91",
        "user_id": "cccccccc-cccc-cccc-cccc-cccccccccccc"
    }
}

async def get_system_user_id(slack_user_id: str) -> str:
    """Slack user_id를 시스템 user_id로 변환"""
    mapping = TEMP_USER_MAPPING.get(slack_user_id)
    if mapping:
        return mapping["user_id"]
    return "default_user"

2.2 ChromaDB 컬렉션 분리

파일: rb10508_micro/app/core/memory/storage.py

def get_user_collection_name(robeing_id: str, user_id: str, memory_type: str) -> str:
    """사용자별 컬렉션 이름 생성"""
    if user_id == "default_user":
        # 매핑 없는 사용자는 공용 컬렉션 사용
        return f"{robeing_id}_{memory_type}"
    else:
        # 사용자별 전용 컬렉션
        return f"{robeing_id}_{user_id[:8]}_{memory_type}"

2.3 Brain 모듈 수정

파일: rb10508_micro/app/core/brain.py

async def think_functional(
    input_data: Dict,
    search_memories_fn,
    store_memory_fn,
    get_identity_fn,
    store_identity_fn,
    get_system_user_id_fn  # 새로 추가
) -> Dict:
    """개선된 think 함수"""
    
    # Slack user_id 변환
    slack_user_id = input_data.get("user_id")
    system_user_id = await get_system_user_id_fn(slack_user_id)
    
    # 사용자별 컬렉션 사용
    collection_name = get_user_collection_name(
        settings.ROBEING_ID,
        system_user_id,
        "episodic"
    )
    
    # 이후 로직은 동일...

3. Auth 서버 API 엔드포인트 설계

3.1 Slack 매핑 조회 API

파일: auth-server/app/api/slack.py

from fastapi import APIRouter, HTTPException
from typing import Optional

router = APIRouter(prefix="/api/slack")

@router.get("/mapping/{slack_user_id}")
async def get_user_mapping(
    slack_user_id: str,
    workspace_id: Optional[str] = None
):
    """Slack user_id로 시스템 사용자 정보 조회"""
    
    query = """
        SELECT 
            sum.user_id,
            u.username,
            u.email,
            wm.robeing_id
        FROM slack_user_mapping sum
        JOIN user u ON sum.user_id = u.id
        LEFT JOIN workspace_member wm ON u.id = wm.user_id
        WHERE sum.slack_user_id = :slack_user_id
    """
    
    result = await database.fetch_one(
        query, 
        {"slack_user_id": slack_user_id}
    )
    
    if not result:
        raise HTTPException(404, "Mapping not found")
    
    return {
        "user_id": str(result["user_id"]),
        "username": result["username"],
        "email": result["email"],
        "robeing_id": result["robeing_id"] or "rb10508_micro"
    }

@router.post("/mapping")
async def create_user_mapping(
    slack_user_id: str,
    slack_workspace_id: int,
    user_id: str
):
    """새로운 Slack 사용자 매핑 생성"""
    
    query = """
        INSERT INTO slack_user_mapping 
        (slack_user_id, slack_workspace_id, user_id)
        VALUES (:slack_user_id, :slack_workspace_id, :user_id)
        ON CONFLICT (slack_user_id, slack_workspace_id) 
        DO UPDATE SET user_id = :user_id
        RETURNING id
    """
    
    result = await database.fetch_one(
        query,
        {
            "slack_user_id": slack_user_id,
            "slack_workspace_id": slack_workspace_id,
            "user_id": user_id
        }
    )
    
    return {"mapping_id": str(result["id"])}

4. Gateway 라우팅 개선

4.1 사용자별 로빙 라우팅

파일: robeing-gateway/app/main.py

@app.post("/api/slack/route")
async def route_slack_message(
    request: Request,
    background_tasks: BackgroundTasks
):
    """Slack 메시지를 적절한 로빙으로 라우팅"""
    
    body = await request.json()
    slack_user_id = body.get("event", {}).get("user")
    
    # Auth 서버에서 매핑 조회
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"http://localhost:8001/api/slack/mapping/{slack_user_id}"
        )
        
        if resp.status_code == 200:
            mapping = resp.json()
            robeing_id = mapping["robeing_id"]
            user_id = mapping["user_id"]
        else:
            # 매핑 없으면 기본 로빙 사용
            robeing_id = "rb10508_micro"
            user_id = "default_user"
    
    # 해당 로빙으로 전달
    robeing_urls = {
        "rb8001": "http://192.168.219.52:8001",
        "rb10508_micro": "http://192.168.219.52:10508",
        "rb10408": "http://192.168.219.52:10408"
    }
    
    target_url = robeing_urls.get(robeing_id)
    if not target_url:
        return {"error": "Unknown robeing"}
    
    # X-System-User-Id 헤더 추가
    headers = {"X-System-User-Id": user_id}
    
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{target_url}/api/slack/events",
            json=body,
            headers=headers
        )
    
    return {"ok": True}

5. SQL 스키마

5.1 slack_user_mapping 테이블

-- Auth DB에 추가할 테이블
CREATE TABLE IF NOT EXISTS slack_user_mapping (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    slack_user_id VARCHAR(100) NOT NULL,
    slack_workspace_id INTEGER REFERENCES slack_workspace(id),
    user_id UUID REFERENCES user(id) NOT NULL,
    workspace_member_id UUID REFERENCES workspace_member(id),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE(slack_user_id, slack_workspace_id)
);

-- 인덱스 추가
CREATE INDEX idx_slack_user_mapping_slack_user 
ON slack_user_mapping(slack_user_id);

CREATE INDEX idx_slack_user_mapping_user 
ON slack_user_mapping(user_id);

-- 초기 데이터
INSERT INTO slack_user_mapping (slack_user_id, slack_workspace_id, user_id)
SELECT 
    'U0925SXQFDK',
    sw.id,
    u.id
FROM user u, slack_workspaces sw
WHERE u.username = 'happybell80' 
AND sw.team_name = 'GoodGang Labs';

6. 테스트 계획

6.1 로컬 테스트

# test_slack_mapping.py
import asyncio
from app.services.slack_service import get_system_user_id

async def test_mapping():
    # 매핑 있는 사용자
    user_id = await get_system_user_id("U0925SXQFDK")
    assert user_id == "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
    
    # 매핑 없는 사용자
    user_id = await get_system_user_id("UNKNOWN")
    assert user_id == "default_user"
    
    print("✅ 모든 테스트 통과")

asyncio.run(test_mapping())

6.2 ChromaDB 컬렉션 확인

# 서버에서 실행
from app.core.memory.storage import get_chroma_client

client = get_chroma_client()
collections = client.list_collections()

for col in collections:
    print(f"{col.name}: {col.count()}개")
    
# 예상 결과:
# rb10508_micro_aaaaaaaa_episodic: N개
# rb10508_micro_bbbbbbbb_episodic: M개
# rb10508_micro_default_user_episodic: K개

7. 배포 순서

Phase 1: 임시 하드코딩 (즉시)

1. rb10508_micro에 TEMP_USER_MAPPING 추가
2. ChromaDB 컬렉션 분리 로직 구현
3. 테스트 후 배포

Phase 2: DB 연동 (1주 내)

1. slack_user_mapping 테이블 생성 (23서버팀 요청)
2. Auth 서버 API 구현
3. rb10508_micro DB 조회 로직 추가

Phase 3: Gateway 통합 (2주 내)

1. Gateway 라우팅 로직 구현
2. 모든 로빙 통합 테스트
3. OAuth 자동 매핑 구현

8. 주의사항

8.1 호환성 유지

• 기존 데이터 마이그레이션 계획 필요
• 매핑 없는 사용자도 서비스 가능하도록

8.2 보안

• Slack 토큰 안전한 관리
• user_id 노출 방지

8.3 성능

• DB 조회 캐싱 고려
• 컬렉션 수 증가에 따른 관리 방안

9. 24서버팀에게 요청사항

1. DB 테이블 생성

   • slack_user_mapping 테이블 생성
   • 초기 데이터 입력

2. 서비스 재배포

   • rb10508_micro 코드 변경 후 재배포
   • Auth 서버 API 추가 후 재배포

3. 테스트 지원

   • Slack 메시지 전송 테스트
   • ChromaDB 데이터 확인

10. 예상 효과

• ✅ 사용자별 개인화된 메모리 관리
• ✅ 멀티 워크스페이스 지원
• ✅ 확장 가능한 아키텍처
• ✅ OAuth 연동 준비

---

이 문서는 24서버팀의 분석을 기반으로 로컬 개발자가 작성한 구현 계획입니다.