UpdateTask

Author: FoseFx

services/tasks-svc/internal/task/aggregate/aggregate.go

@@ -39,6 +39,28 @@ func LoadTaskAggregate(ctx context.Context, as hwes.AggregateStore, id uuid.UUID
 	return taskAggregate, nil
 }
 
+func LoadTaskAggregateWithSnapshotAt(ctx context.Context, as hwes.AggregateStore, id uuid.UUID, pauseAt *uint64) (*TaskAggregate, *models.Task, error) {
+	taskAggregate := NewTaskAggregate(id)
+	var snapshot *models.Task
+
+	if pauseAt != nil {
+		//  load pauseAt+1-many events (version is 0-indexed)
+		if err := as.LoadN(ctx, taskAggregate, *pauseAt+1); err != nil {
+			return nil, nil, err
+		}
+
+		task := *taskAggregate.Task // deref copies model
+		snapshot = &task
+	}
+
+	// continue loading all other events
+	if err := as.Load(ctx, taskAggregate); err != nil {
+		return nil, nil, err
+	}
+
+	return taskAggregate, snapshot, nil
+}
+
 func (a *TaskAggregate) initEventListeners() {
 	a.
 		RegisterEventListener(taskEventsV1.TaskCreated, a.onTaskCreated).

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

@@ -3,15 +3,22 @@ package api
 import (
 	"common"
 	"context"
+	"fmt"
+	commonpb "gen/libs/common/v1"
 	pb "gen/services/tasks_svc/v1"
 	"github.com/google/uuid"
+	zlog "github.com/rs/zerolog"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/timestamppb"
+	"google.golang.org/protobuf/types/known/wrapperspb"
 	"hwes"
 	"hwutil"
 	"strconv"
 	"tasks-svc/internal/task/handlers"
+	"tasks-svc/internal/util"
+	"time"
 )
 
 type TaskGrpcService struct {
@@ -61,15 +68,114 @@ func (s *TaskGrpcService) CreateTask(ctx context.Context, req *pb.CreateTaskRequ
 	}, nil
 }
 
+func timeAlreadyUpdated(was, is *time.Time) bool {
+	if was != nil && is != nil {
+		return !was.Round(time.Second).Equal(is.Round(time.Second))
+	}
+
+	return true
+}
+
 func (s *TaskGrpcService) UpdateTask(ctx context.Context, req *pb.UpdateTaskRequest) (*pb.UpdateTaskResponse, error) {
+	log := zlog.Ctx(ctx)
+
 	taskID, err := uuid.Parse(req.GetId())
 	if err != nil {
 		return nil, err
 	}
 
-	consistency, err := s.handlers.Commands.V1.UpdateTask(ctx, taskID, req.Name, req.Description, req.Status, req.Public, req.DueAt)
-	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.UpdateTask(ctx,
+			taskID, req.Name, req.Description, req.Status, req.Public, req.DueAt, expConsistency)
+		if err != nil {
+			return nil, err
+		}
+		consistency = c
+
+		if conflict == nil {
+			break
+		}
+		conflicts := make(map[string]*commonpb.AttributeConflict)
+
+		// TODO: find a generic approach
+		nameUpdateRequested := req.Name != nil && *req.Name != conflict.Is.Name
+		nameAlreadyUpdated := conflict.Was.Name != conflict.Is.Name
+		if nameUpdateRequested && nameAlreadyUpdated {
+			conflicts["name"], err = util.AttributeConflict(
+				wrapperspb.String(conflict.Is.Name),
+				wrapperspb.String(*req.Name),
+			)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		descrUpdateRequested := req.Description != nil && *req.Description != conflict.Is.Description
+		descrAlreadyUpdated := conflict.Was.Description != conflict.Is.Description
+		if descrUpdateRequested && descrAlreadyUpdated {
+			conflicts["description"], err = util.AttributeConflict(
+				wrapperspb.String(conflict.Is.Description),
+				wrapperspb.String(*req.Description),
+			)
+			if err != nil {
+				return nil, fmt.Errorf("could not marshall description conflict: %w", err)
+			}
+		}
+
+		dueUpdateRequested := req.DueAt != nil &&
+			(conflict.Is.DueAt == nil || !req.DueAt.AsTime().Round(time.Second).Equal(conflict.Is.DueAt.Round(time.Second)))
+		dueAlreadyUpdated := timeAlreadyUpdated(conflict.Was.DueAt, conflict.Is.DueAt)
+		if dueUpdateRequested && dueAlreadyUpdated {
+			var is proto.Message = nil
+			if conflict.Is.DueAt != nil {
+				is = timestamppb.New(*conflict.Is.DueAt)
+			}
+			conflicts["due_at"], err = util.AttributeConflict(
+				is,
+				req.DueAt,
+			)
+			if err != nil {
+				return nil, fmt.Errorf("could not marshall due_at conflict: %w", err)
+			}
+		}
+
+		statusUpdateRequested := req.Status != nil && *req.Status != conflict.Is.Status
+		statusAlreadyUpdated := conflict.Was.Status != conflict.Is.Status
+		if statusUpdateRequested && statusAlreadyUpdated {
+			conflicts["status"], err = util.AttributeConflict(
+				wrapperspb.Int32(int32(conflict.Is.Status)),
+				wrapperspb.Int32(int32(*req.Status)),
+			)
+			if err != nil {
+				return nil, fmt.Errorf("could not marshall status conflict: %w", err)
+			}
+		}
+
+		// bool public can never cause a problem
+		// the user expects public = B, and sets it to \neg B
+		// so either that is the case still, or the update will do nothing anyway
+
+		if len(conflicts) != 0 {
+			return &pb.UpdateTaskResponse{
+				Conflict:    &commonpb.Conflict{ConflictingAttributes: conflicts},
+				Consistency: strconv.FormatUint(conflict.Consistency, 10),
+			}, nil
+		}
+
+		// no conflict? retry with new consistency
+		expConsistency = &conflict.Consistency
 	}
 
 	return &pb.UpdateTaskResponse{

services/tasks-svc/internal/task/commands/v1/update_task.go

@@ -7,8 +7,15 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 	"hwes"
 	"tasks-svc/internal/task/aggregate"
+	"tasks-svc/internal/task/models"
 )
 
+type UpdateTaskConflict struct {
+	Consistency uint64
+	Was         *models.Task
+	Is          *models.Task
+}
+
 type UpdateTaskCommandHandler func(
 	ctx context.Context,
 	taskID uuid.UUID,
@@ -17,45 +24,63 @@ type UpdateTaskCommandHandler func(
 	status *pb.TaskStatus,
 	public *bool,
 	dueAt *timestamppb.Timestamp,
-) (uint64, error)
+	expConsistency *uint64,
+) (uint64, *UpdateTaskConflict, error)
 
 func NewUpdateTaskCommandHandler(as hwes.AggregateStore) UpdateTaskCommandHandler {
-	return func(ctx context.Context, taskID uuid.UUID, name *string, description *string, status *pb.TaskStatus, public *bool, dueAt *timestamppb.Timestamp) (uint64, error) {
-		a, err := aggregate.LoadTaskAggregate(ctx, as, taskID)
+	return func(ctx context.Context,
+		taskID uuid.UUID,
+		name *string,
+		description *string,
+		status *pb.TaskStatus,
+		public *bool,
+		dueAt *timestamppb.Timestamp,
+		expConsistency *uint64,
+	) (uint64, *UpdateTaskConflict, error) {
+		a, oldState, err := aggregate.LoadTaskAggregateWithSnapshotAt(ctx, as, taskID, expConsistency)
 		if err != nil {
-			return 0, err
+			return 0, nil, err
+		}
+
+		if expConsistency != nil && *expConsistency != a.GetVersion() {
+			return 0, &UpdateTaskConflict{
+				Consistency: a.GetVersion(),
+				Was:         oldState,
+				Is:          a.Task,
+			}, nil
 		}
 
 		if name != nil {
 			if err := a.UpdateName(ctx, *name); err != nil {
-				return 0, err
+				return 0, nil, err
 			}
 		}
 
 		if description != nil {
 			if err := a.UpdateDescription(ctx, *description); err != nil {
-				return 0, err
+				return 0, nil, err
 			}
 		}
 
 		if status != nil {
 			if err := a.UpdateStatus(ctx, *status); err != nil {
-				return 0, err
+				return 0, nil, err
 			}
 		}
 
 		if public != nil && a.Task.Public != *public {
 			if err := a.UpdateTaskPublic(ctx, *public); err != nil {
-				return 0, err
+				return 0, nil, err
 			}
 		}
 
 		if dueAt != nil {
 			if err := a.UpdateDueAt(ctx, dueAt.AsTime()); err != nil {
-				return 0, err
+				return 0, nil, err
 			}
 		}
 
-		return as.Save(ctx, a)
+		consistency, err := as.Save(ctx, a)
+		return consistency, nil, err
 	}
 }

services/tasks-svc/stories/PatientCRUD_test.go

@@ -9,6 +9,7 @@ import (
 	"hwtesting"
 	"hwutil"
 	"strconv"
+	"strings"
 	"testing"
 	"time"
 )
@@ -692,6 +693,7 @@ func TestUpdatePatientConflict(t *testing.T) {
 	B := "B"
 	C := "C"
 
+	// Only testing HumanReadableIdentifier
 	testMatrix := []struct {
 		was            string
 		is             string
@@ -741,9 +743,9 @@ func TestUpdatePatientConflict(t *testing.T) {
 			if o.expectConflict {
 				conflict := updateRes.Conflict.ConflictingAttributes["human_readable_identifier"]
 				assert.NotNil(t, conflict)
-				exp := "is:{[type.googleapis.com/google.protobuf.StringValue]:{value:\"B\"}}  " +
+				exp := "is:{[type.googleapis.com/google.protobuf.StringValue]:{value:\"B\"}} " +
 					"want:{[type.googleapis.com/google.protobuf.StringValue]:{value:\"C\"}}"
-				assert.Equal(t, exp, conflict.String())
+				assert.Equal(t, exp, strings.Replace(conflict.String(), "  ", " ", 1))
 			}
 		})
 	}
@@ -830,10 +832,10 @@ func TestAssignBedConflict(t *testing.T) {
 
 				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 + "\"}}  " +
+					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, conflict.String())
+				assert.Equal(t, exp, strings.Replace(conflict.String(), "  ", " ", 1))
 			}
 		})
 	}