func (store *BoltStorage) StoreObject(metaData common.MetaData, data []byte, status string) ([]common.StoreDestinationStatus, common.SyncServiceError) {
	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
		}
		if common.Configuration.NodeType == common.CSS {
			var err error
			dests, deletedDests, err = createDestinationsFromMeta(store, metaData)
			if err != nil {
				return nil, err
			}
		}
	}
	if metaData.DestinationPolicy != nil {
		metaData.DestinationPolicy.Timestamp = time.Now().UTC().UnixNano()
	}
	if metaData.MetaOnly {
		function := func(object boltObject) (boltObject, common.SyncServiceError) {
			if object.Status == common.ConsumedByDest {
				return object, &common.InvalidRequest{"Can't update only the meta data of consumed object"}
			}
			if (object.Meta.DestinationPolicy == nil && metaData.DestinationPolicy != nil) ||
				(object.Meta.DestinationPolicy != nil && metaData.DestinationPolicy == nil) {
				return object, &common.InvalidRequest{"Can't update the existence of Destination Policy"}
			}
			metaData.DataID = object.Meta.DataID        
			metaData.PublicKey = object.Meta.PublicKey  
			metaData.Signature = object.Meta.Signature
			object.Meta = metaData
			object.Status = status
			object.PolicyReceived = false
			object.RemainingConsumers = metaData.ExpectedConsumers
			object.RemainingReceivers = metaData.ExpectedConsumers
			if metaData.DestinationPolicy == nil {
				object.Destinations = dests
			}
			return object, nil
		}
		err := store.updateObjectHelper(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, function)
		if err == nil || err != notFound {
			return deletedDests, err
		}
	}
	var dataPath string
	if !metaData.NoData && data != nil {
		dataPath = createDataPathFromMeta(store.localDataPath, metaData)
		if _, err := dataURI.StoreData(dataPath, bytes.NewReader(data), uint32(len(data))); err != nil {
			return nil, err
		}
	} else if !metaData.MetaOnly {
		if err := dataURI.DeleteStoredData(createDataPathFromMeta(store.localDataPath, metaData)); err != nil {
			return nil, err
		}
	}
	newObject := boltObject{Meta: metaData, Status: status, PolicyReceived: false,
		RemainingConsumers: metaData.ExpectedConsumers, RemainingReceivers: metaData.ExpectedConsumers,
		DataPath: dataPath, Destinations: dests}
	function := func(object boltObject) (boltObject, common.SyncServiceError) {
		if (object.Meta.DestinationPolicy == nil && metaData.DestinationPolicy != nil) ||
			(object.Meta.DestinationPolicy != nil && metaData.DestinationPolicy == nil) {
			return object, &common.InvalidRequest{Message: "Can't update the existence of Destination Policy"}
		}
		if metaData.DestinationPolicy != nil {
			newObject.Destinations = object.Destinations
		}
		return newObject, nil
	}
	err := store.updateObjectHelper(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, function)
	if err == notFound {
		encoded, err := json.Marshal(newObject)
		if err != nil {
			return nil, err
		}
		id := getObjectCollectionID(metaData)
		err = store.db.Update(func(tx *bolt.Tx) error {
			err = tx.Bucket(objectsBucket).Put([]byte(id), []byte(encoded))
			return err
		})
		return deletedDests, err
	}
	return deletedDests, err
}
StoreObject stores an object
If the object already exists, return the changes in its destinations list (for CSS) - return the list of deleted destinations
package storage
import (
	"bytes"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"strings"
	"time"
	"github.com/open-horizon/edge-sync-service/common"
	"github.com/open-horizon/edge-sync-service/core/dataURI"
	"github.com/open-horizon/edge-utilities/logger"
	"github.com/open-horizon/edge-utilities/logger/log"
	"github.com/open-horizon/edge-utilities/logger/trace"
	bolt "go.etcd.io/bbolt"
)
const timebaseBucketName = "syncTimebase"
type BoltStorage struct {
	db            *bolt.DB
	timebase      int64
	lockChannel   chan int
	localDataPath string
}
type boltObject struct {
	Meta                             common.MetaData                 `json:"meta"`
	Status                           string                          `json:"status"`
	PolicyReceived                   bool                            `json:"policy-received"`
	RemainingConsumers               int                             `json:"remaining-consumers"`
	RemainingReceivers               int                             `json:"remaining-receivers"`
	DataPath                         string                          `json:"data-path"`
	ConsumedTimestamp                time.Time                       `json:"consumed-timestamp"`
	Destinations                     []common.StoreDestinationStatus `json:"destinations"`
	RemovedDestinationPolicyServices []common.ServiceID              `json:"removed-destination-policy-services"`
}
type boltDestination struct {
	Destination  common.Destination `json:"destination"`
	LastPingTime time.Time          `json:"last-ping-time"`
}
type boltMessagingGroup struct {
	OrgID      string    `json:"orgid"`
	GroupName  string    `json:"group-name"`
	LastUpdate time.Time `json:"last-update"`
}
type boltACL struct {
	Users   []common.ACLentry `json:"users"`
	OrgID   string            `json:"org-id"`
	ACLType string            `json:"acl-type"`
	Key     string            `json:"key"`
}
var (
	objectsBucket         []byte
	webhooksBucket        []byte
	notificationsBucket   []byte
	timebaseBucket        []byte
	destinationsBucket    []byte
	messagingGroupsBucket []byte
	organizationsBucket   []byte
	aclBucket             []byte
)
func (store *BoltStorage) Init() common.SyncServiceError {
	store.lockChannel = make(chan int, 1)
	store.lockChannel <- 1
	path := common.Configuration.PersistenceRootPath + "/sync/db/"
	err := os.MkdirAll(path, 0750)
	if err != nil {
		return err
	}
	name := "css-sync.db"
	if common.Configuration.NodeType == common.ESS {
		name = "ess-sync.db"
	}
	store.db, err = bolt.Open(path+name, 0600, nil)
	if err != nil {
		return err
	}
	objectsBucket = []byte(objects)
	webhooksBucket = []byte(webhooks)
	notificationsBucket = []byte(notifications)
	timebaseBucket = []byte(timebaseBucketName)
	destinationsBucket = []byte(destinations)
	messagingGroupsBucket = []byte(messagingGroups)
	organizationsBucket = []byte(organizations)
	aclBucket = []byte(acls)
	err = store.db.Update(func(tx *bolt.Tx) error {
		_, err = tx.CreateBucketIfNotExists(objectsBucket)
		if err != nil {
			return err
		}
		_, err = tx.CreateBucketIfNotExists(webhooksBucket)
		if err != nil {
			return err
		}
		_, err = tx.CreateBucketIfNotExists(notificationsBucket)
		if err != nil {
			return err
		}
		_, err = tx.CreateBucketIfNotExists(destinationsBucket)
		if err != nil {
			return err
		}
		_, err = tx.CreateBucketIfNotExists(messagingGroupsBucket)
		if err != nil {
			return err
		}
		_, err = tx.CreateBucketIfNotExists(organizationsBucket)
		if err != nil {
			return err
		}
		_, err = tx.CreateBucketIfNotExists(aclBucket)
		if err != nil {
			return err
		}
		b, err := tx.CreateBucketIfNotExists(timebaseBucket)
		if err != nil {
			return err
		}
		currentTime := time.Now().UnixNano()
		store.timebase = currentTime
		currentTimeInSeconds := currentTime / 1e9
		persistedTimeBase := currentTimeInSeconds
		encoded := b.Get([]byte("timebase"))
		if encoded != nil {
			if err := json.Unmarshal(encoded, &persistedTimeBase); err == nil &&
				currentTimeInSeconds <= persistedTimeBase {
				persistedTimeBase++
				store.timebase = persistedTimeBase * 1e9
			}
		}
		encoded, err = json.Marshal(persistedTimeBase)
		if err != nil {
			return err
		}
		err = tx.Bucket(timebaseBucket).Put([]byte("timebase"), []byte(encoded))
		return err
	})
	if err != nil {
		return err
	}
	if len(common.Configuration.ObjectsDataPath) > 0 {
		path = common.Configuration.ObjectsDataPath
	} else {
		path = common.Configuration.PersistenceRootPath + "/sync/local/"
	}
	err = os.MkdirAll(path, 0750)
	store.localDataPath = "file://" + path
	if err == nil {
		common.HealthStatus.ReconnectedToDatabase()
	}
	return err
}
func (store *BoltStorage) Stop() {
	store.db.Close()
}
func (store *BoltStorage) PerformMaintenance() {
	if common.Configuration.NodeType == common.CSS {
		currentTime := time.Now().UTC().Format(time.RFC3339)
		function := func(object boltObject) bool {
			if object.Meta.Expiration != "" && object.Meta.Expiration <= currentTime &&
				(object.Status == common.ReadyToSend || object.Status == common.NotReadyToSend) {
				return true
			}
			return false
		}
		err := store.deleteObjectsAndNotificationsHelper(function)
		if err != nil {
			if log.IsLogging(logger.ERROR) {
				log.Error("Error in PerformMaintenance: failed to remove expired objects. Error: %s\n", err)
			}
		} else if trace.IsLogging(logger.TRACE) {
			trace.Trace("Removing expired objects")
		}
	}
}
func (store *BoltStorage) Cleanup(isTest bool) common.SyncServiceError {
	var essDbPath string
	if isTest {
		cssDbPath := common.Configuration.PersistenceRootPath + "/sync/db/css-sync.db"
		essDbPath = common.Configuration.PersistenceRootPath + "/sync/db/ess-sync.db"
		if err := os.Remove(cssDbPath); err != nil {
			if log.IsLogging(logger.ERROR) {
				log.Error("Failed to remove css Bolt testing database")
			}
			return err
		}
		if log.IsLogging(logger.DEBUG) {
			log.Debug("CSS bolt db at %s is removed\n", cssDbPath)
		}
	} else {
		essDbPath = common.Configuration.PersistenceRootPath + "/sync/db/ess-sync.db"
	}
	if err := os.Remove(essDbPath); err != nil {
		if log.IsLogging(logger.ERROR) {
			log.Error("Failed to remove ess Bolt database")
		}
		return err
	}
	if log.IsLogging(logger.DEBUG) {
		log.Debug("ESS bolt db at %s is removed\n", essDbPath)
	}
	return nil
}