18553979900

Committed 16 Oct 2025 07:39AM UTC coverage: 57.814% (+0.3%) from 57.553%

Build # 18553979900

Build Type

Pull #5899

github

Committed by

web-flow

Commit Message

Merge 20bdbbb89 into f6fa9e20a

Pull Request Pull Request #5899: Add Provider authentication for DataSources

Run Details

111 of 160 new or added lines in 8 files covered. (69.38%)

1 existing line in 1 file now uncovered.

18837 of 32582 relevant lines covered (57.81%)

37.07 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

76.76

/internal/datasources/service/service.go

// SPDX-FileCopyrightText: Copyright 2024 The Minder Authors
// SPDX-License-Identifier: Apache-2.0

// Package service encodes the business logic for dealing with data sources.
package service

import (
        "context"
        "database/sql"
        "errors"
        "fmt"

        "github.com/google/uuid"
        "github.com/rs/zerolog"
        "google.golang.org/grpc/codes"
        "google.golang.org/protobuf/encoding/protojson"

        "github.com/mindersec/minder/internal/datasources"
        "github.com/mindersec/minder/internal/db"
        "github.com/mindersec/minder/internal/marketplaces/namespaces"
        "github.com/mindersec/minder/internal/util"
        minderv1 "github.com/mindersec/minder/pkg/api/protobuf/go/minder/v1"
        v1datasources "github.com/mindersec/minder/pkg/datasources/v1"
)

//go:generate go run go.uber.org/mock/mockgen -package mock_$GOPACKAGE -destination=./mock/$GOFILE -source=./$GOFILE

// ErrDataSourceAlreadyExists is returned when a data source already exists
var 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.
        GetByName(ctx context.Context, name string, project uuid.UUID, opts *ReadOptions) (*minderv1.DataSource, error)

        // GetByID returns a data source by ID.
        GetByID(ctx context.Context, id uuid.UUID, project uuid.UUID, opts *ReadOptions) (*minderv1.DataSource, error)

        // List lists all data sources in the given project.
        List(ctx context.Context, project uuid.UUID, opts *ReadOptions) ([]*minderv1.DataSource, error)

        // Create creates a new data source.
        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,
                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.
        //
        // Note that one cannot delete a data source that is in use by a rule type.
        Delete(ctx context.Context, id uuid.UUID, project uuid.UUID, opts *Options) error

        // BuildDataSourceRegistry bundles up all data sources referenced in the rule type
        // into a registry.
        BuildDataSourceRegistry(ctx context.Context, rt *minderv1.RuleType, opts *Options) (*v1datasources.DataSourceRegistry, error)
}

type dataSourceService struct {
        store db.Store

        // This is a function that will begin a transaction for the service.
        // We make this a function so that we can mock it in tests.
        txBuilder func(d *dataSourceService, opts txGetter) (serviceTX, error)
}

// NewDataSourceService creates a new data source service.
func NewDataSourceService(store db.Store) *dataSourceService {
        return &dataSourceService{
                store:     store,
                txBuilder: beginTx,
        }
}

// WithTransactionBuilder sets the transaction builder for the data source service.
//
// Note this is mostly just useful for testing.
func (d *dataSourceService) WithTransactionBuilder(txBuilder func(d *dataSourceService, opts txGetter) (serviceTX, error)) {
        d.txBuilder = txBuilder
}

// Ensure that dataSourceService implements DataSourcesService.
var _ DataSourcesService = (*dataSourceService)(nil)

func (d *dataSourceService) GetByName(
        ctx context.Context, name string, project uuid.UUID, opts *ReadOptions,
) (*minderv1.DataSource, error) {
        return d.getDataSourceSomehow(
                ctx, project, opts, func(ctx context.Context, tx db.ExtendQuerier, projs []uuid.UUID,
                ) (db.DataSource, error) {
                        return getByNameQuery(ctx, tx, projs, name)
                })
}

func (d *dataSourceService) GetByID(
        ctx context.Context, id uuid.UUID, project uuid.UUID, opts *ReadOptions,
) (*minderv1.DataSource, error) {
        return d.getDataSourceSomehow(
                ctx, project, opts, func(ctx context.Context, tx db.ExtendQuerier, projs []uuid.UUID,
                ) (db.DataSource, error) {
                        return getByIDQuery(ctx, tx, projs, id)
                })
}

func (d *dataSourceService) List(
        ctx context.Context, project uuid.UUID, opts *ReadOptions,
) ([]*minderv1.DataSource, error) {
        stx, err := d.txBuilder(d, opts)
        if err != nil {
                return nil, fmt.Errorf("failed to start transaction: %w", err)
        }

        defer func() {
                if stx == nil {
                        // already committed
                        return
                }
                if err := stx.Rollback(); err != nil {
                        zerolog.Ctx(ctx).Error().Err(err).Msg("failed to rollback transaction")
                }
        }()

        tx := stx.Q()

        projs, err := listRelevantProjects(ctx, tx, project, opts.canSearchHierarchical())
        if err != nil {
                return nil, fmt.Errorf("failed to list relevant projects: %w", err)
        }

        dss, err := tx.ListDataSources(ctx, projs)
        if err != nil {
                return nil, fmt.Errorf("failed to list data sources: %w", err)
        }

        outDS := make([]*minderv1.DataSource, len(dss))

        for i, ds := range dss {
                dsfuncs, err := tx.ListDataSourceFunctions(ctx, db.ListDataSourceFunctionsParams{
                        DataSourceID: ds.ID,
                        ProjectID:    ds.ProjectID,
                })
                if err != nil {
                        return nil, fmt.Errorf("failed to list data source functions: %w", err)
                }

                dsProtobuf, err := dataSourceDBToProtobuf(ds, dsfuncs)
                if err != nil {
                        return nil, fmt.Errorf("failed to convert data source to protobuf: %w", err)
                }

                outDS[i] = dsProtobuf
        }

        if err := stx.Commit(); err != nil {
                return nil, fmt.Errorf("failed to commit transaction: %w", err)
        }
        stx = nil // Don't try to rollback

        return outDS, nil
}

// Create creates a new data source.
//
// Create handles data source creation by using a transaction to ensure atomicity.
// 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,
        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)
        }

        if err := namespaces.ValidateNamespacedNameRules(ds.GetName(), subscriptionID); err != nil {
                return nil, fmt.Errorf("data source validation failed: %w", err)
        }

        stx, err := d.txBuilder(d, opts)
        if err != nil {
                return nil, fmt.Errorf("failed to start transaction: %w", err)
        }

        defer func() {
                if stx == nil {
                        // already committed
                        return
                }
                if err := stx.Rollback(); err != nil {
                        zerolog.Ctx(ctx).Error().Err(err).Msg("failed to rollback transaction")
                }
        }()

        tx := stx.Q()

        // Check if such data source already exists in project hierarchy
        projs, err := listRelevantProjects(ctx, tx, projectID, true)
        if err != nil {
                return nil, fmt.Errorf("failed to list relevant projects: %w", err)
        }
        existing, err := tx.GetDataSourceByName(ctx, db.GetDataSourceByNameParams{
                Name:     ds.GetName(),
                Projects: projs,
        })
        if err != nil && !errors.Is(err, sql.ErrNoRows) {
                return nil, fmt.Errorf("failed to check for existing data source: %w", err)
        }
        if existing.ID != uuid.Nil {
                return nil, ErrDataSourceAlreadyExists
        }

        // Create data source record
        metadataBytes, err := metadataForDataSource(ds)
        if err != nil {
                return nil, fmt.Errorf("failed to marshal metadata: %w", err)
        }
        dsRecord, err := tx.CreateDataSource(ctx, db.CreateDataSourceParams{
                ProjectID:      projectID,
                Name:           ds.GetName(),
                DisplayName:    ds.GetName(),
                SubscriptionID: uuid.NullUUID{UUID: subscriptionID, Valid: subscriptionID != uuid.Nil},
                Metadata:       metadataBytes,
        })
        if err != nil {
                return nil, fmt.Errorf("failed to create data source: %w", err)
        }

        // Create function records based on driver type
        if err := addDataSourceFunctions(ctx, tx, ds, dsRecord.ID, projectID); err != nil {
                return nil, fmt.Errorf("failed to create data source functions: %w", err)
        }

        if err := stx.Commit(); err != nil {
                return nil, fmt.Errorf("failed to commit transaction: %w", err)
        }
        stx = nil // Don't try to rollback

        ds.Id = dsRecord.ID.String()

        return ds, nil
}

// Update updates an existing data source and its functions.
//
// Update handles data source modifications by using a transaction to ensure atomicity.
// We first validate and verify the data source exists, then update its basic info.
// For functions, we take a "delete and recreate" approach rather than individual updates
// 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,
        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)
        }

        stx, err := d.txBuilder(d, opts)
        if err != nil {
                return nil, fmt.Errorf("failed to start transaction: %w", err)
        }

        defer func() {
                if stx == nil {
                        // already committed
                        return
                }
                if err := stx.Rollback(); err != nil {
                        zerolog.Ctx(ctx).Error().Err(err).Msg("failed to rollback transaction")
                }
        }()

        tx := stx.Q()

        // 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) {
                        return getByNameQuery(ctx, tx, projs, ds.GetName())
                })
        if err != nil {
                return nil, fmt.Errorf("failed to get existing data source from DB: %w", err)
        }
        if err = namespaces.DoesSubscriptionIDMatch(subscriptionID, existingDS.SubscriptionID); err != nil {
                return nil, fmt.Errorf("failed to update data source: %w", err)
        }

        // Validate the data source functions update
        existingFunctions, err := getDataSourceFunctions(ctx, tx, existingDS)
        if err != nil {
                return nil, fmt.Errorf("failed to get existing data source functions: %w", err)
        }
        if err := validateDataSourceFunctionsUpdate(existingDS, existingFunctions, ds); err != nil {
                return nil, err
        }

        metadataBytes, err := metadataForDataSource(ds)
        if err != nil {
                return nil, fmt.Errorf("failed to serialize metadata: %w", err)
        }
        if _, err := tx.UpdateDataSource(ctx, db.UpdateDataSourceParams{
                ID:          existingDS.ID,
                ProjectID:   projectID,
                DisplayName: ds.GetName(),
                Metadata:    metadataBytes,
        }); err != nil {
                return nil, fmt.Errorf("failed to update data source: %w", err)
        }

        if _, err := tx.DeleteDataSourceFunctions(ctx, db.DeleteDataSourceFunctionsParams{
                DataSourceID: existingDS.ID,
                ProjectID:    projectID,
        }); err != nil {
                return nil, fmt.Errorf("failed to delete existing functions: %w", err)
        }

        if err := addDataSourceFunctions(ctx, tx, ds, existingDS.ID, projectID); err != nil {
                return nil, fmt.Errorf("failed to create data source functions: %w", err)
        }

        if err := stx.Commit(); err != nil {
                return nil, fmt.Errorf("failed to commit transaction: %w", err)
        }
        stx = nil // Don't try to rollback

        if ds.Id == "" {
                ds.Id = existingDS.ID.String()
        }

        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 {
        stx, err := d.txBuilder(d, opts)
        if err != nil {
                return fmt.Errorf("failed to start transaction: %w", err)
        }
        defer func() {
                if stx == nil {
                        // already committed
                        return
                }
                if err := stx.Rollback(); err != nil {
                        zerolog.Ctx(ctx).Error().Err(err).Msg("failed to rollback transaction")
                }
        }()

        // Get the transaction querier
        tx := stx.Q()

        // List rule types referencing the data source
        ret, err := tx.ListRuleTypesReferencesByDataSource(ctx, id)
        if err != nil {
                return fmt.Errorf("failed to list rule types referencing data source %s: %w", id, err)
        }

        // Check if the data source is in use by any rule types
        if len(ret) > 0 {
                // Return an error with the rule types that are using the data source
                var existingRefs []string
                for _, r := range ret {
                        existingRefs = append(existingRefs, r.RuleTypeID.String())
                }
                return util.UserVisibleError(codes.FailedPrecondition,
                        "data source %s is in use by the following rule types: %v", id, existingRefs)
        }

        // We don't support the deletion of bundle data sources
        existingDS, err := getDataSourceFromDb(ctx, project, ReadBuilder().WithTransaction(tx), tx,
                func(ctx context.Context, tx db.ExtendQuerier, projs []uuid.UUID) (db.DataSource, error) {
                        return getByIDQuery(ctx, tx, projs, id)
                })
        if err != nil {
                return fmt.Errorf("failed to get data source with id %s: %w", id, err)
        }
        if existingDS.SubscriptionID.Valid {
                return util.UserVisibleError(codes.FailedPrecondition,
                        "data source %s cannot be deleted as it is part of a bundle", id)
        }

        // Delete the data source record
        _, err = tx.DeleteDataSource(ctx, db.DeleteDataSourceParams{
                ID:        id,
                ProjectID: project,
        })
        if err != nil {
                if errors.Is(err, sql.ErrNoRows) {
                        return util.UserVisibleError(codes.NotFound,
                                "data source with id %s not found in project %s", id, project)
                }
                return fmt.Errorf("failed to delete data source with id %s: %w", id, err)
        }

        // Commit the transaction
        if err := stx.Commit(); err != nil {
                return fmt.Errorf("failed to commit transaction: %w", err)
        }
        stx = nil // Don't try to rollback
        return nil
}

// BuildDataSourceRegistry bundles up all data sources referenced in the rule type
// into a registry.
//
// Note that this assumes that the rule type has already been validated.
func (d *dataSourceService) BuildDataSourceRegistry(
        ctx context.Context, rt *minderv1.RuleType, opts *Options,
) (*v1datasources.DataSourceRegistry, error) {
        rawproj := rt.GetContext().GetProject()
        proj, err := uuid.Parse(rawproj)
        if err != nil {
                return nil, fmt.Errorf("failed to parse project UUID: %w", err)
        }

        instantiations := rt.GetDef().GetEval().GetDataSources()
        reg := v1datasources.NewDataSourceRegistry()

        // return early so we don't need to do useless work
        if len(instantiations) == 0 {
                return reg, nil
        }

        stx, err := d.txBuilder(d, opts)
        if err != nil {
                return nil, fmt.Errorf("failed to start transaction: %w", err)
        }

        //nolint:gosec // we'll log this error later.
        defer stx.Rollback()

        tx := stx.Q()

        projectHierarchy, err := tx.GetParentProjects(ctx, proj)
        if err != nil {
                return nil, fmt.Errorf("failed to get project hierarchy: %w", err)
        }

        for _, ref := range instantiations {
                inst, err := d.instantiateDataSource(ctx, ref, projectHierarchy, tx)
                if err != nil {
                        return nil, fmt.Errorf("failed to instantiate data source: %w", err)
                }

                // Get provider from options if available, needed for authenticated data sources
                provider := opts.getProvider()
                impl, err := datasources.BuildFromProtobuf(inst, provider)
                if err != nil {
                        return nil, fmt.Errorf("failed to build data source from protobuf: %w", err)
                }

                if err := reg.RegisterDataSource(getDataSourceReferenceAlias(ref), impl); err != nil {
                        return nil, fmt.Errorf("failed to register data source: %w", err)
                }
        }

        return reg, nil
}

// getDataSourceReferenceAlias gets the alias that the data source will be referred to by
// in the registry.
func getDataSourceReferenceAlias(dsr *minderv1.DataSourceReference) string {
        key := dsr.GetAlias()
        if key == "" {
                return dsr.GetName()
        }
        return key
}

// addDataSourceFunctions adds functions to a data source based on its driver type.
func addDataSourceFunctions(
        ctx context.Context,
        tx db.ExtendQuerier,
        ds *minderv1.DataSource,
        dsID uuid.UUID,
        projectID uuid.UUID,
) error {
        switch drv := ds.GetDriver().(type) {
        case *minderv1.DataSource_Structured:
                for name, def := range drv.Structured.GetDef() {
                        defBytes, err := protojson.Marshal(def)
                        if err != nil {
                                return fmt.Errorf("failed to marshal structured data definition: %w", err)
                        }

                        if _, err := tx.AddDataSourceFunction(ctx, db.AddDataSourceFunctionParams{
                                DataSourceID: dsID,
                                ProjectID:    projectID,
                                Name:         name,
                                Type:         v1datasources.DataSourceDriverStruct,
                                Definition:   defBytes,
                        }); err != nil {
                                return fmt.Errorf("failed to create data source function: %w", err)
                        }
                }
        case *minderv1.DataSource_Rest:
                for name, def := range drv.Rest.GetDef() {
                        defBytes, err := protojson.Marshal(def)
                        if err != nil {
                                return fmt.Errorf("failed to marshal REST definition: %w", err)
                        }

                        if _, err := tx.AddDataSourceFunction(ctx, db.AddDataSourceFunctionParams{
                                DataSourceID: dsID,
                                ProjectID:    projectID,
                                Name:         name,
                                Type:         v1datasources.DataSourceDriverRest,
                                Definition:   defBytes,
                        }); err != nil {
                                return fmt.Errorf("failed to create data source function: %w", err)
                        }
                }
        default:
                return fmt.Errorf("unsupported data source driver type: %T", drv)
        }
        return nil
}

1	// SPDX-FileCopyrightText: Copyright 2024 The Minder Authors
2	// SPDX-License-Identifier: Apache-2.0
3
4	// Package service encodes the business logic for dealing with data sources.
5	package service
6
7	import (
8	"context"
9	"database/sql"
10	"errors"
11	"fmt"
12
13	"github.com/google/uuid"
14	"github.com/rs/zerolog"
15	"google.golang.org/grpc/codes"
16	"google.golang.org/protobuf/encoding/protojson"
17
18	"github.com/mindersec/minder/internal/datasources"
19	"github.com/mindersec/minder/internal/db"
20	"github.com/mindersec/minder/internal/marketplaces/namespaces"
21	"github.com/mindersec/minder/internal/util"
22	minderv1 "github.com/mindersec/minder/pkg/api/protobuf/go/minder/v1"
23	v1datasources "github.com/mindersec/minder/pkg/datasources/v1"
24	)
25
26	//go:generate go run go.uber.org/mock/mockgen -package mock_$GOPACKAGE -destination=./mock/$GOFILE -source=./$GOFILE
27
28	// ErrDataSourceAlreadyExists is returned when a data source already exists
29	var ErrDataSourceAlreadyExists = util.UserVisibleError(codes.AlreadyExists, "data source already exists")
30
31	// DataSourcesService is an interface that defines the methods for the data sources service.
32	type DataSourcesService interface {
33	// GetByName returns a data source by name.
34	GetByName(ctx context.Context, name string, project uuid.UUID, opts ReadOptions) (minderv1.DataSource, error)
35
36	// GetByID returns a data source by ID.
37	GetByID(ctx context.Context, id uuid.UUID, project uuid.UUID, opts ReadOptions) (minderv1.DataSource, error)
38
39	// List lists all data sources in the given project.
40	List(ctx context.Context, project uuid.UUID, opts ReadOptions) ([]minderv1.DataSource, error)
41
42	// Create creates a new data source.
43	Create(
44	ctx context.Context,
45	projectID uuid.UUID,
46	subscriptionID uuid.UUID,
47	ds *minderv1.DataSource,
48	opts *Options,
49	) (*minderv1.DataSource, error)
50
51	// Update updates an existing data source.
52	Update(
53	ctx context.Context,
54	projectID uuid.UUID,
55	subscriptionID uuid.UUID,
56	ds *minderv1.DataSource,
57	opts *Options,
58	) (*minderv1.DataSource, error)
59
60	// Upsert creates a new data source if it does not exist or updates it if it already exists.
61	// This is used in the subscription logic.
62	Upsert(ctx context.Context, projectID uuid.UUID, subscriptionID uuid.UUID, ds minderv1.DataSource, opts Options) error
63
64	// Delete deletes a data source in the given project.
65	//
66	// Note that one cannot delete a data source that is in use by a rule type.
67	Delete(ctx context.Context, id uuid.UUID, project uuid.UUID, opts *Options) error
68
69	// BuildDataSourceRegistry bundles up all data sources referenced in the rule type
70	// into a registry.
71	BuildDataSourceRegistry(ctx context.Context, rt minderv1.RuleType, opts Options) (*v1datasources.DataSourceRegistry, error)
72	}
73
74	type dataSourceService struct {
75	store db.Store
76
77	// This is a function that will begin a transaction for the service.
78	// We make this a function so that we can mock it in tests.
79	txBuilder func(d *dataSourceService, opts txGetter) (serviceTX, error)
80	}
81
82	// NewDataSourceService creates a new data source service.
83	func NewDataSourceService(store db.Store) *dataSourceService {	46✔
84	return &dataSourceService{	46✔
85	store: store,	46✔
86	txBuilder: beginTx,	46✔
87	}	46✔
88	}	46✔
89
90	// WithTransactionBuilder sets the transaction builder for the data source service.
91	//
92	// Note this is mostly just useful for testing.
93	func (d dataSourceService) WithTransactionBuilder(txBuilder func(d dataSourceService, opts txGetter) (serviceTX, error)) {	×
94	d.txBuilder = txBuilder	×
95	}	×
96
97	// Ensure that dataSourceService implements DataSourcesService.
98	var _ DataSourcesService = (*dataSourceService)(nil)
99
100	func (d *dataSourceService) GetByName(
101	ctx context.Context, name string, project uuid.UUID, opts *ReadOptions,
102	) (*minderv1.DataSource, error) {	8✔
103	return d.getDataSourceSomehow(	8✔
104	ctx, project, opts, func(ctx context.Context, tx db.ExtendQuerier, projs []uuid.UUID,	8✔
105	) (db.DataSource, error) {	16✔
106	return getByNameQuery(ctx, tx, projs, name)	8✔
107	})	8✔
108	}
109
110	func (d *dataSourceService) GetByID(
111	ctx context.Context, id uuid.UUID, project uuid.UUID, opts *ReadOptions,
112	) (*minderv1.DataSource, error) {	3✔
113	return d.getDataSourceSomehow(	3✔
114	ctx, project, opts, func(ctx context.Context, tx db.ExtendQuerier, projs []uuid.UUID,	3✔
115	) (db.DataSource, error) {	6✔
116	return getByIDQuery(ctx, tx, projs, id)	3✔
117	})	3✔
118	}
119
120	func (d *dataSourceService) List(
121	ctx context.Context, project uuid.UUID, opts *ReadOptions,
122	) ([]*minderv1.DataSource, error) {	2✔
123	stx, err := d.txBuilder(d, opts)	2✔
124	if err != nil {	2✔
125	return nil, fmt.Errorf("failed to start transaction: %w", err)	×
126	}	×
127
128	defer func() {	4✔
129	if stx == nil {	3✔
130	// already committed	1✔
131	return	1✔
132	}	1✔
133	if err := stx.Rollback(); err != nil {	1✔
134	zerolog.Ctx(ctx).Error().Err(err).Msg("failed to rollback transaction")	×
135	}	×
136	}()
137
138	tx := stx.Q()	2✔
139		2✔
140	projs, err := listRelevantProjects(ctx, tx, project, opts.canSearchHierarchical())	2✔
141	if err != nil {	2✔
142	return nil, fmt.Errorf("failed to list relevant projects: %w", err)	×
143	}	×
144
145	dss, err := tx.ListDataSources(ctx, projs)	2✔
146	if err != nil {	3✔
147	return nil, fmt.Errorf("failed to list data sources: %w", err)	1✔
148	}	1✔
149
150	outDS := make([]*minderv1.DataSource, len(dss))	1✔
151		1✔
152	for i, ds := range dss {	2✔
153	dsfuncs, err := tx.ListDataSourceFunctions(ctx, db.ListDataSourceFunctionsParams{	1✔
154	DataSourceID: ds.ID,	1✔
155	ProjectID: ds.ProjectID,	1✔
156	})	1✔
157	if err != nil {	1✔
158	return nil, fmt.Errorf("failed to list data source functions: %w", err)	×
159	}	×
160
161	dsProtobuf, err := dataSourceDBToProtobuf(ds, dsfuncs)	1✔
162	if err != nil {	1✔
163	return nil, fmt.Errorf("failed to convert data source to protobuf: %w", err)	×
164	}	×
165
166	outDS[i] = dsProtobuf	1✔
167	}
168
169	if err := stx.Commit(); err != nil {	1✔
170	return nil, fmt.Errorf("failed to commit transaction: %w", err)	×
171	}	×
172	stx = nil // Don't try to rollback	1✔
173		1✔
174	return outDS, nil	1✔
175	}
176
177	// Create creates a new data source.
178	//
179	// Create handles data source creation by using a transaction to ensure atomicity.
180	// We first validate the data source name uniqueness, then create the data source record.
181	// Finally, we create function records based on the driver type.
182	func (d *dataSourceService) Create(
183	ctx context.Context,
184	projectID uuid.UUID,
185	subscriptionID uuid.UUID,
186	ds *minderv1.DataSource,
187	opts *Options,
188	) (*minderv1.DataSource, error) {	11✔
189	if err := ds.Validate(); err != nil {	16✔
190	return nil, fmt.Errorf("data source validation failed: %w", err)	5✔
191	}	5✔
192
193	if err := namespaces.ValidateNamespacedNameRules(ds.GetName(), subscriptionID); err != nil {	8✔
194	return nil, fmt.Errorf("data source validation failed: %w", err)	2✔
195	}	2✔
196
197	stx, err := d.txBuilder(d, opts)	4✔
198	if err != nil {	4✔
199	return nil, fmt.Errorf("failed to start transaction: %w", err)	×
200	}	×
201
202	defer func() {	8✔
203	if stx == nil {	6✔
204	// already committed	2✔
205	return	2✔
206	}	2✔
207	if err := stx.Rollback(); err != nil {	2✔
208	zerolog.Ctx(ctx).Error().Err(err).Msg("failed to rollback transaction")	×
209	}	×
210	}()
211
212	tx := stx.Q()	4✔
213		4✔
214	// Check if such data source already exists in project hierarchy	4✔
215	projs, err := listRelevantProjects(ctx, tx, projectID, true)	4✔
216	if err != nil {	4✔
217	return nil, fmt.Errorf("failed to list relevant projects: %w", err)	×
218	}	×
219	existing, err := tx.GetDataSourceByName(ctx, db.GetDataSourceByNameParams{	4✔
220	Name: ds.GetName(),	4✔
221	Projects: projs,	4✔
222	})	4✔
223	if err != nil && !errors.Is(err, sql.ErrNoRows) {	4✔
224	return nil, fmt.Errorf("failed to check for existing data source: %w", err)	×
225	}	×
226	if existing.ID != uuid.Nil {	6✔
227	return nil, ErrDataSourceAlreadyExists	2✔
228	}	2✔
229
230	// Create data source record
231	metadataBytes, err := metadataForDataSource(ds)	2✔
232	if err != nil {	2✔
NEW 233	return nil, fmt.Errorf("failed to marshal metadata: %w", err)	×
NEW 234	}	×
235	dsRecord, err := tx.CreateDataSource(ctx, db.CreateDataSourceParams{	2✔
236	ProjectID: projectID,	2✔
237	Name: ds.GetName(),	2✔
238	DisplayName: ds.GetName(),	2✔
239	SubscriptionID: uuid.NullUUID{UUID: subscriptionID, Valid: subscriptionID != uuid.Nil},	2✔
240	Metadata: metadataBytes,	2✔
241	})	2✔
242	if err != nil {	2✔
243	return nil, fmt.Errorf("failed to create data source: %w", err)	×
244	}	×
245
246	// Create function records based on driver type
247	if err := addDataSourceFunctions(ctx, tx, ds, dsRecord.ID, projectID); err != nil {	2✔
248	return nil, fmt.Errorf("failed to create data source functions: %w", err)	×
249	}	×
250
251	if err := stx.Commit(); err != nil {	2✔
252	return nil, fmt.Errorf("failed to commit transaction: %w", err)	×
253	}	×
254	stx = nil // Don't try to rollback	2✔
255		2✔
256	ds.Id = dsRecord.ID.String()	2✔
257		2✔
258	return ds, nil	2✔
259	}
260
261	// Update updates an existing data source and its functions.
262	//
263	// Update handles data source modifications by using a transaction to ensure atomicity.
264	// We first validate and verify the data source exists, then update its basic info.
265	// For functions, we take a "delete and recreate" approach rather than individual updates
266	// because it's simpler and safer - it ensures consistency and avoids partial updates.
267	// All functions must use the same driver type to maintain data source integrity.
268	func (d *dataSourceService) Update(
269	ctx context.Context,
270	projectID uuid.UUID,
271	subscriptionID uuid.UUID,
272	ds *minderv1.DataSource,
273	opts *Options,
274	) (*minderv1.DataSource, error) {	12✔
275	if err := ds.Validate(); err != nil {	15✔
276	return nil, fmt.Errorf("data source validation failed: %w", err)	3✔
277	}	3✔
278
279	stx, err := d.txBuilder(d, opts)	9✔
280	if err != nil {	9✔
281	return nil, fmt.Errorf("failed to start transaction: %w", err)	×
282	}	×
283
284	defer func() {	18✔
285	if stx == nil {	12✔
286	// already committed	3✔
287	return	3✔
288	}	3✔
289	if err := stx.Rollback(); err != nil {	6✔
290	zerolog.Ctx(ctx).Error().Err(err).Msg("failed to rollback transaction")	×
291	}	×
292	}()
293
294	tx := stx.Q()	9✔
295		9✔
296	// Validate the subscription ID if present	9✔
297	existingDS, err := getDataSourceFromDb(ctx, projectID, ReadBuilder().WithTransaction(tx), tx,	9✔
298	func(ctx context.Context, tx db.ExtendQuerier, projs []uuid.UUID) (db.DataSource, error) {	18✔
299	return getByNameQuery(ctx, tx, projs, ds.GetName())	9✔
300	})	9✔
301	if err != nil {	10✔
302	return nil, fmt.Errorf("failed to get existing data source from DB: %w", err)	1✔
303	}	1✔
304	if err = namespaces.DoesSubscriptionIDMatch(subscriptionID, existingDS.SubscriptionID); err != nil {	9✔
305	return nil, fmt.Errorf("failed to update data source: %w", err)	1✔
306	}	1✔
307
308	// Validate the data source functions update
309	existingFunctions, err := getDataSourceFunctions(ctx, tx, existingDS)	7✔
310	if err != nil {	8✔
311	return nil, fmt.Errorf("failed to get existing data source functions: %w", err)	1✔
312	}	1✔
313	if err := validateDataSourceFunctionsUpdate(existingDS, existingFunctions, ds); err != nil {	7✔
314	return nil, err	1✔
315	}	1✔
316
317	metadataBytes, err := metadataForDataSource(ds)	5✔
318	if err != nil {	5✔
NEW 319	return nil, fmt.Errorf("failed to serialize metadata: %w", err)	×
NEW 320	}	×
321	if _, err := tx.UpdateDataSource(ctx, db.UpdateDataSourceParams{	5✔
322	ID: existingDS.ID,	5✔
323	ProjectID: projectID,	5✔
324	DisplayName: ds.GetName(),	5✔
325	Metadata: metadataBytes,	5✔
326	}); err != nil {	6✔
327	return nil, fmt.Errorf("failed to update data source: %w", err)	1✔
328	}	1✔
329
330	if _, err := tx.DeleteDataSourceFunctions(ctx, db.DeleteDataSourceFunctionsParams{	4✔
331	DataSourceID: existingDS.ID,	4✔
332	ProjectID: projectID,	4✔
333	}); err != nil {	5✔
334	return nil, fmt.Errorf("failed to delete existing functions: %w", err)	1✔
335	}	1✔
336
337	if err := addDataSourceFunctions(ctx, tx, ds, existingDS.ID, projectID); err != nil {	3✔
338	return nil, fmt.Errorf("failed to create data source functions: %w", err)	×
339	}	×
340
341	if err := stx.Commit(); err != nil {	3✔
342	return nil, fmt.Errorf("failed to commit transaction: %w", err)	×
343	}	×
344	stx = nil // Don't try to rollback	3✔
345		3✔
346	if ds.Id == "" {	4✔
347	ds.Id = existingDS.ID.String()	1✔
348	}	1✔
349
350	return ds, nil	3✔
351	}
352
353	// Upsert creates the data source if it does not already exist
354	// or updates it if it already exists. This is used in the subscription
355	// logic.
356	func (d *dataSourceService) Upsert(
357	ctx context.Context,
358	projectID uuid.UUID,
359	subscriptionID uuid.UUID,
360	ds *minderv1.DataSource,
361	opts *Options,
362	) error {	5✔
363	// Simulate upsert semantics by trying to create, then trying to update.	5✔
364	_, err := d.Create(ctx, projectID, subscriptionID, ds, opts)	5✔
365	if err == nil {	6✔
366	// Rule successfully created, we can stop here.	1✔
367	return nil	1✔
368	} else if !errors.Is(err, ErrDataSourceAlreadyExists) {	8✔
369	return fmt.Errorf("error while creating data source: %w", err)	3✔
370	}	3✔
371
372	// If we get here: data source already exists. Let's update it.
373	_, err = d.Update(ctx, projectID, subscriptionID, ds, opts)	1✔
374	if err != nil {	1✔
375	return fmt.Errorf("error while updating data source: %w", err)	×
376	}	×
377	return nil	1✔
378	}
379
380	// Delete deletes a data source in the given project.
381	func (d *dataSourceService) Delete(
382	ctx context.Context, id uuid.UUID, project uuid.UUID, opts *Options,
383	) error {	6✔
384	stx, err := d.txBuilder(d, opts)	6✔
385	if err != nil {	6✔
386	return fmt.Errorf("failed to start transaction: %w", err)	×
387	}	×
388	defer func() {	12✔
389	if stx == nil {	7✔
390	// already committed	1✔
391	return	1✔
392	}	1✔
393	if err := stx.Rollback(); err != nil {	5✔
394	zerolog.Ctx(ctx).Error().Err(err).Msg("failed to rollback transaction")	×
395	}	×
396	}()
397
398	// Get the transaction querier
399	tx := stx.Q()	6✔
400		6✔
401	// List rule types referencing the data source	6✔
402	ret, err := tx.ListRuleTypesReferencesByDataSource(ctx, id)	6✔
403	if err != nil {	7✔
404	return fmt.Errorf("failed to list rule types referencing data source %s: %w", id, err)	1✔
405	}	1✔
406
407	// Check if the data source is in use by any rule types
408	if len(ret) > 0 {	6✔
409	// Return an error with the rule types that are using the data source	1✔
410	var existingRefs []string	1✔
411	for _, r := range ret {	2✔
412	existingRefs = append(existingRefs, r.RuleTypeID.String())	1✔
413	}	1✔
414	return util.UserVisibleError(codes.FailedPrecondition,	1✔
415	"data source %s is in use by the following rule types: %v", id, existingRefs)	1✔
416	}
417
418	// We don't support the deletion of bundle data sources
419	existingDS, err := getDataSourceFromDb(ctx, project, ReadBuilder().WithTransaction(tx), tx,	4✔
420	func(ctx context.Context, tx db.ExtendQuerier, projs []uuid.UUID) (db.DataSource, error) {	8✔
421	return getByIDQuery(ctx, tx, projs, id)	4✔
422	})	4✔
423	if err != nil {	4✔
424	return fmt.Errorf("failed to get data source with id %s: %w", id, err)	×
425	}	×
426	if existingDS.SubscriptionID.Valid {	5✔
427	return util.UserVisibleError(codes.FailedPrecondition,	1✔
428	"data source %s cannot be deleted as it is part of a bundle", id)	1✔
429	}	1✔
430
431	// Delete the data source record
432	_, err = tx.DeleteDataSource(ctx, db.DeleteDataSourceParams{	3✔
433	ID: id,	3✔
434	ProjectID: project,	3✔
435	})	3✔
436	if err != nil {	5✔
437	if errors.Is(err, sql.ErrNoRows) {	3✔
438	return util.UserVisibleError(codes.NotFound,	1✔
439	"data source with id %s not found in project %s", id, project)	1✔
440	}	1✔
441	return fmt.Errorf("failed to delete data source with id %s: %w", id, err)	1✔
442	}
443
444	// Commit the transaction
445	if err := stx.Commit(); err != nil {	1✔
446	return fmt.Errorf("failed to commit transaction: %w", err)	×
447	}	×
448	stx = nil // Don't try to rollback	1✔
449	return nil	1✔
450	}
451
452	// BuildDataSourceRegistry bundles up all data sources referenced in the rule type
453	// into a registry.
454	//
455	// Note that this assumes that the rule type has already been validated.
456	func (d *dataSourceService) BuildDataSourceRegistry(
457	ctx context.Context, rt minderv1.RuleType, opts Options,
458	) (*v1datasources.DataSourceRegistry, error) {	9✔
459	rawproj := rt.GetContext().GetProject()	9✔
460	proj, err := uuid.Parse(rawproj)	9✔
461	if err != nil {	10✔
462	return nil, fmt.Errorf("failed to parse project UUID: %w", err)	1✔
463	}	1✔
464
465	instantiations := rt.GetDef().GetEval().GetDataSources()	8✔
466	reg := v1datasources.NewDataSourceRegistry()	8✔
467		8✔
468	// return early so we don't need to do useless work	8✔
469	if len(instantiations) == 0 {	9✔
470	return reg, nil	1✔
471	}	1✔
472
473	stx, err := d.txBuilder(d, opts)	7✔
474	if err != nil {	7✔
475	return nil, fmt.Errorf("failed to start transaction: %w", err)	×
476	}	×
477
478	//nolint:gosec // we'll log this error later.
479	defer stx.Rollback()	7✔
480		7✔
481	tx := stx.Q()	7✔
482		7✔
483	projectHierarchy, err := tx.GetParentProjects(ctx, proj)	7✔
484	if err != nil {	8✔
485	return nil, fmt.Errorf("failed to get project hierarchy: %w", err)	1✔
486	}	1✔
487
488	for _, ref := range instantiations {	12✔
489	inst, err := d.instantiateDataSource(ctx, ref, projectHierarchy, tx)	6✔
490	if err != nil {	11✔
491	return nil, fmt.Errorf("failed to instantiate data source: %w", err)	5✔
492	}	5✔
493
494	// Get provider from options if available, needed for authenticated data sources
495	provider := opts.getProvider()	1✔
496	impl, err := datasources.BuildFromProtobuf(inst, provider)	1✔
497	if err != nil {	1✔
498	return nil, fmt.Errorf("failed to build data source from protobuf: %w", err)	×
499	}	×
500
501	if err := reg.RegisterDataSource(getDataSourceReferenceAlias(ref), impl); err != nil {	1✔
502	return nil, fmt.Errorf("failed to register data source: %w", err)	×
503	}	×
504	}
505
506	return reg, nil	1✔
507	}
508
509	// getDataSourceReferenceAlias gets the alias that the data source will be referred to by
510	// in the registry.
511	func getDataSourceReferenceAlias(dsr *minderv1.DataSourceReference) string {	1✔
512	key := dsr.GetAlias()	1✔
513	if key == "" {	2✔
514	return dsr.GetName()	1✔
515	}	1✔
516	return key	×
517	}
518
519	// addDataSourceFunctions adds functions to a data source based on its driver type.
520	func addDataSourceFunctions(
521	ctx context.Context,
522	tx db.ExtendQuerier,
523	ds *minderv1.DataSource,
524	dsID uuid.UUID,
525	projectID uuid.UUID,
526	) error {	5✔
527	switch drv := ds.GetDriver().(type) {	5✔
528	case *minderv1.DataSource_Structured:	×
529	for name, def := range drv.Structured.GetDef() {	×
530	defBytes, err := protojson.Marshal(def)	×
531	if err != nil {	×
532	return fmt.Errorf("failed to marshal structured data definition: %w", err)	×
533	}	×
534
535	if _, err := tx.AddDataSourceFunction(ctx, db.AddDataSourceFunctionParams{	×
536	DataSourceID: dsID,	×
537	ProjectID: projectID,	×
538	Name: name,	×
539	Type: v1datasources.DataSourceDriverStruct,	×
540	Definition: defBytes,	×
541	}); err != nil {	×
542	return fmt.Errorf("failed to create data source function: %w", err)	×
543	}	×
544	}
545	case *minderv1.DataSource_Rest:	5✔
546	for name, def := range drv.Rest.GetDef() {	10✔
547	defBytes, err := protojson.Marshal(def)	5✔
548	if err != nil {	5✔
549	return fmt.Errorf("failed to marshal REST definition: %w", err)	×
550	}	×
551
552	if _, err := tx.AddDataSourceFunction(ctx, db.AddDataSourceFunctionParams{	5✔
553	DataSourceID: dsID,	5✔
554	ProjectID: projectID,	5✔
555	Name: name,	5✔
556	Type: v1datasources.DataSourceDriverRest,	5✔
557	Definition: defBytes,	5✔
558	}); err != nil {	5✔
559	return fmt.Errorf("failed to create data source function: %w", err)	×
560	}	×
561	}
562	default:	×
563	return fmt.Errorf("unsupported data source driver type: %T", drv)	×
564	}
565	return nil	5✔
566	}

mindersec / minder / 18553979900

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous