olons_server/internal/events/events.go

package events

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"strconv"
	"strings"
	"time"

	"git.118im.com/im_team/im_workbench/internal/auth"
	"git.118im.com/im_team/im_workbench/internal/config"
	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl"
	"github.com/segmentio/kafka-go/sasl/scram"
)

type Event struct {
	EventID       string         `json:"event_id"`
	TraceID       string         `json:"trace_id,omitempty"`
	TenantID      int64          `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"`
}

type Publisher interface {
	Publish(context.Context, Event) error
	Close() error
}

type NoopPublisher struct{}

func (NoopPublisher) Publish(context.Context, Event) error { return nil }
func (NoopPublisher) Close() error                         { return nil }

type KafkaPublisher struct {
	writer *kafka.Writer
	topic  string
}

func NewKafkaPublisher(cfg config.KafkaConfig) (Publisher, error) {
	if !cfg.Enabled || len(cfg.Brokers) == 0 {
		return NoopPublisher{}, nil
	}
	topic := strings.TrimSpace(cfg.Topic)
	if topic == "" {
		topic = "im.workbench.event"
	}
	transport := &kafka.Transport{}
	if mechanism, err := saslMechanism(cfg); err != nil {
		return nil, err
	} else if mechanism != nil {
		transport.SASL = mechanism
	}
	writer := &kafka.Writer{
		Addr:         kafka.TCP(cfg.Brokers...),
		Topic:        topic,
		Balancer:     &kafka.Hash{},
		Transport:    transport,
		BatchSize:    cfg.BatchSize,
		RequiredAcks: kafka.RequireAll,
		Async:        false,
	}
	if writer.BatchSize <= 0 {
		writer.BatchSize = 1
	}
	return &KafkaPublisher{writer: writer, topic: topic}, nil
}

func (p *KafkaPublisher) Publish(ctx context.Context, event Event) error {
	if p == nil || p.writer == nil {
		return nil
	}
	if strings.TrimSpace(event.EventID) == "" {
		event.EventID = StableEventID(event.TenantID, event.Module, event.Action, event.ResourceID, event.ActorUserID)
	}
	if event.OccurredAt.IsZero() {
		event.OccurredAt = time.Now().UTC().Round(time.Millisecond)
	}
	data, err := json.Marshal(event)
	if err != nil {
		return err
	}
	key := []byte(event.TenantCode + ":" + event.ResourceID)
	if len(key) == 1 {
		key = []byte(event.EventID)
	}
	return p.writer.WriteMessages(ctx, kafka.Message{
		Key:   key,
		Value: data,
		Headers: []kafka.Header{
			{Key: "event_id", Value: []byte(event.EventID)},
			{Key: "event_type", Value: []byte(event.Action)},
			{Key: "tenant_code", Value: []byte(event.TenantCode)},
		},
		Time: event.OccurredAt,
	})
}

func (p *KafkaPublisher) Close() error {
	if p == nil || p.writer == nil {
		return nil
	}
	return p.writer.Close()
}

func NewWorkbenchEvent(p *auth.Principal, traceID, module, action, resourceID string, targets []int64, title, summary string, data map[string]any) Event {
	event := Event{
		TraceID:       strings.TrimSpace(traceID),
		TenantID:      p.TenantID,
		TenantCode:    p.TenantCode,
		ActorUserID:   p.UserID,
		Module:        strings.TrimSpace(module),
		Action:        strings.TrimSpace(action),
		ResourceID:    strings.TrimSpace(resourceID),
		TargetUserIDs: compactInt64s(targets),
		Title:         strings.TrimSpace(title),
		Summary:       strings.TrimSpace(summary),
		OccurredAt:    time.Now().UTC().Round(time.Millisecond),
		Data:          data,
	}
	event.EventID = StableEventID(event.TenantID, event.Module, event.Action, event.ResourceID, event.ActorUserID)
	return event
}

func StableEventID(tenantID int64, module, action, resourceID string, actorUserID int64) string {
	raw := strings.Join([]string{
		int64String(tenantID),
		strings.TrimSpace(module),
		strings.TrimSpace(action),
		strings.TrimSpace(resourceID),
		int64String(actorUserID),
	}, ":")
	sum := sha256.Sum256([]byte(raw))
	return "wbe_" + hex.EncodeToString(sum[:])[:32]
}

func saslMechanism(cfg config.KafkaConfig) (sasl.Mechanism, error) {
	if strings.TrimSpace(cfg.Username) == "" {
		return nil, nil
	}
	switch strings.ToUpper(strings.TrimSpace(cfg.SASL)) {
	case "", "SCRAM-SHA-512":
		return scram.Mechanism(scram.SHA512, cfg.Username, cfg.Password)
	case "SCRAM-SHA-256":
		return scram.Mechanism(scram.SHA256, cfg.Username, cfg.Password)
	default:
		return nil, errors.New("unsupported kafka sasl mechanism")
	}
}

func compactInt64s(values []int64) []int64 {
	if len(values) == 0 {
		return nil
	}
	seen := make(map[int64]bool, len(values))
	items := make([]int64, 0, len(values))
	for _, value := range values {
		if value <= 0 || seen[value] {
			continue
		}
		seen[value] = true
		items = append(items, value)
	}
	return items
}

func int64String(value int64) string {
	return strconv.FormatInt(value, 10)
}