Moved esmongo.

Author: sjohnsonaz

client.go

@@ -0,0 +1,31 @@
+package mongodata
+
+import (
+	"context"
+
+	"go.mongodb.org/mongo-driver/event"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+	"go.mongodb.org/mongo-driver/mongo/readpref"
+)
+
+func Connect(
+	ctx context.Context,
+	uri string,
+	m *event.CommandMonitor,
+) (*mongo.Client, error) {
+	o := options.Client().ApplyURI(uri)
+	if m != nil {
+		o = o.SetMonitor(m)
+	}
+	client, err := mongo.Connect(ctx, o)
+	if err != nil {
+		return nil, err
+	}
+
+	if err := client.Ping(ctx, readpref.Primary()); err != nil {
+		return nil, err
+	}
+
+	return client, nil
+}

collection.go

@@ -0,0 +1,131 @@
+package mongodata
+
+import (
+	"context"
+	"errors"
+	"iter"
+
+	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/mongo"
+	"go.mongodb.org/mongo-driver/mongo/options"
+)
+
+var ErrNotFound = errors.New("not found")
+var ErrVersionMismatch = errors.New("version mismatch")
+
+type Collection[M any] struct {
+	collection *mongo.Collection
+}
+
+func NewCollection[M any](collection *mongo.Collection) *Collection[M] {
+	return &Collection[M]{collection}
+}
+
+func (c *Collection[M]) Insert(ctx context.Context, id string, m *M) error {
+	_, err := c.collection.InsertOne(ctx,
+		m)
+	return err
+}
+
+func (c *Collection[M]) Upsert(ctx context.Context, id string, m *M) error {
+	_, err := c.collection.ReplaceOne(ctx,
+		bson.M{"_id": id},
+		m,
+		options.Replace().SetUpsert(true))
+	return err
+}
+
+func (c *Collection[M]) Replace(ctx context.Context, filter FilterBuilder, m *M) error {
+	result, err := c.collection.ReplaceOne(ctx,
+		filter.Build(),
+		m)
+	if err != nil {
+		return err
+	}
+
+	if result.MatchedCount == 0 {
+		return ErrVersionMismatch
+	}
+
+	return nil
+}
+
+func (c *Collection[M]) Get(ctx context.Context, id string) (*M, error) {
+	m, err := decode[M](c.collection.FindOne(ctx,
+		bson.M{"_id": id}))
+	if errors.Is(err, mongo.ErrNoDocuments) {
+		err = ErrNotFound
+	}
+
+	return &m, err
+}
+
+func (c *Collection[M]) Delete(ctx context.Context, id string) error {
+	_, err := c.collection.DeleteOne(ctx, bson.M{"_id": id})
+	return err
+}
+
+func (c *Collection[M]) GetList(ctx context.Context, filter FilterBuilder, sort *SortBuilder) iter.Seq2[*M, error] {
+	return func(yield func(*M, error) bool) {
+		f := filter.Build()
+		o := sort.Build()
+		result, err := c.collection.Find(ctx, f, o)
+		if err != nil {
+			yield(nil, err)
+			return
+		}
+
+		defer result.Close(ctx)
+
+		for result.Next(ctx) {
+			m, err := decode[M](result)
+			if err != nil {
+				if errors.Is(err, mongo.ErrNoDocuments) {
+					err = ErrNotFound
+				}
+				yield(nil, err)
+				return
+			}
+
+			if !yield(&m, nil) {
+				return
+			}
+		}
+	}
+}
+
+type Change[M any] struct {
+	Value M
+	Token string
+}
+
+func newChange[M any](cs *mongo.ChangeStream) (Change[M], error) {
+	m, err := decode[M](cs)
+	return Change[M]{Value: m, Token: cs.ResumeToken().String()}, err
+}
+
+func (c *Collection[M]) Watch(ctx context.Context, pipeline any, token string) iter.Seq2[Change[M], error] {
+	return func(yield func(Change[M], error) bool) {
+		o := options.ChangeStream()
+		if token != "" {
+			o.SetResumeAfter(bson.Raw(token))
+		}
+
+		cs, err := c.collection.Watch(ctx, pipeline, o)
+		if err != nil {
+			yield(Change[M]{Token: token}, err)
+			return
+		}
+
+		defer cs.Close(ctx)
+
+		for cs.Next(ctx) {
+			if c, err := newChange[M](cs); err != nil {
+				yield(c, err)
+				return
+			} else if !yield(c, nil) {
+				return
+			}
+		}
+	}
+}

dataaccess.go

@@ -0,0 +1,10 @@
+package mongodata
+
+type decoder interface {
+	Decode(any) error
+}
+
+func decode[M any](decoder decoder) (M, error) {
+	var m M
+	return m, decoder.Decode(&m)
+}

decoder.go

@@ -0,0 +1,14 @@
+package mongodata
+
+type Entity[I EntityID] interface {
+	ID() I
+	Version() int
+}
+
+type EntityID interface {
+	String() string
+}
+
+type Event interface {
+	Type() string
+}

entity.go

@@ -0,0 +1,28 @@
+package mongodata
+
+import (
+	"encoding/json"
+	"errors"
+)
+
+type EventDefinition interface {
+	ParseEvent(re modelEvent) (Event, error)
+}
+
+type EventMapper map[string]EventDefinition
+
+func (em EventMapper) ParseEvent(re modelEvent) (Event, error) {
+	parser, ok := em[re.Type]
+	if !ok {
+		return nil, errors.New("no map")
+	}
+
+	return parser.ParseEvent(re)
+}
+
+type EventMap[T Event] struct{}
+
+func (EventMap[T]) ParseEvent(re modelEvent) (Event, error) {
+	var t T
+	return t, json.Unmarshal(re.Data, t)
+}

event_mapper.go

@@ -0,0 +1,40 @@
+package mongodata
+
+import (
+	"encoding/json"
+	"fmt"
+	"testing"
+)
+
+type (
+	entityID string
+)
+
+func (e entityID) String() string { return string(e) }
+
+type createEvent struct {
+	ID       string
+	EntityID entityID
+	Version  int
+}
+
+const (
+	createEventType = "create_event"
+)
+
+func (c createEvent) Type() string { return createEventType }
+
+func TestEventMapper(t *testing.T) {
+	em := EventMapper{
+		createEventType: EventMap[createEvent]{},
+	}
+	d, _ := json.Marshal(createEvent{
+		ID:       "event1",
+		EntityID: "entity1",
+		Version:  1,
+	})
+	e := newModelEvent(createEventType, d)
+	event, err := em.ParseEvent(*e)
+	fmt.Print(err)
+	fmt.Print(event)
+}

event_mapper_test.go

@@ -0,0 +1,52 @@
+package mongodata
+
+import (
+	"go.mongodb.org/mongo-driver/bson"
+)
+
+type FilterBuilder map[string]any
+
+func Filter() FilterBuilder {
+	return FilterBuilder{}
+}
+
+func (fb FilterBuilder) If(condition bool, a func(fb FilterBuilder)) FilterBuilder {
+	if condition {
+		a(fb)
+	}
+	return fb
+}
+
+func (fb FilterBuilder) IfElse(condition bool, a func(fb FilterBuilder), b func(fb FilterBuilder)) FilterBuilder {
+	if condition {
+		a(fb)
+	} else {
+		b(fb)
+	}
+	return fb
+}
+
+func (fb FilterBuilder) Property(key string, value any) FilterBuilder {
+	fb[key] = value
+	return fb
+}
+
+func (fb FilterBuilder) PropertyIf(condition bool, key string, value any) FilterBuilder {
+	if condition {
+		fb[key] = value
+	}
+	return fb
+}
+
+func (fb FilterBuilder) PropertyIfElse(condition bool, key string, a any, b any) FilterBuilder {
+	if condition {
+		fb[key] = a
+	} else {
+		fb[key] = b
+	}
+	return fb
+}
+
+func (fb FilterBuilder) Build() bson.M {
+	return bson.M(fb)
+}

filter_builder.go

@@ -1,29 +1,18 @@
 module github.com/cardboardrobots/mongodata
 
-go 1.19
+go 1.23.1
 
-require (
-	github.com/cardboardrobots/repository v0.0.2
-	github.com/go-playground/validator v9.31.0+incompatible
-	go.mongodb.org/mongo-driver v1.10.3
-)
+require go.mongodb.org/mongo-driver v1.17.1
 
 require (
-	github.com/go-playground/locales v0.14.0 // indirect
-	github.com/go-playground/universal-translator v0.18.0 // indirect
 	github.com/golang/snappy v0.0.4 // indirect
-	github.com/google/go-cmp v0.5.9 // indirect
-	github.com/klauspost/compress v1.15.11 // indirect
-	github.com/leodido/go-urn v1.2.1 // indirect
-	github.com/montanaflynn/stats v0.6.6 // indirect
-	github.com/pkg/errors v0.9.1 // indirect
-	github.com/stretchr/testify v1.8.0 // indirect
+	github.com/klauspost/compress v1.17.10 // indirect
+	github.com/montanaflynn/stats v0.7.1 // indirect
 	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
-	github.com/xdg-go/scram v1.1.1 // indirect
-	github.com/xdg-go/stringprep v1.0.3 // indirect
-	github.com/youmark/pkcs8 v0.0.0-20201027041543-1326539a0a0a // indirect
-	golang.org/x/crypto v0.0.0-20221012134737-56aed061732a // indirect
-	golang.org/x/sync v0.0.0-20220929204114-8fcdb60fdcc0 // indirect
-	golang.org/x/text v0.3.8 // indirect
-	gopkg.in/go-playground/assert.v1 v1.2.1 // indirect
+	github.com/xdg-go/scram v1.1.2 // indirect
+	github.com/xdg-go/stringprep v1.0.4 // indirect
+	github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
+	golang.org/x/crypto v0.27.0 // indirect
+	golang.org/x/sync v0.8.0 // indirect
+	golang.org/x/text v0.18.0 // indirect
 )