266 lines
9.0 KiB
Go
266 lines
9.0 KiB
Go
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"
|
|
)
|
|
|
|
const (
|
|
kafkaFileLifecycleUploadCompleted = "file.upload.completed"
|
|
kafkaFileLifecycleAttachmentsBind = "file.attachments.bind"
|
|
kafkaFileLifecycleGroupFileCreate = "group.file.created"
|
|
kafkaFileLifecycleGroupFileDelete = "group.file.deleted"
|
|
)
|
|
|
|
type kafkaFileLifecyclePayload struct {
|
|
Action string `json:"action"`
|
|
FileID string `json:"file_id,omitempty"`
|
|
FileIDs []string `json:"file_ids,omitempty"`
|
|
AttachmentURLs []string `json:"attachment_urls,omitempty"`
|
|
Topic string `json:"topic,omitempty"`
|
|
MessageID string `json:"message_id,omitempty"`
|
|
SeqID int64 `json:"seq_id,omitempty"`
|
|
UserID string `json:"user_id,omitempty"`
|
|
GroupFileID string `json:"group_file_id,omitempty"`
|
|
FileType string `json:"file_type,omitempty"`
|
|
FileName string `json:"file_name,omitempty"`
|
|
MIMEType string `json:"mime_type,omitempty"`
|
|
FileSize int64 `json:"file_size,omitempty"`
|
|
Storage map[string]any `json:"storage,omitempty"`
|
|
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 {
|
|
return nil
|
|
}
|
|
payload := kafkaFileLifecyclePayload{
|
|
Action: kafkaFileLifecycleUploadCompleted,
|
|
FileID: fd.Uid().String(),
|
|
UserID: types.ParseUid(fd.User).UserId(),
|
|
MIMEType: fd.MimeType,
|
|
FileSize: fd.Size,
|
|
OccurredAt: fd.UpdatedAt,
|
|
Storage: map[string]any{
|
|
"location": fd.Location,
|
|
"etag": fd.ETag,
|
|
},
|
|
}
|
|
return enqueueKafkaFileLifecycle(ctx, tenantID, kafkaFileLifecycleUploadCompleted, fd.Uid().String(), payload)
|
|
}
|
|
|
|
func enqueueKafkaFileAttachmentsBind(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string) error {
|
|
if !kafkaEnabled() || msg == nil || msg.TenantID != tenantID || len(attachmentURLs) == 0 {
|
|
return nil
|
|
}
|
|
payload := kafkaFileLifecyclePayload{
|
|
Action: kafkaFileLifecycleAttachmentsBind,
|
|
AttachmentURLs: append([]string(nil), attachmentURLs...),
|
|
Topic: msg.Topic,
|
|
MessageID: msg.Uid().String(),
|
|
SeqID: msg.SeqId,
|
|
UserID: types.ParseUid(msg.From).UserId(),
|
|
FileType: lifecycleFileTypeFromMessageHead(msg.Head),
|
|
OccurredAt: msg.CreatedAt,
|
|
}
|
|
return enqueueKafkaFileLifecycle(ctx, tenantID, kafkaFileLifecycleAttachmentsBind, fmt.Sprintf("%s:%d", msg.Topic, msg.SeqId), payload)
|
|
}
|
|
|
|
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 == "" {
|
|
return nil
|
|
}
|
|
occurredAt := file.UpdatedAt
|
|
if occurredAt.IsZero() {
|
|
occurredAt = time.Now().UTC().Round(time.Millisecond)
|
|
}
|
|
payload := kafkaFileLifecyclePayload{
|
|
Action: eventType,
|
|
FileID: file.FileID,
|
|
FileIDs: []string{file.FileID},
|
|
Topic: file.Topic,
|
|
GroupFileID: file.GroupFileID,
|
|
SeqID: file.SourceSeq,
|
|
UserID: actor.UserId(),
|
|
FileType: file.FileType,
|
|
FileName: file.FileName,
|
|
MIMEType: file.MIMEType,
|
|
FileSize: file.FileSize,
|
|
OccurredAt: occurredAt,
|
|
}
|
|
return enqueueKafkaFileLifecycle(ctx, tenantID, eventType, firstNonEmpty(file.GroupFileID, file.FileID), payload)
|
|
}
|
|
|
|
func enqueueKafkaFileLifecycle(_ context.Context, tenantID types.TenantID, eventType, bizID string, payload kafkaFileLifecyclePayload) error {
|
|
tenantCode := ""
|
|
if store.Tenants != nil {
|
|
if tenant, err := store.Tenants.GetByID(tenantID); err == nil && tenant != nil {
|
|
tenantCode = tenant.Code
|
|
} else {
|
|
logs.Warn.Printf("kafka file lifecycle: failed to resolve tenant code for tenant %d: %v", tenantID, err)
|
|
}
|
|
}
|
|
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 = enqueueKafkaOutbox(tenantID, "file_lifecycle", bizID, kafkaEnvelope{
|
|
EventID: eventID,
|
|
TenantCode: tenantCode,
|
|
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 := kafkaTenantID(envelope.TenantCode)
|
|
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))
|
|
}
|
|
|
|
func executeKafkaFileLifecycle(tenantID types.TenantID, payload kafkaFileLifecyclePayload) error {
|
|
switch payload.Action {
|
|
case kafkaFileLifecycleAttachmentsBind:
|
|
return executeKafkaFileAttachmentsBind(tenantID, payload)
|
|
case kafkaFileLifecycleUploadCompleted, kafkaFileLifecycleGroupFileCreate, kafkaFileLifecycleGroupFileDelete:
|
|
return nil
|
|
default:
|
|
return types.ErrMalformed
|
|
}
|
|
}
|
|
|
|
func executeKafkaFileAttachmentsBind(tenantID types.TenantID, payload kafkaFileLifecyclePayload) error {
|
|
if payload.Topic == "" || payload.MessageID == "" || len(payload.AttachmentURLs) == 0 {
|
|
return types.ErrMalformed
|
|
}
|
|
msgID := types.ParseUid(payload.MessageID)
|
|
if msgID.IsZero() {
|
|
return types.ErrMalformed
|
|
}
|
|
if types.GetTopicCat(payload.Topic) != types.TopicCatGrp {
|
|
return store.Files.LinkAttachments(tenantID, "", msgID, payload.AttachmentURLs)
|
|
}
|
|
uploader := types.ParseUserId(payload.UserID)
|
|
if uploader.IsZero() {
|
|
return types.ErrMalformed
|
|
}
|
|
for _, url := range payload.AttachmentURLs {
|
|
fid := store.Store.GetMediaHandler().GetIdFromUrl(url)
|
|
if fid.IsZero() {
|
|
continue
|
|
}
|
|
_, err := store.GroupBusiness.CreateFile(tenantID, &types.GroupFile{
|
|
Topic: payload.Topic,
|
|
FileID: fid.String(),
|
|
SourceTopic: payload.Topic,
|
|
SourceSeq: payload.SeqID,
|
|
UploaderID: uploader,
|
|
FileType: lifecycleFileType(payload.FileType),
|
|
IdempotencyKey: fmt.Sprintf("%s:%d:%s", payload.Topic, payload.SeqID, fid.String()),
|
|
})
|
|
if err != nil && !errors.Is(err, types.ErrDuplicate) {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func lifecycleFileTypeFromMessageHead(head types.KVMap) string {
|
|
msgType := lifecycleHeaderString(head, "x-im-type")
|
|
if msgType == "" {
|
|
msgType = lifecycleHeaderString(head, "type")
|
|
}
|
|
if msgType == "" {
|
|
return "file"
|
|
}
|
|
return lifecycleFileType(msgType)
|
|
}
|
|
|
|
func lifecycleFileType(raw string) string {
|
|
switch lifecycleCanonicalMessageType(raw) {
|
|
case "image", "audio", "voice", "video":
|
|
return lifecycleCanonicalMessageType(raw)
|
|
default:
|
|
return "file"
|
|
}
|
|
}
|
|
|
|
func lifecycleHeaderString(head types.KVMap, key string) string {
|
|
if head == nil {
|
|
return ""
|
|
}
|
|
value, ok := head[key]
|
|
if !ok {
|
|
return ""
|
|
}
|
|
str, _ := value.(string)
|
|
return strings.TrimSpace(str)
|
|
}
|
|
|
|
func lifecycleCanonicalMessageType(raw string) string {
|
|
switch strings.ToLower(strings.TrimSpace(raw)) {
|
|
case "audio":
|
|
return "voice"
|
|
default:
|
|
return strings.ToLower(strings.TrimSpace(raw))
|
|
}
|
|
}
|