feat: new queue and workers

Author: AlexanderWangY

apps/api/.env.dev.example

@@ -1,5 +1,6 @@
 DATABASE_URL="postgres://postgres:postgres@postgres:5432/coredb"
 DATABASE_URL_MIGRATION="postgres://postgres:postgres@localhost:5432/coredb"
+REDIS_URL="redis:6379"
 
 # For OAuth
 AUTH_DISCORD_CLIENT_ID=

apps/api/cmd/api/main.go

@@ -4,6 +4,7 @@ import (
 	"net/http"
 	"time"
 
+	"github.com/hibiken/asynq"
 	"github.com/rs/zerolog/log"
 	"github.com/swamphacks/core/apps/api/internal/api"
 	"github.com/swamphacks/core/apps/api/internal/api/handlers"
@@ -38,6 +39,11 @@ func main() {
 		},
 	}
 
+	// Create asynq client
+	taskQueueClient := asynq.NewClient(asynq.RedisClientOpt{
+		Addr: cfg.RedisURL,
+	})
+
 	// Create new middleware injectable
 	mw := middleware.NewMiddleware(database, logger, cfg)
 
@@ -50,9 +56,10 @@ func main() {
 	// Injections into services
 	authService := services.NewAuthService(userRepo, accountRepo, sessionRepo, txm, client, logger, &cfg.Auth)
 	eventInterestService := services.NewEventInterestService(eventInterestRepo, logger)
+	emailService := services.NewEmailService(taskQueueClient, logger)
 
 	// Injections into handlers
-	apiHandlers := handlers.NewHandlers(authService, eventInterestService, cfg, logger)
+	apiHandlers := handlers.NewHandlers(authService, eventInterestService, emailService, cfg, logger)
 
 	api := api.NewAPI(&logger, apiHandlers, mw)
 

apps/api/cmd/email_worker/Dockerfile

@@ -0,0 +1,27 @@
+FROM golang:1.24-alpine AS base
+
+WORKDIR /app
+
+COPY ../../go.mod ../../go.sum ./
+
+RUN go mod download
+
+COPY ../../ ./
+
+# Dev 
+FROM base AS dev
+
+RUN go install github.com/air-verse/air@latest
+
+CMD ["air"]
+
+# Production
+FROM base as prod
+
+RUN CGO_ENABLED=0 GOOS=linux go build -ldflags="-s -w" -o email_worker ./
+
+RUN apk --no-cache add ca-certificates
+
+CMD ["./email_worker"]
+
+

apps/api/cmd/email_worker/main.go

@@ -1,21 +1,39 @@
 package main
 
 import (
+	"fmt"
 	"log"
 
 	"github.com/hibiken/asynq"
+	"github.com/swamphacks/core/apps/api/internal/config"
+	"github.com/swamphacks/core/apps/api/internal/logger"
+	"github.com/swamphacks/core/apps/api/internal/services"
+	"github.com/swamphacks/core/apps/api/internal/tasks"
+	"github.com/swamphacks/core/apps/api/internal/workers"
 )
 
 func main() {
+	logger := logger.New()
+	cfg := config.Load()
+
 	srv := asynq.NewServer(
-		asynq.RedisClientOpt{Addr: "redis:6379"},
+		asynq.RedisClientOpt{Addr: cfg.RedisURL},
 		asynq.Config{
 			Concurrency: 10,
+			Queues: map[string]int{
+				"email": 10,
+			},
 		},
 	)
 
+	emailService := services.NewEmailService(nil, logger)
+	emailWorker := workers.NewEmailWorker(emailService, logger)
+
 	mux := asynq.NewServeMux()
 
+	mux.HandleFunc(tasks.TypeSendEmail, emailWorker.HandleSendEmailTask)
+	fmt.Println("Starting email worker")
+
 	if err := srv.Run(mux); err != nil {
 		log.Fatalf("Failed to run email worker")
 	}

apps/api/internal/api/api.go

@@ -71,6 +71,11 @@ func (api *API) setupRoutes(mw *mw.Middleware) {
 		r.Post("/{eventId}/interest", api.Handlers.EventInterest.AddEmailToEvent)
 	})
 
+	// Email routes
+	api.Router.Route("/email", func(r chi.Router) {
+		r.Post("/queue", api.Handlers.Email.QueueEmail)
+	})
+
 	// Protected test routes
 	api.Router.Route("/protected", func(r chi.Router) {
 		r.Use(mw.Auth.RequireAuth)

apps/api/internal/api/handlers/email.go

@@ -0,0 +1,58 @@
+package handlers
+
+import (
+	"encoding/json"
+	"net/http"
+
+	"github.com/rs/zerolog"
+	res "github.com/swamphacks/core/apps/api/internal/api/response"
+	"github.com/swamphacks/core/apps/api/internal/email"
+	"github.com/swamphacks/core/apps/api/internal/services"
+)
+
+type EmailHandler struct {
+	emailService *services.EmailService
+	logger       zerolog.Logger
+}
+
+func NewEmailHandler(emailService *services.EmailService, logger zerolog.Logger) *EmailHandler {
+	return &EmailHandler{
+		emailService: emailService,
+		logger:       logger.With().Str("handler", "EmailHandler").Str("component", "email").Logger(),
+	}
+}
+
+type QueueEmailRequest struct {
+	To   string `json:"to"`
+	From string `json:"from"`
+	Body string `json:"body"`
+}
+
+func (h *EmailHandler) QueueEmail(w http.ResponseWriter, r *http.Request) {
+	var req QueueEmailRequest
+	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+		res.SendError(w, http.StatusBadRequest, res.NewError("invalid_request", "Could not parse request body"))
+		return
+	}
+
+	if !email.IsValidEmail(req.To) || !email.IsValidEmail(req.From) {
+		res.SendError(w, http.StatusBadRequest, res.NewError("malformed_email", "To and/or From email is malformed or missing"))
+		return
+	}
+
+	if req.Body == "" {
+		res.SendError(w, http.StatusBadRequest, res.NewError("missing_body", "Body is missing or is an empty string."))
+		return
+	}
+
+	taskInfo, err := h.emailService.QueueSendEmail(req.To, req.From, req.Body)
+	if err != nil {
+		h.logger.Err(err).Msg("Failed to queue email sending from EmailHandler")
+		res.SendError(w, http.StatusInternalServerError, res.NewError("internal_err", "The server went kaput while queueing email sending"))
+		return
+	}
+
+	h.logger.Info().Str("TaskID", taskInfo.ID).Str("Task Queue", taskInfo.Queue).Str("Task Type", taskInfo.Type).Msg("Queued Send Email task!")
+
+	w.WriteHeader(http.StatusCreated)
+}

apps/api/internal/api/handlers/handlers.go

@@ -9,11 +9,13 @@ import (
 type Handlers struct {
 	Auth          *AuthHandler
 	EventInterest *EventInterestHandler
+	Email         *EmailHandler
 }
 
-func NewHandlers(authService *services.AuthService, eventInterestService *services.EventInterestService, cfg *config.Config, logger zerolog.Logger) *Handlers {
+func NewHandlers(authService *services.AuthService, eventInterestService *services.EventInterestService, emailService *services.EmailService, cfg *config.Config, logger zerolog.Logger) *Handlers {
 	return &Handlers{
 		Auth:          NewAuthHandler(authService, cfg, logger),
 		EventInterest: NewEventInterestHandler(eventInterestService, cfg, logger),
+		Email:         NewEmailHandler(emailService, logger),
 	}
 }

apps/api/internal/config/config.go

@@ -26,6 +26,7 @@ type AuthConfig struct {
 
 type Config struct {
 	DatabaseURL string `env:"DATABASE_URL"`
+	RedisURL    string `env:"REDIS_URL"`
 	Port        string `env:"PORT" envDefault:"8080"`
 
 	Auth      AuthConfig   `envPrefix:"AUTH_"`

apps/api/internal/services/email.go

@@ -0,0 +1,45 @@
+package services
+
+import (
+	"github.com/hibiken/asynq"
+	"github.com/rs/zerolog"
+	"github.com/swamphacks/core/apps/api/internal/tasks"
+)
+
+type EmailService struct {
+	logger    zerolog.Logger
+	taskQueue *asynq.Client
+	// Add SMTP client later
+}
+
+func NewEmailService(taskQueue *asynq.Client, logger zerolog.Logger) *EmailService {
+	return &EmailService{
+		logger:    logger.With().Str("service", "EmailService").Str("component", "email").Logger(),
+		taskQueue: taskQueue,
+	}
+}
+
+func (s *EmailService) SendEmail(to, from, body string) error {
+	s.logger.Info().Msgf("Sending from %s to %s with body %s", from, to, body)
+	return nil
+}
+
+func (s *EmailService) QueueSendEmail(to, from, body string) (*asynq.TaskInfo, error) {
+	task, err := tasks.NewTaskSendEmail(tasks.SendEmailPayload{
+		To:   to,
+		From: from,
+		Body: body,
+	})
+	if err != nil {
+		s.logger.Err(err).Msg("Failed to create Send Email task")
+		return nil, err
+	}
+
+	info, err := s.taskQueue.Enqueue(task, asynq.Queue("email"))
+	if err != nil {
+		s.logger.Err(err).Msg("Failed to queue Send Email task")
+		return nil, err
+	}
+
+	return info, nil
+}

apps/api/internal/tasks/email.go

@@ -0,0 +1,26 @@
+package tasks
+
+import (
+	"encoding/json"
+
+	"github.com/hibiken/asynq"
+)
+
+const (
+	TypeSendEmail = "email:send"
+)
+
+type SendEmailPayload struct {
+	To   string
+	From string
+	Body string
+}
+
+func NewTaskSendEmail(payload SendEmailPayload) (*asynq.Task, error) {
+	data, err := json.Marshal(payload)
+	if err != nil {
+		return nil, err
+	}
+
+	return asynq.NewTask(TypeSendEmail, data), nil
+}