From 87ec8f98333858240e8a1051a3903257bc2083d9 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Wed, 14 Oct 2026 12:13:33 +0800 Subject: [PATCH] fix kafka attachment message id binding --- server/kafka_audit_test.go | 11 +++++++++-- server/kafka_file_lifecycle.go | 9 ++++++++- 2 files changed, 17 insertions(+), 3 deletions(-) diff --git a/server/kafka_audit_test.go b/server/kafka_audit_test.go index 09687bd..145f7a7 100644 --- a/server/kafka_audit_test.go +++ b/server/kafka_audit_test.go @@ -93,6 +93,12 @@ func TestEnqueueKafkaAuditEventPublishesCrossSystemEvent(t *testing.T) { func TestEnqueueKafkaFileAttachmentsBindRunsSynchronously(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() + var uidGenerator types.UidGenerator + if err := uidGenerator.Init(1, []byte("0123456789abcdef")); err != nil { + t.Fatalf("initialize UID generator: %v", err) + } + store.SetTestUidGenerator(uidGenerator) + defer store.SetTestUidGenerator(types.UidGenerator{}) oldKafkaBus := kafkaBus oldFiles := store.Files @@ -104,7 +110,8 @@ func TestEnqueueKafkaFileAttachmentsBindRunsSynchronously(t *testing.T) { kafkaBus = &kafkaRuntime{} files := mock_store.NewMockFilePersistenceInterface(ctrl) store.Files = files - msgID := types.Uid(101) + databaseMsgID := types.Uid(101) + msgID := store.EncodeUid(int64(databaseMsgID)) msg := &types.Message{ ObjHeader: types.ObjHeader{Id: msgID.String(), CreatedAt: time.Date(2026, 9, 1, 10, 0, 0, 0, time.UTC)}, TenantID: 7, @@ -114,7 +121,7 @@ func TestEnqueueKafkaFileAttachmentsBindRunsSynchronously(t *testing.T) { 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) + files.EXPECT().LinkAttachments(types.TenantID(7), "", databaseMsgID, attachments).Return(nil) if err := enqueueKafkaFileAttachmentsBind(context.Background(), 7, msg, attachments); err != nil { t.Fatalf("enqueueKafkaFileAttachmentsBind returned error: %v", err) diff --git a/server/kafka_file_lifecycle.go b/server/kafka_file_lifecycle.go index 09dad45..d34f4e9 100644 --- a/server/kafka_file_lifecycle.go +++ b/server/kafka_file_lifecycle.go @@ -128,7 +128,14 @@ func executeKafkaFileAttachmentsBind(tenantID types.TenantID, payload kafkaFileL return types.ErrMalformed } if types.GetTopicCat(payload.Topic) != types.TopicCatGrp { - return store.Files.LinkAttachments(tenantID, "", msgID, payload.AttachmentURLs) + // Kafka persistence writes the decoded UID as the SQL message primary key, + // while the lifecycle payload carries the externally encoded UID. The + // file-link adapter expects the raw SQL ID, so decode it before binding. + databaseMsgID := types.Uid(store.DecodeUid(msgID)) + if databaseMsgID.IsZero() { + return types.ErrMalformed + } + return store.Files.LinkAttachments(tenantID, "", databaseMsgID, payload.AttachmentURLs) } uploader := types.ParseUserId(payload.UserID) if uploader.IsZero() {