package filemanager import ( "context" "encoding/json" "errors" "strings" "time" "gorm.io/gorm" "wucher/internal/shared/pkg/uuidv7" ) const ( FileProcessingOutboxStatusPending = "pending" FileProcessingOutboxStatusProcessing = "processing" FileProcessingOutboxStatusPublished = "published" FileProcessingOutboxStatusDead = "dead" FileProcessingOutboxKind = "file_manager.processing" ) type FileProcessingOutboundMessage struct { ID string Body []byte Attributes map[string]string MessageGroupID string DeduplicationID string } type FileProcessingOutboxMessage struct { ID []byte `gorm:"type:binary(16);primaryKey;column:id"` MessageID string `gorm:"type:varchar(64);uniqueIndex;not null;column:message_id"` Kind string `gorm:"type:varchar(64);index;not null;column:kind"` JobType string `gorm:"type:varchar(64);index;not null;column:job_type"` Body []byte `gorm:"type:longblob;not null;column:body"` Attributes []byte `gorm:"type:longblob;column:attributes"` MessageGroupID string `gorm:"type:varchar(128);column:message_group_id"` DeduplicationID string `gorm:"type:varchar(128);column:deduplication_id"` Status string `gorm:"type:varchar(32);index;not null;column:status"` Attempts int `gorm:"not null;default:0;column:attempts"` AvailableAt time.Time `gorm:"index;not null;column:available_at"` LockedAt *time.Time `gorm:"index;column:locked_at"` LockedBy string `gorm:"type:varchar(64);column:locked_by"` PublishedAt *time.Time `gorm:"column:published_at"` LastError string `gorm:"type:text;column:last_error"` CorrelationID string `gorm:"type:varchar(64);index;column:correlation_id"` TraceID string `gorm:"type:varchar(64);index;column:trace_id"` CreatedAt time.Time UpdatedAt time.Time } func (FileProcessingOutboxMessage) TableName() string { return "file_processing_outbox_messages" } func (m *FileProcessingOutboxMessage) BeforeCreate(_ *gorm.DB) error { if len(m.ID) == 0 { m.ID = uuidv7.MustBytes() } if strings.TrimSpace(m.Status) == "" { m.Status = FileProcessingOutboxStatusPending } if m.AvailableAt.IsZero() { m.AvailableAt = time.Now().UTC() } return nil } func NewFileProcessingOutboxMessage(now time.Time, message *FileProcessingOutboundMessage) (*FileProcessingOutboxMessage, error) { if message == nil { return nil, errors.New("outbound message is required") } if strings.TrimSpace(message.ID) == "" { return nil, errors.New("outbound message id is required") } if len(message.Body) == 0 { return nil, errors.New("outbound message body is required") } attributes, err := json.Marshal(message.Attributes) if err != nil { return nil, err } if now.IsZero() { now = time.Now().UTC() } kind := strings.TrimSpace(message.Attributes["message_kind"]) if kind == "" { kind = FileProcessingOutboxKind } jobType := strings.TrimSpace(message.Attributes["job_type"]) if jobType == "" { jobType = "file_processing" } return &FileProcessingOutboxMessage{ MessageID: strings.TrimSpace(message.ID), Kind: kind, JobType: jobType, Body: append([]byte(nil), message.Body...), Attributes: append([]byte(nil), attributes...), MessageGroupID: strings.TrimSpace(message.MessageGroupID), DeduplicationID: strings.TrimSpace(message.DeduplicationID), Status: FileProcessingOutboxStatusPending, AvailableAt: now.UTC(), CorrelationID: strings.TrimSpace(message.Attributes["correlation_id"]), TraceID: strings.TrimSpace(message.Attributes["trace_id"]), }, nil } func (m *FileProcessingOutboxMessage) OutboundMessage() (*FileProcessingOutboundMessage, error) { if m == nil { return nil, errors.New("file processing outbox message is required") } attributes := map[string]string{} if len(m.Attributes) > 0 { if err := json.Unmarshal(m.Attributes, &attributes); err != nil { return nil, err } } if strings.TrimSpace(m.MessageID) == "" { return nil, errors.New("message id is required") } if len(m.Body) == 0 { return nil, errors.New("message body is required") } return &FileProcessingOutboundMessage{ ID: m.MessageID, Body: append([]byte(nil), m.Body...), Attributes: attributes, MessageGroupID: strings.TrimSpace(m.MessageGroupID), DeduplicationID: strings.TrimSpace(m.DeduplicationID), }, nil } type FileProcessingOutboxWriter interface { CreateFileProcessingOutboxMessage(ctx context.Context, message *FileProcessingOutboxMessage) error } type FileProcessingOutboxRepository interface { FileProcessingOutboxWriter ClaimPendingFileProcessingOutboxMessages(ctx context.Context, limit int, now time.Time, lockTTL time.Duration, workerID string) ([]FileProcessingOutboxMessage, error) MarkFileProcessingOutboxMessagePublished(ctx context.Context, id []byte, publishedAt time.Time) error MarkFileProcessingOutboxMessageRetry(ctx context.Context, id []byte, availableAt time.Time, lastErr string) error MarkFileProcessingOutboxMessageDead(ctx context.Context, id []byte, failedAt time.Time, lastErr string) error CountPendingFileProcessingOutboxMessages(ctx context.Context, now time.Time) (int64, error) } type FileTransactionalRepository interface { WithTransaction(ctx context.Context, fn func(repo FileRepository, outbox FileProcessingOutboxWriter) error) error }