v0.0.363 wmo.extraModPipeline as func

wmo/collection.go

@@ -45,7 +45,7 @@ type Coll[TData any] struct {
 	customDecoder       *func(ctx context.Context, dec Decodable) (TData, error) // custom decoding function (useful if TData is an interface)
 	isInterfaceDataType bool                                                     // true if TData is an interface (not a struct)
 	unmarshalHooks      []func(d TData) TData                                    // called for every object after unmarshalling
-	extraModPipeline    mongo.Pipeline                                           // appended to pipelines after filter/limit/skip/sort, used for $lookup, $set, $unset, $project, etc
+	extraModPipeline    []func(ctx context.Context) mongo.Pipeline               // appended to pipelines after filter/limit/skip/sort, used for $lookup, $set, $unset, $project, etc
 }
 
 func (c *Coll[TData]) Collection() *mongo.Collection {
@@ -83,7 +83,13 @@ func (c *Coll[TData]) WithUnmarshalHook(fn func(d TData) TData) *Coll[TData] {
 }
 
 func (c *Coll[TData]) WithModifyingPipeline(p mongo.Pipeline) *Coll[TData] {
-	c.extraModPipeline = append(c.extraModPipeline, p...)
+	c.extraModPipeline = append(c.extraModPipeline, func(ctx context.Context) mongo.Pipeline { return p })
+
+	return c
+}
+
+func (c *Coll[TData]) WithModifyingPipelineFunc(fn func(ctx context.Context) mongo.Pipeline) *Coll[TData] {
+	c.extraModPipeline = append(c.extraModPipeline, fn)
 
 	return c
 }

wmo/queryAggregate.go

@@ -10,7 +10,9 @@ import (
 
 func (c *Coll[TData]) Aggregate(ctx context.Context, pipeline mongo.Pipeline, opts ...*options.AggregateOptions) ([]TData, error) {
 
-	pipeline = langext.ArrConcat(pipeline, c.extraModPipeline)
+	for _, ppl := range c.extraModPipeline {
+		pipeline = langext.ArrConcat(pipeline, ppl(ctx))
+	}
 
 	cursor, err := c.coll.Aggregate(ctx, pipeline, opts...)
 	if err != nil {
@@ -27,7 +29,9 @@ func (c *Coll[TData]) Aggregate(ctx context.Context, pipeline mongo.Pipeline, op
 
 func (c *Coll[TData]) AggregateOneOpt(ctx context.Context, pipeline mongo.Pipeline, opts ...*options.AggregateOptions) (*TData, error) {
 
-	pipeline = langext.ArrConcat(pipeline, c.extraModPipeline)
+	for _, ppl := range c.extraModPipeline {
+		pipeline = langext.ArrConcat(pipeline, ppl(ctx))
+	}
 
 	cursor, err := c.coll.Aggregate(ctx, pipeline, opts...)
 	if err != nil {
@@ -47,7 +51,9 @@ func (c *Coll[TData]) AggregateOneOpt(ctx context.Context, pipeline mongo.Pipeli
 
 func (c *Coll[TData]) AggregateOne(ctx context.Context, pipeline mongo.Pipeline, opts ...*options.AggregateOptions) (TData, error) {
 
-	pipeline = langext.ArrConcat(pipeline, c.extraModPipeline)
+	for _, ppl := range c.extraModPipeline {
+		pipeline = langext.ArrConcat(pipeline, ppl(ctx))
+	}
 
 	cursor, err := c.coll.Aggregate(ctx, pipeline, opts...)
 	if err != nil {

wmo/queryFind.go

@@ -32,7 +32,9 @@ func (c *Coll[TData]) Find(ctx context.Context, filter bson.M, opts ...*options.
 		}
 	}
 
-	pipeline = langext.ArrConcat(pipeline, c.extraModPipeline)
+	for _, ppl := range c.extraModPipeline {
+		pipeline = langext.ArrConcat(pipeline, ppl(ctx))
+	}
 
 	for _, opt := range opts {
 		if opt != nil && opt.Projection != nil {

wmo/queryFindOne.go

@@ -71,7 +71,9 @@ func (c *Coll[TData]) findOneInternal(ctx context.Context, filter bson.M, allowN
 		pipeline = append(pipeline, bson.D{{Key: "$match", Value: filter}})
 		pipeline = append(pipeline, bson.D{{Key: "$limit", Value: 1}})
 
-		pipeline = langext.ArrConcat(pipeline, c.extraModPipeline)
+		for _, ppl := range c.extraModPipeline {
+			pipeline = langext.ArrConcat(pipeline, ppl(ctx))
+		}
 
 		cursor, err := c.coll.Aggregate(ctx, pipeline)
 		if err != nil {

wmo/queryList.go

@@ -6,6 +6,7 @@ import (
 	"go.mongodb.org/mongo-driver/mongo"
 	ct "gogs.mikescher.com/BlackForestBytes/goext/cursortoken"
 	"gogs.mikescher.com/BlackForestBytes/goext/exerr"
+	"gogs.mikescher.com/BlackForestBytes/goext/langext"
 )
 
 func (c *Coll[TData]) List(ctx context.Context, filter ct.Filter, pageSize *int, inTok ct.CursorToken) ([]TData, ct.CursorToken, error) {
@@ -50,7 +51,10 @@ func (c *Coll[TData]) List(ctx context.Context, filter ct.Filter, pageSize *int,
 	}
 
 	pipeline = append(pipeline, paginationPipeline...)
-	pipeline = append(pipeline, c.extraModPipeline...)
+
+	for _, ppl := range c.extraModPipeline {
+		pipeline = langext.ArrConcat(pipeline, ppl(ctx))
+	}
 
 	cursor, err := c.coll.Aggregate(ctx, pipeline)
 	if err != nil {

wmo/queryPaginate.go

@@ -42,7 +42,12 @@ func (c *Coll[TData]) Paginate(ctx context.Context, filter pag.Filter, page int,
 	pipelineCount := mongo.Pipeline{}
 	pipelineCount = append(pipelineCount, bson.D{{Key: "$count", Value: "count"}})
 
-	pipelineList := langext.ArrConcat(pipelineFilter, pipelineSort, pipelinePaginate, c.extraModPipeline, pipelineSort)
+	extrModPipelineResolved := mongo.Pipeline{}
+	for _, ppl := range c.extraModPipeline {
+		extrModPipelineResolved = langext.ArrConcat(extrModPipelineResolved, ppl(ctx))
+	}
+
+	pipelineList := langext.ArrConcat(pipelineFilter, pipelineSort, pipelinePaginate, extrModPipelineResolved, pipelineSort)
 	pipelineTotalCount := langext.ArrConcat(pipelineFilter, pipelineCount)
 
 	cursorList, err := c.coll.Aggregate(ctx, pipelineList)