270 lines
8.7 KiB
Go
270 lines
8.7 KiB
Go
package chat
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"gorm.io/gorm"
|
|
"gorm.io/gorm/clause"
|
|
"hfb_sys/backend/internal/model"
|
|
"hfb_sys/backend/internal/modules/chathub"
|
|
"time"
|
|
)
|
|
|
|
func (r *Repository) Messages(ctx context.Context, principal Principal, conversationID uint64, page, pageSize int) (*PaginatedResult, error) {
|
|
page, pageSize = normalizePagination(page, pageSize)
|
|
db := r.db.WithContext(ctx)
|
|
|
|
// 管理员可以查看任意会话的消息，普通用户需要是 participant
|
|
if principal.Type != "admin" {
|
|
if _, err := r.findParticipant(db, principal, conversationID, false); err != nil {
|
|
return nil, err
|
|
}
|
|
} else {
|
|
// 管理员需要验证会话存在
|
|
var count int64
|
|
if err := db.Model(&model.ChatConversation{}).Where("id = ?", conversationID).Count(&count).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
if count == 0 {
|
|
return nil, ErrConversationNotFound
|
|
}
|
|
}
|
|
|
|
// 构建查询
|
|
query := db.Model(&model.ChatMessage{}).Where("conversation_id = ?", conversationID)
|
|
|
|
var total int64
|
|
if err := query.Count(&total).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
offset := (page - 1) * pageSize
|
|
var rows []model.ChatMessage
|
|
if err := query.Order("id DESC").
|
|
Offset(offset).
|
|
Limit(pageSize).
|
|
Find(&rows).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
for left, right := 0, len(rows)-1; left < right; left, right = left+1, right-1 {
|
|
rows[left], rows[right] = rows[right], rows[left]
|
|
}
|
|
items, err := r.toMessageDTOs(ctx, principal, rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &PaginatedResult{Items: items, Total: total, Page: page, PageSize: pageSize}, nil
|
|
}
|
|
func (r *Repository) SendMessage(ctx context.Context, principal Principal, conversationID uint64, req SendMessageRequest) (*MessageDTO, error) {
|
|
var messageID uint64
|
|
err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
var conversation model.ChatConversation
|
|
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&conversation, conversationID).Error; err != nil {
|
|
return err
|
|
}
|
|
if conversation.Status != "active" {
|
|
return ErrPermissionDenied
|
|
}
|
|
|
|
var senderRole string
|
|
var participant *model.ChatParticipant
|
|
|
|
// 管理员可以在任意会话发送消息，无需是 participant
|
|
if principal.Type == "admin" {
|
|
// 尝试查找管理员的 participant 记录
|
|
var p model.ChatParticipant
|
|
err := tx.Where("conversation_id = ? AND participant_type = ? AND participant_id = ?",
|
|
conversationID, principal.Type, principal.ID).First(&p).Error
|
|
if err == nil {
|
|
// 管理员是 participant，使用其角色
|
|
participant = &p
|
|
senderRole = p.Role
|
|
} else if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
// 管理员不是 participant，使用特殊角色 "admin"
|
|
senderRole = "admin"
|
|
} else {
|
|
return err
|
|
}
|
|
} else {
|
|
// 普通用户必须是 participant
|
|
p, err := r.findParticipant(tx, principal, conversationID, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
participant = p
|
|
senderRole = p.Role
|
|
}
|
|
|
|
message := model.ChatMessage{
|
|
ConversationID: conversation.ID,
|
|
SenderType: principal.Type,
|
|
SenderID: principal.ID,
|
|
SenderRole: senderRole,
|
|
ContentType: "text",
|
|
Content: req.Content,
|
|
AttachmentURLS: encodeStringList(req.AttachmentURLS),
|
|
}
|
|
if principal.Type == "user" {
|
|
message.AdminAttentionType = "user_inquiry"
|
|
}
|
|
if err := tx.Create(&message).Error; err != nil {
|
|
return err
|
|
}
|
|
conversation.LastMessageID = &message.ID
|
|
conversation.LastMessagePreview = messagePreview(message.Content, req.AttachmentURLS)
|
|
conversation.LastMessageAt = &message.CreatedAt
|
|
if err := tx.Save(&conversation).Error; err != nil {
|
|
return err
|
|
}
|
|
|
|
// 更新 participant 的已读时间（仅当是 participant 时）
|
|
if participant != nil {
|
|
now := time.Now()
|
|
if err := tx.Model(participant).Update("last_read_at", now).Error; err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
messageID = message.ID
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var message model.ChatMessage
|
|
if err := r.db.WithContext(ctx).First(&message, messageID).Error; err != nil {
|
|
return nil, err
|
|
}
|
|
items, err := r.toMessageDTOs(ctx, principal, []model.ChatMessage{message})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(items) == 0 {
|
|
return nil, ErrConversationNotFound
|
|
}
|
|
// 推送新消息事件给会话中的在线参与者
|
|
if r.hub != nil {
|
|
msg := items[0]
|
|
event := &chathub.ChatEvent{
|
|
Type: "new_message",
|
|
ConversationID: conversationID,
|
|
Message: &chathub.MessageData{
|
|
ID: msg.ID,
|
|
ConversationID: msg.ConversationID,
|
|
SenderType: msg.SenderType,
|
|
SenderID: msg.SenderID,
|
|
SenderRole: msg.SenderRole,
|
|
SenderName: msg.SenderName,
|
|
ContentType: msg.ContentType,
|
|
Content: msg.Content,
|
|
AttachmentURLS: msg.AttachmentURLS,
|
|
AdminAttentionType: msg.AdminAttentionType,
|
|
CreatedAt: msg.CreatedAt.Format(time.RFC3339),
|
|
},
|
|
}
|
|
r.hub.NotifyConversation(conversationID, event)
|
|
// 同时广播给所有在线客服，使「未分配 / 全部」视图实时刷新；
|
|
// 参与者客服会重复收到，由前端按消息 id 去重。
|
|
r.hub.NotifyAllAdmins(event)
|
|
}
|
|
return &items[0], nil
|
|
}
|
|
func (r *Repository) MarkRead(ctx context.Context, principal Principal, conversationID uint64) error {
|
|
now := time.Now()
|
|
updated := false
|
|
err := r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
if principal.Type == "admin" {
|
|
var conversation model.ChatConversation
|
|
if err := tx.Select("id", "last_message_id").First(&conversation, conversationID).Error; err != nil {
|
|
if errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return ErrConversationNotFound
|
|
}
|
|
return err
|
|
}
|
|
lastMessageID := uint64(0)
|
|
if conversation.LastMessageID != nil {
|
|
lastMessageID = *conversation.LastMessageID
|
|
}
|
|
var existing model.ChatAdminConversationState
|
|
err := tx.Where("conversation_id = ? AND admin_user_id = ?", conversationID, principal.ID).
|
|
First(&existing).Error
|
|
if err == nil && existing.LastReadMessageID >= lastMessageID {
|
|
return nil
|
|
}
|
|
if err != nil && !errors.Is(err, gorm.ErrRecordNotFound) {
|
|
return err
|
|
}
|
|
state := model.ChatAdminConversationState{
|
|
ConversationID: conversationID,
|
|
AdminUserID: principal.ID,
|
|
LastReadMessageID: lastMessageID,
|
|
LastReadAt: &now,
|
|
}
|
|
if err := tx.Clauses(clause.OnConflict{
|
|
Columns: []clause.Column{{Name: "conversation_id"}, {Name: "admin_user_id"}},
|
|
DoUpdates: clause.Assignments(map[string]interface{}{
|
|
"last_read_message_id": lastMessageID,
|
|
"last_read_at": now,
|
|
}),
|
|
}).Create(&state).Error; err != nil {
|
|
return err
|
|
}
|
|
updated = true
|
|
return nil
|
|
}
|
|
|
|
// 普通用户必须是 participant
|
|
participant, err := r.findParticipant(tx, principal, conversationID, true)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
updated = true
|
|
return tx.Model(participant).Update("last_read_at", now).Error
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
// 已读时间发生变化时，通知会话内其他参与者刷新「已读」状态。
|
|
if updated && r.hub != nil {
|
|
event := &chathub.ChatEvent{
|
|
Type: "conversation_read",
|
|
ConversationID: conversationID,
|
|
ReaderType: principal.Type,
|
|
ReaderID: principal.ID,
|
|
ReadAt: now.Format(time.RFC3339),
|
|
}
|
|
r.hub.NotifyConversation(conversationID, event)
|
|
// 同时广播给所有在线客服，覆盖未分配会话中非参与者客服的「已读」刷新。
|
|
r.hub.NotifyAllAdmins(event)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// MarkAllAdminConversationsRead advances one admin's read cursor for every conversation in one transaction.
|
|
func (r *Repository) MarkAllAdminConversationsRead(ctx context.Context, adminID uint64) error {
|
|
now := time.Now()
|
|
return r.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
|
// 先补齐状态行，再更新游标。拆成两步可兼容本地 MySQL/MariaDB 对
|
|
// INSERT ... SELECT 同表读取并 ON DUPLICATE KEY UPDATE 的限制。
|
|
if err := tx.Exec(`
|
|
INSERT INTO chat_admin_conversation_states
|
|
(conversation_id, admin_user_id, remark, last_read_message_id, last_read_at)
|
|
SELECT c.id, ?, COALESCE(existing.remark, ''), COALESCE(c.last_message_id, 0), ?
|
|
FROM chat_conversations AS c
|
|
LEFT JOIN chat_admin_conversation_states AS existing
|
|
ON existing.conversation_id = c.id AND existing.admin_user_id = ?
|
|
WHERE existing.id IS NULL`, adminID, now, adminID).Error; err != nil {
|
|
return err
|
|
}
|
|
return tx.Exec(`
|
|
UPDATE chat_admin_conversation_states AS cas
|
|
JOIN chat_conversations AS c ON c.id = cas.conversation_id
|
|
SET cas.last_read_message_id = c.last_message_id,
|
|
cas.last_read_at = ?,
|
|
cas.updated_at = CURRENT_TIMESTAMP
|
|
WHERE cas.admin_user_id = ?
|
|
AND c.last_message_id IS NOT NULL
|
|
AND cas.last_read_message_id < c.last_message_id`, now, adminID).Error
|
|
})
|
|
}
|