feat: 0.0.1

internal/worker/processor.go

@@ -0,0 +1,67 @@
+package worker
+
+import (
+	"context"
+
+	"github.com/FernandoJVideira/simple_bank/internal/config"
+	db "github.com/FernandoJVideira/simple_bank/internal/db/sqlc"
+	"github.com/FernandoJVideira/simple_bank/internal/mailer"
+	"github.com/hibiken/asynq"
+	"github.com/redis/go-redis/v9"
+	"github.com/rs/zerolog/log"
+)
+
+const (
+	CriticalQueue = "critical"
+	DefaultQueue  = "default"
+)
+
+type TaskProcessor interface {
+	Start() error
+	Shutdown()
+	ProcessTaskSendVerifyEmail(ctx context.Context, task *asynq.Task) error
+}
+
+type RedisTaskProcessor struct {
+	server *asynq.Server
+	store  db.Store
+	mailer mailer.MailSender
+	config config.Config
+}
+
+func NewRedisTaskProcessor(redisOpt asynq.RedisClientOpt, store db.Store, mailer mailer.MailSender, config config.Config) TaskProcessor {
+	logger := NewLogger()
+	redis.SetLogger(logger)
+
+	server := asynq.NewServer(
+		redisOpt,
+		asynq.Config{
+			Queues: map[string]int{
+				CriticalQueue: 10,
+				DefaultQueue:  5,
+			},
+			ErrorHandler: asynq.ErrorHandlerFunc(func(ctx context.Context, task *asynq.Task, err error) {
+				log.Error().Err(err).Str("type", task.Type()).Bytes("payload", task.Payload()).Msg("process task failed")
+			}),
+			Logger: NewLogger(),
+		},
+	)
+	return &RedisTaskProcessor{
+		server: server,
+		store:  store,
+		mailer: mailer,
+		config: config,
+	}
+}
+
+func (r *RedisTaskProcessor) Start() error {
+	mux := asynq.NewServeMux()
+
+	mux.HandleFunc(TaskSendVerifyEmail, r.ProcessTaskSendVerifyEmail)
+
+	return r.server.Start(mux)
+}
+
+func (r *RedisTaskProcessor) Shutdown() {
+	r.server.Shutdown()
+}