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

@@ -15,6 +15,7 @@ import (
 	ingestSrv "github.com/frain-dev/convoy/cmd/ingest"
 	workerSrv "github.com/frain-dev/convoy/cmd/worker"
 	"github.com/frain-dev/convoy/config"
+	"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"
@@ -150,9 +151,9 @@ func startServerComponent(_ context.Context, a *cli.App) error {
 	lo.Info("Starting Convoy data plane")
 
 	userRepo := users.New(a.Logger, a.DB)
-	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)
 	configRepo := configuration.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)
 	err = realm_chain.Init(&cfg.Auth, apiKeyRepo, userRepo, portalLinkRepo, a.Cache, a.Logger)
 	if err != nil {
 		return fmt.Errorf("failed to initialize realm chain: %w", err)

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)
+		backupCron := exporter.DurationToCron(backupInterval)
+
+		s.RegisterTask(backupCron, convoy.ScheduleQueue, convoy.EnqueueBackupJobs)
+		s.RegisterTask(backupCron, convoy.ScheduleQueue, convoy.ProcessBackupJob)
+		s.RegisterTask(backupCron, convoy.ScheduleQueue, convoy.BackupProjectData)
 	}
+	// Retention always runs at 1am
+	s.RegisterTask("0 1 * * *", convoy.ScheduleQueue, convoy.RetentionPolicies)
 
 	err = metrics.RegisterQueueMetrics(a.Queue, a.DB, nil)
 	if err != nil {

cmd/worker/worker.go

@@ -14,6 +14,8 @@ import (
 	"github.com/frain-dev/convoy/config"
 	"github.com/frain-dev/convoy/database/postgres"
 	"github.com/frain-dev/convoy/datastore"
+	"github.com/frain-dev/convoy/datastore/cached"
+	"github.com/frain-dev/convoy/internal/backup_jobs"
 	batch_retries "github.com/frain-dev/convoy/internal/batch_retries"
 	"github.com/frain-dev/convoy/internal/configuration"
 	"github.com/frain-dev/convoy/internal/delivery_attempts"
@@ -23,8 +25,11 @@ import (
 	"github.com/frain-dev/convoy/internal/filters"
 	"github.com/frain-dev/convoy/internal/meta_events"
 	"github.com/frain-dev/convoy/internal/organisations"
+	backup_collector "github.com/frain-dev/convoy/internal/pkg/backup_collector"
 	"github.com/frain-dev/convoy/internal/pkg/billing"
+	blobstore "github.com/frain-dev/convoy/internal/pkg/blob-store"
 	"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/limiter"
@@ -50,8 +55,9 @@ import (
 )
 
 type Worker struct {
-	consumer *worker.Consumer
-	logger   log.Logger
+	consumer        *worker.Consumer
+	backupCollector *backup_collector.BackupCollector // nil if CDC backup disabled
+	logger          log.Logger
 }
 
 // NewWorker initializes all worker components and returns a Worker instance.
@@ -115,16 +121,17 @@ func NewWorker(ctx context.Context, a *cli.App, cfg config.Configuration) (*Work
 		}
 	}
 
-	projectRepo := projects.New(a.Logger, a.DB)
+	projectRepo := cached.NewCachedProjectRepository(projects.New(a.Logger, a.DB), a.Cache, 5*time.Minute, lo)
 	metaEventRepo := meta_events.New(a.Logger, a.DB)
-	endpointRepo := endpoints.New(a.Logger, a.DB)
+	endpointRepo := cached.NewCachedEndpointRepository(endpoints.New(a.Logger, a.DB), a.Cache, 2*time.Minute, lo)
 	eventRepo := events.New(a.Logger, a.DB)
 	jobRepo := postgres.NewJobRepo(a.DB)
 	eventDeliveryRepo := event_deliveries.New(a.Logger, a.DB)
-	subRepo := subscriptions.New(a.Logger, a.DB)
+	subRepo := cached.NewCachedSubscriptionRepository(subscriptions.New(a.Logger, a.DB), a.Cache, 30*time.Second, lo)
 	configRepo := configuration.New(a.Logger, a.DB)
 	attemptRepo := delivery_attempts.New(a.Logger, a.DB)
-	filterRepo := filters.New(a.Logger, a.DB)
+	backupJobRepo := backup_jobs.New(a.Logger, a.DB)
+	filterRepo := cached.NewCachedFilterRepository(filters.New(a.Logger, a.DB), a.Cache, 2*time.Minute, lo)
 	batchRetryRepo := batch_retries.New(lo, a.DB)
 
 	rd, err := rdb.NewClientFromRedisConfig(cfg.Redis)
@@ -352,6 +359,8 @@ func NewWorker(ctx context.Context, a *cli.App, cfg config.Configuration) (*Work
 	if a.Licenser.RetentionPolicy() {
 		consumer.RegisterHandlers(convoy.RetentionPolicies, task.RetentionPolicies(rd.Client(), ret, lo), nil)
 		consumer.RegisterHandlers(convoy.BackupProjectData, task.BackupProjectData(configRepo, projectRepo, eventRepo, eventDeliveryRepo, attemptRepo, rd.Client(), lo), nil)
+		consumer.RegisterHandlers(convoy.EnqueueBackupJobs, task.EnqueueBackupJobs(configRepo, projectRepo, backupJobRepo, lo), nil)
+		consumer.RegisterHandlers(convoy.ProcessBackupJob, task.ProcessBackupJob(configRepo, projectRepo, eventRepo, eventDeliveryRepo, attemptRepo, backupJobRepo, lo), nil)
 	}
 
 	matchSubscriptionsDeps := task.MatchSubscriptionsDeps{
@@ -417,9 +426,31 @@ func NewWorker(ctx context.Context, a *cli.App, cfg config.Configuration) (*Work
 		return nil, fmt.Errorf("failed to register queue metrics: %w", err)
 	}
 
+	// Optionally start CDC-based backup collector
+	var collector *backup_collector.BackupCollector
+	lo.Info(fmt.Sprintf("CDC backup config: enabled=%v, retention=%v", cfg.RetentionPolicy.CDCBackupEnabled, cfg.RetentionPolicy.IsRetentionPolicyEnabled))
+	if cfg.RetentionPolicy.CDCBackupEnabled && cfg.RetentionPolicy.IsRetentionPolicyEnabled {
+		blobStoreClient, blobErr := blobstore.NewBlobStoreClient(configuration.StoragePolicy, lo)
+		if blobErr != nil {
+			return nil, fmt.Errorf("failed to create blob store for CDC backup: %w", blobErr)
+		}
+
+		flushInterval := exporter.ParseBackupInterval(cfg.RetentionPolicy.BackupInterval)
+
+		// ReplicationDSN connects directly to Postgres (bypassing pgbouncer)
+		// for the WAL replication protocol. Falls back to normal DSN if not set.
+		replDSN := cfg.RetentionPolicy.ReplicationDSN
+		if replDSN == "" {
+			replDSN = cfg.Database.BuildDsn()
+		}
+
+		collector = backup_collector.NewBackupCollector(a.DB.GetConn(), replDSN, blobStoreClient, flushInterval, lo)
+	}
+
 	return &Worker{
-		consumer: consumer,
-		logger:   lo,
+		consumer:        consumer,
+		backupCollector: collector,
+		logger:          lo,
 	}, nil
 }
 
@@ -429,13 +460,26 @@ func (w *Worker) Run(ctx context.Context, workerReady chan struct{}) error {
 	}
 	w.logger.Printf("Starting Convoy Consumer Pool")
 
+	// Start CDC backup collector if enabled
+	if w.backupCollector != nil {
+		if err := w.backupCollector.Start(ctx); err != nil {
+			w.logger.Error(fmt.Sprintf("failed to start backup collector: %v", err))
+			// Non-fatal — worker can still process events without CDC backup
+		}
+	}
+
 	if workerReady != nil {
 		close(workerReady)
 	}
 
 	// Wait for context to be canceled before returning
 	<-ctx.Done()
 	w.logger.Printf("Context canceled, stopping Convoy Consumer Pool...")
+
+	if w.backupCollector != nil {
+		w.backupCollector.Stop(ctx)
+	}
+
 	w.consumer.Stop()
 	w.logger.Printf("Convoy Consumer Pool stopped")
 

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,