100 lines
2.8 KiB
Go
100 lines
2.8 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"log/slog"
|
|
"os/signal"
|
|
"syscall"
|
|
|
|
queueapp "wucher/internal/app/queueing"
|
|
appworker "wucher/internal/app/worker"
|
|
"wucher/internal/config"
|
|
"wucher/internal/shared/pkg/logger"
|
|
mysqlrepo "wucher/internal/repository/mysql"
|
|
"wucher/internal/resilience"
|
|
"wucher/internal/service"
|
|
)
|
|
|
|
func runMailWorker(appCfg config.AppConfig) error {
|
|
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
appLogger := logger.New(appCfg.Logging).With("component", "worker", "job", "email")
|
|
resCfg := appCfg.Resilience
|
|
emailCfg := appCfg.Email
|
|
|
|
queueCfg := appCfg.Queue
|
|
sqsCfg := appCfg.SQS
|
|
db, err := mysqlrepo.Connect(appCfg.MySQL)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
authRepo := mysqlrepo.NewAuthRepository(db)
|
|
|
|
sqsBreaker := resilience.NewExecutor("sqs", resCfg.SQS, appLogger, resilience.ClassifySQSError)
|
|
consumer, serializer, err := queueapp.NewEmailConsumer(ctx, queueCfg, sqsCfg, sqsBreaker)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
idempotency := queueapp.NewIdempotencyStore(db, queueCfg)
|
|
var dispatcher *appworker.OutboxDispatcher
|
|
if queueCfg.OutboxEnabled {
|
|
rawProducer, err := queueapp.NewQueueProducer(ctx, sqsCfg, sqsBreaker, "outbox_dispatcher")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
dispatcher = appworker.NewOutboxDispatcher(queueCfg, authRepo, rawProducer, appLogger)
|
|
}
|
|
|
|
emailSender, err := buildEmailSender(ctx, emailCfg, appCfg.SES, resCfg, appLogger)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
handler := service.NewEmailDeliveryHandler(emailSender, appLogger)
|
|
runner := appworker.NewRunner(queueCfg, consumer, serializer, handler, idempotency, dispatcher, appLogger)
|
|
|
|
appLogger.Info("worker started",
|
|
"backend", "sqs",
|
|
"email_provider", emailCfg.Provider,
|
|
"monitor_addr", queueCfg.WorkerMonitorAddr,
|
|
"outbox_enabled", queueCfg.OutboxEnabled,
|
|
)
|
|
if err := runner.Start(ctx); err != nil {
|
|
return err
|
|
}
|
|
appLogger.Info("worker stopped")
|
|
return nil
|
|
}
|
|
|
|
func buildEmailSender(
|
|
ctx context.Context,
|
|
emailCfg config.EmailDeliveryConfig,
|
|
sesCfg config.SESConfig,
|
|
resCfg config.ResilienceConfig,
|
|
appLogger *slog.Logger,
|
|
) (service.EmailSender, error) {
|
|
switch emailCfg.Provider {
|
|
case "", "ses":
|
|
sesBreaker := resilience.NewExecutor("ses", resCfg.SES, appLogger, resilience.ClassifySESError)
|
|
sender, err := service.NewSESSender(ctx, sesCfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
sender = sender.WithExecutor(sesBreaker)
|
|
sender = sender.WithMetricsComponent("email_worker")
|
|
if !sender.Configured() {
|
|
return nil, errors.New("ses is not configured")
|
|
}
|
|
return sender, nil
|
|
case "mock":
|
|
sender := service.NewMockEmailSender(emailCfg.MockDir)
|
|
if !sender.Configured() {
|
|
return nil, errors.New("mock email sender is not configured")
|
|
}
|
|
return sender, nil
|
|
default:
|
|
return nil, errors.New("unsupported email provider: " + emailCfg.Provider)
|
|
}
|
|
}
|