feat: introduces proper job queueing

cmd/digest.go

@@ -85,7 +85,7 @@ func runDigestTest(cmd *cobra.Command, args []string) {
 		sugar.Fatalw("Failed to initialize email service", "error", err)
 	}
 
-	digestService := digest.NewService(db, emailService, cfg, sugar)
+	digestService := digest.NewService(db, emailService, nil, cfg, sugar)
 
 	if dryRun {
 		sugar.Info("Running digest test in DRY-RUN mode (no emails will be sent)...")
@@ -145,7 +145,7 @@ func runDigestPreview(cmd *cobra.Command, args []string) {
 		sugar.Warnw("Failed to initialize email service", "error", err)
 	}
 
-	digestService := digest.NewService(db, emailService, cfg, sugar)
+	digestService := digest.NewService(db, emailService, nil, cfg, sugar)
 
 	summary, err := digestService.GetGlobalEvidenceSummary(ctx)
 	if err != nil {

cmd/run.go

@@ -14,6 +14,7 @@ import (
 	"github.com/compliance-framework/api/internal/service/digest"
 	"github.com/compliance-framework/api/internal/service/email"
 	"github.com/compliance-framework/api/internal/service/scheduler"
+	"github.com/compliance-framework/api/internal/service/worker"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 	"go.uber.org/zap"
@@ -61,42 +62,38 @@ func RunServer(cmd *cobra.Command, args []string) {
 		sugar.Warnw("Failed to initialize email service, digests will be disabled", "error", err)
 	}
 
-	// Initialize digest service
-	digestService := digest.NewService(db, emailService, cfg, sugar)
+	// Initialize digest service (without worker service initially)
+	digestService := digest.NewService(db, emailService, nil, cfg, sugar)
 
-	// Initialize scheduler
-	sched := scheduler.NewCronScheduler(sugar)
+	// Initialize worker service with digest support
+	workerService, err := worker.NewServiceWithDigest(cfg.Worker, db, emailService, digestService, cfg, sugar)
+	if err != nil {
+		sugar.Fatalw("Failed to initialize worker service", "error", err)
+	}
 
-	// Register digest job using config
-	if cfg.DigestEnabled {
-		digestJob := digest.NewGlobalDigestJob(digestService, sugar)
-		if err := sched.ScheduleCron(cfg.DigestSchedule, digestJob); err != nil {
-			sugar.Warnw("Failed to schedule digest job", "schedule", cfg.DigestSchedule, "error", err)
-		} else {
-			sugar.Debugw("Digest job scheduled", "schedule", cfg.DigestSchedule)
-		}
-	} else {
-		sugar.Debugw("Digest scheduler disabled")
+	// Set worker service reference in digest service to avoid circular dependency
+	digestService.SetWorkerService(workerService)
+
+	// Run River migrations
+	if err := workerService.Migrate(ctx); err != nil {
+		sugar.Fatalw("Failed to run River migrations", "error", err)
+	}
+
+	// Start worker service
+	if err := workerService.Start(ctx); err != nil {
+		sugar.Fatalw("Failed to start worker service", "error", err)
 	}
 
-	// Start the scheduler
+	// Initialize scheduler for other jobs (if any)
+	// Note: Digest scheduling is now handled by River's periodic jobs
+	sched := scheduler.NewCronScheduler(sugar)
 	sched.Start()
-	defer func() {
-		stopCtx := sched.Stop()
-		// Wait for jobs to finish gracefully with a 10-second timeout
-		select {
-		case <-stopCtx.Done():
-			sugar.Debug("All scheduled jobs completed gracefully")
-		case <-time.After(10 * time.Second):
-			sugar.Warn("Scheduler shutdown timeout, some jobs may not have completed")
-		}
-	}()
 
 	metrics := api.NewMetricsHandler(ctx, sugar)
 	server := api.NewServer(ctx, sugar, cfg, metrics)
 	handler.RegisterHandlers(server, sugar, db, cfg, digestService, sched)
 	oscal.RegisterHandlers(server, sugar, db, cfg)
-	auth.RegisterHandlers(server, sugar, db, cfg, metrics)
+	auth.RegisterHandlers(server, sugar, db, cfg, metrics, emailService, workerService)
 
 	sugar.Infow("Allowed Origins", "origins", cfg.APIAllowedOrigins)
 	server.PrintRoutes()
@@ -109,4 +106,24 @@ func RunServer(cmd *cobra.Command, args []string) {
 	if err := server.Start(cfg.AppPort); err != nil {
 		sugar.Fatalw("Failed to start server", "error", err)
 	}
+
+	// Note: Defer statements are registered in reverse order of execution.
+	// This ensures proper shutdown order: scheduler -> worker service
+	defer func() {
+		// Stop worker service last (after scheduler has stopped)
+		if err := workerService.Stop(ctx); err != nil {
+			sugar.Errorw("Failed to stop worker service", "error", err)
+		}
+	}()
+
+	defer func() {
+		// Stop scheduler first
+		stopCtx := sched.Stop()
+		select {
+		case <-stopCtx.Done():
+			sugar.Debug("All scheduled jobs completed gracefully")
+		case <-time.After(10 * time.Second):
+			sugar.Warn("Scheduler shutdown timeout, some jobs may not have completed")
+		}
+	}()
 }

go.mod

@@ -13,11 +13,13 @@ require (
 	github.com/go-playground/validator/v10 v10.28.0
 	github.com/golang-jwt/jwt/v5 v5.3.0
 	github.com/google/uuid v1.6.0
-	github.com/jackc/pgx/v5 v5.7.6
+	github.com/jackc/pgx/v5 v5.8.0
 	github.com/joho/godotenv v1.5.1
 	github.com/labstack/echo-contrib v0.17.4
 	github.com/labstack/echo/v4 v4.13.4
 	github.com/prometheus/client_golang v1.23.2
+	github.com/riverqueue/river v0.30.1
+	github.com/riverqueue/river/riverdriver/riverpgxv5 v0.30.1
 	github.com/robfig/cron/v3 v3.0.1
 	github.com/schollz/progressbar/v3 v3.18.0
 	github.com/spf13/cobra v1.10.2
@@ -135,6 +137,9 @@ require (
 	github.com/prometheus/client_model v0.6.2 // indirect
 	github.com/prometheus/common v0.67.4 // indirect
 	github.com/prometheus/procfs v0.19.2 // indirect
+	github.com/riverqueue/river/riverdriver v0.30.1 // indirect
+	github.com/riverqueue/river/rivershared v0.30.1 // indirect
+	github.com/riverqueue/river/rivertype v0.30.1 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/sagikazarmark/locafero v0.12.0 // indirect
@@ -144,9 +149,14 @@ require (
 	github.com/spf13/afero v1.15.0 // indirect
 	github.com/spf13/cast v1.10.0 // indirect
 	github.com/spf13/pflag v1.0.10 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
 	github.com/subosito/gotenv v1.6.0 // indirect
 	github.com/swaggo/files/v2 v2.0.2 // indirect
 	github.com/tdewolff/parse/v2 v2.7.15 // indirect
+	github.com/tidwall/gjson v1.18.0 // indirect
+	github.com/tidwall/match v1.2.0 // indirect
+	github.com/tidwall/pretty v1.2.1 // indirect
+	github.com/tidwall/sjson v1.2.5 // indirect
 	github.com/tklauser/go-sysconf v0.3.12 // indirect
 	github.com/tklauser/numcpus v0.6.1 // indirect
 	github.com/urfave/cli/v2 v2.3.0 // indirect
@@ -159,15 +169,16 @@ require (
 	go.opentelemetry.io/otel/metric v1.35.0 // indirect
 	go.opentelemetry.io/otel/sdk v1.34.0 // indirect
 	go.opentelemetry.io/otel/trace v1.35.0 // indirect
+	go.uber.org/goleak v1.3.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.yaml.in/yaml/v2 v2.4.3 // indirect
 	go.yaml.in/yaml/v3 v3.0.4 // indirect
-	golang.org/x/mod v0.31.0 // indirect
+	golang.org/x/mod v0.32.0 // indirect
 	golang.org/x/net v0.48.0 // indirect
 	golang.org/x/sync v0.19.0 // indirect
 	golang.org/x/sys v0.39.0 // indirect
 	golang.org/x/term v0.38.0 // indirect
-	golang.org/x/text v0.32.0 // indirect
+	golang.org/x/text v0.33.0 // indirect
 	golang.org/x/time v0.14.0 // indirect
 	golang.org/x/tools v0.40.0 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20250218202821-56aae31c358a // indirect

go.sum

@@ -256,12 +256,14 @@ github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2
 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw=
 github.com/invopop/yaml v0.2.0 h1:7zky/qH+O0DwAyoobXUqvVBwgBFRxKoQ/3FjcVpjTMY=
 github.com/invopop/yaml v0.2.0/go.mod h1:2XuRLgs/ouIrW3XNzuNj7J3Nvu/Dig5MXvbCEdiBN3Q=
+github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438 h1:Dj0L5fhJ9F82ZJyVOmBx6msDp/kfd1t9GRfny/mfJA0=
+github.com/jackc/pgerrcode v0.0.0-20240316143900-6e2875d9b438/go.mod h1:a/s9Lp5W7n/DD0VrVoyJ00FbP2ytTPDVOivvn2bMlds=
 github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
 github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
 github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
-github.com/jackc/pgx/v5 v5.7.6 h1:rWQc5FwZSPX58r1OQmkuaNicxdmExaEz5A2DO2hUuTk=
-github.com/jackc/pgx/v5 v5.7.6/go.mod h1:aruU7o91Tc2q2cFp5h4uP3f6ztExVpyVv88Xl/8Vl8M=
+github.com/jackc/pgx/v5 v5.8.0 h1:TYPDoleBBme0xGSAX3/+NujXXtpZn9HBONkQC7IEZSo=
+github.com/jackc/pgx/v5 v5.8.0/go.mod h1:QVeDInX2m9VyzvNeiCJVjCkNFqzsNb43204HshNSZKw=
 github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
 github.com/jdkato/prose v1.2.1 h1:Fp3UnJmLVISmlc57BgKUzdjr0lOtjqTZicL3PaYy6cU=
@@ -377,6 +379,16 @@ github.com/prometheus/common v0.67.4 h1:yR3NqWO1/UyO1w2PhUvXlGQs/PtFmoveVO0KZ4+L
 github.com/prometheus/common v0.67.4/go.mod h1:gP0fq6YjjNCLssJCQp0yk4M8W6ikLURwkdd/YKtTbyI=
 github.com/prometheus/procfs v0.19.2 h1:zUMhqEW66Ex7OXIiDkll3tl9a1ZdilUOd/F6ZXw4Vws=
 github.com/prometheus/procfs v0.19.2/go.mod h1:M0aotyiemPhBCM0z5w87kL22CxfcH05ZpYlu+b4J7mw=
+github.com/riverqueue/river v0.30.1 h1:lpwmDT3zD+iDtF4tD50e/Y23UHpIeBUffVTDr2khN+s=
+github.com/riverqueue/river v0.30.1/go.mod h1:x9tVfiCrbOctSAmaYP00iE5YlO8zh3Y9leFk6wP6aCk=
+github.com/riverqueue/river/riverdriver v0.30.1 h1:p04cz/Ald1Js/STZ9qYrY5/TBJgjQeVPFltxidFYBBo=
+github.com/riverqueue/river/riverdriver v0.30.1/go.mod h1:WBB9w6LftQtoZgRhNstqhP7MyBKt09XJkzluSNwMMoY=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.30.1 h1:nEStDftvm2jvGlJLliJR+n24PCJsoc4CgGzuop2Yzig=
+github.com/riverqueue/river/riverdriver/riverpgxv5 v0.30.1/go.mod h1:4oSf8jYWZaEwmJ3R5LmOMiGlV9uuvCWOJ3uyBfTwWCc=
+github.com/riverqueue/river/rivershared v0.30.1 h1:ytYlTtMppDV2rJRJ2j55mNf9uQDMPFudOmT4le6/9Ig=
+github.com/riverqueue/river/rivershared v0.30.1/go.mod h1:PfmUHWkF6/fJ1CpjC4cG8eKciBXgMuIHgcRcIuHMc34=
+github.com/riverqueue/river/rivertype v0.30.1 h1:jR7M5UlkA7KRxEbII+LOkD9oQMMz60AEdHh2We1APHY=
+github.com/riverqueue/river/rivertype v0.30.1/go.mod h1:rWpgI59doOWS6zlVocROcwc00fZ1RbzRwsRTU8CDguw=
 github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=
 github.com/rivo/uniseg v0.4.7/go.mod h1:FN3SvrM+Zdj16jyLfmOkMNblXMcoc8DfTHruCPUcx88=
 github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
@@ -437,6 +449,17 @@ github.com/testcontainers/testcontainers-go/modules/postgres v0.37.0 h1:hsVwFkS6
 github.com/testcontainers/testcontainers-go/modules/postgres v0.37.0/go.mod h1:Qj/eGbRbO/rEYdcRLmN+bEojzatP/+NS1y8ojl2PQsc=
 github.com/tetratelabs/wazero v1.8.0 h1:iEKu0d4c2Pd+QSRieYbnQC9yiFlMS9D+Jr0LsRmcF4g=
 github.com/tetratelabs/wazero v1.8.0/go.mod h1:yAI0XTsMBhREkM/YDAK/zNou3GoiAce1P6+rp/wQhjs=
+github.com/tidwall/gjson v1.14.2/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
+github.com/tidwall/gjson v1.18.0 h1:FIDeeyB800efLX89e5a8Y0BNH+LOngJyGrIWxG2FKQY=
+github.com/tidwall/gjson v1.18.0/go.mod h1:/wbyibRr2FHMks5tjHJ5F8dMZh3AcwJEMf5vlfC0lxk=
+github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/match v1.2.0 h1:0pt8FlkOwjN2fPt4bIl4BoNxb98gGHN2ObFEDkrfZnM=
+github.com/tidwall/match v1.2.0/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
+github.com/tidwall/pretty v1.2.1 h1:qjsOFOWWQl+N3RsoF5/ssm1pHmJJwhjlSbZ51I6wMl4=
+github.com/tidwall/pretty v1.2.1/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
+github.com/tidwall/sjson v1.2.5 h1:kLy8mja+1c9jlljvWTlSazM7cKDRfJuR/bOJhcY5NcY=
+github.com/tidwall/sjson v1.2.5/go.mod h1:Fvgq9kS/6ociJEDnK0Fk1cpYF4FIW6ZF7LAe+6jwd28=
 github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU=
 github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
 github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk=
@@ -503,8 +526,8 @@ golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
 golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
-golang.org/x/mod v0.31.0 h1:HaW9xtz0+kOcWKwli0ZXy79Ix+UW/vOfmWI5QVd2tgI=
-golang.org/x/mod v0.31.0/go.mod h1:43JraMp9cGx1Rx3AqioxrbrhNsLl2l/iNAvuBkrezpg=
+golang.org/x/mod v0.32.0 h1:9F4d3PHLljb6x//jOyokMv3eX+YDeepZSEo3mFJy93c=
+golang.org/x/mod v0.32.0/go.mod h1:SgipZ/3h2Ci89DlEtEXWUk/HteuRin+HHhN+WbNhguU=
 golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190213061140-3a22650c66bd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
@@ -563,8 +586,8 @@ golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU=
-golang.org/x/text v0.32.0 h1:ZD01bjUt1FQ9WJ0ClOL5vxgxOI/sVCNgX1YtKwcY0mU=
-golang.org/x/text v0.32.0/go.mod h1:o/rUWzghvpD5TXrTIBuJU77MTaN0ljMWE47kxGJQ7jY=
+golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE=
+golang.org/x/text v0.33.0/go.mod h1:LuMebE6+rBincTi9+xWTY8TztLzKHc/9C1uBCG27+q8=
 golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI=
 golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=

internal/api/handler/auth/api.go

@@ -5,21 +5,25 @@ import (
 	"github.com/compliance-framework/api/internal/config"
 	"github.com/compliance-framework/api/internal/service/email"
 	"github.com/compliance-framework/api/internal/service/sso"
+	"github.com/compliance-framework/api/internal/service/worker"
 	"go.uber.org/zap"
 	"gorm.io/gorm"
 )
 
-func RegisterHandlers(server *api.Server, logger *zap.SugaredLogger, db *gorm.DB, cfg *config.Config, metrics *api.PrometheusMetrics) {
+func RegisterHandlers(server *api.Server, logger *zap.SugaredLogger, db *gorm.DB, cfg *config.Config, metrics *api.PrometheusMetrics, emailService *email.Service, workerService *worker.Service) {
 	authGroup := server.API().Group("/auth")
 
-	// Initialize email service
-	emailService, err := email.NewService(cfg.Email, logger)
-	if err != nil {
-		logger.Warnw("Failed to initialize email service", "error", err)
-		emailService = nil // Set to nil so handlers can check if it's available
+	// Use provided email service or create a new one
+	var err error
+	if emailService == nil {
+		emailService, err = email.NewService(cfg.Email, logger)
+		if err != nil {
+			logger.Warnw("Failed to initialize email service", "error", err)
+			emailService = nil // Set to nil so handlers can check if it's available
+		}
 	}
 
-	authHandler := NewAuthHandler(logger, db, cfg, metrics, emailService)
+	authHandler := NewAuthHandler(logger, db, cfg, metrics, emailService, workerService)
 	authHandler.Register(authGroup)
 
 	ssoService, err := sso.NewService(cfg.SSO, logger)

internal/api/handler/auth/auth.go

@@ -14,26 +14,29 @@ import (
 	"github.com/compliance-framework/api/internal/service/email"
 	emailtypes "github.com/compliance-framework/api/internal/service/email/types"
 	"github.com/compliance-framework/api/internal/service/relational"
+	"github.com/compliance-framework/api/internal/service/worker"
 	"github.com/labstack/echo/v4"
 	"go.uber.org/zap"
 	"gorm.io/gorm"
 )
 
 type AuthHandler struct {
-	sugar        *zap.SugaredLogger
-	db           *gorm.DB
-	config       *config.Config
-	metrics      *api.PrometheusMetrics
-	emailService *email.Service
+	sugar         *zap.SugaredLogger
+	db            *gorm.DB
+	config        *config.Config
+	metrics       *api.PrometheusMetrics
+	emailService  *email.Service
+	workerService *worker.Service
 }
 
-func NewAuthHandler(logger *zap.SugaredLogger, db *gorm.DB, config *config.Config, metrics *api.PrometheusMetrics, emailService *email.Service) *AuthHandler {
+func NewAuthHandler(logger *zap.SugaredLogger, db *gorm.DB, config *config.Config, metrics *api.PrometheusMetrics, emailService *email.Service, workerService *worker.Service) *AuthHandler {
 	return &AuthHandler{
-		sugar:        logger,
-		db:           db,
-		config:       config,
-		metrics:      metrics,
-		emailService: emailService,
+		sugar:         logger,
+		db:            db,
+		config:        config,
+		metrics:       metrics,
+		emailService:  emailService,
+		workerService: workerService,
 	}
 }
 
@@ -350,13 +353,33 @@ func (h *AuthHandler) ForgotPassword(ctx echo.Context) error {
 		TextBody: textBody,
 	}
 
-	_, err = h.emailService.Send(ctx.Request().Context(), message)
-	if err != nil {
-		h.sugar.Errorw("Failed to send password reset email", "error", err, "email", user.Email)
-		return ctx.JSON(http.StatusInternalServerError, api.NewError(err))
-	}
+	// Enqueue email job instead of sending directly
+	if h.workerService != nil && h.workerService.IsStarted() {
+		args := &worker.SendEmailArgs{
+			From:     h.getDefaultFromAddress(),
+			To:       message.To,
+			Subject:  message.Subject,
+			HTMLBody: message.HTMLBody,
+			TextBody: message.TextBody,
+		}
 
-	h.sugar.Infow("Password reset email sent", "email", user.Email)
+		err = h.workerService.EnqueueSendEmail(ctx.Request().Context(), args)
+		if err != nil {
+			h.sugar.Errorw("Failed to enqueue password reset email", "error", err, "email", user.Email)
+			return ctx.JSON(http.StatusInternalServerError, api.NewError(err))
+		}
+
+		h.sugar.Infow("Password reset email enqueued", "email", user.Email)
+	} else {
+		// Fallback to direct sending if worker is not available
+		_, err = h.emailService.Send(ctx.Request().Context(), message)
+		if err != nil {
+			h.sugar.Errorw("Failed to send password reset email", "error", err, "email", user.Email)
+			return ctx.JSON(http.StatusInternalServerError, api.NewError(err))
+		}
+
+		h.sugar.Infow("Password reset email sent", "email", user.Email)
+	}
 
 	return ctx.JSON(http.StatusOK, handler.GenericDataResponse[string]{
 		Data: "If an account with this email exists, a password reset link has been sent.",
@@ -433,3 +456,11 @@ func (h *AuthHandler) PasswordReset(ctx echo.Context) error {
 		Data: "Password has been reset successfully",
 	})
 }
+
+// getDefaultFromAddress returns the default From address from the email service configuration
+func (h *AuthHandler) getDefaultFromAddress() string {
+	if h.emailService == nil {
+		return ""
+	}
+	return h.emailService.GetDefaultFromAddress()
+}

internal/api/handler/auth/auth_integration_test.go

@@ -51,7 +51,7 @@ func (suite *AuthAPIIntegrationSuite) SetupSuite() {
 	suite.logger = logger.Sugar()
 	metrics := api.NewMetricsHandler(context.Background(), suite.logger)
 	suite.server = api.NewServer(context.Background(), suite.logger, suite.Config, metrics)
-	RegisterHandlers(suite.server, suite.logger, suite.DB, suite.Config, metrics)
+	RegisterHandlers(suite.server, suite.logger, suite.DB, suite.Config, metrics, nil, nil)
 	fmt.Println("Server initialized")
 }
 

internal/api/handler/auth/password_reset_test.go

@@ -48,7 +48,7 @@ func setupTestAuthHandler(t *testing.T) *AuthHandler {
 	metrics := api.NewMetricsHandler(context.TODO(), logger)
 
 	// Create auth handler without email service for testing
-	authHandler := NewAuthHandler(logger, db, cfg, metrics, nil)
+	authHandler := NewAuthHandler(logger, db, cfg, metrics, nil, nil)
 
 	return authHandler
 }