# Tinode 消息存储机制详解 > 本文档详细介绍 Tinode 消息的存储机制，包括数据结构、存储流程、数据库 Schema 和查询方式。 --- ## 目录 1. [存储架构概览](#存储架构概览) 2. [数据库表结构](#数据库表结构) 3. [消息数据结构](#消息数据结构) 4. [消息存储流程](#消息存储流程) 5. [消息内容格式](#消息内容格式) 6. [消息序号机制](#消息序号机制) 7. [消息删除机制](#消息删除机制) 8. [消息查询](#消息查询) 9. [各数据库实现差异](#各数据库实现差异) --- ## 存储架构概览 ### 整体架构 ``` ┌─────────────────────────────────────────────────────────────────────────────┐ │ 应用层 │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ Topic.handlePublish() -> 消息处理入口 │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 存储抽象层 │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ store.Messages.Save() -> 消息保存接口 │ │ │ │ store.Messages.GetAll() -> 消息查询接口 │ │ │ │ store.Messages.DeleteList() -> 消息删除接口 │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 数据库适配层 │ │ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ │ │ MySQL │ │PostgreSQL│ │ MongoDB │ │RethinkDB│ │ │ │ adapter │ │ adapter │ │ adapter │ │ adapter │ │ │ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 数据库层 │ │ ┌─────────────────────────────────────────────────────────────────────┐ │ │ │ messages 表/集合 - 存储消息内容 │ │ │ │ dellog 表/集合 - 记录删除操作 │ │ │ │ topics 表/集合 - 存储话题元数据 │ │ │ └─────────────────────────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────────────────────────┘ ``` ### 相关文件 | 文件 | 作用 | |------|------| | `server/store/types/types.go` | 消息数据结构定义 | | `server/store/store.go` | 存储抽象层实现 | | `server/db/adapter.go` | 数据库适配器接口 | | `server/db/mysql/adapter.go` | MySQL 实现 | | `server/db/postgres/adapter.go` | PostgreSQL 实现 | | `server/db/mongodb/adapter.go` | MongoDB 实现 | | `server/db/mysql/schema.sql` | MySQL 表结构 | | `server/topic.go` | 消息处理逻辑 | --- ## 数据库表结构 ### MySQL Schema ```sql -- 消息表 CREATE TABLE messages( id INT NOT NULL AUTO_INCREMENT, -- 自增主键 createdat DATETIME(3) NOT NULL, -- 创建时间（毫秒精度） updatedat DATETIME(3) NOT NULL, -- 更新时间 deletedat DATETIME(3), -- 删除时间（软删除） delid INT DEFAULT 0, -- 删除操作ID seqid INT NOT NULL, -- 消息序号（Topic内递增） topic CHAR(25) NOT NULL, -- Topic名称（如 grpXXX） `from` BIGINT NOT NULL, -- 发送者用户ID（数值） head JSON, -- 消息头（元数据） content JSON, -- 消息内容（JSON格式） PRIMARY KEY(id), UNIQUE INDEX messages_topic_seqid (topic, seqid) ); -- 删除日志表（记录软删除） CREATE TABLE dellog( id INT NOT NULL AUTO_INCREMENT, topic CHAR(25) NOT NULL, -- Topic名称 deletedfor BIGINT NOT NULL DEFAULT 0, -- 删除目标用户ID delid INT NOT NULL, -- 删除操作ID low INT NOT NULL, -- 删除范围起始seqid hi INT NOT NULL, -- 删除范围结束seqid（不含） PRIMARY KEY(id), FOREIGN KEY(topic) REFERENCES topics(name), INDEX dellog_topic_delid_deletedfor(topic, delid, deletedfor), INDEX dellog_topic_deletedfor_low_hi(topic, deletedfor, low, hi) ); ``` ### PostgreSQL Schema ```sql CREATE TABLE messages( id SERIAL PRIMARY KEY, createdat TIMESTAMP(3) NOT NULL, updatedat TIMESTAMP(3) NOT NULL, deletedat TIMESTAMP(3), delid INT DEFAULT 0, seqid INT NOT NULL, topic CHAR(25) NOT NULL, "from" BIGINT NOT NULL, head JSONB, content JSONB, UNIQUE INDEX messages_topic_seqid (topic, seqid) ); ``` ### MongoDB Schema ```javascript // messages 集合 { "_id": ObjectId("..."), // MongoDB ObjectId "createdat": ISODate("..."), "updatedat": ISODate("..."), "deletedat": ISODate("..."), "delid": NumberInt(0), "seqid": NumberInt(1), "topic": "grpABC123", "from": "usrXYZ456", "head": { "mime": "text/x-drafty", "attachments": ["file123"] }, "content": { "txt": "Hello World", "fmt": [...] } } // 索引 db.messages.createIndex({ "topic": 1, "seqid": 1 }, { unique: true }) ``` --- ## 消息数据结构 ### Go 结构体定义 ```go // server/store/types/types.go // Message 消息结构体 type Message struct { ObjHeader `bson:",inline"` // 基础头部（ID、时间戳） DeletedAt *time.Time `json:"DeletedAt,omitempty" bson:",omitempty"` DelId int `json:"DelId,omitempty" bson:",omitempty"` DeletedFor []SoftDelete `json:"DeletedFor,omitempty" bson:",omitempty"` SeqId int // 消息序号（Topic内唯一递增） Topic string // Topic名称 From string // 发送者UID（不带 usr 前缀） Head KVMap // 消息头（键值对元数据） Content any // 消息内容（任意JSON结构） } // ObjHeader 基础头部 type ObjHeader struct { Id string `json:"id,omitempty" bson:"_id,omitempty"` CreatedAt time.Time `json:"createdat,omitempty" bson:",omitempty"` UpdatedAt time.Time `json:"updatedat,omitempty" bson:",omitempty"` } // SoftDelete 软删除记录 type SoftDelete struct { User string `json:"user"` SeqId int `json:"seqid,omitempty"` } ``` ### 字段说明 | 字段 | 类型 | 说明 | |------|------|------| | `id` | string | 全局唯一ID（Snowflake算法） | | `createdat` | time.Time | 消息创建时间 | | `updatedat` | time.Time | 消息更新时间（编辑时） | | `deletedat` | *time.Time | 删除时间（硬删除时设置） | | `delid` | int | 删除操作ID（用于追踪删除） | | `seqid` | int | **消息序号**（Topic内递增，核心字段） | | `topic` | string | Topic名称（如 `grpABC123`） | | `from` | string | 发送者UID（如 `usrXYZ456`） | | `head` | KVMap | 消息头（元数据，如 MIME 类型、附件等） | | `content` | any | **消息内容**（任意 JSON 结构） | --- ## 消息存储流程 ### 完整流程图 ``` ┌─────────────────────────────────────────────────────────────────────────────┐ │ 1. 客户端发送消息 │ │ WebSocket: {"pub", "topic":"grpXXX", "content":{"text":"Hello"}} │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 2. WebSocket Handler 处理 (hdl_websock.go) │ │ serveWebSocket() -> sess.readLoop() -> sess.dispatchRaw(raw) │ │ 解析 JSON，识别消息类型为 "pub" │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 3. Session 消息分发 (session.go) │ │ dispatchRaw() -> 构造 ClientComMessage -> 发送到 Hub │ │ msg := &ClientComMessage{ │ │ Pub: &MsgClientPub{Topic: "grpXXX", Content: ...} │ │ } │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 4. Hub 路由分发 (hub.go) │ │ hub.routeCli <- msg │ │ hub.routeClientMessage(msg) -> 根据 msg.RcptTo 找到 Topic │ │ topic.clientMsg <- msg // 发送到 Topic 的消息通道 │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 5. Topic 处理消息 (topic.go) │ │ select { │ │ case msg := <-t.clientMsg: │ │ t.processClientMessage(msg) │ │ } │ │ -> t.handlePublish(msg) │ │ -> t.saveAndBroadcastMessage(msg, ...) │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 6. 保存前检查 (topic.go:saveAndBroadcastMessage) │ │ a. 检查发送者是否有写权限 (IsWriter) │ │ b. 处理消息头 (sender、attachments) │ │ c. 构造 types.Message 对象 │ │ SeqId = t.lastID + 1 // 序号递增 │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 7. 存储层保存 (store/store.go:Messages.Save) │ │ a. msg.InitTimes() // 初始化时间戳 │ │ b. msg.SetUid(Store.GetUid()) // 生成全局唯一ID │ │ c. adp.TopicUpdateOnMessage(msg.Topic, msg) // 更新Topic的seqid │ │ d. adp.MessageSave(msg) // 插入消息到数据库 │ │ e. adp.SubsUpdate() // 更新发送者的 ReadSeqId/RecvSeqId │ │ f. adp.FileLinkAttachments() // 关联附件文件 │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 8. 数据库写入 (db/mysql/adapter.go:MessageSave) │ │ INSERT INTO messages(createdAt, updatedAt, seqid, topic, `from`, │ │ head, content) │ │ VALUES(?, ?, ?, ?, ?, ?, ?) │ │ │ │ 参数: │ │ - createdAt: msg.CreatedAt │ │ - updatedAt: msg.UpdatedAt │ │ - seqid: msg.SeqId (t.lastID + 1) │ │ - topic: msg.Topic │ │ - from: store.DecodeUid(ParseUid(msg.From)) │ │ - head: msg.Head (JSON) │ │ - content: ToJSON(msg.Content) (JSON) │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 9. 更新 Topic 缓存 (topic.go) │ │ t.lastID++ // 递增最后消息序号 │ │ t.touched = msg.Timestamp // 更新最后消息时间 │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 10. 广播给订阅者 (topic.go) │ │ for sess := range t.sessions { │ │ sess.queueOut(&ServerComMessage{ │ │ Data: &MsgServerData{ │ │ Topic: t.name, │ │ SeqId: t.lastID, │ │ From: asUid.UserId(), │ │ Content: content, │ │ }, │ │ }) │ │ } │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 11. 推送通知 (push/push.go) │ │ 如果订阅者离线，发送 FCM 推送通知 │ │ push.Push(userId, &PushData{...}) │ └─────────────────────────────────────────────────────────────────────────────┘ ``` ### 关键代码解析 #### 消息保存入口 (topic.go:999) ```go func (t *Topic) saveAndBroadcastMessage(msg *ClientComMessage, asUid types.Uid, noEcho bool, attachments []string, head map[string]any, content any) error { pud, userFound := t.perUser[asUid] // 1. 权限检查 if t.cat != types.TopicCatSys { if !(pud.modeWant & pud.modeGiven).IsWriter() { msg.sess.queueOut(ErrPermissionDenied(msg.Id, t.original(asUid), msg.Timestamp)) return types.ErrPermissionDenied } } // 2. 处理 sender 头（代理发送） if msg.sess != nil && msg.sess.uid != asUid { if head == nil { head = map[string]any{} } head["sender"] = msg.sess.uid.UserId() } // 3. 构造消息对象并保存 if err, _ := store.Messages.Save( &types.Message{ ObjHeader: types.ObjHeader{CreatedAt: msg.Timestamp}, SeqId: t.lastID + 1, // 关键：序号递增 Topic: t.name, From: asUid.String(), Head: head, Content: content, }, attachments, (pud.modeGiven & pud.modeWant).IsReader()); err != nil { return err } // 4. 更新 Topic 状态 t.lastID++ // 5. 广播给订阅者 // ... } ``` #### 存储层实现 (store/store.go:682) ```go func (messagesMapper) Save(msg *types.Message, attachmentURLs []string, readBySender bool) (error, bool) { // 1. 初始化时间戳 msg.InitTimes() // 2. 生成全局唯一ID（Snowflake算法） msg.SetUid(Store.GetUid()) // 3. 更新 Topic 的 SeqId（数据库层） err := adp.TopicUpdateOnMessage(msg.Topic, msg) if err != nil { return err, false } // 4. 插入消息到数据库 err = adp.MessageSave(msg) if err != nil { return err, false } // 5. 标记发送者已读 markedReadBySender := false if readBySender { fromUid := types.ParseUid(msg.From) if !fromUid.IsZero() { if err := adp.SubsUpdate(msg.Topic, fromUid, map[string]any{ "RecvSeqId": msg.SeqId, "ReadSeqId": msg.SeqId, }); err == nil { markedReadBySender = true } } } // 6. 关联附件文件 if len(attachmentURLs) > 0 { var attachments []string for _, url := range attachmentURLs { if fid := mediaHandler.GetIdFromUrl(url); !fid.IsZero() { attachments = append(attachments, fid.String()) } } if len(attachments) > 0 { return adp.FileLinkAttachments("", types.ZeroUid, msg.Uid(), attachments), markedReadBySender } } return nil, markedReadBySender } ``` #### MySQL 适配器实现 (db/mysql/adapter.go:2661) ```go func (a *adapter) MessageSave(msg *t.Message) error { ctx, cancel := a.getContext() if cancel != nil { defer cancel() } // 执行插入 res, err := a.db.ExecContext(ctx, `INSERT INTO messages(createdAt, updatedAt, seqid, topic, `+"`from`"+`, head, content) VALUES(?, ?, ?, ?, ?, ?, ?)`, msg.CreatedAt, // createdat msg.UpdatedAt, // updatedat msg.SeqId, // seqid msg.Topic, // topic store.DecodeUid(t.ParseUid(msg.From)), // from (数值UID) msg.Head, // head (JSON) common.ToJSON(msg.Content)) // content (JSON) return err } ``` --- ## 消息内容格式 ### 1. 纯文本消息 ```json { "content": "Hello World" } ``` ### 2. Drafty 富文本消息 Drafty 是 Tinode 的富文本格式： ```json { "content": { "txt": "Hello **World**", "fmt": [ {"tp": "SB", "at": 6, "len": 7} ] } } ``` **格式类型**: | tp | 说明 | |----|------| | `ST` | 斜体 | | `SB` | 粗体 | | `CO` | 代码 | | `BR` | 换行 | | `LN` | 链接 | | `MN` | 提及 (@用户) | | `HT` | 隐藏文本 | ### 3. 带附件消息 ```json { "content": { "txt": "Check this image", "fmt": [ { "tp": "IM", "at": 6, "len": 1, "url": "https://example.com/image.jpg", "mime": "image/jpeg", "size": 12345, "width": 800, "height": 600 } ] } } ``` ### 4. 回复/转发消息 ```json { "content": { "txt": "Replying to...", "replyTo": { "seqid": 42, "from": "usrABC", "content": {"txt": "Original message"} } } } ``` ### 5. 消息头 消息头用于存储元数据： ```json { "head": { "mime": "text/x-drafty", // 内容类型 "attachments": ["file123"], // 附件ID列表 "replyTo": 42, // 回复的消息序号 "sender": "usrXYZ", // 代理发送者（如果是代理发送） "replace": 41 // 替换的消息序号（编辑消息） } } ``` --- ## 消息序号机制 ### SeqId 的作用 `seqid` 是消息在 Topic 内的递增序号，是消息同步的核心： ``` Topic: grpABC123 ┌────────┬──────────────┬─────────────────────────────────┐ │ seqid │ from │ content │ ├────────┼──────────────┼─────────────────────────────────┤ │ 1 │ usrAlice │ {"text": "Hello everyone"} │ │ 2 │ usrBob │ {"text": "Hi Alice"} │ │ 3 │ usrCarol │ {"text": "Hey all"} │ │ 4 │ usrAlice │ {"text": "How are you?"} │ │ ... │ ... │ ... │ └────────┴──────────────┴─────────────────────────────────┘ ``` ### SeqId 的用途 #### 1. 消息同步 客户端通过 `since` 和 `before` 参数拉取消息： ```json // 客户端请求 { "get": { "topic": "grpABC123", "data": { "since": 2, "before": 5, "limit": 10 } } } // 服务端返回 seqid >= 2 且 < 5 的消息 ``` #### 2. 已读回执 订阅者表中记录已读位置： ```sql -- subscriptions 表 recvseqid = 3 -- 已收到消息序号 readseqid = 2 -- 已读消息序号 ``` ```json // 客户端发送已读通知 { "note": { "topic": "grpABC123", "what": "read", "seqid": 4 } } ``` #### 3. 未读消息计数 ```go unreadCount = topic.lastID - subscription.readSeqId ``` #### 4. 消息删除范围 ```json // 删除 seqid 3-5 的消息 { "del": { "topic": "grpABC123", "what": "data", "ranges": [{"low": 3, "hi": 6}] } } ``` ### SeqId 分配 ```go // topic.go func (t *Topic) saveAndBroadcastMessage(...) { message.SeqId = t.lastID + 1 // Topic 内递增 t.lastID++ // 更新缓存 } // 数据库层保证唯一性 UNIQUE INDEX messages_topic_seqid (topic, seqid) ``` --- ## 消息删除机制 Tinode 支持两种删除方式：**软删除**和**硬删除**。 ### 1. 软删除 只对当前用户隐藏消息，其他用户仍可见。 #### 数据库记录 ```sql -- dellog 表记录删除操作 INSERT INTO dellog(topic, deletedfor, delid, low, hi) VALUES ('grpABC123', 123, 1, 3, 6); -- 删除 seqid [3, 6) 的消息 ``` #### 查询时过滤 ```sql -- 查询消息时排除已删除的 SELECT m.* FROM messages m LEFT JOIN dellog d ON m.topic = d.topic AND m.seqid >= d.low AND m.seqid < d.hi AND d.deletedfor = ? WHERE m.topic = ? AND (d.id IS NULL) -- 未被软删除 ORDER BY m.seqid DESC LIMIT ? ``` #### 软删除流程 ``` 用户 A 删除消息 │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 1. 检查权限 │ │ - 用户是否有删除权限（D权限） │ │ - 消息是否过期（msg_delete_age 配置） │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 2. 生成删除记录 │ │ dellog{ │ │ topic: "grpABC123", │ │ deletedfor: usrA, │ │ delid: topic.delID++, │ │ low: 3, hi: 6 │ │ } │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 3. 写入数据库 │ │ INSERT INTO dellog(topic, deletedfor, delid, low, hi) │ │ 更新 subscription.delId │ └─────────────────────────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────────────────────────┐ │ 4. 广播删除通知 │ │ {pres: "del", seqid: 3, delid: 1} │ └─────────────────────────────────────────────────────────────────────────────┘ ``` ### 2. 硬删除 物理删除消息，所有用户都不可见。 #### 数据库更新 ```sql -- 标记消息为已删除 UPDATE messages SET deletedat = NOW(), delid = ? WHERE topic = ? AND seqid BETWEEN ? AND ?; ``` #### 硬删除条件 只有以下情况可以硬删除： - Topic 所有者 - 消息发送者（在一定时间内） - `forUser` 为零（管理员操作） ### 3. 删除范围 删除操作支持范围： ```json // 删除单条消息 { "del": { "topic": "grpABC123", "what": "data", "ranges": [{"low": 5, "hi": 0}] // hi=0 表示单条 } } // 删除范围消息 { "del": { "topic": "grpABC123", "what": "data", "ranges": [ {"low": 1, "hi": 5}, // 删除 seqid [1, 5) {"low": 10, "hi": 15} // 删除 seqid [10, 15) ] } } // 删除所有消息 { "del": { "topic": "grpABC123", "what": "data", "ranges": [{"low": 1, "hi": 2147483647}] // 删除全部 } } ``` --- ## 消息查询 ### 查询接口 ```go // store/store.go func (messagesMapper) GetAll(topic string, forUser types.Uid, opt *types.QueryOpt) ([]types.Message, error) ``` ### 查询参数 ```go type QueryOpt struct { Since int // 起始 seqid（包含） Before int // 结束 seqid（不包含） Limit int // 返回数量限制 IdRanges []Range // 指定范围列表 IfModifiedSince *time.Time // 增量查询时间戳 } ``` ### 查询示例 #### 1. 拉取最新消息 ```json { "get": { "topic": "grpABC123", "data": { "limit": 20 } } } ``` ```sql SELECT * FROM messages WHERE topic = 'grpABC123' ORDER BY seqid DESC LIMIT 20; ``` #### 2. 拉取历史消息 ```json { "get": { "topic": "grpABC123", "data": { "before": 50, "limit": 20 } } } ``` ```sql SELECT * FROM messages WHERE topic = 'grpABC123' AND seqid < 50 ORDER BY seqid DESC LIMIT 20; ``` #### 3. 增量同步 ```json { "get": { "topic": "grpABC123", "data": { "since": 100, "limit": 50 } } } ``` ```sql SELECT * FROM messages WHERE topic = 'grpABC123' AND seqid >= 100 ORDER BY seqid ASC LIMIT 50; ``` #### 4. 按范围查询 ```json { "get": { "topic": "grpABC123", "data": { "ranges": [ {"low": 10, "hi": 15}, {"low": 20, "hi": 25} ] } } } ``` ```sql SELECT * FROM messages WHERE topic = 'grpABC123' AND ((seqid >= 10 AND seqid < 15) OR (seqid >= 20 AND seqid < 25)) ORDER BY seqid; ``` --- ## 各数据库实现差异 ### 存储方式对比 | 特性 | MySQL | PostgreSQL | MongoDB | RethinkDB | |------|-------|------------|---------|-----------| | 内容类型 | JSON | JSONB | BSON | 原生JSON | | 主键 | AUTO_INCREMENT | SERIAL | ObjectId | UUID | | 索引 | B+Tree | B+Tree | B+Tree | B+Tree | | 事务 | 支持 | 支持 | 支持 | 支持 | | 分布式 | 否 | 否 | 分片 | 分片 | ### MySQL 适配器 ```go // db/mysql/adapter.go func (a *adapter) MessageSave(msg *t.Message) error { _, err := a.db.ExecContext(ctx, `INSERT INTO messages(createdAt, updatedAt, seqid, topic, `+"`from`"+`, head, content) VALUES(?, ?, ?, ?, ?, ?, ?)`, msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic, store.DecodeUid(t.ParseUid(msg.From)), msg.Head, common.ToJSON(msg.Content)) return err } ``` ### PostgreSQL 适配器 ```go // db/postgres/adapter.go func (a *adapter) MessageSave(msg *t.Message) error { var id int err := a.db.QueryRow(ctx, `INSERT INTO messages(createdAt, updatedAt, seqid, topic, "from", head, content) VALUES($1, $2, $3, $4, $5, $6, $7) RETURNING id`, msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic, store.DecodeUid(t.ParseUid(msg.From)), msg.Head, common.ToJSON(msg.Content)).Scan(&id) return err } ``` ### MongoDB 适配器 ```go // db/mongodb/adapter.go func (a *adapter) MessageSave(msg *t.Message) error { // 直接插入整个结构体 _, err := a.db.Collection("messages").InsertOne(a.ctx, msg) return err } ``` ### 性能优化 #### 索引设计 ```sql -- MySQL/PostgreSQL CREATE UNIQUE INDEX messages_topic_seqid ON messages(topic, seqid); CREATE INDEX messages_topic_created ON messages(topic, createdat); ``` ```javascript // MongoDB db.messages.createIndex({ "topic": 1, "seqid": 1 }, { unique: true }) db.messages.createIndex({ "topic": 1, "createdat": -1 }) ``` #### 分区策略 对于大流量场景，可以考虑按 Topic 分区： ```sql -- MySQL 分区（按 topic hash） CREATE TABLE messages ( ... ) PARTITION BY HASH(topic) PARTITIONS 16; ``` --- ## 常见问题 ### 1. 消息 ID vs SeqId 的区别？ - **id**: 全局唯一，使用 Snowflake 算法生成，用于数据库主键 - **seqid**: Topic 内唯一递增，用于消息同步、已读回执、删除范围 ### 2. 如何实现消息编辑？ 编辑消息实际上是创建新消息并关联原消息： ```json { "pub": { "topic": "grpABC123", "content": {"txt": "Edited content"}, "head": { "replace": 42 // 替换 seqid=42 的消息 } } } ``` ### 3. 如何实现消息撤回？ 在一定时间内（`msg_delete_age` 配置）可以撤回： ```json { "del": { "topic": "grpABC123", "what": "data", "ranges": [{"low": 42, "hi": 0}] } } ``` ### 4. 消息存储限制？ - 最大消息大小：由 `max_message_size` 配置（默认 128KB） - 大文件通过附件存储，消息中只保留引用 --- ## 总结 Tinode 消息存储的核心设计： 1. **JSON 格式存储** - 灵活支持各种消息类型 2. **SeqId 递增** - 每个 Topic 内消息有序，便于同步 3. **Topic 分片** - 消息按 Topic 分组，便于查询 4. **软/硬删除** - 支持两种删除策略 5. **多数据库支持** - 统一接口，适配多种数据库