scm-harmonic-cmts-admin/routers/lock.py

# --- routers/lock.py ---
import time
from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
from typing import Dict, Optional

router = APIRouter()

# ==========================================
# 💡 全域鎖定表 (In-Memory Lock Table)
# 結構: { "path": {"user_id": "...", "username": "...", "expires_at": 1234567890.123} }
# ==========================================
ACTIVE_LOCKS: Dict[str, dict] = {}

# 鎖定超時時間 (秒)：前端需在此時間內發送 Heartbeat，否則自動釋放
LOCK_TIMEOUT = 30  

class LockAcquireReq(BaseModel):
    path: str
    user_id: str      # 前端產生的 UUID (Session ID)
    username: str     # 登入的帳號 (用於 UI 提示，例如 "admin")

class LockActionReq(BaseModel):
    path: str
    user_id: str

def clean_expired_locks():
    """清除已經超時的鎖 (被動式清理)"""
    current_time = time.time()
    expired_paths = [path for path, lock in ACTIVE_LOCKS.items() if lock["expires_at"] < current_time]
    for path in expired_paths:
        del ACTIVE_LOCKS[path]

def is_path_locked_by_others(target_path: str, user_id: str) -> tuple[Optional[dict], str]:
    """
    檢查路徑是否被其他人鎖定，並回傳 (衝突的鎖定資訊, 具體錯誤訊息)
    """
    clean_expired_locks()
    
    for locked_path, lock_info in ACTIVE_LOCKS.items():
        if lock_info["user_id"] == user_id:
            continue # 自己鎖定的不算衝突
            
        # 💡 拆分判斷，回傳更精準的錯誤訊息
        if target_path == locked_path:
            return lock_info, f"此區塊正由 [{lock_info['username']}] 編輯中"
            
        if target_path.startswith(locked_path + "::"):
            return lock_info, f"父層級 [{locked_path}] 已被 [{lock_info['username']}] 鎖定，無法編輯子區塊"
            
        if locked_path.startswith(target_path + "::"):
            return lock_info, f"子層級 [{locked_path}] 已被 [{lock_info['username']}] 鎖定，無法鎖定整個父區塊"
            
    return None, ""

# ==========================================
# 💡 API Endpoints
# ==========================================

@router.post("/acquire")
async def acquire_lock(req: LockAcquireReq):
    """請求鎖定特定路徑"""
    # 接收回傳的鎖定資訊與具體訊息
    conflict_lock, error_msg = is_path_locked_by_others(req.path, req.user_id)
    
    if conflict_lock:
        raise HTTPException(
            status_code=409, 
            detail=error_msg  # 💡 將精準的錯誤訊息傳給前端
        )
        
    # 寫入或更新鎖定狀態
    ACTIVE_LOCKS[req.path] = {
        "user_id": req.user_id,
        "username": req.username,
        "expires_at": time.time() + LOCK_TIMEOUT
    }
    return {"status": "success", "message": f"成功鎖定路徑: {req.path}"}

@router.post("/heartbeat")
async def heartbeat_lock(req: LockActionReq):
    """延長鎖定時間 (心跳偵測)"""
    clean_expired_locks()
    
    if req.path not in ACTIVE_LOCKS:
        raise HTTPException(status_code=404, detail="鎖定已失效，請重新獲取")
        
    if ACTIVE_LOCKS[req.path]["user_id"] != req.user_id:
        raise HTTPException(status_code=403, detail="無權更新他人的鎖定")
        
    # 延長鎖定時間
    ACTIVE_LOCKS[req.path]["expires_at"] = time.time() + LOCK_TIMEOUT
    return {"status": "success", "message": "心跳更新成功"}

@router.post("/release")
async def release_lock(req: LockActionReq):
    """主動釋放鎖定"""
    if req.path in ACTIVE_LOCKS and ACTIVE_LOCKS[req.path]["user_id"] == req.user_id:
        del ACTIVE_LOCKS[req.path]
    return {"status": "success", "message": "鎖定已釋放"}

@router.get("/status")
async def get_lock_status():
    """獲取目前所有被鎖定的路徑 (供前端 UI 標示 '編輯中' 狀態)"""
    clean_expired_locks()
    return {"status": "success", "data": ACTIVE_LOCKS}