153 lines
3.8 KiB
Go
153 lines
3.8 KiB
Go
package core
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"github.com/aws/aws-sdk-go-v2/aws"
|
|
"github.com/aws/aws-sdk-go-v2/config"
|
|
"github.com/aws/aws-sdk-go-v2/service/ses"
|
|
"github.com/aws/aws-sdk-go-v2/service/ses/types"
|
|
"github.com/patrickmn/go-cache"
|
|
"log"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
type EmailClient interface {
|
|
SendEmail(Email) error
|
|
}
|
|
|
|
// Email represents a dummy email structure
|
|
type Email struct {
|
|
Sender string
|
|
Recipient string
|
|
Subject string
|
|
Content string
|
|
}
|
|
|
|
type TestEmailClient struct{}
|
|
|
|
// SendEmail simulates sending an email via AWS SES
|
|
func (c *TestEmailClient) SendEmail(email Email) error {
|
|
// Simulate sending email (replace with actual AWS SES API call)
|
|
fmt.Printf("Sending email to %s\n", email.Recipient)
|
|
return nil
|
|
}
|
|
|
|
type SESClient struct {
|
|
ResetCache *cache.Cache
|
|
}
|
|
|
|
const (
|
|
defaultExpiration = 5 * time.Minute
|
|
cleanupInterval = 10 * time.Minute
|
|
)
|
|
|
|
func NewSESClient() SESClient {
|
|
return SESClient{
|
|
ResetCache: cache.New(defaultExpiration, cleanupInterval),
|
|
}
|
|
}
|
|
|
|
func (s *SESClient) SendEmail(email Email) error {
|
|
if _, exists := s.ResetCache.Get(email.Recipient); exists {
|
|
return fmt.Errorf("email already sent to %s with subject %s", email.Recipient, email.Subject)
|
|
}
|
|
|
|
// Load AWS configuration
|
|
cfg, err := config.LoadDefaultConfig(context.TODO(), config.WithRegion("us-east-1"))
|
|
if err != nil {
|
|
errMsg := fmt.Sprintf("unable to load SDK config, %v", err)
|
|
log.Print(errMsg)
|
|
return errors.New(errMsg)
|
|
}
|
|
|
|
// Create an SES client
|
|
sesClient := ses.NewFromConfig(cfg)
|
|
|
|
// Construct the email message
|
|
input := &ses.SendEmailInput{
|
|
Destination: &types.Destination{
|
|
ToAddresses: []string{email.Recipient},
|
|
},
|
|
Message: &types.Message{
|
|
Body: &types.Body{
|
|
Html: &types.Content{
|
|
Data: aws.String(email.Content),
|
|
},
|
|
},
|
|
Subject: &types.Content{
|
|
Data: aws.String(email.Subject),
|
|
},
|
|
},
|
|
Source: aws.String(email.Sender),
|
|
}
|
|
|
|
// Send the email
|
|
resp, err := sesClient.SendEmail(context.TODO(), input)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to send email, %v", err)
|
|
}
|
|
|
|
// Output the message ID of the sent email
|
|
fmt.Printf("UserEmail sent successfully, Message ID: %s\n", *resp.MessageId)
|
|
return nil
|
|
}
|
|
|
|
// EmailQueue represents the email queue with rate limiting
|
|
type EmailQueue struct {
|
|
emailQueue chan Email // Email queue
|
|
rateLimit <-chan time.Time // Rate limiter
|
|
client EmailClient // SES client to send emails
|
|
wg sync.WaitGroup // To wait for all emails to be processed
|
|
FailedSendCount int
|
|
}
|
|
|
|
// NewEmailQueue creates a new rate-limited email queue
|
|
func NewEmailQueue(bufferSize int, emailsPerSecond int, client EmailClient) *EmailQueue {
|
|
// Create a ticker that ticks every second to limit the rate of sending emails
|
|
rateLimit := time.Tick(time.Second / time.Duration(emailsPerSecond))
|
|
|
|
return &EmailQueue{
|
|
emailQueue: make(chan Email, bufferSize),
|
|
rateLimit: rateLimit,
|
|
client: client,
|
|
FailedSendCount: 0,
|
|
}
|
|
}
|
|
|
|
// AddEmail queues a new email to be sent
|
|
func (q *EmailQueue) AddEmail(email Email) {
|
|
q.wg.Add(1)
|
|
q.emailQueue <- email
|
|
}
|
|
|
|
// Start begins processing the email queue with rate limiting
|
|
func (q *EmailQueue) Start() {
|
|
// Worker goroutine that processes emails from the queue
|
|
go func() {
|
|
for email := range q.emailQueue {
|
|
<-q.rateLimit // Wait for the rate limiter to allow the next email
|
|
q.sendEmail(email)
|
|
q.wg.Done() // Mark the email as processed
|
|
}
|
|
}()
|
|
}
|
|
|
|
// sendEmail sends an email using the SES client
|
|
func (q *EmailQueue) sendEmail(email Email) {
|
|
if err := q.client.SendEmail(email); err != nil {
|
|
q.FailedSendCount += 1
|
|
log.Printf("Failed to send email to %s: %v\n", email.Recipient, err)
|
|
}
|
|
}
|
|
|
|
// Stop stops the queue after all emails have been processed
|
|
func (q *EmailQueue) Stop() {
|
|
// Wait for all emails to be processed
|
|
q.wg.Wait()
|
|
// Close the email queue
|
|
close(q.emailQueue)
|
|
}
|