feat: cache topic and subscription reads

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;
 

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

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)
+	}
+}

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))
+	invalidateSubscriptionCache(tenantID, topic, user)
-	_ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, user, true))
 	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))
+	invalidateSubscriptionCache(tenantID, topic, user)
-	_ = DeleteCachedObject(tenantID, cacheNamespaceSub, subscriptionCacheID(topic, user, true))
 	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