im_server/server/kafka_cache_invalidate.go

package main

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

	"github.com/segmentio/kafka-go"
	"github.com/tinode/chat/server/logs"
	"github.com/tinode/chat/server/store"
	"github.com/tinode/chat/server/store/types"
)

const (
	cacheInvalidateEventAll                 = "cache.invalidate"
	cacheInvalidateEventTenant              = "cache.tenant.invalidate"
	cacheInvalidateEventRegistration        = "cache.registration.invalidate"
	cacheInvalidateEventThirdParty          = "cache.third_party.invalidate"
	cacheInvalidateEventDictionary          = "cache.dictionary.invalidate"
	cacheInvalidateEventRedisObject         = "cache.redis_object.invalidate"
	cacheInvalidateEventGroupHotMessages    = "cache.group_hot_messages.invalidate"
	cacheInvalidateNamespaceTenant          = "tenant"
	cacheInvalidateNamespaceRegistration    = "registration"
	cacheInvalidateNamespaceThirdParty      = "third_party"
	cacheInvalidateNamespaceDictionary      = "dictionary"
	cacheInvalidateNamespaceGroupHotMessage = "group_hot_messages"
)

type cacheInvalidatePayload struct {
	TenantCode string         `json:"tenant_code"`
	TenantID   types.TenantID `json:"tenant_id"`
	Namespace  string         `json:"namespace"`
	Group      string         `json:"group"`
	Key        string         `json:"key"`
	Keys       []string       `json:"keys"`
	Topic      string         `json:"topic"`
	Ranges     []types.Range  `json:"ranges"`
}

var kafkaCacheInvalidateConsumerGroup string

func startKafkaCacheInvalidateConsumer() {
	kafkaCacheInvalidateConsumerGroup = startKafkaBroadcastConsumer("cache_invalidate", handleKafkaCacheInvalidateEvent)
}

func handleKafkaCacheInvalidateEvent(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error {
	if strings.TrimSpace(envelope.EventID) == "" {
		return nil
	}
	tenantID, _ := kafkaTenantIDFromEnvelopeOrPayload(envelope)
	if tenantID.IsZero() {
		tenantID = types.TenantID(1)
	}
	consumerGroup := kafkaCacheInvalidateConsumerGroup
	consume, claimed, err := beginKafkaConsumeForTopicWithGroup(tenantID, envelope, "cache_invalidate", consumerGroup)
	if err != nil {
		return err
	}
	if !claimed {
		logs.Info.Printf("kafka cache invalidate skipped: event=%s group=%s status=%s", envelope.EventID, consumerGroup, consume.Status)
		return nil
	}
	if err = executeKafkaCacheInvalidateEvent(envelope); err != nil {
		reason := kafkaCommandErrorReason(err)
		failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, consumerGroup, reason, err.Error(), time.Now().UTC().Round(time.Millisecond))
		if markErr != nil {
			return markErr
		}
		if failed.RetryCount >= kafkaBus.config.MaxRetryCount {
			return markKafkaDeadLetter(ctx, tenantID, envelope, failed, reason, err.Error())
		}
		return err
	}
	return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, consumerGroup, time.Now().UTC().Round(time.Millisecond))
}

func executeKafkaCacheInvalidateEvent(envelope kafkaEnvelope) error {
	var payload cacheInvalidatePayload
	if len(envelope.Payload) > 0 {
		if err := json.Unmarshal(envelope.Payload, &payload); err != nil {
			return err
		}
	}
	tenantCode := firstNonEmpty(payload.TenantCode, envelope.TenantCode)
	tenantID := payload.TenantID
	if tenantID.IsZero() && tenantCode != "" {
		if tenant, err := store.Tenants.GetByCode(tenantCode); err == nil && tenant != nil {
			tenantID = tenant.ID
		}
	}

	eventType := strings.TrimSpace(envelope.EventType)
	namespace := strings.TrimSpace(payload.Namespace)
	switch {
	case eventType == cacheInvalidateEventTenant || namespace == cacheInvalidateNamespaceTenant:
		store.InvalidateTenantConfigCache(tenantCode, tenantID)
	case eventType == cacheInvalidateEventRegistration || namespace == cacheInvalidateNamespaceRegistration:
		store.InvalidateRegistrationConfigCache()
	case eventType == cacheInvalidateEventThirdParty || eventType == cacheInvalidateEventDictionary ||
		namespace == cacheInvalidateNamespaceThirdParty || namespace == cacheInvalidateNamespaceDictionary:
		store.InvalidateThirdPartyConfigCache(payload.Group)
	case eventType == cacheInvalidateEventGroupHotMessages || namespace == cacheInvalidateNamespaceGroupHotMessage:
		if tenantID.IsValid() && strings.TrimSpace(payload.Topic) != "" {
			if err := store.InvalidateHotMessages(tenantID, strings.TrimSpace(payload.Topic), payload.Ranges); err != nil {
				return err
			}
		}
	case eventType == cacheInvalidateEventRedisObject || namespace != "":
		if tenantID.IsValid() && namespace != "" {
			keys := append([]string{}, payload.Keys...)
			if strings.TrimSpace(payload.Key) != "" {
				keys = append(keys, strings.TrimSpace(payload.Key))
			}
			for _, key := range keys {
				key = strings.TrimSpace(key)
				if key == "" {
					continue
				}
				if err := store.DeleteCachedObject(tenantID, namespace, key); err != nil {
					return err
				}
			}
		}
	default:
		store.InvalidateTenantConfigCache(tenantCode, tenantID)
		store.InvalidateThirdPartyConfigCache(payload.Group)
		store.InvalidateRegistrationConfigCache()
	}
	logs.Info.Printf("kafka cache invalidate applied: event=%s type=%s tenant=%s namespace=%s group=%s", envelope.EventID, envelope.EventType, tenantCode, namespace, payload.Group)
	return nil
}