package model import ( "database/sql/driver" "encoding/json" "errors" "strconv" "sync" "time" "gorm.io/gorm" "with_you/internal/pkg/crypto" "with_you/internal/pkg/utils" ) // 系统消息相关常量 const ( // SystemSenderID 系统消息发送者ID (类似QQ 10000) SystemSenderID int64 = 10000 // SystemSenderIDStr 系统消息发送者ID字符串版本 SystemSenderIDStr string = "10000" // SystemConversationID 系统通知会话ID (string类型) SystemConversationID string = "9999999999" ) // ContentType 消息内容类型 type ContentType string const ( ContentTypeText ContentType = "text" ContentTypeImage ContentType = "image" ContentTypeVideo ContentType = "video" ContentTypeAudio ContentType = "audio" ContentTypeFile ContentType = "file" ) // MessageStatus 消息状态 type MessageStatus string const ( MessageStatusNormal MessageStatus = "normal" // 正常 MessageStatusRecalled MessageStatus = "recalled" // 已撤回 MessageStatusDeleted MessageStatus = "deleted" // 已删除 ) // MessageCategory 消息类别 type MessageCategory string const ( CategoryChat MessageCategory = "chat" // 普通聊天 CategoryNotification MessageCategory = "notification" // 通知类消息 CategoryAnnouncement MessageCategory = "announcement" // 系统公告 CategoryMarketing MessageCategory = "marketing" // 营销消息 ) // SystemMessageType 系统消息类型 (对应原NotificationType) type SystemMessageType string const ( // 互动通知 SystemTypeLikePost SystemMessageType = "like_post" // 点赞帖子 SystemTypeLikeComment SystemMessageType = "like_comment" // 点赞评论 SystemTypeComment SystemMessageType = "comment" // 评论 SystemTypeReply SystemMessageType = "reply" // 回复 SystemTypeFollow SystemMessageType = "follow" // 关注 SystemTypeMention SystemMessageType = "mention" // @提及 // 系统消息 SystemTypeSystem SystemMessageType = "system" // 系统通知 SystemTypeAnnounce SystemMessageType = "announce" // 系统公告 ) // ExtraData 消息额外数据,用于存储系统消息的相关信息 type ExtraData struct { // 操作者信息 ActorID int64 `json:"actor_id,omitempty"` // 操作者ID (数字格式,兼容旧数据) ActorIDStr string `json:"actor_id_str,omitempty"` // 操作者ID (UUID字符串格式) ActorName string `json:"actor_name,omitempty"` // 操作者名称 AvatarURL string `json:"avatar_url,omitempty"` // 操作者头像 // 目标信息 TargetID int64 `json:"target_id,omitempty"` // 目标ID(帖子ID、评论ID等) TargetTitle string `json:"target_title,omitempty"` // 目标标题 TargetType string `json:"target_type,omitempty"` // 目标类型(post/comment等) // 其他信息 ActionURL string `json:"action_url,omitempty"` // 跳转链接 ActionTime string `json:"action_time,omitempty"` // 操作时间 } // Value 实现driver.Valuer接口,用于数据库存储 func (e ExtraData) Value() (driver.Value, error) { return json.Marshal(e) } // Scan 实现sql.Scanner接口,用于数据库读取 func (e *ExtraData) Scan(value any) error { if value == nil { return nil } var data []byte switch v := value.(type) { case []byte: data = v case string: data = []byte(v) default: return errors.New("type assertion to []byte or string failed") } return json.Unmarshal(data, e) } // MessageSegmentData 单个消息段的数据 type MessageSegmentData map[string]any // MessageSegment 消息段 type MessageSegment struct { Type string `json:"type"` Data MessageSegmentData `json:"data"` } // MessageSegments 消息链类型 type MessageSegments []MessageSegment // Value 实现driver.Valuer接口,用于数据库存储 func (s MessageSegments) Value() (driver.Value, error) { return json.Marshal(s) } // Scan 实现sql.Scanner接口,用于数据库读取 func (s *MessageSegments) Scan(value any) error { if value == nil { *s = nil return nil } var data []byte switch v := value.(type) { case []byte: data = v case string: data = []byte(v) default: return errors.New("type assertion to []byte or string failed") } return json.Unmarshal(data, s) } // Message 消息实体 // 使用雪花算法ID(string类型)和seq机制实现消息排序和增量同步 // 支持消息内容加密存储 type Message struct { ID string `gorm:"primaryKey;size:20" json:"id"` // 雪花算法ID(string类型) ConversationID string `gorm:"not null;size:20;index:idx_msg_conversation_seq,priority:1" json:"conversation_id"` // 会话ID(string类型) SenderID string `gorm:"column:sender_id;type:varchar(50);index;not null" json:"sender_id"` // 发送者ID (UUID格式) Seq int64 `gorm:"not null;index:idx_msg_conversation_seq,priority:2" json:"seq"` // 会话内序号,用于排序和增量同步 // 消息内容字段 Segments MessageSegments `gorm:"-" json:"segments"` // 消息链(内存中使用,不映射到数据库) SegmentsEncrypted string `gorm:"column:segments;type:text" json:"-"` // 加密后的消息链(存储在数据库) SegmentsKeyVersion int `gorm:"column:segments_key_version;default:0" json:"segments_key_version"` // 密钥版本,用于密钥轮换 ReplyToID *string `json:"reply_to_id,omitempty"` // 回复的消息ID(string类型) Status MessageStatus `gorm:"type:varchar(20);default:'normal'" json:"status"` // 消息状态 // 新增字段:消息分类和系统消息类型 Category MessageCategory `gorm:"type:varchar(20);default:'chat'" json:"category"` // 消息分类 SystemType SystemMessageType `gorm:"type:varchar(30)" json:"system_type,omitempty"` // 系统消息类型 ExtraData *ExtraData `gorm:"type:json" json:"extra_data,omitempty"` // 额外数据(JSON格式) // 软删除 DeletedAt gorm.DeletedAt `json:"-" gorm:"index"` // 时间戳 CreatedAt time.Time `json:"created_at" gorm:"autoCreateTime"` UpdatedAt time.Time `json:"updated_at" gorm:"autoUpdateTime"` // 内部标记,表示是否已解密 decrypted bool `gorm:"-"` } // SenderIDStr 返回发送者ID字符串(保持兼容性) func (m *Message) SenderIDStr() string { return m.SenderID } // BeforeCreate 创建前生成雪花算法ID并加密消息内容 func (m *Message) BeforeCreate(tx *gorm.DB) error { if m.ID == "" { id, err := utils.GetSnowflake().GenerateID() if err != nil { return err } m.ID = strconv.FormatInt(id, 10) } // 加密消息内容 return m.encryptSegments() } // BeforeUpdate 更新前加密消息内容 func (m *Message) BeforeUpdate(tx *gorm.DB) error { // 只有当 Segments 被修改时才重新加密 if tx.Statement.Changed("Segments") || len(m.Segments) > 0 { return m.encryptSegments() } return nil } // AfterFind 查询后标记需要解密(实际解密延迟到批量处理时) // 这样可以支持批量并行解密优化 func (m *Message) AfterFind(tx *gorm.DB) error { // 标记为未解密状态,等待批量解密 // 如果是单条查询,AfterFind 后会立即使用,需要解密 m.decrypted = false return nil } // Decrypt 解密单条消息(显式调用) func (m *Message) Decrypt() error { return m.decryptSegments() } // AfterCreate 创建后解密消息内容(保持内存中的数据可读) func (m *Message) AfterCreate(tx *gorm.DB) error { m.decrypted = true return nil } // encryptSegments 加密消息段 func (m *Message) encryptSegments() error { // 如果没有消息内容或加密器未初始化,跳过加密 if len(m.Segments) == 0 { return nil } encryptor := crypto.GetMessageEncryptor() if encryptor == nil { // 加密器未初始化时,直接存储JSON(兼容模式) data, err := json.Marshal(m.Segments) if err != nil { return err } m.SegmentsEncrypted = string(data) m.SegmentsKeyVersion = 0 return nil } // 序列化消息段 plaintext, err := json.Marshal(m.Segments) if err != nil { return err } // 加密 ciphertext, err := encryptor.Encrypt(plaintext) if err != nil { return err } m.SegmentsEncrypted = ciphertext m.SegmentsKeyVersion = encryptor.GetKeyVersion() return nil } // decryptSegments 解密消息段 func (m *Message) decryptSegments() error { // 如果已经解密过,跳过 if m.decrypted { return nil } // 如果没有加密内容,跳过 if m.SegmentsEncrypted == "" { m.decrypted = true return nil } encryptor := crypto.GetMessageEncryptor() if encryptor == nil { // 加密器未初始化,尝试直接解析JSON(兼容旧数据) if err := json.Unmarshal([]byte(m.SegmentsEncrypted), &m.Segments); err != nil { return err } m.decrypted = true return nil } // 尝试解密 plaintext, err := encryptor.Decrypt(m.SegmentsEncrypted) if err != nil { // 解密失败,可能是未加密的旧数据,尝试直接解析 if parseErr := json.Unmarshal([]byte(m.SegmentsEncrypted), &m.Segments); parseErr != nil { return err // 返回解密错误 } m.decrypted = true return nil } // 解析解密后的数据 if err := json.Unmarshal(plaintext, &m.Segments); err != nil { return err } m.decrypted = true return nil } func (Message) TableName() string { return "messages" } // IsSystemMessage 判断是否为系统消息 func (m *Message) IsSystemMessage() bool { return m.SenderID == SystemSenderIDStr || m.Category == CategoryNotification || m.Category == CategoryAnnouncement } // IsInteractionNotification 判断是否为互动通知 func (m *Message) IsInteractionNotification() bool { if m.Category != CategoryNotification { return false } switch m.SystemType { case SystemTypeLikePost, SystemTypeLikeComment, SystemTypeComment, SystemTypeReply, SystemTypeFollow, SystemTypeMention: return true default: return false } } // BatchDecryptMessages 批量并行解密消息 // 用于批量查询后提高解密性能,比逐条调用 AfterFind 更高效 // workerCount 指定并行工作数,如果 <= 0 则使用默认值 func BatchDecryptMessages(messages []*Message, workerCount int) error { if len(messages) == 0 { return nil } encryptor := crypto.GetMessageEncryptor() if encryptor == nil { // 加密器未初始化,直接解析 JSON(兼容模式) for _, m := range messages { if m.decrypted || m.SegmentsEncrypted == "" { m.decrypted = true continue } if err := json.Unmarshal([]byte(m.SegmentsEncrypted), &m.Segments); err != nil { return err } m.decrypted = true } return nil } // 小批量直接串行处理 if len(messages) <= 10 { for _, m := range messages { if err := m.decryptSegments(); err != nil { return err } } return nil } // 并行解密 // 先收集需要解密的密文 ciphertexts := make([]string, len(messages)) for i, m := range messages { ciphertexts[i] = m.SegmentsEncrypted } // 批量并行解密 plaintexts := encryptor.BatchDecrypt(ciphertexts, workerCount) // 解析结果 for i, m := range messages { if m.decrypted { continue } if m.SegmentsEncrypted == "" { m.decrypted = true continue } plaintext := plaintexts[i] if plaintext == nil { // 解密失败,尝试直接解析(兼容旧数据) if err := json.Unmarshal([]byte(m.SegmentsEncrypted), &m.Segments); err != nil { continue // 忽略单条错误 } } else { if err := json.Unmarshal(plaintext, &m.Segments); err != nil { continue // 忽略单条错误 } } m.decrypted = true } return nil } // BatchDecryptMessagesParallel 使用 worker pool 并行解密 // 这是一个更高效的版本,适用于大量消息 func BatchDecryptMessagesParallel(messages []*Message) { if len(messages) == 0 { return } encryptor := crypto.GetMessageEncryptor() if encryptor == nil { // 加密器未初始化,串行解析 for _, m := range messages { if !m.decrypted && m.SegmentsEncrypted != "" { _ = json.Unmarshal([]byte(m.SegmentsEncrypted), &m.Segments) m.decrypted = true } } return } // 确定并行度 workerCount := 4 if len(messages) < 20 { workerCount = 2 } else if len(messages) > 100 { workerCount = 8 } jobs := make(chan int, len(messages)) var wg sync.WaitGroup // 启动 workers for w := 0; w < workerCount; w++ { wg.Add(1) go func() { defer wg.Done() for i := range jobs { m := messages[i] if m.decrypted || m.SegmentsEncrypted == "" { m.decrypted = true continue } plaintext, err := encryptor.Decrypt(m.SegmentsEncrypted) if err != nil { // 尝试直接解析 _ = json.Unmarshal([]byte(m.SegmentsEncrypted), &m.Segments) } else { _ = json.Unmarshal(plaintext, &m.Segments) } m.decrypted = true } }() } // 分发任务 for i := range messages { jobs <- i } close(jobs) wg.Wait() }