146 lines
3.6 KiB
Go
146 lines
3.6 KiB
Go
package queue
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log/slog"
|
|
)
|
|
|
|
type ProducerMetrics struct {
|
|
published int64
|
|
publishFailures int64
|
|
secondaryFailures int64
|
|
}
|
|
|
|
type EmailEnqueuer struct {
|
|
serializer MessageSerializer
|
|
producer QueueProducer
|
|
logger *slog.Logger
|
|
}
|
|
|
|
func NewEmailEnqueuer(serializer MessageSerializer, producer QueueProducer, logger *slog.Logger) *EmailEnqueuer {
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
return &EmailEnqueuer{
|
|
serializer: serializer,
|
|
producer: producer,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
func (e *EmailEnqueuer) Enqueue(ctx context.Context, job EmailJob) error {
|
|
message, err := e.serializer.Serialize(ctx, job)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if err := e.producer.Publish(ctx, *message); err != nil {
|
|
e.logger.Error("queue publish failed",
|
|
slog.String("backend", e.producer.Backend()),
|
|
slog.String("message_id", message.ID),
|
|
slog.String("job_type", job.NormalizedType()),
|
|
slog.String("recipient_hash", job.RecipientHash()),
|
|
slog.String("correlation_id", message.Attributes["correlation_id"]),
|
|
slog.Any("error", err),
|
|
)
|
|
return err
|
|
}
|
|
e.logger.Info("queue published",
|
|
slog.String("backend", e.producer.Backend()),
|
|
slog.String("message_id", message.ID),
|
|
slog.String("job_type", job.NormalizedType()),
|
|
slog.String("recipient_hash", job.RecipientHash()),
|
|
slog.String("correlation_id", message.Attributes["correlation_id"]),
|
|
)
|
|
return nil
|
|
}
|
|
|
|
type DualProducer struct {
|
|
primary QueueProducer
|
|
secondary QueueProducer
|
|
secondaryRequired bool
|
|
logger *slog.Logger
|
|
}
|
|
|
|
func NewDualProducer(primary, secondary QueueProducer, secondaryRequired bool, logger *slog.Logger) *DualProducer {
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
return &DualProducer{
|
|
primary: primary,
|
|
secondary: secondary,
|
|
secondaryRequired: secondaryRequired,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
func (p *DualProducer) Backend() string {
|
|
return "dual"
|
|
}
|
|
|
|
func (p *DualProducer) Publish(ctx context.Context, message OutboundMessage) error {
|
|
if p.primary == nil {
|
|
return errors.New("primary producer is required")
|
|
}
|
|
if err := p.primary.Publish(ctx, message); err != nil {
|
|
return err
|
|
}
|
|
if p.secondary == nil {
|
|
return nil
|
|
}
|
|
if err := p.secondary.Publish(ctx, message); err != nil {
|
|
p.logger.Error("queue secondary publish failed",
|
|
slog.String("primary_backend", p.primary.Backend()),
|
|
slog.String("secondary_backend", p.secondary.Backend()),
|
|
slog.String("message_id", message.ID),
|
|
slog.Any("error", err),
|
|
)
|
|
if p.secondaryRequired {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type DualEmailJobProducer struct {
|
|
primary EmailJobProducer
|
|
secondary EmailJobProducer
|
|
secondaryRequired bool
|
|
logger *slog.Logger
|
|
}
|
|
|
|
func NewDualEmailJobProducer(primary, secondary EmailJobProducer, secondaryRequired bool, logger *slog.Logger) *DualEmailJobProducer {
|
|
if logger == nil {
|
|
logger = slog.Default()
|
|
}
|
|
return &DualEmailJobProducer{
|
|
primary: primary,
|
|
secondary: secondary,
|
|
secondaryRequired: secondaryRequired,
|
|
logger: logger,
|
|
}
|
|
}
|
|
|
|
func (p *DualEmailJobProducer) Enqueue(ctx context.Context, job EmailJob) error {
|
|
if p.primary == nil {
|
|
return errors.New("primary email producer is required")
|
|
}
|
|
if err := p.primary.Enqueue(ctx, job); err != nil {
|
|
return err
|
|
}
|
|
if p.secondary == nil {
|
|
return nil
|
|
}
|
|
if err := p.secondary.Enqueue(ctx, job); err != nil {
|
|
p.logger.Error("queue secondary enqueue failed",
|
|
slog.String("job_type", job.NormalizedType()),
|
|
slog.String("recipient_hash", job.RecipientHash()),
|
|
slog.Any("error", err),
|
|
)
|
|
if p.secondaryRequired {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|