Merge pull request #75 from satti-hari-krishna-reddy/newPipelines

making pipelines to show up in the ui

db-connector.go

@@ -8041,12 +8041,12 @@ func GetPipelines(ctx context.Context, OrgId string) ([]Pipeline, error) {
 		return pipelines, err
 
 	} else {
-		// q := datastore.NewQuery(nameKey).Filter("org_id = ", OrgId).Limit(1000)
+		q := datastore.NewQuery(nameKey).Filter("org_id = ", OrgId).Limit(1000)
 
-		// _, err := project.Dbclient.GetAll(ctx, q, &pipelines)
-		// if err != nil && len(pipelines) == 0 {
-		// 	return pipelines, err
-		// }
+		_, err := project.Dbclient.GetAll(ctx, q, &pipelines)
+		if err != nil && len(pipelines) == 0 {
+			return pipelines, err
+		}
 	}
 
 	return pipelines, nil

pipelines.go

@@ -195,6 +195,7 @@ func HandleNewPipelineRegister(resp http.ResponseWriter, request *http.Request)
 		pipelineData.Owner = user.Id
 		pipelineData.Status = "running"
 		pipelineData.TriggerId = pipeline.TriggerId
+		pipelineData.StartNode = pipeline.StartNode
 
 		err = setPipelineTrigger(ctx, pipelineData)
 		if err != nil {

shared.go

@@ -4074,12 +4074,12 @@ func HandleGetTriggers(resp http.ResponseWriter, request *http.Request) {
 	workflowsChan := make(chan []Workflow)
 	schedulesChan := make(chan []ScheduleOld)
 	hooksChan := make(chan []Hook)
-	// pipelinesChan := make(chan []Pipeline)
+	pipelinesChan := make(chan []Pipeline)
 
 	errChan := make(chan error)
 
 	wg := sync.WaitGroup{}
-	wg.Add(3)
+	wg.Add(4)
 
 	go func() {
 		workflows, err := GetAllWorkflowsByQuery(ctx, user)
@@ -4116,16 +4116,16 @@ func HandleGetTriggers(resp http.ResponseWriter, request *http.Request) {
 		hooksChan <- hooks
 	}()
 
-	// go func() {
-	// 	pipelines, err := GetPipelines(ctx, user.ActiveOrg.Id)
-	// 	if err != nil {
-	// 		wg.Done()
-	// 		errChan <- err
-	// 		return
-	// 	}
-	// 	wg.Done()
-	// 	pipelinesChan <- pipelines
-	// }()
+	go func() {
+		pipelines, err := GetPipelines(ctx, user.ActiveOrg.Id)
+		if err != nil {
+			wg.Done()
+			errChan <- err
+			return
+		}
+		wg.Done()
+		pipelinesChan <- pipelines
+	}()
 
 	wg.Wait()
 
@@ -4145,25 +4145,25 @@ func HandleGetTriggers(resp http.ResponseWriter, request *http.Request) {
 	hooks := <-hooksChan
 	schedules := <-schedulesChan
 	workflows := <-workflowsChan
-	// pipelines := <-pipelinesChan
+	pipelines := <-pipelinesChan
 
 	hookMap := map[string]Hook{}
 	scheduleMap := map[string]ScheduleOld{}
-	// pipelineMap := map[string]Pipeline{}
+	pipelineMap := map[string]Pipeline{}
 
 	for _, hook := range hooks {
 		hookMap[hook.Id] = hook
 	}
 	for _, schedule := range schedules {
 		scheduleMap[schedule.Id] = schedule
 	}
-	// for _, pipeline := range pipelines {
-	// 	pipelineMap[pipeline.TriggerId] = pipeline
-	// }
+	for _, pipeline := range pipelines {
+		pipelineMap[pipeline.TriggerId] = pipeline
+	}
 
 	allHooks := []Hook{}
 	allSchedules := []ScheduleOld{}
-	// allPipelines := []Pipeline{}
+	allPipelines := []Pipeline{}
 	// Now loop through the workflow triggers to see if anything is not in sync
 	for _, workflow := range workflows {
 		for _, trigger := range workflow.Triggers {
@@ -4285,47 +4285,46 @@ func HandleGetTriggers(resp http.ResponseWriter, request *http.Request) {
 			case "PIPELINE":
 				{
 
-					// 		storedPipeline, exist := pipelineMap[trigger.ID]
-					// 		if exist && storedPipeline.Status != "uninitialized" {
-					// 			startNode := ""
+							storedPipeline, exist := pipelineMap[trigger.ID]
+							if exist && storedPipeline.Status != "uninitialized" {
+								startNode := ""
 
-					// 			storedPipeline.WorkflowId = workflow.ID
+								storedPipeline.WorkflowId = workflow.ID
 
-					// 			if len(workflow.Branches) != 0 {
-					// 				for _, branch := range workflow.Branches {
-					// 					if branch.SourceID == trigger.ID {
-					// 						startNode = branch.DestinationID
-					// 					}
-					// 				}
-					// 			}
-					// 			if startNode == "" {
-					// 				startNode = workflow.Start
-					// 			}
-					// 			storedPipeline.StartNode = startNode
-					// 			allPipelines = append(allPipelines, storedPipeline)
+								if len(workflow.Branches) != 0 {
+									for _, branch := range workflow.Branches {
+										if branch.SourceID == trigger.ID {
+											startNode = branch.DestinationID
+										}
+									}
+								}
+								if startNode == "" {
+									startNode = workflow.Start
+								}
+								storedPipeline.StartNode = startNode
+								allPipelines = append(allPipelines, storedPipeline)
 
-					// 		}
-					// 	}
+							}
+						}
 				}
 			}
 		}
-	}
 
 	sort.SliceStable(allHooks, func(i, j int) bool {
 		return allHooks[i].Info.Name < allHooks[j].Info.Name
 	})
 	sort.SliceStable(allSchedules, func(i, j int) bool {
 		return allSchedules[i].Name < allSchedules[j].Name
 	})
-	// sort.SliceStable(pipelines, func(i, j int) bool {
-	// 	return pipelines[i].Name < pipelines[j].Name
-	// })
+	sort.SliceStable(pipelines, func(i, j int) bool {
+		return pipelines[i].Name < pipelines[j].Name
+	})
 
 	allTriggersWrapper := AllTriggersWrapper{}
 
 	allTriggersWrapper.WebHooks = allHooks
 	allTriggersWrapper.Schedules = allSchedules
-	// allTriggersWrapper.Pipelines = allPipelines
+	allTriggersWrapper.Pipelines = allPipelines
 
 	newjson, err := json.Marshal(allTriggersWrapper)
 	if err != nil {

structs.go

@@ -2163,7 +2163,7 @@ type SubResponse struct {
 }
 
 type AllTriggersWrapper struct {
-	// Pipelines []Pipeline `json:"pipelines"`
+	Pipelines []Pipeline `json:"pipelines"`
 	WebHooks   []Hook `json:"webhooks"`
 	Schedules []ScheduleOld `json:"schedules"`
 }