updates

server/internal/queue/email_service.go

@@ -0,0 +1,81 @@
+// internal/queue/email_service.go
+package queue
+
+import (
+	"fmt"
+
+	"gopkg.in/gomail.v2"
+)
+
+// EmailService defines the interface for sending emails
+type EmailService interface {
+	Send(to string, subject string, body string) error
+}
+
+// SMTPEmailService implements EmailService using SMTP
+type SMTPEmailService struct {
+	dialer *gomail.Dialer
+	from   string
+}
+
+type SMTPConfig struct {
+	Host     string
+	Port     int
+	Username string
+	Password string
+	From     string
+}
+
+func NewSMTPEmailService(config SMTPConfig) *SMTPEmailService {
+	dialer := gomail.NewDialer(
+		config.Host,
+		config.Port,
+		config.Username,
+		config.Password,
+	)
+
+	return &SMTPEmailService{
+		dialer: dialer,
+		from:   config.From,
+	}
+}
+
+func (s *SMTPEmailService) Send(to string, subject string, body string) error {
+	m := gomail.NewMessage()
+	m.SetHeader("From", s.from)
+	m.SetHeader("To", to)
+	m.SetHeader("Subject", subject)
+	m.SetBody("text/html", body)
+
+	if err := s.dialer.DialAndSend(m); err != nil {
+		return fmt.Errorf("failed to send email: %w", err)
+	}
+
+	return nil
+}
+
+// MockEmailService implements EmailService for testing
+type MockEmailService struct {
+	SentEmails []MockEmail
+}
+
+type MockEmail struct {
+	To      string
+	Subject string
+	Body    string
+}
+
+func NewMockEmailService() *MockEmailService {
+	return &MockEmailService{
+		SentEmails: make([]MockEmail, 0),
+	}
+}
+
+func (s *MockEmailService) Send(to string, subject string, body string) error {
+	s.SentEmails = append(s.SentEmails, MockEmail{
+		To:      to,
+		Subject: subject,
+		Body:    body,
+	})
+	return nil
+}

server/internal/queue/worker.go

@@ -0,0 +1,88 @@
+// internal/queue/worker.go
+package queue
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/hibiken/asynq"
+)
+
+const (
+	TypeEmailDelivery = "email:deliver"
+	TypeDataExport    = "data:export"
+)
+
+type TaskHandler struct {
+	emailService EmailService
+	// Add other services needed for tasks
+}
+
+func NewTaskHandler(emailService EmailService) *TaskHandler {
+	return &TaskHandler{
+		emailService: emailService,
+	}
+}
+
+func NewQueueClient(redisAddr string) *asynq.Client {
+	return asynq.NewClient(asynq.RedisClientOpt{Addr: redisAddr})
+}
+
+func StartWorkerServer(redisAddr string, handler *TaskHandler) {
+	srv := asynq.NewServer(
+		asynq.RedisClientOpt{Addr: redisAddr},
+		asynq.Config{
+			Concurrency: 10,
+			Queues: map[string]int{
+				"critical": 6,
+				"default":  3,
+				"low":      1,
+			},
+		},
+	)
+
+	mux := asynq.NewServeMux()
+	mux.HandleFunc(TypeEmailDelivery, handler.HandleEmailDeliveryTask)
+	mux.HandleFunc(TypeDataExport, handler.HandleDataExportTask)
+
+	if err := srv.Run(mux); err != nil {
+		log.Fatalf("Could not run queue server: %v", err)
+	}
+}
+
+// Task Handlers
+func (h *TaskHandler) HandleEmailDeliveryTask(ctx context.Context, t *asynq.Task) error {
+	var p EmailDeliveryPayload
+	if err := json.Unmarshal(t.Payload(), &p); err != nil {
+		return fmt.Errorf("json.Unmarshal failed: %v", err)
+	}
+
+	// Process the email delivery task
+	return h.emailService.Send(p.To, p.Subject, p.Body)
+}
+
+func (h *TaskHandler) HandleDataExportTask(ctx context.Context, t *asynq.Task) error {
+	var p DataExportPayload
+	if err := json.Unmarshal(t.Payload(), &p); err != nil {
+		return fmt.Errorf("json.Unmarshal failed: %v", err)
+	}
+
+	// Process the data export task
+	return nil
+}
+
+// Task Payloads
+type EmailDeliveryPayload struct {
+	To      string
+	Subject string
+	Body    string
+}
+
+type DataExportPayload struct {
+	UserID    string
+	Format    string
+	Timestamp time.Time
+}