feat(controlplane): refactor object storage

api/api.go

@@ -663,6 +663,10 @@ func (a *ApplicationHandler) mountControlPlaneRoutes(router chi.Router, handler
 			configRouter.Get("/auth", handler.GetAuthConfiguration)
 		})
 
+		uiRouter.Route("/backups", func(backupRouter chi.Router) {
+			backupRouter.Post("/trigger", handler.TriggerBackup)
+		})
+
 		billingHandler := &handlers.BillingHandler{
 			Handler:       handler,
 			BillingClient: a.A.BillingClient,

api/handlers/backup.go

@@ -0,0 +1,92 @@
+package handlers
+
+import (
+	"encoding/json"
+	"net/http"
+	"time"
+
+	"github.com/go-chi/render"
+	"github.com/oklog/ulid/v2"
+
+	"github.com/frain-dev/convoy"
+	"github.com/frain-dev/convoy/config"
+	"github.com/frain-dev/convoy/internal/pkg/exporter"
+	"github.com/frain-dev/convoy/queue"
+	"github.com/frain-dev/convoy/util"
+)
+
+type triggerBackupRequest struct {
+	Start *time.Time `json:"start"`
+	End   *time.Time `json:"end"`
+}
+
+type triggerBackupPayload struct {
+	Start time.Time `json:"start"`
+	End   time.Time `json:"end"`
+}
+
+// TriggerBackup enqueues an asynchronous manual backup job.
+// POST /ui/backups/trigger
+func (h *Handler) TriggerBackup(w http.ResponseWriter, r *http.Request) {
+	if !h.isInstanceAdmin(r) {
+		_ = render.Render(w, r, util.NewErrorResponse("Unauthorized: instance admin access required", http.StatusForbidden))
+		return
+	}
+
+	cfg, err := config.Get()
+	if err != nil {
+		_ = render.Render(w, r, util.NewErrorResponse("failed to load config", http.StatusInternalServerError))
+		return
+	}
+
+	if !cfg.RetentionPolicy.IsRetentionPolicyEnabled {
+		_ = render.Render(w, r, util.NewErrorResponse("backup is not enabled in configuration", http.StatusUnprocessableEntity))
+		return
+	}
+
+	// Default time window: last backup interval to now
+	end := time.Now()
+	start := end.Add(-exporter.ParseBackupInterval(cfg.RetentionPolicy.BackupInterval))
+
+	// Parse optional overrides from request body (empty body is fine)
+	var req triggerBackupRequest
+	if r.ContentLength > 0 {
+		if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
+			_ = render.Render(w, r, util.NewErrorResponse("invalid request body", http.StatusBadRequest))
+			return
+		}
+		if req.Start != nil {
+			start = *req.Start
+		}
+		if req.End != nil {
+			end = *req.End
+		}
+	}
+
+	if !start.Before(end) {
+		_ = render.Render(w, r, util.NewErrorResponse("start must be before end", http.StatusBadRequest))
+		return
+	}
+
+	payload, err := json.Marshal(triggerBackupPayload{Start: start, End: end})
+	if err != nil {
+		_ = render.Render(w, r, util.NewErrorResponse("failed to marshal payload", http.StatusInternalServerError))
+		return
+	}
+
+	job := &queue.Job{
+		ID:      ulid.Make().String(),
+		Payload: payload,
+	}
+
+	if err := h.A.Queue.Write(convoy.ManualBackupJob, convoy.DefaultQueue, job); err != nil {
+		_ = render.Render(w, r, util.NewErrorResponse("failed to enqueue backup job", http.StatusInternalServerError))
+		return
+	}
+
+	_ = render.Render(w, r, util.NewServerResponse("backup job enqueued", map[string]interface{}{
+		"job_id": job.ID,
+		"start":  start.Format(time.RFC3339),
+		"end":    end.Format(time.RFC3339),
+	}, http.StatusAccepted))
+}

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/backup/backup.go

@@ -0,0 +1,108 @@
+package backup
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/spf13/cobra"
+
+	"github.com/frain-dev/convoy/config"
+	"github.com/frain-dev/convoy/internal/configuration"
+	"github.com/frain-dev/convoy/internal/delivery_attempts"
+	"github.com/frain-dev/convoy/internal/event_deliveries"
+	"github.com/frain-dev/convoy/internal/events"
+	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"
+)
+
+func AddBackupCommand(a *cli.App) *cobra.Command {
+	var startFlag, endFlag string
+
+	cmd := &cobra.Command{
+		Use:   "backup",
+		Short: "Run a one-time backup of events, deliveries, and delivery attempts",
+		Annotations: map[string]string{
+			"ShouldBootstrap": "false",
+		},
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx := context.Background()
+
+			cfg, err := config.Get()
+			if err != nil {
+				return fmt.Errorf("failed to get config: %w", err)
+			}
+
+			// Default time window: last backup interval to now
+			end := time.Now()
+			start := end.Add(-exporter.ParseBackupInterval(cfg.RetentionPolicy.BackupInterval))
+
+			// Override with flags if provided
+			if endFlag != "" {
+				end, err = time.Parse(time.RFC3339, endFlag)
+				if err != nil {
+					return fmt.Errorf("invalid --end value (expected RFC3339): %w", err)
+				}
+			}
+			if startFlag != "" {
+				start, err = time.Parse(time.RFC3339, startFlag)
+				if err != nil {
+					return fmt.Errorf("invalid --start value (expected RFC3339): %w", err)
+				}
+			}
+
+			if !start.Before(end) {
+				return fmt.Errorf("--start (%s) must be before --end (%s)", start.Format(time.RFC3339), end.Format(time.RFC3339))
+			}
+
+			fmt.Fprintf(os.Stdout, "Backup window: [%s, %s)\n", start.Format(time.RFC3339), end.Format(time.RFC3339))
+
+			// Create repos
+			configRepo := configuration.New(a.Logger, a.DB)
+			eventRepo := events.New(a.Logger, a.DB)
+			eventDeliveryRepo := event_deliveries.New(a.Logger, a.DB)
+			attemptsRepo := delivery_attempts.New(a.Logger, a.DB)
+
+			// Load DB config for storage policy
+			dbConfig, err := configRepo.LoadConfiguration(ctx)
+			if err != nil {
+				return fmt.Errorf("failed to load configuration: %w", err)
+			}
+
+			// Create blob store
+			store, err := blobstore.NewBlobStoreClient(dbConfig.StoragePolicy, a.Logger)
+			if err != nil {
+				return fmt.Errorf("failed to create blob store: %w", err)
+			}
+
+			// Create exporter with explicit window
+			exp, err := exporter.NewExporterWithWindow(
+				eventRepo, eventDeliveryRepo, dbConfig, attemptsRepo,
+				start, end, a.Logger,
+			)
+			if err != nil {
+				return fmt.Errorf("failed to create exporter: %w", err)
+			}
+
+			// Run export
+			result, err := exp.StreamExport(ctx, store)
+			if err != nil {
+				return fmt.Errorf("backup failed: %w", err)
+			}
+
+			// Print results
+			for table, r := range result {
+				fmt.Fprintf(os.Stdout, "%s: %d records exported → %s\n", table, r.NumDocs, r.ExportFile)
+			}
+
+			fmt.Fprintln(os.Stdout, "Backup complete.")
+			return nil
+		},
+	}
+
+	cmd.Flags().StringVar(&startFlag, "start", "", "Export window start (RFC3339, e.g. 2026-04-01T00:00:00Z)")
+	cmd.Flags().StringVar(&endFlag, "end", "", "Export window end (RFC3339, e.g. 2026-04-02T00:00:00Z)")
+	return cmd
+}