Organize backend into product executable and shared Go modules

This commit is contained in:
Stephan D committed 2026-09-12 14:14:08 +02:00
1 parent 981e84a4b4
commit f0720ec66b
382 files changed
+1944 -6989

No files matched your search

+13
View File
@@ -0,0 +1,13 @@
package db
import (
sharedmodel "git.sendico.io/tech/sxchnge/shared/model"
)
type DBDriver string
const (
Mongo DBDriver = "mongodb"
)
type Config = sharedmodel.DriverConfig[DBDriver]
+59
View File
@@ -0,0 +1,59 @@
package db
import (
"context"
mongoimpl "git.sendico.io/tech/sxchnge/shared/database/internal/mongo"
"git.sendico.io/tech/sxchnge/shared/errors"
"git.sendico.io/tech/sxchnge/shared/logging"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/readpref"
)
// Connection represents a low-level database connection lifecycle.
type Connection interface {
Disconnect(ctx context.Context) error
Ping(ctx context.Context) error
}
// MongoConnection provides direct access to the underlying mongo client.
type MongoConnection struct {
client *mongo.Client
database string
}
func (c *MongoConnection) Client() *mongo.Client {
return c.client
}
func (c *MongoConnection) Database() *mongo.Database {
return c.client.Database(c.database)
}
func (c *MongoConnection) Disconnect(ctx context.Context) error {
return c.client.Disconnect(ctx)
}
func (c *MongoConnection) Ping(ctx context.Context) error {
return c.client.Ping(ctx, readpref.Primary())
}
// ConnectMongo returns a low-level MongoDB connection without constructing repositories.
func ConnectMongo(logger mlogger.Logger, config *Config) (*MongoConnection, error) {
if config == nil {
return nil, merrors.InvalidArgument("database configuration is nil", "config")
}
if config.Driver != Mongo {
return nil, merrors.InvalidArgument("unsupported database driver: "+string(config.Driver), "config.driver")
}
client, _, settings, err := mongoimpl.ConnectClient(logger, config.Settings)
if err != nil {
return nil, err
}
return &MongoConnection{
client: client,
database: settings.Database,
}, nil
}
@@ -0,0 +1,34 @@
package mongo
import (
"git.sendico.io/tech/sxchnge/shared/database"
"git.sendico.io/tech/sxchnge/shared/database/transaction"
"git.sendico.io/tech/sxchnge/shared/errors"
"go.mongodb.org/mongo-driver/v2/mongo"
)
// Dependencies resolves local DB/transaction primitives for edge-local store wiring only.
type Dependencies struct {
Database *mongo.Database
Transactions transaction.Factory
}
type DatabaseProvider interface {
MongoDatabase() *mongo.Database
}
func ResolveDependencies(dbf db.Factory) (*Dependencies, error) {
if dbf == nil {
return nil, merrors.InvalidArgument("database factory is required", "dbFactory")
}
provider, ok := dbf.(DatabaseProvider)
if !ok || provider.MongoDatabase() == nil {
return nil, merrors.Internal("mongo database provider is not available for edge-local storage wiring")
}
return &Dependencies{
Database: provider.MongoDatabase(),
Transactions: dbf.TransactionFactory(),
}, nil
}
+26
View File
@@ -0,0 +1,26 @@
package db
import (
mongoimpl "git.sendico.io/tech/sxchnge/shared/database/internal/mongo"
"git.sendico.io/tech/sxchnge/shared/database/transaction"
"git.sendico.io/tech/sxchnge/shared/errors"
"git.sendico.io/tech/sxchnge/shared/logging"
)
// Factory exposes the local connection and transaction lifecycle.
type Factory interface {
TransactionFactory() transaction.Factory
CloseConnection()
}
// NewConnection builds a Factory backed by the configured driver.
func NewConnection(logger mlogger.Logger, config *Config) (Factory, error) {
if config == nil {
return nil, merrors.InvalidArgument("database configuration is required", "config")
}
if config.Driver == Mongo {
return mongoimpl.NewConnection(logger, config.Settings)
}
return nil, merrors.InvalidArgument("unknown database driver: "+string(config.Driver), "config.driver")
}
@@ -0,0 +1,90 @@
package builderimp
import (
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
)
type literalAccumulatorImp struct {
op builder.MongoOperation
value any
}
func (a *literalAccumulatorImp) Build() bson.D {
return bson.D{{Key: string(a.op), Value: a.value}}
}
func NewAccumulator(op builder.MongoOperation, value any) builder.Accumulator {
return &literalAccumulatorImp{op: op, value: value}
}
func AddToSet(value builder.Expression) builder.Expression {
return newUnaryExpression(builder.AddToSet, value)
}
func Size(value builder.Expression) builder.Expression {
return newUnaryExpression(builder.Size, value)
}
func Ne(left, right builder.Expression) builder.Expression {
return newBinaryExpression(builder.Ne, left, right)
}
func Sum(value any) builder.Accumulator {
return NewAccumulator(builder.Sum, value)
}
func Avg(value any) builder.Accumulator {
return NewAccumulator(builder.Avg, value)
}
func Min(value any) builder.Accumulator {
return NewAccumulator(builder.Min, value)
}
func Max(value any) builder.Accumulator {
return NewAccumulator(builder.Max, value)
}
func Eq(left, right builder.Expression) builder.Expression {
return newBinaryExpression(builder.Eq, left, right)
}
func Gt(left, right builder.Expression) builder.Expression {
return newBinaryExpression(builder.Gt, left, right)
}
func Add(left, right builder.Accumulator) builder.Accumulator {
return newBinaryAccumulator(builder.Add, left, right)
}
func Subtract(left, right builder.Accumulator) builder.Accumulator {
return newBinaryAccumulator(builder.Subtract, left, right)
}
func Multiply(left, right builder.Accumulator) builder.Accumulator {
return newBinaryAccumulator(builder.Multiply, left, right)
}
func Divide(left, right builder.Accumulator) builder.Accumulator {
return newBinaryAccumulator(builder.Divide, left, right)
}
type binaryAccumulator struct {
op builder.MongoOperation
left builder.Accumulator
right builder.Accumulator
}
func newBinaryAccumulator(op builder.MongoOperation, left, right builder.Accumulator) builder.Accumulator {
return &binaryAccumulator{
op: op,
left: left,
right: right,
}
}
func (b *binaryAccumulator) Build() bson.D {
args := []any{b.left.Build(), b.right.Build()}
return bson.D{{Key: string(b.op), Value: args}}
}
@@ -0,0 +1,87 @@
package builderimp
import (
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
)
type aliasImp struct {
lhs builder.Field
rhs any
}
func (a *aliasImp) Field() builder.Field {
return a.lhs
}
func (a *aliasImp) Build() bson.D {
return bson.D{{Key: a.lhs.Build(), Value: a.rhs}}
}
// 1. Null alias (_id: null)
func NewNullAlias(lhs builder.Field) builder.Alias {
return &aliasImp{lhs: lhs, rhs: nil}
}
func NewAlias(lhs builder.Field, rhs any) builder.Alias {
return &aliasImp{lhs: lhs, rhs: rhs}
}
// 2. Simple alias (_id: "$taskRef")
func NewSimpleAlias(lhs, rhs builder.Field) builder.Alias {
return &aliasImp{lhs: lhs, rhs: rhs.Build()}
}
// 3. Complex alias (_id: { aliasName: "$originalField", ... })
type ComplexAlias struct {
lhs builder.Field
rhs []builder.Alias // Correcting handling of slice of aliases
}
func (a *ComplexAlias) Field() builder.Field {
return a.lhs
}
func (a *ComplexAlias) Build() bson.D {
fieldMap := bson.M{}
for _, alias := range a.rhs {
aliasDoc := alias.Build()
for _, field := range aliasDoc {
fieldMap[field.Key] = field.Value
}
}
return bson.D{{Key: a.lhs.Build(), Value: fieldMap}}
}
func NewComplexAlias(lhs builder.Field, rhs []builder.Alias) builder.Alias {
return &ComplexAlias{lhs: lhs, rhs: rhs}
}
type aliasesImp struct {
aliases []builder.Alias
}
func (a *aliasesImp) Field() builder.Field {
if len(a.aliases) > 0 {
return a.aliases[0].Field()
}
return NewFieldImp("")
}
func (a *aliasesImp) Build() bson.D {
results := make([]bson.D, 0)
for _, alias := range a.aliases {
results = append(results, alias.Build())
}
aliases := bson.D{}
for _, r := range results {
aliases = append(aliases, r...)
}
return aliases
}
func NewAliases(aliases ...builder.Alias) builder.Alias {
return &aliasesImp{aliases: aliases}
}
@@ -0,0 +1,27 @@
package builderimp
import (
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
)
type arrayImp struct {
elements []builder.Expression
}
// Build renders the literal array:
//
// [ <expr1>, <expr2>, … ]
func (b *arrayImp) Build() bson.A {
arr := make(bson.A, len(b.elements))
for i, expr := range b.elements {
// each expr.Build() returns the raw value or sub‐expression
arr[i] = expr.Build()
}
return arr
}
// NewArray constructs a new array expression from the given sub‐expressions.
func NewArray(exprs ...builder.Expression) *arrayImp {
return &arrayImp{elements: exprs}
}
@@ -0,0 +1,108 @@
package builderimp
import (
"reflect"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
)
type literalExpression struct {
value any
}
func NewLiteralExpression(value any) builder.Expression {
return &literalExpression{value: value}
}
func (e *literalExpression) Build() any {
return bson.D{{Key: string(builder.Literal), Value: e.value}}
}
type variadicExpression struct {
op builder.MongoOperation
parts []builder.Expression
}
func (e *variadicExpression) Build() any {
args := make([]any, 0, len(e.parts))
for _, p := range e.parts {
args = append(args, p.Build())
}
return bson.D{{Key: string(e.op), Value: args}}
}
func newVariadicExpression(op builder.MongoOperation, exprs ...builder.Expression) builder.Expression {
return &variadicExpression{
op: op,
parts: exprs,
}
}
func newBinaryExpression(op builder.MongoOperation, left, right builder.Expression) builder.Expression {
return &variadicExpression{
op: op,
parts: []builder.Expression{left, right},
}
}
type unaryExpression struct {
op builder.MongoOperation
rhs builder.Expression
}
func (e *unaryExpression) Build() any {
return bson.D{{Key: string(e.op), Value: e.rhs.Build()}}
}
func newUnaryExpression(op builder.MongoOperation, right builder.Expression) builder.Expression {
return &unaryExpression{
op: op,
rhs: right,
}
}
type matchExpression struct {
op builder.MongoOperation
rhs builder.Expression
}
func (e *matchExpression) Build() any {
return bson.E{Key: string(e.op), Value: e.rhs.Build()}
}
func newMatchExpression(op builder.MongoOperation, right builder.Expression) builder.Expression {
return &matchExpression{
op: op,
rhs: right,
}
}
func InRef(value builder.Field) builder.Expression {
return newMatchExpression(builder.In, NewValue(NewRefFieldImp(value).Build()))
}
type inImpl struct {
values []any
}
func (e *inImpl) Build() any {
return bson.D{{Key: string(builder.In), Value: e.values}}
}
func In(values ...any) builder.Expression {
var flattenedValues []any
for _, v := range values {
switch reflect.TypeOf(v).Kind() {
case reflect.Slice:
slice := reflect.ValueOf(v)
for i := range slice.Len() {
flattenedValues = append(flattenedValues, slice.Index(i).Interface())
}
default:
flattenedValues = append(flattenedValues, v)
}
}
return &inImpl{values: flattenedValues}
}
@@ -0,0 +1,71 @@
package builderimp
import (
"strings"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
)
type FieldImp struct {
fields []string
}
func (b *FieldImp) Dot(field string) builder.Field {
newFields := make([]string, len(b.fields), len(b.fields)+1)
copy(newFields, b.fields)
newFields = append(newFields, field)
return &FieldImp{fields: newFields}
}
func (b *FieldImp) CopyWith(field string) builder.Field {
copiedFields := make([]string, 0, len(b.fields)+1)
copiedFields = append(copiedFields, b.fields...)
copiedFields = append(copiedFields, field)
return &FieldImp{
fields: copiedFields,
}
}
func (b *FieldImp) Build() string {
return strings.Join(b.fields, ".")
}
func NewFieldImp(baseName string) builder.Field {
return &FieldImp{
fields: []string{baseName},
}
}
type RefField struct {
imp builder.Field
}
func (b *RefField) Build() string {
return "$" + b.imp.Build()
}
func (b *RefField) CopyWith(field string) builder.Field {
return &RefField{
imp: b.imp.CopyWith(field),
}
}
func (b *RefField) Dot(field string) builder.Field {
return &RefField{
imp: b.imp.Dot(field),
}
}
func NewRefFieldImp(field builder.Field) builder.Field {
return &RefField{
imp: field,
}
}
func NewRootRef() builder.Field {
return NewFieldImp("$$ROOT")
}
func NewRemoveRef() builder.Field {
return NewFieldImp("$$REMOVE")
}
@@ -0,0 +1,137 @@
package builderimp
import (
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
)
type condImp struct {
condition builder.Expression
ifTrue any
ifFalse any
}
func (c *condImp) Build() any {
return bson.D{
{Key: string(builder.Cond), Value: bson.D{
{Key: "if", Value: c.condition.Build()},
{Key: "then", Value: c.ifTrue},
{Key: "else", Value: c.ifFalse},
}},
}
}
func NewCond(condition builder.Expression, ifTrue, ifFalse any) builder.Expression {
return &condImp{
condition: condition,
ifTrue: ifTrue,
ifFalse: ifFalse,
}
}
// setUnionImp implements builder.Expression but takes only builder.Array inputs.
type setUnionImp struct {
inputs []builder.Expression
}
// Build renders the $setUnion stage:
//
// { $setUnion: [ <array1>, <array2>, … ] }
func (s *setUnionImp) Build() any {
arr := make(bson.A, len(s.inputs))
for i, arrayExpr := range s.inputs {
arr[i] = arrayExpr.Build()
}
return bson.D{
{Key: string(builder.SetUnion), Value: arr},
}
}
// NewSetUnion constructs a new $setUnion expression from the given Arrays.
func NewSetUnion(arrays ...builder.Expression) builder.Expression {
return &setUnionImp{inputs: arrays}
}
type assignmentImp struct {
field builder.Field
expression builder.Expression
}
func (a *assignmentImp) Build() bson.D {
// Assign it to the given field name
return bson.D{
{Key: a.field.Build(), Value: a.expression.Build()},
}
}
// NewAssignment creates a projection assignment of the form:
//
// <field>: <expression>
func NewAssignment(field builder.Field, expression builder.Expression) builder.Projection {
return &assignmentImp{
field: field,
expression: expression,
}
}
type computeImp struct {
field builder.Field
expression builder.Expression
}
func (a *computeImp) Build() any {
return bson.D{
{Key: a.field.Build(), Value: a.expression.Build()},
}
}
func NewCompute(field builder.Field, expression builder.Expression) builder.Expression {
return &computeImp{
field: field,
expression: expression,
}
}
func NewIfNull(expression, replacement builder.Expression) builder.Expression {
return newBinaryExpression(builder.IfNull, expression, replacement)
}
func NewPush(expression builder.Expression) builder.Expression {
return newUnaryExpression(builder.Push, expression)
}
func NewAnd(exprs ...builder.Expression) builder.Expression {
return newVariadicExpression(builder.And, exprs...)
}
func NewOr(exprs ...builder.Expression) builder.Expression {
return newVariadicExpression(builder.Or, exprs...)
}
func NewEach(exprs ...builder.Expression) builder.Expression {
return newVariadicExpression(builder.Each, exprs...)
}
func NewLt(left, right builder.Expression) builder.Expression {
return newBinaryExpression(builder.Lt, left, right)
}
func NewNot(expression builder.Expression) builder.Expression {
return newUnaryExpression(builder.Not, expression)
}
func NewSum(expression builder.Expression) builder.Expression {
return newUnaryExpression(builder.Sum, expression)
}
func NewMin(expression builder.Expression) builder.Expression {
return newUnaryExpression(builder.Min, expression)
}
func First(expr builder.Expression) builder.Expression {
return newUnaryExpression(builder.First, expr)
}
func NewType(expr builder.Expression) builder.Expression {
return newUnaryExpression(builder.Type, expr)
}
@@ -0,0 +1,35 @@
package builderimp
import (
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
)
type groupAccumulatorImp struct {
field builder.Field
acc builder.Accumulator
}
// NewGroupAccumulator creates a new GroupAccumulator for the given field using the specified operator and value.
func NewGroupAccumulator(field builder.Field, acc builder.Accumulator) builder.GroupAccumulator {
return &groupAccumulatorImp{
field: field,
acc: acc,
}
}
func (g *groupAccumulatorImp) Field() builder.Field {
return g.field
}
func (g *groupAccumulatorImp) Accumulator() builder.Accumulator {
return g.acc
}
// Build returns a bson.E element for this group accumulator.
func (g *groupAccumulatorImp) Build() bson.D {
return bson.D{{
Key: g.field.Build(),
Value: g.acc.Build(),
}}
}
@@ -0,0 +1,104 @@
package builderimp
import (
"time"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
)
type patchBuilder struct {
updates bson.D
}
func set(field builder.Field, value any) bson.E {
return bson.E{Key: string(builder.Set), Value: bson.D{{Key: field.Build(), Value: value}}}
}
func (u *patchBuilder) Set(field builder.Field, value any) builder.Patch {
u.updates = append(u.updates, set(field, value))
return u
}
func (u *patchBuilder) Inc(field builder.Field, value any) builder.Patch {
u.updates = append(u.updates, bson.E{Key: string(builder.Inc), Value: bson.D{{Key: field.Build(), Value: value}}})
return u
}
func (u *patchBuilder) Unset(field builder.Field) builder.Patch {
u.updates = append(u.updates, bson.E{Key: string(builder.Unset), Value: bson.D{{Key: field.Build(), Value: ""}}})
return u
}
func (u *patchBuilder) Rename(field builder.Field, newName string) builder.Patch {
u.updates = append(u.updates, bson.E{Key: string(builder.Rename), Value: bson.D{{Key: field.Build(), Value: newName}}})
return u
}
func (u *patchBuilder) Push(field builder.Field, value any) builder.Patch {
u.updates = append(u.updates, bson.E{Key: string(builder.Push), Value: bson.D{{Key: field.Build(), Value: value}}})
return u
}
func (u *patchBuilder) Pull(field builder.Field, value any) builder.Patch {
u.updates = append(u.updates, bson.E{Key: string(builder.Pull), Value: bson.D{{Key: field.Build(), Value: value}}})
return u
}
func (u *patchBuilder) AddToSet(field builder.Field, value any) builder.Patch {
u.updates = append(u.updates, bson.E{Key: string(builder.AddToSet), Value: bson.D{{Key: field.Build(), Value: value}}})
return u
}
func (u *patchBuilder) Build() bson.D {
updates := append(bson.D{}, u.updates...)
updates = append(updates, set(NewFieldImp(storable.UpdatedAtField), time.Now()))
return mergePatchOperations(updates)
}
func NewPatchImp() builder.Patch {
return &patchBuilder{updates: bson.D{}}
}
func mergePatchOperations(updates bson.D) bson.D {
operators := make([]string, 0, len(updates))
merged := make(map[string]bson.D, len(updates))
for _, update := range updates {
fields, ok := update.Value.(bson.D)
if !ok {
operators = appendMissingOperator(operators, update.Key)
merged[update.Key] = append(merged[update.Key], bson.E{Key: update.Key, Value: update.Value})
continue
}
operators = appendMissingOperator(operators, update.Key)
for _, field := range fields {
merged[update.Key] = setPatchField(merged[update.Key], field)
}
}
out := make(bson.D, 0, len(operators))
for _, operator := range operators {
out = append(out, bson.E{Key: operator, Value: merged[operator]})
}
return out
}
func appendMissingOperator(operators []string, operator string) []string {
for _, existing := range operators {
if existing == operator {
return operators
}
}
return append(operators, operator)
}
func setPatchField(fields bson.D, field bson.E) bson.D {
for i := range fields {
if fields[i].Key == field.Key {
fields[i] = field
return fields
}
}
return append(fields, field)
}
@@ -0,0 +1,98 @@
package builderimp
import (
"testing"
"time"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
)
func TestPatchBuilderBuildMergesSetOperations(t *testing.T) {
patch := NewPatchImp().
Set(NewFieldImp("name"), "alice").
Inc(NewFieldImp("version"), 1).
Set(NewFieldImp("status"), "active")
built := patch.Build()
if got := countPatchOperator(built, string(builder.Set)); got != 1 {
t.Fatalf("$set operator count = %d, want 1: %#v", got, built)
}
if got := countPatchOperator(built, string(builder.Inc)); got != 1 {
t.Fatalf("$inc operator count = %d, want 1: %#v", got, built)
}
setFields := patchOperatorFields(t, built, string(builder.Set))
for _, key := range []string{"name", "status", storable.UpdatedAtField} {
if countPatchField(setFields, key) != 1 {
t.Fatalf("$set field %s must be present once: %#v", key, setFields)
}
}
}
func TestPatchBuilderBuildOwnsUpdatedAt(t *testing.T) {
manualUpdatedAt := time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)
patch := NewPatchImp().
Set(NewFieldImp(storable.UpdatedAtField), manualUpdatedAt).
Set(NewFieldImp("status"), "active")
built := patch.Build()
setFields := patchOperatorFields(t, built, string(builder.Set))
if got := countPatchField(setFields, storable.UpdatedAtField); got != 1 {
t.Fatalf("updatedAt field count = %d, want 1: %#v", got, setFields)
}
value, ok := patchFieldValue(setFields, storable.UpdatedAtField).(time.Time)
if !ok {
t.Fatalf("updatedAt value type = %T, want time.Time", patchFieldValue(setFields, storable.UpdatedAtField))
}
if value.Equal(manualUpdatedAt) {
t.Fatalf("updatedAt must be repository-owned, got manual value %s", value.Format(time.RFC3339Nano))
}
}
func patchOperatorFields(t *testing.T, built bson.D, operator string) bson.D {
t.Helper()
for _, entry := range built {
if entry.Key != operator {
continue
}
fields, ok := entry.Value.(bson.D)
if !ok {
t.Fatalf("%s value type = %T, want bson.D", operator, entry.Value)
}
return fields
}
t.Fatalf("%s operator missing: %#v", operator, built)
return nil
}
func countPatchOperator(built bson.D, operator string) int {
count := 0
for _, entry := range built {
if entry.Key == operator {
count++
}
}
return count
}
func countPatchField(fields bson.D, key string) int {
count := 0
for _, field := range fields {
if field.Key == key {
count++
}
}
return count
}
func patchFieldValue(fields bson.D, key string) any {
for _, field := range fields {
if field.Key == key {
return field.Value
}
}
return nil
}
@@ -0,0 +1,133 @@
package builderimp
import (
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"git.sendico.io/tech/sxchnge/shared/service"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
)
type unwindOpts = builder.UnwindOpts
// UnwindOption is the same type defined in the builder package.
type UnwindOption = builder.UnwindOption
// NewUnwindOpts applies all UnwindOption's to a fresh unwindOpts.
func NewUnwindOpts(opts ...UnwindOption) *unwindOpts {
cfg := &unwindOpts{}
for _, opt := range opts {
opt(cfg)
}
return cfg
}
type PipelineImp struct {
pipeline mongo.Pipeline
}
func (b *PipelineImp) Match(filter builder.Query) builder.Pipeline {
b.pipeline = append(b.pipeline, filter.BuildPipeline())
return b
}
func (b *PipelineImp) Lookup(from mservice.Type, localField, foreignField, as builder.Field) builder.Pipeline {
b.pipeline = append(b.pipeline, bson.D{{Key: string(builder.Lookup), Value: bson.D{
{Key: string(builder.MKFrom), Value: from},
{Key: string(builder.MKLocalField), Value: localField.Build()},
{Key: string(builder.MKForeignField), Value: foreignField.Build()},
{Key: string(builder.MKAs), Value: as.Build()},
}}})
return b
}
func (b *PipelineImp) LookupWithPipeline(
from mservice.Type,
nested builder.Pipeline,
as builder.Field,
let *map[string]builder.Field,
) builder.Pipeline {
lookupStage := bson.D{
{Key: string(builder.MKFrom), Value: from},
{Key: string(builder.MKPipeline), Value: nested.Build()},
{Key: string(builder.MKAs), Value: as.Build()},
}
// only add "let" if provided and not empty
if let != nil && len(*let) > 0 {
letDoc := bson.D{}
for varName, fld := range *let {
letDoc = append(letDoc, bson.E{Key: varName, Value: fld.Build()})
}
lookupStage = append(lookupStage, bson.E{Key: string(builder.MKLet), Value: letDoc})
}
b.pipeline = append(b.pipeline, bson.D{{Key: string(builder.Lookup), Value: lookupStage}})
return b
}
func (b *PipelineImp) Unwind(path builder.Field, opts ...UnwindOption) builder.Pipeline {
cfg := NewUnwindOpts(opts...)
var stageValue interface{}
// if no options, shorthand
if !cfg.PreserveNullAndEmptyArrays && cfg.IncludeArrayIndex == "" {
stageValue = path.Build()
} else {
d := bson.D{{Key: string(builder.MKPath), Value: path.Build()}}
if cfg.PreserveNullAndEmptyArrays {
d = append(d, bson.E{Key: string(builder.MKPreserveNullAndEmptyArrays), Value: true})
}
if cfg.IncludeArrayIndex != "" {
d = append(d, bson.E{Key: string(builder.MKIncludeArrayIndex), Value: cfg.IncludeArrayIndex})
}
stageValue = d
}
b.pipeline = append(b.pipeline, bson.D{{Key: string(builder.Unwind), Value: stageValue}})
return b
}
func (b *PipelineImp) Count(field builder.Field) builder.Pipeline {
b.pipeline = append(b.pipeline, bson.D{{Key: string(builder.Count), Value: field.Build()}})
return b
}
func (b *PipelineImp) Group(groupBy builder.Alias, accumulators ...builder.GroupAccumulator) builder.Pipeline {
groupDoc := groupBy.Build()
for _, acc := range accumulators {
groupDoc = append(groupDoc, acc.Build()...)
}
b.pipeline = append(b.pipeline, bson.D{
{Key: string(builder.Group), Value: groupDoc},
})
return b
}
func (b *PipelineImp) Project(projections ...builder.Projection) builder.Pipeline {
projDoc := bson.D{}
for _, pr := range projections {
projDoc = append(projDoc, pr.Build()...)
}
b.pipeline = append(b.pipeline, bson.D{{Key: string(builder.Project), Value: projDoc}})
return b
}
func (b *PipelineImp) ReplaceRoot(newRoot builder.Expression) builder.Pipeline {
b.pipeline = append(b.pipeline, bson.D{{Key: string(builder.ReplaceRoot), Value: bson.D{
{Key: string(builder.MKNewRoot), Value: newRoot.Build()},
}}})
return b
}
func (b *PipelineImp) Build() mongo.Pipeline {
return b.pipeline
}
func NewPipelineImp(stages ...bson.D) builder.Pipeline {
pipeline := make(mongo.Pipeline, len(stages))
copy(pipeline, stages)
return &PipelineImp{
pipeline: pipeline,
}
}
@@ -0,0 +1,576 @@
package builderimp
import (
"testing"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"git.sendico.io/tech/sxchnge/shared/service"
"github.com/stretchr/testify/assert"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func TestNewPipelineImp(t *testing.T) {
pipeline := NewPipelineImp()
assert.NotNil(t, pipeline)
assert.IsType(t, &PipelineImp{}, pipeline)
// Build should return empty pipeline initially
built := pipeline.Build()
assert.NotNil(t, built)
assert.Len(t, built, 0)
}
func TestNewPipelineImpSeedsComposableStages(t *testing.T) {
sortStage := bson.D{{Key: "$sort", Value: bson.D{{Key: "createdAt", Value: -1}}}}
pipeline := NewPipelineImp(sortStage)
result := pipeline.Count(&MockField{build: "total"})
assert.Same(t, pipeline, result)
assert.Equal(t, mongo.Pipeline{
sortStage,
bson.D{{Key: "$count", Value: "total"}},
}, pipeline.Build())
}
func TestPipelineImp_Match(t *testing.T) {
pipeline := NewPipelineImp()
mockQuery := &MockQuery{
buildPipeline: bson.D{{Key: "$match", Value: bson.D{{Key: "field", Value: "value"}}}},
}
result := pipeline.Match(mockQuery)
// Should return self for chaining
assert.Same(t, pipeline, result)
built := pipeline.Build()
assert.Len(t, built, 1)
assert.Equal(t, bson.D{{Key: "$match", Value: bson.D{{Key: "field", Value: "value"}}}}, built[0])
}
func TestPipelineImp_Lookup(t *testing.T) {
pipeline := NewPipelineImp()
mockLocalField := &MockField{build: "localField"}
mockForeignField := &MockField{build: "foreignField"}
mockAsField := &MockField{build: "asField"}
result := pipeline.Lookup(mservice.Site, mockLocalField, mockForeignField, mockAsField)
// Should return self for chaining
assert.Same(t, pipeline, result)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Lookup), Value: bson.D{
{Key: string(builder.MKFrom), Value: mservice.Site},
{Key: string(builder.MKLocalField), Value: "localField"},
{Key: string(builder.MKForeignField), Value: "foreignField"},
{Key: string(builder.MKAs), Value: "asField"},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_LookupWithPipeline_WithoutLet(t *testing.T) {
pipeline := NewPipelineImp()
mockNestedPipeline := &MockPipeline{
build: mongo.Pipeline{bson.D{{Key: "$match", Value: bson.D{{Key: "nested", Value: true}}}}},
}
mockAsField := &MockField{build: "asField"}
result := pipeline.LookupWithPipeline(mservice.Site, mockNestedPipeline, mockAsField, nil)
// Should return self for chaining
assert.Same(t, pipeline, result)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Lookup), Value: bson.D{
{Key: string(builder.MKFrom), Value: mservice.Site},
{Key: string(builder.MKPipeline), Value: mockNestedPipeline.build},
{Key: string(builder.MKAs), Value: "asField"},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_LookupWithPipeline_WithLet(t *testing.T) {
pipeline := NewPipelineImp()
mockNestedPipeline := &MockPipeline{
build: mongo.Pipeline{bson.D{{Key: "$match", Value: bson.D{{Key: "nested", Value: true}}}}},
}
mockAsField := &MockField{build: "asField"}
mockLetField := &MockField{build: "$_id"}
letVars := map[string]builder.Field{
"projRef": mockLetField,
}
result := pipeline.LookupWithPipeline(mservice.Site, mockNestedPipeline, mockAsField, &letVars)
// Should return self for chaining
assert.Same(t, pipeline, result)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Lookup), Value: bson.D{
{Key: string(builder.MKFrom), Value: mservice.Site},
{Key: string(builder.MKPipeline), Value: mockNestedPipeline.build},
{Key: string(builder.MKAs), Value: "asField"},
{Key: string(builder.MKLet), Value: bson.D{{Key: "projRef", Value: "$_id"}}},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_LookupWithPipeline_WithEmptyLet(t *testing.T) {
pipeline := NewPipelineImp()
mockNestedPipeline := &MockPipeline{
build: mongo.Pipeline{bson.D{{Key: "$match", Value: bson.D{{Key: "nested", Value: true}}}}},
}
mockAsField := &MockField{build: "asField"}
emptyLetVars := map[string]builder.Field{}
pipeline.LookupWithPipeline(mservice.Site, mockNestedPipeline, mockAsField, &emptyLetVars)
built := pipeline.Build()
assert.Len(t, built, 1)
// Should not include let field when empty
expected := bson.D{{Key: string(builder.Lookup), Value: bson.D{
{Key: string(builder.MKFrom), Value: mservice.Site},
{Key: string(builder.MKPipeline), Value: mockNestedPipeline.build},
{Key: string(builder.MKAs), Value: "asField"},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_Unwind_Simple(t *testing.T) {
pipeline := NewPipelineImp()
mockField := &MockField{build: "$array"}
result := pipeline.Unwind(mockField)
// Should return self for chaining
assert.Same(t, pipeline, result)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Unwind), Value: "$array"}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_Unwind_WithPreserveNullAndEmptyArrays(t *testing.T) {
pipeline := NewPipelineImp()
mockField := &MockField{build: "$array"}
// Mock the UnwindOption function
preserveOpt := func(opts *builder.UnwindOpts) {
opts.PreserveNullAndEmptyArrays = true
}
pipeline.Unwind(mockField, preserveOpt)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Unwind), Value: bson.D{
{Key: string(builder.MKPath), Value: "$array"},
{Key: string(builder.MKPreserveNullAndEmptyArrays), Value: true},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_Unwind_WithIncludeArrayIndex(t *testing.T) {
pipeline := NewPipelineImp()
mockField := &MockField{build: "$array"}
// Mock the UnwindOption function
indexOpt := func(opts *builder.UnwindOpts) {
opts.IncludeArrayIndex = "arrayIndex"
}
pipeline.Unwind(mockField, indexOpt)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Unwind), Value: bson.D{
{Key: string(builder.MKPath), Value: "$array"},
{Key: string(builder.MKIncludeArrayIndex), Value: "arrayIndex"},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_Unwind_WithBothOptions(t *testing.T) {
pipeline := NewPipelineImp()
mockField := &MockField{build: "$array"}
// Mock the UnwindOption functions
preserveOpt := func(opts *builder.UnwindOpts) {
opts.PreserveNullAndEmptyArrays = true
}
indexOpt := func(opts *builder.UnwindOpts) {
opts.IncludeArrayIndex = "arrayIndex"
}
pipeline.Unwind(mockField, preserveOpt, indexOpt)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Unwind), Value: bson.D{
{Key: string(builder.MKPath), Value: "$array"},
{Key: string(builder.MKPreserveNullAndEmptyArrays), Value: true},
{Key: string(builder.MKIncludeArrayIndex), Value: "arrayIndex"},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_Count(t *testing.T) {
pipeline := NewPipelineImp()
mockField := &MockField{build: "totalCount"}
result := pipeline.Count(mockField)
// Should return self for chaining
assert.Same(t, pipeline, result)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Count), Value: "totalCount"}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_Group(t *testing.T) {
pipeline := NewPipelineImp()
mockAlias := &MockAlias{
build: bson.D{{Key: "_id", Value: "$field"}},
field: &MockField{build: "_id"},
}
mockAccumulator := &MockGroupAccumulator{
build: bson.D{{Key: "count", Value: bson.D{{Key: "$sum", Value: 1}}}},
}
result := pipeline.Group(mockAlias, mockAccumulator)
// Should return self for chaining
assert.Same(t, pipeline, result)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Group), Value: bson.D{
{Key: "_id", Value: "$field"},
{Key: "count", Value: bson.D{{Key: "$sum", Value: 1}}},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_Group_MultipleAccumulators(t *testing.T) {
pipeline := NewPipelineImp()
mockAlias := &MockAlias{
build: bson.D{{Key: "_id", Value: "$field"}},
field: &MockField{build: "_id"},
}
mockAccumulator1 := &MockGroupAccumulator{
build: bson.D{{Key: "count", Value: bson.D{{Key: "$sum", Value: 1}}}},
}
mockAccumulator2 := &MockGroupAccumulator{
build: bson.D{{Key: "total", Value: bson.D{{Key: "$sum", Value: "$amount"}}}},
}
pipeline.Group(mockAlias, mockAccumulator1, mockAccumulator2)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Group), Value: bson.D{
{Key: "_id", Value: "$field"},
{Key: "count", Value: bson.D{{Key: "$sum", Value: 1}}},
{Key: "total", Value: bson.D{{Key: "$sum", Value: "$amount"}}},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_Project(t *testing.T) {
pipeline := NewPipelineImp()
mockProjection := &MockProjection{
build: bson.D{{Key: "field1", Value: 1}},
}
result := pipeline.Project(mockProjection)
// Should return self for chaining
assert.Same(t, pipeline, result)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Project), Value: bson.D{
{Key: "field1", Value: 1},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_Project_MultipleProjections(t *testing.T) {
pipeline := NewPipelineImp()
mockProjection1 := &MockProjection{
build: bson.D{{Key: "field1", Value: 1}},
}
mockProjection2 := &MockProjection{
build: bson.D{{Key: "field2", Value: 0}},
}
pipeline.Project(mockProjection1, mockProjection2)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.Project), Value: bson.D{
{Key: "field1", Value: 1},
{Key: "field2", Value: 0},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_ChainedOperations(t *testing.T) {
pipeline := NewPipelineImp()
// Create mocks
mockQuery := &MockQuery{
buildPipeline: bson.D{{Key: "$match", Value: bson.D{{Key: "status", Value: "active"}}}},
}
mockLocalField := &MockField{build: "userId"}
mockForeignField := &MockField{build: "_id"}
mockAsField := &MockField{build: "user"}
mockUnwindField := &MockField{build: "$user"}
mockProjection := &MockProjection{
build: bson.D{{Key: "name", Value: "$user.name"}},
}
// Chain operations
result := pipeline.
Match(mockQuery).
Lookup(mservice.Accounts, mockLocalField, mockForeignField, mockAsField).
Unwind(mockUnwindField).
Project(mockProjection)
// Should return self for chaining
assert.Same(t, pipeline, result)
built := pipeline.Build()
assert.Len(t, built, 4)
// Verify each stage
assert.Equal(t, bson.D{{Key: "$match", Value: bson.D{{Key: "status", Value: "active"}}}}, built[0])
expectedLookup := bson.D{{Key: string(builder.Lookup), Value: bson.D{
{Key: string(builder.MKFrom), Value: mservice.Accounts},
{Key: string(builder.MKLocalField), Value: "userId"},
{Key: string(builder.MKForeignField), Value: "_id"},
{Key: string(builder.MKAs), Value: "user"},
}}}
assert.Equal(t, expectedLookup, built[1])
assert.Equal(t, bson.D{{Key: string(builder.Unwind), Value: "$user"}}, built[2])
expectedProject := bson.D{{Key: string(builder.Project), Value: bson.D{
{Key: "name", Value: "$user.name"},
}}}
assert.Equal(t, expectedProject, built[3])
}
func TestNewUnwindOpts(t *testing.T) {
t.Run("NoOptions", func(t *testing.T) {
opts := NewUnwindOpts()
assert.NotNil(t, opts)
assert.False(t, opts.PreserveNullAndEmptyArrays)
assert.Empty(t, opts.IncludeArrayIndex)
})
t.Run("WithPreserveOption", func(t *testing.T) {
preserveOpt := func(opts *builder.UnwindOpts) {
opts.PreserveNullAndEmptyArrays = true
}
opts := NewUnwindOpts(preserveOpt)
assert.True(t, opts.PreserveNullAndEmptyArrays)
assert.Empty(t, opts.IncludeArrayIndex)
})
t.Run("WithIndexOption", func(t *testing.T) {
indexOpt := func(opts *builder.UnwindOpts) {
opts.IncludeArrayIndex = "index"
}
opts := NewUnwindOpts(indexOpt)
assert.False(t, opts.PreserveNullAndEmptyArrays)
assert.Equal(t, "index", opts.IncludeArrayIndex)
})
t.Run("WithBothOptions", func(t *testing.T) {
preserveOpt := func(opts *builder.UnwindOpts) {
opts.PreserveNullAndEmptyArrays = true
}
indexOpt := func(opts *builder.UnwindOpts) {
opts.IncludeArrayIndex = "index"
}
opts := NewUnwindOpts(preserveOpt, indexOpt)
assert.True(t, opts.PreserveNullAndEmptyArrays)
assert.Equal(t, "index", opts.IncludeArrayIndex)
})
}
// Mock implementations for testing
type MockQuery struct {
buildPipeline bson.D
}
func (m *MockQuery) And(filters ...builder.Query) builder.Query { return m }
func (m *MockQuery) Or(filters ...builder.Query) builder.Query { return m }
func (m *MockQuery) Filter(field builder.Field, value any) builder.Query { return m }
func (m *MockQuery) Expression(value builder.Expression) builder.Query { return m }
func (m *MockQuery) Comparison(field builder.Field, operator builder.MongoOperation, value any) builder.Query {
return m
}
func (m *MockQuery) RegEx(field builder.Field, pattern, options string) builder.Query { return m }
func (m *MockQuery) In(field builder.Field, values ...any) builder.Query { return m }
func (m *MockQuery) NotIn(field builder.Field, values ...any) builder.Query { return m }
func (m *MockQuery) Sort(field builder.Field, ascending bool) builder.Query { return m }
func (m *MockQuery) Limit(limit *int64) builder.Query { return m }
func (m *MockQuery) Offset(offset *int64) builder.Query { return m }
func (m *MockQuery) Archived(isArchived *bool) builder.Query { return m }
func (m *MockQuery) BuildPipeline() bson.D { return m.buildPipeline }
func (m *MockQuery) BuildQuery() bson.D { return bson.D{} }
func (m *MockQuery) BuildOptions() *options.FindOptionsBuilder { return &options.FindOptionsBuilder{} }
type MockField struct {
build string
}
func (m *MockField) Dot(field string) builder.Field { return &MockField{build: m.build + "." + field} }
func (m *MockField) CopyWith(field string) builder.Field { return &MockField{build: field} }
func (m *MockField) Build() string { return m.build }
type MockPipeline struct {
build mongo.Pipeline
}
func (m *MockPipeline) Match(filter builder.Query) builder.Pipeline { return m }
func (m *MockPipeline) Lookup(from mservice.Type, localField, foreignField, as builder.Field) builder.Pipeline {
return m
}
func (m *MockPipeline) LookupWithPipeline(from mservice.Type, pipeline builder.Pipeline, as builder.Field, let *map[string]builder.Field) builder.Pipeline {
return m
}
func (m *MockPipeline) Unwind(path builder.Field, opts ...UnwindOption) builder.Pipeline { return m }
func (m *MockPipeline) Count(field builder.Field) builder.Pipeline { return m }
func (m *MockPipeline) Group(groupBy builder.Alias, accumulators ...builder.GroupAccumulator) builder.Pipeline {
return m
}
func (m *MockPipeline) Project(projections ...builder.Projection) builder.Pipeline { return m }
func (m *MockPipeline) ReplaceRoot(newRoot builder.Expression) builder.Pipeline { return m }
func (m *MockPipeline) Build() mongo.Pipeline { return m.build }
type MockAlias struct {
build bson.D
field builder.Field
}
func (m *MockAlias) Field() builder.Field { return m.field }
func (m *MockAlias) Build() bson.D { return m.build }
type MockGroupAccumulator struct {
build bson.D
}
func (m *MockGroupAccumulator) Build() bson.D { return m.build }
type MockProjection struct {
build bson.D
}
func (m *MockProjection) Build() bson.D { return m.build }
func TestPipelineImp_ReplaceRoot(t *testing.T) {
pipeline := NewPipelineImp()
mockExpr := &MockExpression{build: "$newRoot"}
result := pipeline.ReplaceRoot(mockExpr)
// Should return self for chaining
assert.Same(t, pipeline, result)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.ReplaceRoot), Value: bson.D{
{Key: string(builder.MKNewRoot), Value: "$newRoot"},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_ReplaceRoot_WithNestedField(t *testing.T) {
pipeline := NewPipelineImp()
mockExpr := &MockExpression{build: "$document.data"}
pipeline.ReplaceRoot(mockExpr)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.ReplaceRoot), Value: bson.D{
{Key: string(builder.MKNewRoot), Value: "$document.data"},
}}}
assert.Equal(t, expected, built[0])
}
func TestPipelineImp_ReplaceRoot_WithExpression(t *testing.T) {
pipeline := NewPipelineImp()
// Mock a complex expression like { $mergeObjects: [...] }
mockExpr := &MockExpression{build: bson.D{{Key: "$mergeObjects", Value: bson.A{"$field1", "$field2"}}}}
pipeline.ReplaceRoot(mockExpr)
built := pipeline.Build()
assert.Len(t, built, 1)
expected := bson.D{{Key: string(builder.ReplaceRoot), Value: bson.D{
{Key: string(builder.MKNewRoot), Value: bson.D{{Key: "$mergeObjects", Value: bson.A{"$field1", "$field2"}}}},
}}}
assert.Equal(t, expected, built[0])
}
type MockExpression struct {
build any
}
func (m *MockExpression) Build() any { return m.build }
@@ -0,0 +1,97 @@
package builderimp
import (
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
)
// projectionExprImp is a concrete implementation of builder.Projection
// that projects a field using a custom expression.
type projectionExprImp struct {
expr builder.Expression // The expression for this projection.
field builder.Field // The field name for the projected field.
}
// Field returns the field being projected.
func (p *projectionExprImp) Field() builder.Field {
return p.field
}
// Expression returns the expression for the projection.
func (p *projectionExprImp) Expression() builder.Expression {
return p.expr
}
// Build returns the built expression. If no expression is provided, returns 1.
func (p *projectionExprImp) Build() bson.D {
if p.expr == nil {
return bson.D{{Key: p.field.Build(), Value: 1}}
}
return bson.D{{Key: p.field.Build(), Value: p.expr.Build()}}
}
// NewProjectionExpr creates a new Projection for a given field and expression.
func NewProjectionExpr(field builder.Field, expr builder.Expression) builder.Projection {
return &projectionExprImp{field: field, expr: expr}
}
// aliasProjectionImp is a concrete implementation of builder.Projection
// that projects an alias (renaming a field or expression).
type aliasProjectionImp struct {
alias builder.Alias // The alias for this projection.
}
// Field returns the field being projected (via the alias).
func (p *aliasProjectionImp) Field() builder.Field {
return p.alias.Field()
}
// Expression returns no additional expression for an alias projection.
func (p *aliasProjectionImp) Expression() builder.Expression {
return nil
}
// Build returns the built alias expression.
func (p *aliasProjectionImp) Build() bson.D {
return p.alias.Build()
}
// NewAliasProjection creates a new Projection that renames or wraps an existing field or expression.
func NewAliasProjection(alias builder.Alias) builder.Projection {
return &aliasProjectionImp{alias: alias}
}
// sinkProjectionImp is a simple include/exclude projection (0 or 1).
type sinkProjectionImp struct {
field builder.Field // The field name for the projected field.
val int // 1 to include, 0 to exclude.
}
// Expression returns no expression for a sink projection.
func (p *sinkProjectionImp) Expression() builder.Expression {
return nil
}
// Build returns the include/exclude projection.
func (p *sinkProjectionImp) Build() bson.D {
return bson.D{{Key: p.field.Build(), Value: p.val}}
}
// NewSinkProjection creates a new Projection that includes (true) or excludes (false) a field.
func NewSinkProjection(field builder.Field, include bool) builder.Projection {
val := 0
if include {
val = 1
}
return &sinkProjectionImp{field: field, val: val}
}
// IncludeField returns a projection including the given field.
func IncludeField(field builder.Field) builder.Projection {
return NewSinkProjection(field, true)
}
// ExcludeField returns a projection excluding the given field.
func ExcludeField(field builder.Field) builder.Projection {
return NewSinkProjection(field, false)
}
@@ -0,0 +1,162 @@
package builderimp
import (
"reflect"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
type QueryImp struct {
filter bson.D
sort bson.D
limit *int64
offset *int64
}
func (b *QueryImp) Filter(field builder.Field, value any) builder.Query {
b.filter = append(b.filter, bson.E{Key: field.Build(), Value: value})
return b
}
func (b *QueryImp) And(filters ...builder.Query) builder.Query {
andFilters := bson.A{}
for _, f := range filters {
andFilters = append(andFilters, f.BuildQuery())
}
b.filter = append(b.filter, bson.E{Key: string(builder.And), Value: andFilters})
return b
}
func (b *QueryImp) Or(filters ...builder.Query) builder.Query {
orFilters := bson.A{}
for _, f := range filters {
orFilters = append(orFilters, f.BuildQuery())
}
b.filter = append(b.filter, bson.E{Key: string(builder.Or), Value: orFilters})
return b
}
func (b *QueryImp) Comparison(field builder.Field, operator builder.MongoOperation, value any) builder.Query {
b.filter = append(b.filter, bson.E{Key: field.Build(), Value: bson.M{string(operator): value}})
return b
}
func (b *QueryImp) Expression(value builder.Expression) builder.Query {
b.filter = append(b.filter, bson.E{Key: string(builder.Expr), Value: value.Build()})
return b
}
func (b *QueryImp) RegEx(field builder.Field, pattern, options string) builder.Query {
b.filter = append(b.filter, bson.E{Key: field.Build(), Value: bson.Regex{Pattern: pattern, Options: options}})
return b
}
func (b *QueryImp) opIn(field builder.Field, op builder.MongoOperation, values ...any) builder.Query {
// Mongo requires $in and $nin operands to be arrays. Keep an empty input as
// a non-nil slice so the BSON encoder emits [] instead of null.
flattenedValues := make([]any, 0, len(values))
for _, v := range values {
valueType := reflect.TypeOf(v)
if valueType == nil {
flattenedValues = append(flattenedValues, nil)
continue
}
switch valueType.Kind() {
case reflect.Slice:
slice := reflect.ValueOf(v)
for i := range slice.Len() {
flattenedValues = append(flattenedValues, slice.Index(i).Interface())
}
default:
flattenedValues = append(flattenedValues, v)
}
}
b.filter = append(b.filter, bson.E{Key: field.Build(), Value: bson.M{string(op): flattenedValues}})
return b
}
func (b *QueryImp) NotIn(field builder.Field, values ...any) builder.Query {
return b.opIn(field, builder.NotIn, values...)
}
func (b *QueryImp) In(field builder.Field, values ...any) builder.Query {
return b.opIn(field, builder.In, values...)
}
func (b *QueryImp) Archived(isArchived *bool) builder.Query {
if isArchived == nil {
return b
}
return b.And(NewQueryImp().Filter(NewFieldImp(storable.IsArchivedField), *isArchived))
}
func (b *QueryImp) Sort(field builder.Field, ascending bool) builder.Query {
order := 1
if !ascending {
order = -1
}
b.sort = append(b.sort, bson.E{Key: field.Build(), Value: order})
return b
}
func (b *QueryImp) BuildPipeline() bson.D {
query := bson.D{}
if len(b.filter) > 0 {
query = append(query, bson.E{Key: string(builder.Match), Value: b.filter})
}
if len(b.sort) > 0 {
query = append(query, bson.E{Key: string(builder.Sort), Value: b.sort})
}
if b.limit != nil {
query = append(query, bson.E{Key: string(builder.Limit), Value: *b.limit})
}
if b.offset != nil {
query = append(query, bson.E{Key: string(builder.Skip), Value: *b.offset})
}
return query
}
func (b *QueryImp) BuildQuery() bson.D {
return b.filter
}
func (b *QueryImp) Limit(limit *int64) builder.Query {
b.limit = limit
return b
}
func (b *QueryImp) Offset(offset *int64) builder.Query {
b.offset = offset
return b
}
func (b *QueryImp) BuildOptions() *options.FindOptionsBuilder {
opts := options.Find()
if b.limit != nil {
opts.SetLimit(*b.limit)
}
if b.offset != nil {
opts.SetSkip(*b.offset)
}
if len(b.sort) > 0 {
opts.SetSort(b.sort)
}
return opts
}
func NewQueryImp() builder.Query {
return &QueryImp{
filter: bson.D{},
sort: bson.D{},
}
}
@@ -0,0 +1,31 @@
package builderimp
import (
"testing"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"github.com/stretchr/testify/require"
"go.mongodb.org/mongo-driver/v2/bson"
)
func TestEmptySetComparisonsRemainBSONArrays(t *testing.T) {
field := NewFieldImp("organizationRef")
require.Equal(t,
bson.D{{Key: "organizationRef", Value: bson.M{string(builder.NotIn): []any{}}}},
NewQueryImp().NotIn(field).BuildQuery(),
)
require.Equal(t,
bson.D{{Key: "organizationRef", Value: bson.M{string(builder.In): []any{}}}},
NewQueryImp().In(field).BuildQuery(),
)
}
func TestSetComparisonsAcceptNilElements(t *testing.T) {
field := NewFieldImp("organizationRef")
require.Equal(t,
bson.D{{Key: "organizationRef", Value: bson.M{string(builder.NotIn): []any{"", nil}}}},
NewQueryImp().NotIn(field, "", nil).BuildQuery(),
)
}
@@ -0,0 +1,17 @@
package builderimp
import (
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
)
type valueImp struct {
value any
}
func (v *valueImp) Build() any {
return v.value
}
func NewValue(value any) builder.Value {
return &valueImp{value: value}
}
@@ -0,0 +1,240 @@
package mongo
import (
"context"
"fmt"
sharedmodel "git.sendico.io/tech/sxchnge/shared/model"
"net"
"os"
"strings"
mutil "git.sendico.io/tech/sxchnge/shared/config"
"git.sendico.io/tech/sxchnge/shared/database/internal/transaction"
"git.sendico.io/tech/sxchnge/shared/database/transaction"
"git.sendico.io/tech/sxchnge/shared/logging"
"github.com/go-viper/mapstructure/v2"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/readpref"
"go.uber.org/zap"
)
// Config represents configuration
type Config struct {
Port *string `mapstructure:"port"`
PortEnv *string `mapstructure:"port_env"`
User *string `mapstructure:"user"`
UserEnv *string `mapstructure:"user_env"`
PasswordEnv string `mapstructure:"password_env"`
Database *string `mapstructure:"database"`
DatabaseEnv *string `mapstructure:"database_env"`
Host *string `mapstructure:"host"`
HostEnv *string `mapstructure:"host_env"`
Hosts []string `mapstructure:"hosts,omitempty"`
HostsEnvPrefix *string `mapstructure:"hosts_env_prefix,omitempty"`
HostsEnvPrefixEnv *string `mapstructure:"hosts_env_prefix_env,omitempty"`
PortsEnvPrefix *string `mapstructure:"ports_env_prefix,omitempty"`
PortsEnvPrefixEnv *string `mapstructure:"ports_env_prefix_env,omitempty"`
URI *string `mapstructure:"uri,omitempty"`
URIEnv *string `mapstructure:"uri_env,omitempty"`
AuthSource *string `mapstructure:"auth_source,omitempty"`
AuthSourceEnv *string `mapstructure:"auth_source_env,omitempty"`
AuthMechanism *string `mapstructure:"auth_mechanism,omitempty"`
AuthMechanismEnv *string `mapstructure:"auth_mechanism_env,omitempty"`
ReplicaSet *string `mapstructure:"replica_set,omitempty"`
ReplicaSetEnv *string `mapstructure:"replica_set_env,omitempty"`
}
type DBSettings struct {
Host string
Hosts []string
Port string
User string
Password string
Database string
AuthSource string
AuthMechanism string
ReplicaSet string
URI string
}
func Config2DBSettings(logger mlogger.Logger, config *Config) *DBSettings {
p := new(DBSettings)
p.Port = mutil.GetConfigValue(logger, "port", "port_env", config.Port, config.PortEnv)
p.Database = mutil.GetConfigValue(logger, "database", "database_env", config.Database, config.DatabaseEnv)
p.Password = os.Getenv(config.PasswordEnv)
p.User = mutil.GetConfigValue(logger, "user", "user_env", config.User, config.UserEnv)
p.Host = mutil.GetConfigValue(logger, "host", "host_env", config.Host, config.HostEnv)
p.AuthSource = mutil.GetConfigValue(logger, "auth_source", "auth_source_env", config.AuthSource, config.AuthSourceEnv)
p.AuthMechanism = mutil.GetConfigValue(logger, "auth_mechanism", "auth_mechanism_env", config.AuthMechanism, config.AuthMechanismEnv)
p.ReplicaSet = mutil.GetConfigValue(logger, "replica_set", "replica_set_env", config.ReplicaSet, config.ReplicaSetEnv)
p.URI = mutil.GetConfigValue(logger, "uri", "uri_env", config.URI, config.URIEnv)
hostPrefix := mutil.GetConfigValue(logger, "hosts_env_prefix", "hosts_env_prefix_env", config.HostsEnvPrefix, config.HostsEnvPrefixEnv)
portPrefix := mutil.GetConfigValue(logger, "ports_env_prefix", "ports_env_prefix_env", config.PortsEnvPrefix, config.PortsEnvPrefixEnv)
if hostPrefix == "" && p.ReplicaSet != "" {
hostPrefix = "MONGO_HOSTS_"
}
if portPrefix == "" && p.ReplicaSet != "" {
portPrefix = "MONGO_PORTS_"
}
p.Hosts = collectReplicaHosts(config.Hosts, p.ReplicaSet, p.Port, hostPrefix, portPrefix)
return p
}
func decodeConfig(logger mlogger.Logger, settings sharedmodel.SettingsT) (*Config, *DBSettings, error) {
var config Config
if err := mapstructure.Decode(settings, &config); err != nil {
logger.Warn("Failed to decode settings", zap.Error(err))
return nil, nil, err
}
dbSettings := Config2DBSettings(logger, &config)
return &config, dbSettings, nil
}
func dialMongo(logger mlogger.Logger, dbSettings *DBSettings) (*mongo.Client, error) {
opts := buildOptions(dbSettings)
client, err := mongo.Connect(opts)
if err != nil {
logger.Error("Unable to connect to database", zap.Error(err))
return nil, err
}
if dbSettings.URI != "" {
logger.Info("Connected successfully", zap.Bool("uri_provided", true))
} else {
logger.Info("Connected successfully", zap.String("user", dbSettings.User),
zap.Strings("hosts", opts.Hosts), zap.String("port", dbSettings.Port),
zap.String("database", dbSettings.Database), zap.String("replica_set", dbSettings.ReplicaSet))
}
if err := client.Ping(context.Background(), readpref.Primary()); err != nil {
logger.Error("Unable to ping database", zap.Error(err))
_ = client.Disconnect(context.Background())
return nil, err
}
return client, nil
}
func ConnectClient(logger mlogger.Logger, settings sharedmodel.SettingsT) (*mongo.Client, *Config, *DBSettings, error) {
config, dbSettings, err := decodeConfig(logger, settings)
if err != nil {
return nil, nil, nil, err
}
client, err := dialMongo(logger, dbSettings)
if err != nil {
return nil, nil, nil, err
}
return client, config, dbSettings, nil
}
// DB represents the structure of the database
type DB struct {
logger mlogger.Logger
config *DBSettings
client *mongo.Client
}
func (db *DB) db() *mongo.Database {
return db.client.Database(db.config.Database)
}
func (db *DB) MongoDatabase() *mongo.Database {
return db.db()
}
func (db *DB) TransactionFactory() transaction.Factory {
return transactionimp.CreateFactory(db.client)
}
func collectReplicaHosts(configuredHosts []string, replicaSet, defaultPort, hostPrefix, portPrefix string) []string {
normalize := func(host, port string) string {
host = strings.TrimSpace(host)
port = strings.TrimSpace(port)
if host == "" {
return ""
}
// If host already has a port, keep it; otherwise apply provided/default port.
if _, _, err := net.SplitHostPort(host); err == nil {
return host
}
if port != "" {
return net.JoinHostPort(host, port)
}
if defaultPort != "" {
return net.JoinHostPort(host, defaultPort)
}
return host
}
appendHost := func(list []string, host, port string) []string {
if normalized := normalize(host, port); normalized != "" {
return append(list, normalized)
}
return list
}
var hosts []string
for _, h := range configuredHosts {
hosts = appendHost(hosts, h, "")
}
if replicaSet == "" || hostPrefix == "" {
return hosts
}
index := 0
for {
hostEnv := os.Getenv(fmt.Sprintf("%s%d", hostPrefix, index))
portEnv := ""
if portPrefix != "" {
portEnv = os.Getenv(fmt.Sprintf("%s%d", portPrefix, index))
}
if hostEnv == "" && index == 0 {
hostEnv = os.Getenv(fmt.Sprintf("%s%d", hostPrefix, 1))
if portPrefix != "" {
portEnv = os.Getenv(fmt.Sprintf("%s%d", portPrefix, 1))
}
if hostEnv == "" {
break
}
index = 1
} else if hostEnv == "" {
break
}
hosts = appendHost(hosts, hostEnv, portEnv)
index++
}
return hosts
}
func (db *DB) CloseConnection() {
if err := db.client.Disconnect(context.Background()); err != nil {
db.logger.Warn("Failed to close connection", zap.Error(err))
}
db.logger.Info("Database connection closed")
}
// NewConnection creates a new database connection
func NewConnection(logger mlogger.Logger, settings sharedmodel.SettingsT) (*DB, error) {
client, _, dbSettings, err := ConnectClient(logger, settings)
if err != nil {
return nil, err
}
db := &DB{
logger: logger.Named("db"),
config: dbSettings,
client: client,
}
return db, nil
}
@@ -0,0 +1,49 @@
package mongo
import (
"net"
"strings"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func buildOptions(s *DBSettings) *options.ClientOptions {
opts := options.Client()
if s.URI != "" {
return opts.ApplyURI(s.URI)
}
hosts := make([]string, 0, len(s.Hosts)+1)
for _, h := range s.Hosts {
if trimmed := strings.TrimSpace(h); trimmed != "" {
hosts = append(hosts, trimmed)
}
}
if len(hosts) == 0 && s.Host != "" {
host := s.Host
if _, _, err := net.SplitHostPort(host); err != nil && s.Port != "" {
host = net.JoinHostPort(host, s.Port)
}
hosts = append(hosts, host)
}
if len(hosts) > 0 {
opts.SetHosts(hosts)
}
if s.ReplicaSet != "" {
opts.SetReplicaSet(s.ReplicaSet)
}
cred := options.Credential{
AuthMechanism: s.AuthMechanism,
AuthSource: s.AuthSource,
Username: s.User,
Password: s.Password,
}
opts.SetAuth(cred)
return opts
}
@@ -0,0 +1,143 @@
package repositoryimp
import (
"context"
"fmt"
ri "git.sendico.io/tech/sxchnge/shared/database/repository/index"
"git.sendico.io/tech/sxchnge/shared/errors"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func (r *MongoRepository) CreateIndex(def *ri.Definition) error {
if r.collection == nil {
return merrors.NoData("data collection is not set")
}
if len(def.Keys) == 0 {
return merrors.InvalidArgument("Index definition has no keys", "index.keys")
}
// ----- build BSON keys --------------------------------------------------
keys := bson.D{}
for _, k := range def.Keys {
var value any
switch {
case k.Type != "":
value = k.Type // text, 2dsphere, …
case k.Sort == ri.Desc:
value = int8(-1)
default:
value = int8(1) // default to Asc
}
keys = append(keys, bson.E{Key: k.Field, Value: value})
}
opts := options.Index().
SetUnique(def.Unique)
if def.TTL != nil {
opts.SetExpireAfterSeconds(*def.TTL)
}
if def.Name != "" {
opts.SetName(def.Name)
}
if def.PartialFilter != nil {
opts.SetPartialFilterExpression(def.PartialFilter.BuildQuery())
}
if def.Sparse {
opts.SetSparse(def.Sparse)
}
_, err := r.collection.Indexes().CreateOne(
context.Background(),
mongo.IndexModel{Keys: keys, Options: opts},
)
return err
}
func (r *MongoRepository) DropIndex(ctx context.Context, name string) error {
if r.collection == nil {
return merrors.NoData("data collection is not set")
}
if name == "" {
return merrors.InvalidArgument("index name is required", "index.name")
}
return r.collection.Indexes().DropOne(ctx, name)
}
func (r *MongoRepository) ListIndexDefinitions(ctx context.Context) ([]*ri.Definition, error) {
if r.collection == nil {
return nil, merrors.NoData("data collection is not set")
}
cursor, err := r.collection.Indexes().List(ctx)
if err != nil {
return nil, err
}
defer func() { _ = cursor.Close(ctx) }()
defs := make([]*ri.Definition, 0)
for cursor.Next(ctx) {
var doc mongoIndexDocument
if err := cursor.Decode(&doc); err != nil {
return nil, err
}
defs = append(defs, indexDefinitionFromDocument(doc))
}
if err := cursor.Err(); err != nil {
return nil, err
}
return defs, nil
}
type mongoIndexDocument struct {
Name string `bson:"name"`
Key bson.D `bson:"key"`
Unique bool `bson:"unique"`
Sparse bool `bson:"sparse"`
ExpireAfterSeconds *int32 `bson:"expireAfterSeconds,omitempty"`
}
func indexDefinitionFromDocument(doc mongoIndexDocument) *ri.Definition {
keys := make([]ri.Key, 0, len(doc.Key))
for _, key := range doc.Key {
keys = append(keys, indexKeyFromBSON(key))
}
return &ri.Definition{
Keys: keys,
Unique: doc.Unique,
Sparse: doc.Sparse,
TTL: doc.ExpireAfterSeconds,
Name: doc.Name,
}
}
func indexKeyFromBSON(key bson.E) ri.Key {
out := ri.Key{Field: key.Key}
switch typed := key.Value.(type) {
case string:
out.Type = ri.IndexType(typed)
case int:
out.Sort = indexSort(typed)
case int8:
out.Sort = indexSort(typed)
case int16:
out.Sort = indexSort(typed)
case int32:
out.Sort = indexSort(typed)
case int64:
out.Sort = indexSort(typed)
case float64:
out.Sort = indexSort(typed)
default:
out.Type = ri.IndexType(fmt.Sprint(typed))
}
return out
}
func indexSort[T ~int | ~int8 | ~int16 | ~int32 | ~int64 | ~float64](value T) ri.Sort {
if value < 0 {
return ri.Desc
}
return ri.Asc
}
@@ -0,0 +1,95 @@
//go:build integration
// +build integration
package repositoryimp_test
import (
"context"
"testing"
"time"
"git.sendico.io/tech/sxchnge/shared/database/repository"
ri "git.sendico.io/tech/sxchnge/shared/database/repository/index"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/modules/mongodb"
"github.com/testcontainers/testcontainers-go/wait"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func TestCreateIndex_WithPartialFilter(t *testing.T) {
startCtx, startCancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer startCancel()
mongoContainer, err := mongodb.Run(startCtx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForListeningPort("27017/tcp").WithStartupTimeout(2*time.Minute)),
)
require.NoError(t, err)
mongoURI, err := mongoContainer.ConnectionString(startCtx)
require.NoError(t, err)
client, err := mongo.Connect(options.Client().ApplyURI(mongoURI))
require.NoError(t, err)
defer client.Disconnect(context.Background())
database := client.Database("test_partial_index_" + t.Name())
defer database.Drop(context.Background())
repo := repository.CreateMongoRepository(database, "partial_index_items")
def := &ri.Definition{
Keys: []ri.Key{
{Field: "field", Sort: ri.Asc},
},
Unique: true,
Name: "partial_unique_field_true",
PartialFilter: repository.Filter("flag", true),
}
require.NoError(t, repo.CreateIndex(def))
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
defs, err := repo.ListIndexDefinitions(ctx)
require.NoError(t, err)
listed := false
for _, listedDef := range defs {
if listedDef.Name == def.Name {
listed = true
assert.True(t, listedDef.Unique)
assert.Equal(t, []ri.Key{{Field: "field", Sort: ri.Asc}}, listedDef.Keys)
}
}
assert.True(t, listed, "created index was not returned by ListIndexDefinitions")
cursor, err := database.Collection(repo.Collection()).Indexes().List(ctx)
require.NoError(t, err)
defer cursor.Close(ctx)
found := false
for cursor.Next(ctx) {
var idx bson.M
require.NoError(t, cursor.Decode(&idx))
if idx["name"] == def.Name {
found = true
assert.Equal(t, true, idx["unique"])
assert.Equal(t, bson.M{"field": int32(1)}, idx["key"])
partial, ok := idx["partialFilterExpression"].(bson.M)
require.True(t, ok)
assert.Equal(t, bson.M{"flag": true}, partial)
}
}
assert.True(t, found, "partial unique index was not created")
termCtx, termCancel := context.WithTimeout(context.Background(), 30*time.Second)
defer termCancel()
_ = mongoContainer.Terminate(termCtx)
}
@@ -0,0 +1,52 @@
package repositoryimp
import (
"testing"
ri "git.sendico.io/tech/sxchnge/shared/database/repository/index"
"go.mongodb.org/mongo-driver/v2/bson"
)
func TestIndexDefinitionFromDocumentMapsIndexMetadata(t *testing.T) {
ttl := int32(3600)
def := indexDefinitionFromDocument(mongoIndexDocument{
Name: "account_ref_unique",
Key: bson.D{{Key: "accountRef", Value: int32(1)}, {Key: "version", Value: int64(-1)}},
Unique: true,
Sparse: true,
ExpireAfterSeconds: &ttl,
})
if def.Name != "account_ref_unique" || !def.Unique || !def.Sparse {
t.Fatalf("unexpected definition metadata: %#v", def)
}
if def.TTL == nil || *def.TTL != ttl {
t.Fatalf("TTL = %v, want %d", def.TTL, ttl)
}
wantKeys := []ri.Key{
{Field: "accountRef", Sort: ri.Asc},
{Field: "version", Sort: ri.Desc},
}
if len(def.Keys) != len(wantKeys) {
t.Fatalf("keys = %#v, want %#v", def.Keys, wantKeys)
}
for i := range wantKeys {
if def.Keys[i] != wantKeys[i] {
t.Fatalf("keys[%d] = %#v, want %#v", i, def.Keys[i], wantKeys[i])
}
}
}
func TestIndexDefinitionFromDocumentMapsTypedIndex(t *testing.T) {
def := indexDefinitionFromDocument(mongoIndexDocument{
Name: "text_index",
Key: bson.D{{Key: "description", Value: "text"}},
})
if len(def.Keys) != 1 {
t.Fatalf("keys = %#v", def.Keys)
}
if def.Keys[0] != (ri.Key{Field: "description", Type: ri.IndexTypeText}) {
t.Fatalf("key = %#v", def.Keys[0])
}
}
@@ -0,0 +1,253 @@
package repositoryimp
import (
"context"
"errors"
"fmt"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
rd "git.sendico.io/tech/sxchnge/shared/database/repository/decoder"
"git.sendico.io/tech/sxchnge/shared/database/transaction"
"git.sendico.io/tech/sxchnge/shared/errors"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
type MongoRepository struct {
collectionName string
collection *mongo.Collection
}
func idFilter(id bson.ObjectID) bson.D {
return bson.D{
{Key: storable.IDField, Value: id},
}
}
func NewMongoRepository(db *mongo.Database, collection string) *MongoRepository {
return &MongoRepository{
collectionName: collection,
collection: db.Collection(collection),
}
}
func (r *MongoRepository) Collection() string {
return r.collectionName
}
func prepareForWrite(obj storable.Storable) bson.ObjectID {
id := obj.GetID()
if id == nil || id.IsZero() {
obj.SetID(bson.NewObjectID())
id = obj.GetID()
}
obj.Update()
return *id
}
func (r *MongoRepository) Insert(ctx context.Context, obj storable.Storable, getFilter builder.Query) error {
prepareForWrite(obj)
_, err := r.collection.InsertOne(ctx, obj)
if mongo.IsDuplicateKeyError(err) {
if getFilter != nil {
if err = r.FindOneByFilter(ctx, getFilter, obj); err != nil {
return err
}
}
return merrors.DataConflict("duplicate_key")
}
return err
}
func (r *MongoRepository) InsertMany(ctx context.Context, objects []storable.Storable) error {
if len(objects) == 0 {
return nil
}
docs := make([]interface{}, len(objects))
for i, obj := range objects {
prepareForWrite(obj)
docs[i] = obj
}
_, err := r.collection.InsertMany(ctx, docs)
return err
}
func (r *MongoRepository) findOneByFilterImp(ctx context.Context, filter bson.D, errMessage string, result storable.Storable) error {
err := r.collection.FindOne(ctx, filter).Decode(result)
if errors.Is(err, mongo.ErrNoDocuments) {
return merrors.NoData(errMessage)
}
return err
}
func (r *MongoRepository) Get(ctx context.Context, id bson.ObjectID, result storable.Storable) error {
if id.IsZero() {
return merrors.InvalidArgument("zero id provided while fetching "+result.Collection(), "id")
}
return r.findOneByFilterImp(ctx, idFilter(id), fmt.Sprintf("%s with ID = %s not found", result.Collection(), id.Hex()), result)
}
type QueryFunc func(ctx context.Context, collection *mongo.Collection) (*mongo.Cursor, error)
func (r *MongoRepository) executeQuery(ctx context.Context, queryFunc QueryFunc, decoder rd.DecodingFunc) error {
cursor, err := queryFunc(ctx, r.collection)
if errors.Is(err, mongo.ErrNoDocuments) {
return merrors.NoData("no_items_in_array")
}
if err != nil {
return err
}
defer func() {
_ = cursor.Close(ctx)
}()
for cursor.Next(ctx) {
if err = decoder(cursor); err != nil {
return err
}
}
return nil
}
func (r *MongoRepository) Aggregate(ctx context.Context, pipeline builder.Pipeline, decoder rd.DecodingFunc) error {
queryFunc := func(ctx context.Context, collection *mongo.Collection) (*mongo.Cursor, error) {
return collection.Aggregate(ctx, pipeline.Build())
}
return r.executeQuery(ctx, queryFunc, decoder)
}
func (r *MongoRepository) FindManyByFilter(ctx context.Context, query builder.Query, decoder rd.DecodingFunc) error {
queryFunc := func(ctx context.Context, collection *mongo.Collection) (*mongo.Cursor, error) {
return collection.Find(ctx, query.BuildQuery(), query.BuildOptions())
}
return r.executeQuery(ctx, queryFunc, decoder)
}
func (r *MongoRepository) FindAllByFilter(ctx context.Context, query builder.Query, result any) error {
if query == nil {
cursor, err := r.collection.Find(ctx, bson.D{})
if err != nil {
return err
}
defer func() {
_ = cursor.Close(ctx)
}()
return cursor.All(ctx, result)
}
cursor, err := r.collection.Find(ctx, query.BuildQuery(), query.BuildOptions())
if err != nil {
return err
}
defer func() {
_ = cursor.Close(ctx)
}()
return cursor.All(ctx, result)
}
func (r *MongoRepository) FindOneByFilter(ctx context.Context, query builder.Query, result storable.Storable) error {
return r.findOneByFilterImp(ctx, query.BuildQuery(), result.Collection()+" not found by filter", result)
}
func (r *MongoRepository) Update(ctx context.Context, obj storable.Storable) error {
obj.Update()
return r.collection.FindOneAndReplace(ctx, idFilter(*obj.GetID()), obj).Err()
}
func (r *MongoRepository) ReplaceOneByFilter(ctx context.Context, query builder.Query, obj storable.Storable) (bool, error) {
if query == nil {
return false, merrors.InvalidArgument("filter is required", "filter")
}
if obj == nil {
return false, merrors.InvalidArgument("object is required", "obj")
}
result, err := r.collection.ReplaceOne(ctx, query.BuildQuery(), obj)
if mongo.IsDuplicateKeyError(err) {
return false, merrors.DataConflict("duplicate_key")
}
if err != nil {
return false, err
}
return result.MatchedCount > 0, nil
}
func (r *MongoRepository) Upsert(ctx context.Context, obj storable.Storable) error {
id := prepareForWrite(obj)
_, err := r.collection.ReplaceOne(ctx, idFilter(id), obj, options.Replace().SetUpsert(true))
return err
}
func (r *MongoRepository) Patch(ctx context.Context, id bson.ObjectID, patch builder.Patch) error {
if id.IsZero() {
return merrors.InvalidArgument("zero id provided while patching", "id")
}
_, err := r.collection.UpdateByID(ctx, id, patch.Build())
return err
}
func (r *MongoRepository) PatchMany(ctx context.Context, query builder.Query, patch builder.Patch) (int, error) {
result, err := r.collection.UpdateMany(ctx, query.BuildQuery(), patch.Build())
if err != nil {
return 0, err
}
return int(result.ModifiedCount), nil
}
func (r *MongoRepository) ListIDs(ctx context.Context, query builder.Query) ([]bson.ObjectID, error) {
filter := query.BuildQuery()
findOptions := query.BuildOptions().SetProjection(bson.M{storable.IDField: 1})
cursor, err := r.collection.Find(ctx, filter, findOptions)
if err != nil {
return nil, err
}
defer func() {
_ = cursor.Close(ctx)
}()
var ids []bson.ObjectID
for cursor.Next(ctx) {
var doc struct {
ID bson.ObjectID `bson:"_id"`
}
if err := cursor.Decode(&doc); err != nil {
return nil, err
}
ids = append(ids, doc.ID)
}
if err := cursor.Err(); err != nil {
return nil, err
}
return ids, nil
}
func (r *MongoRepository) Delete(ctx context.Context, id bson.ObjectID) error {
_, err := r.collection.DeleteOne(ctx, idFilter(id))
return err
}
func (r *MongoRepository) DeleteMany(ctx context.Context, query builder.Query) error {
_, err := r.collection.DeleteMany(ctx, query.BuildQuery())
return err
}
func (r *MongoRepository) Name() string {
return r.collection.Name()
}
func (r *MongoRepository) WithTransaction(ctx context.Context, cb transaction.Callback) (any, error) {
if cb == nil {
return nil, merrors.InvalidArgument("transaction callback is required", "callback")
}
session, err := r.collection.Database().Client().StartSession()
if err != nil {
return nil, err
}
defer session.EndSession(ctx)
return session.WithTransaction(ctx, cb)
}
@@ -0,0 +1,510 @@
//go:build integration
// +build integration
package repositoryimp_test
import (
"context"
"errors"
"testing"
"time"
"git.sendico.io/tech/sxchnge/shared/database/internal/builder"
"git.sendico.io/tech/sxchnge/shared/database/internal/repository"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"git.sendico.io/tech/sxchnge/shared/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/modules/mongodb"
"github.com/testcontainers/testcontainers-go/wait"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func TestMongoRepository_Insert(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("Insert_WithoutID", func(t *testing.T) {
testObj := &TestObject{Name: "testInsert"}
// ID should be nil/zero initially
assert.True(t, testObj.GetID().IsZero())
err := repository.Insert(ctx, testObj, nil)
require.NoError(t, err)
// ID should be assigned after insert
assert.False(t, testObj.GetID().IsZero())
assert.NotEmpty(t, testObj.CreatedAt)
assert.NotEmpty(t, testObj.UpdatedAt)
})
t.Run("Insert_WithExistingID", func(t *testing.T) {
existingID := bson.NewObjectID()
testObj := &TestObject{Name: "testInsertWithID"}
testObj.SetID(existingID)
err := repository.Insert(ctx, testObj, nil)
require.NoError(t, err)
// ID should remain the same
assert.Equal(t, existingID, *testObj.GetID())
})
t.Run("Insert_DuplicateKey", func(t *testing.T) {
// Insert first object
testObj1 := &TestObject{Name: "duplicate"}
err := repository.Insert(ctx, testObj1, nil)
require.NoError(t, err)
// Try to insert object with same ID
testObj2 := &TestObject{Name: "duplicate2"}
testObj2.SetID(*testObj1.GetID())
err = repository.Insert(ctx, testObj2, nil)
assert.Error(t, err)
assert.True(t, errors.Is(err, merrors.ErrDataConflict))
})
t.Run("Insert_DuplicateKeyWithGetFilter", func(t *testing.T) {
// Insert first object
testObj1 := &TestObject{Name: "duplicateWithFilter"}
err := repository.Insert(ctx, testObj1, nil)
require.NoError(t, err)
// Try to insert object with same ID, but with getFilter
testObj2 := &TestObject{Name: "duplicateWithFilter2"}
testObj2.SetID(*testObj1.GetID())
getFilter := builderimp.NewQueryImp().Comparison(builderimp.NewFieldImp("_id"), builder.Eq, *testObj1.GetID())
err = repository.Insert(ctx, testObj2, getFilter)
assert.Error(t, err)
assert.True(t, errors.Is(err, merrors.ErrDataConflict))
// But testObj2 should be populated with the existing object data
assert.Equal(t, testObj1.Name, testObj2.Name)
})
}
func TestMongoRepository_Update(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("Update_ExistingObject", func(t *testing.T) {
// Insert object first
testObj := &TestObject{Name: "originalName"}
err := repository.Insert(ctx, testObj, nil)
require.NoError(t, err)
originalUpdatedAt := testObj.UpdatedAt
// Update the object
testObj.Name = "updatedName"
time.Sleep(10 * time.Millisecond) // Ensure time difference
err = repository.Update(ctx, testObj)
require.NoError(t, err)
// Verify the object was updated
result := &TestObject{}
err = repository.Get(ctx, *testObj.GetID(), result)
require.NoError(t, err)
assert.Equal(t, "updatedName", result.Name)
assert.True(t, result.UpdatedAt.After(originalUpdatedAt))
})
t.Run("Update_NonExistentObject", func(t *testing.T) {
nonExistentID := bson.NewObjectID()
testObj := &TestObject{Name: "nonExistent"}
testObj.SetID(nonExistentID)
err := repository.Update(ctx, testObj)
assert.Error(t, err)
assert.True(t, errors.Is(err, mongo.ErrNoDocuments))
})
}
func TestMongoRepository_Delete(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("Delete_ExistingObject", func(t *testing.T) {
// Insert object first
testObj := &TestObject{Name: "toDelete"}
err := repository.Insert(ctx, testObj, nil)
require.NoError(t, err)
// Delete the object
err = repository.Delete(ctx, *testObj.GetID())
require.NoError(t, err)
// Verify the object was deleted
result := &TestObject{}
err = repository.Get(ctx, *testObj.GetID(), result)
assert.Error(t, err)
assert.True(t, errors.Is(err, merrors.ErrNoData))
})
t.Run("Delete_NonExistentObject", func(t *testing.T) {
nonExistentID := bson.NewObjectID()
err := repository.Delete(ctx, nonExistentID)
// Delete should not return error even if object doesn't exist
assert.NoError(t, err)
})
}
func TestMongoRepository_FindOneByFilter(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("FindOneByFilter_MatchingFilter", func(t *testing.T) {
// Insert test objects
testObjs := []*TestObject{
{Name: "findMe"},
{Name: "dontFindMe"},
{Name: "findMeToo"},
}
for _, obj := range testObjs {
err := repository.Insert(ctx, obj, nil)
require.NoError(t, err)
}
// Find by filter
query := builderimp.NewQueryImp().Comparison(builderimp.NewFieldImp("name"), builder.Eq, "findMe")
result := &TestObject{}
err := repository.FindOneByFilter(ctx, query, result)
require.NoError(t, err)
assert.Equal(t, "findMe", result.Name)
})
t.Run("FindOneByFilter_NoMatch", func(t *testing.T) {
query := builderimp.NewQueryImp().Comparison(builderimp.NewFieldImp("name"), builder.Eq, "nonExistentName")
result := &TestObject{}
err := repository.FindOneByFilter(ctx, query, result)
assert.Error(t, err)
assert.True(t, errors.Is(err, merrors.ErrNoData))
})
}
func TestMongoRepository_FindManyByFilter(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("FindManyByFilter_MultipleResults", func(t *testing.T) {
// Insert test objects
testObjs := []*TestObject{
{Name: "findMany1"},
{Name: "findMany2"},
{Name: "dontFind"},
}
for _, obj := range testObjs {
err := repository.Insert(ctx, obj, nil)
require.NoError(t, err)
}
// Find objects with names starting with "findMany"
query := builderimp.NewQueryImp().RegEx(builderimp.NewFieldImp("name"), "^findMany", "")
var results []*TestObject
decoder := func(cursor *mongo.Cursor) error {
var obj TestObject
if err := cursor.Decode(&obj); err != nil {
return err
}
results = append(results, &obj)
return nil
}
err := repository.FindManyByFilter(ctx, query, decoder)
require.NoError(t, err)
assert.Len(t, results, 2)
names := make([]string, len(results))
for i, obj := range results {
names[i] = obj.Name
}
assert.Contains(t, names, "findMany1")
assert.Contains(t, names, "findMany2")
})
t.Run("FindManyByFilter_NoResults", func(t *testing.T) {
query := builderimp.NewQueryImp().Comparison(builderimp.NewFieldImp("name"), builder.Eq, "nonExistentPattern")
var results []*TestObject
decoder := func(cursor *mongo.Cursor) error {
var obj TestObject
if err := cursor.Decode(&obj); err != nil {
return err
}
results = append(results, &obj)
return nil
}
err := repository.FindManyByFilter(ctx, query, decoder)
require.NoError(t, err)
assert.Empty(t, results)
})
}
func TestMongoRepository_DeleteMany(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("DeleteMany_MultipleDocuments", func(t *testing.T) {
// Insert test objects
testObjs := []*TestObject{
{Name: "deleteMany1"},
{Name: "deleteMany2"},
{Name: "keepMe"},
}
for _, obj := range testObjs {
err := repository.Insert(ctx, obj, nil)
require.NoError(t, err)
}
// Delete objects with names starting with "deleteMany"
query := builderimp.NewQueryImp().RegEx(builderimp.NewFieldImp("name"), "^deleteMany", "")
err := repository.DeleteMany(ctx, query)
require.NoError(t, err)
// Verify deletions
queryAll := builderimp.NewQueryImp()
var results []*TestObject
decoder := func(cursor *mongo.Cursor) error {
var obj TestObject
if err := cursor.Decode(&obj); err != nil {
return err
}
results = append(results, &obj)
return nil
}
err = repository.FindManyByFilter(ctx, queryAll, decoder)
require.NoError(t, err)
assert.Len(t, results, 1)
assert.Equal(t, "keepMe", results[0].Name)
})
}
func TestMongoRepository_Name(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "mycollection")
t.Run("Name_ReturnsCollectionName", func(t *testing.T) {
name := repository.Name()
assert.Equal(t, "mycollection", name)
})
}
func TestMongoRepository_UpdateTimestamp(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("Update_Should_Update_Timestamp", func(t *testing.T) {
// Create test object
obj := &TestObject{
Name: "Test Object",
}
// Set ID and initial timestamps
obj.SetID(bson.NewObjectID())
originalCreatedAt := obj.CreatedAt
originalUpdatedAt := obj.UpdatedAt
// Insert the object
err := repository.Insert(ctx, obj, nil)
require.NoError(t, err)
// Wait a moment to ensure timestamp difference
time.Sleep(10 * time.Millisecond)
// Update the object
obj.Name = "Updated Object"
err = repository.Update(ctx, obj)
require.NoError(t, err)
// Verify timestamps
assert.Equal(t, originalCreatedAt, obj.CreatedAt, "CreatedAt should not change")
assert.True(t, obj.UpdatedAt.After(originalUpdatedAt), "UpdatedAt should be updated")
// Verify the object was actually updated in the database
var retrieved TestObject
err = repository.Get(ctx, *obj.GetID(), &retrieved)
require.NoError(t, err)
assert.Equal(t, "Updated Object", retrieved.Name, "Name should be updated")
assert.WithinDuration(t, originalCreatedAt, retrieved.CreatedAt, time.Second, "CreatedAt should not change in DB")
assert.True(t, retrieved.UpdatedAt.After(originalUpdatedAt), "UpdatedAt should be updated in DB")
assert.WithinDuration(t, obj.UpdatedAt, retrieved.UpdatedAt, time.Second, "UpdatedAt should match between object and DB")
})
}
@@ -0,0 +1,59 @@
package repositoryimp
import (
"context"
"errors"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"git.sendico.io/tech/sxchnge/shared/errors"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func (r *MongoRepository) FindFirstByFilter(ctx context.Context, query builder.Query, result storable.Storable) error {
opts := query.BuildOptions().SetLimit(1)
cursor, err := r.collection.Find(ctx, query.BuildQuery(), opts)
if err != nil {
return err
}
defer func() {
_ = cursor.Close(ctx)
}()
if !cursor.Next(ctx) {
if err := cursor.Err(); err != nil {
return err
}
return merrors.NoData(result.Collection() + " not found by filter")
}
return cursor.Decode(result)
}
func (r *MongoRepository) CountByFilter(ctx context.Context, query builder.Query) (int64, error) {
return r.collection.CountDocuments(ctx, query.BuildQuery())
}
func (r *MongoRepository) PatchOneByFilter(ctx context.Context, query builder.Query, patch builder.Patch, result storable.Storable) (bool, error) {
if query == nil {
return false, merrors.InvalidArgument("filter is required", "filter")
}
if patch == nil {
return false, merrors.InvalidArgument("patch is required", "patch")
}
if result == nil {
return false, merrors.InvalidArgument("result is required", "result")
}
err := r.collection.FindOneAndUpdate(
ctx,
query.BuildQuery(),
patch.Build(),
options.FindOneAndUpdate().SetReturnDocument(options.After),
).Decode(result)
if errors.Is(err, mongo.ErrNoDocuments) {
return false, nil
}
if err != nil {
return false, err
}
return true, nil
}
@@ -0,0 +1,153 @@
//go:build integration
// +build integration
package repositoryimp_test
import (
"context"
"testing"
"time"
"git.sendico.io/tech/sxchnge/shared/database/internal/repository"
"git.sendico.io/tech/sxchnge/shared/database/record"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/modules/mongodb"
"github.com/testcontainers/testcontainers-go/wait"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func TestMongoRepository_InsertMany(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("InsertMany_Success", func(t *testing.T) {
objects := []storable.Storable{
&TestObject{Name: "test1"},
&TestObject{Name: "test2"},
&TestObject{Name: "test3"},
}
err := repository.InsertMany(ctx, objects)
require.NoError(t, err)
// Verify all objects were inserted and have IDs
for _, obj := range objects {
assert.NotNil(t, obj.GetID())
assert.False(t, obj.GetID().IsZero())
// Verify we can retrieve each object
result := &TestObject{}
err := repository.Get(ctx, *obj.GetID(), result)
require.NoError(t, err)
assert.Equal(t, obj.(*TestObject).Name, result.Name)
}
})
t.Run("InsertMany_EmptySlice", func(t *testing.T) {
objects := []storable.Storable{}
err := repository.InsertMany(ctx, objects)
require.NoError(t, err)
})
t.Run("InsertMany_WithExistingIDs", func(t *testing.T) {
id1 := bson.NewObjectID()
id2 := bson.NewObjectID()
objects := []storable.Storable{
&TestObject{Base: storable.Base{ID: id1}, Name: "preassigned1"},
&TestObject{Base: storable.Base{ID: id2}, Name: "preassigned2"},
}
err := repository.InsertMany(ctx, objects)
require.NoError(t, err)
// Verify objects were inserted with pre-assigned IDs
result1 := &TestObject{}
err = repository.Get(ctx, id1, result1)
require.NoError(t, err)
assert.Equal(t, "preassigned1", result1.Name)
result2 := &TestObject{}
err = repository.Get(ctx, id2, result2)
require.NoError(t, err)
assert.Equal(t, "preassigned2", result2.Name)
})
t.Run("InsertMany_MixedTypes", func(t *testing.T) {
objects := []storable.Storable{
&TestObject{Name: "test1"},
&AnotherObject{Description: "desc1"},
&TestObject{Name: "test2"},
}
err := repository.InsertMany(ctx, objects)
require.NoError(t, err)
// Verify all objects were inserted
for _, obj := range objects {
assert.NotNil(t, obj.GetID())
assert.False(t, obj.GetID().IsZero())
}
})
t.Run("InsertMany_DuplicateKey", func(t *testing.T) {
id := bson.NewObjectID()
// Insert first object
obj1 := &TestObject{Base: storable.Base{ID: id}, Name: "original"}
err := repository.Insert(ctx, obj1, nil)
require.NoError(t, err)
// Try to insert multiple objects including one with duplicate ID
objects := []storable.Storable{
&TestObject{Name: "test1"},
&TestObject{Base: storable.Base{ID: id}, Name: "duplicate"},
}
err = repository.InsertMany(ctx, objects)
assert.Error(t, err)
assert.True(t, mongo.IsDuplicateKeyError(err))
})
t.Run("InsertMany_UpdateTimestamps", func(t *testing.T) {
objects := []storable.Storable{
&TestObject{Name: "test1"},
&TestObject{Name: "test2"},
}
err := repository.InsertMany(ctx, objects)
require.NoError(t, err)
// Verify timestamps were set
for _, obj := range objects {
testObj := obj.(*TestObject)
assert.NotZero(t, testObj.CreatedAt)
assert.NotZero(t, testObj.UpdatedAt)
}
})
}
@@ -0,0 +1,288 @@
//go:build integration
// +build integration
package repositoryimp_test
import (
"context"
"errors"
"testing"
"time"
"git.sendico.io/tech/sxchnge/shared/database/internal/repository"
"git.sendico.io/tech/sxchnge/shared/database/repository"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"git.sendico.io/tech/sxchnge/shared/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/modules/mongodb"
"github.com/testcontainers/testcontainers-go/wait"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func TestMongoRepository_PatchOperations(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repo := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("Patch_SingleDocument", func(t *testing.T) {
obj := &TestObject{Name: "old"}
err := repo.Insert(ctx, obj, nil)
require.NoError(t, err)
original := obj.UpdatedAt
patch := repository.Patch().Set(repository.Field("name"), "new")
err = repo.Patch(ctx, *obj.GetID(), patch)
require.NoError(t, err)
var result TestObject
err = repo.Get(ctx, *obj.GetID(), &result)
require.NoError(t, err)
assert.Equal(t, "new", result.Name)
assert.True(t, result.UpdatedAt.After(original))
})
t.Run("PatchMany_MultipleDocuments", func(t *testing.T) {
objs := []*TestObject{{Name: "match"}, {Name: "match"}, {Name: "other"}}
for _, o := range objs {
err := repo.Insert(ctx, o, nil)
require.NoError(t, err)
}
query := repository.Query().Comparison(repository.Field("name"), builder.Eq, "match")
patch := repository.Patch().Set(repository.Field("name"), "patched")
modified, err := repo.PatchMany(ctx, query, patch)
require.NoError(t, err)
assert.Equal(t, 2, modified)
verify := repository.Query().Comparison(repository.Field("name"), builder.Eq, "patched")
var results []TestObject
decoder := func(cursor *mongo.Cursor) error {
var obj TestObject
if err := cursor.Decode(&obj); err != nil {
return err
}
results = append(results, obj)
return nil
}
err = repo.FindManyByFilter(ctx, verify, decoder)
require.NoError(t, err)
assert.Len(t, results, 2)
})
t.Run("PatchOneByFilter_MatchesAndReturnsUpdatedDocument", func(t *testing.T) {
obj := &TestObject{Name: "patch-one-match", Tags: []string{"before"}}
err := repo.Insert(ctx, obj, nil)
require.NoError(t, err)
originalUpdatedAt := obj.UpdatedAt
query := repository.Query().Filter(repository.Field("name"), "patch-one-match")
patch := repository.Patch().
Set(repository.Field("name"), "patch-one-updated").
Push(repository.Field("tags"), "after")
var result TestObject
matched, err := repo.PatchOneByFilter(ctx, query, patch, &result)
require.NoError(t, err)
require.True(t, matched)
require.Equal(t, *obj.GetID(), *result.GetID())
require.Equal(t, "patch-one-updated", result.Name)
require.Equal(t, []string{"before", "after"}, result.Tags)
require.True(t, result.UpdatedAt.After(originalUpdatedAt))
var persisted TestObject
err = repo.Get(ctx, *obj.GetID(), &persisted)
require.NoError(t, err)
require.Equal(t, result.Name, persisted.Name)
require.Equal(t, result.Tags, persisted.Tags)
})
t.Run("PatchOneByFilter_NoMatch", func(t *testing.T) {
patch := repository.Patch().Set(repository.Field("name"), "not-applied")
var result TestObject
matched, err := repo.PatchOneByFilter(ctx, repository.Query().Filter(repository.Field("name"), "missing-patch-one"), patch, &result)
require.NoError(t, err)
require.False(t, matched)
require.Empty(t, result.Name)
})
t.Run("PatchOneByFilter_InvalidArguments", func(t *testing.T) {
var result TestObject
_, err := repo.PatchOneByFilter(ctx, nil, repository.Patch().Set(repository.Field("name"), "x"), &result)
require.Error(t, err)
require.True(t, errors.Is(err, merrors.ErrInvalidArg))
_, err = repo.PatchOneByFilter(ctx, repository.Query().Filter(repository.Field("name"), "x"), nil, &result)
require.Error(t, err)
require.True(t, errors.Is(err, merrors.ErrInvalidArg))
_, err = repo.PatchOneByFilter(ctx, repository.Query().Filter(repository.Field("name"), "x"), repository.Patch().Set(repository.Field("name"), "y"), nil)
require.Error(t, err)
require.True(t, errors.Is(err, merrors.ErrInvalidArg))
})
t.Run("Patch_PushArray", func(t *testing.T) {
obj := &TestObject{Name: "test", Tags: []string{"tag1"}}
err := repo.Insert(ctx, obj, nil)
require.NoError(t, err)
patch := repository.Patch().Push(repository.Field("tags"), "tag2")
err = repo.Patch(ctx, *obj.GetID(), patch)
require.NoError(t, err)
var result TestObject
err = repo.Get(ctx, *obj.GetID(), &result)
require.NoError(t, err)
assert.Equal(t, []string{"tag1", "tag2"}, result.Tags)
})
t.Run("Patch_PullArray", func(t *testing.T) {
obj := &TestObject{Name: "test", Tags: []string{"tag1", "tag2", "tag3"}}
err := repo.Insert(ctx, obj, nil)
require.NoError(t, err)
patch := repository.Patch().Pull(repository.Field("tags"), "tag2")
err = repo.Patch(ctx, *obj.GetID(), patch)
require.NoError(t, err)
var result TestObject
err = repo.Get(ctx, *obj.GetID(), &result)
require.NoError(t, err)
assert.Equal(t, []string{"tag1", "tag3"}, result.Tags)
})
t.Run("Patch_AddToSetArray", func(t *testing.T) {
obj := &TestObject{Name: "test", Tags: []string{"tag1"}}
err := repo.Insert(ctx, obj, nil)
require.NoError(t, err)
// Add new tag
patch := repository.Patch().AddToSet(repository.Field("tags"), "tag2")
err = repo.Patch(ctx, *obj.GetID(), patch)
require.NoError(t, err)
var result TestObject
err = repo.Get(ctx, *obj.GetID(), &result)
require.NoError(t, err)
assert.Equal(t, []string{"tag1", "tag2"}, result.Tags)
// Try to add duplicate tag - should not add
patch = repository.Patch().AddToSet(repository.Field("tags"), "tag1")
err = repo.Patch(ctx, *obj.GetID(), patch)
require.NoError(t, err)
err = repo.Get(ctx, *obj.GetID(), &result)
require.NoError(t, err)
assert.Equal(t, []string{"tag1", "tag2"}, result.Tags)
})
t.Run("Patch_PushToEmptyArray", func(t *testing.T) {
obj := &TestObject{Name: "test", Tags: []string{}}
err := repo.Insert(ctx, obj, nil)
require.NoError(t, err)
patch := repository.Patch().Push(repository.Field("tags"), "tag1")
err = repo.Patch(ctx, *obj.GetID(), patch)
require.NoError(t, err)
var result TestObject
err = repo.Get(ctx, *obj.GetID(), &result)
require.NoError(t, err)
assert.Equal(t, []string{"tag1"}, result.Tags)
})
t.Run("Patch_PullFromEmptyArray", func(t *testing.T) {
obj := &TestObject{Name: "test", Tags: []string{}}
err := repo.Insert(ctx, obj, nil)
require.NoError(t, err)
patch := repository.Patch().Pull(repository.Field("tags"), "nonexistent")
err = repo.Patch(ctx, *obj.GetID(), patch)
require.NoError(t, err)
var result TestObject
err = repo.Get(ctx, *obj.GetID(), &result)
require.NoError(t, err)
assert.Equal(t, []string{}, result.Tags)
})
t.Run("Patch_PullNonExistentElement", func(t *testing.T) {
obj := &TestObject{Name: "test", Tags: []string{"tag1", "tag2"}}
err := repo.Insert(ctx, obj, nil)
require.NoError(t, err)
patch := repository.Patch().Pull(repository.Field("tags"), "nonexistent")
err = repo.Patch(ctx, *obj.GetID(), patch)
require.NoError(t, err)
var result TestObject
err = repo.Get(ctx, *obj.GetID(), &result)
require.NoError(t, err)
assert.Equal(t, []string{"tag1", "tag2"}, result.Tags)
})
t.Run("Patch_ChainedArrayOperations", func(t *testing.T) {
obj := &TestObject{Name: "test", Tags: []string{"tag1"}}
err := repo.Insert(ctx, obj, nil)
require.NoError(t, err)
// Note: MongoDB doesn't allow multiple operations on the same array field in a single update
// This test demonstrates that chained array operations on the same field will fail
patch := repository.Patch().
Push(repository.Field("tags"), "tag2").
AddToSet(repository.Field("tags"), "tag3").
Pull(repository.Field("tags"), "tag1")
err = repo.Patch(ctx, *obj.GetID(), patch)
require.Error(t, err) // This should fail due to MongoDB's limitation
assert.Contains(t, err.Error(), "conflict")
})
t.Run("PatchMany_ArrayOperations", func(t *testing.T) {
objs := []*TestObject{
{Name: "obj1", Tags: []string{"tag1"}},
{Name: "obj2", Tags: []string{"tag2"}},
{Name: "obj3", Tags: []string{"tag3"}},
}
for _, o := range objs {
err := repo.Insert(ctx, o, nil)
require.NoError(t, err)
}
query := repository.Query().Comparison(repository.Field("name"), builder.In, []string{"obj1", "obj2"})
patch := repository.Patch().Push(repository.Field("tags"), "common")
modified, err := repo.PatchMany(ctx, query, patch)
require.NoError(t, err)
assert.Equal(t, 2, modified)
// Verify the changes
for _, name := range []string{"obj1", "obj2"} {
var result TestObject
err = repo.FindOneByFilter(ctx, repository.Query().Comparison(repository.Field("name"), builder.Eq, name), &result)
require.NoError(t, err)
assert.Contains(t, result.Tags, "common")
}
})
}
@@ -0,0 +1,188 @@
//go:build integration
// +build integration
package repositoryimp_test
import (
"context"
"errors"
"testing"
"time"
"git.sendico.io/tech/sxchnge/shared/database/internal/builder"
"git.sendico.io/tech/sxchnge/shared/database/internal/repository"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"git.sendico.io/tech/sxchnge/shared/errors"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/modules/mongodb"
"github.com/testcontainers/testcontainers-go/wait"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
type TestObject struct {
storable.Base `bson:",inline" json:",inline"`
Name string `bson:"name"`
Tags []string `bson:"tags"`
}
func (t *TestObject) Collection() string {
return "testObject"
}
type AnotherObject struct {
storable.Base `bson:",inline" json:",inline"`
Description string `bson:"description"`
}
func (a *AnotherObject) Collection() string {
return "anotherObject"
}
func terminate(ctx context.Context, t *testing.T, container *mongodb.MongoDBContainer) {
err := container.Terminate(ctx)
require.NoError(t, err, "failed to terminate MongoDB container")
}
func disconnect(ctx context.Context, t *testing.T, client *mongo.Client) {
err := client.Disconnect(ctx)
require.NoError(t, err, "failed to disconnect from MongoDB")
}
func TestMongoRepository_Get(t *testing.T) {
// Use a context with timeout, so if container spinning or DB ops hang,
// the test won't run indefinitely.
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("Get_Success", func(t *testing.T) {
testObj := &TestObject{Name: "testName"}
err := repository.Insert(ctx, testObj, nil)
require.NoError(t, err)
result := &TestObject{}
err = repository.Get(ctx, testObj.ID, result)
require.NoError(t, err)
assert.Equal(t, testObj.Name, result.Name)
assert.Equal(t, testObj.ID, result.ID)
})
t.Run("Get_NotFound", func(t *testing.T) {
nonExistentID := bson.NewObjectID()
result := &TestObject{}
err := repository.Get(ctx, nonExistentID, result)
assert.Error(t, err)
assert.True(t, errors.Is(err, merrors.ErrNoData))
})
t.Run("Get_InvalidID", func(t *testing.T) {
invalidID := bson.ObjectID{} // zero value
result := &TestObject{}
err := repository.Get(ctx, invalidID, result)
assert.Error(t, err)
assert.True(t, errors.Is(err, merrors.ErrInvalidArg))
})
t.Run("Get_DifferentTypes", func(t *testing.T) {
anotherObj := &AnotherObject{Description: "testDescription"}
err := repository.Insert(ctx, anotherObj, nil)
require.NoError(t, err)
result := &AnotherObject{}
err = repository.Get(ctx, anotherObj.ID, result)
require.NoError(t, err)
assert.Equal(t, anotherObj.Description, result.Description)
assert.Equal(t, anotherObj.ID, result.ID)
})
}
func TestMongoRepository_ListIDs(t *testing.T) {
// Use a context with timeout, so if container spinning or DB ops hang,
// the test won't run indefinitely.
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repository := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("ListIDs_Success", func(t *testing.T) {
// Insert test data
testObjs := []*TestObject{
{Name: "testName1"},
{Name: "testName2"},
{Name: "testName3"},
}
for _, obj := range testObjs {
err := repository.Insert(ctx, obj, nil)
require.NoError(t, err)
}
// Define a query to match all objects
query := builderimp.NewQueryImp()
// Call ListIDs
ids, err := repository.ListIDs(ctx, query)
require.NoError(t, err)
// Assert the IDs are correct
require.Len(t, ids, len(testObjs))
for _, obj := range testObjs {
assert.Contains(t, ids, obj.ID)
}
})
t.Run("ListIDs_EmptyResult", func(t *testing.T) {
// Define a query that matches no objects
query := builderimp.NewQueryImp().Comparison(builderimp.NewFieldImp("name"), builder.Eq, "nonExistentName")
// Call ListIDs
ids, err := repository.ListIDs(ctx, query)
require.NoError(t, err)
// Assert no IDs are returned
assert.Empty(t, ids)
})
}
@@ -0,0 +1,99 @@
//go:build integration
// +build integration
package repositoryimp_test
import (
"context"
"testing"
"time"
"git.sendico.io/tech/sxchnge/shared/database/internal/repository"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"github.com/testcontainers/testcontainers-go"
"github.com/testcontainers/testcontainers-go/modules/mongodb"
"github.com/testcontainers/testcontainers-go/wait"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
func TestMongoRepository_Upsert(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
mongoContainer, err := mongodb.Run(ctx,
"mongo:latest",
mongodb.WithUsername("root"),
mongodb.WithPassword("password"),
testcontainers.WithWaitStrategy(wait.ForLog("Waiting for connections")),
)
require.NoError(t, err, "failed to start MongoDB container")
defer terminate(ctx, t, mongoContainer)
mongoURI, err := mongoContainer.ConnectionString(ctx)
require.NoError(t, err, "failed to get MongoDB connection string")
clientOptions := options.Client().ApplyURI(mongoURI)
client, err := mongo.Connect(clientOptions)
require.NoError(t, err, "failed to connect to MongoDB")
defer disconnect(ctx, t, client)
db := client.Database("testdb")
repo := repositoryimp.NewMongoRepository(db, "testcollection")
t.Run("Upsert_InsertsWhenIDIsZero", func(t *testing.T) {
obj := &TestObject{Name: "inserted-via-upsert"}
require.True(t, obj.GetID().IsZero())
err := repo.Upsert(ctx, obj)
require.NoError(t, err)
assert.False(t, obj.GetID().IsZero())
assert.NotZero(t, obj.CreatedAt)
assert.NotZero(t, obj.UpdatedAt)
var stored TestObject
err = repo.Get(ctx, *obj.GetID(), &stored)
require.NoError(t, err)
assert.Equal(t, "inserted-via-upsert", stored.Name)
})
t.Run("Upsert_UpdatesExistingDocument", func(t *testing.T) {
obj := &TestObject{Name: "before-upsert"}
err := repo.Insert(ctx, obj, nil)
require.NoError(t, err)
createdAt := obj.CreatedAt
updatedAt := obj.UpdatedAt
time.Sleep(10 * time.Millisecond)
obj.Name = "after-upsert"
err = repo.Upsert(ctx, obj)
require.NoError(t, err)
assert.Equal(t, createdAt, obj.CreatedAt)
assert.True(t, obj.UpdatedAt.After(updatedAt))
var stored TestObject
err = repo.Get(ctx, *obj.GetID(), &stored)
require.NoError(t, err)
assert.Equal(t, "after-upsert", stored.Name)
assert.WithinDuration(t, createdAt, stored.CreatedAt, time.Second)
assert.True(t, stored.UpdatedAt.After(updatedAt))
})
t.Run("Upsert_InsertsWhenIDDoesNotExist", func(t *testing.T) {
obj := &TestObject{Name: "upsert-with-preassigned-id"}
obj.SetID(bson.NewObjectID())
err := repo.Upsert(ctx, obj)
require.NoError(t, err)
var stored TestObject
err = repo.Get(ctx, *obj.GetID(), &stored)
require.NoError(t, err)
assert.Equal(t, "upsert-with-preassigned-id", stored.Name)
assert.Equal(t, *obj.GetID(), stored.ID)
})
}
@@ -0,0 +1,18 @@
package transactionimp
import (
"git.sendico.io/tech/sxchnge/shared/database/transaction"
"go.mongodb.org/mongo-driver/v2/mongo"
)
type MongoTransactionFactory struct {
client *mongo.Client
}
func (mtf *MongoTransactionFactory) CreateTransaction() transaction.Transaction {
return Create(mtf.client)
}
func CreateFactory(client *mongo.Client) transaction.Factory {
return &MongoTransactionFactory{client: client}
}
@@ -0,0 +1,34 @@
package transactionimp
import (
"context"
"git.sendico.io/tech/sxchnge/shared/database/transaction"
"go.mongodb.org/mongo-driver/v2/mongo"
)
type MongoTransaction struct {
client *mongo.Client
}
func (mt *MongoTransaction) Execute(ctx context.Context, cb transaction.Callback) (any, error) {
// Reuse an outer customer workflow transaction when a token store joins it.
if transaction.ActiveMongoTransaction(ctx) {
return cb(ctx)
}
session, err := mt.client.StartSession()
if err != nil {
return nil, err
}
defer session.EndSession(ctx)
callback := func(c context.Context) (any, error) {
return cb(c)
}
return session.WithTransaction(ctx, callback)
}
func Create(client *mongo.Client) *MongoTransaction {
return &MongoTransaction{client: client}
}
+37
View File
@@ -0,0 +1,37 @@
package mutil
import (
"context"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository"
"git.sendico.io/tech/sxchnge/shared/logging"
"git.sendico.io/tech/sxchnge/shared/logging/fields"
"go.mongodb.org/mongo-driver/v2/bson"
"go.uber.org/zap"
)
func SetArchived[T storable.Storable](ctx context.Context, logger mlogger.Logger, newArchived bool, objectRef bson.ObjectID, repo repository.Repository) error {
objs, err := GetObjects[T](ctx, logger, repository.IDFilter(objectRef), nil, repo)
if err != nil {
logger.Warn("Failed to fetch object", zap.Error(err), mzap.ObjRef("object_ref", objectRef))
return err
}
if len(objs) == 0 {
logger.Warn("No objects found to archive", mzap.ObjRef("object_ref", objectRef))
return nil
}
// Archive the first object found
obj := objs[0]
if archivable, ok := any(obj).(interface{ SetArchived(bool) }); ok {
archivable.SetArchived(newArchived)
if err := repo.Update(ctx, obj); err != nil {
logger.Warn("Failed to update object archived status", zap.Error(err), mzap.ObjRef("object_ref", objectRef))
return err
}
}
return nil
}
+26
View File
@@ -0,0 +1,26 @@
package mutil
import (
"context"
sharedmodel "git.sendico.io/tech/sxchnge/shared/model"
"git.sendico.io/tech/sxchnge/shared/database/repository"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"git.sendico.io/tech/sxchnge/shared/logging"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.uber.org/zap"
)
func GetObjects[T any](ctx context.Context, logger mlogger.Logger, filter builder.Query, cursor *sharedmodel.ViewCursor, repo repository.Repository) ([]T, error) {
entities := make([]T, 0)
decoder := func(cur *mongo.Cursor) error {
var next T
if e := cur.Decode(&next); e != nil {
logger.Warn("Failed to decode entity", zap.Error(e))
return e
}
entities = append(entities, next)
return nil
}
return entities, repo.FindManyByFilter(ctx, repository.ApplyCursor(filter, cursor), decoder)
}
+20
View File
@@ -0,0 +1,20 @@
package mutil
import (
"context"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository"
"git.sendico.io/tech/sxchnge/shared/logging"
"go.mongodb.org/mongo-driver/v2/bson"
"go.uber.org/zap"
)
func GetObjectByID(ctx context.Context, logger mlogger.Logger, id string, val storable.Storable, repo repository.Repository) error {
p, err := bson.ObjectIDFromHex(id)
if err != nil {
logger.Warn("Failed to decode object reference", zap.String("reference", id), zap.String("collection", val.Collection()))
return err
}
return repo.Get(ctx, p, val)
}
+37
View File
@@ -0,0 +1,37 @@
package storable
import (
"time"
"go.mongodb.org/mongo-driver/v2/bson"
)
const (
IDField = "_id"
IsArchivedField = "isArchived"
UpdatedAtField = "updatedAt"
)
type Base struct {
ID bson.ObjectID `bson:"_id" json:"id"`
CreatedAt time.Time `bson:"createdAt" json:"createdAt"` // Timestamp for when the comment was created
UpdatedAt time.Time `bson:"updatedAt" json:"updatedAt"` // Timestamp for when the comment was last updated (optional)
}
func (b *Base) GetID() *bson.ObjectID {
return &b.ID
}
func (b *Base) SetID(objID bson.ObjectID) {
b.ID = objID
b.CreatedAt = time.Now()
b.UpdatedAt = time.Now()
}
func (b *Base) Update() {
b.UpdatedAt = time.Now()
}
func (b *Base) Collection() string {
return "base"
}
+11
View File
@@ -0,0 +1,11 @@
package storable
import "go.mongodb.org/mongo-driver/v2/bson"
const (
RefField = "ref"
)
type Ref struct {
Ref bson.ObjectID `bson:"ref" json:"ref"`
}
@@ -0,0 +1,10 @@
package storable
import "go.mongodb.org/mongo-driver/v2/bson"
type Storable interface {
GetID() *bson.ObjectID
SetID(objID bson.ObjectID)
Update()
Collection() string
}
@@ -0,0 +1,11 @@
package builder
import "go.mongodb.org/mongo-driver/v2/bson"
type Accumulator interface {
Build() bson.D
}
type GroupAccumulator interface {
Build() bson.D
}
@@ -0,0 +1,8 @@
package builder
import "go.mongodb.org/mongo-driver/v2/bson"
type Alias interface {
Field() Field
Build() bson.D
}
@@ -0,0 +1,7 @@
package builder
import "go.mongodb.org/mongo-driver/v2/bson"
type Array interface {
Build() bson.A
}
@@ -0,0 +1,5 @@
package builder
type Expression interface {
Build() any
}
@@ -0,0 +1,7 @@
package builder
type Field interface {
Dot(field string) Field
CopyWith(field string) Field
Build() string
}
@@ -0,0 +1,16 @@
package builder
type MongoKeyword string
const (
MKAs MongoKeyword = "as"
MKForeignField MongoKeyword = "foreignField"
MKFrom MongoKeyword = "from"
MKIncludeArrayIndex MongoKeyword = "includeArrayIndex"
MKLet MongoKeyword = "let"
MKLocalField MongoKeyword = "localField"
MKPath MongoKeyword = "path"
MKPipeline MongoKeyword = "pipeline"
MKPreserveNullAndEmptyArrays MongoKeyword = "preserveNullAndEmptyArrays"
MKNewRoot MongoKeyword = "newRoot"
)
@@ -0,0 +1,57 @@
package builder
type MongoOperation string
const (
// Comparison operators
Gt MongoOperation = "$gt"
Lt MongoOperation = "$lt"
Gte MongoOperation = "$gte"
Lte MongoOperation = "$lte"
Eq MongoOperation = "$eq"
Ne MongoOperation = "$ne"
In MongoOperation = "$in"
NotIn MongoOperation = "$nin"
Exists MongoOperation = "$exists"
// Logical operators
And MongoOperation = "$and"
Or MongoOperation = "$or"
Not MongoOperation = "$not"
AddToSet MongoOperation = "$addToSet"
Avg MongoOperation = "$avg"
Pull MongoOperation = "$pull"
Count MongoOperation = "$count"
Cond MongoOperation = "$cond"
Each MongoOperation = "$each"
Expr MongoOperation = "$expr"
First MongoOperation = "$first"
Group MongoOperation = "$group"
IfNull MongoOperation = "$ifNull"
Limit MongoOperation = "$limit"
Literal MongoOperation = "$literal"
Lookup MongoOperation = "$lookup"
Match MongoOperation = "$match"
Max MongoOperation = "$max"
Min MongoOperation = "$min"
Push MongoOperation = "$push"
Project MongoOperation = "$project"
Set MongoOperation = "$set"
Inc MongoOperation = "$inc"
Unset MongoOperation = "$unset"
Rename MongoOperation = "$rename"
ReplaceRoot MongoOperation = "$replaceRoot"
SetUnion MongoOperation = "$setUnion"
Size MongoOperation = "$size"
Sort MongoOperation = "$sort"
Skip MongoOperation = "$skip"
Sum MongoOperation = "$sum"
Type MongoOperation = "$type"
Unwind MongoOperation = "$unwind"
Add MongoOperation = "$add"
Subtract MongoOperation = "$subtract"
Multiply MongoOperation = "$multiply"
Divide MongoOperation = "$divide"
)
@@ -0,0 +1,16 @@
package builder
import "go.mongodb.org/mongo-driver/v2/bson"
// Patch defines operations for constructing partial update documents.
// Each builder method returns the same Patch instance to allow chaining.
type Patch interface {
Set(field Field, value any) Patch
Inc(field Field, value any) Patch
Unset(field Field) Patch
Rename(field Field, newName string) Patch
Push(field Field, value any) Patch
Pull(field Field, value any) Patch
AddToSet(field Field, value any) Patch
Build() bson.D
}
@@ -0,0 +1,24 @@
package builder
import (
"git.sendico.io/tech/sxchnge/shared/service"
"go.mongodb.org/mongo-driver/v2/mongo"
)
type Pipeline interface {
Match(filter Query) Pipeline
Lookup(from mservice.Type, localField, foreignField, as Field) Pipeline
LookupWithPipeline(
from mservice.Type,
pipeline Pipeline, // your nested pipeline
as Field,
let *map[string]Field, // optional e.g. {"projRef": Field("$_id")}
) Pipeline
// unwind with functional options
Unwind(path Field, opts ...UnwindOption) Pipeline
Count(field Field) Pipeline
Group(groupBy Alias, accumulators ...GroupAccumulator) Pipeline
Project(projections ...Projection) Pipeline
ReplaceRoot(newRoot Expression) Pipeline
Build() mongo.Pipeline
}
@@ -0,0 +1,7 @@
package builder
import "go.mongodb.org/mongo-driver/v2/bson"
type Projection interface {
Build() bson.D
}
@@ -0,0 +1,24 @@
package builder
import (
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo/options"
)
type Query interface {
Filter(field Field, value any) Query
And(filters ...Query) Query
Or(filters ...Query) Query
Expression(value Expression) Query
Comparison(field Field, operator MongoOperation, value any) Query
RegEx(field Field, pattern, options string) Query
In(field Field, values ...any) Query
NotIn(field Field, values ...any) Query
Sort(field Field, ascending bool) Query
Limit(limit *int64) Query
Offset(offset *int64) Query
Archived(isArchived *bool) Query
BuildPipeline() bson.D
BuildQuery() bson.D
BuildOptions() *options.FindOptionsBuilder
}
@@ -0,0 +1,23 @@
package builder
// UnwindOption is a functional option for configuring the $unwind stage.
type UnwindOption func(*UnwindOpts)
type UnwindOpts struct {
PreserveNullAndEmptyArrays bool
IncludeArrayIndex string
}
// WithPreserveNullAndEmptyArrays tells $unwind to keep docs where the array is null/empty.
func WithPreserveNullAndEmptyArrays() UnwindOption {
return func(o *UnwindOpts) {
o.PreserveNullAndEmptyArrays = true
}
}
// WithIncludeArrayIndex adds an array‐index field named idxField to each unwound doc.
func WithIncludeArrayIndex(idxField string) UnwindOption {
return func(o *UnwindOpts) {
o.IncludeArrayIndex = idxField
}
}
@@ -0,0 +1,5 @@
package builder
type Value interface {
Build() any
}
@@ -0,0 +1,253 @@
package repository
import (
"git.sendico.io/tech/sxchnge/shared/database/internal/builder"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
sharedmodel "git.sendico.io/tech/sxchnge/shared/model"
"go.mongodb.org/mongo-driver/v2/bson"
)
func Query() builder.Query {
return builderimp.NewQueryImp()
}
func Filter(field string, value any) builder.Query {
return Query().Filter(Field(field), value)
}
func Field(baseName string) builder.Field {
return builderimp.NewFieldImp(baseName)
}
func Ref(field builder.Field) builder.Field {
return builderimp.NewRefFieldImp(field)
}
func RootRef() builder.Field {
return builderimp.NewRootRef()
}
func RemoveRef() builder.Field {
return builderimp.NewRemoveRef()
}
func Pipeline(stages ...bson.D) builder.Pipeline {
return builderimp.NewPipelineImp(stages...)
}
func IDField() builder.Field {
return Field(storable.IDField)
}
func NameField() builder.Field {
return Field(sharedmodel.NameField)
}
func DescrtiptionField() builder.Field {
return Field(sharedmodel.DescriptionField)
}
func IsArchivedField() builder.Field {
return Field(storable.IsArchivedField)
}
func IDFilter(ref bson.ObjectID) builder.Query {
return Query().Filter(IDField(), ref)
}
func ArchivedFilter() builder.Query {
return IsArchivedFilter(true)
}
func NotArchivedFilter() builder.Query {
return IsArchivedFilter(false)
}
func IsArchivedFilter(isArchived bool) builder.Query {
return Query().Filter(IsArchivedField(), isArchived)
}
func AccountField() builder.Field {
return Field("accountRef")
}
func AccountFilter(ref bson.ObjectID) builder.Query {
return Query().Filter(AccountField(), ref)
}
func StatusRefField() builder.Field {
return Field("statusRef")
}
func StatusRefFilter(ref bson.ObjectID) builder.Query {
return Query().Filter(StatusRefField(), ref)
}
func PriorityRefField() builder.Field {
return Field("priorityRef")
}
func PriorityRefFilter(ref bson.ObjectID) builder.Query {
return Query().Filter(PriorityRefField(), ref)
}
func IndexField() builder.Field {
return Field("index")
}
func IndexFilter(index int) builder.Query {
return Query().Filter(IndexField(), index)
}
func IndexOpFilter(index int, operation builder.MongoOperation) builder.Query {
return Query().Comparison(IndexField(), operation, index)
}
func Patch() builder.Patch {
return builderimp.NewPatchImp()
}
func Accumulator(operator builder.MongoOperation, value any) builder.Accumulator {
return builderimp.NewAccumulator(operator, value)
}
func GroupAccumulator(field builder.Field, acc builder.Accumulator) builder.GroupAccumulator {
return builderimp.NewGroupAccumulator(field, acc)
}
func Literal(value any) builder.Expression {
return builderimp.NewLiteralExpression(value)
}
func Projection(alias builder.Alias) builder.Projection {
return builderimp.NewAliasProjection(alias)
}
func IncludeField(field builder.Field) builder.Projection {
return builderimp.IncludeField(field)
}
func ExcludeField(field builder.Field) builder.Projection {
return builderimp.ExcludeField(field)
}
func ProjectionExpr(field builder.Field, expr builder.Expression) builder.Projection {
return builderimp.NewProjectionExpr(field, expr)
}
func NullAlias(lhs builder.Field) builder.Alias {
return builderimp.NewNullAlias(lhs)
}
func SimpleAlias(lhs, rhs builder.Field) builder.Alias {
return builderimp.NewSimpleAlias(lhs, rhs)
}
func ComplexAlias(lhs builder.Field, rhs []builder.Alias) builder.Alias {
return builderimp.NewComplexAlias(lhs, rhs)
}
func Aliases(aliases ...builder.Alias) builder.Alias {
return builderimp.NewAliases(aliases...)
}
func AddToSet(value builder.Expression) builder.Expression {
return builderimp.AddToSet(value)
}
func Size(value builder.Expression) builder.Expression {
return builderimp.Size(value)
}
func InRef(value builder.Field) builder.Expression {
return builderimp.InRef(value)
}
func In(values ...any) builder.Expression {
return builderimp.In(values)
}
func Cond(condition builder.Expression, ifTrue, ifFalse any) builder.Expression {
return builderimp.NewCond(condition, ifTrue, ifFalse)
}
func And(exprs ...builder.Expression) builder.Expression {
return builderimp.NewAnd(exprs...)
}
func Or(exprs ...builder.Expression) builder.Expression {
return builderimp.NewOr(exprs...)
}
func Type(expr builder.Expression) builder.Expression {
return builderimp.NewType(expr)
}
func Not(expression builder.Expression) builder.Expression {
return builderimp.NewNot(expression)
}
func Sum(expression builder.Expression) builder.Expression {
return builderimp.NewSum(expression)
}
func Assign(field builder.Field, expression builder.Expression) builder.Projection {
return builderimp.NewAssignment(field, expression)
}
func SetUnion(exprs ...builder.Expression) builder.Expression {
return builderimp.NewSetUnion(exprs...)
}
func Eq(left, right builder.Expression) builder.Expression {
return builderimp.Eq(left, right)
}
func Gt(left, right builder.Expression) builder.Expression {
return builderimp.Gt(left, right)
}
func Lt(left, right builder.Expression) builder.Expression {
return builderimp.NewLt(left, right)
}
func Array(expressions ...builder.Expression) builder.Array {
return builderimp.NewArray(expressions...)
}
func IfNull(cond, replacement builder.Expression) builder.Expression {
return builderimp.NewIfNull(cond, replacement)
}
func Each(exprs ...builder.Expression) builder.Expression {
return builderimp.NewEach(exprs...)
}
func Push(expression builder.Expression) builder.Expression {
return builderimp.NewPush(expression)
}
func Min(expression builder.Expression) builder.Expression {
return builderimp.NewMin(expression)
}
func Ne(left, right builder.Expression) builder.Expression {
return builderimp.Ne(left, right)
}
func Compute(field builder.Field, expression builder.Expression) builder.Expression {
return builderimp.NewCompute(field, expression)
}
func First(expr builder.Expression) builder.Expression {
return builderimp.First(expr)
}
func Value(value any) builder.Value {
return builderimp.NewValue(value)
}
func Exists(field builder.Field, exists bool) builder.Query {
return Query().Comparison(field, builder.Exists, exists)
}
@@ -0,0 +1,19 @@
package repository
import (
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
sharedmodel "git.sendico.io/tech/sxchnge/shared/model"
)
// ApplyCursor adds pagination and archival filters to the provided query.
func ApplyCursor(query builder.Query, cursor *sharedmodel.ViewCursor) builder.Query {
if cursor == nil {
return query
}
query = query.Limit(cursor.Limit)
query = query.Offset(cursor.Offset)
query = query.Archived(cursor.IsArchived)
return query
}
@@ -0,0 +1,5 @@
package repository
import "go.mongodb.org/mongo-driver/v2/mongo"
type DecodingFunc = func(r *mongo.Cursor) error
@@ -0,0 +1,25 @@
package repository
import "git.sendico.io/tech/sxchnge/shared/database/repository/builder"
type Sort int8
const (
Asc Sort = 1
Desc Sort = -1
)
type Key struct {
Field string
Sort Sort // 1 or -1. 0 means “use Type”.
Type IndexType // optional: "text", "2dsphere", ...
}
type Definition struct {
Keys []Key // mandatory, at least one element
Unique bool // unique constraint?
Sparse bool // sparse?
TTL *int32 // seconds; nil means “no TTL”
Name string // optional explicit name
PartialFilter builder.Query // optional: partialFilterExpression for conditional indexes
}
@@ -0,0 +1,36 @@
package repository
// IndexType represents a supported MongoDB index type.
type IndexType string
const (
// IndexTypeNotSet is a default index type
IndexTypeNotSet IndexType = ""
// IndexTypeSingleField is a single-field index.
IndexTypeSingleField IndexType = "single"
// IndexTypeCompound is a compound index on multiple fields.
IndexTypeCompound IndexType = "compound"
// IndexTypeMultikey is an index on array fields (created automatically when needed).
IndexTypeMultikey IndexType = "multikey"
// IndexTypeText is a text index for full-text search.
IndexTypeText IndexType = "text"
// IndexTypeGeo2D is a legacy 2D geospatial index for planar geometry.
IndexTypeGeo2D IndexType = "2d"
// IndexTypeGeo2DSphere is a 2dsphere geospatial index for GeoJSON data.
IndexTypeGeo2DSphere IndexType = "2dsphere"
// IndexTypeHashed is a hashed index for sharding and efficient equality queries.
IndexTypeHashed IndexType = "hashed"
// IndexTypeWildcard is a wildcard index to index all fields or subpaths.
IndexTypeWildcard IndexType = "wildcard"
// IndexTypeClustered is a clustered index that orders the collection on the index key.
IndexTypeClustered IndexType = "clustered"
)
@@ -0,0 +1,58 @@
package repository
import (
"context"
"git.sendico.io/tech/sxchnge/shared/database/internal/repository"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
rd "git.sendico.io/tech/sxchnge/shared/database/repository/decoder"
ri "git.sendico.io/tech/sxchnge/shared/database/repository/index"
"git.sendico.io/tech/sxchnge/shared/database/transaction"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
)
type (
// FilterQuery selects documents to operate on.
FilterQuery = builder.Query
// PatchDoc defines field/value modifications for partial updates.
PatchDoc = builder.Patch
)
type Repository interface {
Aggregate(ctx context.Context, builder builder.Pipeline, decoder rd.DecodingFunc) error
Insert(ctx context.Context, obj storable.Storable, getFilter builder.Query) error
InsertMany(ctx context.Context, objects []storable.Storable) error
Get(ctx context.Context, id bson.ObjectID, result storable.Storable) error
FindOneByFilter(ctx context.Context, builder builder.Query, result storable.Storable) error
FindFirstByFilter(ctx context.Context, builder builder.Query, result storable.Storable) error
FindManyByFilter(ctx context.Context, builder builder.Query, decoder rd.DecodingFunc) error
// FindAllByFilter decodes all matching documents into result, which must be a pointer to a slice.
FindAllByFilter(ctx context.Context, builder builder.Query, result any) error
CountByFilter(ctx context.Context, builder builder.Query) (int64, error)
Update(ctx context.Context, obj storable.Storable) error
// ReplaceOneByFilter replaces one matching document with obj and reports whether a document matched the filter.
// The repository must persist obj as provided and must not mutate its timestamps or identity before replacement.
ReplaceOneByFilter(ctx context.Context, filter builder.Query, obj storable.Storable) (bool, error)
// Upsert replaces the document identified by obj ID, inserting it when no document exists.
Upsert(ctx context.Context, obj storable.Storable) error
// Patch applies partial updates defined by patch to the document identified by id.
Patch(ctx context.Context, id bson.ObjectID, patch PatchDoc) error
// PatchOneByFilter atomically patches one document matching filter and decodes the updated document into result.
PatchOneByFilter(ctx context.Context, filter FilterQuery, patch PatchDoc, result storable.Storable) (bool, error)
// PatchMany applies partial updates defined by patch to all documents matching filter and returns the number of updated documents.
PatchMany(ctx context.Context, filter FilterQuery, patch PatchDoc) (int, error)
Delete(ctx context.Context, id bson.ObjectID) error
DeleteMany(ctx context.Context, query builder.Query) error
CreateIndex(def *ri.Definition) error
DropIndex(ctx context.Context, name string) error
ListIndexDefinitions(ctx context.Context) ([]*ri.Definition, error)
ListIDs(ctx context.Context, query builder.Query) ([]bson.ObjectID, error)
WithTransaction(ctx context.Context, cb transaction.Callback) (any, error)
Collection() string
}
func CreateMongoRepository(db *mongo.Database, collection string) Repository {
return repositoryimp.NewMongoRepository(db, collection)
}
@@ -0,0 +1,21 @@
package template
import (
"context"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"go.mongodb.org/mongo-driver/v2/bson"
)
type DB[T storable.Storable] interface {
Create(ctx context.Context, object T) error
InsertMany(ctx context.Context, objects []T) error
Get(ctx context.Context, objectRef bson.ObjectID, result T) error
Update(ctx context.Context, object T) error
Patch(ctx context.Context, objectRef bson.ObjectID, patch builder.Patch) error
Delete(ctx context.Context, objectRef bson.ObjectID) error
DeleteMany(ctx context.Context, query builder.Query) error
DeleteCascade(ctx context.Context, objectRef bson.ObjectID) error
FindOne(ctx context.Context, query builder.Query, result T) error
}
@@ -0,0 +1,99 @@
package template
import (
"context"
"git.sendico.io/tech/sxchnge/shared/database/record"
"git.sendico.io/tech/sxchnge/shared/database/repository"
"git.sendico.io/tech/sxchnge/shared/database/repository/builder"
"git.sendico.io/tech/sxchnge/shared/errors"
"git.sendico.io/tech/sxchnge/shared/logging"
"git.sendico.io/tech/sxchnge/shared/logging/fields"
"git.sendico.io/tech/sxchnge/shared/service"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
)
type CascadeDeleterT = func(ctx context.Context, objectRef bson.ObjectID) error
type DBImp[T storable.Storable] struct {
Logger mlogger.Logger
Repository repository.Repository
cdeleter CascadeDeleterT
}
func (db *DBImp[T]) Create(ctx context.Context, object T) error {
return db.Repository.Insert(ctx, object, nil)
}
func (db *DBImp[T]) InsertMany(ctx context.Context, objects []T) error {
if len(objects) == 0 {
return nil
}
storables := make([]storable.Storable, len(objects))
for i, obj := range objects {
storables[i] = obj
}
return db.Repository.InsertMany(ctx, storables)
}
func (db *DBImp[T]) Get(ctx context.Context, objectRef bson.ObjectID, result T) error {
return db.Repository.Get(ctx, objectRef, result)
}
func (db *DBImp[T]) Update(ctx context.Context, object T) error {
return db.Repository.Update(ctx, object)
}
func (db *DBImp[T]) Patch(ctx context.Context, objectRef bson.ObjectID, patch builder.Patch) error {
return db.Repository.Patch(ctx, objectRef, patch)
}
func (db *DBImp[T]) PatchMany(ctx context.Context, query builder.Query, patch builder.Patch) (int, error) {
return db.Repository.PatchMany(ctx, query, patch)
}
func (db *DBImp[T]) Delete(ctx context.Context, objectRef bson.ObjectID) error {
return db.Repository.Delete(ctx, objectRef)
}
func (db *DBImp[T]) DeleteMany(ctx context.Context, query builder.Query) error {
return db.Repository.DeleteMany(ctx, query)
}
func (db *DBImp[T]) FindOne(ctx context.Context, query builder.Query, result T) error {
return db.Repository.FindOneByFilter(ctx, query, result)
}
func (db *DBImp[T]) ListIDs(ctx context.Context, query builder.Query) ([]bson.ObjectID, error) {
return db.Repository.ListIDs(ctx, query)
}
func (db *DBImp[T]) SetDeleter(deleter CascadeDeleterT) {
db.Logger.Debug("Custom cascade deletion method installed")
db.cdeleter = deleter
}
func (db *DBImp[T]) deleteCascadeStub(_ context.Context, objectRef bson.ObjectID) error {
db.Logger.Warn("Unimplemented cascade delete call", mzap.ObjRef("object_ref", objectRef))
return merrors.NotImplemented("custom implementation required")
}
func (db *DBImp[T]) DeleteCascade(ctx context.Context, objectRef bson.ObjectID) error {
return db.cdeleter(ctx, objectRef)
}
func Create[T storable.Storable](
logger mlogger.Logger,
collection mservice.Type,
db *mongo.Database,
) *DBImp[T] {
res := &DBImp[T]{
Logger: logger.Named(collection),
Repository: repository.CreateMongoRepository(db, collection),
}
res.cdeleter = res.deleteCascadeStub
return res
}
@@ -0,0 +1,12 @@
package transaction
import (
"context"
"go.mongodb.org/mongo-driver/v2/mongo"
)
func ActiveMongoTransaction(ctx context.Context) bool {
session := mongo.SessionFromContext(ctx)
return session != nil && session.TransactionRunning()
}
@@ -0,0 +1,21 @@
package transaction
import (
"context"
"testing"
"go.mongodb.org/mongo-driver/v2/mongo"
)
func TestActiveMongoTransactionRejectsContextWithoutSession(t *testing.T) {
if ActiveMongoTransaction(context.Background()) {
t.Fatal("context without mongo session must not be active")
}
}
func TestActiveMongoTransactionRejectsSessionWithoutTransaction(t *testing.T) {
ctx := mongo.NewSessionContext(context.Background(), &mongo.Session{})
if ActiveMongoTransaction(ctx) {
t.Fatal("session without transaction must not be active")
}
}
@@ -0,0 +1,5 @@
package transaction
type Factory interface {
CreateTransaction() Transaction
}
@@ -0,0 +1,11 @@
package transaction
import (
"context"
)
type Callback = func(ctx context.Context) (any, error)
type Transaction interface {
Execute(ctx context.Context, cb Callback) (any, error)
}