// query using LSI or GSI
func (t *DynamoTable) GetByIndex(idx string, values ...Any) ([]map[string]interface{}, error) {
	index, ok := t.indexes[idx]
	if !ok {
		log.Error("[DynamoDB] Cannot find the index name, table="+t.name, idx)
		log.Error("[DynamoDB] indexes on table="+t.name, t.indexes)
	}

	hashKey := index.GetHashKeyName()
	rangeKey := index.GetRangeKeyName()

	keys := make(map[string]*SDK.Condition)
	keys[hashKey] = &SDK.Condition{
		AttributeValueList: []*SDK.AttributeValue{createAttributeValue(values[0])},
		ComparisonOperator: String(ComparisonOperatorEQ),
	}
	if len(values) > 1 && rangeKey != "" {
		keys[rangeKey] = &SDK.Condition{
			AttributeValueList: []*SDK.AttributeValue{createAttributeValue(values[1])},
			ComparisonOperator: String(ComparisonOperatorEQ),
		}
	}

	in := &SDK.QueryInput{
		TableName:     String(t.name),
		KeyConditions: keys,
		IndexName:     &idx,
	}
	return t.Query(in)
}

// Get message from the queue with limit
func (q *Queue) Fetch(num int) (*SDK.ReceiveMessageOutput, error) {
	// use long-polling for 1sec when to get multiple messages
	var wait int
	if num > 1 {
		wait = 1
	} else {
		wait = 0
	}

	// receive message from server
	resp, err := q.client.ReceiveMessage(&SDK.ReceiveMessageInput{
		QueueURL:            q.url,
		WaitTimeSeconds:     Long(wait),
		MaxNumberOfMessages: Long(num),
		VisibilityTimeout:   Long(defaultExpireSecond),
	})
	if err != nil {
		log.Error("[SQS] error on `ReceiveMessage` operation, queue="+q.name, err.Error())
	}

	// delete messages automatically
	if q.autoDel && len(resp.Messages) > 0 {
		q.AddDeleteList(resp.Messages)
		defer q.DeleteListItems()
	}

	return resp, err
}

// Send messages in the send spool
func (q *Queue) Send() error {
	// pack the messages ten each to meet the SQS restriction.
	messages := make(map[int][]*SDK.SendMessageBatchRequestEntry)
	if len(q.messages) > 10 {
		for i, msg := range q.messages {
			v := (i + 1) / 10
			messages[v] = append(messages[v], msg)
		}
	} else {
		messages[0] = append(messages[0], q.messages...)
	}

	var err error = nil
	errStr := ""
	// send message
	for i := 0; i < len(messages); i++ {
		e := q.send(messages[i])
		if e != nil {
			log.Error("[SQS] error on `SendMessageBatch` operation, queue="+q.name, e.Error())
			errStr = errStr + "," + e.Error()
		}
	}
	if errStr != "" {
		err = errors.New(errStr)
	}
	return err
}

// get mapped-items with Query operation
func (t *DynamoTable) Query(in *SDK.QueryInput) ([]map[string]interface{}, error) {
	req, err := t.db.client.Query(in)
	if err != nil {
		log.Error("[DynamoDB] Error in `Query` operation, table="+t.name, err)
		return nil, err
	}
	return t.convertItemsToMapArray(req.Items), nil
}

// Create new SQS Queue
func (svc *AmazonSQS) CreateQueue(in *SDK.CreateQueueInput) error {
	data, err := svc.client.CreateQueue(in)
	if err != nil {
		log.Error("[SQS] Error on `CreateQueue` operation, queue="+*in.QueueName, err)
		return err
	}
	log.Info("[SQS] Complete CreateQueue, queue="+*in.QueueName, *(data.QueueURL))
	return nil
}

// Create new DynamoDB table
func (d *AmazonDynamoDB) CreateTable(in *SDK.CreateTableInput) error {
	data, err := d.client.CreateTable(in)
	if err != nil {
		log.Error("[DynamoDB] Error on `CreateTable` operation, table="+*in.TableName, err)
		return err
	}
	log.Info("[DynamoDB] Complete CreateTable, table="+*in.TableName, data.TableDescription.TableStatus)
	return nil
}

// Add message spool from map data
func (q *Queue) AddMessageMap(message map[string]interface{}) error {
	msg, err := json.Marshal(message)
	if err != nil {
		log.Error("[SQS] error on `json.Marshal`, msg="+fmt.Sprint(msg), err.Error())
		return err
	}
	q.AddMessage(string(msg))
	return nil
}

func (t *DynamoTable) Desc() (*SDK.TableDescription, error) {
	req, err := t.db.client.DescribeTable(&SDK.DescribeTableInput{
		TableName: String(t.name),
	})
	if err != nil {
		log.Error("[DynamoDB] Error in `DescribeTable` operation, table="+t.name, err)
		return nil, err
	}
	return req.Table, nil
}

// Delete all messages on the Queue
func (q *Queue) Purge() error {
	_, err := q.client.PurgeQueue(&SDK.PurgeQueueInput{
		QueueURL: q.url,
	})
	if err != nil {
		log.Error("[SQS] error on `PurgeQueue`, queue="+q.name, err.Error())
		return err
	}
	return nil
}

// GetEndpoint
func (svc *AmazonSNS) GetEndpoint(arn string) (*SNSEndpoint, error) {
	in := &SDK.GetEndpointAttributesInput{
		EndpointARN: String(arn),
	}
	resp, err := svc.Client.GetEndpointAttributes(in)
	if err != nil {
		log.Error("[SNS] error on `GetEndpointAttributes` operation, arn="+arn, err.Error())
		return nil, err
	}
	attr := resp.Attributes
	ep := svc.NewApplicationEndpoint(arn)
	ep.token = *attr["Token"]
	ep.enable, err = strconv.ParseBool(*attr["Enabled"])
	if err != nil {
		log.Error("[SNS] error on `Endpoint Attributes` Enabled="+*attr["Enabled"], err.Error())
		ep.enable = false
	}
	return ep, err
}

// Delete a message from server
func (q *Queue) DeleteMessage(msg *SDK.Message) error {
	_, err := q.client.DeleteMessage(&SDK.DeleteMessageInput{
		QueueURL:      q.url,
		ReceiptHandle: msg.ReceiptHandle,
	})
	if err != nil {
		log.Error("[SQS] error on `DeleteMessage`, queue="+q.name, err.Error())
	}
	return err
}

// delete object of target path
func (b *Bucket) DeleteObject(path string) error {
	_, err := b.client.DeleteObject(&SDK.DeleteObjectInput{
		Bucket: String(b.name),
		Key:    String(path),
	})
	if err != nil {
		log.Error("[S3] error on `DeleteObject` operation, bucket="+b.name, err.Error())
	}
	return err
}

// Create Endpoint(add device) and return `EndpointARN`
func (a *SNSApp) createEndpoint(token string) (string, error) {
	in := &SDK.CreatePlatformEndpointInput{
		PlatformApplicationARN: String(a.arn),
		Token: String(token),
	}
	resp, err := a.svc.Client.CreatePlatformEndpoint(in)
	if err != nil {
		log.Error("[SNS] error on `CreatePlatformEndpoint` operation, token="+token, err.Error())
		return "", err
	}
	return *resp.EndpointARN, nil
}

// Delete DynamoDB table
func (d *AmazonDynamoDB) DeleteTable(name string) error {
	in := &SDK.DeleteTableInput{
		TableName: String(name),
	}
	data, err := d.client.DeleteTable(in)
	if err != nil {
		log.Error("[DynamoDB] Error on `DeleteTable` operation, table="+*in.TableName, err)
		return err
	}
	log.Info("[DynamoDB] Complete DeleteTable, table="+*in.TableName, data.TableDescription.TableStatus)
	return nil
}

// fetch object from target S3 path
func (b *Bucket) getObject(path string) (io.Reader, error) {
	req := SDK.GetObjectInput{
		Bucket: &b.name,
		Key:    &path,
	}
	out, err := b.client.GetObject(&req)
	if err != nil {
		log.Error("[S3] error on `GetObject` operation, bucket="+b.name, err.Error())
		return nil, err
	}
	return out.Body, err
}

// Subscribe
func (t *SNSTopic) Subscribe(endpoint *SNSEndpoint) (string, error) {
	resp, err := t.svc.Client.Subscribe(&SDK.SubscribeInput{
		Endpoint: String(endpoint.arn),
		Protocol: String(endpoint.protocol),
		TopicARN: String(t.arn),
	})
	if err != nil {
		log.Error("[SNS] error on `Subscribe` operation, topic="+t.arn, err.Error())
		return "", err
	}
	return *resp.SubscriptionARN, nil
}

// get mapped-items with Scan operation
func (t *DynamoTable) Scan() ([]map[string]interface{}, error) {
	in := &SDK.ScanInput{
		TableName: String(t.name),
		Limit:     Long(1000),
	}
	req, err := t.db.client.Scan(in)
	if err != nil {
		log.Error("[DynamoDB] Error in `Scan` operation, table="+t.name, err)
		return nil, err
	}
	return t.convertItemsToMapArray(req.Items), nil
}

// Create SNS Topic and return `TopicARN`
func (svc *AmazonSNS) createTopic(name string) (string, error) {
	prefix := config.GetConfigValue(snsConfigSectionName, "prefix", defaultPrefix)
	in := &SDK.CreateTopicInput{
		Name: String(prefix + name),
	}
	resp, err := svc.Client.CreateTopic(in)
	if err != nil {
		log.Error("[SNS] error on `CreateTopic` operation, name="+name, err.Error())
		return "", err
	}
	return *resp.TopicARN, nil
}

// Delete a packed message
func (q *Queue) delete(msg []*SDK.DeleteMessageBatchRequestEntry) error {
	if len(msg) < 1 {
		return nil
	}
	res, err := q.client.DeleteMessageBatch(&SDK.DeleteMessageBatchInput{
		Entries:  msg,
		QueueURL: q.url,
	})
	if err != nil {
		log.Error("[SQS] error on `DeleteMessageBatch`, queue="+q.name, err.Error())
		q.failedDelete = append(q.failedDelete, res.Failed...)
	}

	return err
}

// execute put operation for all tables in write spool
func (d *AmazonDynamoDB) PutAll() error {
	var errs []string
	for name, _ := range d.writeTables {
		err := d.tables[name].Put()
		if err != nil {
			errs = append(errs, err.Error())
			log.Error("[DynamoDB] Error on `Put` operation, table="+name, err.Error())
		}
		d.removeWriteTable(name)
	}
	if len(errs) != 0 {
		return errors.New(strings.Join(errs, "\n"))
	}
	return nil
}