fm_be/internal/queue/serializer.go

package queue

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"strings"
	"time"
)

var ErrInvalidEnvelope = errors.New("invalid queue envelope")

type JSONSerializer struct {
	version              string
	enableLegacyFallback bool
	queueType            string
	messageGroupID       string
}

type LegacyJSONSerializer struct{}

func NewJSONSerializer(version string, enableLegacyFallback bool, queueType, messageGroupID string) *JSONSerializer {
	version = strings.TrimSpace(version)
	if version == "" {
		version = "v1"
	}
	queueType = strings.ToLower(strings.TrimSpace(queueType))
	if queueType == "" {
		queueType = QueueTypeStandard
	}
	return &JSONSerializer{
		version:              version,
		enableLegacyFallback: enableLegacyFallback,
		queueType:            queueType,
		messageGroupID:       strings.TrimSpace(messageGroupID),
	}
}

func (s *JSONSerializer) Serialize(ctx context.Context, job EmailJob) (*OutboundMessage, error) {
	envelope, err := NewEnvelope(ctx, s.version, job)
	if err != nil {
		return nil, err
	}
	body, err := json.Marshal(envelope)
	if err != nil {
		return nil, err
	}
	msg := &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":       envelope.Payload.NormalizedType(),
		},
	}
	if s.queueType == QueueTypeFIFO {
		msg.MessageGroupID = s.messageGroupID
		if msg.MessageGroupID == "" {
			msg.MessageGroupID = "default"
		}
		msg.DeduplicationID = envelope.DeduplicationKey()
	}
	return msg, nil
}

func (s *JSONSerializer) Deserialize(body []byte, attributes map[string]string) (*Envelope, error) {
	var envelope Envelope
	if err := json.Unmarshal(body, &envelope); err == nil {
		if err := validateEnvelope(&envelope); err == nil {
			return &envelope, nil
		}
	}

	if !s.enableLegacyFallback {
		return nil, ErrInvalidEnvelope
	}

	var job EmailJob
	if err := json.Unmarshal(body, &job); err != nil {
		return nil, errors.Join(ErrInvalidEnvelope, err)
	}
	job = job.Canonical()
	if err := job.Validate(); err != nil {
		return nil, errors.Join(ErrInvalidEnvelope, err)
	}

	id := messageHash(body)
	correlationID := strings.TrimSpace(attributes["correlation_id"])
	if correlationID == "" {
		correlationID = id
	}
	traceID := strings.TrimSpace(attributes["trace_id"])
	if traceID == "" {
		traceID = correlationID
	}
	return &Envelope{
		Version:        "legacy",
		MessageID:      id,
		Kind:           EmailKind,
		OccurredAt:     time.Now().UTC(),
		CorrelationID:  correlationID,
		TraceID:        traceID,
		IdempotencyKey: id,
		Payload:        job,
	}, nil
}

func (LegacyJSONSerializer) Serialize(ctx context.Context, job EmailJob) (*OutboundMessage, error) {
	job = job.Canonical()
	if err := job.Validate(); err != nil {
		return nil, err
	}
	body, err := json.Marshal(job)
	if err != nil {
		return nil, err
	}
	id := messageHash(body)
	correlationID := CorrelationIDFromContext(ctx)
	if correlationID == "" {
		correlationID = id
	}
	traceID := TraceIDFromContext(ctx)
	if traceID == "" {
		traceID = correlationID
	}
	return &OutboundMessage{
		ID:   id,
		Body: body,
		Attributes: map[string]string{
			"message_kind":   EmailKind,
			"message_id":     id,
			"correlation_id": correlationID,
			"trace_id":       traceID,
			"schema_version": "legacy",
			"job_type":       job.NormalizedType(),
		},
	}, nil
}

func (LegacyJSONSerializer) Deserialize(body []byte, attributes map[string]string) (*Envelope, error) {
	return NewJSONSerializer("v1", true, QueueTypeStandard, "").Deserialize(body, attributes)
}

func validateEnvelope(envelope *Envelope) error {
	if envelope == nil {
		return ErrInvalidEnvelope
	}
	if strings.TrimSpace(envelope.Version) == "" {
		return errors.Join(ErrInvalidEnvelope, errors.New("version is required"))
	}
	if strings.TrimSpace(envelope.MessageID) == "" {
		return errors.Join(ErrInvalidEnvelope, errors.New("message id is required"))
	}
	if strings.TrimSpace(envelope.Kind) == "" {
		return errors.Join(ErrInvalidEnvelope, errors.New("kind is required"))
	}
	if envelope.OccurredAt.IsZero() {
		return errors.Join(ErrInvalidEnvelope, errors.New("occurred at is required"))
	}
	if err := envelope.Payload.Validate(); err != nil {
		return errors.Join(ErrInvalidEnvelope, err)
	}
	envelope.Payload = envelope.Payload.Canonical()
	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
	}
	return nil
}

func messageHash(body []byte) string {
	sum := sha256.Sum256(body)
	return hex.EncodeToString(sum[:16])
}