im_server/server/store/group_message_runtime_test.go

package store

import (
	"context"
	"encoding/json"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/tinode/chat/server/store/types"
)

type testGroupMessageRuntime struct {
	claims          map[string]*groupMessageUniqueRecord
	seqs            map[string]int64
	shortStates     map[string]bool
	objects         map[string][]byte
	deletedPrefixes []string
	deletedObjects  []string
}

func newTestGroupMessageRuntime() *testGroupMessageRuntime {
	return &testGroupMessageRuntime{
		claims:      make(map[string]*groupMessageUniqueRecord),
		seqs:        make(map[string]int64),
		shortStates: make(map[string]bool),
		objects:     make(map[string][]byte),
	}
}

func (r *testGroupMessageRuntime) Close() error { return nil }

func (r *testGroupMessageRuntime) Claim(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) (*groupMessageUniqueRecord, bool, error) {
	key := topic + ":" + from.UserId() + ":" + clientMsgID
	if rec := r.claims[key]; rec != nil {
		return rec, false, nil
	}
	rec := &groupMessageUniqueRecord{Status: "pending", CreatedAt: time.Now().UTC()}
	r.claims[key] = rec
	return rec, true, nil
}

func (r *testGroupMessageRuntime) NextSeq(_ types.TenantID, topic string, baseline int64, candidateMonth string) (int64, string, error) {
	if r.seqs[topic] < baseline {
		r.seqs[topic] = baseline
	}
	r.seqs[topic]++
	return r.seqs[topic], candidateMonth, nil
}

func (r *testGroupMessageRuntime) PutShardRoute(types.TenantID, string, GroupMessageShardRoute) error {
	return nil
}

func (r *testGroupMessageRuntime) GetShardRoutes(types.TenantID, string) ([]GroupMessageShardRoute, bool, error) {
	return nil, false, nil
}

func (r *testGroupMessageRuntime) DeleteShardRoutes(types.TenantID, string) error {
	return nil
}

func (r *testGroupMessageRuntime) PeekSeq(_ types.TenantID, topic string) (int64, error) {
	return r.seqs[topic], nil
}

func (r *testGroupMessageRuntime) BumpSeqFloor(_ types.TenantID, topic string, seq int64) error {
	if r.seqs[topic] < seq {
		r.seqs[topic] = seq
	}
	return nil
}

func (r *testGroupMessageRuntime) MarkDone(_ types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error {
	key := topic + ":" + from.UserId() + ":" + clientMsgID
	r.claims[key] = &groupMessageUniqueRecord{Status: "done", Seq: seq, MessageID: msgID, CreatedAt: time.Now().UTC()}
	return nil
}

func (r *testGroupMessageRuntime) MarkPending(_ types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error {
	key := topic + ":" + from.UserId() + ":" + clientMsgID
	r.claims[key] = &groupMessageUniqueRecord{Status: "pending", Seq: seq, CreatedAt: time.Now().UTC()}
	return nil
}

func (r *testGroupMessageRuntime) MarkFailed(_ types.TenantID, topic string, from types.Uid, clientMsgID string) error {
	key := topic + ":" + from.UserId() + ":" + clientMsgID
	r.claims[key] = &groupMessageUniqueRecord{Status: "failed", CreatedAt: time.Now().UTC()}
	return nil
}

func (r *testGroupMessageRuntime) CacheHotMessage(types.TenantID, *types.Message) error {
	return nil
}

func (r *testGroupMessageRuntime) CachePendingHotMessage(types.TenantID, *types.Message) error {
	return nil
}

func (r *testGroupMessageRuntime) GetHotMessages(types.TenantID, string, *types.QueryOpt, int) ([]types.Message, bool, error) {
	return nil, false, nil
}

func (r *testGroupMessageRuntime) InvalidateHotMessages(types.TenantID, string, []types.Range) error {
	return nil
}

func (r *testGroupMessageRuntime) MarkUserOnline(types.TenantID, types.Uid, string, string, string, string) error {
	return nil
}

func (r *testGroupMessageRuntime) RefreshUserPresence(types.TenantID, types.Uid, string, string) error {
	return nil
}

func (r *testGroupMessageRuntime) MarkUserOffline(types.TenantID, types.Uid, string, string) (int, error) {
	return 0, nil
}

func (r *testGroupMessageRuntime) GetUserPresence(types.TenantID, types.Uid) (*UserPresence, error) {
	return nil, nil
}

func (r *testGroupMessageRuntime) MarkTopicMemberOnline(types.TenantID, string, types.Uid, string) error {
	return nil
}

func (r *testGroupMessageRuntime) MarkTopicMemberOffline(types.TenantID, string, types.Uid, string) error {
	return nil
}

func (r *testGroupMessageRuntime) GetTopicNodeOnlineMembers(types.TenantID, string, string) ([]types.Uid, error) {
	return nil, nil
}

func (r *testGroupMessageRuntime) PublishGroupFanout(string, any) error {
	return nil
}

func (r *testGroupMessageRuntime) StartGroupFanoutSubscriber(context.Context, string, func([]byte)) error {
	return nil
}

func (r *testGroupMessageRuntime) PublishRealtimeFanout(string, types.TenantID, string, string, string, string, string, any) error {
	return nil
}

func (r *testGroupMessageRuntime) StartRealtimeFanoutSubscriber(context.Context, string, func(*RealtimeFanoutEnvelope)) error {
	return nil
}

func (r *testGroupMessageRuntime) CacheObject(tenantID types.TenantID, namespace, id string, value any) error {
	payload, err := json.Marshal(value)
	if err != nil {
		return err
	}
	r.objects[strconv.FormatInt(int64(tenantID), 10)+":"+namespace+":"+id] = payload
	return nil
}

func (r *testGroupMessageRuntime) GetCachedObject(tenantID types.TenantID, namespace, id string, value any) (bool, error) {
	payload, ok := r.objects[strconv.FormatInt(int64(tenantID), 10)+":"+namespace+":"+id]
	if !ok {
		return false, nil
	}
	return true, json.Unmarshal(payload, value)
}

func (r *testGroupMessageRuntime) DeleteCachedObject(tenantID types.TenantID, namespace, id string) error {
	r.deletedObjects = append(r.deletedObjects, namespace+":"+id)
	delete(r.objects, strconv.FormatInt(int64(tenantID), 10)+":"+namespace+":"+id)
	return nil
}

func (r *testGroupMessageRuntime) DeleteCachedObjectsByPrefix(tenantID types.TenantID, namespace, prefix string) error {
	r.deletedPrefixes = append(r.deletedPrefixes, prefix)
	cachePrefix := strconv.FormatInt(int64(tenantID), 10) + ":" + namespace + ":" + prefix
	for key := range r.objects {
		if strings.HasPrefix(key, cachePrefix) {
			delete(r.objects, key)
		}
	}
	return nil
}

func (r *testGroupMessageRuntime) AddCachedSetMember(types.TenantID, string, string, string) error {
	return nil
}

func (r *testGroupMessageRuntime) RemoveCachedSetMember(types.TenantID, string, string, string) error {
	return nil
}

func (r *testGroupMessageRuntime) CachedSetContains(types.TenantID, string, string, string) (bool, error) {
	return false, nil
}

func (r *testGroupMessageRuntime) AllowRate(types.TenantID, string, int, time.Duration) (bool, int, time.Duration, error) {
	return true, 0, 0, nil
}

func (r *testGroupMessageRuntime) ClaimShortState(namespace, id string, _ time.Duration) (bool, error) {
	key := namespace + ":" + id
	if r.shortStates[key] {
		return false, nil
	}
	r.shortStates[key] = true
	return true, nil
}

func (r *testGroupMessageRuntime) ClaimShortStateValue(string, string, string, time.Duration) (bool, bool, error) {
	return true, true, nil
}

func (r *testGroupMessageRuntime) ShortStateExists(namespace, id string) (bool, error) {
	return r.shortStates[namespace+":"+id], nil
}

func (r *testGroupMessageRuntime) StoreLoginSession([]byte, LoginSession, time.Duration) error {
	return nil
}

func TestLoginTokenRevocation(t *testing.T) {
	oldRuntime := groupMsgRuntime
	groupMsgRuntime = newTestGroupMessageRuntime()
	t.Cleanup(func() { groupMsgRuntime = oldRuntime })

	token := []byte("current-device-token")
	if revoked, enabled, err := IsLoginTokenRevoked(7, token); err != nil || !enabled || revoked {
		t.Fatalf("unexpected initial state: revoked=%v enabled=%v err=%v", revoked, enabled, err)
	}
	if enabled, err := RevokeLoginToken(7, token, time.Hour); err != nil || !enabled {
		t.Fatalf("revoke token: enabled=%v err=%v", enabled, err)
	}
	if revoked, enabled, err := IsLoginTokenRevoked(7, token); err != nil || !enabled || !revoked {
		t.Fatalf("revoked token was accepted: revoked=%v enabled=%v err=%v", revoked, enabled, err)
	}
	if revoked, _, err := IsLoginTokenRevoked(7, []byte("other-device-token")); err != nil || revoked {
		t.Fatalf("other device token was revoked: revoked=%v err=%v", revoked, err)
	}
}

func TestGroupMemberPageCacheRoundTripAndInvalidate(t *testing.T) {
	oldRuntime := groupMsgRuntime
	runtime := newTestGroupMessageRuntime()
	groupMsgRuntime = runtime
	t.Cleanup(func() { groupMsgRuntime = oldRuntime })

	opts := types.GroupMemberListOptions{
		Role:        "member",
		Query:       " alice ",
		Cursor:      "42",
		Limit:       20,
		Mentionable: true,
	}
	key := groupMemberPageCacheID("grpAlpha", opts)
	if !strings.HasPrefix(key, "grpAlpha|") {
		t.Fatalf("cache key must be topic-prefixed for invalidation: %q", key)
	}
	CacheGroupMemberPage(7, key, groupMemberPageCache{
		Items: []types.GroupMember{{
			ID:     43,
			Topic:  "grpAlpha",
			UserID: types.Uid(100),
			Role:   types.GroupMemberRoleMember,
		}},
		NextCursor: "43",
	})
	cached, ok := GetCachedGroupMemberPage(7, key)
	if !ok || cached.NextCursor != "43" || len(cached.Items) != 1 || cached.Items[0].UserID != types.Uid(100) {
		t.Fatalf("unexpected cached page: ok=%v page=%#v", ok, cached)
	}

	invalidateGroupMemberPages(7, "grpAlpha")
	if _, ok := GetCachedGroupMemberPage(7, key); ok {
		t.Fatalf("group member page cache survived invalidation")
	}
	if len(runtime.deletedPrefixes) != 1 || runtime.deletedPrefixes[0] != "grpAlpha|" {
		t.Fatalf("unexpected invalidation prefixes: %#v", runtime.deletedPrefixes)
	}
}

func TestRealtimeFanoutDedupClaimIDIsNodeScoped(t *testing.T) {
	tenantID := types.TenantID(7)
	dedupKey := "data:p2pABC:42"

	nodeA := realtimeFanoutDedupClaimID(tenantID, "pod-a", dedupKey)
	nodeB := realtimeFanoutDedupClaimID(tenantID, "pod-b", dedupKey)
	if nodeA == nodeB {
		t.Fatalf("dedup claim id must be node-scoped: %q", nodeA)
	}
	if got, want := realtimeFanoutDedupClaimID(tenantID, "", dedupKey), "7:local:data:p2pABC:42"; got != want {
		t.Fatalf("unexpected local fallback claim id: got %q want %q", got, want)
	}
}

func TestMessageRuntimeSeqAndClientMidIdempotency(t *testing.T) {
	oldRuntime := groupMsgRuntime
	runtime := newTestGroupMessageRuntime()
	groupMsgRuntime = runtime
	t.Cleanup(func() { groupMsgRuntime = oldRuntime })

	tenantID := types.TenantID(7)
	topic := types.Uid(1).P2PName(types.Uid(2))
	from := types.Uid(1)

	claim, claimed, ready, err := ClaimMessageUnique(tenantID, topic, from, "mid-1")
	if err != nil || !ready || !claimed || claim == nil || claim.Status != "pending" {
		t.Fatalf("unexpected first claim: claim=%#v claimed=%v ready=%v err=%v", claim, claimed, ready, err)
	}
	seq, month, ready, err := ReserveMessageSeq(tenantID, topic, 184, "202609")
	if err != nil || !ready || seq != 185 || month != "202609" {
		t.Fatalf("unexpected first seq: seq=%d month=%s ready=%v err=%v", seq, month, ready, err)
	}
	if err = MarkMessageUniqueDone(tenantID, topic, from, "mid-1", seq, ""); err != nil {
		t.Fatalf("mark done: %v", err)
	}

	claim, claimed, ready, err = ClaimMessageUnique(tenantID, topic, from, "mid-1")
	if err != nil || !ready || claimed || claim == nil || claim.Seq != 185 {
		t.Fatalf("unexpected duplicate claim: claim=%#v claimed=%v ready=%v err=%v", claim, claimed, ready, err)
	}

	claim, claimed, ready, err = ClaimMessageUnique(tenantID, topic, from, "mid-2")
	if err != nil || !ready || !claimed || claim == nil {
		t.Fatalf("unexpected second claim: claim=%#v claimed=%v ready=%v err=%v", claim, claimed, ready, err)
	}
	seq, month, ready, err = ReserveMessageSeq(tenantID, topic, 184, "202609")
	if err != nil || !ready || seq != 186 || month != "202609" {
		t.Fatalf("unexpected second seq: seq=%d month=%s ready=%v err=%v", seq, month, ready, err)
	}
}

func TestBumpMessageSeqFloorNeverMovesBackward(t *testing.T) {
	oldRuntime := groupMsgRuntime
	runtime := newTestGroupMessageRuntime()
	groupMsgRuntime = runtime
	t.Cleanup(func() { groupMsgRuntime = oldRuntime })

	tenantID := types.TenantID(7)
	topic := types.Uid(1).P2PName(types.Uid(2))

	seq, _, ready, err := ReserveMessageSeq(tenantID, topic, 0, "")
	if err != nil || !ready || seq != 1 {
		t.Fatalf("unexpected first seq: seq=%d ready=%v err=%v", seq, ready, err)
	}
	if err = BumpMessageSeqFloor(tenantID, topic, 5); err != nil {
		t.Fatalf("bump floor: %v", err)
	}
	if current, err := runtime.PeekSeq(tenantID, topic); err != nil || current != 5 {
		t.Fatalf("floor was not applied: current=%d err=%v", current, err)
	}
	if err = BumpMessageSeqFloor(tenantID, topic, 3); err != nil {
		t.Fatalf("lower bump: %v", err)
	}
	if current, err := runtime.PeekSeq(tenantID, topic); err != nil || current != 5 {
		t.Fatalf("floor must never move backward: current=%d err=%v", current, err)
	}
	// The next allocation must continue above the bumped floor.
	if seq, _, _, err := ReserveMessageSeq(tenantID, topic, 0, ""); err != nil || seq != 6 {
		t.Fatalf("unexpected seq after bump: seq=%d err=%v", seq, err)
	}
}

func TestInvalidateGroupMemberPagesUsesTopicKeyPrefix(t *testing.T) {
	oldRuntime := groupMsgRuntime
	runtime := newTestGroupMessageRuntime()
	groupMsgRuntime = runtime
	t.Cleanup(func() { groupMsgRuntime = oldRuntime })

	invalidateGroupMemberPages(7, "grpAlpha")
	if len(runtime.deletedPrefixes) != 1 || runtime.deletedPrefixes[0] != "grpAlpha|" {
		t.Fatalf("unexpected deleted prefixes: %#v", runtime.deletedPrefixes)
	}
}

func TestInvalidateGroupSettingsUsesTopicCacheKey(t *testing.T) {
	oldRuntime := groupMsgRuntime
	runtime := newTestGroupMessageRuntime()
	groupMsgRuntime = runtime
	t.Cleanup(func() { groupMsgRuntime = oldRuntime })

	invalidateGroupSettings(7, " grpAlpha ")
	if len(runtime.deletedObjects) != 1 || runtime.deletedObjects[0] != "group_settings:grpAlpha" {
		t.Fatalf("unexpected deleted objects: %#v", runtime.deletedObjects)
	}
}