implement sqlite write queue

core/email_queue.go

@@ -51,6 +51,7 @@ func NewSESClient() SESClient {
 }
 
 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)
 	}
@@ -102,6 +103,7 @@ func (s *SESClient) SendEmail(email Email) error {
 
 // EmailQueue represents the email queue with rate limiting
 type EmailQueue struct {
+	stop            bool
 	emailQueue      chan Email       // Email queue
 	rateLimit       <-chan time.Time // Rate limiter
 	client          EmailClient      // SES client to send emails
@@ -115,6 +117,7 @@ func NewEmailQueue(bufferSize int, emailsPerSecond int, client EmailClient) *Ema
 	rateLimit := time.Tick(time.Second / time.Duration(emailsPerSecond))
 
 	return &EmailQueue{
+		stop:            false,
 		emailQueue:      make(chan Email, bufferSize),
 		rateLimit:       rateLimit,
 		client:          client,
@@ -124,12 +127,17 @@ func NewEmailQueue(bufferSize int, emailsPerSecond int, client EmailClient) *Ema
 
 // AddEmail queues a new email to be sent
 func (q *EmailQueue) AddEmail(email Email) {
+	if q.stop {
+		log.Printf("email %s with subject %s not add. Stopping queue", email.Recipient, email.Subject)
+		return
+	}
 	q.wg.Add(1)
 	q.emailQueue <- email
 }
 
 // Start begins processing the email queue with rate limiting
 func (q *EmailQueue) Start() {
+	q.stop = false
 	// Worker goroutine that processes emails from the queue
 	go func() {
 		for email := range q.emailQueue {
@@ -150,6 +158,7 @@ func (q *EmailQueue) sendEmail(email Email) {
 
 // Stop stops the queue after all emails have been processed
 func (q *EmailQueue) Stop() {
+	q.stop = true
 	// Wait for all emails to be processed
 	q.wg.Wait()
 	// Close the email queue