Queue refactor. (#153)

* Queue refactor wip

* still wip

* rework workers

* Queue refactor tests pass

* Update process_message test

* - Enabled Message Retry
- Updated Golden files

* Refactored Implementation

* Clean up

Author: subomi

cmd/main.go

@@ -7,19 +7,21 @@ import (
 	"time"
 	_ "time/tzdata"
 
+	convoyRedis "github.com/frain-dev/convoy/queue/redis"
 	"github.com/getsentry/sentry-go"
 	"github.com/google/uuid"
 	"go.mongodb.org/mongo-driver/bson/primitive"
 
 	"github.com/frain-dev/convoy/util"
+	"github.com/go-redis/redis/v8"
 	log "github.com/sirupsen/logrus"
+	"github.com/vmihailenco/taskq/v3"
 	prefixed "github.com/x-cray/logrus-prefixed-formatter"
 
 	"github.com/frain-dev/convoy"
 	"github.com/frain-dev/convoy/config"
 	"github.com/frain-dev/convoy/datastore"
 	"github.com/frain-dev/convoy/queue"
-	"github.com/frain-dev/convoy/queue/redis"
 	"github.com/spf13/cobra"
 	"go.mongodb.org/mongo-driver/mongo"
 )
@@ -83,10 +85,11 @@ func main() {
 			sentryHook := convoy.NewSentryHook(convoy.DefaultLevels)
 			log.AddHook(sentryHook)
 
-			var queuer queue.Queuer
+			var qFn taskq.Factory
+			var rC *redis.Client
 
 			if cfg.Queue.Type == config.RedisQueueProvider {
-				queuer, err = redis.New(cfg)
+				rC, qFn, err = convoyRedis.NewClient(cfg)
 				if err != nil {
 					return err
 				}
@@ -102,7 +105,8 @@ func main() {
 			app.groupRepo = datastore.NewGroupRepo(conn)
 			app.applicationRepo = datastore.NewApplicationRepo(conn)
 			app.messageRepo = datastore.NewMessageRepository(conn)
-			app.queue = queuer
+			app.scheduleQueue = convoyRedis.NewQueue(rC, qFn, "ScheduleQueue")
+			app.deadLetterQueue = convoyRedis.NewQueue(rC, qFn, "DeadLetterQueue")
 
 			ensureMongoIndices(conn)
 			err = ensureDefaultGroup(context.Background(), app.groupRepo)
@@ -114,7 +118,12 @@ func main() {
 		},
 		PersistentPostRunE: func(cmd *cobra.Command, args []string) error {
 			defer func() {
-				err := app.queue.Close()
+				err := app.scheduleQueue.Close()
+				if err != nil {
+					log.Errorln("failed to close app queue - ", err)
+				}
+
+				err = app.deadLetterQueue.Close()
 				if err != nil {
 					log.Errorln("failed to close app queue - ", err)
 				}
@@ -180,7 +189,8 @@ type app struct {
 	groupRepo       convoy.GroupRepository
 	applicationRepo convoy.ApplicationRepository
 	messageRepo     convoy.MessageRepository
-	queue           queue.Queuer
+	scheduleQueue   queue.Queuer
+	deadLetterQueue queue.Queuer
 }
 
 func getCtx() (context.Context, context.CancelFunc) {

cmd/server.go

@@ -4,10 +4,13 @@ import (
 	"errors"
 	"time"
 
+	"github.com/frain-dev/convoy"
 	"github.com/frain-dev/convoy/config"
+	convoy_queue "github.com/frain-dev/convoy/queue/redis"
 	"github.com/frain-dev/convoy/server"
 	"github.com/frain-dev/convoy/util"
 	"github.com/frain-dev/convoy/worker"
+	convoy_task "github.com/frain-dev/convoy/worker/task"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 )
@@ -36,11 +39,20 @@ func addServerCommand(a *app) *cobra.Command {
 				return errors.New("please provide the HTTP port in the convoy.json file")
 			}
 
-			srv := server.New(cfg, a.messageRepo, a.applicationRepo, a.groupRepo)
+			srv := server.New(cfg, a.messageRepo, a.applicationRepo, a.groupRepo, a.scheduleQueue)
 
-			worker.NewCleaner(a.queue, a.messageRepo).Start()
-			worker.NewScheduler(a.queue, a.messageRepo).Start()
-			worker.NewProducer(a.queue, a.groupRepo, a.applicationRepo, a.messageRepo, cfg.Signature, cfg.SMTP).Start()
+			// register workers.
+			if queue, ok := a.scheduleQueue.(*convoy_queue.RedisQueue); ok {
+				worker.NewProducer(queue).Start()
+			}
+
+			if queue, ok := a.deadLetterQueue.(*convoy_queue.RedisQueue); ok {
+				worker.NewCleaner(queue).Start()
+			}
+
+			// register tasks.
+			convoy_task.CreateTask(convoy.EventProcessor, cfg, convoy_task.ProcessMessages(a.applicationRepo, a.messageRepo, a.groupRepo))
+			convoy_task.CreateTask(convoy.DeadLetterProcessor, cfg, convoy_task.ProcessDeadLetters)
 
 			log.Infof("Started convoy server in %s", time.Since(start))
 			return srv.ListenAndServe()

convoy-docker.json

@@ -1,7 +1,7 @@
 {
   "sentry": {
     "dsn": "<insert-sentry-dsn>"
-  },
+	},
     "database": {
     "environment": "dev",
 		"dsn": "mongodb://root:rootpassword@mongodb:27017"

docs/docs.go

@@ -1,6 +1,6 @@
 // Package docs GENERATED BY THE COMMAND ABOVE; DO NOT EDIT
 // This file was generated by swaggo/swag at
-// 2021-10-15 12:55:36.391586 +0100 WAT m=+25.226284585
+// 2021-10-20 13:55:35.278383 +0100 WAT m=+41.113373043
 package docs
 
 import (
@@ -1028,6 +1028,14 @@ var doc = `{
                     "Group"
                 ],
                 "summary": "Get groups",
+                "parameters": [
+                    {
+                        "type": "string",
+                        "description": "group name",
+                        "name": "name",
+                        "in": "query"
+                    }
+                ],
                 "responses": {
                     "200": {
                         "description": "OK",

docs/swagger.json

@@ -1017,6 +1017,14 @@
                     "Group"
                 ],
                 "summary": "Get groups",
+                "parameters": [
+                    {
+                        "type": "string",
+                        "description": "group name",
+                        "name": "name",
+                        "in": "query"
+                    }
+                ],
                 "responses": {
                     "200": {
                         "description": "OK",

docs/swagger.yaml

@@ -702,6 +702,11 @@ paths:
       consumes:
       - application/json
       description: This endpoint fetches groups
+      parameters:
+      - description: group name
+        in: query
+        name: name
+        type: string
       produces:
       - application/json
       responses:

docs/v3/openapi3.json

@@ -1319,6 +1319,16 @@
 		"/groups": {
 			"get": {
 				"description": "This endpoint fetches groups",
+				"parameters": [
+					{
+						"description": "group name",
+						"in": "query",
+						"name": "name",
+						"schema": {
+							"type": "string"
+						}
+					}
+				],
 				"responses": {
 					"200": {
 						"content": {

docs/v3/openapi3.yaml

@@ -805,6 +805,12 @@ paths:
   /groups:
     get:
       description: This endpoint fetches groups
+      parameters:
+      - description: group name
+        in: query
+        name: name
+        schema:
+          type: string
       responses:
         "200":
           content:

go.mod

@@ -6,17 +6,16 @@ require (
 	github.com/dgrijalva/jwt-go v3.2.0+incompatible
 	github.com/felixge/httpsnoop v1.0.2
 	github.com/getkin/kin-openapi v0.78.0
-	github.com/getsentry/sentry-go v0.11.0 // indirect
+	github.com/getsentry/sentry-go v0.11.0
 	github.com/ghodss/yaml v1.0.0
 	github.com/go-chi/chi/v5 v5.0.3
 	github.com/go-chi/render v1.0.1
-	github.com/go-co-op/gocron v1.7.0
-	github.com/go-redis/redis/v8 v8.8.2
+	github.com/go-redis/redis/v8 v8.11.4
 	github.com/gobeam/mongo-go-pagination v0.0.7
 	github.com/golang/mock v1.4.4
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/uuid v1.2.0
-	github.com/klauspost/compress v1.13.4 // indirect
+	github.com/jarcoal/httpmock v1.0.8
 	github.com/mgutz/ansi v0.0.0-20200706080929-d51e80ef957d // indirect
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/pkg/errors v0.9.1
@@ -27,12 +26,14 @@ require (
 	github.com/spf13/cobra v1.1.3
 	github.com/stretchr/testify v1.7.0
 	github.com/swaggo/swag v1.7.3
+	github.com/vmihailenco/taskq/v3 v3.2.6
 	github.com/x-cray/logrus-prefixed-formatter v0.5.2
 	github.com/youmark/pkcs8 v0.0.0-20201027041543-1326539a0a0a // indirect
 	go.mongodb.org/mongo-driver v1.7.1
 	golang.org/x/crypto v0.0.0-20210813211128-0a44fdfbc16e
-	golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4 // indirect
+	golang.org/x/sync v0.0.0-20210220032951-036812b2e83c // indirect
 	golang.org/x/text v0.3.7 // indirect
 	gopkg.in/alexcesaro/quotedprintable.v3 v3.0.0-20150716171945-2caba252f4dc // indirect
 	gopkg.in/gomail.v2 v2.0.0-20160411212932-81ebce5c23df
+	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b // indirect
 )