user.go

// https://gitlab.com/yoanyombapro/CubeMicroservices/-/blob/08ad700e385853134c5149155e2681c40db1195c/podinfo/pkg/database/user.go
package database

import (
	"context"
	"database/sql"
	"errors"
	"time"

	"github.com/jinzhu/gorm"
	"go.uber.org/zap"

	model "gitlab.com/yoanyombapro/CubeMicroservices/podinfo/pkg/models/proto"
)

// CreateUser creates a user account record in the database
func (db *Database) CreateUser(ctx context.Context, user model.User) (*model.User, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var userOrm model.UserORM

		// check if the user exists in the database based off of
		// email and username
		// Note: Email and Usernames must be unique across the entire database
		recordNotFound := tx.Where(model.UserORM{Email: user.Email, UserName: user.UserName}).First(&userOrm).RecordNotFound()

		// if the user does exist and the user account is not active, reactivate the user account
		// and update the user state in the backend database
		if !recordNotFound && !userOrm.IsActive {
			userOrm.IsActive = true
			if err := tx.Save(&userOrm).Error; err != nil {
				db.Logger.Error("failed to create user", zap.Error(err))
				return nil, err
			}

			return convertUserOrmToGenericUser(ctx, userOrm)
		} else if !recordNotFound {
			db.Logger.Error("failed to create user because user already exists")
			return nil, errors.New("user already exists")
		}

		// convert the input user field to orm
		userOrm, err := user.ToORM(ctx)
		if err != nil {
			db.Logger.Error("failed to create user", zap.Error(err))
			return nil, err
		}

		// generate a random reset token for the user account
		// and set it
		userOrm.ResetToken = GenerateRandomToken(20)
		currentTime := time.Now()
		tokenExpirationTime := currentTime.Add(time.Hour * 24 * 10)
		userOrm.ResetTokenExpiration = &tokenExpirationTime

		// activate user account
		userOrm.IsActive = true

		// save the user to the database
		if err := tx.Create(&userOrm).Error; err != nil {
			db.Logger.Error("failed to create user", zap.Error(err))
			return nil, err
		}

		db.Logger.Info("user successfully created",
			zap.String("id", string(userOrm.Id)),
			zap.String("username", userOrm.UserName),
			zap.String("email", userOrm.Email))

		return convertUserOrmToGenericUser(ctx, userOrm)
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return &model.User{}, err
	}

	createdUser := output.(*model.User)
	return createdUser, nil
}

// convertUserOrmToGenericUser converts a user orm to generic type
func convertUserOrmToGenericUser(ctx context.Context, userOrm model.UserORM) (*model.User, error) {
	createdUser, err := userOrm.ToPB(ctx)
	if err != nil {
		return nil, err
	}

	if err = createdUser.Validate(); err != nil {
		return nil, err
	}
	return &createdUser, nil
}

// GetUserByID queries the database and obtains a user record by id
func (db *Database) GetUserByID(ctx context.Context, userID uint32) (*model.User, error) {
	var userOrm model.UserORM

	if recordNotFound := db.Engine.Where(model.UserORM{Id: userID}).First(&userOrm).RecordNotFound(); recordNotFound {
		db.Logger.Error("user does not exist", zap.String("id", string(userID)))
		return nil, errors.New("user does not exist")
	}

	// convert the obtained user ORM object to a user object and validate all fields are there
	userObj, err := userOrm.ToPB(ctx)
	if err != nil {
		db.Logger.Error("failed to convert fields to protobuf format")
		return nil, err
	}

	// perform field validation
	if err = userObj.Validate(); err != nil {
		db.Logger.Error("field validation failed")
		return nil, err
	}

	db.Logger.Info("user successfully obtained user by id",
		zap.String("id", string(userOrm.Id)),
		zap.String("username", userOrm.UserName),
		zap.String("email", userOrm.Email))

	return &userObj, nil
}

// CreateUserProfile creates a user profile and ties it to a user account record
// if the account record exists.
func (db *Database) CreateUserProfile(ctx context.Context, userID uint32, profile model.Profile) (*model.Profile, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var userOrm *model.UserORM

		// first validate the profile has all necessary fields of interest
		if err := profile.Validate(); err != nil {
			db.Logger.Error("profile field validation failed", zap.Error(err))
			return nil, err
		}

		// check that the user exists
		exists, userOrm, err := db.GetUserIfExists(userID, "", "")
		if !exists {
			db.Logger.Error("user account does not exist. please create one and try again", zap.Error(err))
			return nil, err
		}

		if userOrm.Profile != nil && userOrm.Profile.Id != 0 {
			db.Logger.Error("profile already exists")
			return nil, errors.New("profile already exists")
		}

		// update the user ORM object with the profile ORM object
		profileOrm, err := profile.ToORM(ctx)
		if err != nil {
			db.Logger.Error("failed to convert profile fields to orm type", zap.Error(err))
			return nil, err
		}

		userOrm.Profile = &profileOrm
		// Updates only the relevant fields of interest in a user entity in the database
		if err = tx.Save(&userOrm).Error; err != nil {
			db.Logger.Error("failed to create user profile", zap.Error(err))
			return nil, err
		}
		profile.Id = userOrm.Profile.Id

		db.Logger.Info("successfully created a profile for the user account",
			zap.String("accountId", string(userOrm.Id)),
			zap.String("profileId", string(profile.Id)))

		return &profile, nil
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return nil, err
	}

	createdProfile := output.(*model.Profile)
	return createdProfile, nil
}

// CreateUserSubscription creates a subscription and ties it to a user account record if the
// account record exists, and the subscription does not. if the subscription does indeed exist and is
// inactive, it is reactivated.
func (db *Database) CreateUserSubscription(ctx context.Context, userID uint32, subscription model.Subscriptions) error {
	transaction := func(tx *gorm.DB) error {
		var (
			userOrm           *model.UserORM
			subscriptionExist = false
		)

		// validate the subscription object
		if subscription.SubscriptionName == "" ||
			subscription.StartDate == nil ||
			subscription.EndDate == nil ||
			subscription.SubscriptionStatus == "" {
			db.Logger.Error("invalid subscription. missing subscription name, start date, enddate, or status",
				zap.Any("subscription", subscription))
			return errors.New("invalid subscription")
		}

		// check and make sure the  user account with the specified userid exists
		exists, userOrm, err := db.GetUserIfExists(userID, "", "")
		if !exists {
			db.Logger.Error("user account does not exist", zap.Error(err))
			return err
		}

		// convert the subscription object to an ORM type
		subscriptionOrm, err := subscription.ToORM(ctx)
		if err != nil {
			db.Logger.Error("failed to convert subscription to orm type", zap.Error(err))
			return err
		}

		subscriptions := make([]*model.SubscriptionsORM, len(userOrm.Subscriptions), len(userOrm.Subscriptions))

		for _, oldSubscription := range userOrm.Subscriptions {
			if oldSubscription.SubscriptionName == subscriptionOrm.SubscriptionName && !subscriptionExist {
				// activate the subscription if it is not already active
				oldSubscription.IsActive = true
				oldSubscription.EndDate = subscriptionOrm.EndDate
				subscriptionExist = true

				db.Logger.Info("re-activating subscripion", zap.String("id", string(subscriptionOrm.Id)),
					zap.String("name", subscription.SubscriptionName))
			}
			subscriptions = append(subscriptions, oldSubscription)
		}

		if !subscriptionExist {
			subscriptions = append(subscriptions, &subscriptionOrm)
			db.Logger.Info("new subscripion added to subscriptions list", zap.String("id", string(subscriptionOrm.Id)),
				zap.String("name", subscription.SubscriptionName))
		}

		userOrm.Subscriptions = subscriptions

		// save the user with the updated subscriptions
		if err = tx.Save(&userOrm).Error; err != nil {
			db.Logger.Error("failed to create user subscription", zap.Error(err))
			return err
		}

		db.Logger.Info("successfully created user subscription")
		return nil
	}

	return db.PerformTransaction(transaction)
}

// UpdateUser updates a user record if it already exists in the backend
func (db *Database) UpdateUser(ctx context.Context, userID uint32, user model.User) (*model.User, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		// first and foremost we check for the existence of the user
		exists, _, err := db.GetUserIfExists(userID, "", "")
		if !exists {
			db.Logger.Error("failed to obtain user by id as user does not exist", zap.Error(err))
			return nil, err
		}

		// convert the user to an ORM type
		userOrm, err := user.ToORM(ctx)
		if err != nil {
			db.Logger.Error("failed to convert user object to orm type", zap.Error(err))
			return nil, err
		}

		// update the actual user in the database
		if err := tx.Save(&userOrm).Error; err != nil {
			db.Logger.Error("failed to update user", zap.Error(err))
			return nil, err
		}

		db.Logger.Info("Successfully updated user", zap.String("id", string(userOrm.Id)),
			zap.String("user name", string(userOrm.UserName)))

		return &user, nil
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return &model.User{}, err
	}

	updatedUser := output.(*model.User)
	return updatedUser, nil
}

// UpdateUserSubscription updates a subscription tied to a user account if it exists
func (db *Database) UpdateUserSubscription(ctx context.Context, userID uint32, subscriptionID uint32, newSubscription model.Subscriptions) (*model.Subscriptions, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var (
			userOrm              model.UserORM
			updatedSubscriptions []*model.SubscriptionsORM
		)

		// validate the new version of the subscription
		if err := newSubscription.Validate(); err != nil {
			db.Logger.Error("invalid subscription", zap.Error(err))
			return nil, err
		}

		// convert the subscription to an ORM type
		subscriptionOrm, err := newSubscription.ToORM(ctx)
		if err != nil {
			db.Logger.Error("failed to convert subscription to orm type", zap.Error(err))
			return nil, err
		}

		// we check for the existence of the user account
		// and the subscription
		exist, _, err := db.GetSubscriptionExistById(userID, subscriptionID)
		if !exist {
			db.Logger.Error("subscription does not exist", zap.Error(err))
			return nil, err
		}

		// obtain the user from the database
		if err := tx.Where(model.UserORM{Id: userID}).First(&userOrm).Error; err != nil {
			db.Logger.Error("failed to obtain user from backend database", zap.Error(err))
			return nil, err
		}

		// update the subscription if an existing version exists in the database
		for _, oldSubscription := range userOrm.Subscriptions {
			if oldSubscription.SubscriptionName == subscriptionOrm.SubscriptionName {
				updatedSubscriptions = append(updatedSubscriptions, &subscriptionOrm)
				db.Logger.Info("updated subscription and added to subscription list")
			} else {
				updatedSubscriptions = append(updatedSubscriptions, oldSubscription)
			}
		}

		// update the subscription list tied to the user
		userOrm.Subscriptions = updatedSubscriptions

		// update the actual user in the database
		if err := tx.Save(&userOrm).Error; err != nil {
			db.Logger.Error("failed to successfully update subscription", zap.Error(err))
			return nil, err
		}

		db.Logger.Info("successfully updated subscription",
			zap.String("userId", string(userID)), zap.String("subscriptionId", string(subscriptionID)))

		return &newSubscription, nil
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return nil, err
	}

	updatedSubscription := output.(*model.Subscriptions)
	return updatedSubscription, nil
}

// UpdateUserProfile updates a user profile tied to a user account if such profile exists
func (db *Database) UpdateUserProfile(ctx context.Context, userID, profileID uint32, newProfile model.Profile) (*model.Profile, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var userOrm model.UserORM

		// first and foremost we check for the existence of the user account
		// and the profile
		exist, _, err := db.GetUserProfileIfExists(userID)
		if !exist {
			db.Logger.Error("failed to obtain user profile as it does not exist", zap.Error(err))
			return nil, err
		}

		// obtain the user from the database
		if err := tx.Where(model.UserORM{Id: userID}).First(&userOrm).Error; err != nil {
			db.Logger.Error("failed to obtain user", zap.Error(err))
			return nil, err
		}

		// convert the profile to an ORM type
		profileOrm, err := newProfile.ToORM(ctx)
		if err != nil {
			db.Logger.Error("failed to convert user profile to orm type", zap.Error(err))
			return nil, err
		}

		// update the profile tied to the user
		userOrm.Profile = &profileOrm

		// update the actual user in the database
		if err := tx.Save(&userOrm).Error; err != nil {
			db.Logger.Error("failed to update user profile", zap.Error(err))
			return nil, err
		}

		db.Logger.Info("successfully updated user profile",
			zap.String("userId", string(userID)), zap.String("profileId", string(profileOrm.Id)))

		return newProfile, nil
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return nil, err
	}

	updatedProfile := output.(*model.Profile)
	return updatedProfile, nil
}

// DeleteUser deletes a user account and any reference objects tied to the user
func (db *Database) DeleteUser(ctx context.Context, userID uint32) error {
	transaction := func(tx *gorm.DB) error {
		var userOrm model.UserORM

		if userID == 0 {
			db.Logger.Error("invalid user id")
			return errors.New("invalid user id")
		}

		exist, _, err := db.GetUserIfExists(userID, "", "")
		if !exist {
			db.Logger.Error("failed to obtain user by id. user does not exist", zap.Error(err))
			return err
		}

		if err = tx.Where(model.UserORM{Id: userID}).Delete(&userOrm).Error; err != nil {
			db.Logger.Error("failed to successfully delete user account", zap.Error(err))
			return err
		}

		db.Logger.Info("successfully deleted user account", zap.String("userId", string(userID)))

		return nil
	}

	return db.PerformTransaction(transaction)
}

// DeleteUserProfile deletes a user profile tied to a user account
func (db *Database) DeleteUserProfile(ctx context.Context, userID, profileID uint32) error {
	transaction := func(tx *gorm.DB) error {
		// check the user of interest has a profile to even delete
		exist, profile, err := db.GetUserProfileIfExists(userID)
		if !exist {
			db.Logger.Error("failed to obtain user profile as it does not exist", zap.Error(err))
			return err
		}

		// attempt deletion of the profile
		if err = tx.Where(model.ProfileORM{Id: profileID}).Delete(&profile).Error; err != nil {
			db.Logger.Error("failed to delete user profile", zap.Error(err))
			return err
		}

		db.Logger.Info("successfully deleted user Profile",
			zap.String("userId", string(userID)), zap.String("profileId", string(profileID)))

		return nil
	}

	return db.PerformTransaction(transaction)
}

// DeleteUserSubscription deletes a subscription tied to a user account
func (db *Database) DeleteUserSubscription(ctx context.Context, userID, subscriptionID uint32) error {
	transaction := func(tx *gorm.DB) error {
		// get a subscription by id
		exist, _, err := db.GetSubscriptionExistById(userID, subscriptionID)
		if !exist {
			db.Logger.Error("failed to get subscription by id as it does not exist", zap.Error(err))
			return err
		}

		// delete from subscriptions table in db where id == subscriptions id
		if err = tx.Where(model.SubscriptionsORM{Id: subscriptionID}).Delete(&model.SubscriptionsORM{}).Error; err != nil {
			db.Logger.Error("failed to delete subscription", zap.Error(err))
			return err
		}

		db.Logger.Info("successfully deleted subscription",
			zap.String("userId", string(userID)), zap.String("subscriptionId", string(subscriptionID)))

		return nil
	}

	return db.PerformTransaction(transaction)
}

// GetAllUsers obtains user records. The max number of records returned is defined
// by the limit input parameter.
func (db *Database) GetAllUsers(ctx context.Context, limit uint32) ([]model.User, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var users = make([]model.User, limit, limit+1)
		var iteration uint32 = 0

		rows, err := tx.Limit(limit).Model(&model.UserORM{}).Rows()
		defer rows.Close()

		if err != nil {
			db.Logger.Error("failed to obtain set of user object from the database", zap.Error(err))
			return nil, err
		}

		for rows.Next() {
			var entry model.UserORM
			// scan the data that the row pointer points to into a userOrm object
			if err := tx.ScanRows(rows, &entry); err != nil {
				db.Logger.Error("failed to scan the returned row", zap.Error(err))
				return nil, err
			}

			user, err := entry.ToPB(ctx)
			if err != nil {
				db.Logger.Error("failed to convert orm type to pb type", zap.Error(err))
				return nil, err
			}

			if err = user.Validate(); err != nil {
				db.Logger.Error("user object failed validation check", zap.Error(err))
				return nil, err
			}
			users = append(users, user)
			iteration++

			if iteration == limit {
				break
			}
		}

		db.Logger.Info("successfully obtained users from backend database")

		return users, rows.Close()
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return nil, err
	}

	users := output.([]model.User)
	return users, nil
}

// GetAllUsersByAccountType obtains user records by account type. The upper bound on the number of
// user records returned is defined by the limit input parameter.
func (db *Database) GetAllUsersByAccountType(ctx context.Context, accountType string, limit uint32) ([]model.User, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var (
			users = make([]model.User, limit)
		)

		rows, err := tx.Limit(limit).Where(model.UserORM{UserAccountType: accountType}).Model(&model.UserORM{}).Rows()
		if err != nil {
			db.Logger.Error("failed to obtain users by accounttypes", zap.Error(err))
			return nil, err
		}

		defer rows.Close()

		for rows.Next() {
			entry := model.UserORM{}
			// scan the data that the row pointer points to into a userOrm object
			if err := tx.ScanRows(rows, &entry); err != nil {
				db.Logger.Error("failed to scan row", zap.Error(err))
				return nil, err
			}

			user, err := entry.ToPB(ctx)
			if err != nil {
				db.Logger.Error("failed to convert user orm type to pb type", zap.Error(err))
				return nil, err
			}

			if err = user.Validate(); err != nil {
				db.Logger.Error("user validation failed", zap.Error(err))
				return nil, err
			}
			users = append(users, user)
		}

		db.Logger.Info("successfully obtained users by account type")

		return users, rows.Close()
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return nil, err
	}

	users := output.([]model.User)
	return users, nil
}

// GetAllUsersByIntent queries the database for all user records based on intent. The
// upper bound on the number of user records returned is defined by the limit input parameter.
func (db *Database) GetAllUsersByIntent(ctx context.Context, intent string, limit uint32) ([]model.User, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var users = make([]model.User, limit, limit+10)

		rows, err := tx.Limit(limit).Where(model.UserORM{Intent: intent}).Model(&model.UserORM{}).Rows()
		if err != nil {
			db.Logger.Error("failed to obtain users by intent", zap.Error(err))
			return nil, err
		}

		defer rows.Close()

		for rows.Next() {
			entry := model.UserORM{}
			// scan the data that the row pointer points to into a userOrm object
			if err := tx.ScanRows(rows, &entry); err != nil {
				db.Logger.Error("failed to scan row", zap.Error(err))
				return nil, err
			}

			user, err := entry.ToPB(ctx)
			if err != nil {
				db.Logger.Error("failed to convert user orm type to pb type", zap.Error(err))
				return nil, err
			}

			if err = user.Validate(); err != nil {
				db.Logger.Error("user validation call failed", zap.Error(err))
				return nil, err
			}
			users = append(users, user)
		}

		db.Logger.Info("successfully obtained user by intent")
		return users, rows.Close()
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return nil, err
	}

	users := output.([]model.User)
	return users, nil
}

// GetUserProfile queries the database for a user profile record based on user account id
func (db *Database) GetUserProfile(ctx context.Context, userID uint32) (model.Profile, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		// obtain the user of interest
		user, err := db.GetUserByID(ctx, userID)
		if err != nil {
			db.Logger.Error("failed to obtain user account by id", zap.Error(err))
			return nil, err
		}

		// validate that the user has all necessary fields
		if err = user.Validate(); err != nil {
			db.Logger.Error("user validation failed", zap.Error(err))
			return nil, err
		}

		// from the user object extract the profile object
		profile := user.Profile
		if profile != nil {
			if err = profile.Validate(); err != nil {
				db.Logger.Error("user profile validation failed", zap.Error(err))
				return nil, err
			}
			db.Logger.Info("profile does not exist")
			// return the profile object
			return *profile, nil
		}

		db.Logger.Info("successfully returned user profile")

		return model.Profile{}, errors.New("user profile does not exist")
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return model.Profile{}, err
	}

	profileData := output.(model.Profile)
	return profileData, nil
}

// GetAllUserProfilesByType queries the database and returns all user profile records
// with a specified profile type. The upper bound on the number of records to return
// is defined by the limit input parameter.
func (db *Database) GetAllUserProfilesByType(ctx context.Context, profileType string, limit uint32) ([]model.Profile, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var profiles = make([]model.Profile, limit, limit+10)

		rows, err := tx.Limit(limit).Where(model.ProfileORM{ProfileType: profileType}).Model(&model.ProfileORM{}).Rows()
		if err != nil {
			db.Logger.Error("failed to obtain user by profile type")
			return nil, err
		}
		defer rows.Close()

		// obtain the value of interest from the rows returned by the query
		for rows.Next() {
			entry := model.ProfileORM{}
			// scan the data that the row pointer points to into a userOrm object
			if err := tx.ScanRows(rows, &entry); err != nil {
				db.Logger.Error("failed to scan rows", zap.Error(err))
				return nil, err
			}

			profile, err := entry.ToPB(ctx)
			if err != nil {
				db.Logger.Error("failed to convert user orm type to pb type", zap.Error(err))
				return nil, err
			}

			if err = profile.Validate(); err != nil {
				db.Logger.Error("user profile validation failed", zap.Error(err))
				return nil, err
			}
			profiles = append(profiles, profile)
		}

		db.Logger.Info("successfully returned user profile by profile types")
		return profiles, rows.Close()
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return nil, err
	}

	profileData := output.([]model.Profile)
	return profileData, nil
}

// GetAllUserProfiles queries the database and returns a set of user profile records.
// The upper bound on the number of records returned is defined by the limit
// input parameter.
func (db *Database) GetAllUserProfiles(ctx context.Context, limit uint32) ([]model.Profile, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var profiles = make([]model.Profile, limit, limit+10)

		rows, err := tx.Limit(limit).Model(&model.ProfileORM{}).Rows()
		if err != nil {
			db.Logger.Error(err.Error())
			return nil, err
		}
		defer rows.Close()

		data, err := extractFromRows(rows, tx, model.ProfileORM{})
		if err != nil {
			db.Logger.Error(err.Error())
			return nil, err
		}

		profilesOrm := make([]model.ProfileORM, len(data))

		for _, val := range data {
			profilesOrm = append(profilesOrm, val.(model.ProfileORM))
		}

		for _, profileOrm := range profilesOrm {
			// convert to object type and validate
			profile, err := profileOrm.ToPB(ctx)
			if err != nil {
				db.Logger.Error(err.Error())
				return nil, err
			}

			if err = profile.Validate(); err != nil {
				db.Logger.Error(err.Error())
				return nil, err
			}
			profiles = append(profiles, profile)
		}

		db.Logger.Info("successfully obtained user profiles")
		return profiles, rows.Close()
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return nil, err
	}

	profileData := output.([]model.Profile)
	return profileData, nil
}

// GetAllUserProfilesByNationality queries the database for a list of profiles
// of a certain nationality. The maximal amount of records to return is defined
// by the limit input parameter
func (db *Database) GetAllUserProfilesByNationality(ctx context.Context, nationality string, limit uint32) ([]model.Profile, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var profiles = make([]model.Profile, limit, limit+10)

		// get all profiles by profile type by querying the database
		rows, err := tx.Limit(limit).Where(model.ProfileORM{Nationality: nationality}).Model(&model.ProfileORM{}).Rows()
		if err != nil {
			db.Logger.Error(err.Error())
			return nil, err
		}
		defer rows.Close()

		// obtain the value of interest from the rows returned by the query
		for rows.Next() {
			entry := model.ProfileORM{}
			// scan the data that the row pointer points to into a userOrm object
			if err := tx.ScanRows(rows, &entry); err != nil {
				db.Logger.Error(err.Error())
				return nil, err
			}

			profile, err := entry.ToPB(ctx)
			if err != nil {
				db.Logger.Error(err.Error())
				return nil, err
			}

			if err = profile.Validate(); err != nil {
				db.Logger.Error(err.Error())
				return nil, err
			}
			profiles = append(profiles, profile)
		}

		db.Logger.Info("Successfully obtained user profiles by nationality")
		return profiles, rows.Close()
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return nil, err
	}

	profileData := output.([]model.Profile)
	return profileData, nil
}

// GetAllUserSubscriptions queries the database for a user based off of the
// user id and returns a set of subscriptions the user has
func (db *Database) GetAllUserSubscriptions(ctx context.Context, user_id uint32) ([]model.Subscriptions, error) {
	transaction := func(tx *gorm.DB) (interface{}, error) {
		var subscriptions = make([]model.Subscriptions, 10, 20)

		user, err := db.GetUserByID(ctx, user_id)
		if err != nil {
			return nil, err
		}

		for _, subscription := range user.GetSubscriptions() {
			subscriptions = append(subscriptions, *subscription)
		}

		return subscriptions, nil
	}

	output, err := db.PerformComplexTransaction(transaction)

	if err != nil {
		return nil, err
	}

	subscriptions := output.([]model.Subscriptions)
	return subscriptions, nil
}

// extractFromRows operates on database rows returned by a given query. It transforms the
// obtains row values into an ORM type and returns any errors that may have occurred throughout
// the lifecycle of the function call as well as an abstract type tied to the list of ORM
// types obtained from all rows
func extractFromRows(rows *sql.Rows, tx *gorm.DB, dbModel interface{}) ([]interface{}, error) {
	var data = make([]interface{}, 20)
	defer rows.Close()

	// iterate over all rows obtained from the query
	for rows.Next() {
		// scan the data that the row pointer points to into a userOrm object
		if err := tx.ScanRows(rows, &dbModel); err != nil {
			return nil, err
		}

		data = append(data, dbModel)
	}

	// we call rows.close() again to reduce any potential
	// side effects that may arise. It is much safer to reclose
	// a database handle than to potential leave it open as a sideffect
	// of some mishandled error
	return data, rows.Close()
}