update patient

Author: FoseFx

libs/hwes/aggregate_store.go

@@ -9,6 +9,9 @@ type AggregateStore interface {
 	// Load populates the aggregate to the recent version of the persisted events
 	Load(ctx context.Context, aggregate Aggregate) error
 
+	// LoadN populates the aggregate by N-many persisted events
+	LoadN(ctx context.Context, aggregate Aggregate, N uint64) error
+
 	// Save persists all uncommitted events of the aggregate, returns consistency
 	Save(ctx context.Context, aggregate Aggregate) (uint64, error)
 

libs/hwes/eventstoredb/aggregate_store.go

@@ -89,39 +89,62 @@ func (a *AggregateStore) doSave(ctx context.Context, aggregate hwes.Aggregate, g
 
 // Implements AggregateStore interface
 
-func (a *AggregateStore) Load(ctx context.Context, aggregate hwes.Aggregate) error {
-	stream, err := a.es.ReadStream(ctx, aggregate.GetTypeID(), esdb.ReadStreamOptions{}, math.MaxUint64) // MaxUint64 for "all" events
+func (a *AggregateStore) LoadN(ctx context.Context, aggregate hwes.Aggregate, N uint64) error {
+	if N == 0 {
+		return nil
+	}
+
+	// start from the beginning, unless we have already processed events
+	var from esdb.StreamPosition = esdb.Start{}
+	aE := aggregate.GetAppliedEvents()
+	if len(aE) != 0 {
+		// We cant just use the aggreagate's Version here, as we are unable to
+		// discriminate aggregates where the first event was processed already (version = 0)
+		// from those aggregates where no event was processed yet (version = 0 by default)
+		from = esdb.Revision(aE[len(aE)-1].Version + 1)
+	}
+
+	// open a read stream
+	stream, err := a.es.ReadStream(ctx, aggregate.GetTypeID(), esdb.ReadStreamOptions{
+		Direction: esdb.Forwards,
+		From:      from,
+	}, N)
 	if err != nil {
-		return fmt.Errorf("AggregateStore.Load: could not open stream: %w", err)
+		return fmt.Errorf("AggregateStore.LoadN: could not open stream: %w", err)
 	}
 	defer stream.Close()
 
+	// stream events in
 	for {
 		esdbEvent, err := stream.Recv()
 		if errors.Is(err, io.EOF) {
-			// exit condition for for-loop
+			// all events were handled!
 			break
 		} else if err != nil {
-			return fmt.Errorf("AggregateStore.Load: could not read from stream: %w", err)
+			return fmt.Errorf("AggregateStore.LoadN: could not read from stream: %w", err)
 		}
 
 		event, err := hwes.NewEventFromRecordedEvent(esdbEvent.Event)
 		if err != nil {
-			return fmt.Errorf("AggregateStore.Load: %w", err)
+			return fmt.Errorf("AggregateStore.LoadN: %w", err)
 		}
 
 		if err := aggregate.Progress(event); err != nil {
-			return fmt.Errorf("AggregateStore.Load: Progress failed: %w", err)
+			return fmt.Errorf("AggregateStore.LoadN: Progress failed: %w", err)
 		}
 	}
 
 	if aggregate.IsDeleted() {
-		return fmt.Errorf("AggregateStore.Load: aggregate has been marked as deleted")
+		return fmt.Errorf("AggregateStore.LoadN: aggregate has been marked as deleted")
 	}
 
 	return nil
 }
 
+func (a *AggregateStore) Load(ctx context.Context, aggregate hwes.Aggregate) error {
+	return a.LoadN(ctx, aggregate, math.MaxUint64) // MaxUint64 -> all events
+}
+
 func (a *AggregateStore) Save(ctx context.Context, aggregate hwes.Aggregate) (uint64, error) {
 	// We can switch out the getExpectedRevision strategy for testing optimistic concurrency.
 	// It is not intended to switch the strategy in production.

libs/hwes/eventstoredb/integration_test.go

@@ -0,0 +1,89 @@
+package eventstoredb
+
+import (
+	"context"
+	"fmt"
+	"github.com/google/uuid"
+	zlog "github.com/rs/zerolog/log"
+	"github.com/stretchr/testify/assert"
+	"hwes"
+	"hwtesting"
+	"os"
+	"testing"
+	"time"
+)
+
+var endpoint string
+
+func TestMain(m *testing.M) {
+	ctx, cancel := context.WithCancel(context.Background())
+
+	zlog.Info().Msg("starting containers")
+	endpoints, teardownContainers := hwtesting.StartContainers(ctx, hwtesting.Eventstore)
+
+	endpoint = fmt.Sprintf("esdb://%s:%s@%s?tls=false",
+		hwtesting.EsUser, hwtesting.EsPassword, endpoints.Get(hwtesting.Eventstore))
+
+	exit := m.Run()
+
+	teardownContainers()
+	cancel()
+	os.Exit(exit)
+}
+
+func TestLoadN(t *testing.T) {
+	zlog.Info().Str("endpoint", endpoint).Msg(t.Name())
+	client := SetupEventStore(endpoint)
+	as := NewAggregateStore(client)
+	ctx := context.Background()
+
+	id := uuid.New()
+	eventType := t.Name()
+
+	sendAggregate := hwes.NewAggregateBase(hwes.AggregateType(eventType), id)
+	sendAggregate.RegisterEventListener(eventType, func(evt hwes.Event) error { return nil })
+
+	for i := 0; i < 4; i++ {
+		event, _ := hwes.NewEvent(sendAggregate, eventType, hwes.WithContext(ctx))
+		assert.NoError(t, sendAggregate.Apply(event))
+	}
+
+	_, err := as.Save(ctx, sendAggregate)
+	assert.NoError(t, err)
+
+	time.Sleep(time.Millisecond * 100)
+
+	rxAggregate := hwes.NewAggregateBase(hwes.AggregateType(eventType), id)
+	rxAggregate.RegisterEventListener(eventType, func(evt hwes.Event) error { return nil })
+	assert.Len(t, sendAggregate.GetAppliedEvents(), 0)
+
+	// LoadN 0 does nothing
+	err = as.LoadN(ctx, rxAggregate, 0)
+	assert.NoError(t, err)
+	assert.Len(t, rxAggregate.GetAppliedEvents(), 0)
+
+	// LoadN 1 loads the first event
+	err = as.LoadN(ctx, rxAggregate, 1)
+	assert.NoError(t, err)
+	assert.Len(t, rxAggregate.GetAppliedEvents(), 1)
+	assert.Equal(t, rxAggregate.GetAppliedEvents()[0].Version, uint64(0))
+
+	// LoadN 1 loads another event
+	err = as.LoadN(ctx, rxAggregate, 1)
+	assert.NoError(t, err)
+	assert.Len(t, rxAggregate.GetAppliedEvents(), 2)
+	assert.Equal(t, rxAggregate.GetAppliedEvents()[1].Version, uint64(1))
+
+	// LoadN 2 loads the rest
+	err = as.LoadN(ctx, rxAggregate, 2)
+	assert.NoError(t, err)
+	assert.Len(t, rxAggregate.GetAppliedEvents(), 4)
+	assert.Equal(t, rxAggregate.GetAppliedEvents()[2].Version, uint64(2))
+	assert.Equal(t, rxAggregate.GetAppliedEvents()[3].Version, uint64(3))
+
+	// LoadN 1 now does nothing anymore
+	err = as.LoadN(ctx, rxAggregate, 1)
+	assert.NoError(t, err)
+	assert.Len(t, rxAggregate.GetAppliedEvents(), 4)
+
+}

libs/hwes/go.mod

@@ -5,6 +5,7 @@ go 1.23
 replace (
 	common => ../common
 	hwlocale => ../hwlocale
+	hwtesting => ../hwtesting
 	hwutil => ../hwutil
 	telemetry => ../telemetry
 )
@@ -15,45 +16,87 @@ require (
 	github.com/google/uuid v1.6.0
 	github.com/rs/zerolog v1.33.0
 	github.com/stretchr/testify v1.9.0
+	hwtesting v0.0.0
 	hwutil v0.0.0
 	telemetry v0.0.0
 )
 
 require (
+	dario.cat/mergo v1.0.0 // indirect
+	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
 	github.com/BurntSushi/toml v1.4.0 // indirect
+	github.com/Mariscal6/testcontainers-spicedb-go v0.1.0 // indirect
+	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/containerd/log v0.1.0 // indirect
+	github.com/containerd/platforms v0.2.1 // indirect
 	github.com/coreos/go-oidc v2.2.1+incompatible // indirect
+	github.com/cpuguy83/dockercfg v0.3.1 // indirect
 	github.com/dapr/dapr v1.14.2 // indirect
 	github.com/dapr/go-sdk v1.11.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/distribution/reference v0.6.0 // indirect
+	github.com/docker/docker v27.2.0+incompatible // indirect
+	github.com/docker/go-connections v0.5.0 // indirect
+	github.com/docker/go-units v0.5.0 // indirect
 	github.com/fatih/structs v1.1.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
 	github.com/gabriel-vasile/mimetype v1.4.3 // indirect
 	github.com/go-logr/logr v1.4.2 // indirect
 	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-ole/go-ole v1.2.6 // indirect
 	github.com/go-playground/locales v0.14.1 // indirect
 	github.com/go-playground/universal-translator v0.18.1 // indirect
 	github.com/go-playground/validator/v10 v10.22.1 // indirect
+	github.com/gogo/protobuf v1.3.2 // indirect
+	github.com/golang-migrate/migrate/v4 v4.18.1 // indirect
 	github.com/golang/protobuf v1.5.4 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus v1.0.1 // indirect
 	github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.1.0 // indirect
 	github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
 	github.com/joho/godotenv v1.5.1 // indirect
 	github.com/klauspost/compress v1.17.9 // indirect
 	github.com/leodido/go-urn v1.4.0 // indirect
+	github.com/lib/pq v1.10.9 // indirect
+	github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect
+	github.com/magiconair/properties v1.8.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/moby/docker-image-spec v1.3.1 // indirect
+	github.com/moby/patternmatcher v0.6.0 // indirect
+	github.com/moby/sys/sequential v0.5.0 // indirect
+	github.com/moby/sys/user v0.1.0 // indirect
+	github.com/moby/sys/userns v0.1.0 // indirect
+	github.com/moby/term v0.5.0 // indirect
+	github.com/morikuni/aec v1.0.0 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
 	github.com/nicksnyder/go-i18n/v2 v2.4.0 // indirect
+	github.com/opencontainers/go-digest v1.0.0 // indirect
+	github.com/opencontainers/image-spec v1.1.0 // indirect
 	github.com/openzipkin/zipkin-go v0.4.3 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
+	github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
 	github.com/pquerna/cachecontrol v0.1.0 // indirect
 	github.com/prometheus/client_golang v1.20.3 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
 	github.com/prometheus/common v0.55.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
+	github.com/shirou/gopsutil/v3 v3.23.12 // indirect
+	github.com/shoenig/go-m1cpu v0.1.6 // indirect
+	github.com/sirupsen/logrus v1.9.3 // indirect
+	github.com/testcontainers/testcontainers-go v0.33.0 // indirect
+	github.com/testcontainers/testcontainers-go/modules/postgres v0.33.0 // indirect
+	github.com/testcontainers/testcontainers-go/modules/redis v0.33.0 // indirect
+	github.com/tklauser/go-sysconf v0.3.12 // indirect
+	github.com/tklauser/numcpus v0.6.1 // indirect
+	github.com/yusufpapurcu/wmi v1.2.3 // indirect
 	go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.54.0 // indirect
+	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.54.0 // indirect
 	go.opentelemetry.io/otel v1.30.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.30.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.30.0 // indirect
@@ -63,6 +106,7 @@ require (
 	go.opentelemetry.io/otel/sdk v1.30.0 // indirect
 	go.opentelemetry.io/otel/trace v1.30.0 // indirect
 	go.opentelemetry.io/proto/otlp v1.3.1 // indirect
+	go.uber.org/atomic v1.10.0 // indirect
 	golang.org/x/crypto v0.27.0 // indirect
 	golang.org/x/net v0.29.0 // indirect
 	golang.org/x/oauth2 v0.23.0 // indirect