refactor: Rework definition validation

internal/db/collection_define.go

@@ -25,113 +25,126 @@
 	"github.com/sourcenetwork/defradb/internal/db/description"
 )
 
-func (db *db) createCollection(
+func (db *db) createCollections(
 	ctx context.Context,
-	def client.CollectionDefinition,
 	newDefinitions []client.CollectionDefinition,
-) (client.Collection, error) {
-	schema := def.Schema
-	desc := def.Description
-	txn := mustGetContextTxn(ctx)
-
-	if desc.Name.HasValue() {
-		exists, err := description.HasCollectionByName(ctx, txn, desc.Name.Value())
-		if err != nil {
-			return nil, err
-		}
-		if exists {
-			return nil, ErrCollectionAlreadyExists
-		}
-	}
+) ([]client.CollectionDefinition, error) {
+	returnDescriptions := make([]client.CollectionDefinition, len(newDefinitions))
 
 	existingDefinitions, err := db.getAllActiveDefinitions(ctx)
 	if err != nil {
 		return nil, err
 	}
 
-	schemaByName := map[string]client.SchemaDescription{}
-	for _, existingDefinition := range existingDefinitions {
-		schemaByName[existingDefinition.Schema.Name] = existingDefinition.Schema
-	}
-	for _, newDefinition := range newDefinitions {
-		schemaByName[newDefinition.Schema.Name] = newDefinition.Schema
-	}
+	txn := mustGetContextTxn(ctx)
 
-	_, err = validateUpdateSchemaFields(schemaByName, client.SchemaDescription{}, schema)
-	if err != nil {
-		return nil, err
-	}
+	for i, def := range newDefinitions {
+		schemaByName := map[string]client.SchemaDescription{}
+		for _, existingDefinition := range existingDefinitions {
+			schemaByName[existingDefinition.Schema.Name] = existingDefinition.Schema
+		}
+		for _, newDefinition := range newDefinitions {
+			schemaByName[newDefinition.Schema.Name] = newDefinition.Schema
+		}
 
-	definitionsByName := map[string]client.CollectionDefinition{}
-	for _, existingDefinition := range existingDefinitions {
-		definitionsByName[existingDefinition.GetName()] = existingDefinition
-	}
-	for _, newDefinition := range newDefinitions {
-		definitionsByName[newDefinition.GetName()] = newDefinition
-	}
-	err = db.validateNewCollection(def, definitionsByName)
-	if err != nil {
-		return nil, err
+		schema, err := description.CreateSchemaVersion(ctx, txn, def.Schema)
+		if err != nil {
+			return nil, err
+		}
+		newDefinitions[i].Description.SchemaVersionID = schema.VersionID
+		newDefinitions[i].Schema = schema
 	}
 
-	colSeq, err := db.getSequence(ctx, core.CollectionIDSequenceKey{})
-	if err != nil {
-		return nil, err
-	}
-	colID, err := colSeq.next(ctx)
-	if err != nil {
-		return nil, err
-	}
+	for i, def := range newDefinitions {
+		if len(def.Description.Fields) == 0 {
+			// This is a schema-only definition, we should not create a collection for it
+			continue
+		}
 
-	fieldSeq, err := db.getSequence(ctx, core.NewFieldIDSequenceKey(uint32(colID)))
-	if err != nil {
-		return nil, err
-	}
+		colSeq, err := db.getSequence(ctx, core.CollectionIDSequenceKey{})
+		if err != nil {
+			return nil, err
+		}
+		colID, err := colSeq.next(ctx)
+		if err != nil {
+			return nil, err
+		}
 
-	desc.ID = uint32(colID)
-	desc.RootID = desc.ID
+		fieldSeq, err := db.getSequence(ctx, core.NewFieldIDSequenceKey(uint32(colID)))
+		if err != nil {
+			return nil, err
+		}
 
-	schema, err = description.CreateSchemaVersion(ctx, txn, schema)
-	if err != nil {
-		return nil, err
-	}
-	desc.SchemaVersionID = schema.VersionID
-	for _, localField := range desc.Fields {
-		var fieldID uint64
-		if localField.Name == request.DocIDFieldName {
-			// There is no hard technical requirement for this, we just think it looks nicer
-			// if the doc id is at the zero index.  It makes it look a little nicer in commit
-			// queries too.
-			fieldID = 0
-		} else {
-			fieldID, err = fieldSeq.next(ctx)
-			if err != nil {
-				return nil, err
+		newDefinitions[i].Description.ID = uint32(colID)
+		newDefinitions[i].Description.RootID = newDefinitions[i].Description.ID
+
+		for _, localField := range def.Description.Fields {
+			var fieldID uint64
+			if localField.Name == request.DocIDFieldName {
+				// There is no hard technical requirement for this, we just think it looks nicer
+				// if the doc id is at the zero index.  It makes it look a little nicer in commit
+				// queries too.
+				fieldID = 0
+			} else {
+				fieldID, err = fieldSeq.next(ctx)
+				if err != nil {
+					return nil, err
+				}
 			}
-		}
 
-		for i := range desc.Fields {
-			if desc.Fields[i].Name == localField.Name {
-				desc.Fields[i].ID = client.FieldID(fieldID)
-				break
+			for j := range def.Description.Fields {
+				if def.Description.Fields[j].Name == localField.Name {
+					newDefinitions[i].Description.Fields[j].ID = client.FieldID(fieldID)
+					break
+				}
 			}
 		}
 	}
 
-	desc, err = description.SaveCollection(ctx, txn, desc)
+	err = db.validateNewCollection(
+		ctx,
+		append(
+			append(
+				[]client.CollectionDefinition{},
+				newDefinitions...,
+			),
+			existingDefinitions...,
+		),
+		existingDefinitions,
+	)
 	if err != nil {
 		return nil, err
 	}
 
-	col := db.newCollection(desc, schema)
+	for _, def := range newDefinitions {
+		if len(def.Description.Fields) == 0 {
+			// This is a schema-only definition, we should not create a collection for it
+			returnDescriptions = append(returnDescriptions, def)
+			continue
+		}
+
+		desc, err := description.SaveCollection(ctx, txn, def.Description)
+		if err != nil {
+			return nil, err
+		}
+
+		col := db.newCollection(desc, def.Schema)
+
+		for _, index := range desc.Indexes {
+			if _, err := col.createIndex(ctx, index); err != nil {
+				return nil, err
+			}
+		}
 
-	for _, index := range desc.Indexes {
-		if _, err := col.createIndex(ctx, index); err != nil {
+		result, err := db.getCollectionByID(ctx, desc.ID)
+		if err != nil {
 			return nil, err
 		}
+
+		returnDescriptions = append(returnDescriptions, result.Definition())
 	}
 
-	return db.getCollectionByID(ctx, desc.ID)
+	return returnDescriptions, nil
 }
 
 func (db *db) patchCollection(
@@ -171,7 +184,7 @@
 		return err
 	}
 
-	err = db.validateCollectionChanges(existingColsByID, newColsByID)
+	err = db.validateCollectionChanges(ctx, cols, newColsByID)
 	if err != nil {
 		return err
 	}