fm_be/internal/domain/auth/email_outbox.go

package auth

import (
	"context"
	"encoding/json"
	"errors"
	"strings"
	"time"

	"gorm.io/gorm"

	"wucher/internal/shared/pkg/uuidv7"
	"wucher/internal/queue"
)

const (
	EmailOutboxStatusPending    = "pending"
	EmailOutboxStatusProcessing = "processing"
	EmailOutboxStatusPublished  = "published"
	EmailOutboxStatusDead       = "dead"
)

type EmailOutboxMessage 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 (EmailOutboxMessage) TableName() string { return "email_outbox_messages" }

func (m *EmailOutboxMessage) BeforeCreate(_ *gorm.DB) error {
	if len(m.ID) == 0 {
		m.ID = uuidv7.MustBytes()
	}
	if strings.TrimSpace(m.Status) == "" {
		m.Status = EmailOutboxStatusPending
	}
	if m.AvailableAt.IsZero() {
		m.AvailableAt = time.Now().UTC()
	}
	return nil
}

func NewEmailOutboxMessage(now time.Time, message *queue.OutboundMessage) (*EmailOutboxMessage, 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 = queue.EmailKind
	}
	jobType := strings.TrimSpace(message.Attributes["job_type"])
	if jobType == "" {
		jobType = "generic"
	}

	return &EmailOutboxMessage{
		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:          EmailOutboxStatusPending,
		AvailableAt:     now.UTC(),
		CorrelationID:   strings.TrimSpace(message.Attributes["correlation_id"]),
		TraceID:         strings.TrimSpace(message.Attributes["trace_id"]),
	}, nil
}

func (m *EmailOutboxMessage) OutboundMessage() (*queue.OutboundMessage, error) {
	if m == nil {
		return nil, errors.New("email 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 &queue.OutboundMessage{
		ID:              m.MessageID,
		Body:            append([]byte(nil), m.Body...),
		Attributes:      attributes,
		MessageGroupID:  strings.TrimSpace(m.MessageGroupID),
		DeduplicationID: strings.TrimSpace(m.DeduplicationID),
	}, nil
}

type EmailOutboxWriter interface {
	CreateEmailOutboxMessage(ctx context.Context, message *EmailOutboxMessage) error
}

type EmailOutboxRepository interface {
	EmailOutboxWriter
	ClaimPendingEmailOutboxMessages(ctx context.Context, limit int, now time.Time, lockTTL time.Duration, workerID string) ([]EmailOutboxMessage, error)
	MarkEmailOutboxMessagePublished(ctx context.Context, id []byte, publishedAt time.Time) error
	MarkEmailOutboxMessageRetry(ctx context.Context, id []byte, availableAt time.Time, lastErr string) error
	MarkEmailOutboxMessageDead(ctx context.Context, id []byte, failedAt time.Time, lastErr string) error
	CountPendingEmailOutboxMessages(ctx context.Context, now time.Time) (int64, error)
}

type TransactionalRepository interface {
	WithTransaction(ctx context.Context, fn func(repo Repository, outbox EmailOutboxWriter) error) error
}