im_server/server/kafka_workbench_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"
)

type kafkaExternalWorkbenchEvent struct {
	EventID       string         `json:"event_id"`
	TraceID       string         `json:"trace_id,omitempty"`
	TenantID      types.TenantID `json:"tenant_id,omitempty"`
	TenantCode    string         `json:"tenant_code,omitempty"`
	ActorUserID   int64          `json:"actor_user_id"`
	Module        string         `json:"module"`
	Action        string         `json:"action"`
	ResourceID    string         `json:"resource_id"`
	TargetUserIDs []int64        `json:"target_user_ids,omitempty"`
	Title         string         `json:"title,omitempty"`
	Summary       string         `json:"summary,omitempty"`
	OccurredAt    time.Time      `json:"occurred_at"`
	Data          map[string]any `json:"data,omitempty"`
}

func startKafkaWorkbenchEventConsumer() {
	startKafkaConsumer("workbench_external", handleKafkaWorkbenchEvent)
}

func handleKafkaWorkbenchEvent(ctx context.Context, envelope kafkaEnvelope, msg kafka.Message) error {
	payload, normalized, err := decodeKafkaWorkbenchEvent(envelope, msg)
	if err != nil {
		logs.Warn.Printf("kafka workbench event malformed: topic=%s offset=%d err=%v", msg.Topic, msg.Offset, err)
		return err
	}
	tenantID, err := kafkaTenantIDFromEnvelope(normalized)
	if err != nil {
		return err
	}
	consume, claimed, err := beginKafkaConsumeForTopic(tenantID, normalized, "workbench_external")
	if err != nil {
		return err
	}
	if !claimed {
		logs.Info.Printf("kafka workbench event skipped: event=%s group=%s status=%s", normalized.EventID, kafkaBus.config.ConsumerGroup, consume.Status)
		return nil
	}
	if err = executeKafkaExternalWorkbenchEvent(tenantID, normalized, payload); err != nil {
		reason := kafkaCommandErrorReason(err)
		failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, normalized.EventID, kafkaBus.config.ConsumerGroup, reason, err.Error(), time.Now().UTC().Round(time.Millisecond))
		if markErr != nil {
			return markErr
		}
		if failed != nil && failed.RetryCount >= kafkaBus.config.MaxRetryCount {
			return markKafkaDeadLetter(ctx, tenantID, normalized, failed, reason, err.Error())
		}
		if failed != nil {
			sleepKafkaRetryBackoff(failed.RetryCount)
		}
		return err
	}
	return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, normalized.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond))
}

func decodeKafkaWorkbenchEvent(envelope kafkaEnvelope, msg kafka.Message) (kafkaExternalWorkbenchEvent, kafkaEnvelope, error) {
	var payload kafkaExternalWorkbenchEvent
	if len(envelope.Payload) > 0 {
		if err := json.Unmarshal(envelope.Payload, &payload); err != nil {
			return payload, envelope, err
		}
	} else if err := json.Unmarshal(msg.Value, &payload); err != nil {
		return payload, envelope, err
	}
	if strings.TrimSpace(payload.EventID) == "" {
		payload.EventID = envelope.EventID
	}
	if strings.TrimSpace(payload.Action) == "" && envelope.EventType != "" {
		payload.Action = strings.TrimPrefix(envelope.EventType, "workbench.")
	}
	if strings.TrimSpace(payload.EventID) == "" || strings.TrimSpace(payload.Action) == "" || strings.TrimSpace(payload.ResourceID) == "" {
		return payload, envelope, types.ErrMalformed
	}
	if payload.OccurredAt.IsZero() {
		payload.OccurredAt = time.Now().UTC().Round(time.Millisecond)
	}
	normalized := kafkaEnvelope{
		EventID:       payload.EventID,
		TraceID:       firstNonEmpty(payload.TraceID, envelope.TraceID),
		TenantCode:    firstNonEmpty(payload.TenantCode, envelope.TenantCode),
		TenantID:      firstTenantID(payload.TenantID, envelope.TenantID),
		SourceService: firstNonEmpty(envelope.SourceService, "im-workbench"),
		TargetService: firstNonEmpty(envelope.TargetService, "im-server"),
		EventType:     "workbench." + payload.Action,
		BizType:       firstNonEmpty(payload.Module, envelope.BizType, "workbench"),
		BizID:         firstNonEmpty(payload.ResourceID, envelope.BizID),
		OccurredAt:    payload.OccurredAt,
		Payload:       msg.Value,
	}
	if payload.ActorUserID > 0 {
		normalized.Actor = &kafkaActor{Type: "user", ID: types.Uid(payload.ActorUserID).UserId()}
	}
	return payload, normalized, nil
}

func executeKafkaExternalWorkbenchEvent(tenantID types.TenantID, envelope kafkaEnvelope, payload kafkaExternalWorkbenchEvent) error {
	if store.Security == nil {
		return types.ErrUnsupported
	}
	actor := types.Uid(payload.ActorUserID)
	data := types.KVMap{
		"module":      payload.Module,
		"resource_id": payload.ResourceID,
	}
	for k, v := range payload.Data {
		data[k] = v
	}
	targets := compactWorkbenchTargets(payload.TargetUserIDs, actor)
	for _, target := range targets {
		targetID := target.UserId()
		eventID := stableIMEventID("workbench", envelope.EventID, targetID)
		if target != actor && payload.Title != "" {
			if err := createIMSystemNotification(tenantID, target, stableIMEventID(eventID, "notification"), payload.Action,
				payload.Title, payload.Summary, data); err != nil {
				return err
			}
		}
		if err := enqueueIMBusinessEvent(tenantID, eventID, "user", targetID, actor, payload.Action, data); err != nil {
			return err
		}
	}
	return nil
}

func compactWorkbenchTargets(raw []int64, actor types.Uid) []types.Uid {
	seen := make(map[types.Uid]bool, len(raw)+1)
	targets := make([]types.Uid, 0, len(raw)+1)
	for _, value := range raw {
		uid := types.Uid(value)
		if uid.IsZero() || seen[uid] {
			continue
		}
		seen[uid] = true
		targets = append(targets, uid)
	}
	if len(targets) == 0 && !actor.IsZero() {
		targets = append(targets, actor)
	}
	return targets
}

func firstTenantID(values ...types.TenantID) types.TenantID {
	for _, value := range values {
		if !value.IsZero() {
			return value
		}
	}
	return types.ZeroTenantID
}