113 lines
3.4 KiB
Go
113 lines
3.4 KiB
Go
package queue
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
type FileProcessingJSONSerializer struct {
|
|
version string
|
|
queueType string
|
|
messageGroupID string
|
|
}
|
|
|
|
func NewFileProcessingJSONSerializer(version, queueType, messageGroupID string) *FileProcessingJSONSerializer {
|
|
version = strings.TrimSpace(version)
|
|
if version == "" {
|
|
version = "v1"
|
|
}
|
|
queueType = strings.ToLower(strings.TrimSpace(queueType))
|
|
if queueType == "" {
|
|
queueType = QueueTypeStandard
|
|
}
|
|
return &FileProcessingJSONSerializer{
|
|
version: version,
|
|
queueType: queueType,
|
|
messageGroupID: strings.TrimSpace(messageGroupID),
|
|
}
|
|
}
|
|
|
|
func (s *FileProcessingJSONSerializer) Serialize(ctx context.Context, job FileProcessingJob) (*OutboundMessage, error) {
|
|
envelope, err := NewFileProcessingEnvelope(ctx, s.version, job)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
body, err := json.Marshal(envelope)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
message := &OutboundMessage{
|
|
ID: envelope.MessageID,
|
|
Body: body,
|
|
Attributes: map[string]string{
|
|
"message_kind": envelope.Kind,
|
|
"message_id": envelope.MessageID,
|
|
"correlation_id": envelope.CorrelationID,
|
|
"trace_id": envelope.TraceID,
|
|
"schema_version": envelope.Version,
|
|
"job_type": "file_processing",
|
|
"file_uuid": envelope.Payload.FileUUID,
|
|
},
|
|
}
|
|
if s.queueType == QueueTypeFIFO {
|
|
message.MessageGroupID = s.messageGroupID
|
|
if message.MessageGroupID == "" {
|
|
message.MessageGroupID = "file-processing"
|
|
}
|
|
message.DeduplicationID = envelope.DeduplicationKey()
|
|
}
|
|
return message, nil
|
|
}
|
|
|
|
func (s *FileProcessingJSONSerializer) Deserialize(body []byte, _ map[string]string) (*FileProcessingEnvelope, error) {
|
|
var envelope FileProcessingEnvelope
|
|
if err := json.Unmarshal(body, &envelope); err != nil {
|
|
return nil, errors.Join(ErrInvalidFileProcessingEnvelope, err)
|
|
}
|
|
if err := validateFileProcessingEnvelope(&envelope); err != nil {
|
|
return nil, err
|
|
}
|
|
return &envelope, nil
|
|
}
|
|
|
|
func validateFileProcessingEnvelope(envelope *FileProcessingEnvelope) error {
|
|
if envelope == nil {
|
|
return ErrInvalidFileProcessingEnvelope
|
|
}
|
|
if strings.TrimSpace(envelope.Version) == "" {
|
|
return errors.Join(ErrInvalidFileProcessingEnvelope, errors.New("version is required"))
|
|
}
|
|
if strings.TrimSpace(envelope.MessageID) == "" {
|
|
return errors.Join(ErrInvalidFileProcessingEnvelope, errors.New("message id is required"))
|
|
}
|
|
if strings.TrimSpace(envelope.Kind) == "" {
|
|
return errors.Join(ErrInvalidFileProcessingEnvelope, errors.New("kind is required"))
|
|
}
|
|
if !strings.EqualFold(strings.TrimSpace(envelope.Kind), FileProcessingKind) {
|
|
return errors.Join(ErrInvalidFileProcessingEnvelope, errors.New("unsupported message kind"))
|
|
}
|
|
if envelope.OccurredAt.IsZero() {
|
|
return errors.Join(ErrInvalidFileProcessingEnvelope, errors.New("occurred at is required"))
|
|
}
|
|
envelope.Payload = envelope.Payload.Canonical()
|
|
if err := envelope.Payload.Validate(); err != nil {
|
|
return errors.Join(ErrInvalidFileProcessingEnvelope, err)
|
|
}
|
|
if strings.TrimSpace(envelope.CorrelationID) == "" {
|
|
envelope.CorrelationID = envelope.MessageID
|
|
}
|
|
if strings.TrimSpace(envelope.TraceID) == "" {
|
|
envelope.TraceID = envelope.CorrelationID
|
|
}
|
|
if strings.TrimSpace(envelope.IdempotencyKey) == "" {
|
|
envelope.IdempotencyKey = envelope.MessageID
|
|
}
|
|
if envelope.Payload.CreatedAt.IsZero() {
|
|
envelope.Payload.CreatedAt = time.Now().UTC()
|
|
}
|
|
return nil
|
|
}
|