feat: implemented an in-memory store for data plane (frain-dev#1932)

* feat: completed memorystore refactor

* fix: removed debug lines

* fix: ingest is working

* fix: use context to synchronise and cancel all running consumers

* fix: fixed loader bug

* fix: fixed linter error: check for err

* fix: use rlock for reading table

* fix: fixed sqs to use ctx

* fix: removed unused variable

* add table tests

Author: subomi

cmd/ingest/ingest.go

@@ -1,11 +1,10 @@
 package ingest
 
 import (
-	"context"
-
 	"github.com/frain-dev/convoy/config"
 	"github.com/frain-dev/convoy/database/postgres"
 	"github.com/frain-dev/convoy/internal/pkg/cli"
+	"github.com/frain-dev/convoy/internal/pkg/memorystore"
 	"github.com/frain-dev/convoy/internal/pkg/pubsub"
 	"github.com/frain-dev/convoy/internal/pkg/server"
 	"github.com/frain-dev/convoy/pkg/log"
@@ -56,13 +55,24 @@ func AddIngestCommand(a *cli.App) *cobra.Command {
 
 			lo.SetLevel(lvl)
 
-			sourcePool := pubsub.NewSourcePool(lo)
-			sourceLoader := pubsub.NewSourceLoader(endpointRepo, sourceRepo, projectRepo, a.Queue, sourcePool, lo)
+			sourceLoader := pubsub.NewSourceLoader(endpointRepo, sourceRepo, projectRepo, lo)
+			sourceTable := memorystore.NewTable(memorystore.OptionSyncer(sourceLoader))
+
+			err = memorystore.DefaultStore.Register("sources", sourceTable)
+			if err != nil {
+				return err
+			}
+
+			go memorystore.DefaultStore.Sync(cmd.Context(), interval)
+
+			ingest, err := pubsub.NewIngest(cmd.Context(), sourceTable, a.Queue, lo)
+			if err != nil {
+				return err
+			}
 
-			stop := make(chan struct{})
-			go sourceLoader.Run(context.Background(), interval, stop)
+			go ingest.Run()
 
-			srv := server.NewServer(cfg.Server.HTTP.IngestPort, func() { stop <- struct{}{} })
+			srv := server.NewServer(cfg.Server.HTTP.IngestPort, func() {})
 			srv.SetHandler(chi.NewMux())
 
 			srv.Listen()