From 6a664539f8eee266aaf4ce85cc377645ad573a06 Mon Sep 17 00:00:00 2001 From: Kilian Boute Date: Thu, 13 Aug 2026 14:37:27 +0000 Subject: [PATCH] feat(database): add aggregate capability Add an optional Aggregator interface with MongoDB option forwarding and complete MockDatabase response, queue, call-tracking, reset, and test support. --- pkg/database/aggregate.go | 10 +++++++ pkg/database/mock.go | 57 ++++++++++++++++++++++++++++++++++++ pkg/database/mock_test.go | 42 ++++++++++++++++++++++++++ pkg/database/mongodb.go | 23 +++++++++++++++ pkg/database/mongodb_test.go | 10 +++++++ 5 files changed, 142 insertions(+) create mode 100644 pkg/database/aggregate.go diff --git a/pkg/database/aggregate.go b/pkg/database/aggregate.go new file mode 100644 index 0000000..a675f5a --- /dev/null +++ b/pkg/database/aggregate.go @@ -0,0 +1,10 @@ +package database + +import "context" + +// Aggregator is the optional database capability for executing MongoDB +// aggregation pipelines. Pipeline construction and result decoding remain +// consumer concerns. +type Aggregator interface { + Aggregate(ctx context.Context, db string, collection string, pipeline any, opts ...any) FindResultInterface +} diff --git a/pkg/database/mock.go b/pkg/database/mock.go index aea4d8d..0efdc85 100644 --- a/pkg/database/mock.go +++ b/pkg/database/mock.go @@ -28,6 +28,9 @@ type MockDatabase struct { // CreateIndexesFunc allows customizing index creation behavior. CreateIndexesFunc func(ctx context.Context, db string, collection string, indexes []mongo.IndexModel, opts ...any) ([]string, error) + // AggregateFunc allows customizing Aggregate behavior. + AggregateFunc func(ctx context.Context, db string, collection string, pipeline any, opts ...any) FindResultInterface + // UpdateOneFunc allows customizing UpdateOne behavior UpdateOneFunc func(ctx context.Context, db string, collection string, filter any, update any, opts ...any) (UpdateResultInterface, error) @@ -58,6 +61,7 @@ type MockDatabase struct { CountQueue []CountResponse InsertOneQueue []InsertResponse InsertManyQueue []InsertResponse + AggregateQueue []FindResponse // Call tracking PingCalls []PingCall @@ -71,10 +75,12 @@ type MockDatabase struct { CountCalls []CountCall InsertOneCalls []InsertOneCall InsertManyCalls []InsertManyCall + AggregateCalls []AggregateCall } var _ DatabaseInterface = (*MockDatabase)(nil) var _ IndexManager = (*MockDatabase)(nil) +var _ Aggregator = (*MockDatabase)(nil) // MockSingleResult implements SingleResultInterface for testing type MockSingleResult struct { @@ -286,6 +292,15 @@ type CreateIndexesCall struct { Opts []any } +// AggregateCall records an aggregation call. +type AggregateCall struct { + Ctx context.Context + Db string + Collection string + Pipeline any + Opts []any +} + // UpdateOneCall records a call to UpdateOne type UpdateOneCall struct { Ctx context.Context @@ -362,6 +377,9 @@ func NewMockDatabase() *MockDatabase { CreateIndexesFunc: func(ctx context.Context, db string, collection string, indexes []mongo.IndexModel, opts ...any) ([]string, error) { return nil, nil }, + AggregateFunc: func(ctx context.Context, db string, collection string, pipeline any, opts ...any) FindResultInterface { + return &MockFindResult{results: []any{}, err: nil} + }, UpdateOneFunc: func(ctx context.Context, db string, collection string, filter any, update any, opts ...any) (UpdateResultInterface, error) { return &MockUpdateResult{matchedCount: 1, modifiedCount: 1}, nil }, @@ -385,6 +403,7 @@ func NewMockDatabase() *MockDatabase { CountCalls: []CountCall{}, InsertOneCalls: []InsertOneCall{}, InsertManyCalls: []InsertManyCall{}, + AggregateCalls: []AggregateCall{}, PingQueue: []PingResponse{}, FindQueue: []FindResponse{}, FindOneQueue: []FindOneResponse{}, @@ -396,6 +415,7 @@ func NewMockDatabase() *MockDatabase { CountQueue: []CountResponse{}, InsertOneQueue: []InsertResponse{}, InsertManyQueue: []InsertResponse{}, + AggregateQueue: []FindResponse{}, } } @@ -467,6 +487,27 @@ func (m *MockDatabase) Find(ctx context.Context, db string, collection string, f return &MockFindResult{results: result, err: err} } +// Aggregate implements Aggregator. +func (m *MockDatabase) Aggregate(ctx context.Context, db string, collection string, pipeline any, opts ...any) FindResultInterface { + m.AggregateCalls = append(m.AggregateCalls, AggregateCall{ + Ctx: ctx, + Db: db, + Collection: collection, + Pipeline: pipeline, + Opts: opts, + }) + + if len(m.AggregateQueue) > 0 { + response := m.AggregateQueue[0] + m.AggregateQueue = m.AggregateQueue[1:] + return &MockFindResult{results: response.Result, err: response.Err} + } + if m.AggregateFunc != nil { + return m.AggregateFunc(ctx, db, collection, pipeline, opts...) + } + return &MockFindResult{results: []any{}, err: nil} +} + // FindOne implements DatabaseInterface func (m *MockDatabase) FindOne(ctx context.Context, db string, collection string, filter any, opts ...any) SingleResultInterface { m.FindOneCalls = append(m.FindOneCalls, FindOneCall{ @@ -804,6 +845,7 @@ func (m *MockDatabase) Reset() { m.CountCalls = []CountCall{} m.InsertOneCalls = []InsertOneCall{} m.InsertManyCalls = []InsertManyCall{} + m.AggregateCalls = []AggregateCall{} m.PingQueue = []PingResponse{} m.FindQueue = []FindResponse{} m.FindOneQueue = []FindOneResponse{} @@ -815,6 +857,7 @@ func (m *MockDatabase) Reset() { m.CountQueue = []CountResponse{} m.InsertOneQueue = []InsertResponse{} m.InsertManyQueue = []InsertResponse{} + m.AggregateQueue = []FindResponse{} } // ExpectPing sets up an expectation for Ping @@ -857,6 +900,14 @@ func (m *MockDatabase) ExpectCreateIndexes(names []string, err error) *MockDatab return m } +// ExpectAggregate sets up an aggregation response. +func (m *MockDatabase) ExpectAggregate(result any, err error) *MockDatabase { + m.AggregateFunc = func(ctx context.Context, db string, collection string, pipeline any, opts ...any) FindResultInterface { + return &MockFindResult{results: result, err: err} + } + return m +} + // ExpectUpdateOne sets up an expectation for UpdateOne func (m *MockDatabase) ExpectUpdateOne(matchedCount, modifiedCount, upsertedCount int64, upsertedID any, err error) *MockDatabase { m.UpdateOneFunc = func(ctx context.Context, db string, collection string, filter any, update any, opts ...any) (UpdateResultInterface, error) { @@ -925,6 +976,12 @@ func (m *MockDatabase) QueueCreateIndexes(names []string, err error) *MockDataba return m } +// QueueAggregate adds an aggregation response to the queue. +func (m *MockDatabase) QueueAggregate(result any, err error) *MockDatabase { + m.AggregateQueue = append(m.AggregateQueue, FindResponse{Result: result, Err: err}) + return m +} + // QueueUpdateOne adds an UpdateOne response to the queue for sequential calls func (m *MockDatabase) QueueUpdateOne(matchedCount, modifiedCount, upsertedCount int64, upsertedID any, err error) *MockDatabase { m.UpdateOneQueue = append(m.UpdateOneQueue, UpdateOneResponse{ diff --git a/pkg/database/mock_test.go b/pkg/database/mock_test.go index 5633931..0806f1b 100644 --- a/pkg/database/mock_test.go +++ b/pkg/database/mock_test.go @@ -911,3 +911,45 @@ func TestMockDatabaseInsert(t *testing.T) { } }) } + +func TestMockDatabaseAggregate(t *testing.T) { + t.Run("ExpectAggregateTracksPipelineAndOptions", func(t *testing.T) { + mock := NewMockDatabase().ExpectAggregate([]bson.M{{"count": 2}}, nil) + pipeline := mongo.Pipeline{{{Key: "$group", Value: bson.D{{Key: "_id", Value: "$status"}}}}} + opts := options.Aggregate().SetAllowDiskUse(true) + + var result []bson.M + err := mock.Aggregate(context.Background(), "testdb", "events", pipeline, opts).All(&result) + if err != nil { + t.Fatalf("Aggregate error: %v", err) + } + if len(result) != 1 || result[0]["count"] != float64(2) { + t.Fatalf("Aggregate result = %#v", result) + } + if len(mock.AggregateCalls) != 1 { + t.Fatalf("Aggregate calls = %d", len(mock.AggregateCalls)) + } + call := mock.AggregateCalls[0] + if call.Db != "testdb" || call.Collection != "events" || len(call.Opts) != 1 || call.Opts[0] != opts { + t.Fatalf("unexpected Aggregate call: %#v", call) + } + }) + + t.Run("QueueAggregatePreservesErrorAndReset", func(t *testing.T) { + expectedErr := errors.New("aggregate failed") + mock := NewMockDatabase().QueueAggregate(nil, expectedErr) + + result := mock.Aggregate(context.Background(), "testdb", "events", mongo.Pipeline{}) + if !errors.Is(result.Err(), expectedErr) { + t.Fatalf("Aggregate error = %v, want %v", result.Err(), expectedErr) + } + if len(mock.AggregateCalls) != 1 { + t.Fatalf("Aggregate calls = %d", len(mock.AggregateCalls)) + } + + mock.Reset() + if len(mock.AggregateCalls) != 0 || len(mock.AggregateQueue) != 0 { + t.Fatal("expected Aggregate state to be cleared") + } + }) +} diff --git a/pkg/database/mongodb.go b/pkg/database/mongodb.go index f8b7051..42616c3 100644 --- a/pkg/database/mongodb.go +++ b/pkg/database/mongodb.go @@ -231,6 +231,7 @@ type MongoClient struct { var _ DatabaseInterface = (*MongoClient)(nil) var _ IndexManager = (*MongoClient)(nil) +var _ Aggregator = (*MongoClient)(nil) // NewMongoClient creates a new MongoClient with the provided MongoDB settings func NewMongoClient(options *MongoOptions) (DatabaseInterface, error) { @@ -450,6 +451,28 @@ func (m *MongoClient) Find(ctx context.Context, db string, collection string, fi return &FindResult{cursor: cursor, ctx: ctx, err: err} } +// Aggregate executes an aggregation pipeline on the specified collection. +// Results support both slice-wide decoding and optional cursor iteration. +// Supports *moptions.AggregateOptions in opts. +func (m *MongoClient) Aggregate(ctx context.Context, db string, collection string, pipeline any, opts ...any) FindResultInterface { + coll := m.Client.Database(db).Collection(collection) + + cursor, err := coll.Aggregate(ctx, pipeline, aggregateOptions(opts...)...) + return &FindResult{cursor: cursor, ctx: ctx, err: err} +} + +// aggregateOptions retains supported MongoDB options and ignores unrelated +// wrapper options, consistent with the other optional capabilities. +func aggregateOptions(opts ...any) []*moptions.AggregateOptions { + var aggregateOpts []*moptions.AggregateOptions + for _, opt := range opts { + if ao, ok := opt.(*moptions.AggregateOptions); ok { + aggregateOpts = append(aggregateOpts, ao) + } + } + return aggregateOpts +} + // FindOne executes a findOne query on the specified database and collection. // Returns a SingleResult that can be used with .Into() or .Raw() for fluent decoding. // Supports *moptions.FindOneOptions and *Projection in opts. diff --git a/pkg/database/mongodb_test.go b/pkg/database/mongodb_test.go index afe7533..186392e 100644 --- a/pkg/database/mongodb_test.go +++ b/pkg/database/mongodb_test.go @@ -38,6 +38,16 @@ func TestInsertOptions(t *testing.T) { }) } +func TestAggregateOptions(t *testing.T) { + first := moptions.Aggregate().SetAllowDiskUse(true) + second := moptions.Aggregate().SetBatchSize(100) + + got := aggregateOptions(first, moptions.Find(), second) + if len(got) != 2 || got[0] != first || got[1] != second { + t.Fatalf("aggregate options = %#v", got) + } +} + func envVarsSet(keys ...string) bool { for _, key := range keys { if os.Getenv(key) == "" {