consumer.go

package main

import (
	"fmt"
	"log"
	"net/smtp"
	"sync"
	"time"
)

func emailWorker(id int, ch <-chan DispatchJob, retryCh chan<- DispatchJob, wg *sync.WaitGroup, taskWg *sync.WaitGroup, limiter <-chan time.Time) {
	defer wg.Done()
	for job := range ch {
		<-limiter
		start := time.Now()
		smtpHost := "localhost"
		smtpPort := "1025"
		recipient := job.Recipient
		t := job.Template
		// formattedMsg := fmt.Sprintf("To: %s\r\nSubject: Test Email\r\n\r\n%s\r\n", recipient.Email, "Just testing our email campaign.")
		// msg := []byte(formattedMsg)

		msg, err := executeTemplate(t, recipient)
		if err != nil {
			log.Printf("Worker %d: Error executing template for %s: %v\n",
				id, recipient.Email, err)
			emailsFailedTotal.Inc()
			taskWg.Done()
			continue
		}
		fmt.Printf("Worker %d: Sending email to %s\n", id, recipient.Email)
		err = smtp.SendMail(smtpHost+":"+smtpPort, nil, "devflow.pro27@gmail.com", []string{recipient.Email}, []byte(msg))
		if err != nil {
			log.Printf("Worker %d: Failed to send email to %s: %v\n",
				id, recipient.Email, err)
			if recipient.Retries < 3 {
				recipient.Retries++
				emailsRetriedTotal.Inc()
				log.Printf("Worker %d: Retrying %s (attempt %d)\n", id, recipient.Email, recipient.Retries)
				go func(j DispatchJob) {
					time.Sleep(1 * time.Second)
					retryCh <- j
				}(job)
			} else {
				log.Printf("Worker %d: Permanent failure for %s after 3 retries\n", id, recipient.Email)
				emailsFailedTotal.Inc()
				taskWg.Done()
			}
			continue
		}
		time.Sleep(50 * time.Millisecond)
		fmt.Printf("Worker %d: Sent email to %s\n", id, recipient.Email)
		emailsSentTotal.Inc()
		emailProcessingDuration.Observe(time.Since(start).Seconds())
		taskWg.Done()
	}
}