refactor(controlplane,dataplane): migrate jobs and feature flags from postgres to sqlc

api/handlers/organisation.go

@@ -15,7 +15,6 @@ import (
 	"github.com/frain-dev/convoy/api/models"
 	"github.com/frain-dev/convoy/api/policies"
 	"github.com/frain-dev/convoy/auth"
-	"github.com/frain-dev/convoy/database/postgres"
 	"github.com/frain-dev/convoy/datastore"
 	"github.com/frain-dev/convoy/internal/event_deliveries"
 	"github.com/frain-dev/convoy/internal/organisation_members"
@@ -289,7 +288,8 @@ func (h *Handler) GetEarlyAdopterFeatures(w http.ResponseWriter, r *http.Request
 	features := fflag.GetEarlyAdopterFeatures()
 	responseFeatures := make([]models.EarlyAdopterFeature, 0, len(features))
 
-	earlyAdopterFeatures, err := postgres.LoadEarlyAdopterFeaturesByOrg(r.Context(), h.A.DB, org.UID)
+	ffService := h.A.FeatureFlagService
+	earlyAdopterFeatures, err := ffService.LoadEarlyAdopterFeaturesByOrg(r.Context(), org.UID)
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return
@@ -378,7 +378,8 @@ func (h *Handler) updateFeatureFlag(w http.ResponseWriter, r *http.Request, feat
 		feature.EnabledAt = null.TimeFrom(time.Now())
 	}
 
-	err := postgres.UpsertEarlyAdopterFeature(r.Context(), h.A.DB, feature)
+	ffService := h.A.FeatureFlagService
+	err := ffService.UpsertEarlyAdopterFeature(r.Context(), feature)
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return err
@@ -423,7 +424,8 @@ func (h *Handler) GetAllFeatureFlags(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	flags, err := postgres.LoadFeatureFlags(r.Context(), h.A.DB)
+	ffService := h.A.FeatureFlagService
+	flags, err := ffService.LoadFeatureFlags(r.Context())
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return
@@ -482,7 +484,8 @@ func (h *Handler) GetOrganisationOverrides(w http.ResponseWriter, r *http.Reques
 		return
 	}
 
-	overrides, err := postgres.LoadFeatureFlagOverridesByOwner(r.Context(), h.A.DB, "organisation", orgID)
+	ffService := h.A.FeatureFlagService
+	overrides, err := ffService.LoadFeatureFlagOverridesByOwner(r.Context(), "organisation", orgID)
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return
@@ -496,7 +499,7 @@ func (h *Handler) GetOrganisationOverrides(w http.ResponseWriter, r *http.Reques
 
 	enrichedOverrides := make([]OverrideWithKey, 0, len(overrides))
 	for i := range overrides {
-		featureFlag, err := postgres.FetchFeatureFlagByID(r.Context(), h.A.DB, overrides[i].FeatureFlagID)
+		featureFlag, err := ffService.FetchFeatureFlagByID(r.Context(), overrides[i].FeatureFlagID)
 		if err != nil {
 			h.A.Logger.WarnContext(r.Context(), fmt.Sprintf("Failed to fetch feature flag for override: %s: %v", overrides[i].FeatureFlagID, err))
 			continue
@@ -537,10 +540,12 @@ func (h *Handler) UpdateOrganisationOverride(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
+	ffService := h.A.FeatureFlagService
+
 	// Fetch the feature flag
-	featureFlag, err := postgres.FetchFeatureFlagByKey(r.Context(), h.A.DB, overrideRequest.FeatureKey)
+	featureFlag, err := ffService.FetchFeatureFlagByKey(r.Context(), overrideRequest.FeatureKey)
 	if err != nil {
-		if errors.Is(err, postgres.ErrFeatureFlagNotFound) {
+		if errors.Is(err, datastore.ErrFeatureFlagNotFound) {
 			_ = render.Render(w, r, util.NewErrorResponse("Feature flag not found: "+overrideRequest.FeatureKey, http.StatusBadRequest))
 			return
 		}
@@ -566,7 +571,7 @@ func (h *Handler) UpdateOrganisationOverride(w http.ResponseWriter, r *http.Requ
 		override.EnabledAt = null.TimeFrom(time.Now())
 	}
 
-	err = postgres.UpsertFeatureFlagOverride(r.Context(), h.A.DB, override)
+	err = ffService.UpsertFeatureFlagOverride(r.Context(), override)
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return
@@ -594,18 +599,20 @@ func (h *Handler) DeleteOrganisationOverride(w http.ResponseWriter, r *http.Requ
 		return
 	}
 
+	ffService := h.A.FeatureFlagService
+
 	// Fetch the feature flag to get its ID
-	featureFlag, err := postgres.FetchFeatureFlagByKey(r.Context(), h.A.DB, featureKey)
+	featureFlag, err := ffService.FetchFeatureFlagByKey(r.Context(), featureKey)
 	if err != nil {
-		if errors.Is(err, postgres.ErrFeatureFlagNotFound) {
+		if errors.Is(err, datastore.ErrFeatureFlagNotFound) {
 			_ = render.Render(w, r, util.NewErrorResponse("Feature flag not found: "+featureKey, http.StatusBadRequest))
 			return
 		}
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return
 	}
 
-	err = postgres.DeleteFeatureFlagOverride(r.Context(), h.A.DB, "organisation", orgID, featureFlag.UID)
+	err = ffService.DeleteFeatureFlagOverride(r.Context(), "organisation", orgID, featureFlag.UID)
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return
@@ -904,10 +911,12 @@ func (h *Handler) UpdateFeatureFlag(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
+	ffService := h.A.FeatureFlagService
+
 	// Fetch the feature flag
-	featureFlag, err := postgres.FetchFeatureFlagByKey(r.Context(), h.A.DB, featureKey)
+	featureFlag, err := ffService.FetchFeatureFlagByKey(r.Context(), featureKey)
 	if err != nil {
-		if errors.Is(err, postgres.ErrFeatureFlagNotFound) {
+		if errors.Is(err, datastore.ErrFeatureFlagNotFound) {
 			_ = render.Render(w, r, util.NewErrorResponse("Feature flag not found: "+featureKey, http.StatusBadRequest))
 			return
 		}
@@ -917,14 +926,14 @@ func (h *Handler) UpdateFeatureFlag(w http.ResponseWriter, r *http.Request) {
 
 	// Update enabled state if provided
 	if updateRequest.Enabled != nil {
-		err = postgres.UpdateFeatureFlag(r.Context(), h.A.DB, featureFlag.UID, *updateRequest.Enabled)
+		err = ffService.UpdateFeatureFlag(r.Context(), featureFlag.UID, *updateRequest.Enabled)
 		if err != nil {
 			_ = render.Render(w, r, util.NewServiceErrResponse(err))
 			return
 		}
 	}
 
-	updatedFlag, err := postgres.FetchFeatureFlagByID(r.Context(), h.A.DB, featureFlag.UID)
+	updatedFlag, err := ffService.FetchFeatureFlagByID(r.Context(), featureFlag.UID)
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return

api/oauth2_integration_test.go

@@ -25,10 +25,10 @@ import (
 	mcache "github.com/frain-dev/convoy/cache/memory"
 	"github.com/frain-dev/convoy/config"
 	"github.com/frain-dev/convoy/database"
-	"github.com/frain-dev/convoy/database/postgres"
 	"github.com/frain-dev/convoy/datastore"
 	"github.com/frain-dev/convoy/internal/api_keys"
 	"github.com/frain-dev/convoy/internal/endpoints"
+	"github.com/frain-dev/convoy/internal/feature_flags"
 	"github.com/frain-dev/convoy/internal/pkg/fflag"
 	"github.com/frain-dev/convoy/internal/pkg/keys"
 	"github.com/frain-dev/convoy/internal/pkg/metrics"
@@ -589,6 +589,8 @@ func (s *OAuth2IntegrationTestSuite) Test_CreateEndpoint_WithOAuth2ClientAsserti
 func enableOAuth2FeatureFlag(t *testing.T, db database.Database, orgID string) error {
 	t.Helper()
 
+	ffService := feature_flags.New(log.New("convoy", log.LevelError), db)
+
 	// Create or update early adopter feature
 	feature := &datastore.EarlyAdopterFeature{
 		OrganisationID: orgID,
@@ -597,7 +599,7 @@ func enableOAuth2FeatureFlag(t *testing.T, db database.Database, orgID string) e
 		EnabledAt:      null.TimeFrom(time.Now()),
 	}
 
-	return postgres.UpsertEarlyAdopterFeature(context.Background(), db, feature)
+	return ffService.UpsertEarlyAdopterFeature(context.Background(), feature)
 }
 
 func TestOAuth2IntegrationTestSuite(t *testing.T) {

api/oss_login_integration_test.go

@@ -16,10 +16,10 @@ import (
 	"github.com/frain-dev/convoy/api/types"
 	rcache "github.com/frain-dev/convoy/cache/redis"
 	"github.com/frain-dev/convoy/config"
-	"github.com/frain-dev/convoy/database/postgres"
 	"github.com/frain-dev/convoy/datastore"
 	"github.com/frain-dev/convoy/internal/api_keys"
 	"github.com/frain-dev/convoy/internal/configuration"
+	"github.com/frain-dev/convoy/internal/feature_flags"
 	"github.com/frain-dev/convoy/internal/pkg/fflag"
 	rlimiter "github.com/frain-dev/convoy/internal/pkg/limiter/redis"
 	"github.com/frain-dev/convoy/internal/portal_links"
@@ -134,6 +134,7 @@ func (s *OSSLoginIntegrationTestSuite) buildServerWithMockLicenser(t *testing.T,
 
 	noopCache := rcache.NewRedisCacheFromClient(tl.Redis)
 	limiter := rlimiter.NewLimiterFromRedisClient(tl.Redis)
+	ffService := feature_flags.New(tl.Logger, db)
 
 	ah, err := NewApplicationHandler(
 		&types.APIOptions{
@@ -143,8 +144,8 @@ func (s *OSSLoginIntegrationTestSuite) buildServerWithMockLicenser(t *testing.T,
 			Logger:                     tl.Logger,
 			Cache:                      noopCache,
 			FFlag:                      fflag.NewFFlag([]string{string(fflag.Prometheus), string(fflag.FullTextSearch)}),
-			FeatureFlagFetcher:         postgres.NewFeatureFlagFetcher(db),
-			EarlyAdopterFeatureFetcher: postgres.NewEarlyAdopterFeatureFetcher(db),
+			FeatureFlagFetcher:         ffService,
+			EarlyAdopterFeatureFetcher: ffService,
 			Rate:                       limiter,
 			ConfigRepo:                 configuration.New(tl.Logger, db),
 			Licenser:                   licenser,

api/server_suite_test.go

@@ -30,6 +30,7 @@ import (
 	"github.com/frain-dev/convoy/database/postgres"
 	"github.com/frain-dev/convoy/datastore"
 	"github.com/frain-dev/convoy/internal/configuration"
+	"github.com/frain-dev/convoy/internal/feature_flags"
 	"github.com/frain-dev/convoy/internal/pkg/fflag"
 	"github.com/frain-dev/convoy/internal/pkg/keys"
 	noopLicenser "github.com/frain-dev/convoy/internal/pkg/license/noop"
@@ -182,6 +183,7 @@ func buildServer(t *testing.T) *ApplicationHandler {
 
 	noopCache := rcache.NewRedisCacheFromClient(tl.Redis)
 	limiter := rlimiter.NewLimiterFromRedisClient(tl.Redis)
+	ffService := feature_flags.New(tl.Logger, db)
 
 	ah, err := NewApplicationHandler(
 		&types.APIOptions{
@@ -191,8 +193,8 @@ func buildServer(t *testing.T) *ApplicationHandler {
 			Logger:                     tl.Logger,
 			Cache:                      noopCache,
 			FFlag:                      fflag.NewFFlag([]string{string(fflag.Prometheus), string(fflag.FullTextSearch)}),
-			FeatureFlagFetcher:         postgres.NewFeatureFlagFetcher(db),
-			EarlyAdopterFeatureFetcher: postgres.NewEarlyAdopterFeatureFetcher(db),
+			FeatureFlagFetcher:         ffService,
+			EarlyAdopterFeatureFetcher: ffService,
 			Rate:                       limiter,
 			ConfigRepo:                 configuration.New(tl.Logger, db),
 			Licenser:                   noopLicenser.NewLicenser(),

api/types/types.go

@@ -9,6 +9,7 @@ import (
 	"github.com/frain-dev/convoy/config"
 	"github.com/frain-dev/convoy/database"
 	"github.com/frain-dev/convoy/datastore"
+	"github.com/frain-dev/convoy/internal/feature_flags"
 	"github.com/frain-dev/convoy/internal/pkg/billing"
 	"github.com/frain-dev/convoy/internal/pkg/fflag"
 	"github.com/frain-dev/convoy/internal/pkg/license"
@@ -21,6 +22,7 @@ type APIOptions struct {
 	FFlag                      *fflag.FFlag
 	FeatureFlagFetcher         fflag.FeatureFlagFetcher
 	EarlyAdopterFeatureFetcher fflag.EarlyAdopterFeatureFetcher
+	FeatureFlagService         *feature_flags.Service
 	DB                         database.Database
 	Redis                      redis.UniversalClient
 	Queue                      queue.Queuer

cmd/hooks/hooks.go

@@ -103,7 +103,7 @@ func PreRun(app *cli.App, db *postgres.Postgres) func(cmd *cobra.Command, args [
 
 		lo := log.New("convoy", logLevel)
 
-		postgresDB, err := postgres.NewDB(cfg)
+		postgresDB, err := postgres.NewDB(cfg, lo)
 		if err != nil {
 			return errors.New("failed to connect to postgres with err: " + err.Error())
 		}

cmd/main.go

@@ -22,6 +22,7 @@ import (
 	"github.com/frain-dev/convoy/config"
 	"github.com/frain-dev/convoy/database/postgres"
 	"github.com/frain-dev/convoy/internal/pkg/cli"
+	log "github.com/frain-dev/convoy/pkg/logger"
 )
 
 func main() {
@@ -34,6 +35,7 @@ func main() {
 	app := &cli.App{}
 	app.Version = convoy.GetVersion()
 	db := &postgres.Postgres{}
+	app.Logger = log.New("convoy", log.LevelDebug)
 
 	c := cli.NewCli(app)