assign beds

Author: FoseFx

proto/libs/common/v1/conflict.proto

@@ -21,6 +21,6 @@ message Conflict {
 }
 
 message AttributeConflict {
-  google.protobuf.Any is = 1;
+  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;
 }

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

@@ -10,6 +10,7 @@ import (
 	zlog "github.com/rs/zerolog/log"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"google.golang.org/protobuf/proto"
 	"google.golang.org/protobuf/types/known/wrapperspb"
 	"hwdb"
 	"hwdb/locale"
@@ -394,11 +395,23 @@ func (s *PatientGrpcService) UpdatePatient(ctx context.Context, req *pb.UpdatePa
 		return nil, common.UnparsableConsistencyError(ctx, "consistency")
 	}
 
-	consistency, conflict, err := s.handlers.Commands.V1.UpdatePatient(ctx, patientID, expConsistency, req.HumanReadableIdentifier, req.Notes)
-	if err != nil {
-		return nil, err
-	}
-	if conflict != nil {
+	var consistency uint64
+
+	for i := 0; true; i++ {
+		if i > 10 {
+			zlog.Ctx(ctx).Warn().Msg("UpdatePatient: conflict circuit breaker triggered")
+			return nil, fmt.Errorf("failed conflict resolution")
+		}
+
+		c, conflict, err := s.handlers.Commands.V1.UpdatePatient(ctx, patientID, expConsistency, req.HumanReadableIdentifier, req.Notes)
+		consistency = c
+
+		if err != nil {
+			return nil, err
+		}
+		if conflict == nil {
+			break
+		}
 		conflicts := make(map[string]*commonpb.AttributeConflict)
 
 		// TODO: find a generic approach
@@ -432,6 +445,9 @@ func (s *PatientGrpcService) UpdatePatient(ctx context.Context, req *pb.UpdatePa
 				Consistency: strconv.FormatUint(conflict.Consistency, 10),
 			}, nil
 		}
+
+		// no conflict? retry with new consistency
+		expConsistency = &conflict.Consistency
 	}
 
 	tracking.AddPatientToRecentActivity(ctx, patientID.String())
@@ -457,17 +473,64 @@ func (s *PatientGrpcService) AssignBed(ctx context.Context, req *pb.AssignBedReq
 		return nil, err
 	}
 
-	consistency, err := s.handlers.Commands.V1.AssignBed(ctx, patientID, bedID)
-	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("AssignBed: conflict circuit breaker triggered")
+			return nil, fmt.Errorf("failed conflict resolution")
+		}
+
+		c, conflict, err := s.handlers.Commands.V1.AssignBed(ctx, patientID, bedID, expConsistency)
+		consistency = c
+
+		if err != nil {
+			return nil, err
+		}
+		if conflict == nil {
+			break
+		}
+		conflicts := make(map[string]*commonpb.AttributeConflict)
+
+		// TODO: find a generic approach
+		bedUpdateRequested := req.BedId != conflict.Is.BedID.UUID.String()
+		bedAlreadyUpdated := conflict.Was.BedID != conflict.Is.BedID
+		if bedUpdateRequested && bedAlreadyUpdated {
+			var is proto.Message = nil
+			if conflict.Is.BedID.Valid {
+				is = wrapperspb.String(conflict.Is.BedID.UUID.String())
+			}
+			conflicts["bed_id"], err = util.AttributeConflict(
+				is,
+				wrapperspb.String(req.BedId),
+			)
+			if err != nil {
+				return nil, err
+			}
+		}
+
+		if len(conflicts) != 0 {
+			return &pb.AssignBedResponse{
+				Conflict:    &commonpb.Conflict{ConflictingAttributes: conflicts},
+				Consistency: strconv.FormatUint(conflict.Consistency, 10),
+			}, nil
+		}
+
+		// no conflict? retry with new consistency
+		expConsistency = &conflict.Consistency
 	}
 
 	log.Info().Str("patientID", patientID.String()).Str("bedID", bedID.String()).Msg("assigned bed to patient")
 
 	tracking.AddWardToRecentActivity(ctx, patientID.String())
 
 	return &pb.AssignBedResponse{
-		Conflict:    nil, // TODO
+		Conflict:    nil,
 		Consistency: strconv.FormatUint(consistency, 10),
 	}, nil
 }

services/tasks-svc/internal/patient/commands/v1/assign_bed.go

@@ -5,20 +5,38 @@ import (
 	"github.com/google/uuid"
 	"hwes"
 	"tasks-svc/internal/patient/aggregate"
+	"tasks-svc/internal/patient/models"
 )
 
-type AssignBedCommandHandler func(ctx context.Context, patientID uuid.UUID, bedID uuid.UUID) (uint64, error)
+type AssignBedConflict struct {
+	Consistency uint64
+	Was         *models.Patient
+	Is          *models.Patient
+}
+
+type AssignBedCommandHandler func(ctx context.Context, patientID uuid.UUID, bedID uuid.UUID, expConsistency *uint64) (uint64, *AssignBedConflict, error)
 
 func NewAssignBedCommandHandler(as hwes.AggregateStore) AssignBedCommandHandler {
-	return func(ctx context.Context, patientID uuid.UUID, bedID uuid.UUID) (uint64, error) {
-		a, err := aggregate.LoadPatientAggregate(ctx, as, patientID)
+	return func(ctx context.Context, patientID uuid.UUID, bedID uuid.UUID, expConsistency *uint64) (uint64, *AssignBedConflict, error) {
+		a, oldState, err := aggregate.LoadPatientAggregateWithSnapshotAt(ctx, as, patientID, expConsistency)
 		if err != nil {
-			return 0, err
+			return 0, nil, err
+		}
+
+		// update happened since
+		if expConsistency != nil && *expConsistency != a.GetVersion() {
+			return 0, &AssignBedConflict{
+				Consistency: a.GetVersion(),
+				Was:         oldState,
+				Is:          a.Patient,
+			}, nil
 		}
 
 		if err := a.AssignBed(ctx, bedID); 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/internal/util/conflict.go

@@ -10,13 +10,22 @@ import (
 // AttributeConflict is a constructor for commonpb.AttributeConflicts
 // I'd love to move this somewhere else, but I also don't want common to depend on gen (and thus hwdb, hwes, ...)
 func AttributeConflict(is, want proto.Message) (*commonpb.AttributeConflict, error) {
-	wantAny, err := anypb.New(want)
-	if err != nil {
-		return nil, fmt.Errorf("AttributeConflict could not marshal want: %w", err)
+	var err error
+
+	var wantAny *anypb.Any
+	if want != nil {
+		wantAny, err = anypb.New(want)
+		if err != nil {
+			return nil, fmt.Errorf("AttributeConflict could not marshal want: %w", err)
+		}
 	}
-	isAny, err := anypb.New(is)
-	if err != nil {
-		return nil, fmt.Errorf("AttributeConflict could not marshal is: %w", err)
+
+	var isAny *anypb.Any
+	if is != nil {
+		isAny, err = anypb.New(is)
+		if err != nil {
+			return nil, fmt.Errorf("AttributeConflict could not marshal is: %w", err)
+		}
 	}
 
 	return &commonpb.AttributeConflict{

services/tasks-svc/stories/PatientCRUD_test.go

@@ -738,6 +738,103 @@ func TestUpdatePatientConflict(t *testing.T) {
 
 			// EXPECT
 			assert.Equal(t, o.expectConflict, updateRes.Conflict != nil)
+			if o.expectConflict {
+				conflict := updateRes.Conflict.ConflictingAttributes["human_readable_identifier"]
+				assert.NotNil(t, conflict)
+				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())
+			}
+		})
+	}
+}
+
+func TestAssignBedConflict(t *testing.T) {
+	ctx := context.Background()
+	patientClient := patientServiceClient()
+
+	wardId, _ := prepareWard(t, ctx, "")
+	roomId, _ := prepareRoom(t, ctx, wardId, "")
+
+	A, _ := prepareBed(t, ctx, roomId, "A")
+	B, _ := prepareBed(t, ctx, roomId, "B")
+	C, _ := prepareBed(t, ctx, roomId, "C")
+
+	testMatrix := []struct {
+		was            string
+		is             *string
+		want           string
+		expectConflict bool
+	}{
+		{A, &B, B, false},
+		{A, &B, C, true},
+		{A, &A, C, false},
+		{A, nil, C, true},
+	}
+
+	for i, o := range testMatrix {
+		t.Run(t.Name()+"_"+strconv.Itoa(i), func(t *testing.T) {
+			// WAS
+			patientRes, err := patientClient.CreatePatient(ctx, &pb.CreatePatientRequest{
+				HumanReadableIdentifier: t.Name(),
+				Notes:                   hwutil.PtrTo("A patient for test " + t.Name()),
+			})
+			assert.NoError(t, err)
+			time.Sleep(time.Millisecond * 100)
+
+			initialAssignment, err := patientClient.AssignBed(ctx, &pb.AssignBedRequest{
+				Id:          patientRes.Id,
+				BedId:       o.was,
+				Consistency: &patientRes.Consistency,
+			})
+			assert.NoError(t, err)
+			assert.Nil(t, initialAssignment.Conflict)
+
+			id := patientRes.Id
+			initialConsistency := initialAssignment.Consistency
+
+			time.Sleep(time.Millisecond * 100)
+
+			// IS
+			if o.is != nil {
+				a, err := patientClient.AssignBed(ctx, &pb.AssignBedRequest{
+					Id:          id,
+					BedId:       *o.is,
+					Consistency: &initialConsistency,
+				})
+				assert.NoError(t, err)
+				assert.Nil(t, a.Conflict)
+			} else {
+				u, err := patientClient.UnassignBed(ctx, &pb.UnassignBedRequest{
+					Id:          id,
+					Consistency: &initialConsistency,
+				})
+				assert.NoError(t, err)
+				assert.Nil(t, u.Conflict)
+			}
+			time.Sleep(time.Millisecond * 100)
+
+			// WANT
+			updateRes, err := patientClient.AssignBed(ctx, &pb.AssignBedRequest{
+				Id:          id,
+				BedId:       o.want,
+				Consistency: &initialConsistency,
+			})
+			assert.NoError(t, err)
+
+			// EXPECT
+			assert.Equal(t, o.expectConflict, updateRes.Conflict != nil)
+			if o.expectConflict {
+				conflict := updateRes.Conflict.ConflictingAttributes["bed_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, conflict.String())
+			}
 		})
 	}
 }