pro/im_server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix kafka attachment message id binding

Browse Source
This commit is contained in:
richjohnrambo
2026-10-14 12:13:45 +08:00
parent 6930b23cce
commit 87ec8f9833
2 changed files with 17 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/kafka_audit_test.go
+9 -2
View File
@@ -93,6 +93,12 @@ func TestEnqueueKafkaAuditEventPublishesCrossSystemEvent(t *testing.T) {
func TestEnqueueKafkaFileAttachmentsBindRunsSynchronously(t *testing.T) { func TestEnqueueKafkaFileAttachmentsBindRunsSynchronously(t *testing.T) {
ctrl := gomock.NewController(t) ctrl := gomock.NewController(t)
defer ctrl.Finish() 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 oldKafkaBus := kafkaBus
oldFiles := store.Files oldFiles := store.Files
@@ -104,7 +110,8 @@ func TestEnqueueKafkaFileAttachmentsBindRunsSynchronously(t *testing.T) {
kafkaBus = &kafkaRuntime{} kafkaBus = &kafkaRuntime{}
files := mock_store.NewMockFilePersistenceInterface(ctrl) files := mock_store.NewMockFilePersistenceInterface(ctrl)
store.Files = files store.Files = files
msgID := types.Uid(101) databaseMsgID := types.Uid(101)
msgID := store.EncodeUid(int64(databaseMsgID))
msg := &types.Message{ msg := &types.Message{
ObjHeader: types.ObjHeader{Id: msgID.String(), CreatedAt: time.Date(2026, 9, 1, 10, 0, 0, 0, time.UTC)}, ObjHeader: types.ObjHeader{Id: msgID.String(), CreatedAt: time.Date(2026, 9, 1, 10, 0, 0, 0, time.UTC)},
TenantID: 7, TenantID: 7,
@@ -114,7 +121,7 @@ func TestEnqueueKafkaFileAttachmentsBindRunsSynchronously(t *testing.T) {
Head: types.KVMap{"x-im-type": "image"}, Head: types.KVMap{"x-im-type": "image"},
} }
attachments := []string{"https://api.118im.com/v1/im/files/file_001"} 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 { if err := enqueueKafkaFileAttachmentsBind(context.Background(), 7, msg, attachments); err != nil {
t.Fatalf("enqueueKafkaFileAttachmentsBind returned error: %v", err) t.Fatalf("enqueueKafkaFileAttachmentsBind returned error: %v", err)
server/kafka_file_lifecycle.go
+8 -1
View File
@@ -128,7 +128,14 @@ func executeKafkaFileAttachmentsBind(tenantID types.TenantID, payload kafkaFileL
return types.ErrMalformed return types.ErrMalformed
} }
if types.GetTopicCat(payload.Topic) != types.TopicCatGrp { 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) uploader := types.ParseUserId(payload.UserID)
if uploader.IsZero() { if uploader.IsZero() {