diff --git a/cmd/server/wire.go b/cmd/server/wire.go index dc90369..46125f7 100644 --- a/cmd/server/wire.go +++ b/cmd/server/wire.go @@ -47,6 +47,7 @@ func ProvideRouter( adminLogHandler *handler.AdminLogHandler, qrcodeHandler *handler.QRCodeHandler, materialHandler *handler.MaterialHandler, + callHandler *handler.CallHandler, logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, @@ -76,6 +77,7 @@ func ProvideRouter( adminLogHandler, qrcodeHandler, materialHandler, + callHandler, logService, activityService, casbinService, diff --git a/cmd/server/wire_gen.go b/cmd/server/wire_gen.go index c6d9edb..a7f1494 100644 --- a/cmd/server/wire_gen.go +++ b/cmd/server/wire_gen.go @@ -117,10 +117,13 @@ func InitializeApp() (*App, error) { qrCodeHandler := handler.NewQRCodeHandler(qrCodeLoginService) materialSubjectRepository := repository.NewMaterialSubjectRepository(db) materialFileRepository := repository.NewMaterialFileRepository(db) + callRepository := repository.NewCallRepository(db) materialService := wire.ProvideMaterialService(materialSubjectRepository, materialFileRepository) + callService := wire.ProvideCallService(callRepository, hub, config) materialHandler := handler.NewMaterialHandler(materialService) - wsHandler := wire.ProvideWSHandler(hub, chatService, groupService, jwtService) - router := ProvideRouter(userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, voteHandler, channelHandler, scheduleHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, qrCodeHandler, materialHandler, logService, userActivityService, casbinService, wsHandler) + callHandler := handler.NewCallHandler(callService) + wsHandler := wire.ProvideWSHandler(hub, chatService, groupService, jwtService, callService) + router := ProvideRouter(userHandler, postHandler, commentHandler, messageHandler, notificationHandler, uploadHandler, jwtService, pushHandler, systemMessageHandler, groupHandler, stickerHandler, voteHandler, channelHandler, scheduleHandler, roleHandler, adminUserHandler, adminPostHandler, adminCommentHandler, adminGroupHandler, adminDashboardHandler, adminLogHandler, qrCodeHandler, materialHandler, callHandler, logService, userActivityService, casbinService, wsHandler) hotRankWorker := wire.ProvideHotRankWorker(config, postRepository, cache) app := NewApp(config, db, router, pushService, hotRankWorker, server, logger) return app, nil @@ -153,6 +156,7 @@ func ProvideRouter( adminLogHandler *handler.AdminLogHandler, qrcodeHandler *handler.QRCodeHandler, materialHandler *handler.MaterialHandler, + callHandler *handler.CallHandler, logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, @@ -182,6 +186,7 @@ func ProvideRouter( adminLogHandler, qrcodeHandler, materialHandler, + callHandler, logService, activityService, casbinService, diff --git a/go.mod b/go.mod index 1aeb383..adf5a16 100644 --- a/go.mod +++ b/go.mod @@ -12,6 +12,7 @@ require ( github.com/golang-jwt/jwt/v5 v5.3.1 github.com/google/uuid v1.6.0 github.com/google/wire v0.7.0 + github.com/gorilla/websocket v1.5.3 github.com/minio/minio-go/v7 v7.0.99 github.com/redis/go-redis/v9 v9.18.0 github.com/spf13/viper v1.21.0 @@ -51,7 +52,6 @@ require ( github.com/goccy/go-yaml v1.19.2 // indirect github.com/golang-sql/civil v0.0.0-20220223132316-b832511892a9 // indirect github.com/golang-sql/sqlexp v0.1.0 // indirect - github.com/gorilla/websocket v1.5.3 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect github.com/jackc/pgx/v5 v5.8.0 // indirect diff --git a/go.sum b/go.sum index 60a4add..af86b37 100644 --- a/go.sum +++ b/go.sum @@ -115,8 +115,6 @@ github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e h1:ijClszYn+mADRFY17kjQEVQ1XRhq2/JR1M3sGqeJoxs= github.com/google/pprof v0.0.0-20250317173921-a4b03ec1a45e/go.mod h1:boTsfXsheKC2y+lKOCMpSfarhxDeIzfZG1jqGcPl3cA= -github.com/google/subcommands v1.2.0 h1:vWQspBTo2nEqTUFita5/KeEWlUL8kQObDFbub/EN9oE= -github.com/google/subcommands v1.2.0/go.mod h1:ZjhPrFU+Olkh9WazFPsl27BQ4UPiG37m3yTrtFlrHVk= github.com/google/uuid v1.3.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= diff --git a/internal/config/config.go b/internal/config/config.go index 43601b7..9aeea0b 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -29,6 +29,7 @@ type Config struct { Casbin CasbinConfig `mapstructure:"casbin"` Encryption EncryptionConfig `mapstructure:"encryption"` HotRank HotRankConfig `mapstructure:"hot_rank"` + WebRTC WebRTCConfig `mapstructure:"webrtc"` } // Load 加载配置文件 @@ -150,6 +151,10 @@ 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]interface{}{ + {"urls": []string{"stun:stun.l.google.com:19302"}}, + }) if err := viper.ReadInConfig(); err != nil { return nil, fmt.Errorf("failed to read config: %w", err) diff --git a/internal/config/webrtc.go b/internal/config/webrtc.go new file mode 100644 index 0000000..984a1a6 --- /dev/null +++ b/internal/config/webrtc.go @@ -0,0 +1,13 @@ +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"` +} diff --git a/internal/dto/call_dto.go b/internal/dto/call_dto.go new file mode 100644 index 0000000..5d7b523 --- /dev/null +++ b/internal/dto/call_dto.go @@ -0,0 +1,191 @@ +package dto + +import ( + "time" + + "carrot_bbs/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"` +} + +// ==================== Converter Functions ==================== + +// ConvertCallSessionToResponse 转换通话会话为响应格式 +func ConvertCallSessionToResponse(call *model.CallSession, caller *model.User, participants []*model.User) *CallSessionResponse { + resp := &CallSessionResponse{ + ID: call.ID, + ConversationID: call.ConversationID, + CallerID: call.CallerID, + CallType: call.CallType, + Status: call.Status, + StartedAt: call.StartedAt, + EndedAt: call.EndedAt, + Duration: call.Duration, + CreatedAt: call.CreatedAt, + } + + if call.GroupID != nil { + resp.GroupID = *call.GroupID + } + + if caller != nil { + resp.Caller = ConvertUserToResponse(caller) + } + + // 转换参与者 + resp.Participants = make([]CallParticipantResponse, 0, len(call.Participants)) + userMap := make(map[string]*model.User) + for _, u := range participants { + userMap[u.ID] = u + } + + for _, p := range call.Participants { + pr := CallParticipantResponse{ + UserID: p.UserID, + Status: p.Status, + JoinedAt: p.JoinedAt, + LeftAt: p.LeftAt, + } + if u, ok := userMap[p.UserID]; ok { + pr.User = ConvertUserToResponse(u) + } + resp.Participants = append(resp.Participants, pr) + } + + return resp +} diff --git a/internal/handler/call_handler.go b/internal/handler/call_handler.go new file mode 100644 index 0000000..9ec2c35 --- /dev/null +++ b/internal/handler/call_handler.go @@ -0,0 +1,60 @@ +package handler + +import ( + "strconv" + + "github.com/gin-gonic/gin" + + "carrot_bbs/internal/pkg/response" + "carrot_bbs/internal/service" +) + +// CallHandler 通话记录处理器 +type CallHandler struct { + callService service.CallService +} + +// NewCallHandler 创建通话处理器 +func NewCallHandler(callService service.CallService) *CallHandler { + return &CallHandler{callService: callService} +} + +// GetCallHistory 获取通话记录 +// GET /api/v1/calls/history?page=1&page_size=20 +func (h *CallHandler) GetCallHistory(c *gin.Context) { + userID, exists := c.Get("user_id") + if !exists { + response.Unauthorized(c, "unauthorized") + return + } + + page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) + pageSize, _ := strconv.Atoi(c.DefaultQuery("page_size", "20")) + if page < 1 { + page = 1 + } + if pageSize < 1 || pageSize > 50 { + pageSize = 20 + } + + calls, total, err := h.callService.GetCallHistory(c.Request.Context(), userID.(string), page, pageSize) + if err != nil { + response.InternalServerError(c, "failed to get call history") + return + } + + response.Success(c, gin.H{ + "items": calls, + "total": total, + "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, + }) +} diff --git a/internal/handler/ws_handler.go b/internal/handler/ws_handler.go index 7b760a0..0b7a036 100644 --- a/internal/handler/ws_handler.go +++ b/internal/handler/ws_handler.go @@ -36,6 +36,7 @@ type WSHandler struct { chatService service.ChatService groupService service.GroupService jwtService *service.JWTService + callService service.CallService clientSeq uint64 } @@ -45,12 +46,14 @@ func NewWSHandler( chatService service.ChatService, groupService service.GroupService, jwtService *service.JWTService, + callService service.CallService, ) *WSHandler { return &WSHandler{ wsHub: wsHub, chatService: chatService, groupService: groupService, jwtService: jwtService, + callService: callService, } } @@ -151,7 +154,7 @@ func (h *WSHandler) readPump(conn *websocket.Conn, client *ws.Client) { conn.Close() }() - conn.SetReadLimit(64 * 1024) // 64KB + conn.SetReadLimit(256 * 1024) // 256KB (SDP/ICE candidates can be large) conn.SetReadDeadline(time.Now().Add(60 * time.Second)) conn.SetPongHandler(func(string) error { zap.L().Debug("WebSocket pong received", @@ -260,6 +263,23 @@ func (h *WSHandler) handleMessage(client *ws.Client, msg *ws.Message) { h.handleRead(ctx, client, msg.Payload) case "recall": h.handleRecall(ctx, client, msg.Payload) + // 通话信令 + case "call_invite": + h.handleCallInvite(ctx, client, msg.Payload) + case "call_answer": + h.handleCallAnswer(ctx, client, msg.Payload) + case "call_reject": + 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_end": + h.handleCallEnd(ctx, client, msg.Payload) + case "call_mute": + h.handleCallMute(ctx, client, msg.Payload) default: h.wsHub.SendError(client, "unknown_type", fmt.Sprintf("unknown message type: %s", msg.Type)) } @@ -421,3 +441,199 @@ func (h *WSHandler) handleRecall(ctx context.Context, client *ws.Client, payload } const defaultUserBufferSize = 128 + +// ==================== 通话信令处理 ==================== + +// handleCallInvite 处理呼叫邀请 +func (h *WSHandler) handleCallInvite(ctx context.Context, client *ws.Client, payload json.RawMessage) { + var req struct { + CalleeID string `json:"callee_id"` + ConversationID string `json:"conversation_id"` + } + if err := json.Unmarshal(payload, &req); err != nil { + h.wsHub.SendError(client, "parse_error", "invalid call_invite format") + return + } + if req.CalleeID == "" || req.ConversationID == "" { + h.wsHub.SendError(client, "invalid_params", "callee_id and conversation_id are required") + return + } + + call, err := h.callService.Invite(ctx, client.UserID, req.CalleeID, req.ConversationID) + if err != nil { + zap.L().Warn("Failed to invite call", + zap.String("caller_id", client.UserID), + zap.String("callee_id", req.CalleeID), + zap.Error(err), + ) + h.wsHub.SendError(client, "call_invite_failed", err.Error()) + return + } + + resp := ws.ResponseMessage{ + EventID: h.wsHub.NextID(), + Type: "call_invited", + TS: time.Now().UnixMilli(), + Payload: map[string]interface{}{ + "call_id": call.ID, + "conversation_id": req.ConversationID, + "callee_id": req.CalleeID, + }, + } + data, _ := json.Marshal(resp) + select { + case <-client.Quit: + case client.Send <- data: + } +} + +// handleCallAnswer 处理接听 +func (h *WSHandler) handleCallAnswer(ctx context.Context, client *ws.Client, payload json.RawMessage) { + var req struct { + CallID string `json:"call_id"` + } + if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { + h.wsHub.SendError(client, "invalid_params", "call_id is required") + return + } + + call, err := h.callService.Accept(ctx, req.CallID, client.UserID) + if err != nil { + h.wsHub.SendError(client, "call_accept_failed", err.Error()) + return + } + + resp := ws.ResponseMessage{ + EventID: h.wsHub.NextID(), + Type: "call_accepted", + TS: time.Now().UnixMilli(), + Payload: map[string]interface{}{ + "call_id": call.ID, + "started_at": call.StartedAt, + }, + } + data, _ := json.Marshal(resp) + select { + case <-client.Quit: + case client.Send <- data: + } +} + +// handleCallReject 处理拒绝 +func (h *WSHandler) handleCallReject(ctx context.Context, client *ws.Client, payload json.RawMessage) { + var req struct { + CallID string `json:"call_id"` + } + if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { + h.wsHub.SendError(client, "invalid_params", "call_id is required") + return + } + + if err := h.callService.Reject(ctx, req.CallID, client.UserID); err != nil { + h.wsHub.SendError(client, "call_reject_failed", err.Error()) + } +} + +// handleCallBusy 处理忙线 +func (h *WSHandler) handleCallBusy(ctx context.Context, client *ws.Client, payload json.RawMessage) { + var req struct { + CallID string `json:"call_id"` + } + if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { + h.wsHub.SendError(client, "invalid_params", "call_id is required") + return + } + + if err := h.callService.Busy(ctx, req.CallID, client.UserID); err != nil { + h.wsHub.SendError(client, "call_busy_failed", err.Error()) + } +} + +// handleCallSDP 转发 SDP +func (h *WSHandler) handleCallSDP(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.wsHub.SendError(client, "invalid_params", "call_id and sdp_type are required") + return + } + + signalPayload, _ := json.Marshal(map[string]interface{}{ + "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.wsHub.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"` + } + if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { + h.wsHub.SendError(client, "invalid_params", "call_id is required") + return + } + + signalPayload, _ := json.Marshal(map[string]interface{}{ + "candidate": json.RawMessage(req.Candidate), + }) + if err := h.callService.RelaySignal(ctx, req.CallID, client.UserID, "call_ice", signalPayload); err != nil { + h.wsHub.SendError(client, "call_ice_failed", err.Error()) + } +} + +// handleCallEnd 处理结束通话 +func (h *WSHandler) handleCallEnd(ctx context.Context, client *ws.Client, payload json.RawMessage) { + var req struct { + CallID string `json:"call_id"` + Reason string `json:"reason"` + } + if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { + h.wsHub.SendError(client, "invalid_params", "call_id is required") + return + } + + call, err := h.callService.End(ctx, req.CallID, client.UserID, req.Reason) + if err != nil { + h.wsHub.SendError(client, "call_end_failed", err.Error()) + return + } + + resp := ws.ResponseMessage{ + EventID: h.wsHub.NextID(), + Type: "call_ended", + TS: time.Now().UnixMilli(), + Payload: map[string]interface{}{ + "call_id": call.ID, + "duration": call.Duration, + }, + } + data, _ := json.Marshal(resp) + select { + case <-client.Quit: + case client.Send <- data: + } +} + +// handleCallMute 处理静音 +func (h *WSHandler) handleCallMute(ctx context.Context, client *ws.Client, payload json.RawMessage) { + var req struct { + CallID string `json:"call_id"` + Muted bool `json:"muted"` + } + if err := json.Unmarshal(payload, &req); err != nil || req.CallID == "" { + h.wsHub.SendError(client, "invalid_params", "call_id is required") + return + } + + if err := h.callService.SetMuted(ctx, req.CallID, client.UserID, req.Muted); err != nil { + h.wsHub.SendError(client, "call_mute_failed", err.Error()) + } +} diff --git a/internal/model/init.go b/internal/model/init.go index a18e740..f53a8d5 100644 --- a/internal/model/init.go +++ b/internal/model/init.go @@ -163,6 +163,10 @@ func autoMigrate(db *gorm.DB) error { // 学习资料相关 &MaterialSubject{}, &MaterialFile{}, + + // 通话相关 + &CallSession{}, + &CallParticipant{}, ) if err != nil { return err diff --git a/internal/repository/call_repo.go b/internal/repository/call_repo.go new file mode 100644 index 0000000..52910b6 --- /dev/null +++ b/internal/repository/call_repo.go @@ -0,0 +1,156 @@ +package repository + +import ( + "context" + + "carrot_bbs/internal/model" + + "gorm.io/gorm" +) + +// CallRepository 通话仓库接口 +type CallRepository interface { + // 通话会话操作 + CreateCall(call *model.CallSession) error + GetCallByID(id string) (*model.CallSession, error) + GetCallByIDWithParticipants(id string) (*model.CallSession, error) + UpdateCall(call *model.CallSession) error + GetActiveCallByConversationID(conversationID string) (*model.CallSession, error) + + // 通话参与者操作 + CreateCallParticipant(participant *model.CallParticipant) error + GetCallParticipant(callID, userID string) (*model.CallParticipant, error) + GetCallParticipants(callID string) ([]model.CallParticipant, error) + UpdateCallParticipant(participant *model.CallParticipant) error + + // 通话历史 + GetCallHistory(userID string, page, pageSize int) ([]model.CallSession, int64, error) + + // 事务操作 + CreateCallWithParticipants(ctx context.Context, call *model.CallSession, participants []*model.CallParticipant) error +} + +// callRepository 通话仓库实现 +type callRepository struct { + db *gorm.DB +} + +// NewCallRepository 创建通话仓库 +func NewCallRepository(db *gorm.DB) CallRepository { + return &callRepository{db: db} +} + +// CreateCall 创建通话会话 +func (r *callRepository) CreateCall(call *model.CallSession) error { + return r.db.Create(call).Error +} + +// GetCallByID 根据ID获取通话会话 +func (r *callRepository) GetCallByID(id string) (*model.CallSession, error) { + var call model.CallSession + err := r.db.First(&call, "id = ?", id).Error + if err != nil { + return nil, err + } + return &call, nil +} + +// GetCallByIDWithParticipants 根据ID获取通话会话（包含参与者） +func (r *callRepository) GetCallByIDWithParticipants(id string) (*model.CallSession, error) { + var call model.CallSession + err := r.db.Preload("Participants").First(&call, "id = ?", id).Error + if err != nil { + return nil, err + } + return &call, nil +} + +// UpdateCall 更新通话会话 +func (r *callRepository) UpdateCall(call *model.CallSession) error { + return r.db.Save(call).Error +} + +// GetActiveCallByConversationID 获取会话中的活跃通话 +func (r *callRepository) GetActiveCallByConversationID(conversationID string) (*model.CallSession, error) { + var call model.CallSession + err := r.db.Where("conversation_id = ? AND status IN ?", conversationID, []model.CallStatus{ + model.CallStatusCalling, + model.CallStatusConnected, + }).First(&call).Error + if err != nil { + return nil, err + } + return &call, nil +} + +// CreateCallParticipant 创建通话参与者 +func (r *callRepository) CreateCallParticipant(participant *model.CallParticipant) error { + return r.db.Create(participant).Error +} + +// GetCallParticipant 获取通话参与者 +func (r *callRepository) GetCallParticipant(callID, userID string) (*model.CallParticipant, error) { + var participant model.CallParticipant + err := r.db.Where("call_id = ? AND user_id = ?", callID, userID).First(&participant).Error + if err != nil { + return nil, err + } + return &participant, nil +} + +// GetCallParticipants 获取通话参与者列表 +func (r *callRepository) GetCallParticipants(callID string) ([]model.CallParticipant, error) { + var participants []model.CallParticipant + err := r.db.Where("call_id = ?", callID).Find(&participants).Error + if err != nil { + return nil, err + } + return participants, nil +} + +// UpdateCallParticipant 更新通话参与者 +func (r *callRepository) UpdateCallParticipant(participant *model.CallParticipant) error { + return r.db.Save(participant).Error +} + +// GetCallHistory 获取用户通话历史 +func (r *callRepository) GetCallHistory(userID string, page, pageSize int) ([]model.CallSession, int64, error) { + var calls []model.CallSession + var total int64 + + // 子查询获取用户参与的通话ID + subQuery := r.db.Model(&model.CallParticipant{}). + Select("DISTINCT call_id"). + Where("user_id = ?", userID) + + query := r.db.Model(&model.CallSession{}). + Where("id IN (?)", subQuery). + Order("created_at DESC") + + if err := query.Count(&total).Error; err != nil { + return nil, 0, err + } + + offset := (page - 1) * pageSize + if err := query.Preload("Participants").Offset(offset).Limit(pageSize).Find(&calls).Error; err != nil { + return nil, 0, err + } + + return calls, total, nil +} + +// CreateCallWithParticipants 创建通话会话和参与者（事务） +func (r *callRepository) CreateCallWithParticipants(ctx context.Context, call *model.CallSession, participants []*model.CallParticipant) error { + return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { + if err := tx.Create(call).Error; err != nil { + return err + } + for _, p := range participants { + p.CallID = call.ID + if err := tx.Create(p).Error; err != nil { + return err + } + } + return nil + }) +} diff --git a/internal/router/router.go b/internal/router/router.go index 31de36a..87f4e1e 100644 --- a/internal/router/router.go +++ b/internal/router/router.go @@ -34,6 +34,7 @@ type Router struct { adminLogHandler *handler.AdminLogHandler qrcodeHandler *handler.QRCodeHandler materialHandler *handler.MaterialHandler + callHandler *handler.CallHandler wsHandler *handler.WSHandler logService *service.LogService jwtService *service.JWTService @@ -65,6 +66,7 @@ func New( adminLogHandler *handler.AdminLogHandler, qrcodeHandler *handler.QRCodeHandler, materialHandler *handler.MaterialHandler, + callHandler *handler.CallHandler, logService *service.LogService, activityService service.UserActivityService, casbinService service.CasbinService, @@ -99,6 +101,7 @@ func New( adminLogHandler: adminLogHandler, qrcodeHandler: qrcodeHandler, materialHandler: materialHandler, + callHandler: callHandler, logService: logService, jwtService: jwtService, casbinService: casbinService, @@ -335,6 +338,16 @@ func (r *Router) setupRoutes() { realtime.GET("/ws", r.wsHandler.HandleWebSocket) } + // 通话路由 + if r.callHandler != nil { + calls := v1.Group("/calls") + calls.Use(authMiddleware) + { + calls.GET("/history", r.callHandler.GetCallHistory) + calls.GET("/ice-servers", r.callHandler.GetICEServers) + } + } + // 消息操作路由 messages := v1.Group("/messages") messages.Use(authMiddleware) diff --git a/internal/service/call_service.go b/internal/service/call_service.go new file mode 100644 index 0000000..2e76e74 --- /dev/null +++ b/internal/service/call_service.go @@ -0,0 +1,319 @@ +package service + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "time" + + "carrot_bbs/internal/config" + "carrot_bbs/internal/model" + "carrot_bbs/internal/pkg/ws" + "carrot_bbs/internal/repository" +) + +var ( + ErrCallNotFound = errors.New("call not found") + ErrCallNotActive = errors.New("call is not active") + ErrCallInProgress = errors.New("call already in progress between users") + ErrNotParticipant = errors.New("user is not a participant of this call") + ErrInvalidCallState = errors.New("invalid call state for this operation") +) + +// CallService 通话服务接口 +type CallService interface { + Invite(ctx context.Context, callerID, calleeID, conversationID string) (*model.CallSession, error) + Accept(ctx context.Context, callID, userID string) (*model.CallSession, error) + 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) (*model.CallSession, error) + RelaySignal(ctx context.Context, callID, fromUserID string, signalType string, payload json.RawMessage) error + SetMuted(ctx context.Context, callID, userID string, muted bool) error + GetCallHistory(ctx context.Context, userID string, page, pageSize int) ([]model.CallSession, int64, error) + GetICEServers() []config.ICEServerConfig +} + +type callService struct { + callRepo repository.CallRepository + hub *ws.Hub + config *config.Config +} + +// NewCallService 创建通话服务 +func NewCallService( + callRepo repository.CallRepository, + hub *ws.Hub, + cfg *config.Config, +) CallService { + return &callService{ + callRepo: callRepo, + hub: hub, + config: cfg, + } +} + +func (s *callService) Invite(ctx context.Context, callerID, calleeID, conversationID string) (*model.CallSession, error) { + active, err := s.callRepo.GetActiveCallByConversationID(conversationID) + if err == nil && active != nil { + return nil, fmt.Errorf("%w: call %s", ErrCallInProgress, active.ID) + } + + now := time.Now() + call := &model.CallSession{ + ConversationID: conversationID, + CallerID: callerID, + CallType: model.CallTypePrivate, + Status: model.CallStatusCalling, + CreatedAt: now, + UpdatedAt: now, + } + + participants := []*model.CallParticipant{ + {UserID: callerID, Status: model.ParticipantStatusJoined, JoinedAt: &now, CreatedAt: now, UpdatedAt: now}, + {UserID: calleeID, Status: model.ParticipantStatusInvited, CreatedAt: now, UpdatedAt: now}, + } + + if err := s.callRepo.CreateCallWithParticipants(ctx, call, participants); err != nil { + return nil, fmt.Errorf("create call session: %w", err) + } + + call, err = s.callRepo.GetCallByIDWithParticipants(call.ID) + if err != nil { + return nil, fmt.Errorf("get call with participants: %w", err) + } + + payload := map[string]interface{}{ + "call_id": call.ID, + "conversation_id": conversationID, + "caller_id": callerID, + "call_type": call.CallType, + "created_at": now.UnixMilli(), + "ice_servers": s.config.WebRTC.ICEServers, + } + s.hub.PublishToUser(calleeID, "call_incoming", payload) + + return call, nil +} + +func (s *callService) Accept(ctx context.Context, callID, userID string) (*model.CallSession, error) { + call, err := s.callRepo.GetCallByIDWithParticipants(callID) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrCallNotFound, err) + } + if !call.IsActive() { + return nil, ErrCallNotActive + } + if !isParticipant(call, userID) { + return nil, ErrNotParticipant + } + + now := time.Now() + call.Status = model.CallStatusConnected + call.StartedAt = &now + call.UpdatedAt = now + + if err := s.callRepo.UpdateCall(call); err != nil { + return nil, fmt.Errorf("update call connected: %w", err) + } + + participant, err := s.callRepo.GetCallParticipant(callID, userID) + if err == nil { + participant.Status = model.ParticipantStatusJoined + participant.JoinedAt = &now + s.callRepo.UpdateCallParticipant(participant) + } + + s.hub.PublishToUser(call.CallerID, "call_accepted", map[string]interface{}{ + "call_id": callID, + "started_at": now.UnixMilli(), + "ice_servers": s.config.WebRTC.ICEServers, + }) + + return call, nil +} + +func (s *callService) Reject(ctx context.Context, callID, userID string) error { + call, err := s.callRepo.GetCallByIDWithParticipants(callID) + if err != nil { + return fmt.Errorf("%w: %v", ErrCallNotFound, err) + } + if call.Status != model.CallStatusCalling { + return ErrInvalidCallState + } + if !isParticipant(call, userID) { + return ErrNotParticipant + } + + now := time.Now() + call.Status = model.CallStatusRejected + call.EndedAt = &now + call.UpdatedAt = now + + if err := s.callRepo.UpdateCall(call); err != nil { + return fmt.Errorf("update call rejected: %w", err) + } + + participant, err := s.callRepo.GetCallParticipant(callID, userID) + if err == nil { + participant.Status = model.ParticipantStatusRejected + participant.LeftAt = &now + s.callRepo.UpdateCallParticipant(participant) + } + + s.hub.PublishToUser(call.CallerID, "call_rejected", map[string]interface{}{ + "call_id": callID, + "reason": "rejected", + }) + return nil +} + +func (s *callService) Busy(ctx context.Context, callID, userID string) error { + call, err := s.callRepo.GetCallByIDWithParticipants(callID) + if err != nil { + return fmt.Errorf("%w: %v", ErrCallNotFound, err) + } + if call.Status != model.CallStatusCalling { + return ErrInvalidCallState + } + if !isParticipant(call, userID) { + return ErrNotParticipant + } + + now := time.Now() + call.Status = model.CallStatusMissed + call.EndedAt = &now + call.UpdatedAt = now + + if err := s.callRepo.UpdateCall(call); err != nil { + return fmt.Errorf("update call missed: %w", err) + } + + s.hub.PublishToUser(call.CallerID, "call_busy", map[string]interface{}{ + "call_id": callID, + }) + return nil +} + +func (s *callService) End(ctx context.Context, callID, userID string, reason string) (*model.CallSession, error) { + call, err := s.callRepo.GetCallByIDWithParticipants(callID) + if err != nil { + return nil, fmt.Errorf("%w: %v", ErrCallNotFound, err) + } + if !call.IsActive() { + return nil, ErrCallNotActive + } + if !isParticipant(call, userID) { + return nil, ErrNotParticipant + } + + now := time.Now() + var duration int64 + if call.StartedAt != nil { + duration = int64(now.Sub(*call.StartedAt).Seconds()) + } + + endStatus := model.CallStatusEnded + if call.Status == model.CallStatusCalling { + endStatus = model.CallStatusCancelled + } + + call.Status = endStatus + call.EndedAt = &now + call.Duration = duration + call.UpdatedAt = now + + if err := s.callRepo.UpdateCall(call); err != nil { + return nil, fmt.Errorf("end call: %w", err) + } + + participant, err := s.callRepo.GetCallParticipant(callID, userID) + if err == nil { + participant.Status = model.ParticipantStatusLeft + participant.LeftAt = &now + s.callRepo.UpdateCallParticipant(participant) + } + + participants, _ := s.callRepo.GetCallParticipants(callID) + for _, p := range participants { + if p.UserID != userID { + s.hub.PublishToUser(p.UserID, "call_ended", map[string]interface{}{ + "call_id": callID, + "ended_by": userID, + "reason": reason, + "duration": duration, + "ended_at": now.UnixMilli(), + }) + } + } + + return call, nil +} + +func (s *callService) RelaySignal(ctx context.Context, callID, fromUserID string, signalType string, payload json.RawMessage) error { + call, err := s.callRepo.GetCallByIDWithParticipants(callID) + if err != nil { + return fmt.Errorf("%w: %v", ErrCallNotFound, err) + } + if !call.IsActive() { + return ErrCallNotActive + } + if !isParticipant(call, fromUserID) { + return ErrNotParticipant + } + + participants, _ := s.callRepo.GetCallParticipants(callID) + for _, p := range participants { + if p.UserID != fromUserID { + s.hub.PublishToUser(p.UserID, signalType, map[string]interface{}{ + "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, err := s.callRepo.GetCallByIDWithParticipants(callID) + if err != nil { + return fmt.Errorf("%w: %v", ErrCallNotFound, err) + } + if !call.IsActive() { + return ErrCallNotActive + } + if !isParticipant(call, userID) { + return ErrNotParticipant + } + + participants, _ := s.callRepo.GetCallParticipants(callID) + for _, p := range participants { + if p.UserID != userID { + s.hub.PublishToUser(p.UserID, "call_peer_muted", map[string]interface{}{ + "call_id": callID, + "user_id": userID, + "muted": muted, + }) + } + } + 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 +} + +// isParticipant 检查用户是否是通话参与者 +func isParticipant(call *model.CallSession, userID string) bool { + for _, p := range call.Participants { + if p.UserID == userID { + return true + } + } + return false +} diff --git a/internal/wire/handler.go b/internal/wire/handler.go index 4a84583..f33a130 100644 --- a/internal/wire/handler.go +++ b/internal/wire/handler.go @@ -27,6 +27,7 @@ var HandlerSet = wire.NewSet( handler.NewAdminLogHandler, handler.NewQRCodeHandler, handler.NewMaterialHandler, + handler.NewCallHandler, // 需要特殊处理的 Handler ProvideUserHandler, @@ -126,6 +127,7 @@ func ProvideWSHandler( chatService service.ChatService, groupService service.GroupService, jwtService *service.JWTService, + callService service.CallService, ) *handler.WSHandler { - return handler.NewWSHandler(wsHub, chatService, groupService, jwtService) + return handler.NewWSHandler(wsHub, chatService, groupService, jwtService, callService) } diff --git a/internal/wire/repository.go b/internal/wire/repository.go index 8fcadca..03ad027 100644 --- a/internal/wire/repository.go +++ b/internal/wire/repository.go @@ -26,6 +26,7 @@ var RepositorySet = wire.NewSet( repository.NewScheduleRepository, repository.NewMaterialSubjectRepository, repository.NewMaterialFileRepository, + repository.NewCallRepository, ProvideUserActivityRepository, // 日志相关仓储 diff --git a/internal/wire/service.go b/internal/wire/service.go index 0b3ff2d..ad80254 100644 --- a/internal/wire/service.go +++ b/internal/wire/service.go @@ -55,6 +55,7 @@ var ServiceSet = wire.NewSet( ProvideQRCodeLoginService, ProvideHotRankWorker, ProvideMaterialService, + ProvideCallService, // 日志服务 ProvideAsyncLogManager, @@ -380,3 +381,12 @@ func ProvideMaterialService( func parseDuration(s string) (time.Duration, error) { return time.ParseDuration(s) } + +// ProvideCallService 提供通话服务 +func ProvideCallService( + callRepo repository.CallRepository, + wsHub *ws.Hub, + cfg *config.Config, +) service.CallService { + return service.NewCallService(callRepo, wsHub, cfg) +}