feat(controlplane): refactor object storage

api/handlers/handlers.go

@@ -15,6 +15,7 @@ import (
 	"github.com/frain-dev/convoy/auth"
 	"github.com/frain-dev/convoy/config"
 	"github.com/frain-dev/convoy/datastore"
+	"github.com/frain-dev/convoy/datastore/cached"
 	"github.com/frain-dev/convoy/internal/organisation_members"
 	"github.com/frain-dev/convoy/internal/organisations"
 	"github.com/frain-dev/convoy/internal/pkg/middleware"
@@ -56,7 +57,7 @@ func (h *Handler) retrieveProject(r *http.Request) (*datastore.Project, error) {
 	var project *datastore.Project
 	var err error
 
-	projectRepo := projects.New(h.A.Logger, h.A.DB)
+	projectRepo := cached.NewCachedProjectRepository(projects.New(h.A.Logger, h.A.DB), h.A.Cache, 5*time.Minute, h.A.Logger)
 
 	switch {
 	case h.IsReqWithJWT(authUser), h.IsReqWithPersonalAccessToken(authUser):
@@ -81,29 +82,10 @@ func (h *Handler) retrieveProject(r *http.Request) (*datastore.Project, error) {
 
 		projectID := apiKey.Role.Project
 
-		var p datastore.Project
-
-		cacheKey := convoy.ProjectCacheKey.Get(projectID)
-		cacheErr := h.A.Cache.Get(r.Context(), cacheKey.String(), &p)
-
-		// If cache hit with valid data, return it
-		if cacheErr == nil && p.UID != "" {
-			h.A.Logger.Info("found item in cache")
-			return &p, nil
-		}
-
-		// Cache miss - fetch from database
-		pp, fetchErr := projectRepo.FetchProjectByID(r.Context(), projectID)
-		if fetchErr != nil {
-			return nil, fetchErr
-		}
-
-		cacheErr = h.A.Cache.Set(r.Context(), cacheKey.String(), &pp, time.Hour)
-		if cacheErr != nil {
-			h.A.Logger.Error("failed to cache item", "error", cacheErr)
+		project, err = projectRepo.FetchProjectByID(r.Context(), projectID)
+		if err != nil {
+			return nil, err
 		}
-
-		return pp, nil
 	case h.IsReqWithPortalLinkToken(authUser):
 		if len(authUser.Credential.Token) > 0 { // this is the legacy static token type
 			svc := portal_links.New(h.A.Logger, h.A.DB)

api/handlers/middleware.go

@@ -4,12 +4,14 @@ import (
 	"context"
 	"errors"
 	"net/http"
+	"time"
 
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/render"
 
 	"github.com/frain-dev/convoy"
 	"github.com/frain-dev/convoy/datastore"
+	"github.com/frain-dev/convoy/datastore/cached"
 	"github.com/frain-dev/convoy/internal/organisations"
 	"github.com/frain-dev/convoy/util"
 )
@@ -61,7 +63,7 @@ func (h *Handler) RequireEnabledOrganisation() func(next http.Handler) http.Hand
 				if cachedOrg := r.Context().Value(convoy.OrganisationCtx); cachedOrg != nil {
 					org = cachedOrg.(*datastore.Organisation)
 				} else {
-					orgRepo := organisations.New(h.A.Logger, h.A.DB)
+					orgRepo := cached.NewCachedOrganisationRepository(organisations.New(h.A.Logger, h.A.DB), h.A.Cache, 5*time.Minute, h.A.Logger)
 					org, err = orgRepo.FetchOrganisationByID(r.Context(), project.OrganisationID)
 					if err != nil {
 						h.A.Logger.Error("Failed to fetch organisation for disabled check", "error", err)

api/models/configuration.go

@@ -47,14 +47,17 @@ type ConfigurationResponse struct {
 }
 
 type StoragePolicyConfiguration struct {
-	// Storage policy type e.g on_prem or s3
+	// Storage policy type e.g on_prem, s3, or azure_blob
 	Type datastore.StorageType `json:"type,omitempty" valid:"supported_storage~please provide a valid storage type,required"`
 
 	// S3 Bucket creds
 	S3 *S3Storage `json:"s3"`
 
 	// On_Prem directory
 	OnPrem *OnPremStorage `json:"on_prem"`
+
+	// Azure Blob Storage creds
+	AzureBlob *AzureBlobStorage `json:"azure_blob"`
 }
 
 func (sc *StoragePolicyConfiguration) Transform() *datastore.StoragePolicyConfiguration {
@@ -63,9 +66,10 @@ func (sc *StoragePolicyConfiguration) Transform() *datastore.StoragePolicyConfig
 	}
 
 	return &datastore.StoragePolicyConfiguration{
-		Type:   sc.Type,
-		S3:     sc.S3.transform(),
-		OnPrem: sc.OnPrem.transform(),
+		Type:      sc.Type,
+		S3:        sc.S3.transform(),
+		OnPrem:    sc.OnPrem.transform(),
+		AzureBlob: sc.AzureBlob.transform(),
 	}
 }
 
@@ -119,3 +123,25 @@ func (os *OnPremStorage) transform() *datastore.OnPremStorage {
 
 	return &datastore.OnPremStorage{Path: os.Path}
 }
+
+type AzureBlobStorage struct {
+	AccountName   null.String `json:"account_name"`
+	AccountKey    null.String `json:"account_key,omitempty"`
+	ContainerName null.String `json:"container_name"`
+	Endpoint      null.String `json:"endpoint,omitempty"`
+	Prefix        null.String `json:"prefix,omitempty"`
+}
+
+func (az *AzureBlobStorage) transform() *datastore.AzureBlobStorage {
+	if az == nil {
+		return nil
+	}
+
+	return &datastore.AzureBlobStorage{
+		AccountName:   az.AccountName,
+		AccountKey:    az.AccountKey,
+		ContainerName: az.ContainerName,
+		Endpoint:      az.Endpoint,
+		Prefix:        az.Prefix,
+	}
+}

cmd/agent/agent.go

@@ -2,6 +2,7 @@ package agent
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"os"
 	"os/signal"
@@ -48,8 +49,8 @@ func AddAgentCommand(a *cli.App) *cobra.Command {
 				return err
 			}
 
-			if err := config.Override(cliConfig); err != nil {
-				return err
+			if oErr := config.Override(cliConfig); oErr != nil {
+				return oErr
 			}
 
 			cfg, err := config.Get()
@@ -71,9 +72,9 @@ func AddAgentCommand(a *cli.App) *cobra.Command {
 			case <-quit:
 				cancel()
 				return nil
-			case err := <-runtimeErr:
-				if err != nil && err != context.Canceled {
-					return err
+			case eRrr := <-runtimeErr:
+				if eRrr != nil && !errors.Is(eRrr, context.Canceled) {
+					return eRrr
 				}
 				return nil
 			case <-ctx.Done():

cmd/hooks/hooks.go

@@ -364,10 +364,19 @@ func ensureInstanceConfig(ctx context.Context, a *cli.App, cfg config.Configurat
 		Path: null.NewString(cfg.StoragePolicy.OnPrem.Path, true),
 	}
 
+	azureBlob := datastore.AzureBlobStorage{
+		AccountName:   null.NewString(cfg.StoragePolicy.AzureBlob.AccountName, true),
+		AccountKey:    null.NewString(cfg.StoragePolicy.AzureBlob.AccountKey, true),
+		ContainerName: null.NewString(cfg.StoragePolicy.AzureBlob.ContainerName, true),
+		Endpoint:      null.NewString(cfg.StoragePolicy.AzureBlob.Endpoint, true),
+		Prefix:        null.NewString(cfg.StoragePolicy.AzureBlob.Prefix, true),
+	}
+
 	storagePolicy := &datastore.StoragePolicyConfiguration{
-		Type:   datastore.StorageType(cfg.StoragePolicy.Type),
-		S3:     &s3,
-		OnPrem: &onPrem,
+		Type:      datastore.StorageType(cfg.StoragePolicy.Type),
+		S3:        &s3,
+		OnPrem:    &onPrem,
+		AzureBlob: &azureBlob,
 	}
 
 	retentionPolicy := &datastore.RetentionPolicyConfiguration{

cmd/server/server.go

@@ -14,9 +14,11 @@ import (
 	"github.com/frain-dev/convoy/auth/realm_chain"
 	"github.com/frain-dev/convoy/config"
 	"github.com/frain-dev/convoy/database/postgres"
+	"github.com/frain-dev/convoy/datastore/cached"
 	"github.com/frain-dev/convoy/internal/api_keys"
 	"github.com/frain-dev/convoy/internal/configuration"
 	"github.com/frain-dev/convoy/internal/pkg/cli"
+	"github.com/frain-dev/convoy/internal/pkg/exporter"
 	"github.com/frain-dev/convoy/internal/pkg/fflag"
 	"github.com/frain-dev/convoy/internal/pkg/keys"
 	"github.com/frain-dev/convoy/internal/pkg/metrics"
@@ -116,9 +118,9 @@ func StartConvoyServer(a *cli.App) error {
 		return err
 	}
 
-	apiKeyRepo := api_keys.New(a.Logger, a.DB)
+	apiKeyRepo := cached.NewCachedAPIKeyRepository(api_keys.New(a.Logger, a.DB), a.Cache, 5*time.Minute, a.Logger)
 	userRepo := users.New(a.Logger, a.DB)
-	portalLinkRepo := portal_links.New(a.Logger, a.DB)
+	portalLinkRepo := cached.NewCachedPortalLinkRepository(portal_links.New(a.Logger, a.DB), a.Cache, 5*time.Minute, a.Logger)
 	configRepo := configuration.New(a.Logger, a.DB)
 	err = realm_chain.Init(&cfg.Auth, apiKeyRepo, userRepo, portalLinkRepo, a.Cache, a.Logger)
 	if err != nil {
@@ -187,13 +189,18 @@ func StartConvoyServer(a *cli.App) error {
 	// 	lo.Infof("Registered metrics materialized view refresh every %d min", refreshInterval)
 	// }
 
-	// ensures that project data is backed up about 2 hours before they are deleted
-	if a.Licenser.RetentionPolicy() {
-		// runs at 10pm
-		s.RegisterTask("0 22 * * *", convoy.ScheduleQueue, convoy.BackupProjectData)
-		// runs at 1am
-		s.RegisterTask("0 1 * * *", convoy.ScheduleQueue, convoy.RetentionPolicies)
+	// Register cron-based backup tasks only when CDC backup is not enabled.
+	// When CDC is active, the BackupCollector in the worker handles exports continuously.
+	if !cfg.RetentionPolicy.CDCBackupEnabled {
+		backupInterval := exporter.ParseBackupInterval(cfg.RetentionPolicy.BackupInterval)
+		enqueueCron := exporter.DurationToCron(backupInterval)
+		processCron := exporter.DurationToCronOffset(backupInterval, 1) // +1 min offset so enqueue runs first
+
+		s.RegisterTask(enqueueCron, convoy.ScheduleQueue, convoy.EnqueueBackupJobs)
+		s.RegisterTask(processCron, convoy.ScheduleQueue, convoy.ProcessBackupJob)
 	}
+	// Retention always runs at 1am
+	s.RegisterTask("0 1 * * *", convoy.ScheduleQueue, convoy.RetentionPolicies)
 
 	err = metrics.RegisterQueueMetrics(a.Queue, a.DB, nil)
 	if err != nil {

config/config.go

@@ -74,6 +74,7 @@ var DefaultConfiguration = Configuration{
 	RetentionPolicy: RetentionPolicyConfiguration{
 		Policy:                   "720h",
 		IsRetentionPolicyEnabled: false,
+		BackupInterval:           "1h",
 	},
 	CircuitBreaker: CircuitBreakerConfiguration{
 		SampleRate:                  30,
@@ -375,6 +376,9 @@ type DatadogConfiguration struct {
 type RetentionPolicyConfiguration struct {
 	Policy                   string `json:"policy" envconfig:"CONVOY_RETENTION_POLICY"`
 	IsRetentionPolicyEnabled bool   `json:"enabled" envconfig:"CONVOY_RETENTION_POLICY_ENABLED"`
+	BackupInterval           string `json:"backup_interval" envconfig:"CONVOY_BACKUP_INTERVAL"`
+	CDCBackupEnabled         bool   `json:"cdc_backup_enabled" envconfig:"CONVOY_CDC_BACKUP_ENABLED"`
+	ReplicationDSN           string `json:"replication_dsn" envconfig:"CONVOY_REPLICATION_DSN"`
 }
 
 type CircuitBreakerConfiguration struct {
@@ -393,9 +397,10 @@ type AnalyticsConfiguration struct {
 }
 
 type StoragePolicyConfiguration struct {
-	Type   string        `json:"type" envconfig:"CONVOY_STORAGE_POLICY_TYPE"`
-	S3     S3Storage     `json:"s3"`
-	OnPrem OnPremStorage `json:"on_prem"`
+	Type      string           `json:"type" envconfig:"CONVOY_STORAGE_POLICY_TYPE"`
+	S3        S3Storage        `json:"s3"`
+	OnPrem    OnPremStorage    `json:"on_prem"`
+	AzureBlob AzureBlobStorage `json:"azure_blob"`
 }
 
 type S3Storage struct {
@@ -412,6 +417,14 @@ type OnPremStorage struct {
 	Path string `json:"path" envconfig:"CONVOY_STORAGE_PREM_PATH"`
 }
 
+type AzureBlobStorage struct {
+	AccountName   string `json:"account_name" envconfig:"CONVOY_STORAGE_AZURE_ACCOUNT_NAME"`
+	AccountKey    string `json:"account_key" envconfig:"CONVOY_STORAGE_AZURE_ACCOUNT_KEY"`
+	ContainerName string `json:"container_name" envconfig:"CONVOY_STORAGE_AZURE_CONTAINER_NAME"`
+	Endpoint      string `json:"endpoint" envconfig:"CONVOY_STORAGE_AZURE_ENDPOINT"`
+	Prefix        string `json:"prefix" envconfig:"CONVOY_STORAGE_AZURE_PREFIX"`
+}
+
 type MetricsConfiguration struct {
 	IsEnabled  bool                           `json:"enabled" envconfig:"CONVOY_METRICS_ENABLED"`
 	Backend    MetricsBackend                 `json:"metrics_backend" envconfig:"CONVOY_METRICS_BACKEND"`

config/config_test.go

@@ -125,6 +125,7 @@ func TestLoadConfig(t *testing.T) {
 				RetentionPolicy: RetentionPolicyConfiguration{
 					Policy:                   "720h",
 					IsRetentionPolicyEnabled: true,
+					BackupInterval:           "1h",
 				},
 				CircuitBreaker: CircuitBreakerConfiguration{
 					SampleRate:                  30,
@@ -208,7 +209,7 @@ func TestLoadConfig(t *testing.T) {
 				APIVersion:       DefaultAPIVersion,
 				Host:             "localhost:5005",
 				ConsumerPoolSize: 100,
-				RetentionPolicy:  RetentionPolicyConfiguration{Policy: "720h"},
+				RetentionPolicy:  RetentionPolicyConfiguration{Policy: "720h", BackupInterval: "1h"},
 				Database: DatabaseConfiguration{
 					Type:               PostgresDatabaseProvider,
 					Scheme:             "postgres",
@@ -307,7 +308,7 @@ func TestLoadConfig(t *testing.T) {
 			wantCfg: Configuration{
 				APIVersion:       DefaultAPIVersion,
 				Host:             "localhost:5005",
-				RetentionPolicy:  RetentionPolicyConfiguration{Policy: "720h"},
+				RetentionPolicy:  RetentionPolicyConfiguration{Policy: "720h", BackupInterval: "1h"},
 				ConsumerPoolSize: 100,
 				CircuitBreaker: CircuitBreakerConfiguration{
 					SampleRate:                  30,

configs/local/conf/.env

@@ -12,10 +12,11 @@ PGBOUNCER_AUTH_TYPE=trust
 PGBOUNCER_USERLIST_FILE=/bitnami/userlists.txt
 PGBOUNCER_DATABASE=${POSTGRESQL_DATABASE}
 PGBOUNCER_AUTH_USER=convoy
-PGBOUNCER_POOL_MODE=session
-PGBOUNCER_MAX_CLIENT_CONN=120
-PGBOUNCER_DEFAULT_POOL_SIZE=20
+PGBOUNCER_POOL_MODE=transaction
+PGBOUNCER_MAX_CLIENT_CONN=500
+PGBOUNCER_DEFAULT_POOL_SIZE=50
 PGBOUNCER_MAX_DB_CONNECTIONS=250
+PGBOUNCER_MAX_PREPARED_STATEMENTS=100
 PGBOUNCER_IGNORE_STARTUP_PARAMETERS=extra_float_digits
 
 # host should be your db host address

configs/local/docker-compose.yml

@@ -61,8 +61,10 @@ services:
                 condition: service_healthy
 
     postgres:
-        image: postgres:15.2-alpine
+        image: postgres:18-alpine
         restart: unless-stopped
+        ports:
+            - "5433:5432"
         environment:
             POSTGRES_DB: convoy
             POSTGRES_USER: convoy
@@ -76,11 +78,17 @@ services:
             timeout: 5s
             retries: 5
             start_period: 10s
+        command:
+            - "postgres"
+            - "-c"
+            - "wal_level=logical"
 
     pgbouncer:
         image: bitnamilegacy/pgbouncer
         hostname: pgbouncer
         restart: unless-stopped
+        ports:
+            - "6432:6432"
         depends_on:
             postgres:
                 condition: service_healthy
@@ -101,8 +109,37 @@ services:
             - ./conf/redis/redis-server-cert.pem:/usr/local/etc/redis/certs/redis-server-cert.pem:ro
             - ./conf/redis/redis-server-key.pem:/usr/local/etc/redis/certs/redis-server-key.pem:ro
 
+    minio:
+        image: minio/minio:RELEASE.2024-01-16T16-07-38Z
+        restart: unless-stopped
+        ports:
+            - "9000:9000"
+            - "9001:9001"
+        environment:
+            MINIO_ROOT_USER: convoyadmin
+            MINIO_ROOT_PASSWORD: convoyadmin
+        command: server /data --console-address ":9001"
+        volumes:
+            - minio_data:/data
+        healthcheck:
+            test: ["CMD", "mc", "ready", "local"]
+            interval: 5s
+            timeout: 5s
+            retries: 5
+
+    azurite:
+        image: mcr.microsoft.com/azure-storage/azurite:3.31.0
+        restart: unless-stopped
+        ports:
+            - "10000:10000"
+        command: ["azurite-blob", "--blobHost", "0.0.0.0", "--blobPort", "10000", "--skipApiVersionCheck"]
+        volumes:
+            - azurite_data:/data
+
 
 volumes:
     postgres_data:
     redis_data:
     caddy_data:
+    minio_data:
+    azurite_data: