im_server/server/im_group_events_test.go

package main

import (
	"encoding/json"
	"testing"
	"time"

	"github.com/golang/mock/gomock"
	"github.com/tinode/chat/server/store"
	"github.com/tinode/chat/server/store/mock_store"
	"github.com/tinode/chat/server/store/types"
)

func TestPublishIMGroupRealtimeEventRestrictsMemberRemoval(t *testing.T) {
	oldGroupBusiness := store.GroupBusiness
	oldSessions := globals.sessionStore
	oldHub := globals.hub
	admin := types.Uid(1)
	affected := types.Uid(2)
	regular := types.Uid(3)
	actor := types.Uid(4)
	store.GroupBusiness = &testGroupBusinessStore{listMembers: []types.GroupMember{
		{UserID: admin, Role: types.GroupMemberRoleAdmin, MemberStatus: types.GroupMemberStatusActive},
	}}
	globals.sessionStore = NewSessionStore(time.Minute)
	globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 2)}
	t.Cleanup(func() {
		store.GroupBusiness = oldGroupBusiness
		globals.sessionStore = oldSessions
		globals.hub = oldHub
	})

	sessions := map[types.Uid]*Session{}
	for _, uid := range []types.Uid{admin, affected, regular} {
		sess := &Session{
			sid: uid.UserId(), uid: uid, tenantID: 7, proto: WEBSOCK,
			send: make(chan any, 2), subs: make(map[string]*Subscription),
		}
		globals.sessionStore.sessCache[sess.sid] = sess
		sessions[uid] = sess
	}

	publishIMGroupRealtimeEvent(7, "grpRestricted", actor, "group.member.left",
		map[string]any{"user_id": affected.UserId()})

	for _, uid := range []types.Uid{admin, affected} {
		select {
		case raw := <-sessions[uid].send:
			msg, ok := raw.(*ServerComMessage)
			if !ok || msg.Info == nil || msg.Info.Event != "group.member.left" || msg.Info.Topic != "grpRestricted" {
				t.Fatalf("unexpected event for %s: %#v", uid.UserId(), raw)
			}
			var payload map[string]any
			if err := json.Unmarshal(msg.Info.Payload, &payload); err != nil {
				t.Fatal(err)
			}
			if payload["member_sync"] != "remove" || payload["visibility"] != "admins_and_affected" {
				t.Fatalf("unexpected payload for %s: %#v", uid.UserId(), payload)
			}
		default:
			t.Fatalf("expected event for %s", uid.UserId())
		}
	}
	select {
	case raw := <-sessions[regular].send:
		t.Fatalf("regular member received restricted event: %#v", raw)
	default:
	}
	if len(globals.hub.routeSrv) != 0 {
		t.Fatalf("restricted event was broadcast to topic")
	}
}

func TestPublishIMGroupRealtimeEventBroadcastsMemberJoin(t *testing.T) {
	oldHub := globals.hub
	globals.hub = &Hub{routeSrv: make(chan *ServerComMessage, 1)}
	t.Cleanup(func() { globals.hub = oldHub })

	publishIMGroupRealtimeEvent(7, "grpPublic", types.Uid(1), "group.member.joined",
		map[string]any{"user_id": types.Uid(2).UserId()})

	select {
	case msg := <-globals.hub.routeSrv:
		if msg.Info == nil || msg.Info.Event != "group.member.joined" {
			t.Fatalf("unexpected joined event: %#v", msg)
		}
		var payload map[string]any
		if err := json.Unmarshal(msg.Info.Payload, &payload); err != nil {
			t.Fatal(err)
		}
		if payload["member_sync"] != "single" || payload["visibility"] != "all_members" {
			t.Fatalf("unexpected joined payload: %#v", payload)
		}
	default:
		t.Fatal("joined event was not broadcast")
	}
}

func TestSendPushForStoredGroupMessageIncludesEverySubscription(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()
	oldTopics := store.Topics
	oldUsersUpdate := globals.usersUpdate
	oldCluster := globals.cluster
	topics := mock_store.NewMockTopicsPersistenceInterface(ctrl)
	store.Topics = topics
	globals.usersUpdate = make(chan *UserCacheReq, 1)
	globals.cluster = nil
	t.Cleanup(func() {
		store.Topics = oldTopics
		globals.usersUpdate = oldUsersUpdate
		globals.cluster = oldCluster
	})

	subs := make([]types.Subscription, 0, 1500)
	for i := 1; i <= 1500; i++ {
		sub := types.Subscription{
			User: types.Uid(i).String(), Topic: "grpLarge", ModeWant: types.ModeCFull, ModeGiven: types.ModeCFull,
		}
		if i == 1500 {
			sub.Private = map[string]any{"mute_notifications": true}
		}
		subs = append(subs, sub)
	}
	topics.EXPECT().GetSubs(types.TenantID(7), "grpLarge", nil).Return(subs, nil)
	msg := &types.Message{
		ObjHeader: types.ObjHeader{CreatedAt: time.Now().UTC()}, TenantID: 7,
		Topic: "grpLarge", SeqId: 99, Content: map[string]any{"text": "成员加入了群聊"},
	}

	sendPushForStoredGroupMessage(7, msg)

	select {
	case req := <-globals.usersUpdate:
		if req.PushRcpt == nil || len(req.PushRcpt.To) != len(subs) {
			t.Fatalf("push recipients: got %d want %d", len(req.PushRcpt.To), len(subs))
		}
		for _, recipient := range req.PushRcpt.To {
			if !recipient.ShouldIncrementUnreadCountInCache {
				t.Fatal("system group message did not increment per-user unread")
			}
		}
		if !req.PushRcpt.To[types.Uid(1500)].Silent {
			t.Fatal("muted group recipient must receive a silent push")
		}
	default:
		t.Fatal("group system message did not enqueue push")
	}
}