im_server/server/kafka_workbench_audit_test.go

package main

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	"github.com/golang/mock/gomock"
	"github.com/tinode/chat/server/store"
	"github.com/tinode/chat/server/store/mock_store"
	"github.com/tinode/chat/server/store/types"
)

func TestKafkaTopicsKeepOnlyMessageInternalAsyncTopics(t *testing.T) {
	topics := kafkaTopics(nil)
	if topics["message_persist"] != "im.message.persist" {
		t.Fatalf("unexpected private message persist topic: %q", topics["message_persist"])
	}
	if topics["group_message_persist"] != "im.group.message.persist" {
		t.Fatalf("unexpected group message persist topic: %q", topics["group_message_persist"])
	}
	if topics["audit_event"] != "im.audit.event" {
		t.Fatalf("unexpected audit event topic: %q", topics["audit_event"])
	}
	for _, removed := range []string{"file_lifecycle", "workbench_event"} {
		if _, ok := topics[removed]; ok {
			t.Fatalf("internal sync topic %q must not be configured", removed)
		}
	}
}

func TestNormalizeKafkaConfigUsesLowLatencyBatchTimeout(t *testing.T) {
	cfg := kafkaConfig{}
	normalizeKafkaConfig(&cfg)
	if cfg.BatchTimeoutMS != 10 {
		t.Fatalf("unexpected kafka batch timeout: %d", cfg.BatchTimeoutMS)
	}
	if cfg.MaxRetryCount != 3 {
		t.Fatalf("unexpected kafka max retry count: %d", cfg.MaxRetryCount)
	}
}

func TestEnqueueKafkaAuditEventPublishesCrossSystemEvent(t *testing.T) {
	oldKafkaBus := kafkaBus
	oldPublisher := publishKafkaAuditEventEnvelope
	defer func() {
		kafkaBus = oldKafkaBus
		publishKafkaAuditEventEnvelope = oldPublisher
	}()

	kafkaBus = &kafkaRuntime{}
	var publishedTopic string
	var publishedKey string
	var publishedEnvelope kafkaEnvelope
	publishKafkaAuditEventEnvelope = func(_ context.Context, topicName, key string, envelope kafkaEnvelope) error {
		publishedTopic = topicName
		publishedKey = key
		publishedEnvelope = envelope
		return nil
	}

	err := enqueueKafkaAuditEvent(context.Background(), 7, "admin", types.Uid(42),
		"group.update", "group", "grp-test", "succeeded", "", map[string]any{"topic": "grp-test"})
	if err != nil {
		t.Fatalf("enqueueKafkaAuditEvent returned error: %v", err)
	}
	if publishedTopic != "audit_event" {
		t.Fatalf("unexpected audit publish topic: %q", publishedTopic)
	}
	if publishedKey != "group:grp-test" {
		t.Fatalf("unexpected audit publish key: %q", publishedKey)
	}
	if publishedEnvelope.TargetService != "hotgo-admin" || publishedEnvelope.EventType != "audit.group.update" {
		t.Fatalf("unexpected audit envelope: %#v", publishedEnvelope)
	}
	if publishedEnvelope.Actor == nil || publishedEnvelope.Actor.Type != "admin" || publishedEnvelope.Actor.ID != types.Uid(42).UserId() {
		t.Fatalf("unexpected audit actor: %#v", publishedEnvelope.Actor)
	}
	var payload kafkaAuditPayload
	if err = json.Unmarshal(publishedEnvelope.Payload, &payload); err != nil {
		t.Fatalf("invalid audit payload: %v", err)
	}
	if payload.Resource != "group" || payload.ResourceID != "grp-test" || payload.Result != "succeeded" {
		t.Fatalf("unexpected audit payload: %#v", payload)
	}
}

func TestEnqueueKafkaFileAttachmentsBindRunsSynchronously(t *testing.T) {
	ctrl := gomock.NewController(t)
	defer ctrl.Finish()

	oldKafkaBus := kafkaBus
	oldFiles := store.Files
	defer func() {
		kafkaBus = oldKafkaBus
		store.Files = oldFiles
	}()

	kafkaBus = &kafkaRuntime{}
	files := mock_store.NewMockFilePersistenceInterface(ctrl)
	store.Files = files
	msgID := types.Uid(101)
	msg := &types.Message{
		ObjHeader: types.ObjHeader{Id: msgID.String(), CreatedAt: time.Date(2026, 9, 1, 10, 0, 0, 0, time.UTC)},
		TenantID:  7,
		Topic:     "usr42",
		SeqId:     3,
		From:      types.Uid(42).String(),
		Head:      types.KVMap{"x-im-type": "image"},
	}
	attachments := []string{"https://api.118im.com/v1/im/files/file_001"}
	files.EXPECT().LinkAttachments(types.TenantID(7), "", msgID, attachments).Return(nil)

	if err := enqueueKafkaFileAttachmentsBind(context.Background(), 7, msg, attachments); err != nil {
		t.Fatalf("enqueueKafkaFileAttachmentsBind returned error: %v", err)
	}
}

func TestEnqueueKafkaWorkbenchEventRunsSynchronously(t *testing.T) {
	oldKafkaBus := kafkaBus
	oldInfrastructure := store.HTTPInfrastructure
	oldSecurity := store.Security
	defer func() {
		kafkaBus = oldKafkaBus
		store.HTTPInfrastructure = oldInfrastructure
		store.Security = oldSecurity
	}()

	kafkaBus = &kafkaRuntime{}
	infrastructure := &testHTTPInfrastructureStore{}
	security := &testSecurityStore{}
	store.HTTPInfrastructure = infrastructure
	store.Security = security

	err := enqueueKafkaWorkbenchEvent(context.Background(), 7, types.Uid(42), "meeting", "meeting.created", "meeting-1",
		[]types.Uid{42, 43}, "会议通知", "周会", map[string]any{"meeting_id": "meeting-1"})
	if err != nil {
		t.Fatalf("enqueueKafkaWorkbenchEvent returned error: %v", err)
	}
	if len(security.notifications) != 1 {
		t.Fatalf("expected one notification for non-actor target, got %#v", security.notifications)
	}
	if security.notifications[0].UserID != types.Uid(43) || security.notifications[0].EventType != "meeting.created" {
		t.Fatalf("unexpected notification: %#v", security.notifications[0])
	}
	if len(infrastructure.events) != 2 {
		t.Fatalf("expected business events for both targets, got %#v", infrastructure.events)
	}
}