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

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

router = APIRouter(prefix="/locks", tags=["Lock Management"])

# ==========================================
# 💡 全域鎖定表 (In-Memory Lock Table)
# 🌟 結構升級: { "host@@path": {"user_id": "...", "username": "...", "expires_at": ...} }
# ⚠️ 嚴格規則：Lock Key 絕對不包含 config_type，確保 running 與 full 視圖共用同一把鎖！
# ==========================================
ACTIVE_LOCKS: Dict[str, dict] = {}
LOCK_TIMEOUT = 120  

# 🌟 請求模型嚴格排除 config_type
class LockAcquireReq(BaseModel):
    host: str
    path: str
    user_id: str      
    username: str     

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

def clean_expired_locks():
    current_time = time.time()
    expired_keys = [k for k, lock in ACTIVE_LOCKS.items() if lock["expires_at"] < current_time]
    for k in expired_keys:
        del ACTIVE_LOCKS[k]

def is_path_locked_by_others(target_host: str, target_path: str, user_id: str) -> tuple[Optional[dict], str]:
    clean_expired_locks()
    prefix = f"{target_host}@@"
    
    for locked_key, lock_info in ACTIVE_LOCKS.items():
        # 🌟 只檢查同一台設備 (IP) 的鎖定
        if not locked_key.startswith(prefix):
            continue
            
        locked_path = locked_key.split("@@", 1)[1]
        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, ""

@router.post("/acquire")
async def acquire_lock(req: LockAcquireReq):
    conflict_lock, error_msg = is_path_locked_by_others(req.host, req.path, req.user_id)
    
    if conflict_lock:
        raise HTTPException(status_code=409, detail=error_msg)
        
    # ⚠️ 嚴格綁定 host 與 path，跨視圖共用
    lock_key = f"{req.host}@@{req.path}"
    ACTIVE_LOCKS[lock_key] = {
        "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()
    lock_key = f"{req.host}@@{req.path}"
    
    if lock_key not in ACTIVE_LOCKS:
        raise HTTPException(status_code=404, detail="鎖定已失效，請重新獲取")
        
    if ACTIVE_LOCKS[lock_key]["user_id"] != req.user_id:
        raise HTTPException(status_code=403, detail="無權更新他人的鎖定")
        
    ACTIVE_LOCKS[lock_key]["expires_at"] = time.time() + LOCK_TIMEOUT
    return {"status": "success", "message": "心跳更新成功"}

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

@router.get("/status")
async def get_lock_status(host: str = Query(None)):
    """獲取特定設備的鎖定狀態"""
    clean_expired_locks()
    if host:
        prefix = f"{host}@@"
        # 🌟 貼心設計：拔除 host@@ 前綴，讓前端收到的依然是乾淨的 path，UI 不用改！
        filtered_locks = {k.split("@@", 1)[1]: v for k, v in ACTIVE_LOCKS.items() if k.startswith(prefix)}
        return {"status": "success", "data": filtered_locks}
        
    return {"status": "success", "data": {}}