Merge pull request #21 from one-project-one-month/feature/email-service

Feature/email service

Author: SwanHtetAungPhyo

.gitignore

@@ -13,7 +13,7 @@
 .Trashes
 ehthumbs.db
 Thumbs.db
-
+backend/**/.env
 # Terraform
 infra/.terraform/
 infra/.terraform.lock.hcl

backend/email-service/Dockerfile

@@ -0,0 +1,18 @@
+FROM golang:1.21-alpine AS builder
+
+WORKDIR /app
+COPY go.mod go.sum ./
+RUN go mod download
+
+COPY . .
+RUN CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -o main .
+
+FROM alpine:latest
+RUN apk --no-cache add ca-certificates
+WORKDIR /root/
+
+COPY --from=builder /app/main .
+COPY --from=builder /app/templates ./templates/
+
+EXPOSE 8080
+CMD ["./main"]

backend/email-service/cmd/main.go

@@ -0,0 +1,177 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"os"
+	"os/signal"
+	"runtime"
+	"sync"
+	"syscall"
+	"time"
+
+	"github.com/multi-tenants-cms-golang/email-service/config"
+	"github.com/multi-tenants-cms-golang/email-service/internal/email"
+	"github.com/multi-tenants-cms-golang/email-service/internal/health"
+	natss "github.com/multi-tenants-cms-golang/email-service/internal/nats"
+	"github.com/nats-io/nats.go"
+	"go.uber.org/zap"
+)
+
+func main() {
+	logger, err := zap.NewProduction()
+	if err != nil {
+		panic(err)
+	}
+	defer func(logger *zap.Logger) {
+		err := logger.Sync()
+		if err != nil {
+			panic(err.Error())
+		}
+	}(logger)
+
+	cfg, err := config.Load()
+	if err != nil {
+		logger.Fatal("failed to load configuration", zap.Error(err))
+	}
+
+	// Connect to NATS
+	nc, err := nats.Connect(cfg.NATS.URL,
+		nats.ReconnectWait(time.Second),
+		nats.MaxReconnects(-1),
+		nats.DisconnectErrHandler(func(nc *nats.Conn, err error) {
+			logger.Error("NATS disconnected", zap.Error(err))
+		}),
+		nats.ReconnectHandler(func(nc *nats.Conn) {
+			logger.Info("NATS reconnected")
+		}),
+	)
+	if err != nil {
+		logger.Fatal("failed to connect to NATS", zap.Error(err))
+	}
+	defer nc.Close()
+
+	emailCh := make(chan natss.EmailRequest, 100)
+
+	// Create email service
+	emailService, err := email.NewService(
+		logger,
+		cfg.SMTP.Host,
+		cfg.SMTP.Port,
+		cfg.SMTP.User,
+		cfg.SMTP.Password,
+		cfg.SMTP.FromAddr,
+		cfg.SMTP.TemplateDir,
+	)
+	if err != nil {
+		logger.Fatal("failed to create email service", zap.Error(err))
+	}
+
+	// Create NATS consumer
+	natsConsumer, err := natss.NewConsumer(
+		nc,
+		logger,
+		emailCh,
+		cfg.NATS.StreamName,
+		cfg.NATS.Subject,
+		cfg.NATS.ConsumerName,
+	)
+	if err != nil {
+		logger.Fatal("failed to create NATS consumer", zap.Error(err))
+	}
+
+	// Create contexts
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	var wg sync.WaitGroup
+
+	// Start worker goroutines
+	workerCount := runtime.NumCPU()
+	logger.Info("starting email workers", zap.Int("count", workerCount))
+
+	for i := 0; i < workerCount; i++ {
+		wg.Add(1)
+		go func(workerID int) {
+			defer wg.Done()
+			logger.Info("worker started", zap.Int("id", workerID))
+
+			for {
+				select {
+				case req, ok := <-emailCh:
+					if !ok {
+						logger.Info("worker stopping", zap.Int("id", workerID))
+						return
+					}
+
+					if err := emailService.Send(ctx, req); err != nil {
+						logger.Error("failed to send email",
+							zap.Int("worker", workerID),
+							zap.String("to", req.To),
+							zap.String("template", req.Template),
+							zap.Error(err))
+					}
+				case <-ctx.Done():
+					logger.Info("worker stopping due to context cancellation", zap.Int("id", workerID))
+					return
+				}
+			}
+		}(i)
+	}
+
+	// Start NATS consumer
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := natsConsumer.Start(ctx); err != nil {
+			logger.Error("NATS consumer failed", zap.Error(err))
+		}
+	}()
+
+	// Start health server
+	healthServer := health.NewServer(cfg.Server.Port, logger)
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		if err := healthServer.Start(); !errors.Is(err, http.ErrServerClosed) && err != nil {
+			logger.Error("health server failed", zap.Error(err))
+		}
+	}()
+
+	// Wait for shutdown signal
+	sigCh := make(chan os.Signal, 1)
+	signal.Notify(sigCh, syscall.SIGINT, syscall.SIGTERM)
+	<-sigCh
+
+	logger.Info("shutdown signal received, starting graceful shutdown")
+
+	// Create shutdown context
+	shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), cfg.Server.ShutdownTimeout)
+	defer shutdownCancel()
+
+	// Stop health server
+	if err := healthServer.Stop(shutdownCtx); err != nil {
+		logger.Error("failed to stop health server gracefully", zap.Error(err))
+	}
+
+	// Cancel main context to stop all goroutines
+	cancel()
+
+	close(emailCh)
+
+	done := make(chan struct{})
+	go func() {
+		wg.Wait()
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		logger.Info("all goroutines stopped")
+	case <-shutdownCtx.Done():
+		logger.Warn("shutdown timeout exceeded, forcing exit")
+	}
+
+	logger.Info("shutdown complete")
+}

backend/email-service/config/config.go

@@ -0,0 +1,94 @@
+package config
+
+import (
+	"fmt"
+	"os"
+	"strconv"
+	"time"
+)
+
+type Config struct {
+	Server ServerConfig
+	NATS   NATSConfig
+	SMTP   SMTPConfig
+}
+
+type ServerConfig struct {
+	Port            string
+	ShutdownTimeout time.Duration
+}
+
+type NATSConfig struct {
+	URL          string
+	StreamName   string
+	Subject      string
+	ConsumerName string
+}
+
+type SMTPConfig struct {
+	Host        string
+	Port        int
+	User        string
+	Password    string
+	FromAddr    string
+	TemplateDir string
+}
+
+func Load() (*Config, error) {
+	smtpPortStr := os.Getenv("SMTP_PORT")
+	if smtpPortStr == "" {
+		smtpPortStr = "587"
+	}
+
+	smtpPort, err := strconv.Atoi(smtpPortStr)
+	if err != nil {
+		return nil, fmt.Errorf("invalid SMTP_PORT: %w", err)
+	}
+
+	cfg := &Config{
+		Server: ServerConfig{
+			Port:            getEnvOrDefault("SERVER_PORT", "8080"),
+			ShutdownTimeout: 30 * time.Second,
+		},
+		NATS: NATSConfig{
+			URL:          getEnvOrDefault("NATS_URL", "nats://localhost:4222"),
+			StreamName:   getEnvOrDefault("STREAM_NAME", "email-stream"),
+			Subject:      getEnvOrDefault("SUBJECT_TO", "email.send"),
+			ConsumerName: getEnvOrDefault("CONSUMER_NAME", "email-consumer"),
+		},
+		SMTP: SMTPConfig{
+			Host:        getEnvOrDefault("SMTP_HOST", "localhost"),
+			Port:        smtpPort,
+			User:        os.Getenv("SMTP_USER"),
+			Password:    os.Getenv("SMTP_PASSWORD"),
+			FromAddr:    os.Getenv("FROM_ADDR"),
+			TemplateDir: getEnvOrDefault("TEMPLATE_DIR", "./templates"),
+		},
+	}
+
+	if err := cfg.validate(); err != nil {
+		return nil, fmt.Errorf("config validation failed: %w", err)
+	}
+
+	return cfg, nil
+}
+
+func (c *Config) validate() error {
+	if c.SMTP.User == "" {
+		return fmt.Errorf("SMTP_USER is required")
+	}
+	if c.SMTP.Password == "" {
+		return fmt.Errorf("SMTP_PASSWORD is required")
+	}
+	if c.SMTP.FromAddr == "" {
+		return fmt.Errorf("FROM_ADDR is required")
+	}
+	return nil
+}
+
+func getEnvOrDefault(key, defaultValue string) string {
+	if value := os.Getenv(key); value != "" {
+		return value
+	}
+	return defaultValue
+}

backend/email-service/go.mod

@@ -0,0 +1,19 @@
+module github.com/multi-tenants-cms-golang/email-service
+
+go 1.24
+
+require (
+	github.com/nats-io/nats.go v1.43.0
+	github.com/wneessen/go-mail v0.6.2
+	go.uber.org/zap v1.27.0
+)
+
+require (
+	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/nats-io/nkeys v0.4.11 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
+	go.uber.org/multierr v1.10.0 // indirect
+	golang.org/x/crypto v0.37.0 // indirect
+	golang.org/x/sys v0.32.0 // indirect
+	golang.org/x/text v0.24.0 // indirect
+)