OSAC-704: Add catalog_item support to Cluster and ComputeInstance create flow

go.mod

@@ -86,6 +86,7 @@ require (
 	github.com/prometheus/common v0.66.1 // indirect
 	github.com/prometheus/procfs v0.17.0 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 // indirect
+	github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 // indirect
 	github.com/segmentio/asm v1.2.1 // indirect
 	github.com/sirupsen/logrus v1.9.4 // indirect
 	github.com/tchap/go-patricia/v2 v2.3.3 // indirect

go.sum

@@ -257,6 +257,8 @@ github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9/go.mod h1:bCqn
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
+github.com/santhosh-tekuri/jsonschema/v6 v6.0.2 h1:KRzFb2m7YtdldCEkzs6KqmJw4nqEVZGK7IN2kJkjTuQ=
+github.com/santhosh-tekuri/jsonschema/v6 v6.0.2/go.mod h1:JXeL+ps8p7/KNMjDQk3TCwPpBy0wYklyWTfbkIzdIFU=
 github.com/segmentio/asm v1.2.1 h1:DTNbBqs57ioxAD4PrArqftgypG4/qNpXoJx8TVXxPR0=
 github.com/segmentio/asm v1.2.1/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr5aAcs=
 github.com/segmentio/encoding v0.3.6 h1:E6lVLyDPseWEulBmCmAKPanDd3jiyGDo5gMcugCRwZQ=

internal/servers/catalog_item_validation.go

@@ -0,0 +1,201 @@
+/*
+Copyright (c) 2025 Red Hat Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
+License. You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific
+language governing permissions and limitations under the License.
+*/
+
+package servers
+
+import (
+	"encoding/json"
+	"strings"
+
+	"github.com/santhosh-tekuri/jsonschema/v6"
+	grpccodes "google.golang.org/grpc/codes"
+	grpcstatus "google.golang.org/grpc/status"
+	"google.golang.org/protobuf/encoding/protojson"
+	"google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/types/known/structpb"
+
+	privatev1 "github.com/osac-project/fulfillment-service/internal/api/osac/private/v1"
+)
+
+// catalogItem is implemented by both ClusterCatalogItem and ComputeInstanceCatalogItem.
+type catalogItem interface {
+	proto.Message
+	GetPublished() bool
+	GetTemplate() string
+	GetFieldDefinitions() []*privatev1.FieldDefinition
+	GetMetadata() *privatev1.Metadata
+}
+
+// applyFieldDefinitions processes field definitions from a catalog item against a resource spec.
+// For non-editable fields: overrides user-provided values with the catalog item default.
+// For editable fields with user values: validates against the JSON Schema.
+// For editable fields without user values: applies the catalog item default.
+func applyFieldDefinitions(
+	spec proto.Message,
+	fieldDefinitions []*privatev1.FieldDefinition,
+) error {
+	if len(fieldDefinitions) == 0 {
+		return nil
+	}
+
+	marshaller := protojson.MarshalOptions{UseProtoNames: true}
+	specJSON, err := marshaller.Marshal(spec)
+	if err != nil {
+		return grpcstatus.Errorf(grpccodes.Internal, "failed to marshal spec: %v", err)
+	}
+
+	var specMap map[string]any
+	if err := json.Unmarshal(specJSON, &specMap); err != nil {
+		return grpcstatus.Errorf(grpccodes.Internal, "failed to parse spec: %v", err)
+	}
+
+	compiler := jsonschema.NewCompiler()
+
+	for _, fd := range fieldDefinitions {
+		path := fd.GetPath()
+		if path == "" {
+			continue
+		}
+
+		defaultVal := fd.GetDefault()
+		userVal, userHasValue := getNestedValue(specMap, path)
+
+		if !fd.GetEditable() {
+			if defaultVal == nil {
+				return grpcstatus.Errorf(grpccodes.Internal,
+					"catalog item misconfigured: non-editable field '%s' has no default value", path)
+			}
+			if err := applyDefault(specMap, path, defaultVal); err != nil {
+				return err
+			}
+		} else {
+			if userHasValue && userVal != nil {
+				schema := fd.GetValidationSchema()
+				if schema != "" {
+					if err := validateAgainstSchema(compiler, path, userVal, schema); err != nil {
+						return err
+					}
+				}
+			} else {
+				if err := applyDefault(specMap, path, defaultVal); err != nil {
+					return err
+				}
+			}
+		}
+	}
+
+	updatedJSON, err := json.Marshal(specMap)
+	if err != nil {
+		return grpcstatus.Errorf(grpccodes.Internal, "failed to serialize updated spec: %v", err)
+	}
+
+	proto.Reset(spec)
+	if err := protojson.Unmarshal(updatedJSON, spec); err != nil {
+		return grpcstatus.Errorf(grpccodes.Internal, "failed to apply updated spec: %v", err)
+	}
+
+	return nil
+}
+
+// validateCatalogItemAccess checks that a catalog item is published and not deleted.
+// Tenant visibility is enforced by the GenericDAO's tenancy logic at the query level.
+func validateCatalogItemAccess(item catalogItem, ref string) error {
+	if item.GetMetadata().HasDeletionTimestamp() {
+		return grpcstatus.Errorf(grpccodes.InvalidArgument,
+			"catalog item '%s' has been deleted", ref)
+	}
+	if !item.GetPublished() {
+		return grpcstatus.Errorf(grpccodes.NotFound,
+			"catalog item '%s' is not published", ref)
+	}
+	return nil
+}
+
+func applyDefault(specMap map[string]any, path string, defaultVal *structpb.Value) error {
+	if defaultVal == nil {
+		return nil
+	}
+	defaultAny, err := defaultVal.MarshalJSON()
+	if err != nil {
+		return grpcstatus.Errorf(grpccodes.Internal,
+			"failed to marshal default for field '%s': %v", path, err)
+	}
+	var parsed any
+	if err := json.Unmarshal(defaultAny, &parsed); err != nil {
+		return grpcstatus.Errorf(grpccodes.Internal,
+			"failed to parse default for field '%s': %v", path, err)
+	}
+	setNestedValue(specMap, path, parsed)
+	return nil
+}
+
+func validateAgainstSchema(compiler *jsonschema.Compiler, path string, value any, schemaStr string) error {
+	resourceName := "schema_" + strings.ReplaceAll(path, ".", "_") + ".json"
+	var schemaDoc any
+	if err := json.Unmarshal([]byte(schemaStr), &schemaDoc); err != nil {
+		return grpcstatus.Errorf(grpccodes.Internal,
+			"invalid validation schema for field '%s': %v", path, err)
+	}
+	if err := compiler.AddResource(resourceName, schemaDoc); err != nil {
+		return grpcstatus.Errorf(grpccodes.Internal,
+			"invalid validation schema for field '%s': %v", path, err)
+	}
+	schema, err := compiler.Compile(resourceName)
+	if err != nil {
+		return grpcstatus.Errorf(grpccodes.Internal,
+			"failed to compile validation schema for field '%s': %v", path, err)
+	}
+	if err := schema.Validate(value); err != nil {
+		return grpcstatus.Errorf(grpccodes.InvalidArgument,
+			"validation failed for field '%s': %v", path, err)
+	}
+	return nil
+}
+
+func getNestedValue(m map[string]any, path string) (any, bool) {
+	parts := strings.Split(path, ".")
+	current := any(m)
+	for _, part := range parts {
+		currentMap, ok := current.(map[string]any)
+		if !ok {
+			return nil, false
+		}
+		current, ok = currentMap[part]
+		if !ok {
+			return nil, false
+		}
+	}
+	return current, true
+}
+
+func setNestedValue(m map[string]any, path string, value any) {
+	parts := strings.Split(path, ".")
+	current := m
+	for i, part := range parts {
+		if i == len(parts)-1 {
+			current[part] = value
+			return
+		}
+		next, ok := current[part]
+		if !ok {
+			next = map[string]any{}
+			current[part] = next
+		}
+		currentMap, ok := next.(map[string]any)
+		if !ok {
+			currentMap = map[string]any{}
+			current[part] = currentMap
+		}
+		current = currentMap
+	}
+}

internal/servers/private_clusters_server.go

@@ -52,10 +52,11 @@ var _ privatev1.ClustersServer = (*PrivateClustersServer)(nil)
 
 type PrivateClustersServer struct {
 	privatev1.UnimplementedClustersServer
-	logger       *slog.Logger
-	templatesDao *dao.GenericDAO[*privatev1.ClusterTemplate]
-	hostTypesDao *dao.GenericDAO[*privatev1.HostType]
-	generic      *GenericServer[*privatev1.Cluster]
+	logger          *slog.Logger
+	templatesDao    *dao.GenericDAO[*privatev1.ClusterTemplate]
+	catalogItemsDao *dao.GenericDAO[*privatev1.ClusterCatalogItem]
+	hostTypesDao    *dao.GenericDAO[*privatev1.HostType]
+	generic         *GenericServer[*privatev1.Cluster]
 }
 
 func NewPrivateClustersServer() *PrivateClustersServerBuilder {
@@ -110,6 +111,16 @@ func (b *PrivateClustersServerBuilder) Build() (result *PrivateClustersServer, e
 		return
 	}
 
+	// Create the catalog items DAO:
+	catalogItemsDao, err := dao.NewGenericDAO[*privatev1.ClusterCatalogItem]().
+		SetLogger(b.logger).
+		SetTenancyLogic(b.tenancyLogic).
+		SetMetricsRegisterer(b.metricsRegisterer).
+		Build()
+	if err != nil {
+		return
+	}
+
 	// Create the host types DAO:
 	hostTypesDao, err := dao.NewGenericDAO[*privatev1.HostType]().
 		SetLogger(b.logger).
@@ -135,10 +146,11 @@ func (b *PrivateClustersServerBuilder) Build() (result *PrivateClustersServer, e
 
 	// Create and populate the object:
 	result = &PrivateClustersServer{
-		logger:       b.logger,
-		templatesDao: templatesDao,
-		hostTypesDao: hostTypesDao,
-		generic:      generic,
+		logger:          b.logger,
+		templatesDao:    templatesDao,
+		catalogItemsDao: catalogItemsDao,
+		hostTypesDao:    hostTypesDao,
+		generic:         generic,
 	}
 	return
 }
@@ -180,11 +192,25 @@ func (s *PrivateClustersServer) Create(ctx context.Context,
 		return
 	}
 
-	// Validate template and perform transformations:
-	err = s.validateAndTransformCluster(ctx, request.GetObject())
-	if err != nil {
+	// Dispatch between catalog item and template paths:
+	catalogItemRef := spec.GetCatalogItem()
+	templateRef := spec.GetTemplate()
+	if catalogItemRef != "" && templateRef != "" {
+		err = grpcstatus.Errorf(grpccodes.InvalidArgument,
+			"catalog_item and template are mutually exclusive")
 		return
 	}
+	if catalogItemRef != "" {
+		err = s.validateAndTransformCatalogItem(ctx, request.GetObject())
+		if err != nil {
+			return
+		}
+	} else {
+		err = s.validateAndTransformCluster(ctx, request.GetObject())
+		if err != nil {
+			return
+		}
+	}
 
 	err = s.generic.Create(ctx, request, &response)
 	return
@@ -458,13 +484,14 @@ func (s *PrivateClustersServer) validateNodeSetHostTypeImmutability(
 // cannot be changed after cluster creation.
 func (s *PrivateClustersServer) validateTemplateImmutability(ctx context.Context,
 	request *privatev1.ClustersUpdateRequest) error {
-	// Check if template or template_parameters are being updated:
+	// Check if template, template_parameters, or catalog_item are being updated:
 	updateMask := request.GetUpdateMask()
 	updatingTemplate := s.isFieldInMask(updateMask, "spec.template")
 	updatingTemplateParams := s.isFieldInMask(updateMask, "spec.template_parameters")
+	updatingCatalogItem := s.isFieldInMask(updateMask, "spec.catalog_item")
 
-	// If neither field is being updated, no validation needed:
-	if !updatingTemplate && !updatingTemplateParams {
+	// If none of the immutable fields are being updated, no validation needed:
+	if !updatingTemplate && !updatingTemplateParams && !updatingCatalogItem {
 		return nil
 	}
 
@@ -504,6 +531,15 @@ func (s *PrivateClustersServer) validateTemplateImmutability(ctx context.Context
 		}
 	}
 
+	if updatingCatalogItem && existingSpec.GetCatalogItem() != newSpec.GetCatalogItem() {
+		return grpcstatus.Errorf(
+			grpccodes.InvalidArgument,
+			"cannot change spec.catalog_item from '%s' to '%s': catalog item is immutable",
+			existingSpec.GetCatalogItem(),
+			newSpec.GetCatalogItem(),
+		)
+	}
+
 	return nil
 }
 
@@ -696,3 +732,68 @@ func (s *PrivateClustersServer) validateAndTransformCluster(ctx context.Context,
 
 	return nil
 }
+
+func (s *PrivateClustersServer) validateAndTransformCatalogItem(ctx context.Context, cluster *privatev1.Cluster) error {
+	if cluster == nil {
+		return grpcstatus.Errorf(grpccodes.InvalidArgument, "object is mandatory")
+	}
+	catalogItemRef := cluster.GetSpec().GetCatalogItem()
+	if catalogItemRef == "" {
+		return grpcstatus.Errorf(grpccodes.InvalidArgument, "catalog_item is mandatory")
+	}
+
+	catalogItem, err := s.lookupCatalogItem(ctx, catalogItemRef)
+	if err != nil {
+		return err
+	}
+
+	if err := validateCatalogItemAccess(catalogItem, catalogItemRef); err != nil {
+		return err
+	}
+
+	templateRef := catalogItem.GetTemplate()
+	if templateRef != "" {
+		cluster.GetSpec().SetTemplate(templateRef)
+	}
+
+	if err := applyFieldDefinitions(cluster.GetSpec(), catalogItem.GetFieldDefinitions()); err != nil {
+		return err
+	}
+
+	if err := utils.ValidateClusterSpecFields(cluster.GetSpec()); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (s *PrivateClustersServer) lookupCatalogItem(ctx context.Context,
+	key string) (result *privatev1.ClusterCatalogItem, err error) {
+	if key == "" {
+		return
+	}
+	response, err := s.catalogItemsDao.List().
+		SetFilter(fmt.Sprintf("this.id == %[1]s || this.metadata.name == %[1]s", strconv.Quote(key))).
+		SetLimit(1).
+		Do(ctx)
+	if err != nil {
+		var deniedErr *dao.ErrDenied
+		if errors.As(err, &deniedErr) {
+			err = grpcstatus.Errorf(grpccodes.PermissionDenied, "%s", deniedErr.Reason)
+			return
+		}
+		s.logger.ErrorContext(ctx, "Failed to lookup catalog item",
+			slog.String("key", key),
+			slog.Any("error", err))
+		err = grpcstatus.Errorf(grpccodes.Internal, "failed to lookup catalog item")
+		return
+	}
+	items := response.GetItems()
+	if len(items) == 0 {
+		err = grpcstatus.Errorf(grpccodes.NotFound,
+			"there is no catalog item with identifier or name '%s'", key)
+		return
+	}
+	result = items[0]
+	return
+}