package nodes import ( "context" "fmt" "time" "code.fbi.h-da.de/danet/gosdn/controller/customerrs" query "code.fbi.h-da.de/danet/gosdn/controller/store" "github.com/google/uuid" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/bson/primitive" "go.mongodb.org/mongo-driver/mongo" "go.mongodb.org/mongo-driver/mongo/options" "go.mongodb.org/mongo-driver/mongo/writeconcern" ) const storeName = "node-store.json" // Store defines a NodeStore interface. type Store interface { Add(context.Context, Node) error Update(context.Context, Node) error Delete(context.Context, Node) error Get(context.Context, query.Query) (Node, error) GetAll(context.Context) ([]Node, error) } // DatabaseNodeStore is a database store for nodes. type DatabaseNodeStore struct { collection *mongo.Collection } // NewDatabaseNodeStore returns a NodeStore. func NewDatabaseNodeStore(db *mongo.Database) Store { collection := db.Collection(storeName) return &DatabaseNodeStore{ collection: collection, } } // Get takes a nodes's UUID or name and returns the nodes. func (s *DatabaseNodeStore) Get(ctx context.Context, query query.Query) (Node, error) { var loadedNode Node if query.ID.String() != "" { loadedNode, err := s.getByID(ctx, query.ID) if err != nil { return loadedNode, customerrs.CouldNotFindError{ID: query.ID, Name: query.Name} } return loadedNode, nil } loadedNode, err := s.getByName(ctx, query.Name) if err != nil { return loadedNode, customerrs.CouldNotFindError{ID: query.ID, Name: query.Name} } return loadedNode, nil } func (s *DatabaseNodeStore) getByID(ctx context.Context, idOfNode uuid.UUID) (loadedNode Node, err error) { idAsByteArray, _ := idOfNode.MarshalBinary() result := s.collection.FindOne(ctx, bson.D{primitive.E{Key: "_id", Value: idAsByteArray}}) if result == nil { return loadedNode, customerrs.CouldNotFindError{ID: idOfNode} } err = result.Decode(&loadedNode) if err != nil { return loadedNode, customerrs.CouldNotMarshallError{Identifier: idOfNode, Type: loadedNode, Err: err} } return loadedNode, nil } func (s *DatabaseNodeStore) getByName(ctx context.Context, nameOfNode string) (loadedNode Node, err error) { result := s.collection.FindOne(ctx, bson.D{primitive.E{Key: "name", Value: nameOfNode}}) if result == nil { return loadedNode, customerrs.CouldNotFindError{Name: nameOfNode} } err = result.Decode(&loadedNode) if err != nil { return loadedNode, customerrs.CouldNotMarshallError{Identifier: nameOfNode, Type: loadedNode, Err: err} } return loadedNode, nil } // GetAll returns all stored nodes. func (s *DatabaseNodeStore) GetAll(ctx context.Context) (loadedNode []Node, err error) { cursor, err := s.collection.Find(ctx, bson.D{}) if err != nil { return []Node{}, err } defer func() { if ferr := cursor.Close(ctx); ferr != nil { fErrString := ferr.Error() err = fmt.Errorf("InternalError=%w DeferError=%+s", err, fErrString) } }() err = cursor.All(ctx, &loadedNode) if err != nil { return loadedNode, customerrs.CouldNotMarshallError{Type: loadedNode, Err: err} } return loadedNode, nil } // Add adds a node to the node store. func (s *DatabaseNodeStore) Add(ctx context.Context, node Node) (err error) { node.Metadata.ResourceVersion = 0 node.Metadata.CreatedAt = time.Now() node.Metadata.LastUpdated = time.Now() _, err = s.collection. InsertOne(ctx, node) if err != nil { return customerrs.CouldNotCreateError{Identifier: node.ID, Type: node, Err: err} } return nil } // Update updates a existing node. func (s *DatabaseNodeStore) Update(ctx context.Context, node Node) (err error) { var updatedLoadedNodes Node wc := writeconcern.Majority() txnOptions := options.Transaction().SetWriteConcern(wc) // Starts a session on the client session, err := s.collection.Database().Client().StartSession() if err != nil { return err } // Defers ending the session after the transaction is committed or ended defer session.EndSession(ctx) // Transaction callback := func(sessCtx mongo.SessionContext) (interface{}, error) { // 1. Fetch exisiting Entity existingNode, err := s.getByID(ctx, node.ID) if err != nil { return nil, err } // 2. Check if Entity.Metadata.ResourceVersion == UpdatedEntity.Metadata.ResourceVersion if node.Metadata.ResourceVersion != existingNode.Metadata.ResourceVersion { // 2.1 End transaction // 2.2 If no -> return error return nil, fmt.Errorf( "resource version %d of provided node %s is older or newer than %d in the store", node.Metadata.ResourceVersion, node.ID.String(), existingNode.Metadata.ResourceVersion, ) } // Important: You must pass sessCtx as the Context parameter to the operations for them to be executed in the // transaction. update := bson.D{primitive.E{Key: "$set", Value: node}} upsert := false after := options.After opt := options.FindOneAndUpdateOptions{ Upsert: &upsert, ReturnDocument: &after, } err = s.collection. FindOneAndUpdate( ctx, bson.M{"_id": node.ID.String()}, update, &opt). Decode(&updatedLoadedNodes) if err != nil { return nil, customerrs.CouldNotUpdateError{Identifier: node.ID, Type: node, Err: err} } // 3. End transaction return "", nil } _, err = session.WithTransaction(ctx, callback, txnOptions) if err != nil { return err } return nil } // Delete deletes a node from the node store. func (s *DatabaseNodeStore) Delete(ctx context.Context, node Node) (err error) { _, err = s.collection.DeleteOne(ctx, bson.D{primitive.E{Key: node.ID.String()}}) if err != nil { return err } return nil }