From d293de34f597e6459a9792fa74cd5bc6aec8a1b1 Mon Sep 17 00:00:00 2001 From: "DESKTOP-L8OUL9F\\gfl0034" <1003614916@qq.com> Date: Fri, 16 Oct 2026 18:22:08 +0800 Subject: [PATCH] =?UTF-8?q?=E5=AE=9E=E7=8E=B0=20TCP=20READ=EF=BC=88?= =?UTF-8?q?=E5=91=BD=E4=BB=A4=208=EF=BC=89=EF=BC=8C=E5=B7=B2=E8=AF=BB?= =?UTF-8?q?=E6=B0=B4=E4=BD=8D=E5=89=8D=E8=BF=9B=E6=97=B6=E6=B8=85=E6=9C=AA?= =?UTF-8?q?=E8=AF=BB=E5=B9=B6=E9=80=9A=E7=9F=A5=E5=85=B6=E4=BB=96=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 补上会员价格按平台筛选，并修正广场机器人测试的 Feed 调用，使 go test 能通过。 Co-authored-by: Cursor --- README.md | 2 +- docs/architecture-v3.md | 10 ++ docs/daily/2026-10-16-change-list.md | 36 ++++++ internal/bootstrap/app.go | 4 +- internal/module/client/logic/plans.go | 63 ++++++++++ internal/module/client/logic/service.go | 4 +- internal/module/client/logic/service_test.go | 4 +- .../conversation/model/conversation_member.go | 29 +++++ internal/module/conversation/model/memory.go | 20 +++ internal/module/inbox/logic/store.go | 1 + internal/module/inbox/logic/store_test.go | 1 + internal/module/inbox/model/memory.go | 30 +++++ internal/module/inbox/model/redis.go | 28 +++++ internal/module/message/logic/message_test.go | 77 ++++++++++++ internal/module/message/logic/read.go | 118 ++++++++++++++++++ internal/module/message/logic/service.go | 17 ++- .../module/squarebot/logic/service_test.go | 5 +- internal/transport/http/docs/openapi.yaml | 8 +- internal/transport/tcp/handlers.go | 1 + internal/transport/tcp/message.go | 25 ++++ internal/transport/tcp/read_test.go | 95 ++++++++++++++ 21 files changed, 569 insertions(+), 9 deletions(-) create mode 100644 docs/daily/2026-10-16-change-list.md create mode 100644 internal/module/client/logic/plans.go create mode 100644 internal/module/message/logic/read.go create mode 100644 internal/transport/tcp/read_test.go diff --git a/README.md b/README.md index e6f667e..459409a 100644 --- a/README.md +++ b/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 | → 某会话消息增量 | diff --git a/docs/architecture-v3.md b/docs/architecture-v3.md index 1ccc0fe..100314b 100644 --- a/docs/architecture-v3.md +++ b/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。 diff --git a/docs/daily/2026-10-16-change-list.md b/docs/daily/2026-10-16-change-list.md new file mode 100644 index 0000000..45fc5c8 --- /dev/null +++ b/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 并发布到新加坡 diff --git a/internal/bootstrap/app.go b/internal/bootstrap/app.go index a844903..40c8923 100644 --- a/internal/bootstrap/app.go +++ b/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) } diff --git a/internal/module/client/logic/plans.go b/internal/module/client/logic/plans.go new file mode 100644 index 0000000..df0ba67 --- /dev/null +++ b/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 +} diff --git a/internal/module/client/logic/service.go b/internal/module/client/logic/service.go index 23a290f..c6de63f 100644 --- a/internal/module/client/logic/service.go +++ b/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), diff --git a/internal/module/client/logic/service_test.go b/internal/module/client/logic/service_test.go index b1517c6..95bd49b 100644 --- a/internal/module/client/logic/service_test.go +++ b/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) } diff --git a/internal/module/conversation/model/conversation_member.go b/internal/module/conversation/model/conversation_member.go index 98a0cab..dd980ea 100644 --- a/internal/module/conversation/model/conversation_member.go +++ b/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 diff --git a/internal/module/conversation/model/memory.go b/internal/module/conversation/model/memory.go index 5111dfc..97e2778 100644 --- a/internal/module/conversation/model/memory.go +++ b/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() diff --git a/internal/module/inbox/logic/store.go b/internal/module/inbox/logic/store.go index 4a6c74f..7d03ff0 100644 --- a/internal/module/inbox/logic/store.go +++ b/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) diff --git a/internal/module/inbox/logic/store_test.go b/internal/module/inbox/logic/store_test.go index 778a7de..cdf9b99 100644 --- a/internal/module/inbox/logic/store_test.go +++ b/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 diff --git a/internal/module/inbox/model/memory.go b/internal/module/inbox/model/memory.go index 45d9e2f..ea6f5f2 100644 --- a/internal/module/inbox/model/memory.go +++ b/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() diff --git a/internal/module/inbox/model/redis.go b/internal/module/inbox/model/redis.go index 5540dfd..be378e5 100644 --- a/internal/module/inbox/model/redis.go +++ b/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 diff --git a/internal/module/message/logic/message_test.go b/internal/module/message/logic/message_test.go index 33a3812..f1166f7 100644 --- a/internal/module/message/logic/message_test.go +++ b/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() diff --git a/internal/module/message/logic/read.go b/internal/module/message/logic/read.go new file mode 100644 index 0000000..84e546a --- /dev/null +++ b/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), + }) +} diff --git a/internal/module/message/logic/service.go b/internal/module/message/logic/service.go index 547dbee..d1a44df 100644 --- a/internal/module/message/logic/service.go +++ b/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 { diff --git a/internal/module/squarebot/logic/service_test.go b/internal/module/squarebot/logic/service_test.go index 61df9c3..232a94b 100644 --- a/internal/module/squarebot/logic/service_test.go +++ b/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) } diff --git a/internal/transport/http/docs/openapi.yaml b/internal/transport/http/docs/openapi.yaml index 751df2b..0ffab20 100644 --- a/internal/transport/http/docs/openapi.yaml +++ b/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"] diff --git a/internal/transport/tcp/handlers.go b/internal/transport/tcp/handlers.go index a02065f..e992410 100644 --- a/internal/transport/tcp/handlers.go +++ b/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) diff --git a/internal/transport/tcp/message.go b/internal/transport/tcp/message.go index b029732..a4f4b52 100644 --- a/internal/transport/tcp/message.go +++ b/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 diff --git a/internal/transport/tcp/read_test.go b/internal/transport/tcp/read_test.go new file mode 100644 index 0000000..cde2caf --- /dev/null +++ b/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 +}