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

Refine Kafka persistence topic boundaries

Browse Source
This commit is contained in:
richjohnrambo
2026-09-01 23:23:08 +08:00
parent f8a286cec2
commit 318ce04f7d
13 changed files with 294 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
server/kafka_file_lifecycle.go
+4 -68
View File
@@ -2,14 +2,11 @@ package main
import (
"context"
"encoding/json"
"errors"
"fmt"
"strings"
"time"
"github.com/segmentio/kafka-go"
"github.com/tinode/chat/server/logs"
"github.com/tinode/chat/server/store"
"github.com/tinode/chat/server/store/types"
)
@@ -39,12 +36,8 @@ type kafkaFileLifecyclePayload struct {
OccurredAt time.Time `json:"occurred_at"`
}
func startKafkaFileLifecycleConsumer() {
startKafkaConsumer("file_lifecycle", handleKafkaFileLifecycle)
}
func enqueueKafkaFileUploaded(ctx context.Context, tenantID types.TenantID, fd *types.FileDef) error {
if !kafkaEnabled() || fd == nil || fd.TenantID != tenantID || fd.Status != types.UploadCompleted {
if fd == nil || fd.TenantID != tenantID || fd.Status != types.UploadCompleted {
return nil
}
payload := kafkaFileLifecyclePayload{
@@ -63,7 +56,7 @@ func enqueueKafkaFileUploaded(ctx context.Context, tenantID types.TenantID, fd *
}
func enqueueKafkaFileAttachmentsBind(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string) error {
if !kafkaEnabled() || msg == nil || msg.TenantID != tenantID || len(attachmentURLs) == 0 {
if msg == nil || msg.TenantID != tenantID || len(attachmentURLs) == 0 {
return nil
}
payload := kafkaFileLifecyclePayload{
@@ -80,7 +73,7 @@ func enqueueKafkaFileAttachmentsBind(ctx context.Context, tenantID types.TenantI
}
func enqueueKafkaGroupFileLifecycle(ctx context.Context, tenantID types.TenantID, eventType string, file *types.GroupFile, actor types.Uid) error {
if !kafkaEnabled() || file == nil || file.Topic == "" || file.FileID == "" {
if file == nil || file.Topic == "" || file.FileID == "" {
return nil
}
occurredAt := file.UpdatedAt
@@ -108,64 +101,7 @@ func enqueueKafkaFileLifecycle(ctx context.Context, tenantID types.TenantID, eve
if payload.OccurredAt.IsZero() {
payload.OccurredAt = time.Now().UTC().Round(time.Millisecond)
}
encoded, err := json.Marshal(payload)
if err != nil {
return err
}
eventID := stableIMEventID("file-lifecycle", tenantID, eventType, bizID, payload.SeqID, strings.Join(payload.AttachmentURLs, ","))
err = publishKafkaEnvelope(ctx, "file_lifecycle", bizID, kafkaEnvelope{
EventID: eventID,
TenantID: tenantID,
SourceService: "im-server",
TargetService: "im-server",
EventType: eventType,
BizType: "file",
BizID: bizID,
OccurredAt: payload.OccurredAt,
Payload: encoded,
})
if err != nil {
return err
}
actor := types.ParseUserId(payload.UserID)
if auditErr := enqueueKafkaAuditEvent(context.Background(), tenantID, "user", actor, eventType, "file", bizID, "succeeded", "", map[string]any{
"file_id": payload.FileID,
"file_ids": payload.FileIDs,
"topic": payload.Topic,
"group_file_id": payload.GroupFileID,
"seq_id": payload.SeqID,
}); auditErr != nil {
logs.Warn.Printf("kafka file lifecycle: failed to enqueue audit event type=%s biz=%s: %v", eventType, bizID, auditErr)
}
return nil
}
func handleKafkaFileLifecycle(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error {
tenantID, err := kafkaTenantIDFromEnvelope(envelope)
if err != nil {
return err
}
consume, claimed, err := beginKafkaConsumeForTopic(tenantID, envelope, "file_lifecycle")
if err != nil || !claimed {
return err
}
var payload kafkaFileLifecyclePayload
if err = json.Unmarshal(envelope.Payload, &payload); err != nil {
return markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error())
}
if err = executeKafkaFileLifecycle(tenantID, payload); err != nil {
failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup,
kafkaCommandErrorReason(err), err.Error(), time.Now().UTC().Round(time.Millisecond))
if markErr != nil {
return markErr
}
if failed != nil && failed.RetryCount >= kafkaBus.config.MaxRetryCount {
return markKafkaDeadLetter(ctx, tenantID, envelope, failed, kafkaCommandErrorReason(err), err.Error())
}
sleepKafkaRetryBackoff(failed.RetryCount)
return err
}
return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond))
return executeKafkaFileLifecycle(tenantID, payload)
}
func executeKafkaFileLifecycle(tenantID types.TenantID, payload kafkaFileLifecyclePayload) error {