im_server/server/kafka_config_events.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 (
	configEventTenantUpdated         = "tenant.updated"
	configEventTenantStatusChanged   = "tenant.status.changed"
	configEventTenantLimitsUpdated   = "tenant.limits.updated"
	configEventTenantFeaturesUpdated = "tenant.features.updated"
	configEventDictionaryUpdated     = "dictionary.updated"
	configEventThirdPartyUpdated     = "third_party.updated"
	configEventRegistrationUpdated   = "registration.config.updated"
	configEventAppReleaseUpdated     = "app.release.updated"
	configEventSecurityPolicyUpdated = "security.policy.updated"
	configEventSensitiveWordsUpdated = "sensitive_words.updated"
)

type configEventPayload struct {
	TenantCode string         `json:"tenant_code"`
	TenantID   types.TenantID `json:"tenant_id"`
	Group      string         `json:"group"`
	Keys       []string       `json:"keys"`
}

func startKafkaConfigEventConsumer() {
	startKafkaConsumer("config_event", handleKafkaConfigEvent)
}

func handleKafkaConfigEvent(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)
	}
	consume, claimed, err := beginKafkaConsumeForTopic(tenantID, envelope, "config_event")
	if err != nil {
		return err
	}
	if !claimed {
		logs.Info.Printf("kafka config event skipped: event=%s group=%s status=%s", envelope.EventID, kafkaBus.config.ConsumerGroup, consume.Status)
		return nil
	}
	if err = executeKafkaConfigEvent(envelope); err != nil {
		reason := kafkaCommandErrorReason(err)
		failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.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())
		}
		sleepKafkaRetryBackoff(failed.RetryCount)
		return err
	}
	return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond))
}

func executeKafkaConfigEvent(envelope kafkaEnvelope) error {
	var payload configEventPayload
	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
		}
	}

	switch strings.TrimSpace(envelope.EventType) {
	case configEventTenantUpdated, configEventTenantStatusChanged, configEventTenantLimitsUpdated, configEventTenantFeaturesUpdated:
		store.InvalidateTenantConfigCache(tenantCode, tenantID)
		if strings.TrimSpace(envelope.EventType) == configEventTenantFeaturesUpdated {
			store.InvalidateModerationConfigCache(tenantCode, false, true, false)
		}
	case configEventDictionaryUpdated, configEventThirdPartyUpdated:
		store.InvalidateThirdPartyConfigCache(payload.Group)
		if payload.Group == "moderation" {
			store.InvalidateModerationConfigCache(tenantCode, true, false, false)
		}
	case configEventRegistrationUpdated:
		store.InvalidateRegistrationConfigCache()
	case configEventAppReleaseUpdated:
		store.InvalidateAppReleaseConfigCache()
	case configEventSecurityPolicyUpdated:
		store.InvalidateModerationConfigCache(tenantCode, false, true, false)
	case configEventSensitiveWordsUpdated:
		store.InvalidateModerationConfigCache(tenantCode, false, false, true)
	default:
		store.InvalidateTenantConfigCache(tenantCode, tenantID)
		store.InvalidateThirdPartyConfigCache(payload.Group)
		store.InvalidateRegistrationConfigCache()
		store.InvalidateAppReleaseConfigCache()
		store.InvalidateModerationConfigCache(tenantCode, true, true, true)
	}
	logs.Info.Printf("kafka config event applied: event=%s type=%s tenant=%s group=%s", envelope.EventID, envelope.EventType, tenantCode, payload.Group)
	return nil
}

func kafkaTenantIDFromEnvelopeOrPayload(envelope kafkaEnvelope) (types.TenantID, error) {
	var payload configEventPayload
	if len(envelope.Payload) > 0 {
		_ = json.Unmarshal(envelope.Payload, &payload)
	}
	if payload.TenantID.IsValid() {
		return payload.TenantID, nil
	}
	return kafkaTenantID(firstNonEmpty(envelope.TenantCode, payload.TenantCode))
}

func beginKafkaConsumeForTopic(tenantID types.TenantID, envelope kafkaEnvelope, topicName string) (*types.KafkaConsumeRecord, bool, error) {
	if kafkaBus == nil {
		return nil, false, types.ErrUnsupported
	}
	deadline := time.Now().UTC().Add(5 * time.Minute).Round(time.Millisecond)
	return store.HTTPInfrastructure.BeginKafkaConsume(tenantID, &types.KafkaConsumeRecord{
		EventID:       envelope.EventID,
		ConsumerGroup: kafkaBus.config.ConsumerGroup,
		SourceTopic:   kafkaTopic(topicName),
		EventType:     envelope.EventType,
		BizType:       envelope.BizType,
		BizID:         envelope.BizID,
	}, deadline)
}