139 lines
3.6 KiB
Go
139 lines
3.6 KiB
Go
package queueing
|
|
|
|
import (
|
|
"context"
|
|
"log/slog"
|
|
|
|
"gorm.io/gorm"
|
|
|
|
"wucher/internal/config"
|
|
"wucher/internal/queue"
|
|
mysqlrepo "wucher/internal/repository/mysql"
|
|
sqsrepo "wucher/internal/repository/sqs"
|
|
"wucher/internal/resilience"
|
|
)
|
|
|
|
func NewEmailProducer(
|
|
ctx context.Context,
|
|
queueCfg config.QueueConfig,
|
|
sqsCfg config.SQSConfig,
|
|
executor *resilience.Executor,
|
|
logger *slog.Logger,
|
|
) (queue.EmailJobProducer, error) {
|
|
producer, err := NewQueueProducer(ctx, sqsCfg, executor, "api")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return queue.NewEmailEnqueuer(NewOutboxSerializer(queueCfg, sqsCfg), producer, logger), nil
|
|
}
|
|
|
|
func NewQueueProducer(ctx context.Context, sqsCfg config.SQSConfig, executor *resilience.Executor, component string) (queue.QueueProducer, error) {
|
|
client, err := sqsrepo.NewClient(ctx, sqsCfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return sqsrepo.NewQueue(
|
|
sqsrepo.NewResilientAPI(client, executor),
|
|
sqsCfg,
|
|
sqsrepo.WithMetricsComponent(component),
|
|
)
|
|
}
|
|
|
|
func NewOutboxSerializer(queueCfg config.QueueConfig, sqsCfg config.SQSConfig) queue.MessageSerializer {
|
|
return queue.NewJSONSerializer(
|
|
queueCfg.MessageSchemaVersion,
|
|
queueCfg.EnableLegacyPayloadFallback,
|
|
sqsCfg.QueueType,
|
|
sqsCfg.MessageGroupID,
|
|
)
|
|
}
|
|
|
|
func NewEmailConsumer(
|
|
ctx context.Context,
|
|
queueCfg config.QueueConfig,
|
|
sqsCfg config.SQSConfig,
|
|
executor *resilience.Executor,
|
|
) (queue.QueueConsumer, queue.MessageSerializer, error) {
|
|
client, err := sqsrepo.NewClient(ctx, sqsCfg)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
consumer, err := sqsrepo.NewEmailQueue(
|
|
sqsrepo.NewResilientAPI(client, executor),
|
|
sqsCfg,
|
|
sqsrepo.WithMetricsComponent("email_worker"),
|
|
)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
serializer := queue.NewJSONSerializer(
|
|
queueCfg.MessageSchemaVersion,
|
|
queueCfg.EnableLegacyPayloadFallback,
|
|
sqsCfg.QueueType,
|
|
sqsCfg.MessageGroupID,
|
|
)
|
|
return consumer, serializer, nil
|
|
}
|
|
|
|
func NewFileProcessingProducer(
|
|
ctx context.Context,
|
|
queueCfg config.FileQueueConfig,
|
|
sqsCfg config.SQSConfig,
|
|
executor *resilience.Executor,
|
|
logger *slog.Logger,
|
|
) (queue.FileProcessingJobProducer, error) {
|
|
producer, err := NewQueueProducer(ctx, sqsCfg, executor, "api")
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return queue.NewFileProcessingEnqueuer(NewFileProcessingSerializer(queueCfg, sqsCfg), producer, logger), nil
|
|
}
|
|
|
|
func NewFileProcessingSerializer(queueCfg config.FileQueueConfig, sqsCfg config.SQSConfig) queue.FileProcessingMessageSerializer {
|
|
return queue.NewFileProcessingJSONSerializer(
|
|
queueCfg.MessageSchemaVersion,
|
|
sqsCfg.QueueType,
|
|
sqsCfg.MessageGroupID,
|
|
)
|
|
}
|
|
|
|
func NewFileProcessingConsumer(
|
|
ctx context.Context,
|
|
queueCfg config.FileQueueConfig,
|
|
sqsCfg config.SQSConfig,
|
|
executor *resilience.Executor,
|
|
) (queue.QueueConsumer, queue.FileProcessingMessageSerializer, error) {
|
|
client, err := sqsrepo.NewClient(ctx, sqsCfg)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
consumer, err := sqsrepo.NewQueue(
|
|
sqsrepo.NewResilientAPI(client, executor),
|
|
sqsCfg,
|
|
sqsrepo.WithMetricsComponent("file_worker"),
|
|
)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
serializer := queue.NewFileProcessingJSONSerializer(
|
|
queueCfg.MessageSchemaVersion,
|
|
sqsCfg.QueueType,
|
|
sqsCfg.MessageGroupID,
|
|
)
|
|
return consumer, serializer, nil
|
|
}
|
|
|
|
func NewIdempotencyStore(db *gorm.DB, queueCfg config.QueueConfig) queue.IdempotencyStore {
|
|
if db == nil {
|
|
return queue.NewMemoryIdempotencyStore()
|
|
}
|
|
return mysqlrepo.NewQueueIdempotencyStore(db, queueCfg.IdempotencyKeyPrefix)
|
|
}
|
|
|
|
func NewIdempotencyStoreWithPrefix(db *gorm.DB, keyPrefix string) queue.IdempotencyStore {
|
|
if db == nil {
|
|
return queue.NewMemoryIdempotencyStore()
|
|
}
|
|
return mysqlrepo.NewQueueIdempotencyStore(db, keyPrefix)
|
|
}
|