assigntask

Author: FoseFx

gen/dart/lib/libs/common/v1/conflict.pb.dart

@@ -21,6 +21,10 @@ message Conflict {
 }
 
 message AttributeConflict {
-  google.protobuf.Any is = 1; // CAUTION: may be missing, if the is underlying value is missing (e.g., unassigned beds)
-  google.protobuf.Any want = 2; // CAUTION: may be missing, if the requested value is missing (e.g., unassignment of a bed)
+  // CAUTION: may be missing, if the is underlying value is missing (e.g., unassigned beds)
+  // Enums are returned as Int32s
+  google.protobuf.Any is = 1;
+  // CAUTION: may be missing, if the requested value is missing (e.g., unassignment of a bed)
+  // Enums are returned as Int32s
+  google.protobuf.Any want = 2;
 }

gen/go/libs/common/v1/conflict.pb.go

@@ -8,6 +8,7 @@ import (
 	pb "gen/services/tasks_svc/v1"
 	"github.com/google/uuid"
 	zlog "github.com/rs/zerolog"
+	"github.com/rs/zerolog/log"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
 	"google.golang.org/protobuf/proto"
@@ -194,9 +195,55 @@ func (s *TaskGrpcService) AssignTask(ctx context.Context, req *pb.AssignTaskRequ
 		return nil, err
 	}
 
-	consistency, err := s.handlers.Commands.V1.AssignTask(ctx, taskID, userID)
-	if err != nil {
-		return nil, err
+	expConsistency, ok := hwutil.ParseConsistency(req.Consistency)
+	if !ok {
+		return nil, common.UnparsableConsistencyError(ctx, "consistency")
+	}
+
+	var consistency uint64
+
+	for i := 0; true; i++ {
+		if i > 10 {
+			log.Warn().Msg("UpdatePatient: conflict circuit breaker triggered")
+			return nil, fmt.Errorf("failed conflict resolution")
+		}
+
+		c, conflict, err := s.handlers.Commands.V1.AssignTask(ctx, taskID, userID, expConsistency)
+		if err != nil {
+			return nil, err
+		}
+		consistency = c
+		if conflict == nil {
+			break
+		}
+		conflicts := make(map[string]*commonpb.AttributeConflict)
+
+		// TODO: find a generic approach
+		userUpdateRequested := req.UserId != conflict.Is.AssignedUser.UUID.String()
+		userAlreadyUpdated := conflict.Was.AssignedUser != conflict.Is.AssignedUser
+		if userUpdateRequested && userAlreadyUpdated {
+			var is proto.Message = nil
+			if conflict.Is.AssignedUser.Valid {
+				is = wrapperspb.String(conflict.Is.AssignedUser.UUID.String())
+			}
+			conflicts["user_id"], err = util.AttributeConflict(
+				is,
+				wrapperspb.String(req.UserId),
+			)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		if len(conflicts) != 0 {
+			return &pb.AssignTaskResponse{
+				Conflict:    &commonpb.Conflict{ConflictingAttributes: conflicts},
+				Consistency: strconv.FormatUint(conflict.Consistency, 10),
+			}, nil
+		}
+
+		// no conflict? retry with new consistency
+		expConsistency = &conflict.Consistency
 	}
 
 	return &pb.AssignTaskResponse{

proto/libs/common/v1/conflict.proto

@@ -5,22 +5,38 @@ import (
 	"github.com/google/uuid"
 	"hwes"
 	"tasks-svc/internal/task/aggregate"
+	"tasks-svc/internal/task/models"
 )
 
-type AssignTaskCommandHandler func(ctx context.Context, taskID, userID uuid.UUID) (uint64, error)
+type AssignTaskConflict struct {
+	Consistency uint64
+	Was         *models.Task
+	Is          *models.Task
+}
+
+type AssignTaskCommandHandler func(ctx context.Context, taskID, userID uuid.UUID, expConsistency *uint64) (uint64, *AssignTaskConflict, error)
 
 func NewAssignTaskCommandHandler(as hwes.AggregateStore) AssignTaskCommandHandler {
-	return func(ctx context.Context, taskID, userID uuid.UUID) (uint64, error) {
-		task, err := aggregate.LoadTaskAggregate(ctx, as, taskID)
+	return func(ctx context.Context, taskID, userID uuid.UUID, expConsistency *uint64) (uint64, *AssignTaskConflict, error) {
+		task, oldState, err := aggregate.LoadTaskAggregateWithSnapshotAt(ctx, as, taskID, expConsistency)
 		if err != nil {
-			return 0, err
+			return 0, nil, err
+		}
+
+		if expConsistency != nil && *expConsistency != task.GetVersion() {
+			return 0, &AssignTaskConflict{
+				Consistency: task.GetVersion(),
+				Was:         oldState,
+				Is:          task.Task,
+			}, nil
 		}
 
 		// TODO: Handle SelfAssignTask when common.GetUserID() is testable
 		if err := task.AssignTask(ctx, userID); err != nil {
-			return 0, err
+			return 0, nil, err
 		}
 
-		return as.Save(ctx, task)
+		consistency, err := as.Save(ctx, task)
+		return consistency, nil, err
 	}
 }

services/tasks-svc/internal/task/api/grpc.go

@@ -3,6 +3,7 @@ package stories
 import (
 	"context"
 	pb "gen/services/tasks_svc/v1"
+	"github.com/google/uuid"
 	"github.com/stretchr/testify/assert"
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"math"
@@ -22,7 +23,6 @@ func TestUpdateTaskConflict_Name(t *testing.T) {
 	B := "B"
 	C := "C"
 
-	// Only testing HumanReadableIdentifier
 	testMatrix := []struct {
 		was            string
 		is             string
@@ -92,7 +92,6 @@ func TestUpdateTaskConflict_Description(t *testing.T) {
 	B := "B"
 	C := "C"
 
-	// Only testing HumanReadableIdentifier
 	testMatrix := []struct {
 		was            string
 		is             string
@@ -162,7 +161,6 @@ func TestUpdateTaskConflict_DueAt(t *testing.T) {
 	B := timestamppb.New(time.Now().Add(time.Hour))
 	C := timestamppb.New(time.Now().Add(time.Hour * 2))
 
-	// Only testing HumanReadableIdentifier
 	testMatrix := []struct {
 		was            *timestamppb.Timestamp
 		is             *timestamppb.Timestamp
@@ -237,7 +235,6 @@ func TestUpdateTaskConflict_Status(t *testing.T) {
 	B := pb.TaskStatus_TASK_STATUS_IN_PROGRESS
 	C := pb.TaskStatus_TASK_STATUS_DONE
 
-	// Only testing HumanReadableIdentifier
 	testMatrix := []struct {
 		was            pb.TaskStatus
 		is             *pb.TaskStatus
@@ -300,3 +297,87 @@ func TestUpdateTaskConflict_Status(t *testing.T) {
 		})
 	}
 }
+
+func TestAssignTaskConflict(t *testing.T) {
+	ctx := context.Background()
+	taskClient := taskServiceClient()
+
+	patientId := preparePatient(t, ctx, "")
+
+	A := uuid.New().String()
+	B := uuid.New().String()
+	C := uuid.New().String()
+
+	testMatrix := []struct {
+		was            *string
+		is             *string
+		want           string
+		expectConflict bool
+	}{
+		{&A, &B, B, false},
+		{&A, &B, C, true},
+		{&A, &A, C, false},
+		{nil, &A, C, true},
+		{&A, nil, C, true},
+	}
+
+	for i, o := range testMatrix {
+		t.Run(t.Name()+"_"+strconv.Itoa(i), func(t *testing.T) {
+			// WAS
+			task, err := taskClient.CreateTask(ctx, &pb.CreateTaskRequest{
+				Name:           t.Name(),
+				Description:    nil,
+				PatientId:      patientId,
+				Public:         nil,
+				DueAt:          nil,
+				InitialStatus:  nil,
+				AssignedUserId: o.was,
+				Subtasks:       nil,
+			})
+			assert.NoError(t, err)
+
+			id := task.Id
+			initialConsistency := task.Consistency
+
+			// IS
+			if o.is != nil {
+				a, err := taskClient.AssignTask(ctx, &pb.AssignTaskRequest{
+					TaskId:      id,
+					UserId:      *o.is,
+					Consistency: &initialConsistency,
+				})
+				assert.NoError(t, err)
+				assert.Nil(t, a.Conflict)
+			} else {
+				a, err := taskClient.UnassignTask(ctx, &pb.UnassignTaskRequest{
+					TaskId:      id,
+					UserId:      *o.was,
+					Consistency: &initialConsistency,
+				})
+				assert.NoError(t, err)
+				assert.Nil(t, a.Conflict)
+			}
+
+			// WANT
+			updateRes, err := taskClient.AssignTask(ctx, &pb.AssignTaskRequest{
+				TaskId:      id,
+				UserId:      o.want,
+				Consistency: &initialConsistency,
+			})
+			assert.NoError(t, err)
+
+			// EXPECT
+			assert.Equal(t, o.expectConflict, updateRes.Conflict != nil)
+			if o.expectConflict {
+				conflict := updateRes.Conflict.ConflictingAttributes["user_id"]
+				assert.NotNil(t, conflict)
+				exp := "want:{[type.googleapis.com/google.protobuf.StringValue]:{value:\"" + o.want + "\"}}"
+				if o.is != nil {
+					exp = "is:{[type.googleapis.com/google.protobuf.StringValue]:{value:\"" + *o.is + "\"}} " +
+						"want:{[type.googleapis.com/google.protobuf.StringValue]:{value:\"" + o.want + "\"}}"
+				}
+				assert.Equal(t, exp, strings.Replace(conflict.String(), "  ", " ", 1))
+			}
+		})
+	}
+}