feat(call): integrate LiveKit for voice and video calling

Replace the manual WebRTC signaling implementation with LiveKit SFU. This includes: - Adding LiveKit service, handler, and configuration. - Updating Docker Compose to include LiveKit server, Redis, and PostgreSQL. - Refactoring `CallService` and `WSHandler` to support LiveKit room readiness instead of raw SDP/ICE relaying. - Adding new API endpoints for LiveKit token generation and webhooks. - Removing deprecated WebRTC configuration and manual signaling DTOs.
2026-06-01 13:41:02 +08:00
parent 9356cb6876
commit 14114db68d
22 changed files with 731 additions and 452 deletions
--- a/internal/config/config.go
+++ b/internal/config/config.go
@@ -27,7 +27,7 @@ type Config struct {
 	Casbin            CasbinConfig            `mapstructure:"casbin"`
 	Encryption        EncryptionConfig        `mapstructure:"encryption"`
 	HotRank           HotRankConfig           `mapstructure:"hot_rank"`
-	WebRTC            WebRTCConfig            `mapstructure:"webrtc"`
+	LiveKit           LiveKitConfig           `mapstructure:"livekit"`
 	Report            ReportConfig            `mapstructure:"report"`
 	Sensitive         SensitiveConfig         `mapstructure:"sensitive"`
 	JPush             JPushConfig             `mapstructure:"jpush"`
@@ -168,10 +168,6 @@ func Load(configPath string) (*Config, error) {
 	viper.SetDefault("hot_rank.recent_window_hours", 168)
 	viper.SetDefault("hot_rank.recent_fetch_limit", 500)
 	viper.SetDefault("hot_rank.candidate_cap", 800)
-	// WebRTC 默认值（Google 公共 STUN 服务器）
-	viper.SetDefault("webrtc.ice_servers", []map[string]any{
-		{"urls": []string{"stun:stun.l.google.com:19302"}},
-	})
 	// Report 默认值
 	viper.SetDefault("report.auto_hide_threshold", 3)
 	viper.SetDefault("jpush.enabled", false)
@@ -213,6 +209,13 @@ func Load(configPath string) (*Config, error) {
 	viper.SetDefault("push_worker.max_retries", 3)
 	viper.SetDefault("push_worker.max_stream_len", 100000)
 	viper.SetDefault("push_worker.idle_timeout_ms", 30000)
+	// LiveKit 默认值
+	viper.SetDefault("livekit.enabled", false)
+	viper.SetDefault("livekit.url", "http://localhost:7880")
+	viper.SetDefault("livekit.api_key", "devkey")
+	viper.SetDefault("livekit.api_secret", "")
+	viper.SetDefault("livekit.token_ttl", 600)
+	viper.SetDefault("livekit.webhook_secret", "")

 	if err := viper.ReadInConfig(); err != nil {
 		return nil, fmt.Errorf("failed to read config: %w", err)
--- a/internal/config/livekit.go
+++ b/internal/config/livekit.go
@@ -0,0 +1,11 @@
+package config
+
+// LiveKitConfig LiveKit SFU 配置
+type LiveKitConfig struct {
+	Enabled       bool   `mapstructure:"enabled"`
+	URL           string `mapstructure:"url"`
+	APIKey        string `mapstructure:"api_key"`
+	APISecret     string `mapstructure:"api_secret"`
+	TokenTTL      int    `mapstructure:"token_ttl"`
+	WebhookSecret string `mapstructure:"webhook_secret"`
+}
--- a/internal/config/webrtc.go
+++ b/internal/config/webrtc.go
@@ -1,13 +0,0 @@
-package config
-
-// WebRTCConfig WebRTC ICE 服务器配置
-type WebRTCConfig struct {
-	ICEServers []ICEServerConfig `mapstructure:"ice_servers"`
-}
-
-// ICEServerConfig ICE 服务器配置
-type ICEServerConfig struct {
-	URLs       []string `mapstructure:"urls"`
-	Username   string   `mapstructure:"username"`
-	Credential string   `mapstructure:"credential"`
-}
--- a/internal/dto/call_dto.go
+++ b/internal/dto/call_dto.go
@@ -1,145 +0,0 @@
-package dto
-
-import (
-	"time"
-
-	"with_you/internal/model"
-)
-
-// ==================== Call Request DTOs ====================
-
-// StartCallRequest 发起通话请求
-type StartCallRequest struct {
-	ConversationID string         `json:"conversation_id" binding:"required"` // 会话ID
-	CallType       model.CallType `json:"call_type" binding:"required"`       // 通话类型: private, group
-}
-
-// AnswerCallRequest 接听通话请求
-type AnswerCallRequest struct {
-	SDP string `json:"sdp" binding:"required"` // SDP Answer
-}
-
-// SendSDPRequest 发送SDP请求
-type SendSDPRequest struct {
-	SDP  string `json:"sdp" binding:"required"`  // SDP Offer/Answer
-	Type string `json:"type" binding:"required"` // offer 或 answer
-}
-
-// SendICECandidateRequest 发送ICE候选请求
-type SendICECandidateRequest struct {
-	Candidate     string `json:"candidate" binding:"required"`      // candidate 字符串
-	SDPMid        string `json:"sdp_mid"`                           // SDP mid
-	SDPMLineIndex int16  `json:"sdp_mline_index"`                   // SDP m-line index
-}
-
-// ==================== Call Response DTOs ====================
-
-// CallSessionResponse 通话会话响应
-type CallSessionResponse struct {
-	ID             string                  `json:"id"`              // 通话ID
-	ConversationID string                  `json:"conversation_id"` // 会话ID
-	GroupID        string                  `json:"group_id,omitempty"`
-	CallerID       string                  `json:"caller_id"`       // 发起者ID
-	CallType       model.CallType          `json:"call_type"`       // 通话类型
-	Status         model.CallStatus        `json:"status"`          // 通话状态
-	StartedAt      *time.Time              `json:"started_at,omitempty"`
-	EndedAt        *time.Time              `json:"ended_at,omitempty"`
-	Duration       int64                   `json:"duration"`        // 通话时长(秒)
-	CreatedAt      time.Time               `json:"created_at"`
-	Participants   []CallParticipantResponse `json:"participants"`  // 参与者列表
-	Caller         *UserResponse           `json:"caller,omitempty"` // 发起者信息
-}
-
-// CallParticipantResponse 通话参与者响应
-type CallParticipantResponse struct {
-	UserID   string                 `json:"user_id"`
-	Status   model.ParticipantStatus `json:"status"`
-	JoinedAt *time.Time             `json:"joined_at,omitempty"`
-	LeftAt   *time.Time             `json:"left_at,omitempty"`
-	User     *UserResponse          `json:"user,omitempty"`
-}
-
-// ==================== Call SSE Event DTOs ====================
-
-// CallInviteEvent 来电邀请事件
-type CallInviteEvent struct {
-	CallID       string         `json:"call_id"`
-	ConversationID string       `json:"conversation_id"`
-	GroupID       string        `json:"group_id,omitempty"`
-	CallerID      string        `json:"caller_id"`
-	CallType      model.CallType `json:"call_type"`
-	Caller        *UserResponse `json:"caller,omitempty"`
-	CreatedAt     time.Time     `json:"created_at"`
-}
-
-// CallAnswerEvent 接听响应事件
-type CallAnswerEvent struct {
-	CallID    string `json:"call_id"`
-	UserID    string `json:"user_id"`
-	SDP       string `json:"sdp"`       // SDP Answer
-}
-
-// CallRejectEvent 拒绝通话事件
-type CallRejectEvent struct {
-	CallID string `json:"call_id"`
-	UserID string `json:"user_id"`
-}
-
-// CallEndEvent 结束通话事件
-type CallEndEvent struct {
-	CallID   string `json:"call_id"`
-	UserID   string `json:"user_id"`
-	Duration int64  `json:"duration"`
-}
-
-// CallICECandidateEvent ICE候选事件
-type CallICECandidateEvent struct {
-	CallID        string `json:"call_id"`
-	UserID        string `json:"user_id"`
-	Candidate     string `json:"candidate"`
-	SDPMid        string `json:"sdp_mid"`
-	SDPMLineIndex int16  `json:"sdp_mline_index"`
-}
-
-// CallSDPOfferEvent SDP Offer事件
-type CallSDPOfferEvent struct {
-	CallID string `json:"call_id"`
-	UserID string `json:"user_id"`
-	SDP    string `json:"sdp"`
-}
-
-// CallSDPAnswerEvent SDP Answer事件
-type CallSDPAnswerEvent struct {
-	CallID string `json:"call_id"`
-	UserID string `json:"user_id"`
-	SDP    string `json:"sdp"`
-}
-
-// CallUserJoinedEvent 用户加入通话事件(群聊)
-type CallUserJoinedEvent struct {
-	CallID string        `json:"call_id"`
-	UserID string        `json:"user_id"`
-	User   *UserResponse `json:"user,omitempty"`
-}
-
-// CallUserLeftEvent 用户离开通话事件(群聊)
-type CallUserLeftEvent struct {
-	CallID string `json:"call_id"`
-	UserID string `json:"user_id"`
-}
-
-// ==================== ICE Server Config ====================
-
-// ICEServerConfig ICE服务器配置
-type ICEServerConfig struct {
-	URLs           []string `json:"urls"`
-	Username       string   `json:"username,omitempty"`
-	Credential     string   `json:"credential,omitempty"`
-	CredentialType string   `json:"credential_type,omitempty"`
-}
-
-// CallConfigResponse 通话配置响应
-type CallConfigResponse struct {
-	ICEServers []ICEServerConfig `json:"ice_servers"`
-}
-
--- a/internal/handler/call_handler.go
+++ b/internal/handler/call_handler.go
@@ -49,12 +49,3 @@ func (h *CallHandler) GetCallHistory(c *gin.Context) {
 		"page":  page,
 	})
 }
-
-// GetICEServers 获取 ICE 服务器配置
-// GET /api/v1/calls/ice-servers
-func (h *CallHandler) GetICEServers(c *gin.Context) {
-	servers := h.callService.GetICEServers()
-	response.Success(c, gin.H{
-		"ice_servers": servers,
-	})
-}
--- a/internal/handler/livekit_handler.go
+++ b/internal/handler/livekit_handler.go
@@ -0,0 +1,155 @@
+package handler
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"strconv"
+
+	"github.com/gin-gonic/gin"
+	"github.com/livekit/protocol/livekit"
+	"go.uber.org/zap"
+
+	"with_you/internal/config"
+	"with_you/internal/pkg/response"
+	"with_you/internal/service"
+)
+
+// LiveKitHandler LiveKit 令牌和 Webhook 处理器
+type LiveKitHandler struct {
+	liveKitService service.LiveKitService
+	callService    service.CallService
+	config         *config.LiveKitConfig
+	logger         *zap.Logger
+}
+
+// NewLiveKitHandler 创建 LiveKit 处理器
+func NewLiveKitHandler(
+	liveKitService service.LiveKitService,
+	callService service.CallService,
+	cfg *config.Config,
+	logger *zap.Logger,
+) *LiveKitHandler {
+	return &LiveKitHandler{
+		liveKitService: liveKitService,
+		callService:    callService,
+		config:         &cfg.LiveKit,
+		logger:          logger,
+	}
+}
+
+// GetToken 生成 LiveKit 访问令牌
+// GET /api/v1/calls/token?room=<callID>
+func (h *LiveKitHandler) GetToken(c *gin.Context) {
+	if !h.config.Enabled {
+		response.Forbidden(c, "livekit is not enabled")
+		return
+	}
+
+	userID, exists := c.Get("user_id")
+	if !exists {
+		response.Unauthorized(c, "unauthorized")
+		return
+	}
+
+	room := c.Query("room")
+	if room == "" {
+		response.BadRequest(c, "room parameter is required")
+		return
+	}
+
+	// 验证用户是该通话的参与者
+	activeCall := h.callService.GetActiveCall(c.Request.Context(), room)
+	if activeCall == nil {
+		response.NotFound(c, "call not found or already ended")
+		return
+	}
+
+	isParticipant := false
+	for _, p := range activeCall.Participants {
+		if p.UserID == userID.(string) {
+			isParticipant = true
+			break
+		}
+	}
+	if !isParticipant && activeCall.CallerID != userID.(string) && activeCall.CalleeID != userID.(string) {
+		response.Forbidden(c, "not a participant of this call")
+		return
+	}
+
+	canPublish := true
+	canSubscribe := true
+	if cp := c.Query("can_publish"); cp != "" {
+		canPublish, _ = strconv.ParseBool(cp)
+	}
+	if cs := c.Query("can_subscribe"); cs != "" {
+		canSubscribe, _ = strconv.ParseBool(cs)
+	}
+
+	token, err := h.liveKitService.GenerateToken(room, userID.(string), canPublish, canSubscribe)
+	if err != nil {
+		h.logger.Error("failed to generate livekit token", zap.Error(err))
+		response.InternalServerError(c, "failed to generate token")
+		return
+	}
+
+	response.Success(c, gin.H{
+		"token": token,
+		"url":   h.config.URL,
+	})
+}
+
+// HandleWebhook 处理 LiveKit Webhook 回调
+// POST /api/v1/calls/webhook
+func (h *LiveKitHandler) HandleWebhook(c *gin.Context) {
+	if !h.config.Enabled {
+		c.JSON(http.StatusNotFound, gin.H{"error": "livekit is not enabled"})
+		return
+	}
+
+	body, err := io.ReadAll(c.Request.Body)
+	if err != nil {
+		h.logger.Error("failed to read webhook body", zap.Error(err))
+		c.JSON(http.StatusBadRequest, gin.H{"error": "failed to read body"})
+		return
+	}
+
+	authHeader := c.GetHeader("Authorization")
+	event, err := h.liveKitService.ProcessWebhook(c.Request.Context(), body, authHeader)
+	if err != nil {
+		h.logger.Error("failed to process webhook", zap.Error(err))
+		c.JSON(http.StatusBadRequest, gin.H{"error": "invalid webhook"})
+		return
+	}
+
+	// 根据 webhook 事件类型处理
+	switch event.Event {
+	case "room_finished":
+		h.handleRoomFinished(c.Request.Context(), event)
+	case "participant_left":
+		h.logger.Info("participant left room",
+			zap.String("room", event.Room.GetName()),
+			zap.String("identity", event.Participant.GetIdentity()),
+		)
+	}
+
+	c.JSON(http.StatusOK, gin.H{"status": "ok"})
+}
+
+// handleRoomFinished 处理 room_finished 事件，作为通话历史持久化的安全兜底
+func (h *LiveKitHandler) handleRoomFinished(ctx context.Context, event *livekit.WebhookEvent) {
+	roomName := event.Room.GetName()
+	if roomName == "" {
+		return
+	}
+
+	// 尝试结束通话（如果尚未结束）
+	// callService.End 会检查调用是否仍然活跃，如果已经结束则返回 nil
+	_, err := h.callService.End(ctx, roomName, "", "room_finished")
+	if err != nil {
+		h.logger.Debug("room_finished: call end returned error (may already be ended)",
+			zap.String("room", roomName),
+			zap.Error(err),
+		)
+	}
+}
--- a/internal/handler/ws_handler.go
+++ b/internal/handler/ws_handler.go
@@ -212,7 +212,7 @@ func (h *WSHandler) readPump(conn *websocket.Conn, client *ws.Client) {
 		conn.Close()
 	}()

-	conn.SetReadLimit(256 * 1024) // 256KB (SDP/ICE candidates can be large)
+	conn.SetReadLimit(256 * 1024) // 256KB
 	conn.SetReadDeadline(time.Now().Add(60 * time.Second))
 	conn.SetPongHandler(func(string) error {
 		zap.L().Debug("WebSocket pong received",
@@ -383,10 +383,8 @@ func (h *WSHandler) handleMessage(client *ws.Client, msg *ws.Message) {
 		h.handleCallReject(ctx, client, msg.Payload)
 	case "call_busy":
 		h.handleCallBusy(ctx, client, msg.Payload)
-	case "call_sdp":
-		h.handleCallSDP(ctx, client, msg.Payload)
-	case "call_ice":
-		h.handleCallICE(ctx, client, msg.Payload)
+	case "call_ready":
+		h.handleCallReady(ctx, client, msg.Payload)
 	case "call_end":
 		h.handleCallEnd(ctx, client, msg.Payload)
 	case "call_mute":
@@ -717,43 +715,18 @@ func (h *WSHandler) handleCallBusy(ctx context.Context, client *ws.Client, paylo
 	}
 }

-// handleCallSDP 转发 SDP
-func (h *WSHandler) handleCallSDP(ctx context.Context, client *ws.Client, payload json.RawMessage) {
+// handleCallReady 处理通话就绪（LiveKit 房间连接成功）
+func (h *WSHandler) handleCallReady(ctx context.Context, client *ws.Client, payload json.RawMessage) {
 	var req struct {
-		CallID  string          `json:"call_id"`
-		SDPType string          `json:"sdp_type"`
-		SDP     json.RawMessage `json:"sdp"`
-	}
-	if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" || req.SDPType == "" {
-		h.publisher.SendError(client, "invalid_params", "call_id and sdp_type are required")
-		return
-	}
-
-	signalPayload, _ := json.Marshal(map[string]any{
-		"sdp_type": req.SDPType,
-		"sdp":      json.RawMessage(req.SDP),
-	})
-	if err := h.callService.RelaySignal(ctx, req.CallID, client.UserID, "call_sdp", signalPayload); err != nil {
-		h.publisher.SendError(client, "call_sdp_failed", err.Error())
-	}
-}
-
-// handleCallICE 转发 ICE candidate
-func (h *WSHandler) handleCallICE(ctx context.Context, client *ws.Client, payload json.RawMessage) {
-	var req struct {
-		CallID    string          `json:"call_id"`
-		Candidate json.RawMessage `json:"candidate"`
+		CallID string `json:"call_id"`
 	}
 	if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" {
 		h.publisher.SendError(client, "invalid_params", "call_id is required")
 		return
 	}

-	signalPayload, _ := json.Marshal(map[string]any{
-		"candidate": json.RawMessage(req.Candidate),
-	})
-	if err := h.callService.RelaySignal(ctx, req.CallID, client.UserID, "call_ice", signalPayload); err != nil {
-		h.publisher.SendError(client, "call_ice_failed", err.Error())
+	if err := h.callService.Ready(ctx, req.CallID, client.UserID); err != nil {
+		h.publisher.SendError(client, "call_ready_failed", err.Error())
 	}
 }

--- a/internal/model/audit_log.go
+++ b/internal/model/audit_log.go
@@ -1,119 +0,0 @@
-package model
-
-import (
-	"time"
-
-	"gorm.io/gorm"
-)
-
-// AuditTargetType 审核对象类型
-type AuditTargetType string
-
-const (
-	AuditTargetTypePost        AuditTargetType = "post"         // 帖子
-	AuditTargetTypeComment     AuditTargetType = "comment"      // 评论
-	AuditTargetTypeMessage     AuditTargetType = "message"      // 私信
-	AuditTargetTypeUser        AuditTargetType = "user"         // 用户资料
-	AuditTargetTypeImage       AuditTargetType = "image"        // 图片
-	AuditTargetTypeAvatar      AuditTargetType = "avatar"       // 头像
-	AuditTargetTypeCover       AuditTargetType = "cover"        // 背景图
-	AuditTargetTypeBio         AuditTargetType = "bio"          // 个性签名
-	AuditTargetTypeUserProfile AuditTargetType = "user_profile" // 用户资料审核任务
-)
-
-// AuditResult 审核结果
-type AuditResult string
-
-const (
-	AuditResultPass    AuditResult = "pass"    // 通过
-	AuditResultReview  AuditResult = "review"  // 需人工复审
-	AuditResultBlock   AuditResult = "block"   // 违规拦截
-	AuditResultUnknown AuditResult = "unknown" // 未知
-)
-
-// AuditRiskLevel 风险等级
-type AuditRiskLevel string
-
-const (
-	AuditRiskLevelLow    AuditRiskLevel = "low"    // 低风险
-	AuditRiskLevelMedium AuditRiskLevel = "medium" // 中风险
-	AuditRiskLevelHigh   AuditRiskLevel = "high"   // 高风险
-)
-
-// AuditSource 审核来源
-type AuditSource string
-
-const (
-	AuditSourceAuto     AuditSource = "auto"     // 自动审核
-	AuditSourceManual   AuditSource = "manual"   // 人工审核
-	AuditSourceCallback AuditSource = "callback" // 回调审核
-)
-
-// AuditLog 审核日志实体
-type AuditLog struct {
-	ID           string          `json:"id" gorm:"type:varchar(36);primaryKey"`
-	TargetType   AuditTargetType `json:"target_type" gorm:"type:varchar(50);index"`
-	TargetID     string          `json:"target_id" gorm:"type:varchar(255);index"`
-	Content      string          `json:"content" gorm:"type:text"`             // 待审核内容
-	ContentType  string          `json:"content_type" gorm:"type:varchar(50)"` // 内容类型: text, image
-	ContentURL   string          `json:"content_url" gorm:"type:text"`         // 图片/文件URL
-	AuditType    string          `json:"audit_type" gorm:"type:varchar(50)"`   // 审核类型: porn, violence, ad, political, fraud, gamble
-	Result       AuditResult     `json:"result" gorm:"type:varchar(50);index"`
-	RiskLevel    AuditRiskLevel  `json:"risk_level" gorm:"type:varchar(20)"`
-	Labels       string          `json:"labels" gorm:"type:text"`                 // JSON数组，标签列表
-	Suggestion   string          `json:"suggestion" gorm:"type:varchar(50)"`      // pass, review, block
-	Detail       string          `json:"detail" gorm:"type:text"`                 // 详细说明
-	ThirdPartyID string          `json:"third_party_id" gorm:"type:varchar(255)"` // 第三方审核服务返回的ID
-	Source       AuditSource     `json:"source" gorm:"type:varchar(20);default:auto"`
-	ReviewerID   string          `json:"reviewer_id" gorm:"type:varchar(255)"`           // 审核人ID（人工审核时使用）
-	ReviewerName string          `json:"reviewer_name" gorm:"type:varchar(100)"`         // 审核人名称
-	ReviewTime   *time.Time      `json:"review_time" gorm:"index"`                       // 审核时间
-	UserID       string          `json:"user_id" gorm:"type:varchar(255);index"`         // 内容发布者ID
-	UserIP       string          `json:"user_ip" gorm:"type:varchar(45)"`                // 用户IP
-	Status       string          `json:"status" gorm:"type:varchar(20);default:pending"` // pending, completed, failed
-	RejectReason string          `json:"reject_reason" gorm:"type:text"`                 // 拒绝原因（人工审核时使用）
-	ExtraData    string          `json:"extra_data" gorm:"type:text"`                    // 额外数据，JSON格式
-	CreatedAt    time.Time       `json:"created_at" gorm:"autoCreateTime"`
-	UpdatedAt    time.Time       `json:"updated_at" gorm:"autoUpdateTime"`
-	DeletedAt    gorm.DeletedAt  `json:"-" gorm:"index"`
-}
-
-// BeforeCreate 创建前生成UUID
-func (al *AuditLog) BeforeCreate(tx *gorm.DB) error {
-	SetUUIDIfEmpty(&al.ID)
-	return nil
-}
-
-func (AuditLog) TableName() string {
-	return "audit_logs"
-}
-
-// AuditLogRequest 创建审核日志请求
-type AuditLogRequest struct {
-	TargetType  AuditTargetType `json:"target_type" validate:"required"`
-	TargetID    string          `json:"target_id" validate:"required"`
-	Content     string          `json:"content"`
-	ContentType string          `json:"content_type"`
-	ContentURL  string          `json:"content_url"`
-	AuditType   string          `json:"audit_type"`
-	UserID      string          `json:"user_id"`
-	UserIP      string          `json:"user_ip"`
-}
-
-// AuditLogListItem 审核日志列表项
-type AuditLogListItem struct {
-	ID          string          `json:"id"`
-	TargetType  AuditTargetType `json:"target_type"`
-	TargetID    string          `json:"target_id"`
-	Content     string          `json:"content"`
-	ContentType string          `json:"content_type"`
-	Result      AuditResult     `json:"result"`
-	RiskLevel   AuditRiskLevel  `json:"risk_level"`
-	Suggestion  string          `json:"suggestion"`
-	Source      AuditSource     `json:"source"`
-	ReviewerID  string          `json:"reviewer_id"`
-	ReviewTime  *time.Time      `json:"review_time"`
-	UserID      string          `json:"user_id"`
-	Status      string          `json:"status"`
-	CreatedAt   time.Time       `json:"created_at"`
-}
--- a/internal/model/init.go
+++ b/internal/model/init.go
@@ -167,7 +167,7 @@ func autoMigrate(db *gorm.DB) error {

 		// 敏感词和审核
 		&SensitiveWord{},
-		&AuditLog{},
+		// &AuditLog{}, // TODO: define AuditLog model

 		// 举报
 		&Report{},
--- a/internal/pkg/crypto/batch_decrypt.go
+++ b/internal/pkg/crypto/batch_decrypt.go
@@ -1,76 +0,0 @@
-package crypto
-
-import (
-	"sync"
-)
-
-// BatchDecryptResult 批量解密结果
-type BatchDecryptResult struct {
-	Index     int
-	Plaintext []byte
-	Error     error
-}
-
-// BatchDecrypt 批量并行解密
-// workerCount 指定并行工作数，如果 <= 0 则使用 CPU 核数
-func (e *MessageEncryptor) BatchDecrypt(ciphertexts []string, workerCount int) [][]byte {
-	if e == nil || len(ciphertexts) == 0 {
-		return nil
-	}
-
-	results := make([][]byte, len(ciphertexts))
-
-	// 小批量直接串行处理
-	if len(ciphertexts) <= 10 {
-		for i, ct := range ciphertexts {
-			if ct == "" {
-				continue
-			}
-			plaintext, err := e.Decrypt(ct)
-			if err == nil {
-				results[i] = plaintext
-			}
-		}
-		return results
-	}
-
-	// 并行处理
-	jobs := make(chan int, len(ciphertexts))
-	var wg sync.WaitGroup
-
-	// 确定工作数
-	if workerCount <= 0 {
-		workerCount = 4 // 默认4个并行
-	}
-	if workerCount > len(ciphertexts) {
-		workerCount = len(ciphertexts)
-	}
-
-	// 启动 worker
-	for w := 0; w < workerCount; w++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			for i := range jobs {
-				ct := ciphertexts[i]
-				if ct == "" {
-					continue
-				}
-				plaintext, err := e.Decrypt(ct)
-				if err == nil {
-					results[i] = plaintext
-				}
-			}
-		}()
-	}
-
-	// 分发任务
-	for i := range ciphertexts {
-		jobs <- i
-	}
-	close(jobs)
-
-	wg.Wait()
-	return results
-}
-
--- a/internal/pkg/ws/hub.go
+++ b/internal/pkg/ws/hub.go
@@ -343,7 +343,7 @@ func (h *Hub) PublishToUserOnline(userID string, eventType string, payload any)
 }

 // PublishToUserOnlineReliable 可靠地发送事件给在线用户（阻塞发送）
-// 用于重要的信令消息（如 SDP, ICE candidate），确保消息送达
+// 用于重要的信令消息（如通话邀请、接听），确保消息送达
 // 如果用户不在线，返回 false
 func (h *Hub) PublishToUserOnlineReliable(userID string, eventType string, payload any) bool {
 	h.mu.RLock()
--- a/internal/router/router.go
+++ b/internal/router/router.go
@@ -39,6 +39,7 @@ type RouterDeps struct {
 	QRCodeHandler            *handler.QRCodeHandler
 	MaterialHandler          *handler.MaterialHandler
 	CallHandler              *handler.CallHandler
+	LiveKitHandler           *handler.LiveKitHandler
 	ReportHandler            *handler.ReportHandler
 	AdminReportHandler       *handler.AdminReportHandler
 	VerificationHandler      *handler.VerificationHandler
@@ -83,6 +84,7 @@ type Router struct {
 	qrcodeHandler            *handler.QRCodeHandler
 	materialHandler          *handler.MaterialHandler
 	callHandler              *handler.CallHandler
+	liveKitHandler           *handler.LiveKitHandler
 	reportHandler            *handler.ReportHandler
 	adminReportHandler       *handler.AdminReportHandler
 	verificationHandler      *handler.VerificationHandler
@@ -130,6 +132,7 @@ func New(deps RouterDeps) *Router {
 		qrcodeHandler:            deps.QRCodeHandler,
 		materialHandler:          deps.MaterialHandler,
 		callHandler:              deps.CallHandler,
+			liveKitHandler:           deps.LiveKitHandler,
 		reportHandler:            deps.ReportHandler,
 		adminReportHandler:       deps.AdminReportHandler,
 		verificationHandler:      deps.VerificationHandler,
@@ -462,10 +465,20 @@ func (r *Router) setupRoutes() {
 			calls.Use(authMiddleware)
 			{
 				calls.GET("/history", r.callHandler.GetCallHistory)
-				calls.GET("/ice-servers", r.callHandler.GetICEServers)
 			}
 		}

+		// LiveKit 通话路由
+		if r.liveKitHandler != nil {
+			lk := v1.Group("/calls")
+			lk.Use(authMiddleware)
+			{
+				lk.GET("/token", r.liveKitHandler.GetToken)
+			}
+			// Webhook 不需要认证（LiveKit 服务端调用）
+			v1.POST("/calls/webhook", r.liveKitHandler.HandleWebhook)
+		}
+
 		// 消息操作路由
 		messages := v1.Group("/messages")
 		messages.Use(authMiddleware)
--- a/internal/service/call_service.go
+++ b/internal/service/call_service.go
@@ -59,8 +59,6 @@ type ActiveCall struct {
 	Duration       int64 // 通话时长（秒）
 	// 参与者状态
 	Participants map[string]*ActiveParticipant
-	// ICE Servers
-	ICEServers []config.ICEServerConfig
 }

 // ActiveParticipant 内存中的活跃参与者
@@ -77,10 +75,10 @@ type CallService interface {
 	Reject(ctx context.Context, callID, userID string) error
 	Busy(ctx context.Context, callID, userID string) error
 	End(ctx context.Context, callID, userID string, reason string) (*ActiveCall, error)
-	RelaySignal(ctx context.Context, callID, fromUserID string, signalType string, payload json.RawMessage) error
 	SetMuted(ctx context.Context, callID, userID string, muted bool) error
+	Ready(ctx context.Context, callID, userID string) error
+	GetActiveCall(ctx context.Context, callID string) *ActiveCall
 	GetCallHistory(ctx context.Context, userID string, page, pageSize int) ([]model.CallSession, int64, error)
-	GetICEServers() []config.ICEServerConfig
 	StartCleanupTicker()
 }

@@ -217,8 +215,7 @@ func (s *callService) redisGetCall(callID string) *ActiveCall {
 		CallType:       model.CallType(data.CallType),
 		Status:         model.CallStatus(data.Status),
 		MediaType:      data.MediaType,
-		ICEServers:     s.config.WebRTC.ICEServers,
-		Participants: map[string]*ActiveParticipant{
+				Participants: map[string]*ActiveParticipant{
 			data.CallerID: {UserID: data.CallerID, Status: model.ParticipantStatusJoined},
 			data.CalleeID: {UserID: data.CalleeID, Status: model.ParticipantStatusInvited},
 		},
@@ -390,8 +387,7 @@ func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversati
 			callerID: {UserID: callerID, Status: model.ParticipantStatusJoined, JoinedAt: &now},
 			calleeID: {UserID: calleeID, Status: model.ParticipantStatusInvited},
 		},
-		ICEServers: s.config.WebRTC.ICEServers,
-	}
+			}

 	// 存储到内存
 	s.storeActiveCall(call)
@@ -411,8 +407,7 @@ func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversati
 		"media_type":      mediaType,
 		"created_at":      now.UnixMilli(),
 		"lifetime":        CallLifetimeMs,
-		"ice_servers":     s.config.WebRTC.ICEServers,
-	}
+			}

 	online := s.hub.PublishToUserOnline(calleeID, "call_incoming", payload)

@@ -473,8 +468,7 @@ func (s *callService) Accept(ctx context.Context, callID, userID string) (*Activ
 	s.hub.PublishToUserOnline(call.CallerID, "call_accepted", map[string]any{
 		"call_id":     callID,
 		"started_at":  now.UnixMilli(),
-		"ice_servers": s.config.WebRTC.ICEServers,
-	})
+			})

 	// 通知被叫方其他设备
 	s.hub.PublishToUserOnline(userID, "call_answered_elsewhere", map[string]any{
@@ -626,35 +620,6 @@ func (s *callService) End(ctx context.Context, callID, userID string, reason str
 	return call, nil
 }

-func (s *callService) RelaySignal(ctx context.Context, callID, fromUserID string, signalType string, payload json.RawMessage) error {
-	call := s.getActiveCall(callID)
-	if call == nil {
-		return apperrors.ErrCallNotFound
-	}
-
-	s.mu.RLock()
-	if call.Status != model.CallStatusCalling && call.Status != model.CallStatusConnected {
-		s.mu.RUnlock()
-		return apperrors.ErrCallNotActive
-	}
-	if !isParticipant(call, fromUserID) {
-		s.mu.RUnlock()
-		return apperrors.ErrNotCallParticipant
-	}
-	s.mu.RUnlock()
-
-	for pUserID := range call.Participants {
-		if pUserID != fromUserID {
-			s.hub.PublishToUserOnlineReliable(pUserID, signalType, map[string]any{
-				"call_id": callID,
-				"from_id": fromUserID,
-				"payload": json.RawMessage(payload),
-			})
-		}
-	}
-	return nil
-}
-
 func (s *callService) SetMuted(ctx context.Context, callID, userID string, muted bool) error {
 	call := s.getActiveCall(callID)
 	if call == nil {
@@ -684,13 +649,41 @@ func (s *callService) SetMuted(ctx context.Context, callID, userID string, muted
 	return nil
 }

+// GetActiveCall 获取活跃通话（公开方法，供 LiveKit handler 调用）
+func (s *callService) GetActiveCall(ctx context.Context, callID string) *ActiveCall {
+	return s.getActiveCall(callID)
+}
+
+// Ready 标记参与者已加入 LiveKit 房间
+func (s *callService) Ready(ctx context.Context, callID, userID string) error {
+	call := s.getActiveCall(callID)
+	if call == nil {
+		return apperrors.ErrCallNotFound
+	}
+
+	s.mu.RLock()
+	if !isParticipant(call, userID) {
+		s.mu.RUnlock()
+		return apperrors.ErrNotCallParticipant
+	}
+	s.mu.RUnlock()
+
+	// 通知其他参与者该用户已就绪
+	for pUserID := range call.Participants {
+		if pUserID != userID {
+			s.hub.PublishToUserOnline(pUserID, "call_ready", map[string]any{
+				"call_id": callID,
+				"user_id": userID,
+			})
+		}
+	}
+	return nil
+}
+
 func (s *callService) GetCallHistory(ctx context.Context, userID string, page, pageSize int) ([]model.CallSession, int64, error) {
 	return s.callRepo.GetCallHistory(userID, page, pageSize)
 }

-func (s *callService) GetICEServers() []config.ICEServerConfig {
-	return s.config.WebRTC.ICEServers
-}

 // saveCallHistory 保存通话历史到数据库
 func (s *callService) saveCallHistory(call *ActiveCall, status model.CallStatus, duration int64) {
--- a/internal/service/livekit_service.go
+++ b/internal/service/livekit_service.go
@@ -0,0 +1,191 @@
+package service
+
+import (
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	"with_you/internal/config"
+
+	"github.com/livekit/protocol/auth"
+	"github.com/livekit/protocol/livekit"
+	"go.uber.org/zap"
+	"google.golang.org/protobuf/encoding/protojson"
+)
+
+// LiveKitService LiveKit 令牌生成和 Webhook 处理服务
+type LiveKitService interface {
+	GenerateToken(roomName, userID string, canPublish, canSubscribe bool) (string, error)
+	ProcessWebhook(ctx context.Context, body []byte, authHeader string) (*livekit.WebhookEvent, error)
+}
+
+type liveKitService struct {
+	config *config.LiveKitConfig
+	logger *zap.Logger
+}
+
+// NewLiveKitService 创建 LiveKit 服务
+func NewLiveKitService(cfg *config.Config, logger *zap.Logger) LiveKitService {
+	return &liveKitService{
+		config: &cfg.LiveKit,
+		logger: logger,
+	}
+}
+
+// GenerateToken 为指定 room 和 user 生成 LiveKit 访问令牌
+func (s *liveKitService) GenerateToken(roomName, userID string, canPublish, canSubscribe bool) (string, error) {
+	if !s.config.Enabled {
+		return "", fmt.Errorf("livekit is not enabled")
+	}
+
+	ttl := s.config.TokenTTL
+	if ttl <= 0 {
+		ttl = 600 // 默认 10 分钟
+	}
+
+	at := auth.NewAccessToken(s.config.APIKey, s.config.APISecret)
+	at.SetName(userID)
+	at.SetIdentity(userID)
+	at.SetValidFor(time.Duration(ttl) * time.Second)
+
+	grant := &auth.VideoGrant{
+		RoomJoin:       true,
+		Room:           roomName,
+		CanPublish:     &canPublish,
+		CanSubscribe:   &canSubscribe,
+		CanPublishData: &canSubscribe,
+	}
+	at.SetVideoGrant(grant)
+
+	token, err := at.ToJWT()
+	if err != nil {
+		s.logger.Error("failed to generate livekit token", zap.Error(err))
+		return "", fmt.Errorf("generate token: %w", err)
+	}
+
+	return token, nil
+}
+
+// ProcessWebhook 验证并解析 LiveKit Webhook 事件
+func (s *liveKitService) ProcessWebhook(ctx context.Context, body []byte, authHeader string) (*livekit.WebhookEvent, error) {
+	if s.config.WebhookSecret != "" {
+		if err := s.verifyWebhookSignature(body, authHeader); err != nil {
+			return nil, fmt.Errorf("webhook signature verification failed: %w", err)
+		}
+	}
+
+	var event livekit.WebhookEvent
+	if err := protojson.Unmarshal(body, &event); err != nil {
+		return nil, fmt.Errorf("failed to unmarshal webhook event: %w", err)
+	}
+
+	s.logger.Info("received livekit webhook",
+		zap.String("event", event.Event),
+		zap.String("room_name", event.Room.GetName()),
+	)
+
+	return &event, nil
+}
+
+// verifyWebhookSignature 验证 Webhook 签名
+func (s *liveKitService) verifyWebhookSignature(body []byte, authHeader string) error {
+	// LiveKit webhook 使用 SHA256 HMAC 签名
+	// Authorization header 格式: "sha256=<base64 encoded signature>"
+	if authHeader == "" {
+		return fmt.Errorf("missing authorization header")
+	}
+
+	expectedPrefix := "sha256="
+	if len(authHeader) < len(expectedPrefix) || authHeader[:len(expectedPrefix)] != expectedPrefix {
+		return fmt.Errorf("invalid authorization header format")
+	}
+
+	expectedSig, err := base64.StdEncoding.DecodeString(authHeader[len(expectedPrefix):])
+	if err != nil {
+		return fmt.Errorf("failed to decode signature: %w", err)
+	}
+
+	mac := hmac.New(sha256.New, []byte(s.config.WebhookSecret))
+	mac.Write(body)
+	actualSig := mac.Sum(nil)
+
+	if !hmac.Equal(actualSig, expectedSig) {
+		return fmt.Errorf("signature mismatch")
+	}
+
+	return nil
+}
+
+// WebhookRoomFinished 处理 room_finished 事件的负载
+type WebhookRoomFinished struct {
+	RoomName   string `json:"room_name"`
+	Duration   int64  `json:"duration"`
+	StartedAt  int64  `json:"started_at"`
+	EndedAt    int64  `json:"ended_at"`
+}
+
+// ParseRoomFinished 从 webhook event 中解析 room_finished 数据
+func ParseRoomFinished(event *livekit.WebhookEvent) (*WebhookRoomFinished, error) {
+	if event.Event != "room_finished" {
+		return nil, fmt.Errorf("not a room_finished event: %s", event.Event)
+	}
+
+	room := event.Room
+	if room == nil {
+		return nil, fmt.Errorf("room is nil in webhook event")
+	}
+
+	duration := int64(0)
+	startedAt := int64(0)
+	endedAt := int64(0)
+
+	if room.CreationTime > 0 {
+		startedAt = room.CreationTime
+	}
+
+	endedAt = time.Now().Unix()
+	if startedAt > 0 {
+		duration = endedAt - startedAt
+	}
+
+	return &WebhookRoomFinished{
+		RoomName:  room.Name,
+		Duration:  duration,
+		StartedAt: startedAt,
+		EndedAt:   endedAt,
+	}, nil
+}
+
+// WebhookEventJSON 用于 JSON 序列化的 webhook 事件
+type WebhookEventJSON struct {
+	Event    string               `json:"event"`
+	Room     *WebhookRoomJSON     `json:"room,omitempty"`
+	RoomName string               `json:"room_name,omitempty"`
+}
+
+// WebhookRoomJSON webhook 房间信息 JSON
+type WebhookRoomJSON struct {
+	Sid          string `json:"sid,omitempty"`
+	Name         string `json:"name,omitempty"`
+	CreationTime int64  `json:"creation_time,omitempty"`
+	NumParticipants uint32 `json:"num_participants,omitempty"`
+}
+
+// WebhookEventToJSON 将 webhook event 转为自定义 JSON 结构（避免 protobuf 内部字段）
+func WebhookEventToJSON(event *livekit.WebhookEvent) (*WebhookEventJSON, error) {
+	data, err := protojson.Marshal(event)
+	if err != nil {
+		return nil, err
+	}
+
+	var result WebhookEventJSON
+	if err := json.Unmarshal(data, &result); err != nil {
+		return nil, err
+	}
+
+	return &result, nil
+}
--- a/internal/wire/handler.go
+++ b/internal/wire/handler.go
@@ -1,12 +1,14 @@
 package wire

 import (
+	"with_you/internal/config"
 	"with_you/internal/handler"
 	"with_you/internal/pkg/ws"
 	"with_you/internal/repository"
 	"with_you/internal/service"

 	"github.com/google/wire"
+	"go.uber.org/zap"
 )

 // HandlerSet Handler 层 Provider Set
@@ -29,6 +31,7 @@ var HandlerSet = wire.NewSet(
 	handler.NewQRCodeHandler,
 	handler.NewMaterialHandler,
 	handler.NewCallHandler,
+	ProvideLiveKitHandler,
 	handler.NewReportHandler,
 	handler.NewAdminReportHandler,
 	handler.NewVerificationHandler,
@@ -106,6 +109,16 @@ func ProvideWSHandler(
 	return handler.NewWSHandler(publisher, chatService, groupService, jwtService, callService, userRepo)
 }

+// ProvideLiveKitHandler 提供 LiveKit 处理器
+func ProvideLiveKitHandler(
+	liveKitService service.LiveKitService,
+	callService service.CallService,
+	cfg *config.Config,
+	logger *zap.Logger,
+) *handler.LiveKitHandler {
+	return handler.NewLiveKitHandler(liveKitService, callService, cfg, logger)
+}
+
 // ProvideAdminVerificationHandler 提供管理端身份认证处理器
 func ProvideAdminVerificationHandler(
 	adminVerificationService service.AdminVerificationService,
--- a/internal/wire/service.go
+++ b/internal/wire/service.go
@@ -62,6 +62,7 @@ var ServiceSet = wire.NewSet(
 	ProvideHotRankWorker,
 	ProvideMaterialService,
 	ProvideCallService,
+	ProvideLiveKitService,
 	ProvideReportService,
 	ProvideAdminReportService,
 	ProvideVerificationService,
@@ -463,6 +464,14 @@ func ProvideCallService(
 	return service.NewCallService(callRepo, publisher, cfg, db, redisClient)
 }

+// ProvideLiveKitService 提供 LiveKit 服务
+func ProvideLiveKitService(
+	cfg *config.Config,
+	logger *zap.Logger,
+) service.LiveKitService {
+	return service.NewLiveKitService(cfg, logger)
+}
+
 // ProvideReportService 提供举报服务
 func ProvideReportService(
 	reportRepo repository.ReportRepository,