feat: scaffold independent workbench service

internal/events/events.go

@@ -0,0 +1,181 @@
+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{
+		Topic: p.topic,
+		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)
+}