feat(contacts): add MongoDB repository

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Copilot-Session: 47a18375-9f8b-4f08-9014-ef78dcb72780

Author: AchoArnold

api/pkg/di/container.go

@@ -925,12 +925,22 @@ func (container *Container) MessageThreadRepository() (repository repositories.M
 
 // ContactRepository creates a new instance of repositories.ContactRepository
 func (container *Container) ContactRepository() (repository repositories.ContactRepository) {
-	container.logger.Debug("creating GORM repositories.ContactRepository")
-	return repositories.NewGormContactRepository(
-		container.Logger(),
-		container.Tracer(),
-		container.DB(),
-	)
+	switch os.Getenv("CONTACT_DB_BACKEND") {
+	case "mongodb":
+		container.logger.Debug("creating MongoDB repositories.ContactRepository")
+		return repositories.NewMongoContactRepository(
+			container.Logger(),
+			container.Tracer(),
+			container.MongoDB(),
+		)
+	default:
+		container.logger.Debug("creating GORM repositories.ContactRepository")
+		return repositories.NewGormContactRepository(
+			container.Logger(),
+			container.Tracer(),
+			container.DB(),
+		)
+	}
 }
 
 // HeartbeatMonitorRepository creates a new instance of repositories.HeartbeatMonitorRepository

api/pkg/entities/contact.go

@@ -63,14 +63,14 @@ func (p *ContactProperties) Scan(src any) error {
 
 // Contact represents a saved contact belonging to a user.
 type Contact struct {
-	ID           uuid.UUID         `json:"id" gorm:"primaryKey;type:uuid" example:"32343a19-da5e-4b1b-a767-3298a73703cb"`
-	UserID       UserID            `json:"user_id" gorm:"index" example:"WB7DRDWrJZRGbYrv2CKGkqbzvqdC"`
-	Name         string            `json:"name" example:"Alice Smith"`
-	Emails       pq.StringArray    `json:"emails" gorm:"type:text[]" swaggertype:"array,string" example:"alice@example.com"`
-	PhoneNumbers pq.StringArray    `json:"phone_numbers" gorm:"type:text[]" swaggertype:"array,string" example:"+18005550199,+18005550100"`
-	Properties   ContactProperties `json:"properties" gorm:"type:jsonb" swaggertype:"object,string"`
-	CreatedAt    time.Time         `json:"created_at" example:"2022-06-05T14:26:02.302718+03:00"`
-	UpdatedAt    time.Time         `json:"updated_at" example:"2022-06-05T14:26:02.302718+03:00"`
+	ID           uuid.UUID         `json:"id" gorm:"primaryKey;type:uuid" bson:"_id" example:"32343a19-da5e-4b1b-a767-3298a73703cb"`
+	UserID       UserID            `json:"user_id" gorm:"index" bson:"user_id" example:"WB7DRDWrJZRGbYrv2CKGkqbzvqdC"`
+	Name         string            `json:"name" bson:"name" example:"Alice Smith"`
+	Emails       pq.StringArray    `json:"emails" gorm:"type:text[]" bson:"emails" swaggertype:"array,string" example:"alice@example.com"`
+	PhoneNumbers pq.StringArray    `json:"phone_numbers" gorm:"type:text[]" bson:"phone_numbers" swaggertype:"array,string" example:"+18005550199,+18005550100"`
+	Properties   ContactProperties `json:"properties" gorm:"type:jsonb" bson:"properties" swaggertype:"object,string"`
+	CreatedAt    time.Time         `json:"created_at" bson:"created_at" example:"2022-06-05T14:26:02.302718+03:00"`
+	UpdatedAt    time.Time         `json:"updated_at" bson:"updated_at" example:"2022-06-05T14:26:02.302718+03:00"`
 }
 
 // TableName overrides the table name used by Contact.

api/pkg/entities/contact_test.go

@@ -1,11 +1,32 @@
 package entities
 
 import (
+	"reflect"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
 )
 
+func TestContact_BSONFieldNames(t *testing.T) {
+	contactType := reflect.TypeOf(Contact{})
+	expected := map[string]string{
+		"ID":           "_id",
+		"UserID":       "user_id",
+		"Name":         "name",
+		"Emails":       "emails",
+		"PhoneNumbers": "phone_numbers",
+		"Properties":   "properties",
+		"CreatedAt":    "created_at",
+		"UpdatedAt":    "updated_at",
+	}
+
+	for fieldName, bsonName := range expected {
+		field, found := contactType.FieldByName(fieldName)
+		assert.True(t, found)
+		assert.Equal(t, bsonName, field.Tag.Get("bson"))
+	}
+}
+
 func TestContactProperties_ValueScanRoundTrip(t *testing.T) {
 	cases := []ContactProperties{
 		nil,

api/pkg/repositories/mongo_contact_repository.go

@@ -0,0 +1,239 @@
+package repositories
+
+import (
+	"context"
+	"fmt"
+	"regexp"
+
+	"github.com/NdoleStudio/httpsms/pkg/entities"
+	"github.com/NdoleStudio/httpsms/pkg/telemetry"
+	"github.com/NdoleStudio/stacktrace"
+	"github.com/google/uuid"
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/mongo/options"
+)
+
+// mongoContactRepository is responsible for persisting entities.Contact in MongoDB.
+type mongoContactRepository struct {
+	logger     telemetry.Logger
+	tracer     telemetry.Tracer
+	collection *mongo.Collection
+}
+
+// NewMongoContactRepository creates the MongoDB version of the ContactRepository.
+func NewMongoContactRepository(
+	logger telemetry.Logger,
+	tracer telemetry.Tracer,
+	db *mongo.Database,
+) ContactRepository {
+	return &mongoContactRepository{
+		logger:     logger.WithService(fmt.Sprintf("%T", &mongoContactRepository{})),
+		tracer:     tracer,
+		collection: db.Collection(collectionContacts),
+	}
+}
+
+func (repository *mongoContactRepository) Store(ctx context.Context, contacts []*entities.Contact) error {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	if len(contacts) == 0 {
+		return nil
+	}
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	documents := make([]any, len(contacts))
+	for index, contact := range contacts {
+		documents[index] = contact
+	}
+
+	session, err := repository.collection.Database().Client().StartSession()
+	if err != nil {
+		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot start transaction to store [%d] contacts", len(contacts)))
+	}
+	defer session.EndSession(ctx)
+
+	_, err = session.WithTransaction(ctx, func(transactionCtx context.Context) (any, error) {
+		return repository.collection.InsertMany(transactionCtx, documents)
+	})
+	if err != nil {
+		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot store [%d] contacts", len(contacts)))
+	}
+
+	return nil
+}
+
+func (repository *mongoContactRepository) Update(ctx context.Context, contact *entities.Contact) error {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	filter := mongoContactIDFilter(contact.UserID, contact.ID)
+	if _, err := repository.collection.ReplaceOne(ctx, filter, contact); err != nil {
+		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot update contact with ID [%s]", contact.ID))
+	}
+
+	return nil
+}
+
+func (repository *mongoContactRepository) Load(ctx context.Context, userID entities.UserID, contactID uuid.UUID) (*entities.Contact, error) {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	contact := new(entities.Contact)
+	err := repository.collection.FindOne(ctx, mongoContactIDFilter(userID, contactID)).Decode(contact)
+	if err == mongo.ErrNoDocuments {
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.PropagateWithCodef(err, ErrCodeNotFound, "contact with ID [%s] for user [%s] does not exist", contactID, userID))
+	}
+	if err != nil {
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot load contact with ID [%s] for user [%s]", contactID, userID))
+	}
+
+	return contact, nil
+}
+
+func (repository *mongoContactRepository) Index(ctx context.Context, userID entities.UserID, params IndexParams) (*[]entities.Contact, error) {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	findOptions := options.Find().
+		SetSort(mongoContactSort(params)).
+		SetSkip(int64(params.Skip)).
+		SetLimit(int64(params.Limit))
+	cursor, err := repository.collection.Find(ctx, mongoContactFilter(userID, params.Query), findOptions)
+	if err != nil {
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot index contacts for user [%s] with params [%+#v]", userID, params))
+	}
+	defer cursor.Close(ctx)
+
+	contacts := make([]entities.Contact, 0)
+	if err = cursor.All(ctx, &contacts); err != nil {
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode contacts for user [%s]", userID))
+	}
+
+	return &contacts, nil
+}
+
+func (repository *mongoContactRepository) Count(ctx context.Context, userID entities.UserID, params IndexParams) (int64, error) {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	count, err := repository.collection.CountDocuments(ctx, mongoContactFilter(userID, params.Query))
+	if err != nil {
+		return 0, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot count contacts for user [%s] with query [%s]", userID, params.Query))
+	}
+
+	return count, nil
+}
+
+func (repository *mongoContactRepository) FetchByPhoneNumbers(ctx context.Context, userID entities.UserID, phoneNumbers []string) (*[]entities.Contact, error) {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	findOptions := options.Find().SetSort(bson.D{
+		{Key: "updated_at", Value: 1},
+		{Key: "_id", Value: 1},
+	})
+	cursor, err := repository.collection.Find(ctx, mongoContactPhoneNumbersFilter(userID, phoneNumbers), findOptions)
+	if err != nil {
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot fetch contacts for user [%s] by phone numbers [%v]", userID, phoneNumbers))
+	}
+	defer cursor.Close(ctx)
+
+	contacts := make([]entities.Contact, 0)
+	if err = cursor.All(ctx, &contacts); err != nil {
+		return nil, repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot decode contacts for user [%s] by phone numbers", userID))
+	}
+
+	return &contacts, nil
+}
+
+func (repository *mongoContactRepository) Delete(ctx context.Context, userID entities.UserID, contactID uuid.UUID) error {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	if _, err := repository.collection.DeleteOne(ctx, mongoContactIDFilter(userID, contactID)); err != nil {
+		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot delete contact with ID [%s] for user [%s]", contactID, userID))
+	}
+
+	return nil
+}
+
+func (repository *mongoContactRepository) DeleteAllForUser(ctx context.Context, userID entities.UserID) error {
+	ctx, span := repository.tracer.Start(ctx)
+	defer span.End()
+
+	ctx, cancel := context.WithTimeout(ctx, dbOperationDuration)
+	defer cancel()
+
+	if _, err := repository.collection.DeleteMany(ctx, bson.D{{Key: "user_id", Value: string(userID)}}); err != nil {
+		return repository.tracer.WrapErrorSpan(span, stacktrace.Propagatef(err, "cannot delete all contacts for user [%s]", userID))
+	}
+
+	return nil
+}
+
+func mongoContactFilter(userID entities.UserID, query string) bson.D {
+	filter := bson.D{{Key: "user_id", Value: string(userID)}}
+	if query == "" {
+		return filter
+	}
+
+	expression := bson.Regex{Pattern: regexp.QuoteMeta(query), Options: "i"}
+	return append(filter, bson.E{Key: "$or", Value: bson.A{
+		bson.D{{Key: "name", Value: expression}},
+		bson.D{{Key: "emails", Value: expression}},
+		bson.D{{Key: "phone_numbers", Value: expression}},
+	}})
+}
+
+func mongoContactSort(params IndexParams) bson.D {
+	sortBy := "updated_at"
+	if params.SortBy == "name" {
+		sortBy = "name"
+	}
+
+	direction := 1
+	if params.SortBy == "" || params.SortDescending {
+		direction = -1
+	}
+
+	return bson.D{
+		{Key: sortBy, Value: direction},
+		{Key: "_id", Value: direction},
+	}
+}
+
+func mongoContactPhoneNumbersFilter(userID entities.UserID, phoneNumbers []string) bson.D {
+	return bson.D{
+		{Key: "user_id", Value: string(userID)},
+		{Key: "phone_numbers", Value: bson.D{{Key: "$in", Value: phoneNumbers}}},
+	}
+}
+
+func mongoContactIDFilter(userID entities.UserID, contactID uuid.UUID) bson.D {
+	return bson.D{
+		{Key: "user_id", Value: string(userID)},
+		{Key: "_id", Value: contactID.String()},
+	}
+}