feat: updated grpc for dependencies service

core/controllers/base_v2.go

@@ -86,7 +86,6 @@ func (ctr *BaseControllerV2[T]) PutById(c *gin.Context) {
 
 	u := GetUserFromContextV2(c)
 	m := any(&model).(interfaces.ModelV2)
-	m.SetId(primitive.NewObjectID())
 	m.SetUpdated(u.Id)
 
 	if err := ctr.modelSvc.ReplaceById(id, model); err != nil {
@@ -170,14 +169,19 @@ func (ctr *BaseControllerV2[T]) DeleteList(c *gin.Context) {
 }
 
 func (ctr *BaseControllerV2[T]) getAll(c *gin.Context) {
-	models, err := ctr.modelSvc.GetMany(nil, &mongo.FindOptions{
-		Sort: bson.D{{"_id", -1}},
+	query := MustGetFilterQuery(c)
+	sort := MustGetSortOption(c)
+	if sort == nil {
+		sort = bson.D{{"_id", -1}}
+	}
+	models, err := ctr.modelSvc.GetMany(query, &mongo.FindOptions{
+		Sort: sort,
 	})
 	if err != nil {
 		HandleErrorInternalServerError(c, err)
 		return
 	}
-	total, err := ctr.modelSvc.Count(nil)
+	total, err := ctr.modelSvc.Count(query)
 	if err != nil {
 		HandleErrorInternalServerError(c, err)
 		return

core/entity/dependency.go

@@ -0,0 +1,14 @@
+package entity
+
+import "go.mongodb.org/mongo-driver/bson/primitive"
+
+type DependencyResult struct {
+	Name          string               `json:"name,omitempty" bson:"name,omitempty"`
+	NodeIds       []primitive.ObjectID `json:"node_ids,omitempty" bson:"node_ids,omitempty"`
+	Versions      []string             `json:"versions,omitempty" bson:"versions,omitempty"`
+	LatestVersion string               `json:"latest_version" bson:"latest_version"`
+	Count         int                  `json:"count,omitempty" bson:"count,omitempty"`
+	Upgradable    bool                 `json:"upgradable" bson:"upgradable"`
+	Downgradable  bool                 `json:"downgradable" bson:"downgradable"`
+	Installable   bool                 `json:"installable" bson:"installable"`
+}

core/grpc/client/client_v2.go

@@ -40,7 +40,7 @@ type GrpcClientV2 struct {
 	NodeClient               grpc2.NodeServiceClient
 	TaskClient               grpc2.TaskServiceClient
 	ModelBaseServiceV2Client grpc2.ModelBaseServiceV2Client
-	DependenciesClient       grpc2.DependencyServiceV2Client
+	DependenciesClient       grpc2.DependenciesServiceV2Client
 }
 
 func (c *GrpcClientV2) Init() (err error) {
@@ -95,7 +95,7 @@ func (c *GrpcClientV2) Register() {
 	c.NodeClient = grpc2.NewNodeServiceClient(c.conn)
 	c.ModelBaseServiceV2Client = grpc2.NewModelBaseServiceV2Client(c.conn)
 	c.TaskClient = grpc2.NewTaskServiceClient(c.conn)
-	c.DependenciesClient = grpc2.NewDependencyServiceV2Client(c.conn)
+	c.DependenciesClient = grpc2.NewDependenciesServiceV2Client(c.conn)
 
 	// log
 	log.Infof("[GrpcClient] grpc client registered client services")

core/grpc/server/dependencies_server_v2.go

@@ -2,30 +2,41 @@ package server
 
 import (
 	"context"
+	"errors"
+	"github.com/apex/log"
 	"github.com/crawlab-team/crawlab/core/models/models"
 	"github.com/crawlab-team/crawlab/core/models/service"
+	mongo2 "github.com/crawlab-team/crawlab/db/mongo"
 	"github.com/crawlab-team/crawlab/grpc"
+	"github.com/crawlab-team/crawlab/trace"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/bson/primitive"
+	"go.mongodb.org/mongo-driver/mongo"
 	"io"
 	"sync"
+	"time"
 )
 
 type DependenciesServerV2 struct {
 	grpc.UnimplementedDependenciesServiceV2Server
 	mu      *sync.Mutex
-	streams map[string]grpc.DependenciesServiceV2_ConnectServer
+	streams map[string]*grpc.DependenciesServiceV2_ConnectServer
 }
 
-func (svr DependenciesServerV2) Connect(stream grpc.DependenciesServiceV2_ConnectServer) (err error) {
+func (svr DependenciesServerV2) Connect(req *grpc.DependenciesServiceV2ConnectRequest, stream grpc.DependenciesServiceV2_ConnectServer) (err error) {
 	svr.mu.Lock()
-	defer svr.mu.Unlock()
-	req, err := stream.Recv()
-	if err != nil {
-		return err
+	svr.streams[req.NodeKey] = &stream
+	svr.mu.Unlock()
+	log.Info("[DependenciesServerV2] connected: " + req.NodeKey)
+
+	// Keep this scope alive because once this scope exits - the stream is closed
+	for {
+		select {
+		case <-stream.Context().Done():
+			log.Info("[DependenciesServerV2] disconnected: " + req.NodeKey)
+			return nil
+		}
 	}
-	svr.streams[req.NodeKey] = stream
-	return nil
 }
 
 func (svr DependenciesServerV2) Sync(ctx context.Context, request *grpc.DependenciesServiceV2SyncRequest) (response *grpc.Response, err error) {
@@ -33,20 +44,75 @@ func (svr DependenciesServerV2) Sync(ctx context.Context, request *grpc.Dependen
 	if err != nil {
 		return nil, err
 	}
-	var deps []models.DependencyV2
+
+	depsDb, err := service.NewModelServiceV2[models.DependencyV2]().GetMany(bson.M{
+		"node_id": n.Id,
+		"type":    request.Lang,
+	}, nil)
+	if err != nil {
+		if !errors.Is(err, mongo.ErrNoDocuments) {
+			log.Errorf("[DependenciesServiceV2] get dependencies from db error: %v", err)
+			return nil, err
+		}
+	}
+
+	depsDbMap := make(map[string]*models.DependencyV2)
+	for _, d := range depsDb {
+		depsDbMap[d.Name] = &d
+	}
+
+	var depsToInsert []models.DependencyV2
+	depsMap := make(map[string]*models.DependencyV2)
 	for _, dep := range request.Dependencies {
-		deps = append(deps, models.DependencyV2{
+		d := models.DependencyV2{
 			Name:    dep.Name,
 			NodeId:  n.Id,
 			Type:    request.Lang,
 			Version: dep.Version,
-		})
+		}
+		d.SetCreatedAt(time.Now())
+
+		depsMap[d.Name] = &d
+
+		_, ok := depsDbMap[d.Name]
+		if !ok {
+			depsToInsert = append(depsToInsert, d)
+		}
 	}
-	_, err = service.NewModelServiceV2[models.DependencyV2]().InsertMany(deps)
-	if err != nil {
-		return nil, err
+
+	var depIdsToDelete []primitive.ObjectID
+	for _, d := range depsDb {
+		_, ok := depsMap[d.Name]
+		if !ok {
+			depIdsToDelete = append(depIdsToDelete, d.Id)
+		}
 	}
-	return nil, nil
+
+	err = mongo2.RunTransaction(func(ctx mongo.SessionContext) (err error) {
+		if len(depIdsToDelete) > 0 {
+			err = service.NewModelServiceV2[models.DependencyV2]().DeleteMany(bson.M{
+				"_id": bson.M{"$in": depIdsToDelete},
+			})
+			if err != nil {
+				log.Errorf("[DependenciesServerV2] delete dependencies in db error: %v", err)
+				trace.PrintError(err)
+				return err
+			}
+		}
+
+		if len(depsToInsert) > 0 {
+			_, err = service.NewModelServiceV2[models.DependencyV2]().InsertMany(depsToInsert)
+			if err != nil {
+				log.Errorf("[DependenciesServerV2] insert dependencies in db error: %v", err)
+				trace.PrintError(err)
+				return err
+			}
+		}
+
+		return nil
+	})
+
+	return nil, err
 }
 
 func (svr DependenciesServerV2) UpdateTaskLog(stream grpc.DependenciesServiceV2_UpdateTaskLogServer) (err error) {
@@ -70,18 +136,45 @@ func (svr DependenciesServerV2) UpdateTaskLog(stream grpc.DependenciesServiceV2_
 				return err
 			}
 		}
-		l := models.DependencyLogV2{
-			TaskId:  taskId,
-			Content: req.Content,
+		var logs []models.DependencyLogV2
+		for _, line := range req.LogLines {
+			l := models.DependencyLogV2{
+				TaskId:  taskId,
+				Content: line,
+			}
+			l.SetCreated(t.CreatedBy)
+			logs = append(logs, l)
 		}
-		l.SetCreated(t.CreatedBy)
-		_, err = service.NewModelServiceV2[models.DependencyLogV2]().InsertOne(l)
+		_, err = service.NewModelServiceV2[models.DependencyLogV2]().InsertMany(logs)
 		if err != nil {
 			return err
 		}
 	}
 }
 
-func NewDependenciesServerV2() *DependenciesServerV2 {
-	return &DependenciesServerV2{}
+func (svr DependenciesServerV2) GetStream(key string) (stream *grpc.DependenciesServiceV2_ConnectServer, err error) {
+	svr.mu.Lock()
+	defer svr.mu.Unlock()
+	stream, ok := svr.streams[key]
+	if !ok {
+		return nil, errors.New("stream not found")
+	}
+	return stream, nil
+}
+
+func NewDependenciesServerV2() *DependenciesServerV2 {
+	return &DependenciesServerV2{
+		mu:      new(sync.Mutex),
+		streams: make(map[string]*grpc.DependenciesServiceV2_ConnectServer),
+	}
+}
+
+var depSvc *DependenciesServerV2
+
+func GetDependenciesServerV2() *DependenciesServerV2 {
+	if depSvc != nil {
+		return depSvc
+	}
+	depSvc = NewDependenciesServerV2()
+	return depSvc
 }

core/grpc/server/server_v2.go

@@ -42,10 +42,10 @@ type GrpcServerV2 struct {
 	nodeCfgSvc interfaces.NodeConfigService
 
 	// servers
-	nodeSvr             *NodeServerV2
-	taskSvr             *TaskServerV2
-	modelBaseServiceSvr *ModelBaseServiceServerV2
-	dependenciesSvr     *DependenciesServerV2
+	NodeSvr             *NodeServerV2
+	TaskSvr             *TaskServerV2
+	ModelBaseServiceSvr *ModelBaseServiceServerV2
+	DependenciesSvr     *DependenciesServerV2
 }
 
 func (svr *GrpcServerV2) GetConfigPath() (path string) {
@@ -115,9 +115,10 @@ func (svr *GrpcServerV2) Stop() (err error) {
 }
 
 func (svr *GrpcServerV2) Register() (err error) {
-	grpc2.RegisterNodeServiceServer(svr.svr, *svr.nodeSvr) // node service
-	grpc2.RegisterModelBaseServiceV2Server(svr.svr, *svr.modelBaseServiceSvr)
-	grpc2.RegisterTaskServiceServer(svr.svr, *svr.taskSvr)
+	grpc2.RegisterNodeServiceServer(svr.svr, *svr.NodeSvr)
+	grpc2.RegisterModelBaseServiceV2Server(svr.svr, *svr.ModelBaseServiceSvr)
+	grpc2.RegisterTaskServiceServer(svr.svr, *svr.TaskSvr)
+	grpc2.RegisterDependenciesServiceV2Server(svr.svr, *svr.DependenciesSvr)
 
 	return nil
 }
@@ -206,16 +207,16 @@ func NewGrpcServerV2() (svr *GrpcServerV2, err error) {
 
 	svr.nodeCfgSvc = nodeconfig.GetNodeConfigService()
 
-	svr.nodeSvr, err = NewNodeServerV2()
+	svr.NodeSvr, err = NewNodeServerV2()
 	if err != nil {
 		return nil, err
 	}
-	svr.modelBaseServiceSvr = NewModelBaseServiceV2Server()
-	svr.taskSvr, err = NewTaskServerV2()
+	svr.ModelBaseServiceSvr = NewModelBaseServiceV2Server()
+	svr.TaskSvr, err = NewTaskServerV2()
 	if err != nil {
 		return nil, err
 	}
-	svr.dependenciesSvr = NewDependenciesServerV2()
+	svr.DependenciesSvr = GetDependenciesServerV2()
 
 	// recovery options
 	recoveryOpts := []grpc_recovery.Option{

core/models/common/index_service_v2.go

@@ -0,0 +1,160 @@
+package common
+
+import (
+	"github.com/crawlab-team/crawlab/core/models/models"
+	"github.com/crawlab-team/crawlab/core/models/service"
+	"github.com/crawlab-team/crawlab/db/mongo"
+	"go.mongodb.org/mongo-driver/bson"
+	mongo2 "go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+func CreateIndexesV2() {
+	// nodes
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.NodeV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"key": 1}},       // key
+		{Keys: bson.M{"name": 1}},      // name
+		{Keys: bson.M{"is_master": 1}}, // is_master
+		{Keys: bson.M{"status": 1}},    // status
+		{Keys: bson.M{"enabled": 1}},   // enabled
+		{Keys: bson.M{"active": 1}},    // active
+	})
+
+	// projects
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.ProjectV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"name": 1}},
+	})
+
+	// spiders
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.SpiderV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"name": 1}},
+		{Keys: bson.M{"type": 1}},
+		{Keys: bson.M{"col_id": 1}},
+		{Keys: bson.M{"project_id": 1}},
+	})
+
+	// tasks
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.TaskV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"spider_id": 1}},
+		{Keys: bson.M{"status": 1}},
+		{Keys: bson.M{"node_id": 1}},
+		{Keys: bson.M{"schedule_id": 1}},
+		{Keys: bson.M{"type": 1}},
+		{Keys: bson.M{"mode": 1}},
+		{Keys: bson.M{"priority": 1}},
+		{Keys: bson.M{"parent_id": 1}},
+		{Keys: bson.M{"has_sub": 1}},
+		{Keys: bson.M{"create_ts": -1}},
+	})
+
+	// task stats
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.TaskStatV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"create_ts": 1}},
+	})
+
+	// schedules
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.ScheduleV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"name": 1}},
+		{Keys: bson.M{"spider_id": 1}},
+		{Keys: bson.M{"enabled": 1}},
+	})
+
+	// users
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.UserV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"username": 1}},
+		{Keys: bson.M{"role": 1}},
+		{Keys: bson.M{"email": 1}},
+	})
+
+	// settings
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.SettingV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"key": 1}},
+	})
+
+	// tokens
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.TokenV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"name": 1}},
+	})
+
+	// variables
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.VariableV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"key": 1}},
+	})
+
+	// data sources
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.DataSourceV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"name": 1}},
+	})
+
+	// data collections
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.DataCollectionV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.M{"name": 1}},
+	})
+
+	// roles
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.RoleV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.D{{"key", 1}}, Options: options.Index().SetUnique(true)},
+	})
+
+	// user role relations
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.UserRoleV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.D{{"user_id", 1}, {"role_id", 1}}, Options: options.Index().SetUnique(true)},
+		{Keys: bson.D{{"role_id", 1}, {"user_id", 1}}, Options: options.Index().SetUnique(true)},
+	})
+
+	// permissions
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.PermissionV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.D{{"key", 1}}, Options: options.Index().SetUnique(true)},
+	})
+
+	// role permission relations
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.RolePermissionV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{Keys: bson.D{{"role_id", 1}, {"permission_id", 1}}, Options: options.Index().SetUnique(true)},
+		{Keys: bson.D{{"permission_id", 1}, {"role_id", 1}}, Options: options.Index().SetUnique(true)},
+	})
+
+	// dependencies
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.DependencyV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{
+			Keys: bson.D{
+				{"type", 1},
+				{"node_id", 1},
+				{"name", 1},
+			},
+			Options: (&options.IndexOptions{}).SetUnique(true),
+		},
+	})
+
+	// dependency settings
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.DependencySettingV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{
+			Keys: bson.D{
+				{"type", 1},
+				{"node_id", 1},
+				{"name", 1},
+			},
+			Options: (&options.IndexOptions{}).SetUnique(true),
+		},
+	})
+
+	// dependency logs
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.DependencyLogV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{
+			Keys: bson.D{{"task_id", 1}},
+		},
+		{
+			Keys:    bson.D{{"update_ts", 1}},
+			Options: (&options.IndexOptions{}).SetExpireAfterSeconds(60 * 60 * 24),
+		},
+	})
+
+	// dependency tasks
+	mongo.GetMongoCol(service.GetCollectionNameByInstance(models.DependencyTaskV2{})).MustCreateIndexes([]mongo2.IndexModel{
+		{
+			Keys: bson.D{
+				{"update_ts", 1},
+			},
+			Options: (&options.IndexOptions{}).SetExpireAfterSeconds(60 * 60 * 24),
+		},
+	})
+}

core/models/models/dependency_v2.go

@@ -1,14 +1,18 @@
 package models
 
-import "go.mongodb.org/mongo-driver/bson/primitive"
+import (
+	"github.com/crawlab-team/crawlab/core/entity"
+	"go.mongodb.org/mongo-driver/bson/primitive"
+)
 
 type DependencyV2 struct {
 	any                       `collection:"dependencies"`
 	BaseModelV2[DependencyV2] `bson:",inline"`
-	Name                      string             `json:"name" bson:"name"`
-	Description               string             `json:"description" bson:"description"`
-	NodeId                    primitive.ObjectID `json:"node_id" bson:"node_id"`
-	Type                      string             `json:"type" bson:"type"`
-	LatestVersion             string             `json:"latest_version" bson:"latest_version"`
-	Version                   string             `json:"version" bson:"version"`
+	Name                      string                  `json:"name" bson:"name"`
+	Description               string                  `json:"description" bson:"description"`
+	NodeId                    primitive.ObjectID      `json:"node_id" bson:"node_id"`
+	Type                      string                  `json:"type" bson:"type"`
+	LatestVersion             string                  `json:"latest_version" bson:"latest_version"`
+	Version                   string                  `json:"version" bson:"version"`
+	Result                    entity.DependencyResult `json:"result" bson:"-"`
 }

core/models/models/git_v2.go

@@ -4,6 +4,7 @@ type GitV2 struct {
 	any                `collection:"gits"`
 	BaseModelV2[GitV2] `bson:",inline"`
 	Url                string `json:"url" bson:"url"`
+	Name               string `json:"name" bson:"name"`
 	AuthType           string `json:"auth_type" bson:"auth_type"`
 	Username           string `json:"username" bson:"username"`
 	Password           string `json:"password" bson:"password"`

core/node/service/master_service_v2.go

@@ -54,7 +54,7 @@ func (svc *MasterServiceV2) Init() (err error) {
 
 func (svc *MasterServiceV2) Start() {
 	// create indexes
-	common.CreateIndexes()
+	common.CreateIndexesV2()
 
 	// start grpc server
 	if err := svc.server.Start(); err != nil {