305 lines
8.4 KiB
Go
305 lines
8.4 KiB
Go
package main
|
|
|
|
import (
|
|
"encoding/json"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/tinode/chat/server/store"
|
|
"github.com/tinode/chat/server/store/types"
|
|
)
|
|
|
|
func TestRealtimeEnvelopeDeliversToUserSessions(t *testing.T) {
|
|
oldStore := globals.sessionStore
|
|
defer func() { globals.sessionStore = oldStore }()
|
|
|
|
tenantID := types.TenantID(1)
|
|
uid := types.Uid(1001)
|
|
globals.sessionStore = realtimeTestSessionStore(
|
|
realtimeTestSession("sid-1", tenantID, uid, ""),
|
|
realtimeTestSession("sid-2", tenantID, uid, ""),
|
|
realtimeTestSession("sid-other", tenantID, types.Uid(1002), ""),
|
|
)
|
|
msg := &ServerComMessage{
|
|
Data: &MsgServerData{Topic: uid.UserId(), From: types.Uid(1002).UserId(), SeqId: 12, Timestamp: time.Now()},
|
|
}
|
|
raw, err := json.Marshal(msg)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
deliverRealtimeFanoutEnvelope(&store.RealtimeFanoutEnvelope{
|
|
TenantID: tenantID,
|
|
Kind: realtimeKindMessage,
|
|
TargetType: realtimeTargetUser,
|
|
Target: uid.UserId(),
|
|
Message: raw,
|
|
})
|
|
|
|
for _, sid := range []string{"sid-1", "sid-2"} {
|
|
sess := globals.sessionStore.Get(sid)
|
|
select {
|
|
case got := <-sess.send:
|
|
if got.(*ServerComMessage).TenantID != tenantID {
|
|
t.Fatalf("tenant not restored for %s", sid)
|
|
}
|
|
default:
|
|
t.Fatalf("expected delivery to %s", sid)
|
|
}
|
|
}
|
|
if len(globals.sessionStore.Get("sid-other").send) != 0 {
|
|
t.Fatal("unexpected delivery to other user")
|
|
}
|
|
}
|
|
|
|
func TestRealtimeUserPresenceSkipsSessionsAttachedToTopic(t *testing.T) {
|
|
oldStore := globals.sessionStore
|
|
defer func() { globals.sessionStore = oldStore }()
|
|
|
|
tenantID := types.TenantID(1)
|
|
sender := types.Uid(1001)
|
|
recipient := types.Uid(1002)
|
|
topic := sender.P2PName(recipient)
|
|
if topic == "" {
|
|
t.Fatal("failed to create p2p topic")
|
|
}
|
|
visibleTopic, err := types.P2PNameForUser(recipient, topic)
|
|
if err != nil || visibleTopic == "" {
|
|
t.Fatalf("failed to create visible topic: %v", err)
|
|
}
|
|
globals.sessionStore = realtimeTestSessionStore(
|
|
realtimeTestSession("sid-attached", tenantID, recipient, visibleTopic),
|
|
realtimeTestSession("sid-me-only", tenantID, recipient, ""),
|
|
)
|
|
msg := &ServerComMessage{
|
|
Pres: &MsgServerPres{
|
|
Topic: "me",
|
|
Src: topic,
|
|
What: "msg",
|
|
SeqId: 35,
|
|
SkipTopic: topic,
|
|
},
|
|
}
|
|
raw, err := json.Marshal(msg)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
deliverRealtimeFanoutEnvelope(&store.RealtimeFanoutEnvelope{
|
|
TenantID: tenantID,
|
|
Kind: realtimeKindMessage,
|
|
TargetType: realtimeTargetUser,
|
|
Target: recipient.UserId(),
|
|
SkipTopic: topic,
|
|
Message: raw,
|
|
})
|
|
|
|
if len(globals.sessionStore.Get("sid-attached").send) != 0 {
|
|
t.Fatal("attached p2p session must not receive duplicate me presence")
|
|
}
|
|
select {
|
|
case got := <-globals.sessionStore.Get("sid-me-only").send:
|
|
pres := got.(*ServerComMessage).Pres
|
|
if pres == nil || pres.Topic != "me" || pres.What != "msg" || pres.Src != topic || pres.SeqId != 35 {
|
|
t.Fatalf("unexpected presence: %#v", pres)
|
|
}
|
|
default:
|
|
t.Fatal("me-only session must receive message presence")
|
|
}
|
|
}
|
|
|
|
func TestRealtimeP2PTopicFallbackRewritesVisibleTopic(t *testing.T) {
|
|
oldStore := globals.sessionStore
|
|
oldHub := globals.hub
|
|
defer func() {
|
|
globals.sessionStore = oldStore
|
|
globals.hub = oldHub
|
|
}()
|
|
|
|
tenantID := types.TenantID(1)
|
|
uid1 := types.Uid(1001)
|
|
uid2 := types.Uid(1002)
|
|
topic := uid1.P2PName(uid2)
|
|
if topic == "" {
|
|
t.Fatal("failed to create p2p topic")
|
|
}
|
|
globals.hub = nil
|
|
globals.sessionStore = realtimeTestSessionStore(
|
|
realtimeTestSession("sid-u1", tenantID, uid1, uid2.UserId()),
|
|
realtimeTestSession("sid-u2", tenantID, uid2, uid1.UserId()),
|
|
)
|
|
msg := &ServerComMessage{
|
|
Data: &MsgServerData{Topic: topic, From: uid1.UserId(), SeqId: 13, Timestamp: time.Now()},
|
|
}
|
|
raw, err := json.Marshal(msg)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
deliverRealtimeFanoutEnvelope(&store.RealtimeFanoutEnvelope{
|
|
TenantID: tenantID,
|
|
Kind: realtimeKindMessage,
|
|
TargetType: realtimeTargetTopic,
|
|
Target: topic,
|
|
Message: raw,
|
|
})
|
|
|
|
expectTopic := map[string]string{"sid-u1": uid2.UserId(), "sid-u2": uid1.UserId()}
|
|
for sid, visibleTopic := range expectTopic {
|
|
sess := globals.sessionStore.Get(sid)
|
|
select {
|
|
case got := <-sess.send:
|
|
data := got.(*ServerComMessage).Data
|
|
if data.Topic != visibleTopic {
|
|
t.Fatalf("%s got topic %q, want %q", sid, data.Topic, visibleTopic)
|
|
}
|
|
default:
|
|
t.Fatalf("expected p2p delivery to %s", sid)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestRealtimeSessionControlStaleEvictDropped(t *testing.T) {
|
|
oldStore := globals.sessionStore
|
|
defer func() { globals.sessionStore = oldStore }()
|
|
|
|
tenantID := types.TenantID(1)
|
|
uid := types.Uid(1001)
|
|
globals.sessionStore = realtimeTestSessionStore(
|
|
realtimeTestPlatformSession("sid-live", tenantID, uid, "web"),
|
|
)
|
|
|
|
envelope := func(ts time.Time) *store.RealtimeFanoutEnvelope {
|
|
msg := &ServerComMessage{
|
|
Ctrl: &MsgServerCtrl{
|
|
Code: 205,
|
|
Text: "evicted",
|
|
Timestamp: ts,
|
|
Params: map[string]any{"reason": "replaced_by_new_login"},
|
|
},
|
|
}
|
|
raw, err := json.Marshal(msg)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return &store.RealtimeFanoutEnvelope{
|
|
TenantID: tenantID,
|
|
Kind: realtimeKindSessionControl,
|
|
TargetType: realtimeTargetPlatf,
|
|
Target: uid.UserId() + ":web",
|
|
SkipSid: "sid-new",
|
|
Message: raw,
|
|
}
|
|
}
|
|
|
|
// A stale replay (old skipSid != current session sid) must not evict.
|
|
deliverRealtimeFanoutEnvelope(envelope(time.Now().Add(-2 * fanoutSessionControlStaleLimit)))
|
|
if globals.sessionStore.Get("sid-live") == nil {
|
|
t.Fatal("stale evict must not terminate the session")
|
|
}
|
|
|
|
// A fresh evict must evict as usual.
|
|
deliverRealtimeFanoutEnvelope(envelope(time.Now()))
|
|
if globals.sessionStore.Get("sid-live") != nil {
|
|
t.Fatal("fresh evict must terminate the session")
|
|
}
|
|
}
|
|
|
|
func TestRealtimeP2PTopicFallbackRewritesVisibleInfoTopic(t *testing.T) {
|
|
oldStore := globals.sessionStore
|
|
oldHub := globals.hub
|
|
defer func() {
|
|
globals.sessionStore = oldStore
|
|
globals.hub = oldHub
|
|
}()
|
|
|
|
tenantID := types.TenantID(1)
|
|
uid1 := types.Uid(1001)
|
|
uid2 := types.Uid(1002)
|
|
topic := uid1.P2PName(uid2)
|
|
if topic == "" {
|
|
t.Fatal("failed to create p2p topic")
|
|
}
|
|
globals.hub = nil
|
|
globals.sessionStore = realtimeTestSessionStore(
|
|
realtimeTestSession("sid-u1", tenantID, uid1, uid2.UserId()),
|
|
realtimeTestSession("sid-u2", tenantID, uid2, uid1.UserId()),
|
|
)
|
|
msg := &ServerComMessage{
|
|
Info: &MsgServerInfo{Topic: topic, From: uid1.UserId(), What: "read", SeqId: 13},
|
|
}
|
|
raw, err := json.Marshal(msg)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
|
|
deliverRealtimeFanoutEnvelope(&store.RealtimeFanoutEnvelope{
|
|
TenantID: tenantID,
|
|
Kind: realtimeKindMessage,
|
|
TargetType: realtimeTargetTopic,
|
|
Target: topic,
|
|
Message: raw,
|
|
})
|
|
|
|
expectTopic := map[string]string{"sid-u1": uid2.UserId(), "sid-u2": uid1.UserId()}
|
|
for sid, visibleTopic := range expectTopic {
|
|
sess := globals.sessionStore.Get(sid)
|
|
select {
|
|
case got := <-sess.send:
|
|
info := got.(*ServerComMessage).Info
|
|
if info.Topic != visibleTopic {
|
|
t.Fatalf("%s got info topic %q, want %q", sid, info.Topic, visibleTopic)
|
|
}
|
|
if info.What != "read" || info.SeqId != 13 || info.From != uid1.UserId() {
|
|
t.Fatalf("%s got unexpected info %#v", sid, info)
|
|
}
|
|
default:
|
|
t.Fatalf("expected p2p info delivery to %s", sid)
|
|
}
|
|
}
|
|
}
|
|
|
|
func TestRealtimeInfoDedupKeySeparatesReadRecv(t *testing.T) {
|
|
read := realtimeMessageDedupKey("p2pABC", &ServerComMessage{
|
|
Info: &MsgServerInfo{From: "usrA", What: "read", SeqId: 12},
|
|
})
|
|
recv := realtimeMessageDedupKey("p2pABC", &ServerComMessage{
|
|
Info: &MsgServerInfo{From: "usrA", What: "recv", SeqId: 12},
|
|
})
|
|
if read == recv {
|
|
t.Fatalf("read and recv info dedup keys must differ: %q", read)
|
|
}
|
|
if read != "info:p2pABC:read:usrA::12" {
|
|
t.Fatalf("unexpected read dedup key %q", read)
|
|
}
|
|
}
|
|
|
|
func realtimeTestSessionStore(sessions ...*Session) *SessionStore {
|
|
ss := NewSessionStore(time.Minute)
|
|
for _, sess := range sessions {
|
|
ss.sessCache[sess.sid] = sess
|
|
}
|
|
return ss
|
|
}
|
|
|
|
func realtimeTestSession(sid string, tenantID types.TenantID, uid types.Uid, sub string) *Session {
|
|
sess := &Session{
|
|
sid: sid,
|
|
tenantID: tenantID,
|
|
uid: uid,
|
|
send: make(chan any, 4),
|
|
stop: make(chan any, 1),
|
|
subs: make(map[string]*Subscription),
|
|
}
|
|
if sub != "" {
|
|
sess.subs[sub] = &Subscription{}
|
|
}
|
|
return sess
|
|
}
|
|
|
|
func realtimeTestPlatformSession(sid string, tenantID types.TenantID, uid types.Uid, platf string) *Session {
|
|
sess := realtimeTestSession(sid, tenantID, uid, "")
|
|
sess.platf = platf
|
|
return sess
|
|
}
|