func ListObjectsWithDestinationPolicyByService(orgID, serviceOrgID, serviceName string) ([]common.ObjectDestinationPolicy, common.SyncServiceError) {
	apiLock.RLock()
	defer apiLock.RUnlock()
	common.HealthStatus.ClientRequestReceived()
	objects, err := store.RetrieveObjectsWithDestinationPolicyByService(orgID, serviceOrgID, serviceName)
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("In ListObjectsWithDestinationPolicyByService. Get %s/%s. Returned %d objects\n",
			serviceOrgID, serviceName, len(objects))
	}
	return objects, err
}
ListObjectsWithDestinationPolicyByService provides a list of objects that have a DestinationPolicy and are
associated with a service
package base
import (
	"bytes"
	"encoding/base64"
	"fmt"
	"io"
	"math"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"
	"github.com/open-horizon/edge-sync-service/common"
	"github.com/open-horizon/edge-sync-service/core/communications"
	"github.com/open-horizon/edge-sync-service/core/dataURI"
	"github.com/open-horizon/edge-sync-service/core/dataVerifier"
	"github.com/open-horizon/edge-sync-service/core/storage"
	"github.com/open-horizon/edge-utilities/logger"
	"github.com/open-horizon/edge-utilities/logger/log"
	"github.com/open-horizon/edge-utilities/logger/trace"
)
var apiLock sync.RWMutex
var apiObjectLocks common.Locks
func init() {
	apiObjectLocks = *common.NewLocks("api")
}
func UpdateObject(orgID string, objectType string, objectID string, metaData common.MetaData, data []byte) common.SyncServiceError {
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("In UpdateObject. Update %s %s %s\n", orgID, objectType, objectID)
	}
	common.HealthStatus.ClientRequestReceived()
	if !common.IsValidName(orgID) {
		return &common.InvalidRequest{Message: fmt.Sprintf("Organization ID (%s) contains invalid characters", orgID)}
	}
	if metaData.ObjectID == "" {
		return &common.InvalidRequest{Message: "Object's meta data does not contain object's ID"}
	}
	if objectID != metaData.ObjectID {
		message := fmt.Sprintf("Object ID (%+v) in the URL doesn't match the object-id (%+v) in the payload", objectID, metaData.ObjectID)
		return &common.InvalidRequest{Message: message}
	}
	if !common.IsValidName(objectID) {
		return &common.InvalidRequest{Message: fmt.Sprintf("Object ID (%s) contains invalid characters", objectID)}
	}
	if metaData.ObjectType == "" {
		return &common.InvalidRequest{Message: "Object's meta data does not contain object's type"}
	}
	if objectType != metaData.ObjectType {
		message := fmt.Sprintf("Object type (%+v) in the URL doesn't match the object-type (%+v) in the payload", objectType, metaData.ObjectType)
		return &common.InvalidRequest{Message: message}
	}
	if !common.IsValidName(objectType) {
		return &common.InvalidRequest{Message: fmt.Sprintf("Object type (%s) contains invalid characters", objectType)}
	}
	if metaData.HashAlgorithm != "" && metaData.HashAlgorithm != common.Sha1 && metaData.HashAlgorithm != common.Sha256 {
		return &common.InvalidRequest{Message: fmt.Sprintf("Hash Algorithm (%s) is not supported. Please use %s or %s if specify", metaData.HashAlgorithm, common.Sha1, common.Sha256)}
	}
	if metaData.PublicKey != "" {
		if _, err := base64.StdEncoding.DecodeString(metaData.PublicKey); err != nil {
			return &common.InvalidRequest{Message: "PublicKey is not base64 encoded. Error: " + err.Error()}
		}
	}
	if metaData.Signature != "" {
		if _, err := base64.StdEncoding.DecodeString(metaData.Signature); err != nil {
			return &common.InvalidRequest{Message: "Signature is not base64 encoded. Error: " + err.Error()}
		}
	}
	if metaData.Expiration != "" {
		if common.Configuration.NodeType == common.ESS {
			return &common.InvalidRequest{Message: "Object expiration is disabled on ESS"}
		}
		expiration, err := time.Parse(time.RFC3339, metaData.Expiration)
		if err != nil {
			return &common.InvalidRequest{Message: "Failed to parse expiration in object's meta data. Error: " + err.Error()}
		}
		if time.Now().After(expiration) {
			return &common.InvalidRequest{Message: "Invalid expiration time in object's meta data"}
		}
	}
	if metaData.Version != "" && !common.IsValidName(metaData.Version) {
		return &common.InvalidRequest{Message: fmt.Sprintf("Version (%s) contains one/some invalid characters (eg: <, >, =, ', \", &, space, \\, /)", metaData.Version)}
	}
	if metaData.Description != "" && common.IsInvalidDescription(metaData.Description) {
		return &common.InvalidRequest{Message: fmt.Sprintf("Description (%s) contains one/some invalid characters (<, >, =, ', \", &, space, \\, /)", metaData.Description)}
	}
	if metaData.MetaOnly && len(data) != 0 {
		return &common.InvalidRequest{Message: "Can't update data if MetaOnly is true"}
	}
	if metaData.DestID != "" && metaData.DestType == "" {
		return &common.InvalidRequest{Message: "Destination ID provided without destination type in object's meta data"}
	}
	if metaData.DestinationsList != nil && common.Configuration.NodeType == common.ESS {
		return &common.InvalidRequest{Message: "Destinations list is not supported for ESS"}
	}
	if metaData.DestinationsList != nil && metaData.DestType != "" {
		return &common.InvalidRequest{Message: "Both destinations list and destination type are specified"}
	}
	if validatedDestList, _ := common.ValidateDestinationListInput(metaData.DestinationsList); validatedDestList == false {
		return &common.InvalidRequest{Message: "Unsupported char <, > in destinationsList."}
	}
	if metaData.DestinationPolicy != nil {
		if metaData.DestType != "" {
			return &common.InvalidRequest{Message: "Both destination policy and destination type are specified"}
		}
		if metaData.DestinationsList != nil {
			return &common.InvalidRequest{Message: "Both destination policy and destination list are specified"}
		}
		properties := metaData.DestinationPolicy.Properties
		for _, property := range properties {
			if len(property.Name) == 0 {
				return &common.InvalidRequest{Message: "A property in the DestinationPolicy must have a name"}
			}
		}
		services := metaData.DestinationPolicy.Services
		for _, service := range services {
			if len(service.OrgID) == 0 || len(service.Arch) == 0 || len(service.ServiceName) == 0 || len(service.Version) == 0 {
				return &common.InvalidRequest{
					Message: "A service in a DestinationPolicy must have an organization ID, architecture, service name, and version specified"}
			}
			if _, err := common.ParseSemVerRange(service.Version); err != nil {
				return &common.InvalidRequest{
					Message: fmt.Sprintf("A service in the DestinationPolicy has an invalid version `%s`", service.Version)}
			}
		}
	}
	if metaData.DestType != "" && !common.IsValidName(metaData.DestType) {
		return &common.InvalidRequest{Message: fmt.Sprintf("Destination type (%s) contains invalid characters", metaData.DestType)}
	}
	if metaData.AutoDelete && metaData.DestinationsList == nil && metaData.DestID == "" {
		return &common.InvalidRequest{Message: "AutoDelete can be used only for objects with DestinationsList or DestID set"}
	}
	if metaData.ActivationTime != "" && metaData.Inactive {
		activation, err := time.Parse(time.RFC3339, metaData.ActivationTime)
		if err != nil {
			return &common.InvalidRequest{Message: "Failed to parse activation time in object's meta data. Error: " + err.Error()}
		}
		if time.Now().After(activation) {
			return &common.InvalidRequest{Message: "Invalid activation time in object's meta data"}
		}
	}
	if metaData.Deleted {
		return &common.InvalidRequest{Message: "Object marked as deleted"}
	}
	if metaData.DestinationDataURI != "" {
		if common.Configuration.NodeType == common.ESS {
			return &common.InvalidRequest{Message: "Data URI is disabled on CSS"}
		}
		uri, err := url.Parse(metaData.DestinationDataURI)
		if err != nil || !strings.EqualFold(uri.Scheme, "file") || uri.Host != "" {
			return &common.InvalidRequest{Message: "Invalid destination data URI"}
		}
	}
	if metaData.SourceDataURI != "" {
		if common.Configuration.NodeType == common.CSS {
			return &common.InvalidRequest{Message: "Data URI is disabled on CSS"}
		}
		if data != nil {
			return &common.InvalidRequest{Message: "Both source data URI and data are set"}
		}
		uri, err := url.Parse(metaData.SourceDataURI)
		if err != nil || !strings.EqualFold(uri.Scheme, "file") || uri.Host != "" {
			return &common.InvalidRequest{Message: "Invalid source data URI"}
		}
		if fi, err := os.Stat(uri.Path); err == nil {
			metaData.ObjectSize = fi.Size()
		} else {
			log.Error(" Invalid source data URI: %s, failed to get file information for the file, err= %v\n", metaData.SourceDataURI, err)
			return &common.InvalidRequest{Message: "Invalid source data URI"}
		}
	}
	if metaData.OriginType == "" || metaData.OriginID == "" {
		metaData.OriginType = common.Configuration.DestinationType
		metaData.OriginID = common.Configuration.DestinationID
	} else {
		if !common.IsValidName(metaData.OriginType) {
			return &common.InvalidRequest{Message: fmt.Sprintf("OriginType (%s) contains one/some invalid characters (eg: <, >, =, ', \", &, space, \\, /)", metaData.OriginType)}
		} else if !common.IsValidName(metaData.OriginID) {
			return &common.InvalidRequest{Message: fmt.Sprintf("OriginID (%s) contains one/some invalid characters (eg: <, >, =, ', \", &, space, \\, /)", metaData.OriginID)}
		}
	}
	if metaData.DestOrgID == "" {
		metaData.DestOrgID = orgID
	} else if !common.IsValidName(metaData.DestOrgID) {
		return &common.InvalidRequest{Message: fmt.Sprintf("DestOrgID (%s) contains one/some invalid characters (eg: <, >, =, ', \", &, space, \\, /)", metaData.DestOrgID)}
	}
	if metaData.ExpectedConsumers == 0 {
		metaData.ExpectedConsumers = 1
	} else if metaData.ExpectedConsumers == -1 {
		metaData.ExpectedConsumers = math.MaxInt32
	}
	status := common.NotReadyToSend
	if data != nil || metaData.Link != "" || metaData.NoData || metaData.SourceDataURI != "" {
		status = common.ReadyToSend
	} else if metaData.MetaOnly {
		reader, err := store.RetrieveObjectData(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
		if err != nil {
			return err
		}
		if reader != nil {
			status = common.ReadyToSend
			store.CloseDataReader(reader)
		}
	}
	lockIndex := common.HashStrings(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
	apiObjectLocks.Lock(lockIndex)
	common.ObjectLocks.Lock(lockIndex)
	if metaData.NoData {
		data = nil
		metaData.Link = ""
		metaData.SourceDataURI = ""
		metaData.PublicKey = ""
		metaData.Signature = ""
	} else if data != nil {
		if common.IsValidHashAlgorithm(metaData.HashAlgorithm) && metaData.PublicKey != "" && metaData.Signature != "" {
			dataReader := bytes.NewReader(data)
			dataVf := dataVerifier.NewDataVerifier(metaData.HashAlgorithm, metaData.PublicKey, metaData.Signature)
			if success, err := dataVf.VerifyDataSignature(dataReader, orgID, objectType, objectID, ""); !success || err != nil {
				if trace.IsLogging(logger.ERROR) {
					trace.Error("Failed to verify data for object %s %s, remove temp data\n", objectType, objectID)
				}
				dataVf.RemoveTempData(orgID, objectType, objectID, "")
				common.ObjectLocks.Unlock(lockIndex)
				apiObjectLocks.Unlock(lockIndex)
				return err
			}
			dataVf.RemoveTempData(orgID, objectType, objectID, "")
		}
		metaData.ObjectSize = int64(len(data))
	}
	metaData.ChunkSize = common.Configuration.MaxDataChunkSize
	_, existingObjStatus, _ := store.RetrieveObjectAndStatus(orgID, objectType, objectID)
	if existingObjStatus != "" && existingObjStatus != common.ReadyToSend && existingObjStatus != common.NotReadyToSend {
		common.ObjectLocks.Unlock(lockIndex)
		apiObjectLocks.Unlock(lockIndex)
		return &common.InvalidRequest{Message: "Can't update object of the receiving side"}
	}
	deletedDestinations, err := store.StoreObject(metaData, data, status)
	if err != nil {
		common.ObjectLocks.Unlock(lockIndex)
		apiObjectLocks.Unlock(lockIndex)
		return err
	}
	store.DeleteNotificationRecords(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID, "", "")
	if status == common.NotReadyToSend || metaData.Inactive {
		common.ObjectLocks.Unlock(lockIndex)
		apiObjectLocks.Unlock(lockIndex)
		return nil
	}
	updatedMetaData, err := store.RetrieveObject(metaData.DestOrgID, metaData.ObjectType, metaData.ObjectID)
	if err != nil {
		common.ObjectLocks.Unlock(lockIndex)
		apiObjectLocks.Unlock(lockIndex)
		return err
	}
	common.ObjectLocks.Unlock(lockIndex)
	apiObjectLocks.Unlock(lockIndex)
	if len(deletedDestinations) != 0 {
		if trace.IsLogging(logger.DEBUG) {
			trace.Debug("In UpdateObject. Send object to objectQueue for delete%s %s\n", objectType, objectID)
		}
		objectInQueue := common.ObjectInQueue{NotificationAction: common.Delete, NotificationType: common.TypeDestination, Object: *updatedMetaData, Destinations: deletedDestinations}
		objectQueue.SendObjectToQueue(objectInQueue)
		if trace.IsLogging(logger.DEBUG) {
			trace.Debug("In UpdateObject. Continue to updateDestination %s %s\n", objectType, objectID)
		}
	}
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("In UpdateObject. Send object to objectQueue for update%s %s\n", objectType, objectID)
	}
	objectInQueue := common.ObjectInQueue{NotificationAction: common.Update, NotificationType: common.TypeObject, Object: *updatedMetaData, Destinations: []common.StoreDestinationStatus{}}
	objectQueue.SendObjectToQueue(objectInQueue)
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("In UpdateObject. Return response for UpdateObject %s %s\n", objectType, objectID)
	}
	return nil
}
func GetObjectStatus(orgID string, objectType string, objectID string) (string, common.SyncServiceError) {
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("In GetObjectStatus. Get status of %s %s\n", objectType, objectID)
	}
	common.HealthStatus.ClientRequestReceived()
	lockIndex := common.HashStrings(orgID, objectType, objectID)
	apiObjectLocks.RLock(lockIndex)
	defer apiObjectLocks.RUnlock(lockIndex)
	return store.RetrieveObjectStatus(orgID, objectType, objectID)
}
func ListUpdatedObjects(orgID string, objectType string, received bool) ([]common.MetaData, common.SyncServiceError) {
	apiLock.RLock()
	defer apiLock.RUnlock()
	common.HealthStatus.ClientRequestReceived()
	updatedObjects, err := store.RetrieveUpdatedObjects(orgID, objectType, received)
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("In ListUpdatedObjects. Get %s %s. returned %d objects\n", orgID, objectType, len(updatedObjects))
	}
	return updatedObjects, err
}
func ListObjectsWithDestinationPolicy(orgID string, received bool) ([]common.ObjectDestinationPolicy, common.SyncServiceError) {
	apiLock.RLock()
	defer apiLock.RUnlock()
	common.HealthStatus.ClientRequestReceived()
	objects, err := store.RetrieveObjectsWithDestinationPolicy(orgID, received)
	if trace.IsLogging(logger.DEBUG) {
		trace.Debug("In ListObjectsWithDestinationPolicy. Get %s. Returned %d objects\n", orgID, len(objects))
	}
	return objects, err
}