Merge pull request cockroachdb#31 from crowdflux/gopher

Work Flow Builder Changes

app/DAL/clients/postgres/postgres.go

@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"reflect"
 
+	"github.com/crowdflux/angel/app/DAL/repositories"
 	"github.com/crowdflux/angel/app/DAL/repositories/queries"
 	"gopkg.in/gorp.v1"
 )
@@ -16,6 +17,8 @@ type postgres_db struct {
 	gorpDbMap *gorp.DbMap
 }
 
+var _ repositories.IDatabase = &postgres_db{}
+
 func (pg *postgres_db) Insert(list ...interface{}) (err error) {
 	err = pg.gorpDbMap.Insert(list...)
 	return

app/DAL/repositories/projects_repo/interface.go

@@ -13,6 +13,7 @@ type IProjectsRepo interface {
 	Add(models.Project) error
 	Update(models.Project) error
 	Delete(id uuid.UUID) error
+	IfIdExist(uuid.UUID) (bool, error)
 }
 
 //=============================================================================================//
@@ -44,6 +45,9 @@ func (i *inMemProjectRepo) Update(models.Project) error {
 func (i *inMemProjectRepo) Delete(uuid.UUID) error {
 	return nil
 }
+func (i *inMemProjectRepo) IfIdExist(uuid.UUID) (bool, error) {
+	return false, nil
+}
 
 func (i *inMemProjectRepo) Save(pr models.Project) error {
 	i.Lock()

app/DAL/repositories/projects_repo/project_repo.go

@@ -80,3 +80,10 @@ func (i *projectsRepo) Delete(id uuid.UUID) error {
 	}
 	return err
 }
+func (i *projectsRepo) IfIdExist(id uuid.UUID) (ifExist bool, err error) {
+	err = i.pg.SelectOne(&ifExist, `select exists(select 1 from projects where id=$1)`, id)
+	if err != nil {
+		return
+	}
+	return
+}

app/DAL/repositories/step_repo/error.go

@@ -0,0 +1,5 @@
+package step_repo
+
+import "github.com/crowdflux/angel/app/services/plerrors"
+
+var ErrSteptNotFound = plerrors.ServiceError{"SR_0001", "Step not found"}

app/DAL/repositories/step_repo/export.go

@@ -10,6 +10,10 @@ type IStepRepo interface {
 	GetById(id uuid.UUID) (models.Step, error)
 	GetStartStep(projectId uuid.UUID) (models.Step, error)
 	GetEndStep(projectId uuid.UUID) (models.Step, error)
+	GetStepsByWorkflowId(id uuid.UUID) ([]models.Step, error)
+	AddMany([]models.Step) error
+	UpdateMany([]models.Step) (int64, error)
+	DeleteMany([]models.Step) (int64, error)
 }
 
 func New() IStepRepo {

app/DAL/repositories/step_repo/step_repo.go

@@ -12,11 +12,19 @@ type stepRepo struct {
 
 const stepTable = "step"
 
+var _ IStepRepo = &stepRepo{}
+
 func (s *stepRepo) GetById(id uuid.UUID) (step models.Step, err error) {
 	err = s.Db.SelectById(&step, id)
 	return
 }
 
+func (s *stepRepo) GetStepsByWorkflowId(workFlowId uuid.UUID) (steps []models.Step, err error) {
+
+	_, err = s.Db.Select(&steps, `select * from step where work_flow_id = $1 `, workFlowId.String())
+	return
+}
+
 func (s *stepRepo) GetStartStep(projectId uuid.UUID) (step models.Step, err error) {
 
 	err = s.Db.SelectOne(&step, `
@@ -33,3 +41,32 @@ func (s *stepRepo) GetEndStep(projectId uuid.UUID) (step models.Step, err error)
 	where w.project_id = $1 and is_end is true`, projectId.String())
 	return
 }
+
+func (s *stepRepo) AddMany(steps []models.Step) (err error) {
+	var stepsInterface []interface{} = make([]interface{}, len(steps))
+	for i, _ := range steps {
+		stepsInterface[i] = &steps[i]
+	}
+
+	err = s.Db.Insert(stepsInterface...)
+	return
+}
+
+func (s *stepRepo) UpdateMany(steps []models.Step) (result int64, err error) {
+	var stepsInterface []interface{} = make([]interface{}, len(steps))
+	for i, _ := range steps {
+		stepsInterface[i] = &steps[i]
+	}
+
+	result, err = s.Db.Update(stepsInterface...)
+	return
+}
+func (s *stepRepo) DeleteMany(steps []models.Step) (result int64, err error) {
+	var stepsInterface []interface{} = make([]interface{}, len(steps))
+	for i, _ := range steps {
+		stepsInterface[i] = &steps[i]
+	}
+
+	result, err = s.Db.Delete(stepsInterface...)
+	return
+}

app/DAL/repositories/step_router_repo/export.go

@@ -1,6 +1,20 @@
 package step_router_repo
 
-import "github.com/crowdflux/angel/app/DAL/clients/postgres"
+import (
+	"github.com/crowdflux/angel/app/DAL/clients/postgres"
+	"github.com/crowdflux/angel/app/models"
+	"github.com/crowdflux/angel/app/models/uuid"
+)
+
+type IStepRoutesRepo interface {
+	GetRoutesByStepId(stepId uuid.UUID) ([]models.Route, error)
+	GetRoutesWithLogicByStepId(stepId uuid.UUID) ([]models.RouteWithLogicGate, error)
+	Save(models.Route) error
+	UpdateMany([]models.Route) (int64, error)
+	DeleteMany([]models.Route) (int64, error)
+	AddMany([]models.Route) error
+	GetRoutesByWorkFlowId(workFlowId uuid.UUID) ([]models.Route, error)
+}
 
 func New() IStepRoutesRepo {
 	return &stepRouteRepo{

app/DAL/repositories/step_router_repo/interface.go

@@ -7,14 +7,6 @@ import (
 	"github.com/crowdflux/angel/app/models/uuid"
 )
 
-type IStepRoutesRepo interface {
-	GetRoutesByStepId(stepId uuid.UUID) ([]models.Route, error)
-	GetRoutesWithLogicByStepId(stepId uuid.UUID) ([]models.RouteWithLogicGate, error)
-	Save(models.Route) error
-}
-
-//========================================================================================================================================================//
-
 type inMemStepRouteRepo struct {
 	sync.RWMutex
 	stepRoutes map[uuid.UUID]models.Route
@@ -41,14 +33,25 @@ func (i *inMemStepRouteRepo) GetRoutesWithLogicByStepId(stepId uuid.UUID) (route
 	panic("not implemented")
 	return routes, nil
 }
+func (i *inMemStepRouteRepo) GetRoutesByWorkFlowId(stepId uuid.UUID) (routes []models.Route, err error) {
+	return
+}
 
 func (i *inMemStepRouteRepo) Save(r models.Route) error {
 	i.Lock()
 	defer i.Unlock()
 	i.stepRoutes[r.ID] = r
 	return nil
 }
-
+func (i *inMemStepRouteRepo) AddMany(routes []models.Route) error {
+	return nil
+}
+func (i *inMemStepRouteRepo) UpdateMany(routes []models.Route) (result int64, err error) {
+	return
+}
+func (i *inMemStepRouteRepo) DeleteMany(routes []models.Route) (result int64, err error) {
+	return
+}
 func Mock() IStepRoutesRepo {
 	return &inMemStepRouteRepo{
 		stepRoutes: make(map[uuid.UUID]models.Route),

app/DAL/repositories/step_router_repo/step_router_repo.go

@@ -58,3 +58,39 @@ func (s *stepRouteRepo) Save(r models.Route) error {
 	panic("not implemented")
 	return nil
 }
+
+func (s *stepRouteRepo) GetRoutesByWorkFlowId(workFlowId uuid.UUID) (routes []models.Route, err error) {
+
+	_, err = s.Db.Select(&routes, `select r.* from routes r inner join
+	step s on s.id = r.step_id where s.work_flow_id = $1
+	`, workFlowId.String())
+	return
+}
+
+func (s *stepRouteRepo) AddMany(routes []models.Route) (err error) {
+	var routesInterface []interface{} = make([]interface{}, len(routes))
+	for i, _ := range routes {
+		routesInterface[i] = &routes[i]
+	}
+
+	err = s.Db.Insert(routesInterface...)
+	return
+}
+func (s *stepRouteRepo) UpdateMany(routes []models.Route) (response int64, err error) {
+	var routesInterface []interface{} = make([]interface{}, len(routes))
+	for i, _ := range routes {
+		routesInterface[i] = &routes[i]
+	}
+
+	response, err = s.Db.Update(routesInterface...)
+	return
+}
+func (s *stepRouteRepo) DeleteMany(routes []models.Route) (response int64, err error) {
+	var routesInterface []interface{} = make([]interface{}, len(routes))
+	for i, _ := range routes {
+		routesInterface[i] = &routes[i]
+	}
+
+	response, err = s.Db.Delete(routesInterface...)
+	return
+}

app/DAL/repositories/workflow_repo/error.go

@@ -0,0 +1,5 @@
+package workflow_repo
+
+import "github.com/crowdflux/angel/app/services/plerrors"
+
+var ErrWorkflowNotFound = plerrors.ServiceError{"WR_0001", "WorkFlow not found"}

app/DAL/repositories/workflow_repo/export.go

@@ -7,9 +7,11 @@ import (
 )
 
 type IWorkflowRepo interface {
-	Add(models.WorkFlow) error
+	Add(*models.WorkFlow) error
 	Update(models.WorkFlow) error
 	Delete(uuid.UUID) error
+	GetById(id uuid.UUID) (models.WorkFlow, error)
+	IfIdExist(uuid.UUID) (bool, error)
 }
 
 func New() IWorkflowRepo {

app/DAL/repositories/workflow_repo/workflow_repo.go

@@ -15,8 +15,8 @@ type workflow_repo struct {
 
 var _ IWorkflowRepo = &workflow_repo{}
 
-func (wr *workflow_repo) Add(wf models.WorkFlow) error {
-	return wr.db.Insert(&wf)
+func (wr *workflow_repo) Add(wf *models.WorkFlow) error {
+	return wr.db.Insert(wf)
 }
 
 func (wr *workflow_repo) Update(wf models.WorkFlow) error {
@@ -35,3 +35,15 @@ func (wr *workflow_repo) Delete(id uuid.UUID) error {
 	}
 	return err
 }
+func (wr *workflow_repo) GetById(id uuid.UUID) (wf models.WorkFlow, err error) {
+	wf = models.WorkFlow{}
+	err = wr.db.SelectById(&wf, id)
+	return
+}
+func (i *workflow_repo) IfIdExist(id uuid.UUID) (ifExist bool, err error) {
+	err = i.db.SelectOne(&ifExist, `select exists(select 1 from work_flow where id=$1)`, id)
+	if err != nil {
+		return
+	}
+	return
+}

app/api/builder.go

@@ -14,6 +14,7 @@ import (
 
 	"time"
 
+	"github.com/crowdflux/angel/app/services/work_flow_builder_svc"
 	"github.com/crowdflux/angel/app/services/work_flow_svc/step/crowdsourcing_step_svc"
 	"github.com/crowdflux/angel/app/services/work_flow_svc/step/manual_step_svc"
 	"github.com/itsjamie/gin-cors"
@@ -79,6 +80,10 @@ func Build() {
 		flu_svc_transport.AddHttpTransport(authorized)
 	}
 
+	workFlow := r.Group("/api/v0")
+	{
+		work_flow_builder_svc.AddHttpTransport(workFlow)
+	}
 	var _ image_svc1.IImageService
 
 	r.Run(":8999") // listen and serve on 127.0.0.1:8999

app/models/db_struct.go

@@ -495,6 +495,7 @@ type Step struct {
 	CreatedAt  pq.NullTime        `db:"created_at" json:"created_at" bson:"created_at"`
 	UpdatedAt  pq.NullTime        `db:"updated_at" json:"updated_at" bson:"updated_at"`
 	IsStart    bool               `db:"is_start" json:"is_start" bson:"is_start"`
+	Config     JsonF              `db:"config" json:"config" bson:"config"`
 }
 
 type StepTypes struct {

app/models/other.go

@@ -47,3 +47,10 @@ type RouteWithLogicGate struct {
 	Route
 	LogicGate LogicGate
 }
+
+type WorkflowContainer struct {
+	WorkFlow
+	Steps []Step `json:"steps"`
+	//Routes []RouteWithStepLabel
+	Routes []Route `json:"routes"`
+}

app/models/step_type/step_type.go

@@ -20,8 +20,8 @@ const (
 	Error
 )
 
-func (s StepType) Value() (driver.Value, error) {
-	return uint(s), nil
+func (s *StepType) Value() (driver.Value, error) {
+	return uint(*s), nil
 }
 func (s *StepType) Scan(src interface{}) error {
 

app/models/step_type/step_type_test.go

@@ -3,6 +3,7 @@ package step_type
 import (
 	"testing"
 
+	"github.com/notnow/src/encoding/json"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -16,3 +17,17 @@ func TestStepType_Scan(t *testing.T) {
 	assert.NoError(t, err)
 	assert.Equal(t, CrowdSourcing, s)
 }
+
+func TestStepType_Value(t *testing.T) {
+
+	type Name struct {
+		A StepType `json:"A"`
+	}
+
+	str := `{ "A" : 1 }`
+
+	n := Name{}
+
+	err := json.Unmarshal([]byte(str), &n)
+	assert.NoError(t, err)
+}

app/services/work_flow_builder_svc/export.go

@@ -0,0 +1,26 @@
+package work_flow_builder_svc
+
+import (
+	"github.com/crowdflux/angel/app/DAL/repositories/projects_repo"
+	"github.com/crowdflux/angel/app/DAL/repositories/step_repo"
+	"github.com/crowdflux/angel/app/DAL/repositories/step_router_repo"
+	"github.com/crowdflux/angel/app/DAL/repositories/workflow_repo"
+	"github.com/crowdflux/angel/app/models"
+	"github.com/crowdflux/angel/app/models/uuid"
+)
+
+type IWorkflowBuilderService interface {
+	GetWorkflowContainer(uuid.UUID) (models.WorkflowContainer, error)
+	AddWorkflowContainer(models.WorkflowContainer) (models.WorkflowContainer, error)
+	InitWorkflowContainer(uuid.UUID) (models.WorkflowContainer, error)
+	UpdateWorkflowContainer(models.WorkflowContainer) (models.WorkflowContainer, error)
+}
+
+func New() IWorkflowBuilderService {
+	return &workFlowBuilderService{
+		stepRepo:       step_repo.New(),
+		stepRouterRepo: step_router_repo.New(),
+		workflowRepo:   workflow_repo.New(),
+		projectsRep:    projects_repo.New(),
+	}
+}