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/push"
	"github.com/tinode/chat/server/store"
	"github.com/tinode/chat/server/store/types"
)

type kafkaWorkbenchEvent struct {
	EventID       string         `json:"event_id"`
	TraceID       string         `json:"trace_id,omitempty"`
	TenantID      types.TenantID `json:"tenant_id"`
	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 {
	var event kafkaWorkbenchEvent
	if err := json.Unmarshal(msg.Value, &event); err != nil {
		return err
	}
	event.normalize()
	if event.EventID == "" || event.TenantID.IsZero() || event.Action == "" || event.ResourceID == "" {
		return types.ErrMalformed
	}
	envelope = event.kafkaEnvelope(envelope)
	_, claimed, err := beginKafkaConsumeForTopic(event.TenantID, envelope, "workbench_external")
	if err != nil || !claimed {
		return err
	}
	if err = executeKafkaWorkbenchEvent(event); err != nil {
		reason := kafkaCommandErrorReason(err)
		failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(event.TenantID, event.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, event.TenantID, envelope, failed, reason, err.Error())
		}
		if failed != nil {
			sleepKafkaRetryBackoff(failed.RetryCount)
		}
		return err
	}
	return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(event.TenantID, event.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond))
}

func executeKafkaWorkbenchEvent(event kafkaWorkbenchEvent) error {
	actor := store.EncodeUid(event.ActorUserID)
	payload, err := event.realtimePayload()
	if err != nil {
		return err
	}
	pushTargets := make(map[types.Uid]push.Recipient)
	for _, targetID := range event.TargetUserIDs {
		target := store.EncodeUid(targetID)
		if target.IsZero() {
			continue
		}
		if target != actor && event.Title != "" {
			action := event.notificationAction()
			if err = createIMSystemNotification(event.TenantID, target, stableIMEventID(event.EventID, "notification", target.UserId()),
				event.Action, event.Title, event.Summary, action); err != nil && err != types.ErrDuplicate {
				return err
			}
			pushTargets[target] = push.Recipient{}
		}
		if err = enqueueIMBusinessEvent(event.TenantID, event.EventID, "user", target.UserId(), actor, event.Action, payload); err != nil {
			return err
		}
	}
	if len(pushTargets) > 0 {
		sendPush(&push.Receipt{TenantID: event.TenantID, To: pushTargets, Payload: push.Payload{
			What: push.ActMsg, Topic: "sys", Timestamp: time.Now().UTC(), ContentType: "application/json",
			Content: map[string]any{"event_id": event.EventID, "event_type": event.Action, "title": event.Title, "summary": event.Summary, "action": event.notificationAction()},
		}})
	}
	return nil
}

func (e *kafkaWorkbenchEvent) normalize() {
	e.EventID = strings.TrimSpace(e.EventID)
	e.TenantCode = strings.TrimSpace(e.TenantCode)
	e.Module = strings.TrimSpace(e.Module)
	e.Action = strings.TrimSpace(e.Action)
	e.ResourceID = strings.TrimSpace(e.ResourceID)
	e.Title = strings.TrimSpace(e.Title)
	e.Summary = strings.TrimSpace(e.Summary)
	if e.OccurredAt.IsZero() {
		e.OccurredAt = time.Now().UTC().Round(time.Millisecond)
	}
	e.TargetUserIDs = compactInt64IDs(e.TargetUserIDs)
}

func (e kafkaWorkbenchEvent) kafkaEnvelope(base kafkaEnvelope) kafkaEnvelope {
	base.EventID = e.EventID
	base.TraceID = firstNonEmpty(base.TraceID, e.TraceID)
	base.TenantCode = firstNonEmpty(base.TenantCode, e.TenantCode)
	base.TenantID = e.TenantID
	base.EventType = e.Action
	base.BizType = e.Module
	base.BizID = e.ResourceID
	base.OccurredAt = e.OccurredAt
	return base
}

func (e kafkaWorkbenchEvent) realtimePayload() (map[string]any, error) {
	payload := make(map[string]any, len(e.Data)+3)
	for key, value := range e.Data {
		if strings.TrimSpace(key) != "" {
			payload[key] = value
		}
	}
	payload["module"] = e.Module
	payload["resource_id"] = e.ResourceID
	payload["event_id"] = e.EventID
	return payload, nil
}

func (e kafkaWorkbenchEvent) notificationAction() types.KVMap {
	action := types.KVMap{
		"module":      e.Module,
		"resource_id": e.ResourceID,
		"event_id":    e.EventID,
	}
	for key, value := range e.Data {
		if strings.TrimSpace(key) != "" {
			action[key] = value
		}
	}
	return action
}

func compactInt64IDs(ids []int64) []int64 {
	out := make([]int64, 0, len(ids))
	seen := make(map[int64]bool, len(ids))
	for _, id := range ids {
		if id <= 0 || seen[id] {
			continue
		}
		seen[id] = true
		out = append(out, id)
	}
	return out
}