im_server/server/store/group_message_runtime_test.go

package store

import (
	"context"
	"testing"
	"time"

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

type testGroupMessageRuntime struct {
	claims map[string]*groupMessageUniqueRecord
	seqs   map[string]int64
}

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

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) (int64, error) {
	if r.seqs[topic] < baseline {
		r.seqs[topic] = baseline
	}
	r.seqs[topic]++
	return r.seqs[topic], 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) 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) error {
	return 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) CacheObject(types.TenantID, string, string, any) error {
	return nil
}

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

func (r *testGroupMessageRuntime) DeleteCachedObject(types.TenantID, string, string) error {
	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(string, string, time.Duration) (bool, error) {
	return true, nil
}

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

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, ready, err := ReserveMessageSeq(tenantID, topic, 184)
	if err != nil || !ready || seq != 185 {
		t.Fatalf("unexpected first seq: seq=%d ready=%v err=%v", seq, 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, ready, err = ReserveMessageSeq(tenantID, topic, 184)
	if err != nil || !ready || seq != 186 {
		t.Fatalf("unexpected second seq: seq=%d ready=%v err=%v", seq, ready, err)
	}
}