fix(controlplane): enforce cache-invalidating project repo and wire bootstrap licenser (#2704)

wire the api layer's ProjectRepo through a single cached repository so
every write path invalidates the projects:<id> cache key. previously
several handlers constructed their own uncached repo, so config updates
via the api were served stale for up to the cache ttl.

also pass the licenser into CreateOrganisationService from the bootstrap
command; it was omitted and caused a nil-pointer panic on convoy
bootstrap, and guard the service against a nil licenser with a clear
error.

Author: mekilis

api/api_options_repositories.go

@@ -2,13 +2,17 @@ package api
 
 import (
 	"github.com/frain-dev/convoy/api/types"
+	"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/projects"
 )
 
 // ensureAPIRepositories wires organisation and project repositories when the caller
 // omitted them but provided DB and Logger (e.g. cmd/server or dataplane startup).
+// The project repository is wrapped in the read-through cache whenever a cache is
+// available so every write path invalidates "projects:<id>" instead of serving
+// stale config until the TTL expires.
 func ensureAPIRepositories(a *types.APIOptions) {
 	if a == nil || a.DB == nil || a.Logger == nil {
 		return
@@ -20,6 +24,11 @@ func ensureAPIRepositories(a *types.APIOptions) {
 		a.OrgMemberRepo = organisation_members.New(a.Logger, a.DB)
 	}
 	if a.ProjectRepo == nil {
-		a.ProjectRepo = projects.New(a.Logger, a.DB)
+		projectRepo := projects.New(a.Logger, a.DB)
+		if a.Cache != nil {
+			a.ProjectRepo = cached.NewCachedProjectRepository(projectRepo, a.Cache, cached.DefaultProjectTTL, a.Logger)
+		} else {
+			a.ProjectRepo = projectRepo
+		}
 	}
 }

api/api_options_repositories_test.go

@@ -8,6 +8,7 @@ import (
 	"go.uber.org/mock/gomock"
 
 	"github.com/frain-dev/convoy/api/types"
+	"github.com/frain-dev/convoy/datastore/cached"
 	"github.com/frain-dev/convoy/mocks"
 	"github.com/frain-dev/convoy/pkg/logger"
 )
@@ -54,6 +55,49 @@ func TestEnsureAPIRepositories_preservesExplicitRepos(t *testing.T) {
 	require.NotNil(t, a.ProjectRepo)
 }
 
+// Regression: project config updates via the API must invalidate the
+// "projects:<id>" cache key. That only holds when the wired ProjectRepo is the
+// cache-invalidating repository, since handlers write through it.
+func TestEnsureAPIRepositories_wrapsProjectRepoInCacheWhenAvailable(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	db := mocks.NewMockDatabase(ctrl)
+	db.EXPECT().GetConn().Return(nil).AnyTimes()
+	db.EXPECT().GetHook().Return(nil).AnyTimes()
+
+	lo := logger.New("test", slog.LevelError)
+
+	a := &types.APIOptions{
+		DB:     db,
+		Logger: lo,
+		Cache:  mocks.NewMockCache(ctrl),
+	}
+
+	ensureAPIRepositories(a)
+
+	require.IsType(t, &cached.CachedProjectRepository{}, a.ProjectRepo,
+		"ProjectRepo must be the cache-invalidating repository when a cache is available")
+}
+
+func TestEnsureAPIRepositories_plainProjectRepoWithoutCache(t *testing.T) {
+	ctrl := gomock.NewController(t)
+	db := mocks.NewMockDatabase(ctrl)
+	db.EXPECT().GetConn().Return(nil).AnyTimes()
+	db.EXPECT().GetHook().Return(nil).AnyTimes()
+
+	lo := logger.New("test", slog.LevelError)
+
+	a := &types.APIOptions{
+		DB:     db,
+		Logger: lo,
+	}
+
+	ensureAPIRepositories(a)
+
+	require.NotNil(t, a.ProjectRepo)
+	_, isCached := a.ProjectRepo.(*cached.CachedProjectRepository)
+	require.False(t, isCached, "no cache available, so the plain repository is wired")
+}
+
 func TestEnsureAPIRepositories_noOpWithoutDB(t *testing.T) {
 	lo := logger.New("test", slog.LevelError)
 	a := &types.APIOptions{Logger: lo}

api/handlers/endpoint.go

@@ -18,7 +18,6 @@ import (
 	"github.com/frain-dev/convoy/internal/event_deliveries"
 	"github.com/frain-dev/convoy/internal/pkg/cbenablement"
 	"github.com/frain-dev/convoy/internal/pkg/middleware"
-	"github.com/frain-dev/convoy/internal/projects"
 	convoynet "github.com/frain-dev/convoy/net"
 	"github.com/frain-dev/convoy/pkg/circuit_breaker"
 	"github.com/frain-dev/convoy/pkg/constants"
@@ -97,7 +96,7 @@ func (h *Handler) CreateEndpoint(w http.ResponseWriter, r *http.Request) {
 
 	ce := services.CreateEndpointService{
 		EndpointRepo:               endpointsvc.New(h.A.Logger, h.A.DB),
-		ProjectRepo:                projects.New(h.A.Logger, h.A.DB),
+		ProjectRepo:                h.projectRepo(),
 		Licenser:                   h.A.Licenser,
 		E:                          e,
 		ProjectID:                  project.UID,
@@ -481,7 +480,7 @@ func (h *Handler) UpdateEndpoint(w http.ResponseWriter, r *http.Request) {
 	ce := services.UpdateEndpointService{
 		Cache:                      h.A.Cache,
 		EndpointRepo:               endpointsvc.New(h.A.Logger, h.A.DB),
-		ProjectRepo:                projects.New(h.A.Logger, h.A.DB),
+		ProjectRepo:                h.projectRepo(),
 		Licenser:                   h.A.Licenser,
 		FeatureFlag:                h.A.FFlag,
 		FeatureFlagFetcher:         h.A.FeatureFlagFetcher,
@@ -612,7 +611,7 @@ func (h *Handler) ExpireSecret(w http.ResponseWriter, r *http.Request) {
 		Queuer:       h.A.Queue,
 		Cache:        h.A.Cache,
 		EndpointRepo: endpointsvc.New(h.A.Logger, h.A.DB),
-		ProjectRepo:  projects.New(h.A.Logger, h.A.DB),
+		ProjectRepo:  h.projectRepo(),
 		S:            e,
 		Endpoint:     endpoint,
 		Project:      project,

api/handlers/handlers.go

@@ -15,7 +15,6 @@ 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/events"
 	"github.com/frain-dev/convoy/internal/organisation_members"
 	"github.com/frain-dev/convoy/internal/organisations"
@@ -41,7 +40,10 @@ func (h *Handler) orgRepo() datastore.OrganisationRepository {
 }
 
 // projectRepo returns the injected project repository, falling back to a freshly
-// constructed one when none was wired (e.g. in tests).
+// constructed one when none was wired (e.g. in tests). In production wiring
+// (ensureAPIRepositories) this is the cached repository, so all project reads
+// share the "projects:<id>" cache and all writes invalidate it. Handlers must
+// use this accessor instead of constructing their own project repository.
 func (h *Handler) projectRepo() datastore.ProjectRepository {
 	if h.A.ProjectRepo != nil {
 		return h.A.ProjectRepo
@@ -95,7 +97,7 @@ func (h *Handler) retrieveProject(r *http.Request) (*datastore.Project, error) {
 	var project *datastore.Project
 	var err error
 
-	projectRepo := cached.NewCachedProjectRepository(projects.New(h.A.Logger, h.A.DB), h.A.Cache, 5*time.Minute, h.A.Logger)
+	projectRepo := h.projectRepo()
 
 	switch {
 	case h.IsReqWithJWT(authUser), h.IsReqWithPersonalAccessToken(authUser):
@@ -205,7 +207,7 @@ func (h *Handler) retrieveOrganisation(r *http.Request) (*datastore.Organisation
 		orgRepo := organisations.New(h.A.Logger, h.A.DB)
 		org, err = orgRepo.FetchOrganisationByID(r.Context(), project.OrganisationID)
 	} else if projectID := chi.URLParam(r, "projectID"); projectID != "" {
-		projectRepo := projects.New(h.A.Logger, h.A.DB)
+		projectRepo := h.projectRepo()
 		var project *datastore.Project
 		project, err = projectRepo.FetchProjectByID(r.Context(), projectID)
 		if err == nil {

api/handlers/organisation.go

@@ -25,7 +25,6 @@ import (
 	"github.com/frain-dev/convoy/internal/pkg/cbenablement"
 	fflag "github.com/frain-dev/convoy/internal/pkg/fflag"
 	m "github.com/frain-dev/convoy/internal/pkg/middleware"
-	"github.com/frain-dev/convoy/internal/projects"
 	"github.com/frain-dev/convoy/services"
 	"github.com/frain-dev/convoy/util"
 	"github.com/frain-dev/convoy/worker/task"
@@ -675,7 +674,7 @@ func (h *Handler) GetOrganisationCircuitBreakerConfig(w http.ResponseWriter, r *
 		return
 	}
 
-	projectRepo := projects.New(h.A.Logger, h.A.DB)
+	projectRepo := h.projectRepo()
 	projects, err := projectRepo.LoadProjects(r.Context(), &datastore.ProjectFilter{OrgID: orgID})
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
@@ -747,7 +746,7 @@ func (h *Handler) UpdateOrganisationCircuitBreakerConfig(w http.ResponseWriter,
 		return
 	}
 
-	projectRepo := projects.New(h.A.Logger, h.A.DB)
+	projectRepo := h.projectRepo()
 	projects, err := projectRepo.LoadProjects(r.Context(), &datastore.ProjectFilter{OrgID: orgID})
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
@@ -818,7 +817,7 @@ func (h *Handler) GetProjectCircuitBreakerConfig(w http.ResponseWriter, r *http.
 		return
 	}
 
-	projectRepo := projects.New(h.A.Logger, h.A.DB)
+	projectRepo := h.projectRepo()
 	project, err := projectRepo.FetchProjectByID(r.Context(), projectID)
 	if err != nil {
 		if errors.Is(err, datastore.ErrProjectNotFound) {
@@ -877,7 +876,7 @@ func (h *Handler) UpdateProjectCircuitBreakerConfig(w http.ResponseWriter, r *ht
 		return
 	}
 
-	projectRepo := projects.New(h.A.Logger, h.A.DB)
+	projectRepo := h.projectRepo()
 	project, err := projectRepo.FetchProjectByID(r.Context(), projectID)
 	if err != nil {
 		if errors.Is(err, datastore.ErrProjectNotFound) {

api/handlers/project.go

@@ -12,7 +12,6 @@ import (
 	"github.com/frain-dev/convoy/internal/event_deliveries"
 	"github.com/frain-dev/convoy/internal/event_types"
 	"github.com/frain-dev/convoy/internal/events"
-	"github.com/frain-dev/convoy/internal/projects"
 	"github.com/frain-dev/convoy/services"
 	"github.com/frain-dev/convoy/util"
 )
@@ -21,7 +20,10 @@ const errBillingRequired = "complete billing setup to create projects: add a sub
 
 func createProjectService(h *Handler) *services.ProjectService {
 	apiKeyRepo := api_keys.New(h.A.Logger, h.A.DB)
-	projectRepo := projects.New(h.A.Logger, h.A.DB)
+	// Must be the cache-invalidating repository: ProjectService.UpdateProject
+	// persists config changes (meta events URL, signature versions, etc.) that
+	// the API and dataplane read through the "projects:<id>" cache.
+	projectRepo := h.projectRepo()
 	eventRepo := events.New(h.A.Logger, h.A.DB)
 	eventDeliveryRepo := event_deliveries.New(h.A.Logger, h.A.DB)
 	eventTypesRepo := event_types.New(h.A.Logger, h.A.DB)
@@ -57,7 +59,7 @@ func (h *Handler) GetProjectStatistics(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	err = projects.New(h.A.Logger, h.A.DB).FillProjectsStatistics(r.Context(), project)
+	err = h.projectRepo().FillProjectsStatistics(r.Context(), project)
 	if err != nil {
 		h.A.Logger.ErrorContext(r.Context(), "failed to count project statistics", "error", err)
 		_ = render.Render(w, r, util.NewErrorResponse("failed to count project statistics", http.StatusBadRequest))
@@ -79,7 +81,7 @@ func (h *Handler) DeleteProject(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	err = projects.New(h.A.Logger, h.A.DB).DeleteProject(r.Context(), project.UID)
+	err = h.projectRepo().DeleteProject(r.Context(), project.UID)
 	if err != nil {
 		h.A.Logger.ErrorContext(r.Context(), "failed to delete project", "error", err)
 		_ = render.Render(w, r, util.NewErrorResponse("failed to delete project", http.StatusBadRequest))
@@ -137,7 +139,7 @@ func (h *Handler) CreateProject(w http.ResponseWriter, r *http.Request) {
 		}
 		limitDeps := services.OrgProjectLimitDeps{
 			BillingClient: h.A.BillingClient,
-			ProjectRepo:   projects.New(h.A.Logger, h.A.DB),
+			ProjectRepo:   h.projectRepo(),
 			Cfg:           h.A.Cfg,
 			Logger:        h.A.Logger,
 		}
@@ -215,7 +217,7 @@ func (h *Handler) GetProjects(w http.ResponseWriter, r *http.Request) {
 	}
 
 	filter := &datastore.ProjectFilter{OrgID: org.UID}
-	projectsList, err := projects.New(h.A.Logger, h.A.DB).LoadProjects(r.Context(), filter)
+	projectsList, err := h.projectRepo().LoadProjects(r.Context(), filter)
 	if err != nil {
 		h.A.Logger.ErrorContext(r.Context(), "failed to load projects", "error", err)
 		_ = render.Render(w, r, util.NewErrorResponse("an error occurred while fetching projects", http.StatusBadRequest))

api/handlers/project_test.go

@@ -8,9 +8,11 @@ import (
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	"go.uber.org/mock/gomock"
 
 	"github.com/frain-dev/convoy/api/types"
 	"github.com/frain-dev/convoy/config"
+	"github.com/frain-dev/convoy/mocks"
 	log "github.com/frain-dev/convoy/pkg/logger"
 )
 
@@ -35,3 +37,30 @@ func TestHandler_CreateProject_InvalidBody_Returns400(t *testing.T) {
 	require.NoError(t, err)
 	require.Contains(t, response["message"], "Invalid request format")
 }
+
+// Regression: ProjectService.UpdateProject persists config the API and
+// dataplane read through the "projects:<id>" cache, so the service must be
+// built on the shared (cache-invalidating) repository from h.projectRepo(),
+// never on a freshly constructed uncached one that skips invalidation.
+func TestCreateProjectService_UsesSharedProjectRepo(t *testing.T) {
+	ctrl := gomock.NewController(t)
+
+	db := mocks.NewMockDatabase(ctrl)
+	db.EXPECT().GetConn().Return(nil).AnyTimes()
+	db.EXPECT().GetHook().Return(nil).AnyTimes()
+
+	projectRepo := mocks.NewMockProjectRepository(ctrl)
+
+	handler := &Handler{
+		A: &types.APIOptions{
+			Logger:      log.New("convoy", log.LevelInfo),
+			DB:          db,
+			ProjectRepo: projectRepo,
+		},
+	}
+
+	svc := createProjectService(handler)
+
+	require.Same(t, projectRepo, svc.ProjectRepo,
+		"createProjectService must reuse the wired project repository")
+}

api/handlers/security.go

@@ -12,7 +12,6 @@ import (
 	"github.com/frain-dev/convoy/datastore"
 	"github.com/frain-dev/convoy/internal/api_keys"
 	m "github.com/frain-dev/convoy/internal/pkg/middleware"
-	"github.com/frain-dev/convoy/internal/projects"
 	"github.com/frain-dev/convoy/internal/users"
 	"github.com/frain-dev/convoy/services"
 	"github.com/frain-dev/convoy/util"
@@ -33,7 +32,7 @@ func (h *Handler) CreatePersonalAPIKey(w http.ResponseWriter, r *http.Request) {
 	}
 
 	cpk := &services.CreatePersonalAPIKeyService{
-		ProjectRepo: projects.New(h.A.Logger, h.A.DB),
+		ProjectRepo: h.projectRepo(),
 		UserRepo:    users.New(h.A.Logger, h.A.DB),
 		APIKeyRepo:  api_keys.New(h.A.Logger, h.A.DB),
 		User:        user,
@@ -75,7 +74,7 @@ func (h *Handler) RevokePersonalAPIKey(w http.ResponseWriter, r *http.Request) {
 	}
 
 	rvk := &services.RevokePersonalAPIKeyService{
-		ProjectRepo: projects.New(h.A.Logger, h.A.DB),
+		ProjectRepo: h.projectRepo(),
 		UserRepo:    users.New(h.A.Logger, h.A.DB),
 		APIKeyRepo:  api_keys.New(h.A.Logger, h.A.DB),
 		UID:         chi.URLParam(r, "keyID"),
@@ -111,7 +110,7 @@ func (h *Handler) RegenerateProjectAPIKey(w http.ResponseWriter, r *http.Request
 	}
 
 	rgp := &services.RegenerateProjectAPIKeyService{
-		ProjectRepo: projects.New(h.A.Logger, h.A.DB),
+		ProjectRepo: h.projectRepo(),
 		UserRepo:    users.New(h.A.Logger, h.A.DB),
 		APIKeyRepo:  api_keys.New(h.A.Logger, h.A.DB),
 		Project:     project,

api/handlers/subscription.go

@@ -12,7 +12,6 @@ import (
 	"github.com/frain-dev/convoy/internal/endpoints"
 	"github.com/frain-dev/convoy/internal/organisations"
 	"github.com/frain-dev/convoy/internal/pkg/middleware"
-	"github.com/frain-dev/convoy/internal/projects"
 	"github.com/frain-dev/convoy/internal/sources"
 	"github.com/frain-dev/convoy/internal/subscriptions"
 	"github.com/frain-dev/convoy/pkg/transform"
@@ -380,7 +379,7 @@ func (h *Handler) UpdateSubscription(w http.ResponseWriter, r *http.Request) {
 	us := services.UpdateSubscriptionService{
 		SubRepo:        subscriptions.New(h.A.Logger, h.A.DB),
 		EndpointRepo:   endpoints.New(h.A.Logger, h.A.DB),
-		ProjectRepo:    projects.New(h.A.Logger, h.A.DB),
+		ProjectRepo:    h.projectRepo(),
 		SourceRepo:     sources.New(h.A.Logger, h.A.DB),
 		Licenser:       h.A.Licenser,
 		ProjectId:      project.UID,

api/ingest.go

@@ -23,7 +23,6 @@ import (
 	"github.com/frain-dev/convoy/internal/pkg/crc"
 	"github.com/frain-dev/convoy/internal/pkg/dedup"
 	"github.com/frain-dev/convoy/internal/pkg/license"
-	"github.com/frain-dev/convoy/internal/projects"
 	"github.com/frain-dev/convoy/internal/sources"
 	"github.com/frain-dev/convoy/pkg/httpheader"
 	"github.com/frain-dev/convoy/pkg/msgpack"
@@ -51,8 +50,8 @@ func (a *ApplicationHandler) IngestEvent(w http.ResponseWriter, r *http.Request)
 	}
 
 	// 2. Retrieve source using mask ID.
-	projectRepo := projects.New(a.A.Logger, a.A.DB)
-	project, err := projectRepo.FetchProjectByID(r.Context(), source.ProjectID)
+	// ProjectRepo is the cached repository wired by ensureAPIRepositories.
+	project, err := a.A.ProjectRepo.FetchProjectByID(r.Context(), source.ProjectID)
 	if err != nil {
 		_ = render.Render(w, r, util.NewServiceErrResponse(err))
 		return