Add In-Memory Queue Support

.github/workflows/go.yml

@@ -13,6 +13,7 @@ jobs:
         go-version: [1.16.x, 1.17.x]
         os: [ubuntu-latest, macos-latest]
         mongodb-version: ["4.0", "4.2", "4.4"]
+        redis-version: ["6.2.6"]
 
     runs-on: ubuntu-latest
     steps:
@@ -21,6 +22,12 @@ jobs:
         with:
           mongodb-version: ${{ matrix.mongodb-version }}
 
+      - name: Start Redis v${{ matrix.redis-version }}
+        uses: supercharge/[email protected]
+        with:
+          redis-version: ${{ matrix.redis-version }}
+          redis-port: 6379
+
       - name: Get the version
         id: get_version
         run: echo ::set-output name=tag::$(echo ${GITHUB_SHA:8})
@@ -49,8 +56,9 @@ jobs:
       - name: Go vet
         run: go vet ./...
 
-      - name: Run Mongo integration tests
+      - name: Run integration tests
         run: go test -tags integration -v ./...
         env:
           TEST_BOLT_DSN: "../../test.db"
           TEST_MONGO_DSN: "mongodb://localhost:27017/testdb"
+          TEST_REDIS_DSN: "redis://localhost:6379"

cmd/main.go

@@ -8,15 +8,16 @@ import (
 	"time"
 	_ "time/tzdata"
 
-	convoyRedis "github.com/frain-dev/convoy/queue/redis"
+	memqueue "github.com/frain-dev/convoy/queue/memqueue"
+	redisqueue "github.com/frain-dev/convoy/queue/redis"
 	"github.com/frain-dev/convoy/worker/task"
 	"github.com/getsentry/sentry-go"
+	"github.com/go-redis/redis/v8"
 	"github.com/google/uuid"
 	prefixed "github.com/x-cray/logrus-prefixed-formatter"
 	"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"
 
@@ -106,12 +107,31 @@ func main() {
 
 			var qFn taskq.Factory
 			var rC *redis.Client
+			var lS queue.Storage
+			var opts queue.QueueOptions
 
 			if cfg.Queue.Type == config.RedisQueueProvider {
-				rC, qFn, err = convoyRedis.NewClient(cfg)
+				rC, qFn, err = redisqueue.NewClient(cfg)
 				if err != nil {
 					return err
 				}
+				opts = queue.QueueOptions{
+					Type:    "redis",
+					Redis:   rC,
+					Factory: qFn,
+				}
+			}
+
+			if cfg.Queue.Type == config.InMemoryQueueProvider {
+				lS, qFn, err = memqueue.NewClient(cfg)
+				if err != nil {
+					return err
+				}
+				opts = queue.QueueOptions{
+					Type:    "in-memory",
+					Storage: lS,
+					Factory: qFn,
+				}
 			}
 
 			if util.IsStringEmpty(string(cfg.GroupConfig.Signature.Header)) {
@@ -125,8 +145,8 @@ func main() {
 			app.applicationRepo = db.AppRepo()
 			app.eventDeliveryRepo = db.EventDeliveryRepo()
 
-			app.eventQueue = convoyRedis.NewQueue(rC, qFn, "EventQueue")
-			app.deadLetterQueue = convoyRedis.NewQueue(rC, qFn, "DeadLetterQueue")
+			app.eventQueue = NewQueue(opts, "EventQueue")
+			app.deadLetterQueue = NewQueue(opts, "DeadLetterQueue")
 
 			err = ensureDefaultGroup(context.Background(), cfg, app)
 			if err != nil {
@@ -173,6 +193,23 @@ func main() {
 	}
 }
 
+func NewQueue(opts queue.QueueOptions, name string) queue.Queuer {
+	optsType := opts.Type
+	var convoyQueue queue.Queuer
+	switch optsType {
+	case "in-memory":
+		opts.Name = name
+		convoyQueue = memqueue.NewQueue(opts)
+
+	case "redis":
+		opts.Name = name
+		convoyQueue = redisqueue.NewQueue(opts)
+	default:
+		log.Errorf("Invalid queue type: %v", optsType)
+	}
+	return convoyQueue
+}
+
 func ensureDefaultGroup(ctx context.Context, cfg config.Configuration, a *app) error {
 	var filter *datastore.GroupFilter
 	var groups []*datastore.Group

cmd/server.go

@@ -8,7 +8,6 @@ import (
 	"github.com/frain-dev/convoy/worker/task"
 
 	"github.com/frain-dev/convoy/config"
-	convoyQueue "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"
@@ -55,12 +54,13 @@ func addServerCommand(a *app) *cobra.Command {
 			}
 
 			// register workers.
-			if queue, ok := a.eventQueue.(*convoyQueue.RedisQueue); ok {
-				worker.NewProducer(queue).Start()
-			}
+			producer := worker.NewProducer(a.eventQueue)
+
+			cleaner := worker.NewCleaner(a.deadLetterQueue)
 
-			if queue, ok := a.deadLetterQueue.(*convoyQueue.RedisQueue); ok {
-				worker.NewCleaner(queue).Start()
+			if cfg.Queue.Type != config.InMemoryQueueProvider {
+				producer.Start()
+				cleaner.Start()
 			}
 
 			log.Infof("Started convoy server in %s", time.Since(start))

config/config.go

@@ -92,6 +92,7 @@ const (
 
 const (
 	RedisQueueProvider      QueueProvider           = "redis"
+	InMemoryQueueProvider   QueueProvider           = "in-memory"
 	DefaultStrategyProvider StrategyProvider        = "default"
 	DefaultSignatureHeader  SignatureHeaderProvider = "X-Convoy-Signature"
 )
@@ -270,6 +271,10 @@ func ensureQueueConfig(queueCfg QueueConfiguration) error {
 		if queueCfg.Redis.DSN == "" {
 			return errors.New("redis queue dsn is empty")
 		}
+
+	case InMemoryQueueProvider:
+		return nil
+
 	default:
 		return fmt.Errorf("unsupported queue type: %s", queueCfg.Type)
 	}

go.mod

@@ -18,6 +18,7 @@ require (
 	github.com/golang/mock v1.6.0
 	github.com/golang/snappy v0.0.4 // indirect
 	github.com/google/uuid v1.3.0
+	github.com/hashicorp/golang-lru v0.5.4
 	github.com/jarcoal/httpmock v1.0.8
 	github.com/kelseyhightower/envconfig v1.4.0
 	github.com/mattn/go-colorable v0.1.11 // indirect

go.sum

@@ -474,6 +474,7 @@ github.com/russross/blackfriday v1.5.2/go.mod h1:JO/DiYxRf+HjHt06OyowR9PTA263kcR
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 github.com/ryanuber/columnize v0.0.0-20160712163229-9b3edd62028f/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
 github.com/ryanuber/columnize v2.1.0+incompatible/go.mod h1:sm1tb6uqfes/u+d4ooFouqFdy9/2g9QGwK3SQygK0Ts=
+github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
 github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
 github.com/schollz/closestmatch v2.1.0+incompatible/go.mod h1:RtP1ddjLong6gTkbtmuhtR2uUrrJOpYzYRvbcPAid+g=
 github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529/go.mod h1:DxrIzT+xaE7yg65j358z/aeFdxmN0P9QXhEzd20vsDc=

queue/memqueue/client.go

@@ -0,0 +1,75 @@
+package memqueue
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"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/vmihailenco/taskq/v3"
+	"github.com/vmihailenco/taskq/v3/memqueue"
+)
+
+type MemQueue struct {
+	Name      string
+	queue     *memqueue.Queue
+	inner     taskq.Factory
+	closeChan chan struct{}
+}
+
+func NewClient(cfg config.Configuration) (queue.Storage, taskq.Factory, error) {
+	if cfg.Queue.Type != config.InMemoryQueueProvider {
+		return nil, nil, errors.New("please select the in-memory queue in your config")
+	}
+
+	qFn := memqueue.NewFactory()
+
+	storage := queue.NewLocalStorage()
+
+	return storage, qFn, nil
+}
+
+func NewQueue(opts queue.QueueOptions) queue.Queuer {
+	q := opts.Factory.RegisterQueue(&taskq.QueueOptions{
+		Name:    opts.Name,
+		Storage: opts.Storage,
+	})
+
+	return &MemQueue{
+		Name:  opts.Name,
+		inner: opts.Factory,
+		queue: q.(*memqueue.Queue),
+	}
+}
+
+func (q *MemQueue) Close() error {
+	q.closeChan <- struct{}{}
+	return q.inner.Close()
+}
+
+func (q *MemQueue) Write(ctx context.Context, name convoy.TaskName, e *datastore.EventDelivery, delay time.Duration) error {
+	job := &queue.Job{
+		ID: e.UID,
+	}
+
+	m := &taskq.Message{
+		Ctx:      ctx,
+		TaskName: string(name),
+		Args:     []interface{}{job},
+		Delay:    delay,
+	}
+
+	err := q.queue.Add(m)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (q *MemQueue) Consumer() taskq.QueueConsumer {
+	return q.queue.Consumer()
+}

queue/memqueue/client_test.go

@@ -0,0 +1,88 @@
+//go:build integration
+// +build integration
+
+package memqueue
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"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/google/uuid"
+	"github.com/vmihailenco/taskq/v3"
+)
+
+func TestMemqueueQueue(t *testing.T) {
+	for scenario, fn := range map[string]func(t *testing.T){
+		"memqueue queue write": testWritetoQueue,
+	} {
+		t.Run(scenario, func(t *testing.T) {
+			fn(t)
+		})
+	}
+}
+
+func testWritetoQueue(t *testing.T) {
+	configfile := "../testdata/convoy_memqueue.json"
+
+	appID := uuid.NewString()
+	eventID := uuid.NewString()
+	eventDeliveryID := uuid.NewString()
+
+	eventDelivery := &datastore.EventDelivery{
+		UID: eventDeliveryID,
+		EventMetadata: &datastore.EventMetadata{
+			UID: eventID,
+		},
+		Status: datastore.SuccessEventStatus,
+		AppMetadata: &datastore.AppMetadata{
+			UID: appID,
+		},
+	}
+	taskName := convoy.TaskName(uuid.NewString())
+	configFile := configfile
+	err := config.LoadConfig(configFile, new(config.Configuration))
+	if err != nil {
+		t.Fatalf("Failed to load config file: %v", err)
+	}
+	cfg, err := config.Get()
+	if err != nil {
+		t.Fatalf("Failed to get config")
+
+	}
+
+	var qFn taskq.Factory
+	var lS queue.Storage
+	var opts queue.QueueOptions
+
+	lS, qFn, err = NewClient(cfg)
+	if err != nil {
+		t.Fatalf("Failed to load new client")
+	}
+	opts = queue.QueueOptions{
+		Name:    uuid.NewString(),
+		Type:    "in-memory",
+		Storage: lS,
+		Factory: qFn,
+	}
+
+	eventQueue := NewQueue(opts)
+
+	err = eventQueue.Write(context.TODO(), taskName, eventDelivery, 0)
+	if err != nil {
+		t.Fatalf("Failed to get queue length")
+	}
+	queueLength, err := eventQueue.Consumer().Queue().Len()
+
+	if err != nil {
+		t.Fatalf("Failed to get queue length")
+	}
+	if fmt.Sprint(queueLength) != "1" {
+		t.Fatalf("Length = %q, Want: %v", queueLength, 1)
+
+	}
+}