spark-server/internal/module/message/logic/message_test.go

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
	for _, it := range hist.Items {
		if it.MsgID == first.MsgID && it.Kind == msgentity.KindPrivate {
			foundKind = true
		}
	}
	if !foundKind {
		t.Fatal("private_message kind")
	}
	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
}