Support data sources in bundles

internal/controlplane/handlers_datasource.go

@@ -13,6 +13,7 @@ import (
 	"github.com/mindersec/minder/internal/datasources/service"
 	"github.com/mindersec/minder/internal/engine/engcontext"
 	"github.com/mindersec/minder/internal/flags"
+	"github.com/mindersec/minder/internal/util"
 	minderv1 "github.com/mindersec/minder/pkg/api/protobuf/go/minder/v1"
 )
 
@@ -25,6 +26,14 @@ func (s *Server) CreateDataSource(ctx context.Context,
 		return nil, status.Errorf(codes.Unavailable, "DataSources feature is disabled")
 	}
 
+	entityCtx := engcontext.EntityFromContext(ctx)
+	err := entityCtx.ValidateProject(ctx, s.store)
+	if err != nil {
+		return nil, util.UserVisibleError(codes.InvalidArgument, "error in entity context: %v", err)
+	}
+
+	projectID := entityCtx.Project.ID
+
 	// Get the data source from the request
 	dsReq := in.GetDataSource()
 	if dsReq == nil {
@@ -36,7 +45,7 @@ func (s *Server) CreateDataSource(ctx context.Context,
 	}
 
 	// Process the request
-	ret, err := s.dataSourcesService.Create(ctx, uuid.Nil, dsReq, nil)
+	ret, err := s.dataSourcesService.Create(ctx, projectID, uuid.Nil, dsReq, nil)
 	if err != nil {
 		return nil, err
 	}
@@ -156,6 +165,14 @@ func (s *Server) UpdateDataSource(ctx context.Context,
 		return nil, status.Errorf(codes.Unavailable, "DataSources feature is disabled")
 	}
 
+	entityCtx := engcontext.EntityFromContext(ctx)
+	err := entityCtx.ValidateProject(ctx, s.store)
+	if err != nil {
+		return nil, status.Errorf(codes.InvalidArgument, "error in entity context: %v", err)
+	}
+
+	projectID := entityCtx.Project.ID
+
 	// Get the data source from the request
 	dsReq := in.GetDataSource()
 	if dsReq == nil {
@@ -167,7 +184,7 @@ func (s *Server) UpdateDataSource(ctx context.Context,
 	}
 
 	// Process the request
-	ret, err := s.dataSourcesService.Update(ctx, uuid.Nil, dsReq, nil)
+	ret, err := s.dataSourcesService.Update(ctx, projectID, uuid.Nil, dsReq, nil)
 	if err != nil {
 		return nil, err
 	}

internal/controlplane/handlers_datasource_test.go

@@ -37,7 +37,7 @@ func TestCreateDataSource(t *testing.T) {
 			setupMocks: func(dsService *mock_service.MockDataSourcesService, featureClient *flags.FakeClient) {
 				featureClient.Data = map[string]interface{}{"data_sources": true}
 				dsService.EXPECT().
-					Create(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+					Create(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 					Return(&minderv1.DataSource{Name: "test-ds"}, nil)
 			},
 			request: &minderv1.CreateDataSourceRequest{
@@ -417,7 +417,7 @@ func TestUpdateDataSource(t *testing.T) {
 			setupMocks: func(dsService *mock_service.MockDataSourcesService, featureClient *flags.FakeClient, _ *mockdb.MockStore) {
 				featureClient.Data = map[string]interface{}{"data_sources": true}
 				dsService.EXPECT().
-					Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+					Update(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
 					Return(&minderv1.DataSource{Id: dsIDStr, Name: "updated-ds"}, nil)
 			},
 			request: &minderv1.UpdateDataSourceRequest{

internal/controlplane/handlers_user.go

@@ -114,7 +114,7 @@ func (s *Server) CreateUser(ctx context.Context,
 	}, nil
 }
 
-func (s *Server) claimGitHubInstalls(ctx context.Context, qtx db.Querier) []*db.Project {
+func (s *Server) claimGitHubInstalls(ctx context.Context, qtx db.ExtendQuerier) []*db.Project {
 	ghId, ok := jwt.GetUserClaimFromContext[string](ctx, "gh_id")
 	if !ok || ghId == "" {
 		return nil
@@ -484,7 +484,7 @@ func (s *Server) acceptInvitation(ctx context.Context, userInvite db.GetInvitati
 	return nil
 }
 
-func ensureUser(ctx context.Context, s *Server, store db.Querier) (db.User, error) {
+func ensureUser(ctx context.Context, s *Server, store db.ExtendQuerier) (db.User, error) {
 	sub := jwt.GetUserSubjectFromContext(ctx)
 	if sub == "" {
 		return db.User{}, status.Error(codes.Internal, "failed to get user subject")

internal/datasources/service/mock/fixtures/service.go

@@ -75,3 +75,9 @@ func WithFailedGetByName() func(DataSourcesSvcMock) {
 			Return(nil, errDefault)
 	}
 }
+
+func WithSuccessfulUpsertDataSource(mock DataSourcesSvcMock) {
+	mock.EXPECT().
+		Upsert(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).
+		Return(nil)
+}

internal/datasources/service/service.go

@@ -25,6 +25,11 @@ import (
 
 //go:generate go run go.uber.org/mock/mockgen -package mock_$GOPACKAGE -destination=./mock/$GOFILE -source=./$GOFILE
 
+var (
+	// ErrDataSourceAlreadyExists is returned when a data source already exists
+	ErrDataSourceAlreadyExists = util.UserVisibleError(codes.AlreadyExists, "data source already exists")
+)
+
 // DataSourcesService is an interface that defines the methods for the data sources service.
 type DataSourcesService interface {
 	// GetByName returns a data source by name.
@@ -37,10 +42,26 @@ type DataSourcesService interface {
 	List(ctx context.Context, project uuid.UUID, opts *ReadOptions) ([]*minderv1.DataSource, error)
 
 	// Create creates a new data source.
-	Create(ctx context.Context, subscriptionID uuid.UUID, ds *minderv1.DataSource, opts *Options) (*minderv1.DataSource, error)
+	Create(
+		ctx context.Context,
+		projectID uuid.UUID,
+		subscriptionID uuid.UUID,
+		ds *minderv1.DataSource,
+		opts *Options,
+	) (*minderv1.DataSource, error)
 
 	// Update updates an existing data source.
-	Update(ctx context.Context, subscriptionID uuid.UUID, ds *minderv1.DataSource, opts *Options) (*minderv1.DataSource, error)
+	Update(
+		ctx context.Context,
+		projectID uuid.UUID,
+		subscriptionID uuid.UUID,
+		ds *minderv1.DataSource,
+		opts *Options,
+	) (*minderv1.DataSource, error)
+
+	// Upsert creates a new data source if it does not exist or updates it if it already exists.
+	// This is used in the subscription logic.
+	Upsert(ctx context.Context, projectID uuid.UUID, subscriptionID uuid.UUID, ds *minderv1.DataSource, opts *Options) error
 
 	// Delete deletes a data source in the given project.
 	//
@@ -150,7 +171,12 @@ func (d *dataSourceService) List(
 // We first validate the data source name uniqueness, then create the data source record.
 // Finally, we create function records based on the driver type.
 func (d *dataSourceService) Create(
-	ctx context.Context, subscriptionID uuid.UUID, ds *minderv1.DataSource, opts *Options) (*minderv1.DataSource, error) {
+	ctx context.Context,
+	projectID uuid.UUID,
+	subscriptionID uuid.UUID,
+	ds *minderv1.DataSource,
+	opts *Options,
+) (*minderv1.DataSource, error) {
 	if err := ds.Validate(); err != nil {
 		return nil, fmt.Errorf("data source validation failed: %w", err)
 	}
@@ -173,11 +199,6 @@ func (d *dataSourceService) Create(
 
 	tx := stx.Q()
 
-	projectID, err := uuid.Parse(ds.GetContext().GetProjectId())
-	if err != nil {
-		return nil, fmt.Errorf("invalid project ID: %w", err)
-	}
-
 	// Check if such data source already exists in project hierarchy
 	projs, err := listRelevantProjects(ctx, tx, projectID, true)
 	if err != nil {
@@ -191,8 +212,7 @@ func (d *dataSourceService) Create(
 		return nil, fmt.Errorf("failed to check for existing data source: %w", err)
 	}
 	if existing.ID != uuid.Nil {
-		return nil, util.UserVisibleError(codes.AlreadyExists,
-			"data source with name %s already exists", ds.GetName())
+		return nil, ErrDataSourceAlreadyExists
 	}
 
 	// Create data source record
@@ -226,7 +246,12 @@ func (d *dataSourceService) Create(
 // because it's simpler and safer - it ensures consistency and avoids partial updates.
 // All functions must use the same driver type to maintain data source integrity.
 func (d *dataSourceService) Update(
-	ctx context.Context, subscriptionID uuid.UUID, ds *minderv1.DataSource, opts *Options) (*minderv1.DataSource, error) {
+	ctx context.Context,
+	projectID uuid.UUID,
+	subscriptionID uuid.UUID,
+	ds *minderv1.DataSource,
+	opts *Options,
+) (*minderv1.DataSource, error) {
 	if err := ds.Validate(); err != nil {
 		return nil, fmt.Errorf("data source validation failed: %w", err)
 	}
@@ -245,11 +270,6 @@ func (d *dataSourceService) Update(
 
 	tx := stx.Q()
 
-	projectID, err := uuid.Parse(ds.GetContext().GetProjectId())
-	if err != nil {
-		return nil, fmt.Errorf("invalid project ID: %w", err)
-	}
-
 	// Validate the subscription ID if present
 	existingDS, err := getDataSourceFromDb(ctx, projectID, ReadBuilder().WithTransaction(tx), tx,
 		func(ctx context.Context, tx db.ExtendQuerier, projs []uuid.UUID) (db.DataSource, error) {
@@ -301,6 +321,33 @@ func (d *dataSourceService) Update(
 	return ds, nil
 }
 
+// Upsert creates the data source if it does not already exist
+// or updates it if it already exists. This is used in the subscription
+// logic.
+func (d *dataSourceService) Upsert(
+	ctx context.Context,
+	projectID uuid.UUID,
+	subscriptionID uuid.UUID,
+	ds *minderv1.DataSource,
+	opts *Options,
+) error {
+	// Simulate upsert semantics by trying to create, then trying to update.
+	_, err := d.Create(ctx, projectID, subscriptionID, ds, opts)
+	if err == nil {
+		// Rule successfully created, we can stop here.
+		return nil
+	} else if !errors.Is(err, ErrDataSourceAlreadyExists) {
+		return fmt.Errorf("error while creating data source: %w", err)
+	}
+
+	// If we get here: data source already exists. Let's update it.
+	_, err = d.Update(ctx, projectID, subscriptionID, ds, opts)
+	if err != nil {
+		return fmt.Errorf("error while updating data source: %w", err)
+	}
+	return nil
+}
+
 // Delete deletes a data source in the given project.
 func (d *dataSourceService) Delete(
 	ctx context.Context, id uuid.UUID, project uuid.UUID, opts *Options) error {