updatesubtask

Author: FoseFx

gen/dart/lib/services/tasks_svc/v1/task_svc.pb.dart

@@ -219,7 +219,6 @@ message AssignTaskResponse {
 message UnassignTaskRequest {
   string task_id = 1; // @gotags: validate:"uuid4"
   string user_id = 2; // @gotags: validate:"uuid4"
-  optional string consistency = 3; // no conflict detection, if not provided
 }
 
 message UnassignTaskResponse {

gen/dart/lib/services/tasks_svc/v1/task_svc.pbjson.dart

@@ -6,6 +6,7 @@ import (
 	pb "gen/services/tasks_svc/v1"
 	"github.com/google/uuid"
 	"hwes"
+	"hwutil"
 	taskEventsV1 "tasks-svc/internal/task/events/v1"
 	"tasks-svc/internal/task/models"
 	"time"
@@ -50,6 +51,17 @@ func LoadTaskAggregateWithSnapshotAt(ctx context.Context, as hwes.AggregateStore
 		}
 
 		task := *taskAggregate.Task // deref copies model
+
+		// also copy pointer values
+		if task.DueAt != nil {
+			task.DueAt = hwutil.PtrTo(*task.DueAt)
+		}
+		subtasks := make(map[uuid.UUID]models.Subtask)
+		for key, value := range task.Subtasks {
+			subtasks[key] = value
+		}
+		task.Subtasks = subtasks
+
 		snapshot = &task
 	}
 

gen/go/services/tasks_svc/v1/task_svc.pb.go

@@ -8,7 +8,6 @@ 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"
@@ -18,6 +17,7 @@ import (
 	"hwutil"
 	"strconv"
 	"tasks-svc/internal/task/handlers"
+	"tasks-svc/internal/task/models"
 	"tasks-svc/internal/util"
 	"time"
 )
@@ -204,7 +204,7 @@ func (s *TaskGrpcService) AssignTask(ctx context.Context, req *pb.AssignTaskRequ
 
 	for i := 0; true; i++ {
 		if i > 10 {
-			log.Warn().Msg("UpdatePatient: conflict circuit breaker triggered")
+			zlog.Ctx(ctx).Warn().Msg("UpdatePatient: conflict circuit breaker triggered")
 			return nil, fmt.Errorf("failed conflict resolution")
 		}
 
@@ -526,9 +526,66 @@ func (s *TaskGrpcService) UpdateSubtask(ctx context.Context, req *pb.UpdateSubta
 		return nil, err
 	}
 
-	consistency, err := s.handlers.Commands.V1.UpdateSubtask(ctx, taskID, subtaskID, req.Subtask.Name, req.Subtask.Done)
-	if err != nil {
-		return nil, err
+	expConsistency, ok := hwutil.ParseConsistency(req.TaskConsistency)
+	if !ok {
+		return nil, common.UnparsableConsistencyError(ctx, "task_consistency")
+	}
+
+	var consistency uint64
+
+	for i := 0; true; i++ {
+		if i > 10 {
+			zlog.Ctx(ctx).Warn().Msg("UpdateSubtask: conflict circuit breaker triggered")
+			return nil, fmt.Errorf("failed conflict resolution")
+		}
+
+		c, conflict, err := s.handlers.Commands.V1.UpdateSubtask(ctx, taskID, subtaskID, req.Subtask.Name, req.Subtask.Done, expConsistency)
+		if err != nil {
+			return nil, err
+		}
+		consistency = c
+
+		if conflict == nil {
+			break
+		}
+
+		conflicts := make(map[string]*commonpb.AttributeConflict)
+
+		var was models.Subtask
+		if w, ok := conflict.Was.Subtasks[subtaskID]; ok {
+			was = w
+		} else {
+			return nil, fmt.Errorf("subtask did not exist at expConsistency")
+		}
+
+		is := conflict.Is.Subtasks[subtaskID] // handler would have failed if non-existent
+
+		// TODO: find a generic approach
+		nameUpdateRequested := req.Subtask.Name != nil && *req.Subtask.Name != is.Name
+		nameAlreadyUpdated := was.Name != is.Name
+		if nameUpdateRequested && nameAlreadyUpdated {
+			conflicts["name"], err = util.AttributeConflict(
+				wrapperspb.String(is.Name),
+				wrapperspb.String(*req.Subtask.Name),
+			)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		if len(conflicts) != 0 {
+			return &pb.UpdateSubtaskResponse{
+				Conflict:        &commonpb.Conflict{ConflictingAttributes: conflicts},
+				TaskConsistency: strconv.FormatUint(conflict.Consistency, 10),
+			}, nil
+		}
+
+		// bool done can never cause a problem
+		// the user expects done = B, and sets it to \neg B
+		// so either that is the case still, or the update will do nothing anyway
+
+		// no conflict? retry with new consistency
+		expConsistency = &conflict.Consistency
 	}
 
 	return &pb.UpdateSubtaskResponse{

gen/ts/services/tasks_svc/v1/task_svc_pb.d.ts

@@ -5,26 +5,19 @@ import (
 	"github.com/google/uuid"
 	"hwes"
 	"tasks-svc/internal/task/aggregate"
-	"tasks-svc/internal/task/models"
 )
 
-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)
+type AssignTaskCommandHandler func(ctx context.Context, taskID, userID uuid.UUID, expConsistency *uint64) (uint64, *UpdateTaskConflict, error)
 
 func NewAssignTaskCommandHandler(as hwes.AggregateStore) AssignTaskCommandHandler {
-	return func(ctx context.Context, taskID, userID uuid.UUID, expConsistency *uint64) (uint64, *AssignTaskConflict, error) {
+	return func(ctx context.Context, taskID, userID uuid.UUID, expConsistency *uint64) (uint64, *UpdateTaskConflict, error) {
 		task, oldState, err := aggregate.LoadTaskAggregateWithSnapshotAt(ctx, as, taskID, expConsistency)
 		if err != nil {
 			return 0, nil, err
 		}
 
 		if expConsistency != nil && *expConsistency != task.GetVersion() {
-			return 0, &AssignTaskConflict{
+			return 0, &UpdateTaskConflict{
 				Consistency: task.GetVersion(),
 				Was:         oldState,
 				Is:          task.Task,

gen/ts/services/tasks_svc/v1/task_svc_pb.js

@@ -8,32 +8,41 @@ import (
 	"tasks-svc/internal/task/aggregate"
 )
 
-type UpdateSubtaskCommandHandler func(ctx context.Context, taskID, subtaskID uuid.UUID, name *string, done *bool) (uint64, error)
+type UpdateSubtaskCommandHandler func(ctx context.Context, taskID, subtaskID uuid.UUID, name *string, done *bool, expConsistency *uint64) (uint64, *UpdateTaskConflict, error)
 
 func NewUpdateSubtaskCommandHandler(as hwes.AggregateStore) UpdateSubtaskCommandHandler {
-	return func(ctx context.Context, taskID, subtaskID uuid.UUID, name *string, done *bool) (uint64, error) {
-		a, err := aggregate.LoadTaskAggregate(ctx, as, taskID)
+	return func(ctx context.Context, taskID, subtaskID uuid.UUID, name *string, done *bool, expConsistency *uint64) (uint64, *UpdateTaskConflict, error) {
+		a, oldState, err := aggregate.LoadTaskAggregateWithSnapshotAt(ctx, as, taskID, expConsistency)
 		if err != nil {
-			return 0, err
+			return 0, nil, err
 		}
 
 		currentSubtask, found := a.Task.Subtasks[subtaskID]
 		if !found {
-			return 0, fmt.Errorf("subtask with ID: %s not found on Task with ID: %s", subtaskID, taskID)
+			return 0, nil, fmt.Errorf("subtask with ID: %s not found on Task with ID: %s", subtaskID, taskID)
+		}
+
+		if expConsistency != nil && *expConsistency != a.GetVersion() {
+			return 0, &UpdateTaskConflict{
+				Consistency: a.GetVersion(),
+				Was:         oldState,
+				Is:          a.Task,
+			}, err
 		}
 
 		if name != nil && *name != currentSubtask.Name {
 			if err := a.UpdateSubtaskName(ctx, subtaskID, *name); err != nil {
-				return 0, err
+				return 0, nil, err
 			}
 		}
 
 		if done != nil && *done != currentSubtask.Done {
 			if err := a.UpdateSubtaskDone(ctx, subtaskID, *done); err != nil {
-				return 0, err
+				return 0, nil, err
 			}
 		}
 
-		return as.Save(ctx, a)
+		consistency, err := as.Save(ctx, a)
+		return consistency, nil, err
 	}
 }