feat: esdb version change & projection template code

.github/workflows/go-test.yml

@@ -0,0 +1,17 @@
+name: Go-test
+on: [push, pull_request]
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+
+    steps:
+      - uses: actions/checkout@v4
+      - name: Setup Go
+        uses: actions/setup-go@v4
+        with:
+          go-version: '1.21.x'
+      - name: Build
+        run: go build -v ./...
+      - name: Test with the Go CLI
+        run: go test ./... 

.github/workflows/release.yml

@@ -0,0 +1,29 @@
+name: goreleaser
+
+on:
+  push:
+    tags:
+      - "v*.*.*"
+
+permissions:
+  contents: write
+
+jobs:
+  goreleaser:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - name: Set up Go
+        uses: actions/setup-go@v4
+        with:
+          go-version: '1.21.1'
+      -
+        name: Run GoReleaser
+        uses: goreleaser/goreleaser-action@v5.0.0
+        with:
+          distribution: goreleaser
+          version: ${{ env.GITHUB_REF_NAME }}
+          args: release --clean
+          workdir: ./
+        env:
+          GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

docker-compose.yml

@@ -1,28 +1,25 @@
-version: "3.8"
+version: "3.4"
 
 services:
   eventstore.db:
-    image: eventstore/eventstore:22.10.3-buster-slim
+    image: eventstore/eventstore:24.2.0-alpha-arm64v8
     environment:
       - EVENTSTORE_CLUSTER_SIZE=1
       - EVENTSTORE_RUN_PROJECTIONS=All
       - EVENTSTORE_START_STANDARD_PROJECTIONS=true
-      - EVENTSTORE_EXT_TCP_PORT=1113
       - EVENTSTORE_HTTP_PORT=2113
       - EVENTSTORE_INSECURE=true
-      - EVENTSTORE_ENABLE_EXTERNAL_TCP=true
       - EVENTSTORE_ENABLE_ATOM_PUB_OVER_HTTP=true
     ports:
-      - "1113:1113"
       - "2113:2113"
-    #volumes:
-    #  - type: volume
-    #    source: eventsourcing-volume-data
-    #    target: /var/lib/eventsourcing
-    #  - type: volume
-    #    source: eventsourcing-volume-logs
-    #    target: /var/log/eventsourcing
-
-#volumes:
-#  eventstore-volume-data:
-#  eventstore-volume-logs:
+      - "1113:1113"
+    volumes:
+      - type: volume
+        source: eventstore-volume-data
+        target: /var/lib/eventstore
+      - type: volume
+        source: eventstore-volume-logs
+        target: /var/log/eventstore
+volumes:
+  eventstore-volume-data:
+  eventstore-volume-logs:

go.mod

@@ -3,32 +3,35 @@ module secap-input
 go 1.22.1
 
 require (
-	github.com/EventStore/EventStore-Client-Go v1.0.2
+	github.com/EventStore/EventStore-Client-Go/v4 v4.0.0
 	github.com/gofiber/fiber/v2 v2.52.4
 	github.com/gofrs/uuid v4.4.0+incompatible
+	github.com/google/uuid v1.6.0
 	github.com/joho/godotenv v1.5.1
 	github.com/json-iterator/go v1.1.12
+	github.com/pkg/errors v0.9.1
+	golang.org/x/sync v0.6.0
 )
 
 require (
 	github.com/andybalholm/brotli v1.1.0 // indirect
-	github.com/golang/mock v1.6.0 // indirect
-	github.com/golang/protobuf v1.5.0 // indirect
-	github.com/google/uuid v1.6.0 // indirect
+	github.com/golang/protobuf v1.5.3 // indirect
 	github.com/klauspost/compress v1.17.7 // indirect
 	github.com/mattn/go-colorable v0.1.13 // indirect
 	github.com/mattn/go-isatty v0.0.20 // indirect
 	github.com/mattn/go-runewidth v0.0.15 // indirect
-	github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/rivo/uniseg v0.4.7 // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
 	github.com/valyala/fasthttp v1.52.0 // indirect
 	github.com/valyala/tcplisten v1.0.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
 	golang.org/x/net v0.21.0 // indirect
 	golang.org/x/sys v0.18.0 // indirect
 	golang.org/x/text v0.14.0 // indirect
-	google.golang.org/genproto v0.0.0-20200815001618-f69a88009b70 // indirect
-	google.golang.org/grpc v1.35.0 // indirect
-	google.golang.org/protobuf v1.27.1 // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20240227224415-6ceb2ff114de // indirect
+	google.golang.org/grpc v1.61.0 // indirect
+	google.golang.org/protobuf v1.32.0 // indirect
 )

internal/common/esdb/esdb.go

@@ -1,11 +1,13 @@
 package esdb
 
 import (
-	"fmt"
-	"github.com/EventStore/EventStore-Client-Go/esdb"
+	"context"
+	"github.com/EventStore/EventStore-Client-Go/v4/esdb"
 	"github.com/gofiber/fiber/v2/log"
 )
 
+type SubscriptionWorker func(ctx context.Context, stream *esdb.PersistentSubscription, workerID int) error
+
 func ConnectESDB() *esdb.Client {
 	cfg, err := esdb.ParseConnectionString("esdb://localhost:2113?tls=false")
 	if err != nil {
@@ -17,7 +19,5 @@ func ConnectESDB() *esdb.Client {
 		log.Fatal(err)
 	}
 
-	fmt.Printf("client.Config.Address: %v\n", client.Config.Address)
-
 	return client
 }

internal/common/eventsourcing/aggregate.go

@@ -3,7 +3,7 @@ package eventsourcing
 import (
 	"slices"
 
-	uuid "github.com/gofrs/uuid"
+	"github.com/google/uuid"
 )
 
 const (

internal/common/eventsourcing/command.go

@@ -1,6 +1,6 @@
 package eventsourcing
 
-import "github.com/gofrs/uuid"
+import "github.com/google/uuid"
 
 type Command interface {
 	GetAggregateId() uuid.UUID

internal/common/eventsourcing/event.go

@@ -2,10 +2,10 @@ package eventsourcing
 
 import (
 	"fmt"
+	"github.com/google/uuid"
 	"time"
 
-	"github.com/EventStore/EventStore-Client-Go/esdb"
-	uuid "github.com/gofrs/uuid"
+	"github.com/EventStore/EventStore-Client-Go/v4/esdb"
 	jsoniter "github.com/json-iterator/go"
 )
 
@@ -22,7 +22,7 @@ type Event struct {
 
 func NewEvent(a *AggregateBase, eventType string) *Event {
 	return &Event{
-		EventID:       uuid.Must(uuid.NewV4()),
+		EventID:       uuid.New(),
 		EventType:     eventType,
 		AggregateID:   a.GetAggregateId(),
 		AggregateType: a.GetType(),
@@ -31,13 +31,16 @@ func NewEvent(a *AggregateBase, eventType string) *Event {
 	}
 }
 
+// This may cause performance issues?
 func NewEventFromRecordedEvent(re *esdb.RecordedEvent) *Event {
+	aggrId, _ := uuid.Parse(re.StreamID)
+
 	return &Event{
 		EventID:     re.EventID,
 		EventType:   re.EventType,
 		Data:        re.Data,
 		Timestamp:   re.CreatedDate,
-		AggregateID: uuid.FromStringOrNil(re.StreamID),
+		AggregateID: aggrId,
 		Version:     int64(re.EventNumber),
 		Metadata:    re.UserMetadata,
 	}
@@ -98,7 +101,7 @@ func (e *Event) ToEventData() esdb.EventData {
 	return esdb.EventData{
 		EventID:     e.EventID,
 		EventType:   e.EventType,
-		ContentType: esdb.JsonContentType,
+		ContentType: esdb.ContentTypeJson,
 		Data:        e.Data,
 		Metadata:    e.Metadata,
 	}

internal/common/infrastructure/repository/aggregate.go

@@ -3,13 +3,13 @@ package repository
 import (
 	"context"
 	"errors"
-	"github.com/gofrs/uuid"
+	"github.com/google/uuid"
 	"io"
 	"log/slog"
 	"math"
 	"secap-input/internal/common/eventsourcing"
 
-	"github.com/EventStore/EventStore-Client-Go/esdb"
+	"github.com/EventStore/EventStore-Client-Go/v4/esdb"
 )
 
 const readCount = math.MaxUint64
@@ -36,10 +36,17 @@ func (r *AggregateRepository) Load(ctx context.Context, a eventsourcing.Aggregat
 			break
 		}
 
-		if errors.Is(err, esdb.ErrStreamNotFound) {
+		var esdbErr *esdb.Error
+		errors.As(err, &esdbErr)
+
+		if esdbErr != nil && esdbErr.IsErrorCode(esdb.ErrorCodeResourceNotFound) {
 			slog.Error("stream not found", err)
 			return err
 		}
+		//if errors.Is(err, esdb.ErrorCodeResourceNotFound) {
+		//	slog.Error("stream not found", err)
+		//	return err
+		//}
 
 		e := eventsourcing.NewEventFromRecordedEvent(re.Event)
 
@@ -85,12 +92,13 @@ func (r *AggregateRepository) Save(ctx context.Context, a eventsourcing.Aggregat
 	return nil
 }
 
-func (r *AggregateRepository) Exists(ctx context.Context, aggregateId uuid.UUID) error {
+func (r *AggregateRepository) Exists(ctx context.Context, aggregateId uuid.UUID) bool {
 	readStreamOptions := esdb.ReadStreamOptions{Direction: esdb.Backwards, From: esdb.Revision(1)}
+	// error nil means stream exists?
+	_, err := r.db.ReadStream(ctx, aggregateId.String(), readStreamOptions, 1)
 
-	if _, err := r.db.ReadStream(ctx, aggregateId.String(), readStreamOptions, 1); !errors.Is(err, esdb.ErrStreamNotFound) {
-		return err
+	if err == nil {
+		return false
 	}
-
-	return nil
+	return true
 }

internal/common/infrastructure/repository/event.go

@@ -7,7 +7,7 @@ import (
 	"log/slog"
 	"secap-input/internal/common/eventsourcing"
 
-	"github.com/EventStore/EventStore-Client-Go/esdb"
+	"github.com/EventStore/EventStore-Client-Go/v4/esdb"
 )
 
 type EventRepository struct {

internal/domain/building/application/create_building.go

@@ -2,7 +2,8 @@ package application
 
 import (
 	"context"
-	"github.com/gofrs/uuid"
+	"github.com/google/uuid"
+	"github.com/pkg/errors"
 	"secap-input/internal/common/infrastructure/repository"
 	"secap-input/internal/domain/building/core/aggregate"
 )
@@ -16,13 +17,13 @@ func NewCreateBuildingCommandHandler(repo *repository.AggregateRepository) *Crea
 }
 
 func (h *CreateBuildingCommandHandler) Handle(cmd *aggregate.CreateBuildingCommand) (uuid.UUID, error) {
-	err := h.repo.Exists(context.Background(), cmd.AggregateId)
-	if err != nil {
-		return uuid.Nil, err
+	exist := h.repo.Exists(context.Background(), cmd.AggregateId)
+	if exist {
+		return uuid.Nil, errors.Errorf("building with id %s already exists", cmd.AggregateId.String())
 	}
 
 	a := aggregate.NewBuildingAggregateWithId(cmd.AggregateId)
-	err = a.CreateBuildingCommandHandler(cmd)
+	err := a.CreateBuildingCommandHandler(cmd)
 	if err != nil {
 		return uuid.Nil, err
 	}

internal/domain/building/application/measure_building.go

@@ -2,6 +2,7 @@ package application
 
 import (
 	"context"
+	"github.com/pkg/errors"
 	"secap-input/internal/common/infrastructure/repository"
 	"secap-input/internal/domain/building/core/aggregate"
 )
@@ -16,8 +17,8 @@ func NewMeasureBuildingCommandHandler(repo *repository.AggregateRepository) *Mea
 
 func (h *MeasureBuildingCommandHandler) Handle(cmd *aggregate.MeasureBuildingCommand) error {
 	ctx := context.Background()
-	if err := h.repo.Exists(ctx, cmd.AggregateId); err != nil {
-		return err
+	if exist := h.repo.Exists(ctx, cmd.AggregateId); !exist {
+		return errors.Errorf("building with id %s does not exist", cmd.AggregateId.String())
 	}
 
 	a := aggregate.NewBuildingAggregateWithId(cmd.AggregateId)

internal/domain/building/core/aggregate/aggregate.go

@@ -1,7 +1,7 @@
 package aggregate
 
 import (
-	"github.com/gofrs/uuid"
+	"github.com/google/uuid"
 	"secap-input/internal/common/eventsourcing"
 	"secap-input/internal/domain/building/core/events"
 	"secap-input/internal/domain/building/core/model"

internal/domain/building/core/aggregate/command_handlers.go

@@ -2,7 +2,7 @@ package aggregate
 
 import (
 	"errors"
-	"github.com/gofrs/uuid"
+	"github.com/google/uuid"
 	"secap-input/internal/common/eventsourcing"
 	"secap-input/internal/domain/building/core/events"
 	"secap-input/internal/domain/building/core/model"

internal/domain/building/core/model/measurement.go

@@ -2,32 +2,29 @@ package model
 
 import (
 	"fmt"
+	"secap-input/internal/domain/building/core/vo"
 )
 
 // Q: Should esdb.Events be same as Measurement?
 // A: No, the NewMeasurementEvent containts the Measurements.
 type Measurement struct {
-	Unit            string          `json:"unit"`
-	Value           float64         `json:"value"`
-	MeasurementType MeasurementType `json:"measurementType"`
+	Unit            string  `json:"unit"`
+	Value           float64 `json:"value"`
+	MeasurementType string  `json:"measurementType"`
 }
 
-func NewMeasurement(unit string, value float64, measurementType string) (*Measurement, error) {
+func NewMeasurement(unit string, value float64, measurementType vo.MeasurementType) (*Measurement, error) {
 	if err := validateValue(value); err != nil {
 		return nil, err
 	}
 	if err := validateUnit(unit); err != nil {
 		return nil, err
 	}
-	mt, err := MeasurementTypeFromString(measurementType)
-	if err != nil {
-		return nil, err
-	}
 
 	return &Measurement{
 		Unit:            unit,
 		Value:           value,
-		MeasurementType: mt,
+		MeasurementType: measurementType.String(),
 	}, nil
 }