diff --git a/docs/sql/mysql-multitenant-tenant-id-review.sql b/docs/sql/mysql-multitenant-tenant-id-review.sql index 656b966..7cefad8 100644 --- a/docs/sql/mysql-multitenant-tenant-id-review.sql +++ b/docs/sql/mysql-multitenant-tenant-id-review.sql @@ -249,9 +249,9 @@ ALTER TABLE im_topictags ALTER TABLE im_subscriptions ADD UNIQUE KEY uk_subscriptions_tenant_topic_user (tenant_id, topic, userid), - ADD KEY idx_subscriptions_tenant_user_topic_deleted + ADD KEY idx_subscriptions_tenant_user_topic_deletedat (tenant_id, userid, topic, deletedat), - ADD KEY idx_subscriptions_tenant_topic_deleted + ADD KEY idx_subscriptions_tenant_topic_deletedat (tenant_id, topic, deletedat), DROP INDEX subscriptions_topic_userid; diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index 187bc93..1314e42 100644 --- a/server/store/group_message_runtime.go +++ b/server/store/group_message_runtime.go @@ -1497,7 +1497,9 @@ func (r *redisGroupMessageRuntime) loginSessionKey(tokenHash string) string { func (r *redisGroupMessageRuntime) cacheTTL(namespace string) time.Duration { ttl := r.cfg.CacheTTL - if namespace == "verification" || namespace == "action_token" || namespace == "login_lockout" { + if namespace == cacheNamespaceTopic || namespace == cacheNamespaceSub { + ttl = 60 + } else if namespace == "verification" || namespace == "action_token" || namespace == "login_lockout" { ttl = r.cfg.ShortTTL } return time.Duration(ttl) * time.Second diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index a334358..2231d43 100644 --- a/server/store/group_message_runtime_test.go +++ b/server/store/group_message_runtime_test.go @@ -409,3 +409,28 @@ func TestInvalidateGroupSettingsUsesTopicCacheKey(t *testing.T) { t.Fatalf("unexpected deleted objects: %#v", runtime.deletedObjects) } } + +func TestTopicAndSubscriptionCacheTTLIsShort(t *testing.T) { + runtime := &redisGroupMessageRuntime{cfg: redisConfig{CacheTTL: 300, ShortTTL: 600}} + for _, namespace := range []string{cacheNamespaceTopic, cacheNamespaceSub} { + if got := runtime.cacheTTL(namespace); got != time.Minute { + t.Fatalf("cache TTL for %s = %v, want 1m", namespace, got) + } + } + if got := runtime.cacheTTL(cacheNamespaceGroupSettings); got != 5*time.Minute { + t.Fatalf("cache TTL for group settings = %v, want 5m", got) + } +} + +func TestInvalidateSubscriptionCacheWithZeroUserDeletesTopicPrefix(t *testing.T) { + oldRuntime := groupMsgRuntime + runtime := newTestGroupMessageRuntime() + groupMsgRuntime = runtime + t.Cleanup(func() { groupMsgRuntime = oldRuntime }) + + invalidateSubscriptionCache(7, "grpAlpha", types.ZeroUid) + want := []string{"grpAlpha:", "chnAlpha:"} + if len(runtime.deletedPrefixes) != len(want) || runtime.deletedPrefixes[0] != want[0] || runtime.deletedPrefixes[1] != want[1] { + t.Fatalf("unexpected deleted prefixes: %#v", runtime.deletedPrefixes) + } +} diff --git a/server/store/store.go b/server/store/store.go index 4d51448..588d027 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -1395,7 +1395,11 @@ func (topicsMapper) UpdateSubCnt(tenantID types.TenantID, topic string) error { if err != nil { return err } - return tenantAdp.TenantTopicUpdateSubCnt(tenantID, topic) + if err = tenantAdp.TenantTopicUpdateSubCnt(tenantID, topic); err != nil { + return err + } + invalidateTopicCache(tenantID, topic) + return nil } // Update is a generic topic update. @@ -1410,7 +1414,7 @@ func (topicsMapper) Update(tenantID types.TenantID, topic string, update map[str if err = tenantAdp.TenantTopicUpdate(tenantID, topic, update); err != nil { return err } - _ = DeleteCachedObject(tenantID, cacheNamespaceTopic, topic) + invalidateTopicCache(tenantID, topic) return nil } @@ -1421,7 +1425,7 @@ func (topicsMapper) BumpMutationVersion(tenantID types.TenantID, topic string, a } version, err := tenantAdp.TenantTopicBumpMutationVersion(tenantID, topic, at) if err == nil { - _ = DeleteCachedObject(tenantID, cacheNamespaceTopic, topic) + invalidateTopicCache(tenantID, topic) } return version, err } @@ -1435,7 +1439,7 @@ func (topicsMapper) OwnerChange(tenantID types.TenantID, topic string, newOwner if err = tenantAdp.TenantTopicOwnerChange(tenantID, topic, newOwner); err != nil { return err } - _ = DeleteCachedObject(tenantID, cacheNamespaceTopic, topic) + invalidateTopicCache(tenantID, topic) return nil } @@ -1448,7 +1452,8 @@ func (topicsMapper) Delete(tenantID types.TenantID, topic string, isChan, hard b if err = tenantAdp.TenantTopicDelete(tenantID, topic, isChan, hard); err != nil { return err } - _ = DeleteCachedObject(tenantID, cacheNamespaceTopic, topic) + invalidateTopicCache(tenantID, topic) + invalidateTopicSubscriptions(tenantID, topic) if hard && types.GetTopicCat(topic) == types.TopicCatGrp { _ = DeleteGroupMessageShardRoutes(tenantID, topic) } @@ -1541,8 +1546,7 @@ func (subsMapper) Update(tenantID types.TenantID, topic string, user types.Uid, if err = tenantAdp.TenantSubsUpdate(tenantID, topic, user, update); err != nil { return err } - _ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, user, false)) - _ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, user, true)) + invalidateSubscriptionCache(tenantID, topic, user) return nil } @@ -1556,8 +1560,7 @@ func (subsMapper) Delete(tenantID types.TenantID, topic string, user types.Uid) if err = tenantAdp.TenantSubsDelete(tenantID, topic, user); err != nil { return err } - _ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, user, false)) - _ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, user, true)) + invalidateSubscriptionCache(tenantID, topic, user) return nil } @@ -1568,6 +1571,32 @@ func subscriptionCacheID(topic string, user types.Uid, keepDeleted bool) string return topic + ":" + user.String() + ":active" } +func invalidateTopicCache(tenantID types.TenantID, topic string) { + topic = strings.TrimSpace(topic) + if topic == "" { + return + } + if err := DeleteCachedObject(tenantID, cacheNamespaceTopic, topic); err != nil { + logs.Warn.Printf("cache topic[%s]: delete failed: %v", topic, err) + } +} + +func invalidateSubscriptionCache(tenantID types.TenantID, topic string, user types.Uid) { + topic = strings.TrimSpace(topic) + if topic == "" { + return + } + if user.IsZero() { + invalidateTopicSubscriptions(tenantID, topic) + return + } + for _, keepDeleted := range []bool{false, true} { + if err := DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, user, keepDeleted)); err != nil { + logs.Warn.Printf("cache sub[%s]: delete failed: %v", subscriptionCacheID(topic, user, keepDeleted), err) + } + } +} + // MessagesPersistenceInterface is an interface which defines methods for persistent storage of messages. type MessagesPersistenceInterface interface { Save(tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) (*types.MessageSaveResult, error) @@ -3459,6 +3488,7 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message if err != nil { return nil, err } + invalidateTopicCache(tenantID, msg.Topic) return &types.MessageSaveResult{SeqId: msg.SeqId}, nil } @@ -3536,6 +3566,7 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg); err != nil { return nil, err } + invalidateTopicCache(tenantID, msg.Topic) if isGroupMessage && groupMsgRuntime != nil && clientMsgID != "" { if err := groupMsgRuntime.MarkDone(tenantID, msg.Topic, fromUid, clientMsgID, msg.SeqId, msg.Uid().String()); err != nil { logs.Warn.Printf("topic[%s]: failed to mark group message unique key done: %v", msg.Topic, err) @@ -3564,6 +3595,7 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag "ReadSeqId": msg.SeqId}); subErr != nil { logs.Warn.Printf("topic[%s]: failed to mark message (seq: %d) read by sender - err: %+v", msg.Topic, msg.SeqId, subErr) } else { + invalidateSubscriptionCache(tenantID, msg.Topic, fromUid) markedReadBySender = true } } @@ -3659,6 +3691,7 @@ func retryTenantMessageSaveAfterSeqConflict(tenantAdp adapter.TenantBusinessAdap if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg); err != nil { return err } + invalidateTopicCache(tenantID, msg.Topic) if isGroupMessage { err = tenantAdp.TenantGroupMessageSave(tenantID, msg) } else { @@ -3959,6 +3992,7 @@ func (messagesMapper) DeleteList(tenantID types.TenantID, topic string, delID in if err != nil { return err } + invalidateTopicCache(tenantID, topic) // Soft-deleting will update one subscription, hard-deleting will ipdate all. // Soft- or hard- is defined by the forUser being defined. @@ -3966,6 +4000,7 @@ func (messagesMapper) DeleteList(tenantID types.TenantID, topic string, delID in if err != nil { return err } + invalidateSubscriptionCache(tenantID, topic, forUser) } return err