Get task from task service if worker allowed before assignment creation

edshadi · edshadi · commit b075861c04f3 · 2020-02-27T12:07:53.000+01:00
diff --git a/pkg/api/assignmentcreator/endpoint.go b/pkg/api/assignmentcreator/endpoint.go
@@ -6,6 +6,7 @@ import (
 	"github.com/gemsorg/assignment/pkg/apierror"
 	"github.com/gemsorg/assignment/pkg/assignment"
 	"github.com/gemsorg/assignment/pkg/authentication"
+	ds "github.com/gemsorg/assignment/pkg/datastore"
 	"github.com/gemsorg/assignment/pkg/service"
 	"github.com/go-kit/kit/endpoint"
 )
@@ -17,7 +18,9 @@ func makeAssignmentCreatorEndpoint(svc service.AssignmentService) endpoint.Endpo
 		req := request.(assignment.NewAssignment)
 		settings, err := svc.GetSettings(req.JobID)
 		if err != nil {
-			return nil, errorResponse(err)
+			if _, ok := err.(ds.NoRowErr); !ok {
+				return nil, errorResponse(err)
+			}
 		}
 
 		saved, err := svc.CreateAssignment(req, settings)
diff --git a/pkg/assignment/assignment.go b/pkg/assignment/assignment.go
@@ -28,7 +28,7 @@ type NewAssignment struct {
 	JobID                  uint64 `json:"job_id"`
 	TaskID                 uint64 `json:"task_id"`
 	WorkerID               uint64 `json:"worker_id"`
-	WorkerAssignmentCount  int    `json:"worker_assignment_count"`
+	WorkerAssignmentCount  int    // This comes internally from this service
 	OnboardingSuccess      bool   `json:"onboarding_success"`
 	WorkerAlreadyResponded bool   `json:"worker_already_responded"`
 	WorkerAlreadyAssigned  bool   // This comes internally from this service
diff --git a/pkg/authentication/authentication.go b/pkg/authentication/authentication.go
@@ -21,6 +21,7 @@ type AuthData struct {
 	Expiration int64
 	Issuer     string
 	UserID     uint64
+	Token      string
 }
 
 func ParseAuthData(ctx context.Context) (AuthData, error) {
@@ -37,9 +38,10 @@ func ParseAuthData(ctx context.Context) (AuthData, error) {
 	}
 
 	return AuthData{
-		int64(claims[ExpirationKey].(float64)),
-		claims[IssuerKey].(string),
-		uint64(claims[UserIDKey].(float64)),
+		Expiration: int64(claims[ExpirationKey].(float64)),
+		Issuer:     claims[IssuerKey].(string),
+		UserID:     uint64(claims[UserIDKey].(float64)),
+		Token:      tokenString,
 	}, nil
 }
 
diff --git a/pkg/authentication/authentication_test.go b/pkg/authentication/authentication_test.go
@@ -43,7 +43,7 @@ func TestParseAuthData(t *testing.T) {
 		{
 			"it returns auth data",
 			args{ctx},
-			AuthData{1591960106, "http://localhost:3000", 1},
+			AuthData{1591960106, "http://localhost:3000", 1, "abc"},
 			false,
 		},
 	}
diff --git a/pkg/authorization/authorization.go b/pkg/authorization/authorization.go
@@ -10,6 +10,7 @@ import (
 type Authorizer interface {
 	SetAuthData(data authentication.AuthData)
 	IsModerator() (bool, error)
+	GetAuthToken() string
 }
 
 type authorizor struct {
@@ -33,3 +34,7 @@ func (a *authorizor) IsModerator() (bool, error) {
 	}
 	return true, nil
 }
+
+func (a *authorizor) GetAuthToken() string {
+	return a.authData.Token
+}
diff --git a/pkg/authorization/authorization_test.go b/pkg/authorization/authorization_test.go
@@ -27,7 +27,7 @@ func TestNewAuthorizer(t *testing.T) {
 }
 
 func TestSetAuthData(t *testing.T) {
-	authData := authentication.AuthData{1591960106, "http://localhost:3000", 8}
+	authData := authentication.AuthData{1591960106, "http://localhost:3000", 8, "abc"}
 	type fields struct {
 		authData authentication.AuthData
 	}
diff --git a/pkg/datastore/assignment.go b/pkg/datastore/assignment.go
@@ -79,7 +79,7 @@ func (as *AssignmentStore) GetAssignment(id string) (*assignment.Assignment, err
 
 func (as *AssignmentStore) CreateAssignment(a assignment.NewAssignment) (*assignment.Assignment, error) {
 	result, err := as.DB.Exec(
-		"INSERT INTO assignments (job_id, task_id, worker_id) VALUES (?,?,?)",
+		"INSERT INTO assignments (job_id, task_id, worker_id, expires_at) VALUES (?,?,?, DATE_ADD(CURRENT_TIMESTAMP, INTERVAL 2 HOUR))",
 		a.JobID, a.TaskID, a.WorkerID)
 
 	if err != nil {
@@ -108,14 +108,18 @@ func (as *AssignmentStore) CreateAssignment(a assignment.NewAssignment) (*assign
 }
 
 func (as *AssignmentStore) GetSettings(jobID uint64) (*assignment.Settings, error) {
-	set := &assignment.Settings{}
-	err := as.DB.Get(set, "SELECT * FROM settings WHERE job_id = ?", jobID)
+	set := []*assignment.Settings{}
+	err := as.DB.Select(&set, "SELECT * FROM settings WHERE job_id = ?", jobID)
 
 	if err != nil {
 		return nil, err
 	}
 
-	return set, nil
+	if len(set) == 0 {
+		return nil, NoRowErr{}
+	}
+
+	return set[0], nil
 }
 
 func (as *AssignmentStore) GetWhitelist(jobID uint64, workerID uint64) (*whitelist.Whitelist, error) {
diff --git a/pkg/datastore/errors.go b/pkg/datastore/errors.go
@@ -16,3 +16,9 @@ type AssignmentNotFound struct {
 func (err AssignmentNotFound) Error() string {
 	return fmt.Sprintf("No Record found for worker_id: %d, job_id: %d", err.WorkerID, err.JobID)
 }
+
+type NoRowErr struct{}
+
+func (err NoRowErr) Error() string {
+	return "Query returned no rows"
+}
diff --git a/pkg/service/service.go b/pkg/service/service.go
@@ -5,6 +5,7 @@ import (
 	"github.com/gemsorg/assignment/pkg/authentication"
 	"github.com/gemsorg/assignment/pkg/authorization"
 	"github.com/gemsorg/assignment/pkg/datastore"
+	"github.com/gemsorg/assignment/pkg/tasksvc"
 )
 
 type AssignmentService interface {
@@ -47,22 +48,37 @@ func (s *service) GetAssignment(id string) (*assignment.Assignment, error) {
 }
 
 func (s *service) CreateAssignment(a assignment.NewAssignment, set *assignment.Settings) (*assignment.Assignment, error) {
+	if set == nil {
+		return s.store.CreateAssignment(a)
+	}
 	// if job has a whitelist, check if worker is part of it
 	if set.Whitelist {
 		wl, err := s.store.GetWhitelist(a.JobID, a.WorkerID)
 		if wl == nil || err != nil {
 			return nil, assignment.WorkerNotWhitelisted{}
 		}
 	}
+
+	// Get worker's assignment for this job
+	assigned, err := s.store.WorkerAlreadyAssigned(a.JobID, a.WorkerID)
+	a.WorkerAlreadyAssigned = assigned
+
 	// Check if the assignment is allowed
 	allowed, err := a.IsAllowed(set)
 	if !allowed {
 		return nil, err
 	}
 
-	// Get worker's assignment for this job
-	assigned, err := s.store.WorkerAlreadyAssigned(a.JobID, a.WorkerID)
-	a.WorkerAlreadyAssigned = assigned
+	// Get a task from the task service
+	params := tasksvc.SearchParams{
+		Random: true,
+	}
+	result, err := tasksvc.Search(a.JobID, s.authorizor.GetAuthToken(), params)
+
+	if err != nil {
+		return nil, err
+	}
+	a.TaskID = result.ID
 
 	return s.store.CreateAssignment(a)
 }
diff --git a/pkg/tasksvc/errors.go b/pkg/tasksvc/errors.go
@@ -0,0 +1,11 @@
+package tasksvc
+
+import "fmt"
+
+type NoAvailableTasks struct {
+	JobID uint64
+}
+
+func (err NoAvailableTasks) Error() string {
+	return fmt.Sprintf("No available tasks for job: %d", err.JobID)
+}
diff --git a/pkg/tasksvc/request.go b/pkg/tasksvc/request.go
@@ -0,0 +1,71 @@
+package tasksvc
+
+import (
+	"encoding/json"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"net/http"
+	"os"
+
+	"github.com/gemsorg/assignment/pkg/apierror"
+)
+
+type SearchParams struct {
+	Random bool
+}
+
+type SearchResult struct {
+	ID uint64 `json:"id"`
+}
+
+func Search(jobID uint64, authToken string, params SearchParams) (SearchResult, error) {
+	r := SearchResult{}
+	p := fmt.Sprintf("random=%t", params.Random)
+	route := fmt.Sprintf("jobs/%d/tasks/search?%s", jobID, p)
+
+	result, err := serviceRequest("GET", route, authToken, nil)
+	if err != nil {
+		return r, err
+	}
+
+	err = json.Unmarshal(result, &r)
+	if err != nil {
+		return r, err
+	}
+
+	if r.ID == 0 {
+		return r, NoAvailableTasks{JobID: jobID}
+	}
+
+	return r, nil
+}
+
+func serviceRequest(action, route, authToken string, reqBody io.Reader) ([]byte, error) {
+	client := &http.Client{}
+	serviceURL := fmt.Sprintf("%s/%s", os.Getenv("TASK_SVC_URL"), route)
+
+	req, err := http.NewRequest(action, serviceURL, reqBody)
+	if err != nil {
+		return nil, errorResponse(err)
+	}
+	// req.Header.Add("Authorization", authToken)
+	req.AddCookie(&http.Cookie{Name: "JWT", Value: authToken})
+
+	r, err := client.Do(req)
+	if err != nil {
+		return nil, errorResponse(err)
+	}
+
+	body, err := ioutil.ReadAll(r.Body)
+	// decoder := json.NewDecoder(r.Body)
+	// err = decoder.Decode(&body)
+	if err != nil {
+		return nil, errorResponse(err)
+	}
+	return body, nil
+}
+
+func errorResponse(err error) *apierror.APIError {
+	return apierror.New(500, err.Error(), err)
+}

Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,7 @@ type AuthData struct {`
`21`	`21`	`Expiration int64`
`22`	`22`	`Issuer string`
`23`	`23`	`UserID uint64`
	`24`	`+ Token string`
`24`	`25`	`}`
`25`	`26`
`26`	`27`	`func ParseAuthData(ctx context.Context) (AuthData, error) {`
`@@ -37,9 +38,10 @@ func ParseAuthData(ctx context.Context) (AuthData, error) {`
`37`	`38`	`}`
`38`	`39`
`39`	`40`	`return AuthData{`
`40`		`- int64(claims[ExpirationKey].(float64)),`
`41`		`- claims[IssuerKey].(string),`
`42`		`- uint64(claims[UserIDKey].(float64)),`
	`41`	`+ Expiration: int64(claims[ExpirationKey].(float64)),`
	`42`	`+ Issuer: claims[IssuerKey].(string),`
	`43`	`+ UserID: uint64(claims[UserIDKey].(float64)),`
	`44`	`+ Token: tokenString,`
`43`	`45`	`}, nil`
`44`	`46`	`}`
`45`	`47`
Original file line number	Diff line number	Diff line change
`@@ -43,7 +43,7 @@ func TestParseAuthData(t *testing.T) {`
`43`	`43`	`{`
`44`	`44`	`"it returns auth data",`
`45`	`45`	`args{ctx},`
`46`		`- AuthData{1591960106, "http://localhost:3000", 1},`
	`46`	`+ AuthData{1591960106, "http://localhost:3000", 1, "abc"},`
`47`	`47`	`false,`
`48`	`48`	`},`
`49`	`49`	`}`
Original file line number	Diff line number	Diff line change
`@@ -10,6 +10,7 @@ import (`
`10`	`10`	`type Authorizer interface {`
`11`	`11`	`SetAuthData(data authentication.AuthData)`
`12`	`12`	`IsModerator() (bool, error)`
	`13`	`+ GetAuthToken() string`
`13`	`14`	`}`
`14`	`15`
`15`	`16`	`type authorizor struct {`
`@@ -33,3 +34,7 @@ func (a *authorizor) IsModerator() (bool, error) {`
`33`	`34`	`}`
`34`	`35`	`return true, nil`
`35`	`36`	`}`
	`37`	`+`
	`38`	`+func (a *authorizor) GetAuthToken() string {`
	`39`	`+ return a.authData.Token`
	`40`	`+}`
Original file line number	Diff line number	Diff line change
`@@ -27,7 +27,7 @@ func TestNewAuthorizer(t *testing.T) {`
`27`	`27`	`}`
`28`	`28`
`29`	`29`	`func TestSetAuthData(t *testing.T) {`
`30`		`- authData := authentication.AuthData{1591960106, "http://localhost:3000", 8}`
	`30`	`+ authData := authentication.AuthData{1591960106, "http://localhost:3000", 8, "abc"}`
`31`	`31`	`type fields struct {`
`32`	`32`	`authData authentication.AuthData`
`33`	`33`	`}`