306 lines
7.7 KiB
Go
306 lines
7.7 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"net/mail"
|
|
"net/url"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
awsv2 "github.com/aws/aws-sdk-go-v2/aws"
|
|
awsmiddleware "github.com/aws/aws-sdk-go-v2/aws/middleware"
|
|
awsconfig "github.com/aws/aws-sdk-go-v2/config"
|
|
"github.com/aws/aws-sdk-go-v2/service/sesv2"
|
|
sestypes "github.com/aws/aws-sdk-go-v2/service/sesv2/types"
|
|
|
|
"wucher/internal/config"
|
|
"wucher/internal/queue"
|
|
"wucher/internal/resilience"
|
|
"wucher/internal/shared/pkg/metrics"
|
|
)
|
|
|
|
const utf8Charset = "UTF-8"
|
|
|
|
type sesAPI interface {
|
|
SendEmail(ctx context.Context, params *sesv2.SendEmailInput, optFns ...func(*sesv2.Options)) (*sesv2.SendEmailOutput, error)
|
|
}
|
|
|
|
type SESSender struct {
|
|
cfg config.SESConfig
|
|
api sesAPI
|
|
executor *resilience.Executor
|
|
component string
|
|
}
|
|
|
|
func NewSESSender(ctx context.Context, cfg config.SESConfig) (*SESSender, error) {
|
|
api, err := newSESAPI(ctx, cfg)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return &SESSender{
|
|
cfg: cfg,
|
|
api: api,
|
|
component: "worker",
|
|
}, nil
|
|
}
|
|
|
|
func NewSESSenderWithAPI(cfg config.SESConfig, api sesAPI) *SESSender {
|
|
return &SESSender{
|
|
cfg: cfg,
|
|
api: api,
|
|
component: "worker",
|
|
}
|
|
}
|
|
|
|
func (s *SESSender) WithExecutor(executor *resilience.Executor) *SESSender {
|
|
if s == nil {
|
|
return nil
|
|
}
|
|
s.executor = executor
|
|
return s
|
|
}
|
|
|
|
func (s *SESSender) WithMetricsComponent(component string) *SESSender {
|
|
if s == nil {
|
|
return nil
|
|
}
|
|
component = strings.TrimSpace(component)
|
|
if component == "" {
|
|
return s
|
|
}
|
|
s.component = component
|
|
return s
|
|
}
|
|
|
|
func (s *SESSender) Configured() bool {
|
|
return s != nil && strings.TrimSpace(s.cfg.Region) != "" && strings.TrimSpace(s.cfg.From) != ""
|
|
}
|
|
|
|
func (s *SESSender) Send(ctx context.Context, job queue.EmailJob) (EmailDeliveryResult, error) {
|
|
return resilience.Do(ctx, s.executor, func(ctx context.Context) (EmailDeliveryResult, error) {
|
|
if !s.Configured() {
|
|
return EmailDeliveryResult{}, errors.New("ses is not configured")
|
|
}
|
|
if s.api == nil {
|
|
return EmailDeliveryResult{}, errors.New("ses client is required")
|
|
}
|
|
job = job.Canonical()
|
|
if err := job.Validate(); err != nil {
|
|
return EmailDeliveryResult{}, err
|
|
}
|
|
input, err := s.buildSendInput(job)
|
|
if err != nil {
|
|
return EmailDeliveryResult{}, err
|
|
}
|
|
|
|
sendCtx := ctx
|
|
cancel := func() {}
|
|
if s.cfg.SendTimeout > 0 {
|
|
sendCtx, cancel = context.WithTimeout(ctx, s.cfg.SendTimeout)
|
|
}
|
|
defer cancel()
|
|
|
|
started := time.Now()
|
|
output, err := s.api.SendEmail(sendCtx, input)
|
|
metrics.ObserveAWSRequest(s.component, "ses", "SendEmail", err, started)
|
|
metrics.ObserveEmailSend(s.component, "ses", err, started)
|
|
if err != nil {
|
|
return EmailDeliveryResult{}, err
|
|
}
|
|
|
|
result := EmailDeliveryResult{
|
|
Provider: "ses",
|
|
ProviderMessageID: awsv2.ToString(output.MessageId),
|
|
}
|
|
if requestID, ok := awsmiddleware.GetRequestIDMetadata(output.ResultMetadata); ok {
|
|
result.RequestID = requestID
|
|
}
|
|
return result, nil
|
|
})
|
|
}
|
|
|
|
func (s *SESSender) buildSendInput(job queue.EmailJob) (*sesv2.SendEmailInput, error) {
|
|
from, err := mail.ParseAddress(strings.TrimSpace(s.cfg.From))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
to, err := normalizeEmailAddresses([]string{job.To})
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
cc, err := normalizeEmailAddresses(job.CC)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
bcc, err := normalizeEmailAddresses(job.BCC)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
replyTo, err := normalizeEmailAddresses(job.ReplyTo)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
input := &sesv2.SendEmailInput{
|
|
FromEmailAddress: awsv2.String(from.String()),
|
|
Destination: &sestypes.Destination{
|
|
ToAddresses: to,
|
|
CcAddresses: cc,
|
|
BccAddresses: bcc,
|
|
},
|
|
ReplyToAddresses: replyTo,
|
|
EmailTags: buildSESTags(job.Tags),
|
|
}
|
|
if cfgSet := strings.TrimSpace(s.cfg.ConfigurationSetName); cfgSet != "" {
|
|
input.ConfigurationSetName = awsv2.String(cfgSet)
|
|
}
|
|
|
|
content, err := buildSESContent(from.String(), to, cc, replyTo, job)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
input.Content = content
|
|
return input, nil
|
|
}
|
|
|
|
func buildSESContent(from string, to, cc, replyTo []string, job queue.EmailJob) (*sestypes.EmailContent, error) {
|
|
if len(job.Attachments) > 0 {
|
|
return buildSESRawContent(from, to, cc, replyTo, job)
|
|
}
|
|
if strings.TrimSpace(job.TemplateName) != "" && strings.TrimSpace(job.TextBodyValue()) == "" && strings.TrimSpace(job.HTMLBody) == "" {
|
|
templateData := "{}"
|
|
if len(job.TemplateData) > 0 {
|
|
encoded, err := json.Marshal(job.TemplateData)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
templateData = string(encoded)
|
|
}
|
|
return &sestypes.EmailContent{
|
|
Template: &sestypes.Template{
|
|
TemplateName: awsv2.String(job.TemplateName),
|
|
TemplateData: awsv2.String(templateData),
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
body := &sestypes.Body{}
|
|
if textBody := strings.TrimSpace(job.TextBodyValue()); textBody != "" {
|
|
body.Text = &sestypes.Content{
|
|
Data: awsv2.String(textBody),
|
|
Charset: awsv2.String(utf8Charset),
|
|
}
|
|
}
|
|
if htmlBody := strings.TrimSpace(job.HTMLBody); htmlBody != "" {
|
|
body.Html = &sestypes.Content{
|
|
Data: awsv2.String(htmlBody),
|
|
Charset: awsv2.String(utf8Charset),
|
|
}
|
|
}
|
|
return &sestypes.EmailContent{
|
|
Simple: &sestypes.Message{
|
|
Subject: &sestypes.Content{
|
|
Data: awsv2.String(sanitizeSubject(job.Subject)),
|
|
Charset: awsv2.String(utf8Charset),
|
|
},
|
|
Body: body,
|
|
},
|
|
}, nil
|
|
}
|
|
|
|
func normalizeEmailAddresses(addresses []string) ([]string, error) {
|
|
if len(addresses) == 0 {
|
|
return nil, nil
|
|
}
|
|
out := make([]string, 0, len(addresses))
|
|
for _, address := range addresses {
|
|
address = strings.TrimSpace(address)
|
|
if address == "" {
|
|
continue
|
|
}
|
|
parsed, err := mail.ParseAddress(address)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, parsed.Address)
|
|
}
|
|
if len(out) == 0 {
|
|
return nil, nil
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func buildSESTags(tags map[string]string) []sestypes.MessageTag {
|
|
if len(tags) == 0 {
|
|
return nil
|
|
}
|
|
keys := make([]string, 0, len(tags))
|
|
for key := range tags {
|
|
keys = append(keys, key)
|
|
}
|
|
sort.Strings(keys)
|
|
out := make([]sestypes.MessageTag, 0, len(keys))
|
|
for _, key := range keys {
|
|
value := strings.TrimSpace(tags[key])
|
|
if strings.TrimSpace(key) == "" || value == "" {
|
|
continue
|
|
}
|
|
out = append(out, sestypes.MessageTag{
|
|
Name: awsv2.String(strings.TrimSpace(key)),
|
|
Value: awsv2.String(value),
|
|
})
|
|
}
|
|
if len(out) == 0 {
|
|
return nil
|
|
}
|
|
return out
|
|
}
|
|
|
|
func sanitizeSubject(subject string) string {
|
|
subject = strings.TrimSpace(subject)
|
|
subject = strings.ReplaceAll(subject, "\r", "")
|
|
subject = strings.ReplaceAll(subject, "\n", "")
|
|
return subject
|
|
}
|
|
|
|
func newSESAPI(ctx context.Context, cfg config.SESConfig) (sesAPI, error) {
|
|
if err := validateSESConfig(cfg); err != nil {
|
|
return nil, err
|
|
}
|
|
loadOptions := make([]func(*awsconfig.LoadOptions) error, 0, 1)
|
|
if strings.TrimSpace(cfg.Region) != "" {
|
|
loadOptions = append(loadOptions, awsconfig.WithRegion(strings.TrimSpace(cfg.Region)))
|
|
}
|
|
awsCfg, err := awsconfig.LoadDefaultConfig(ctx, loadOptions...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return sesv2.NewFromConfig(awsCfg, func(options *sesv2.Options) {
|
|
if endpoint := strings.TrimSpace(cfg.Endpoint); endpoint != "" {
|
|
options.EndpointResolver = sesv2.EndpointResolverFromURL(endpoint)
|
|
}
|
|
}), nil
|
|
}
|
|
|
|
func validateSESConfig(cfg config.SESConfig) error {
|
|
if strings.TrimSpace(cfg.Region) == "" {
|
|
return errors.New("aws region is required")
|
|
}
|
|
if strings.TrimSpace(cfg.From) == "" {
|
|
return errors.New("ses from address is required")
|
|
}
|
|
if endpoint := strings.TrimSpace(cfg.Endpoint); endpoint != "" {
|
|
parsed, err := url.Parse(endpoint)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if parsed.Scheme != "https" && !cfg.AllowInsecureEndpoint {
|
|
return errors.New("insecure ses endpoint is disabled")
|
|
}
|
|
}
|
|
return nil
|
|
}
|