// Copyright 2022 E99p1ant. All rights reserved. // Use of this source code is governed by a MIT-style // license that can be found in the LICENSE file. package db import ( "context" "time" "github.com/pkg/errors" "github.com/sirupsen/logrus" "github.com/thanhpk/randstr" "gorm.io/gorm" "gorm.io/gorm/clause" "github.com/tamakyi/TamaBox/internal/dbutil" ) var Conversations ConversationStore var ( ErrDailyMessageLimit = errors.New("已达到每日私聊上限") ErrConversationLimit = errors.New("会话已达到消息上限或已关闭") ) type ConversationStore interface { Create(ctx context.Context, opts CreateConversationOptions) (*Conversation, error) // CreateWithFirstMessage 事务化地建会话 + 首条消息 + 消息计数。 // 单独写首条消息(用于「发第一条私聊」这类一次性动作),避免失败时留下空会话。 CreateWithFirstMessage(ctx context.Context, opts CreateConversationOptions, senderID uint, content string) (*Conversation, *Message, error) SendMessage(ctx context.Context, opts SendConversationMessageOptions) (*Message, error) GetByID(ctx context.Context, id uint) (*Conversation, error) GetByToken(ctx context.Context, token string) (*Conversation, error) GetByIDAndOwner(ctx context.Context, id, ownerID uint) (*Conversation, error) GetByIDAndAsker(ctx context.Context, id, askerID uint) (*Conversation, error) IncrementMessageCount(ctx context.Context, id uint) error IncrementMessageCountIfBelow(ctx context.Context, id uint, max int) (bool, error) DecrementMessageCount(ctx context.Context, id uint) error MaybeClose(ctx context.Context, id uint, max int) error Close(ctx context.Context, id uint) error // ListByOwner 返回某箱主拥有的会话。limit>0 时启用分页(取 limit 条、跳过 offset 条), // 用于私聊列表分页;导出场景传 limit=0 取全量,行为与改造前一致。 ListByOwner(ctx context.Context, ownerID uint, limit, offset int) ([]*Conversation, error) // CountByOwner 统计某箱主拥有的会话总数(分页看板与"加载更多"判定用)。 CountByOwner(ctx context.Context, ownerID uint) (int64, error) // ListByAsker 返回某提问者发起的会话;分页语义同 ListByOwner。 ListByAsker(ctx context.Context, askerID uint, limit, offset int) ([]*Conversation, error) // CountByAsker 统计某提问者发起的会话总数。 CountByAsker(ctx context.Context, askerID uint) (int64, error) MarkDeletedByOwner(ctx context.Context, id uint) error } func NewConversationStore(db *gorm.DB) ConversationStore { return &conversations{db} } type conversations struct { *gorm.DB } type Conversation struct { dbutil.Model OwnerID uint `gorm:"index:idx_conversation_owner" json:"owner_id"` AskerID uint `gorm:"index:idx_conversation_asker" json:"asker_id"` Token string `gorm:"size:6;uniqueIndex" json:"token"` MaxMessages int `gorm:"default:20" json:"max_messages"` MessageCount int `gorm:"default:0" json:"message_count"` IsClosed bool `gorm:"default:false" json:"is_closed"` IsDeletedByOwner bool `gorm:"default:false" json:"is_deleted_by_owner"` // 新增 } func (c *Conversation) BeforeCreate(tx *gorm.DB) error { c.MessageCount = 0 c.Token = randstr.String(6) return nil } type CreateConversationOptions struct { OwnerID uint AskerID uint MaxMessages int } type SendConversationMessageOptions struct { ConversationID uint SenderID uint Content string Image *CreateMessageImageOptions } func reserveDailyMessageTx(tx *gorm.DB, ownerID, senderID uint) error { var sender User if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id").First(&sender, senderID).Error; err != nil { return errors.Wrap(err, "lock message sender") } var owner User if err := tx.Select("daily_message_limit").First(&owner, ownerID).Error; err != nil { return errors.Wrap(err, "get owner daily message limit") } if owner.DailyMessageLimit <= 0 { return nil } now := time.Now() start := time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 0, now.Location()) var count int64 if err := tx.Table("messages").Joins("JOIN conversations ON conversations.id = messages.conversation_id"). Where("conversations.owner_id = ? AND messages.sender_id = ? AND messages.created_at >= ?", ownerID, senderID, start). Count(&count).Error; err != nil { return errors.Wrap(err, "count today messages") } if count >= int64(owner.DailyMessageLimit) { return ErrDailyMessageLimit } return nil } func (db *conversations) Create(ctx context.Context, opts CreateConversationOptions) (*Conversation, error) { conversation := &Conversation{ OwnerID: opts.OwnerID, AskerID: opts.AskerID, MaxMessages: opts.MaxMessages, } if err := db.WithContext(ctx).Create(conversation).Error; err != nil { return nil, errors.Wrap(err, "create conversation") } return conversation, nil } // CreateWithFirstMessage 在一个事务里完成「建会话 + 写首条消息 + 消息计数」。 // // 原来这三步是三次独立写入:第二步失败时会话已经提交,于是留下一条 // message_count=0、没有任何消息的「空会话」——箱主的私聊列表里能看到它, // 点进去什么都没有;更麻烦的是 Conversations.Create 的 BeforeCreate 已经 // 生成并占用了一个 6 字符 token(唯一索引),孤儿会话会永久占着这个 token。 func (db *conversations) CreateWithFirstMessage( ctx context.Context, opts CreateConversationOptions, senderID uint, content string, ) (*Conversation, *Message, error) { var conv *Conversation var msg *Message err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { if senderID == opts.AskerID { if err := reserveDailyMessageTx(tx, opts.OwnerID, senderID); err != nil { return err } } conv = &Conversation{ OwnerID: opts.OwnerID, AskerID: opts.AskerID, MaxMessages: opts.MaxMessages, } // 走 tx 而不是 db:BeforeCreate 里生成 token 的 hook 同样会被触发。 if err := tx.Create(conv).Error; err != nil { return errors.Wrap(err, "create conversation") } msg = &Message{ ConversationID: conv.ID, SenderID: senderID, Content: content, } if err := tx.Create(msg).Error; err != nil { return errors.Wrap(err, "create first message") } // 原子自增,与 IncrementMessageCount 保持同一写法 if err := tx.Model(&Conversation{}). Where("id = ?", conv.ID). UpdateColumn("message_count", gorm.Expr("message_count + 1")).Error; err != nil { return errors.Wrap(err, "increment message count") } // 私聊定量收集:新建会话即计一条(后续往来消息不重复计数)。 // 放在事务里而不是 handler 里:原来 handler 在事务之后调 IncrementQuota 且 // 失败只打 Warn,会话建成了但计数没加,箱主的「已收集」偏少、上限被绕过。 if err := incrementQuotaTx(tx, opts.OwnerID, FeatureChat, 1); err != nil { return errors.Wrap(err, "increment chat quota") } return nil }) if err != nil { return nil, nil, err } return conv, msg, nil } func (db *conversations) SendMessage(ctx context.Context, opts SendConversationMessageOptions) (*Message, error) { msg := &Message{ConversationID: opts.ConversationID, SenderID: opts.SenderID, Content: opts.Content} err := db.WithContext(ctx).Transaction(func(tx *gorm.DB) error { // 先锁发送者,再锁会话,保证与 CreateWithFirstMessage 的锁顺序一致, // 避免并发下两条路径互相等待造成配额检查窗口。 var sender User if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Select("id").First(&sender, opts.SenderID).Error; err != nil { return errors.Wrap(err, "lock message sender") } var conv Conversation if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&conv, opts.ConversationID).Error; err != nil { return errors.Wrap(err, "lock conversation") } // 发送者必须是会话双方之一 if conv.OwnerID != opts.SenderID && conv.AskerID != opts.SenderID { return errors.New("sender is not part of the conversation") } // 只有 asker 发送才消耗 owner 的每日配额 if conv.AskerID == opts.SenderID { if err := reserveDailyMessageTx(tx, conv.OwnerID, opts.SenderID); err != nil { return err } } res := tx.Model(&Conversation{}).Where("id = ? AND message_count < max_messages AND is_closed = ?", conv.ID, false). UpdateColumn("message_count", gorm.Expr("message_count + 1")) if res.Error != nil { return errors.Wrap(res.Error, "increment message count") } if res.RowsAffected == 0 { return ErrConversationLimit } if err := tx.Create(msg).Error; err != nil { return errors.Wrap(err, "create message") } if opts.Image != nil { image := *opts.Image image.MessageID = msg.ID if err := tx.Create(&MessageImage{MessageID: image.MessageID, URL: image.URL, ObjectKey: image.ObjectKey, FileName: image.FileName, FileSize: image.FileSize}).Error; err != nil { return errors.Wrap(err, "create message image") } } return nil }) if err != nil { return nil, err } return msg, nil } func (db *conversations) GetByID(ctx context.Context, id uint) (*Conversation, error) { var conv Conversation if err := db.WithContext(ctx).First(&conv, id).Error; err != nil { return nil, errors.Wrap(err, "get conversation") } return &conv, nil } func (db *conversations) GetByToken(ctx context.Context, token string) (*Conversation, error) { var conv Conversation if err := db.WithContext(ctx).Where("token = ?", token).First(&conv).Error; err != nil { return nil, errors.Wrap(err, "get conversation by token") } return &conv, nil } func (db *conversations) GetByIDAndOwner(ctx context.Context, id, ownerID uint) (*Conversation, error) { var conv Conversation if err := db.WithContext(ctx).Where("id = ? AND owner_id = ?", id, ownerID).First(&conv).Error; err != nil { return nil, errors.Wrap(err, "get by owner") } return &conv, nil } func (db *conversations) GetByIDAndAsker(ctx context.Context, id, askerID uint) (*Conversation, error) { var conv Conversation if err := db.WithContext(ctx).Where("id = ? AND asker_id = ?", id, askerID).First(&conv).Error; err != nil { return nil, errors.Wrap(err, "get by asker") } return &conv, nil } func (db *conversations) IncrementMessageCount(ctx context.Context, id uint) error { return db.WithContext(ctx).Model(&Conversation{}).Where("id = ?", id).UpdateColumn("message_count", gorm.Expr("message_count + 1")).Error } // IncrementMessageCountIfBelow 原子地检查并递增消息计数: // 仅当 message_count < max 且未关闭时才 +1,返回 true。 // 并发场景下(同一对话两端同时发送)可防止消息数超出上限。 func (db *conversations) IncrementMessageCountIfBelow(ctx context.Context, id uint, max int) (bool, error) { res := db.WithContext(ctx).Model(&Conversation{}). Where("id = ? AND message_count < ? AND is_closed = ?", id, max, false). UpdateColumn("message_count", gorm.Expr("message_count + 1")) if res.Error != nil { return false, errors.Wrap(res.Error, "increment message count if below") } return res.RowsAffected > 0, nil } // DecrementMessageCount 回滚一次消息计数(计数已递增但消息落库失败时调用)。 func (db *conversations) DecrementMessageCount(ctx context.Context, id uint) error { return db.WithContext(ctx).Model(&Conversation{}). Where("id = ? AND message_count > 0", id). UpdateColumn("message_count", gorm.Expr("message_count - 1")).Error } // MaybeClose 当消息数达到上限时关闭对话(幂等,供发送消息后调用)。 func (db *conversations) MaybeClose(ctx context.Context, id uint, max int) error { return db.WithContext(ctx).Model(&Conversation{}). Where("id = ? AND message_count >= ? AND is_closed = ?", id, max, false). Update("is_closed", true).Error } func (db *conversations) Close(ctx context.Context, id uint) error { return db.WithContext(ctx).Model(&Conversation{}).Where("id = ?", id).Update("is_closed", true).Error } func (db *conversations) ListByOwner(ctx context.Context, ownerID uint, limit, offset int) ([]*Conversation, error) { var convs []*Conversation q := db.WithContext(ctx).Where("owner_id = ? AND is_deleted_by_owner = ? AND token != ''", ownerID, false). Order("created_at DESC") if limit > 0 { q = q.Limit(limit) if offset > 0 { q = q.Offset(offset) } } if err := q.Find(&convs).Error; err != nil { return nil, errors.Wrap(err, "list by owner") } return convs, nil } func (db *conversations) CountByOwner(ctx context.Context, ownerID uint) (int64, error) { var count int64 if err := db.WithContext(ctx).Model(&Conversation{}). Where("owner_id = ? AND is_deleted_by_owner = ? AND token != ''", ownerID, false). Count(&count).Error; err != nil { return 0, errors.Wrap(err, "count by owner") } return count, nil } func (db *conversations) ListByAsker(ctx context.Context, askerID uint, limit, offset int) ([]*Conversation, error) { var convs []*Conversation q := db.WithContext(ctx).Where("asker_id = ? AND is_deleted_by_owner = ? AND token != ''", askerID, false). Order("created_at DESC") if limit > 0 { q = q.Limit(limit) if offset > 0 { q = q.Offset(offset) } } if err := q.Find(&convs).Error; err != nil { return nil, errors.Wrap(err, "list by asker") } return convs, nil } func (db *conversations) CountByAsker(ctx context.Context, askerID uint) (int64, error) { var count int64 if err := db.WithContext(ctx).Model(&Conversation{}). Where("asker_id = ? AND is_deleted_by_owner = ? AND token != ''", askerID, false). Count(&count).Error; err != nil { return 0, errors.Wrap(err, "count by asker") } return count, nil } func (db *conversations) MarkDeletedByOwner(ctx context.Context, id uint) error { var conv Conversation if err := db.WithContext(ctx).First(&conv, id).Error; err != nil { if errors.Is(err, gorm.ErrRecordNotFound) { // 已删除或不存在:幂等,不影响调用方。 return nil } return errors.Wrap(err, "get conversation for delete") } if conv.IsDeletedByOwner { // 已经是删除状态:幂等返回,避免重复请求多次回退配额计数。 return nil } // 用「条件更新 + RowsAffected」代替纯 Update: // 原来的写法是 check-then-act,双击「删除会话」或前端重复提交时,两个请求 // 都能读到 is_deleted_by_owner=false,于是都执行 UPDATE、都执行一次 // IncrementQuota(-1)。IncrementQuota 有 quota_count>0 兜底不会变负, // 但会白吃掉本用户其他会话的计数 —— 表现为定量收集提前关闭, // 而箱主重置计数也校准不回来。 res := db.WithContext(ctx).Model(&Conversation{}). Where("id = ? AND is_deleted_by_owner = ?", id, false). Update("is_deleted_by_owner", true) if res.Error != nil { return errors.Wrap(res.Error, "mark conversation deleted by owner") } if res.RowsAffected == 0 { // 并发里被另一个请求抢先删掉了:它已经做过配额回退,这里不能再来一次。 return nil } // 回退定量收集计数:删除私聊会话等价于少一条已收集内容, // 否则达上限后即便删掉旧会话,功能仍会显示 quota_reached 关闭,须手动重置计数。 if conv.OwnerID != 0 { // 用本 store 的 *gorm.DB 构造,避免依赖包级全局单例(详见 DeleteByID 注释)。 if err := NewFeatureControlsStore(db.DB).IncrementQuota(ctx, conv.OwnerID, FeatureChat, -1); err != nil { logrus.WithContext(ctx).WithError(err).Errorf("Failed to decrement chat quota for user %d", conv.OwnerID) } } return nil }