func (store *MongoStorage) AddObjectDestinations(orgID string, objectType string, objectID string, destinationsList []string) (*common.MetaData, string, []common.StoreDestinationStatus, common.SyncServiceError) {
	result := object{}
	id := createObjectCollectionID(orgID, objectType, objectID)
	selector := bson.M{"metadata": bson.ElementDocument, "destinations": bson.ElementArray, "last-update": bson.ElementTimestamp, "status": bson.ElementString}
	for i := 0; i < maxUpdateTries; i++ {
		if err := store.fetchOne(objects, bson.M{"_id": id}, selector, &result); err != nil {
			return nil, "", nil, &Error{fmt.Sprintf("Failed to retrieve object's destinations. Error: %s.", err)}
		}
		updatedDests, addedDests, err := getDestinationsForAdd(orgID, store, result.Destinations, destinationsList)
		if err != nil {
			return nil, "", nil, err
		}
		query := bson.M{
			"$set":         bson.M{"destinations": updatedDests},
			"$currentDate": bson.M{"last-update": bson.M{"$type": "timestamp"}},
		}
		if err := store.update(objects, bson.M{"_id": id, "last-update": result.LastUpdate}, query); err != nil {
			if err == mgo.ErrNotFound {
				time.Sleep(time.Duration(sleepInMS) * time.Millisecond)
				continue
			}
			return nil, "", nil, &Error{fmt.Sprintf("Failed to add destinations to object's destinations list. Error: %s.", err)}
		}
		return &result.MetaData, result.Status, addedDests, nil
	}
	return nil, "", nil, &Error{"Failed to add destinations to object's destination list."}
}
AddObjectdestinations adds the destinations to object's destination list
Returns the metadata, object's status, an array of added destinations after removing the overlapped destinations
package storage
import (
	"crypto/tls"
	"crypto/x509"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"os"
	"strings"
	"time"
	"github.com/globalsign/mgo"
	"github.com/globalsign/mgo/bson"
	"github.com/open-horizon/edge-sync-service/common"
	"github.com/open-horizon/edge-utilities/logger"
	"github.com/open-horizon/edge-utilities/logger/log"
	"github.com/open-horizon/edge-utilities/logger/trace"
)
type fileHandle struct {
	file    *mgo.GridFile
	session *mgo.Session
	offset  int64
	chunks  map[int64][]byte
}
type MongoStorage struct {
	session      *mgo.Session
	dialInfo     *mgo.DialInfo
	openFiles    map[string]*fileHandle
	connected    bool
	lockChannel  chan int
	mapLock      chan int
	sessionCache []*mgo.Session
	cacheSize    int
	cacheIndex   int
}
type object struct {
	ID                 string                          `bson:"_id"`
	MetaData           common.MetaData                 `bson:"metadata"`
	Status             string                          `bson:"status"`
	PolicyReceived     bool                            `bson:"policy-received"`
	RemainingConsumers int                             `bson:"remaining-consumers"`
	RemainingReceivers int                             `bson:"remaining-receivers"`
	Destinations       []common.StoreDestinationStatus `bson:"destinations"`
	LastUpdate         bson.MongoTimestamp             `bson:"last-update"`
}
type destinationObject struct {
	ID           string              `bson:"_id"`
	Destination  common.Destination  `bson:"destination"`
	LastPingTime bson.MongoTimestamp `bson:"last-ping-time"`
}
type notificationObject struct {
	ID           string              `bson:"_id"`
	Notification common.Notification `bson:"notification"`
}
type leaderDocument struct {
	ID               int32               `bson:"_id"`
	UUID             string              `bson:"uuid"`
	LastHeartbeatTS  bson.MongoTimestamp `bson:"last-heartbeat-ts"`
	HeartbeatTimeout int32               `bson:"heartbeat-timeout"`
	Version          int64               `bson:"version"`
}
type isMasterResult struct {
	IsMaster  bool      `bson:"isMaster"`
	LocalTime time.Time `bson:"localTime"`
	OK        bool      `bson:"ok"`
}
type messagingGroupObject struct {
	ID         string              `bson:"_id"`
	GroupName  string              `bson:"group-name"`
	LastUpdate bson.MongoTimestamp `bson:"last-update"`
}
type organizationObject struct {
	ID           string              `bson:"_id"`
	Organization common.Organization `bson:"org"`
	LastUpdate   bson.MongoTimestamp `bson:"last-update"`
}
type webhookObject struct {
	ID         string              `bson:"_id"`
	Hooks      []string            `bson:"hooks"`
	LastUpdate bson.MongoTimestamp `bson:"last-update"`
}
type aclObject struct {
	ID         string              `bson:"_id"`
	Users      []common.ACLentry   `bson:"users"`
	OrgID      string              `bson:"org-id"`
	ACLType    string              `bson:"acl-type"`
	LastUpdate bson.MongoTimestamp `bson:"last-update"`
}
const maxUpdateTries = 5
var sleepInMS int
func (store *MongoStorage) Init() common.SyncServiceError {
	store.lockChannel = make(chan int, 1)
	store.lockChannel <- 1
	store.mapLock = make(chan int, 1)
	store.mapLock <- 1
	store.dialInfo = &mgo.DialInfo{
		Addrs:        strings.Split(common.Configuration.MongoAddressCsv, ","),
		Source:       common.Configuration.MongoAuthDbName,
		Username:     common.Configuration.MongoUsername,
		Password:     common.Configuration.MongoPassword,
		Timeout:      time.Duration(20 * time.Second),
		ReadTimeout:  time.Duration(60 * time.Second),
		WriteTimeout: time.Duration(60 * time.Second),
	}
	if common.Configuration.MongoUseSSL {
		tlsConfig := &tls.Config{}
		if common.Configuration.MongoCACertificate != "" {
			var caFile string
			if strings.HasPrefix(common.Configuration.MongoCACertificate, "/") {
				caFile = common.Configuration.MongoCACertificate
			} else {
				caFile = common.Configuration.PersistenceRootPath + common.Configuration.MongoCACertificate
			}
			serverCaCert, err := ioutil.ReadFile(caFile)
			if err != nil {
				if _, ok := err.(*os.PathError); ok {
					serverCaCert = []byte(common.Configuration.MongoCACertificate)
					err = nil
				} else {
					message := fmt.Sprintf("Failed to find mongo SSL CA file. Error: %s.", err)
					return &Error{message}
				}
			}
			caCertPool := x509.NewCertPool()
			caCertPool.AppendCertsFromPEM(serverCaCert)
			tlsConfig.RootCAs = caCertPool
		}
		if common.Configuration.MongoAllowInvalidCertificates {
			tlsConfig.InsecureSkipVerify = true
		}
		store.dialInfo.DialServer = func(addr *mgo.ServerAddr) (net.Conn, error) {
			return tls.Dial("tcp", addr.String(), tlsConfig)
		}
	}
	var session *mgo.Session
	var err error
	if trace.IsLogging(logger.INFO) {
		trace.Info("Connecting to mongo...")
	}
	for connectTime := 0; connectTime < common.Configuration.DatabaseConnectTimeout; connectTime += 10 {
		session, err = mgo.DialWithInfo(store.dialInfo)
		if err == nil {
			break
		}
		if connectTime == 0 && trace.IsLogging(logger.ERROR) {
			trace.Error("Failed to dial mgo. Error: " + err.Error())
		}
		if strings.HasPrefix(err.Error(), "unauthorized") ||
			strings.HasPrefix(err.Error(), "not authorized") ||
			strings.HasPrefix(err.Error(), "auth fail") ||
			strings.HasPrefix(err.Error(), "Authentication failed") {
			break
		}
		if connectTime == 0 && trace.IsLogging(logger.ERROR) {
			trace.Error("Retrying to connect to mongo")
		}
	}
	if session == nil {
		message := fmt.Sprintf("Failed to dial mgo. Error: %s.", err)
		return &Error{message}
	}
	store.connected = true
	common.HealthStatus.ReconnectedToDatabase()
	if trace.IsLogging(logger.INFO) {
		trace.Info("Connected to the database")
	}
	if log.IsLogging(logger.INFO) {
		log.Info("Connected to the database")
	}
	session.SetSafe(&mgo.Safe{})
	db := session.DB(common.Configuration.MongoDbName)
	db.C(destinations).EnsureIndexKey("destination.destination-org-id")
	notificationsCollection := db.C(notifications)
	notificationsCollection.EnsureIndexKey("notification.destination-org-id", "notification.destination-id", "notification.destination-type")
	notificationsCollection.EnsureIndexKey("notification.resend-time", "notification.status")
	objectsCollection := db.C(objects)
	objectsCollection.EnsureIndexKey("metadata.destination-org-id")
	err = objectsCollection.EnsureIndex(
		mgo.Index{
			Key: []string{
				"metadata.destination-org-id",
				"metadata.destination-policy.services.org-id",
				"metadata.destination-policy.services.service-name",
			},
			Name:       "syncObjects-destination-policy.services.service-id",
			Unique:     false,
			DropDups:   false,
			Background: false,
			Sparse:     true,
		})
	if err != nil {
		log.Error("Failed to create an index on %s. Error: %s", objects, err)
	}
	err = objectsCollection.EnsureIndex(
		mgo.Index{
			Key: []string{
				"metadata.destination-org-id",
				"metadata.destination-policy.timestamp",
			},
			Unique:     false,
			DropDups:   false,
			Background: false,
			Sparse:     true,
		})
	if err != nil {
		log.Error("Failed to create an index on %s. Error: %s", objects, err)
	}
	db.C(acls).EnsureIndexKey("org-id", "acl-type")
	store.session = session
	store.cacheSize = common.Configuration.MongoSessionCacheSize
	if store.cacheSize > 1 {
		store.sessionCache = make([]*mgo.Session, store.cacheSize)
		for i := 0; i < store.cacheSize; i++ {
			store.sessionCache[i] = store.session.Copy()
		}
	}
	store.openFiles = make(map[string]*fileHandle)
	sleepInMS = common.Configuration.MongoSleepTimeBetweenRetry
	if trace.IsLogging(logger.TRACE) {
		trace.Trace("Successfully initialized mongo driver")
	}
	return nil
}
func (store *MongoStorage) Stop() {
	if store.cacheSize > 1 {
		for i := 0; i < store.cacheSize; i++ {
			store.sessionCache[i].Close()
		}
	}
	store.session.Close()
}
func (store *MongoStorage) PerformMaintenance() {
	store.checkObjects()
}
func (store *MongoStorage) Cleanup(isTest bool) common.SyncServiceError {
	return nil
}
func (store *MongoStorage) GetObjectsToActivate() ([]common.MetaData, common.SyncServiceError) {
	currentTime := time.Now().UTC().Format(time.RFC3339)
	query := bson.M{"$or": []bson.M{
		bson.M{"status": common.NotReadyToSend},
		bson.M{"status": common.ReadyToSend}},
		"metadata.inactive": true,
		"$and": []bson.M{
			bson.M{"metadata.activation-time": bson.M{"$ne": ""}},
			bson.M{"metadata.activation-time": bson.M{"$lte": currentTime}}}}
	selector := bson.M{"metadata": bson.ElementDocument}
	result := []object{}
	if err := store.fetchAll(objects, query, selector, &result); err != nil {
		return nil, err
	}
	metaDatas := make([]common.MetaData, len(result))
	for i, r := range result {
		metaDatas[i] = r.MetaData
	}
	return metaDatas, nil
}
func (store *MongoStorage) StoreObject(metaData common.MetaData, data []byte, status string) ([]common.StoreDestinationStatus, common.SyncServiceError) {
	id := getObjectCollectionID(metaData)
	if !metaData.NoData && data != nil {
		if err := store.storeDataInFile(id, data); err != nil {
			return nil, err
		}
	} else if !metaData.MetaOnly {
		store.removeFile(id)
	}
	if metaData.DestinationPolicy != nil {
		metaData.DestinationPolicy.Timestamp = time.Now().UTC().UnixNano()
	}
	var dests []common.StoreDestinationStatus
	var deletedDests []common.StoreDestinationStatus
	if status == common.NotReadyToSend || status == common.ReadyToSend {
		newID := store.getInstanceID()
		metaData.InstanceID = newID
		if data != nil && !metaData.NoData && !metaData.MetaOnly {
			metaData.DataID = newID
		}
		var err error
		dests, deletedDests, err = createDestinationsFromMeta(store, metaData)
		if err != nil {
			return nil, err
		}
	}
	existingObject := &object{}
	if err := store.fetchOne(objects, bson.M{"_id": id}, nil, existingObject); err != nil {
		if err != mgo.ErrNotFound {
			return nil, &Error{fmt.Sprintf("Failed to retrieve object's status. Error: %s.", err)}
		}
		existingObject = nil
	}
	if existingObject != nil {
		if (metaData.DestinationPolicy != nil && existingObject.MetaData.DestinationPolicy == nil) ||
			(metaData.DestinationPolicy == nil && existingObject.MetaData.DestinationPolicy != nil) {
			return nil, &common.InvalidRequest{Message: "Can't update the existence of Destination Policy"}
		}
		if metaData.MetaOnly {
			metaData.DataID = existingObject.MetaData.DataID
			metaData.ObjectSize = existingObject.MetaData.ObjectSize
			metaData.ChunkSize = existingObject.MetaData.ChunkSize
			metaData.PublicKey = existingObject.MetaData.PublicKey
			metaData.Signature = existingObject.MetaData.Signature
		}
		if metaData.DestinationPolicy != nil {
			dests = existingObject.Destinations
		}
	}
	newObject := object{ID: id, MetaData: metaData, Status: status, PolicyReceived: false,
		RemainingConsumers: metaData.ExpectedConsumers,
		RemainingReceivers: metaData.ExpectedConsumers, Destinations: dests}
	if err := store.upsert(objects, bson.M{"_id": id, "metadata.destination-org-id": metaData.DestOrgID}, newObject); err != nil {
		return nil, &Error{fmt.Sprintf("Failed to store an object. Error: %s.", err)}
	}
	return deletedDests, nil
}
func (store *MongoStorage) GetObjectDestinations(metaData common.MetaData) ([]common.Destination, common.SyncServiceError) {
	result := object{}
	id := getObjectCollectionID(metaData)
	if err := store.fetchOne(objects, bson.M{"_id": id}, bson.M{"destinations": bson.ElementArray}, &result); err != nil {
		switch err {
		case mgo.ErrNotFound:
			return nil, nil
		default:
			return nil, &Error{fmt.Sprintf("Failed to retrieve object's destinations. Error: %s.", err)}
		}
	}
	dests := make([]common.Destination, 0)
	for _, d := range result.Destinations {
		dests = append(dests, d.Destination)
	}
	return dests, nil
}
func (store *MongoStorage) GetObjectDestinationsList(orgID string, objectType string,
	objectID string) ([]common.StoreDestinationStatus, common.SyncServiceError) {
	result := object{}
	id := createObjectCollectionID(orgID, objectType, objectID)
	if err := store.fetchOne(objects, bson.M{"_id": id}, bson.M{"destinations": bson.ElementArray}, &result); err != nil {
		switch err {
		case mgo.ErrNotFound:
			return nil, nil
		default:
			return nil, &Error{fmt.Sprintf("Failed to retrieve object's destinations. Error: %s.", err)}
		}
	}
	return result.Destinations, nil
}
func (store *MongoStorage) UpdateObjectDestinations(orgID string, objectType string, objectID string, destinationsList []string) (*common.MetaData, string,
	[]common.StoreDestinationStatus, []common.StoreDestinationStatus, common.SyncServiceError) {
	result := object{}
	id := createObjectCollectionID(orgID, objectType, objectID)
	selector := bson.M{"metadata": bson.ElementDocument, "destinations": bson.ElementArray, "last-update": bson.ElementTimestamp, "status": bson.ElementString}
	for i := 0; i < maxUpdateTries; i++ {
		if err := store.fetchOne(objects, bson.M{"_id": id}, selector, &result); err != nil {
			return nil, "", nil, nil, &Error{fmt.Sprintf("Failed to retrieve object's destinations. Error: %s.", err)}
		}
		dests, deletedDests, addedDests, err := createDestinations(orgID, store, result.Destinations, destinationsList)
		if err != nil {
			return nil, "", nil, nil, err
		}
		query := bson.M{
			"$set":         bson.M{"destinations": dests},
			"$currentDate": bson.M{"last-update": bson.M{"$type": "timestamp"}},
		}
		if err := store.update(objects, bson.M{"_id": id, "last-update": result.LastUpdate}, query); err != nil {
			if err == mgo.ErrNotFound {
				time.Sleep(time.Duration(sleepInMS) * time.Millisecond)
				continue
			}
			return nil, "", nil, nil, &Error{fmt.Sprintf("Failed to update object's destinations. Error: %s.", err)}
		}
		return &result.MetaData, result.Status, deletedDests, addedDests, nil
	}
	return nil, "", nil, nil, &Error{"Failed to update object's destinations."}
}