273 lines
7.2 KiB
Go
273 lines
7.2 KiB
Go
package usecase
|
||
|
||
import (
|
||
"backend/pkg/chat/domain/entity"
|
||
"backend/pkg/chat/domain/repository"
|
||
"backend/pkg/chat/domain/usecase"
|
||
"backend/pkg/library/centrifugo"
|
||
errs "backend/pkg/library/errors"
|
||
"backend/pkg/utils"
|
||
"context"
|
||
"crypto/md5"
|
||
"encoding/hex"
|
||
"fmt"
|
||
"math"
|
||
"time"
|
||
|
||
"github.com/gocql/gocql"
|
||
"github.com/google/uuid"
|
||
)
|
||
|
||
const (
|
||
defaultPageSize = 20
|
||
maxPageSize = 100 // 設置最大分頁大小
|
||
)
|
||
|
||
type MessageUseCaseParam struct {
|
||
MessageRepo repository.MessageRepository
|
||
RoomRepo repository.RoomRepository
|
||
MsgClient *centrifugo.Client
|
||
Logger errs.Logger
|
||
}
|
||
|
||
type MessageUseCase struct {
|
||
MessageUseCaseParam
|
||
}
|
||
|
||
// NewMessageUseCase 創建新的訊息 UseCase
|
||
func NewMessageUseCase(param MessageUseCaseParam) usecase.MessageUseCase {
|
||
return &MessageUseCase{
|
||
param,
|
||
}
|
||
}
|
||
|
||
func (use *MessageUseCase) SendMessage(ctx context.Context, param usecase.SendMessageReq) error {
|
||
// 驗證輸入參數
|
||
if err := use.validateSendMessageReq(param); err != nil {
|
||
return err
|
||
}
|
||
|
||
// 驗證使用者是否在房間中（優先檢查，避免不必要的去重操作）
|
||
if err := use.verifyRoomMembership(ctx, param.UID, param.RoomID); err != nil {
|
||
return err
|
||
}
|
||
|
||
// 去重檢查
|
||
now := time.Now().UTC()
|
||
bucketSec := now.Unix()
|
||
contentMD5 := calculateMD5(param.Content)
|
||
|
||
isDuplicate, err := use.MessageRepo.CheckAndInsertDedup(ctx, repository.CheckDupReq{
|
||
RoomID: param.RoomID,
|
||
UID: param.UID, // 修復：應該是 param.UID 而不是 param.RoomID
|
||
BucketSec: bucketSec,
|
||
ContentMD5: contentMD5,
|
||
})
|
||
if err != nil {
|
||
return use.logError("messageRepo.CheckAndInsertDedup", param, err, "failed to check message deduplication")
|
||
}
|
||
|
||
if isDuplicate {
|
||
return errs.InputInvalidFormatError("duplicate message detected")
|
||
}
|
||
|
||
// 建立並儲存訊息
|
||
msg, err := use.createMessage(param, now)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
if err = use.MessageRepo.Insert(ctx, msg); err != nil {
|
||
return use.logError("messageRepo.Insert", param, err, "failed to insert message")
|
||
}
|
||
|
||
// 發布到 Centrifugo（非阻塞，失敗不影響主流程）
|
||
use.publishToCentrifugo(ctx, param.RoomID, msg, now)
|
||
|
||
return nil
|
||
}
|
||
|
||
// validateSendMessageReq 驗證發送訊息的請求參數
|
||
func (use *MessageUseCase) validateSendMessageReq(param usecase.SendMessageReq) error {
|
||
if param.Content == "" {
|
||
return errs.InputInvalidFormatError("content cannot be empty")
|
||
}
|
||
if param.RoomID == "" {
|
||
return errs.InputInvalidFormatError("room_id cannot be empty")
|
||
}
|
||
if param.UID == "" {
|
||
return errs.InputInvalidFormatError("uid cannot be empty")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// verifyRoomMembership 驗證使用者是否在房間中
|
||
func (use *MessageUseCase) verifyRoomMembership(ctx context.Context, uid, roomID string) error {
|
||
isMember, err := use.RoomRepo.IsUserInRoom(ctx, uid, roomID)
|
||
if err != nil {
|
||
return use.logError("roomRepo.IsUserInRoom", map[string]interface{}{
|
||
"uid": uid,
|
||
"roomID": roomID,
|
||
}, err, "failed to check room membership")
|
||
}
|
||
|
||
if !isMember {
|
||
return errs.AuthForbiddenErrorL(
|
||
use.Logger,
|
||
[]errs.LogField{
|
||
{Key: "uid", Val: uid},
|
||
{Key: "roomID", Val: roomID},
|
||
},
|
||
"user is not a member of the room")
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
// createMessage 建立訊息實體
|
||
func (use *MessageUseCase) createMessage(param usecase.SendMessageReq, now time.Time) (*entity.Message, error) {
|
||
roomID, err := gocql.ParseUUID(param.RoomID)
|
||
if err != nil {
|
||
return nil, errs.InputInvalidFormatError("invalid room_id format").Wrap(err)
|
||
}
|
||
|
||
msgID, err := uuid.NewV7()
|
||
if err != nil {
|
||
return nil, errs.SysInternalError("failed to generate message id").Wrap(err)
|
||
}
|
||
|
||
return &entity.Message{
|
||
RoomID: roomID,
|
||
BucketDay: utils.GetBucketDay(now),
|
||
UID: param.UID,
|
||
MsgID: msgID,
|
||
Content: param.Content,
|
||
}, nil
|
||
}
|
||
|
||
// publishToCentrifugo 發布訊息到 Centrifugo
|
||
func (use *MessageUseCase) publishToCentrifugo(ctx context.Context, roomID string, msg *entity.Message, now time.Time) {
|
||
channel := fmt.Sprintf("room:%s", roomID)
|
||
messageData := map[string]interface{}{
|
||
"msg_id": msg.MsgID.String(),
|
||
"uid": msg.UID,
|
||
"content": msg.Content,
|
||
"timestamp": now.UnixNano(), // 使用實際時間戳，而不是 msg.TS（可能為 0）
|
||
"room_id": msg.RoomID.String(),
|
||
}
|
||
|
||
if _, err := use.MsgClient.PublishJSON(ctx, channel, messageData); err != nil {
|
||
// 記錄錯誤但不影響主流程，因為訊息已經成功儲存
|
||
if use.Logger != nil {
|
||
use.Logger.WithFields(
|
||
errs.LogField{Key: "roomID", Val: roomID},
|
||
errs.LogField{Key: "msgID", Val: msg.MsgID.String()},
|
||
errs.LogField{Key: "error", Val: err.Error()},
|
||
).Error(fmt.Sprintf("failed to publish message to Centrifugo: %v", err))
|
||
}
|
||
}
|
||
}
|
||
|
||
// logError 統一的錯誤記錄方法
|
||
func (use *MessageUseCase) logError(funcName string, param interface{}, err error, message string) error {
|
||
return errs.DBErrorErrorL(
|
||
use.Logger,
|
||
[]errs.LogField{
|
||
{Key: "param", Val: param},
|
||
{Key: "func", Val: funcName},
|
||
{Key: "err", Val: err.Error()},
|
||
},
|
||
message)
|
||
}
|
||
|
||
func (use *MessageUseCase) ListMessages(ctx context.Context, req usecase.ListMessagesReq) ([]usecase.Message, int64, error) {
|
||
// 驗證輸入參數
|
||
if err := use.validateListMessagesReq(req); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
|
||
// 驗證使用者是否在房間中
|
||
if err := use.verifyRoomMembership(ctx, req.UID, req.RoomID); err != nil {
|
||
return nil, 0, err
|
||
}
|
||
|
||
// 取得 bucket_day（如果未提供則使用今天）
|
||
bucketDay := req.BucketDay
|
||
if bucketDay == "" {
|
||
bucketDay = utils.GetTodayBucketDay()
|
||
}
|
||
|
||
// 防止 PageSize overflow 並設置合理的範圍
|
||
pageSize := use.normalizePageSize(req.PageSize)
|
||
|
||
// 查詢訊息
|
||
messages, err := use.MessageRepo.ListMessages(ctx, repository.ListMessagesReq{
|
||
RoomID: req.RoomID,
|
||
BucketDay: bucketDay,
|
||
PageSize: pageSize,
|
||
LastTS: req.LastTS,
|
||
})
|
||
if err != nil {
|
||
return nil, 0, use.logError("messageRepo.ListMessages", req, err, "failed to list messages")
|
||
}
|
||
|
||
// 轉換為 usecase.Message
|
||
result := make([]usecase.Message, 0, len(messages))
|
||
for _, msg := range messages {
|
||
result = append(result, usecase.Message{
|
||
RoomID: msg.RoomID.String(),
|
||
BucketDay: msg.BucketDay,
|
||
TS: msg.TS,
|
||
UID: msg.UID,
|
||
Content: msg.Content,
|
||
})
|
||
}
|
||
|
||
// 計算總數（只在第一頁時計算）
|
||
var total int64
|
||
if req.LastTS == 0 {
|
||
total, err = use.MessageRepo.Count(ctx, req.RoomID)
|
||
if err != nil {
|
||
return nil, 0, use.logError("messageRepo.Count", req, err, "failed to count messages")
|
||
}
|
||
}
|
||
|
||
return result, total, nil
|
||
}
|
||
|
||
// validateListMessagesReq 驗證查詢訊息的請求參數
|
||
func (use *MessageUseCase) validateListMessagesReq(req usecase.ListMessagesReq) error {
|
||
if req.RoomID == "" {
|
||
return errs.InputInvalidFormatError("room_id cannot be empty")
|
||
}
|
||
if req.UID == "" {
|
||
return errs.InputInvalidFormatError("uid cannot be empty")
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// normalizePageSize 正規化分頁大小
|
||
func (use *MessageUseCase) normalizePageSize(pageSize int64) int {
|
||
// 檢查是否超過 int 的最大值
|
||
if pageSize > int64(math.MaxInt) {
|
||
return maxPageSize
|
||
}
|
||
|
||
size := int(pageSize)
|
||
if size <= 0 {
|
||
return defaultPageSize
|
||
}
|
||
|
||
if size > maxPageSize {
|
||
return maxPageSize
|
||
}
|
||
|
||
return size
|
||
}
|
||
|
||
// calculateMD5 計算字串的 MD5 雜湊值
|
||
func calculateMD5(content string) string {
|
||
hash := md5.Sum([]byte(content))
|
||
return hex.EncodeToString(hash[:])
|
||
}
|