实现 TCP READ（命令 8），已读水位前进时清未读并通知其他设备。

补上会员价格按平台筛选，并修正广场机器人测试的 Feed 调用，使 go test 能通过。

Co-authored-by: Cursor <cursoragent@cursor.com>

README.md

@@ -185,7 +185,7 @@ offset 23  Body       bytes   UTF-8 JSON
 | 5 | SEND_ACK | ← 仅 4 的成功应答 |
 | 6 | MESSAGE | ← 推送 |
 | 7 | PUSH_ACK | → 空；`SequenceID` = 推送帧的 SequenceID |
-| 8 | READ_MESSAGE | → |
+| 8 | READ_MESSAGE | → `{"conversation_id","msg_id"}`；成功 `data` 同名字段，`msg_id` 为落库后的 `last_read_msg_id`（只增不减）。清该成员 Redis 未读，并把会话从 unread ZSET 挪到 read ZSET。水位前进时给该用户其他在线设备推 Command 8（不系统推送、不写 mailbox）。缺 `conversation_id` 或 `msg_id` 为 `40001` |
 | 9 | RECALL_MESSAGE | → `{"msg_id"}` |
 | 10 | SYNC_INBOX | → 会话列表增量 |
 | 12 | SYNC_MESSAGES | → 某会话消息增量 |

docs/architecture-v3.md

@@ -647,6 +647,16 @@ PUSH_ACK    传输层，不落业务库
 READ        更新 last_read_msg_id，清 Redis 未读，入队通知其他设备
 ```
 
+请求（命令 8，业务 JSON，无信封）：
+
+```json
+{"conversation_id": 1001, "msg_id": 2002}
+```
+
+`msg_id` 是成员已读到的消息 id。成功应答仍是命令 8，Body 为 V3 信封，`data.msg_id` 为写入后的 `last_read_msg_id`（只增不减；更小的 id 不回退）。缺 `conversation_id` 或 `msg_id`（≤0）返回 `40001`。消息不存在为 `40401`；消息不属于该会话为 `40001`；不是成员为 `40301`。
+
+水位前进时入队：`ToUserIDs` 仅为当前用户，`ExcludeDevice` 为当前设备，`NotPush` 与 `NotMailbox` 均为真。其他在线设备收到命令 8。
+
 不做业务 DELIVERY_ACK。
 
 已读：MySQL `last_read_msg_id` + Redis `HDEL unreadcnt` + 从 unread ZSET 挪到 read ZSET。

docs/daily/2026-10-16-change-list.md

@@ -0,0 +1,36 @@
+# 每日变动 2026-10-16
+
+## 18:15 — TCP READ（命令 8）已读
+
+### 概要
+
+- 注册 TCP 命令 8。请求带 `conversation_id` 与 `msg_id` 时更新 `last_read_msg_id`（只增不减），清 Redis 未读并把会话从 unread ZSET 挪到 read ZSET。
+- 水位前进时入队通知该用户其他在线设备（不系统推送、不写 mailbox）。缺必填字段仍返回 `40001`；合法请求不再无条件 `40001`。
+- 补齐当前 `main` 上会员价格按平台筛选的缺失函数，并修正广场机器人测试对 `Feed` 的调用，否则整包测试无法编译。
+
+### 变更范围
+
+| 类型 | 说明 |
+|------|------|
+| 功能 | TCP READ / MarkRead |
+| 修复 | 客户端订阅计划编译、squarebot 测试签名 |
+| 文档 | README、架构、OpenAPI 写明命令 8 字段 |
+
+### 主要文件
+
+- `internal/module/message/logic/read.go` — 已读水位、清未读、通知其他设备
+- `internal/transport/tcp/handlers.go` — 注册命令 8
+- `internal/module/inbox/model/redis.go` — `ClearUnread`
+- `internal/module/conversation/model/conversation_member.go` — `UpdateLastRead`
+- `internal/module/client/logic/plans.go` — 包名归一与平台筛选
+- `README.md`、`docs/architecture-v3.md`、`internal/transport/http/docs/openapi.yaml`
+
+### 验证
+
+- `go test ./...`：pass
+- logic 覆盖率均 ≥ 65%（message/logic 69.5%）
+- `make test`：本机无 `make`；golangci-lint v1.64.8 无法类型检查 Go 1.27 标准库，lint 未通过
+
+### 关联
+
+- 修复 TCP READ 并发布到新加坡

internal/bootstrap/app.go

@@ -208,7 +208,7 @@ func finishApp(cfg *Config, log *zap.Logger, logClose func() error, idgen *id.Ge
 	friends := wireFriends(idgen, db, users, deliverQueue, chat)
 	groups := wireGroups(idgen, db, users, friends, deliverQueue, objects, chat)
 	moderation := wireModeration(db, users, idgen)
-	messages := wireMessages(idgen, friends, groups, deliverQueue, chat, moderation, users)
+	messages := wireMessages(idgen, friends, groups, deliverQueue, chat, moderation, users, inbox)
 	files := wireFiles(cfg, log, objects, groups)
 	favorites := wireFavorites(idgen, db, objects, chat)
 	moments := wireMoments(idgen, db, objects, friends, users)
@@ -520,11 +520,13 @@ func wireMessages(
 	chat chatStores,
 	moderation *modlogic.Service,
 	users *userlogic.Service,
+	inbox inboxlogic.Store,
 ) *msglogic.Service {
 	if idgen == nil {
 		return nil
 	}
 	svc := msglogic.NewService(chat.msgs, chat.chats, chat.p2p, friends, groups, idgen, q)
+	svc.BindInbox(inbox)
 	if moderation != nil && users != nil {
 		svc.BindModeration(users, moderation, moderation)
 	}

internal/module/client/logic/plans.go

@@ -0,0 +1,63 @@
+package logic
+
+import (
+	"context"
+	"strings"
+
+	"spark-server/internal/module/client/entity"
+	"spark-server/internal/pkg/i18n"
+)
+
+func canonicalHealthPackage(pkg string) string {
+	return entity.CanonicalHealthPackage(pkg)
+}
+
+func normalizePlatform(platform string) string {
+	switch strings.ToLower(strings.TrimSpace(platform)) {
+	case entity.PlatformIOS, "iphone", "ipad":
+		return entity.PlatformIOS
+	case entity.PlatformAndroid:
+		return entity.PlatformAndroid
+	case entity.PlatformWeb:
+		return entity.PlatformWeb
+	default:
+		return ""
+	}
+}
+
+func filterPlansByPlatform(items []entity.SubscriptionPlan, platform string) []entity.SubscriptionPlan {
+	platform = normalizePlatform(platform)
+	if platform == "" || len(items) == 0 {
+		return items
+	}
+	exact := make([]entity.SubscriptionPlan, 0, len(items))
+	fallback := make([]entity.SubscriptionPlan, 0, len(items))
+	for _, p := range items {
+		got := normalizePlatform(p.Platform)
+		if got == platform {
+			exact = append(exact, p)
+		} else if got == "" {
+			fallback = append(fallback, p)
+		}
+	}
+	if len(exact) > 0 {
+		return exact
+	}
+	return fallback
+}
+
+func (s *Service) clientPlans(ctx context.Context, lang, packageName, platform string) ([]entity.SubscriptionPlan, error) {
+	plans, err := s.repo.ListPlans(ctx, lang, packageName)
+	if err != nil {
+		return nil, err
+	}
+	picked := filterPlansByPlatform(plans, platform)
+	if len(picked) > 0 || lang == string(i18n.ZhCN) {
+		return picked, nil
+	}
+	plans, err = s.repo.ListPlans(ctx, string(i18n.ZhCN), packageName)
+	if err != nil {
+		return nil, err
+	}
+	return filterPlansByPlatform(plans, platform), nil
+}

internal/module/client/logic/service.go

@@ -193,7 +193,9 @@ func (s *Service) SubscriptionPlans(ctx context.Context, packageName, platform s
 		if err != nil {
 			return nil, err
 		}
-		return &entity.SubscriptionConfig{Headline: headline, Subheadline: sub, Plans: plans}, nil
+		if len(plans) > 0 {
+			return &entity.SubscriptionConfig{Headline: headline, Subheadline: sub, Plans: plans}, nil
+		}
 	}
 	return &entity.SubscriptionConfig{
 		Headline: headline, Subheadline: sub, Plans: filterPlansByPlatform(defaultPlans(lang), platform),

internal/module/client/logic/service_test.go

@@ -183,7 +183,7 @@ func TestLegalPlansFeedbackFallback(t *testing.T) {
 	if err != nil || legal.Privacy.Title == "" || legal.Terms.Title == "" {
 		t.Fatalf("legal %+v err=%v", legal, err)
 	}
-	plans, err := svc.SubscriptionPlans(t.Context(), "")
+	plans, err := svc.SubscriptionPlans(t.Context(), "", "")
 	if err != nil || len(plans.Plans) != 4 {
 		t.Fatalf("plans %+v err=%v", plans, err)
 	}
@@ -331,7 +331,7 @@ func TestDeveloperAppsAdmin(t *testing.T) {
 	if err != nil || len(apps.Items) != 0 {
 		t.Fatalf("exclude package %+v %v", apps, err)
 	}
-	cfg, err := svc.SubscriptionPlans(ctxLang(string(i18n.ZhTW)), "")
+	cfg, err := svc.SubscriptionPlans(ctxLang(string(i18n.ZhTW)), "", "")
 	if err != nil || len(cfg.Plans) == 0 {
 		t.Fatalf("plans %+v %v", cfg, err)
 	}

internal/module/conversation/model/conversation_member.go

@@ -51,6 +51,35 @@ func (m *Repo) ListMemberIDs(ctx context.Context, conversationID int64) ([]int64
 	return ids, nil
 }
 
+func (m *Repo) UpdateLastRead(ctx context.Context, conversationID, userID, msgID int64) error {
+	if conversationID <= 0 || userID <= 0 || msgID <= 0 {
+		return errs.ErrInvalidArgument
+	}
+	db, err := m.session(ctx)
+	if err != nil {
+		return err
+	}
+	res := db.Model(&ConversationMember{}).
+		Where("conversation_id = ? AND user_id = ? AND last_read_msg_id < ?", conversationID, userID, msgID).
+		Update("last_read_msg_id", msgID)
+	if res.Error != nil {
+		return mapDBErr(res.Error)
+	}
+	if res.RowsAffected > 0 {
+		return nil
+	}
+	var n int64
+	if err := db.Model(&ConversationMember{}).
+		Where("conversation_id = ? AND user_id = ?", conversationID, userID).
+		Count(&n).Error; err != nil {
+		return mapDBErr(err)
+	}
+	if n == 0 {
+		return errs.ErrNotFound
+	}
+	return nil
+}
+
 func (m *Repo) GetMember(ctx context.Context, conversationID, userID int64) (*entity.Member, error) {
 	if conversationID <= 0 || userID <= 0 {
 		return nil, errs.ErrInvalidArgument

internal/module/conversation/model/memory.go

@@ -110,6 +110,26 @@ func (m *MemoryP2P) GetMember(_ context.Context, conversationID, userID int64) (
 	return &cp, nil
 }
 
+func (m *MemoryP2P) UpdateLastRead(_ context.Context, conversationID, userID, msgID int64) error {
+	if conversationID <= 0 || userID <= 0 || msgID <= 0 {
+		return errs.ErrInvalidArgument
+	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	byUser, ok := m.members[conversationID]
+	if !ok {
+		return errs.ErrNotFound
+	}
+	mem, ok := byUser[userID]
+	if !ok {
+		return errs.ErrNotFound
+	}
+	if msgID > mem.LastReadMsgID {
+		mem.LastReadMsgID = msgID
+	}
+	return nil
+}
+
 func (m *MemoryP2P) UpdateLastMsg(_ context.Context, id, msgID, atMS int64) error {
 	m.mu.Lock()
 	defer m.mu.Unlock()

internal/module/inbox/logic/store.go

@@ -11,6 +11,7 @@ type Store interface {
 	TouchUnread(ctx context.Context, userID, conversationID int64, score int64) error
 	TouchRead(ctx context.Context, userID, conversationID int64, score int64) error
 	IncrUnread(ctx context.Context, userID, conversationID int64, delta int64) error
+	ClearUnread(ctx context.Context, userID, conversationID, score int64) error
 	SetPreview(ctx context.Context, conversationID int64, preview []byte) error
 	ListUpdatedSince(ctx context.Context, userID, score int64, limit int) ([]entity.DialogHit, error)
 	LoadDialogs(ctx context.Context, userID int64, hits []entity.DialogHit) ([]entity.Dialog, error)

internal/module/inbox/logic/store_test.go

@@ -12,6 +12,7 @@ type nopStore struct{}
 func (nopStore) TouchUnread(context.Context, int64, int64, int64) error { return nil }
 func (nopStore) TouchRead(context.Context, int64, int64, int64) error   { return nil }
 func (nopStore) IncrUnread(context.Context, int64, int64, int64) error  { return nil }
+func (nopStore) ClearUnread(context.Context, int64, int64, int64) error { return nil }
 func (nopStore) SetPreview(context.Context, int64, []byte) error        { return nil }
 func (nopStore) ListUpdatedSince(context.Context, int64, int64, int) ([]entity.DialogHit, error) {
 	return nil, nil

internal/module/inbox/model/memory.go

@@ -51,6 +51,25 @@ func (s *MemoryStore) TouchRead(_ context.Context, userID, conversationID, score
 	return nil
 }
 
+func (s *MemoryStore) ClearUnread(_ context.Context, userID, conversationID, score int64) error {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	d := s.user(userID)
+	if score <= 0 {
+		if sc, ok := d.unread[conversationID]; ok {
+			score = sc
+		} else if sc, ok := d.read[conversationID]; ok {
+			score = sc
+		}
+	}
+	delete(d.unread, conversationID)
+	delete(d.unreadCnt, conversationID)
+	if score > 0 {
+		d.read[conversationID] = score
+	}
+	return nil
+}
+
 func (s *MemoryStore) IncrUnread(_ context.Context, userID, conversationID, delta int64) error {
 	s.mu.Lock()
 	defer s.mu.Unlock()
@@ -125,6 +144,17 @@ func (s *MemoryStore) Preview(conversationID int64) []byte {
 	return append([]byte(nil), s.previews[conversationID]...)
 }
 
+func (s *MemoryStore) ReadScore(userID, conversationID int64) (int64, bool) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	d := s.byUser[userID]
+	if d == nil {
+		return 0, false
+	}
+	sc, ok := d.read[conversationID]
+	return sc, ok
+}
+
 func (s *MemoryStore) UnreadCount(userID, conversationID int64) int64 {
 	s.mu.Lock()
 	defer s.mu.Unlock()

internal/module/inbox/model/redis.go

@@ -50,6 +50,34 @@ func (s *RedisStore) TouchRead(ctx context.Context, userID, conversationID, scor
 	return err
 }
 
+func (s *RedisStore) ClearUnread(ctx context.Context, userID, conversationID, score int64) error {
+	if s == nil || s.rdb == nil {
+		return nil
+	}
+	member := strconv.FormatInt(conversationID, 10)
+	if score <= 0 {
+		score = s.dialogScore(ctx, userID, member)
+	}
+	pipe := s.rdb.Pipeline()
+	if score > 0 {
+		pipe.ZAdd(ctx, readKey(userID), redis.Z{Score: float64(score), Member: member})
+	}
+	pipe.ZRem(ctx, unreadKey(userID), member)
+	pipe.HDel(ctx, countKey(userID), member)
+	_, err := pipe.Exec(ctx)
+	return err
+}
+
+func (s *RedisStore) dialogScore(ctx context.Context, userID int64, member string) int64 {
+	if z, err := s.rdb.ZScore(ctx, unreadKey(userID), member).Result(); err == nil {
+		return int64(z)
+	}
+	if z, err := s.rdb.ZScore(ctx, readKey(userID), member).Result(); err == nil {
+		return int64(z)
+	}
+	return 0
+}
+
 func (s *RedisStore) IncrUnread(ctx context.Context, userID, conversationID, delta int64) error {
 	if s == nil || s.rdb == nil {
 		return nil

internal/module/message/logic/message_test.go

@@ -14,6 +14,7 @@ import (
 	delivermodel "spark-server/internal/module/deliver/model"
 	friendlogic "spark-server/internal/module/friend/logic"
 	friendmodel "spark-server/internal/module/friend/model"
+	inboxmodel "spark-server/internal/module/inbox/model"
 	msgentity "spark-server/internal/module/message/entity"
 	msgmodel "spark-server/internal/module/message/model"
 	userentity "spark-server/internal/module/user/entity"
@@ -238,6 +239,82 @@ func TestP2PRejectsGroupMentions(t *testing.T) {
 	}
 }
 
+func TestMarkReadClearsUnreadAndNotifies(t *testing.T) {
+	svc, users, friends, q := testMessageSvc(t)
+	inbox := inboxmodel.NewMemoryStore()
+	svc.BindInbox(inbox)
+	ctx := reqctx.With(t.Context(), &reqctx.RequestContext{Language: "zh-CN", RealIP: "203.0.113.10"})
+	a, b := twoUsers(t, users)
+	applied, err := friends.Apply(ctx, a.ID, &friendlogic.ApplyInput{UID: b.ID, Reason: "hi"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := friends.Accept(ctx, b.ID, friendlogic.AcceptInput{RequestID: applied.RequestID}); err != nil {
+		t.Fatal(err)
+	}
+	text := json.RawMessage(`{"t":1,"text":{"body":"hi"}}`)
+	sent, err := svc.SendMessage(ctx, a.ID, 1, &SendInput{PeerID: b.ID, ClientMsgID: "c-read", Content: text})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := inbox.TouchUnread(ctx, b.ID, sent.ConversationID, sent.ServerTimeMS); err != nil {
+		t.Fatal(err)
+	}
+	if err := inbox.IncrUnread(ctx, b.ID, sent.ConversationID, 2); err != nil {
+		t.Fatal(err)
+	}
+	out, err := svc.MarkRead(ctx, b.ID, 9, &ReadInput{ConversationID: sent.ConversationID, MsgID: sent.MsgID})
+	if err != nil || out.MsgID != sent.MsgID || out.ConversationID != sent.ConversationID {
+		t.Fatalf("read %+v %v", out, err)
+	}
+	if inbox.UnreadCount(b.ID, sent.ConversationID) != 0 {
+		t.Fatalf("unread %d", inbox.UnreadCount(b.ID, sent.ConversationID))
+	}
+	if _, ok := inbox.ReadScore(b.ID, sent.ConversationID); !ok {
+		t.Fatal("dialog still unread")
+	}
+	mem, err := svc.chats.GetMember(ctx, sent.ConversationID, b.ID)
+	if err != nil || mem.LastReadMsgID != sent.MsgID {
+		t.Fatalf("cursor %+v %v", mem, err)
+	}
+	job, err := takeCmd(ctx, q, cmdRead, 0)
+	if err != nil || job == nil || job.ExcludeDevice != 9 || !job.NotPush || !job.NotMailbox {
+		t.Fatalf("read job %+v %v", job, err)
+	}
+	if len(job.ToUserIDs) != 1 || job.ToUserIDs[0] != b.ID {
+		t.Fatalf("targets %+v", job.ToUserIDs)
+	}
+	again, err := svc.MarkRead(ctx, b.ID, 9, &ReadInput{ConversationID: sent.ConversationID, MsgID: sent.MsgID})
+	if err != nil || again.MsgID != sent.MsgID {
+		t.Fatalf("idempotent %+v %v", again, err)
+	}
+	if _, err := takeCmd(ctx, q, cmdRead, 0); !errors.Is(err, errs.ErrNotFound) {
+		t.Fatalf("second notify %v", err)
+	}
+	if _, err := svc.MarkRead(ctx, a.ID, 1, &ReadInput{MsgID: sent.MsgID}); !errors.Is(err, errs.ErrInvalidArgument) {
+		t.Fatalf("missing conversation %v", err)
+	}
+	otherID, err := svc.ids.NextID()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := svc.repo.Create(ctx, &msgentity.Message{
+		ID: otherID, ConversationID: sent.ConversationID + 1, SenderID: a.ID, ClientMsgID: "other",
+		Type: msgentity.TypeText, Content: []byte(`{"t":1,"text":{"body":"x"}}`), Status: msgentity.StatusNormal,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := svc.MarkRead(ctx, a.ID, 1, &ReadInput{ConversationID: sent.ConversationID, MsgID: otherID}); !errors.Is(err, errs.ErrInvalidArgument) {
+		t.Fatalf("wrong conversation %v", err)
+	}
+	if _, err := svc.MarkRead(ctx, 1, 1, &ReadInput{ConversationID: sent.ConversationID, MsgID: sent.MsgID}); !errors.Is(err, errs.ErrForbidden) {
+		t.Fatalf("non member %v", err)
+	}
+	if _, err := svc.MarkRead(ctx, 0, 1, &ReadInput{ConversationID: 1, MsgID: 1}); !errors.Is(err, errs.ErrUnauthorized) {
+		t.Fatalf("anon %v", err)
+	}
+}
+
 func TestMessageGuards(t *testing.T) {
 	svc, _, _, _ := testMessageSvc(t)
 	ctx := t.Context()

internal/module/message/logic/read.go

@@ -0,0 +1,118 @@
+package logic
+
+import (
+	"context"
+	"errors"
+
+	conventity "spark-server/internal/module/conversation/entity"
+	deliverentity "spark-server/internal/module/deliver/entity"
+	"spark-server/internal/pkg/errs"
+)
+
+const cmdRead uint16 = 8
+
+// ReadInput is TCP command 8. msg_id is the message the member has read.
+type ReadInput struct {
+	ConversationID int64 `json:"conversation_id"`
+	MsgID          int64 `json:"msg_id"`
+}
+
+// ReadResult is the stored cursor. msg_id is last_read_msg_id and never moves backward.
+type ReadResult struct {
+	ConversationID int64 `json:"conversation_id"`
+	MsgID          int64 `json:"msg_id"`
+}
+
+func (s *Service) MarkRead(ctx context.Context, userID int64, deviceID uint64, in *ReadInput) (*ReadResult, error) {
+	if !s.ready() || userID <= 0 {
+		return nil, errs.ErrUnauthorized
+	}
+	if in == nil || in.ConversationID <= 0 || in.MsgID <= 0 {
+		return nil, errs.ErrInvalidArgument
+	}
+	mem, err := s.readMember(ctx, in.ConversationID, userID)
+	if err != nil {
+		return nil, err
+	}
+	if err := s.readTarget(ctx, in); err != nil {
+		return nil, err
+	}
+	cursor, advanced, err := s.advanceRead(ctx, mem, in)
+	if err != nil {
+		return nil, err
+	}
+	if err := s.clearDialogUnread(ctx, userID, in.ConversationID); err != nil {
+		return nil, err
+	}
+	if advanced {
+		if err := s.enqueueRead(ctx, userID, deviceID, in.ConversationID, cursor); err != nil {
+			return nil, err
+		}
+	}
+	return &ReadResult{ConversationID: in.ConversationID, MsgID: cursor}, nil
+}
+
+func (s *Service) readMember(ctx context.Context, conversationID, userID int64) (*conventity.Member, error) {
+	mem, err := s.chats.GetMember(ctx, conversationID, userID)
+	if err != nil {
+		if errors.Is(err, errs.ErrNotFound) {
+			return nil, errs.ErrForbidden
+		}
+		return nil, err
+	}
+	return mem, nil
+}
+
+func (s *Service) readTarget(ctx context.Context, in *ReadInput) error {
+	msg, err := s.repo.GetByID(ctx, in.MsgID)
+	if err != nil {
+		return err
+	}
+	if msg.ConversationID != in.ConversationID {
+		return errs.ErrInvalidArgument
+	}
+	return nil
+}
+
+func (s *Service) advanceRead(ctx context.Context, mem *conventity.Member, in *ReadInput) (int64, bool, error) {
+	if mem.LastReadMsgID >= in.MsgID {
+		return mem.LastReadMsgID, false, nil
+	}
+	if err := s.chats.UpdateLastRead(ctx, in.ConversationID, mem.UserID, in.MsgID); err != nil {
+		if errors.Is(err, errs.ErrNotFound) {
+			return 0, false, errs.ErrForbidden
+		}
+		return 0, false, err
+	}
+	return in.MsgID, true, nil
+}
+
+func (s *Service) clearDialogUnread(ctx context.Context, userID, conversationID int64) error {
+	if s.inbox == nil {
+		return nil
+	}
+	var score int64
+	if conv, err := s.chats.Get(ctx, conversationID); err == nil && conv != nil {
+		score = conv.LastMsgAt
+	}
+	if err := s.inbox.ClearUnread(ctx, userID, conversationID, score); err != nil {
+		return errs.ErrInner.WithError("inbox", err)
+	}
+	return nil
+}
+
+func (s *Service) enqueueRead(ctx context.Context, userID int64, deviceID uint64, conversationID, cursor int64) error {
+	if s.queue == nil {
+		return nil
+	}
+	jobID, err := s.ids.NextID()
+	if err != nil {
+		return err
+	}
+	return s.queue.Push(ctx, &deliverentity.Job{
+		MsgID: jobID, Cmd: cmdRead, ConversationID: conversationID,
+		FromUserID: userID, FromDeviceID: deviceID, ExcludeDevice: deviceID,
+		ToUserIDs: []int64{userID}, NotPush: true, NotMailbox: true,
+		Payload: controlBytes(cursor, conversationID),
+	})
+}

internal/module/message/logic/service.go

@@ -30,6 +30,12 @@ type Conversations interface {
 	Get(ctx context.Context, id int64) (*conventity.Conversation, error)
 	GetMember(ctx context.Context, conversationID, userID int64) (*conventity.Member, error)
 	ListMemberIDs(ctx context.Context, conversationID int64) ([]int64, error)
+	UpdateLastRead(ctx context.Context, conversationID, userID, msgID int64) error
+}
+
+// DialogUnread clears one conversation's Redis unread (HDEL + unread ZSET → read ZSET).
+type DialogUnread interface {
+	ClearUnread(ctx context.Context, userID, conversationID, score int64) error
 }
 
 type P2POpener interface {
@@ -55,7 +61,7 @@ type DeliverQueue interface {
 	Push(ctx context.Context, job *deliverentity.Job) error
 }
 
-// Service is send / recall / delete / history. HTTP and TCP share it.
+// Service is send / recall / delete / read / history. HTTP and TCP share it.
 type Service struct {
 	repo    Repository
 	chats   Conversations
@@ -64,6 +70,7 @@ type Service struct {
 	groups  GroupGate
 	ids     IDGen
 	queue   DeliverQueue
+	inbox   DialogUnread
 	account AccountGate
 	filter  TextFilter
 	penalty Penalty
@@ -73,6 +80,14 @@ func NewService(repo Repository, chats Conversations, p2p P2POpener, friends Fri
 	return &Service{repo: repo, chats: chats, p2p: p2p, friends: friends, groups: groups, ids: ids, queue: queue}
 }
 
+// BindInbox wires the Redis dialog list. Nil skips the unread clear.
+func (s *Service) BindInbox(inbox DialogUnread) {
+	if s == nil {
+		return
+	}
+	s.inbox = inbox
+}
+
 // BindModeration wires account gate and text filter (optional; nil skips).
 func (s *Service) BindModeration(account AccountGate, filter TextFilter, penalty Penalty) {
 	if s == nil {

internal/module/squarebot/logic/service_test.go

@@ -10,6 +10,7 @@ import (
 	"spark-server/internal/infrastructure/id"
 	friendlogic "spark-server/internal/module/friend/logic"
 	friendmodel "spark-server/internal/module/friend/model"
+	squareentity "spark-server/internal/module/square/entity"
 	squarelogic "spark-server/internal/module/square/logic"
 	squaremodel "spark-server/internal/module/square/model"
 	userentity "spark-server/internal/module/user/entity"
@@ -70,7 +71,7 @@ func TestPublishBotMediaTTLAndDedupe(t *testing.T) {
 	if !errors.Is(err, errs.ErrDuplicate) {
 		t.Fatalf("want duplicate got %v", err)
 	}
-	feed, err := square.Feed(ctx, u.ID, 0, 20)
+	feed, err := square.Feed(ctx, u.ID, squareentity.ChannelSquare, squarelogic.FeedQuery{Limit: 20})
 	if err != nil || len(feed.Items) < 1 {
 		t.Fatalf("feed %+v %v", feed, err)
 	}
@@ -78,7 +79,7 @@ func TestPublishBotMediaTTLAndDedupe(t *testing.T) {
 	if err != nil || n < 1 {
 		t.Fatalf("expire %d %v", n, err)
 	}
-	feed2, err := square.Feed(ctx, u.ID, 0, 20)
+	feed2, err := square.Feed(ctx, u.ID, squareentity.ChannelSquare, squarelogic.FeedQuery{Limit: 20})
 	if err != nil {
 		t.Fatal(err)
 	}

internal/transport/http/docs/openapi.yaml

@@ -3556,7 +3556,13 @@ components:
       description: |
         Frame.Command：1 AUTH 2 HEARTBEAT 3 GENERATE_MSG_ID 4 SEND_MESSAGE 5 SEND_ACK
         6 MESSAGE 7 PUSH_ACK 8 READ 9 RECALL 10 SYNC_INBOX 11 SYNC_INBOX_FINISH
-        12 SYNC_MESSAGES 13 SYNC_MESSAGES_FINISH 14 KICK 15 DELETE_MESSAGE
+        12 SYNC_MESSAGES 13 SYNC_MESSAGES_FINISH 14 KICK 15 DELETE_MESSAGE。
+        命令 8 READ 请求体为 `{"conversation_id","msg_id"}`（无信封）。成功应答命令仍为 8，
+        Body 为 Envelope，`data` 为 `{"conversation_id","msg_id"}`，其中 `msg_id` 是
+        `t_conversation_members.last_read_msg_id`（只增不减）。服务端清该用户 Redis
+        `im:dialog:unreadcnt` 字段，并把会话从 `im:dialog:unread` 挪到 `im:dialog:read`。
+        水位前进时通知该用户其他在线设备（命令 8，不系统推送、不写 mailbox）。
+        缺 `conversation_id` 或 `msg_id` 时 `code` 为 `40001`。
     EnvelopeCode:
       type: string
       enum: ["0", "40001", "40002", "40003", "40004", "40005", "40101", "40102", "40103", "40301", "40302", "40303", "40401", "40402", "40901", "42901", "42902", "42903", "49901", "50001", "50002", "50003", "50301", "50401"]

internal/transport/tcp/handlers.go

@@ -13,6 +13,7 @@ func (s *Server) registerDefaultHandlers() {
 	s.router.Register(protocol.CmdAuth, s.handleAuth)
 	s.router.Register(protocol.CmdGenerateMsgID, s.handleGenerateMsgID)
 	s.router.Register(protocol.CmdSendMessage, s.handleSendMessage)
+	s.router.Register(protocol.CmdReadMessage, s.handleReadMessage)
 	s.router.Register(protocol.CmdRecallMessage, s.handleRecallMessage)
 	s.router.Register(protocol.CmdDeleteMessage, s.handleDeleteMessage)
 	s.router.Register(protocol.CmdSyncInbox, s.handleSyncInbox)

internal/transport/tcp/message.go

@@ -15,6 +15,11 @@ type msgIDBody struct {
 	MsgID int64 `json:"msg_id"`
 }
 
+type readBody struct {
+	ConversationID int64 `json:"conversation_id"`
+	MsgID          int64 `json:"msg_id"`
+}
+
 type syncInboxBody struct {
 	MailboxSeq  int64 `json:"mailbox_seq"`
 	DialogScore int64 `json:"dialog_score"`
@@ -45,6 +50,26 @@ func (s *Server) handleSendMessage(ctx context.Context, conn *session.Conn, fram
 	}, nil
 }
 
+func (s *Server) handleReadMessage(ctx context.Context, conn *session.Conn, frame protocol.Frame) (*protocol.Frame, error) {
+	if s.messages == nil {
+		return nil, errs.ErrAvailable
+	}
+	var in readBody
+	if err := json.Unmarshal(frame.Body, &in); err != nil || in.ConversationID <= 0 || in.MsgID <= 0 {
+		return nil, errs.ErrInvalidArgument
+	}
+	out, err := s.messages.MarkRead(ctx, conn.UserID(), conn.DeviceID(), &msglogic.ReadInput{
+		ConversationID: in.ConversationID, MsgID: in.MsgID,
+	})
+	if err != nil {
+		return nil, err
+	}
+	return &protocol.Frame{
+		Version: protocol.CurrentVersion, Command: protocol.CmdReadMessage,
+		RequestID: frame.RequestID, SequenceID: frame.SequenceID, Body: protocol.EncodeOK(conn.RequestID(), out),
+	}, nil
+}
+
 func (s *Server) handleRecallMessage(ctx context.Context, conn *session.Conn, frame protocol.Frame) (*protocol.Frame, error) {
 	if s.messages == nil {
 		return nil, errs.ErrAvailable

internal/transport/tcp/read_test.go

@@ -0,0 +1,95 @@
+package tcp
+
+import (
+	"encoding/json"
+	"errors"
+	"strconv"
+	"testing"
+
+	"spark-server/internal/infrastructure/id"
+	convlogic "spark-server/internal/module/conversation/logic"
+	convmodel "spark-server/internal/module/conversation/model"
+	delivermodel "spark-server/internal/module/deliver/model"
+	friendlogic "spark-server/internal/module/friend/logic"
+	friendmodel "spark-server/internal/module/friend/model"
+	inboxmodel "spark-server/internal/module/inbox/model"
+	msglogic "spark-server/internal/module/message/logic"
+	msgmodel "spark-server/internal/module/message/model"
+	userentity "spark-server/internal/module/user/entity"
+	"spark-server/internal/module/user/logictest"
+	"spark-server/internal/pkg/errs"
+	"spark-server/internal/pkg/reqctx"
+	"spark-server/internal/transport/tcp/protocol"
+	"spark-server/internal/transport/tcp/session"
+)
+
+func TestReadMessageRegistered(t *testing.T) {
+	svc, b, convID, msgID := readFixture(t)
+	srv := &Server{messages: svc, router: NewRouter()}
+	srv.registerDefaultHandlers()
+	conn := session.NewConn("c", "127.0.0.1", 4)
+	conn.Authenticate(b, 9)
+	conn.SetRequestID("req-read")
+
+	missing, err := srv.router.Dispatch(t.Context(), conn, protocol.Frame{
+		Version: protocol.CurrentVersion, Command: protocol.CmdReadMessage, Body: []byte(`{}`),
+	})
+	if missing != nil || !errors.Is(err, errs.ErrInvalidArgument) {
+		t.Fatalf("missing fields frame=%v err=%v", missing, err)
+	}
+
+	resp, err := srv.router.Dispatch(t.Context(), conn, protocol.Frame{
+		Version: protocol.CurrentVersion, Command: protocol.CmdReadMessage, SequenceID: 4,
+		Body: []byte(`{"conversation_id":` + strconv.FormatInt(convID, 10) + `,"msg_id":` + strconv.FormatInt(msgID, 10) + `}`),
+	})
+	if err != nil || resp == nil || resp.Command != protocol.CmdReadMessage {
+		t.Fatalf("read %+v %v", resp, err)
+	}
+	env, err := protocol.DecodeEnvelope(resp.Body)
+	if err != nil || env.Code != protocol.CodeOK {
+		t.Fatalf("envelope %+v %v", env, err)
+	}
+	var data msglogic.ReadResult
+	if err := json.Unmarshal(env.Data, &data); err != nil || data.MsgID != msgID || data.ConversationID != convID {
+		t.Fatalf("data %+v %v", data, err)
+	}
+}
+
+func readFixture(t *testing.T) (*msglogic.Service, int64, int64, int64) {
+	t.Helper()
+	ids, err := id.New(1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	users := logictest.Users(ids, nil)
+	chats := convmodel.NewMemoryP2P()
+	p2p := convlogic.NewP2PService(chats, ids)
+	q := delivermodel.NewMemoryQueue()
+	msgs := msgmodel.NewMemoryRepo()
+	friends := friendlogic.NewService(friendmodel.NewMemoryRepo(), users, ids, friendlogic.NewGreeter(p2p, msgs, ids, q))
+	svc := msglogic.NewService(msgs, chats, p2p, friends, nil, ids, q)
+	svc.BindInbox(inboxmodel.NewMemoryStore())
+	ctx := reqctx.With(t.Context(), &reqctx.RequestContext{Language: "zh-CN"})
+	a, err := users.AuthenticateEmail(ctx, "test01@spark.com", "1", &userentity.LoginDevice{DeviceID: 1, DeviceType: userentity.DeviceIOS})
+	if err != nil {
+		t.Fatal(err)
+	}
+	b, err := users.AuthenticateEmail(ctx, "test02@spark.com", "1", &userentity.LoginDevice{DeviceID: 2, DeviceType: userentity.DeviceWeb})
+	if err != nil {
+		t.Fatal(err)
+	}
+	applied, err := friends.Apply(ctx, a.ID, &friendlogic.ApplyInput{UID: b.ID})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := friends.Accept(ctx, b.ID, friendlogic.AcceptInput{RequestID: applied.RequestID}); err != nil {
+		t.Fatal(err)
+	}
+	sent, err := svc.SendMessage(ctx, a.ID, 1, &msglogic.SendInput{
+		PeerID: b.ID, ClientMsgID: "read-1", Content: json.RawMessage(`{"t":1,"text":{"body":"hi"}}`),
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	return svc, b.ID, sent.ConversationID, sent.MsgID
+}