package logic import ( "context" "encoding/json" "errors" "testing" "time" "spark-server/internal/infrastructure/id" convlogic "spark-server/internal/module/conversation/logic" convmodel "spark-server/internal/module/conversation/model" deliverentity "spark-server/internal/module/deliver/entity" 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" userlogic "spark-server/internal/module/user/logic" "spark-server/internal/module/user/logictest" "spark-server/internal/pkg/errs" "spark-server/internal/pkg/reqctx" ) func testMessageSvc(t *testing.T) (*Service, *userlogic.Service, *friendlogic.Service, *delivermodel.MemoryQueue) { 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), ) return NewService(msgs, chats, p2p, friends, nil, ids, q), users, friends, q } func twoUsers(t *testing.T, users *userlogic.Service) (*userentity.User, *userentity.User) { t.Helper() ctx := t.Context() 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) } return a, b } func TestSendReplyRecallDelete(t *testing.T) { svc, users, friends, q := testMessageSvc(t) 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"}}`) first, err := svc.SendMessage(ctx, a.ID, 1, &SendInput{PeerID: b.ID, ClientMsgID: "c1", Content: text}) if err != nil { t.Fatal(err) } if first.ConversationID == 0 { t.Fatal("conversation_id") } job, err := takeCmd(ctx, q, cmdMessage, first.MsgID) if err != nil || job == nil || job.ExcludeDevice != 1 { t.Fatalf("deliver %+v %v", job, err) } again, err := svc.SendMessage(ctx, a.ID, 1, &SendInput{PeerID: b.ID, ClientMsgID: "c1", Content: text}) if err != nil || again.MsgID != first.MsgID { t.Fatalf("idempotent %+v %v", again, err) } replyJSON, err := json.Marshal(map[string]any{"t": 1, "reply_to": first.MsgID, "text": map[string]string{"body": "re"}}) if err != nil { t.Fatal(err) } replied, err := svc.SendMessage(ctx, b.ID, 2, &SendInput{PeerID: a.ID, ClientMsgID: "c2", Content: replyJSON}) if err != nil { t.Fatal(err) } msg, err := svc.repo.GetByID(ctx, first.MsgID) if err != nil { t.Fatal(err) } if msg.ReceiverID != b.ID { t.Fatalf("receiver %d want %d", msg.ReceiverID, b.ID) } hist, err := svc.GetHistory(ctx, a.ID, HistoryRequest{ConversationID: msg.ConversationID}) if err != nil || len(hist.Items) < 2 { t.Fatalf("history %+v %v", hist, err) } foundKind := false foundSystem := false for _, it := range hist.Items { if it.PeerID != b.ID { t.Fatalf("viewer peer %d", it.PeerID) } if it.MsgID == first.MsgID && it.Kind == msgentity.KindPrivate { foundKind = true } if it.SenderID == 0 && it.MessageType == msgentity.TypeSystem { foundSystem = true if it.ReceiverID != 0 { t.Fatalf("system receiver %d", it.ReceiverID) } } } if !foundKind { t.Fatal("private_message kind") } if !foundSystem { t.Fatal("system peer") } caught, err := svc.GetHistory(ctx, a.ID, HistoryRequest{ ConversationID: msg.ConversationID, LastMsgID: hist.Items[len(hist.Items)-1].MsgID, }) if err != nil || len(caught.Items) != 0 || caught.HasMore { t.Fatalf("incremental %+v %v", caught, err) } if _, err := svc.SendMessage(ctx, a.ID, 1, &SendInput{ PeerID: b.ID, ClientMsgID: "bad", Content: json.RawMessage(`{"t":1,"reply_to":1,"text":{"body":"x"}}`), }); !errors.Is(err, errs.ErrInvalidArgument) { t.Fatalf("missing reply target %v", err) } if err := svc.RecallMessage(ctx, b.ID, 2, first.MsgID); !errors.Is(err, errs.ErrForbidden) { t.Fatalf("not sender %v", err) } if err := svc.RecallMessage(ctx, a.ID, 1, first.MsgID); err != nil { t.Fatal(err) } recJob, err := takeCmd(ctx, q, cmdRecall, 0) if err != nil || recJob == nil || recJob.MsgID == first.MsgID { t.Fatalf("recall job %+v %v", recJob, err) } got, err := svc.repo.GetByID(ctx, first.MsgID) if err != nil || got.Status != msgentity.StatusRecalled { t.Fatalf("recalled %+v %v", got, err) } oldID, err := svc.ids.NextID() if err != nil { t.Fatal(err) } old := &msgentity.Message{ ID: oldID, ConversationID: msg.ConversationID, SenderID: a.ID, ClientMsgID: "old", Type: msgentity.TypeText, Content: []byte(`{"t":1,"text":{"body":"old"}}`), Status: msgentity.StatusNormal, CreatedAtMS: time.Now().Add(-3 * time.Minute).UnixMilli(), } if err := svc.repo.Create(ctx, old); err != nil { t.Fatal(err) } if err := svc.RecallMessage(ctx, a.ID, 1, old.ID); !errors.Is(err, errs.ErrForbidden) { t.Fatalf("window %v", err) } if err := svc.DeleteMessage(ctx, a.ID, 1, replied.MsgID); err != nil { t.Fatal(err) } delJob, err := takeCmd(ctx, q, cmdDelete, 0) if err != nil || delJob == nil || len(delJob.ToUserIDs) != 1 || delJob.ToUserIDs[0] != a.ID { t.Fatalf("delete job %+v %v", delJob, err) } after, err := svc.GetHistory(ctx, a.ID, HistoryRequest{ConversationID: msg.ConversationID}) if err != nil { t.Fatal(err) } for _, it := range after.Items { if it.MsgID == replied.MsgID { t.Fatal("hidden for deleter") } } peer, err := svc.GetHistory(ctx, b.ID, HistoryRequest{ConversationID: msg.ConversationID}) if err != nil { t.Fatal(err) } seen := false for _, it := range peer.Items { if it.MsgID == replied.MsgID { seen = true } } if !seen { t.Fatal("peer still sees message") } } func TestSendRequiresFriend(t *testing.T) { svc, users, _, _ := testMessageSvc(t) ctx := t.Context() a, b := twoUsers(t, users) _, err := svc.SendMessage(ctx, a.ID, 1, &SendInput{ PeerID: b.ID, ClientMsgID: "x", Content: json.RawMessage(`{"t":1,"text":{"body":"no"}}`), }) if !errors.Is(err, errs.ErrForbidden) { t.Fatalf("want forbidden got %v", err) } } func TestSendRejectsStaleMsgID(t *testing.T) { svc, users, friends, _ := testMessageSvc(t) ctx := t.Context() a, b := twoUsers(t, users) 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) } _, err = svc.SendMessage(ctx, a.ID, 1, &SendInput{ PeerID: b.ID, ClientMsgID: "stale", MsgID: 1, Content: json.RawMessage(`{"t":1,"text":{"body":"x"}}`), }) if !errors.Is(err, errs.ErrInvalidArgument) { t.Fatalf("stale msg_id %v", err) } } func TestP2PRejectsGroupMentions(t *testing.T) { svc, users, friends, _ := testMessageSvc(t) ctx := t.Context() a, b := twoUsers(t, users) 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) } _, err = svc.SendMessage(ctx, a.ID, 1, &SendInput{ PeerID: b.ID, ClientMsgID: "at", Content: json.RawMessage(`{"t":1,"at_all":true,"text":{"body":"hi"}}`), }) if !errors.Is(err, errs.ErrInvalidArgument) { t.Fatalf("p2p at-all %v", err) } } 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() if _, err := svc.GetHistory(ctx, 0, HistoryRequest{}); !errors.Is(err, errs.ErrUnauthorized) { t.Fatalf("hist %v", err) } if err := svc.DeleteMessage(ctx, 0, 1, 1); !errors.Is(err, errs.ErrUnauthorized) { t.Fatalf("del %v", err) } if err := svc.RecallMessage(ctx, 0, 1, 1); !errors.Is(err, errs.ErrUnauthorized) { t.Fatalf("recall %v", err) } _, err := svc.SendMessage(ctx, 1, 1, &SendInput{ClientMsgID: "x", Content: json.RawMessage(`{"t":1,"text":{"body":"x"}}`)}) if !errors.Is(err, errs.ErrInvalidArgument) { t.Fatalf("no dest %v", err) } if err := svc.DeleteMessage(ctx, 1, 1, 0); !errors.Is(err, errs.ErrInvalidArgument) && !errors.Is(err, errs.ErrUnauthorized) { t.Fatalf("del zero %v", err) } } func takeCmd(ctx context.Context, q *delivermodel.MemoryQueue, cmd uint16, msgID int64) (*deliverentity.Job, error) { for i := 0; i < 8; i++ { job, err := q.Take(ctx, 50*time.Millisecond) if err != nil { return nil, err } if job == nil { return nil, errs.ErrNotFound } if job.Cmd != cmd || (msgID != 0 && job.MsgID != msgID) { continue } return job, nil } return nil, errs.ErrNotFound }