// 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)
}

func (q *Query) query(in *SDK.QueryInput) (*QueryResult, error) {
	if !q.HasCondition() {
		errData := &DynamoError{}
		errData.AddMessage("condition is missing, you must specify at least one condition")
		return nil, errData
	}
	in.KeyConditionExpression = q.formatConditions()
	in.ExpressionAttributeValues = q.formatConditionValues()
	in.ExpressionAttributeNames = q.formatConditionNames()

	if q.HasIndex() {
		in.IndexName = String(q.index)
	}
	if q.HasLimit() {
		in.Limit = Long(q.limit)
	}

	req, err := q.table.db.client.Query(in)
	if err != nil {
		log.Error("[DynamoDB] Error in `Query` operation, table="+q.table.name, err)
		return nil, err
	}

	res := &QueryResult{
		Items:            req.Items,
		LastEvaluatedKey: req.LastEvaluatedKey,
		Count:            *req.Count,
		ScannedCount:     *req.ScannedCount,
	}
	return res, nil
}

// 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
}

// 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...)
	}

	err := &SQSError{}
	// 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())
			err.AddMessage(e.Error())
		}
	}

	if err.HasError() {
		return err
	}
	return nil
}

// make sns message for Apple Push Notification Service
func composeMessageAPNS(msg string, opt map[string]interface{}) string {
	aps := make(map[string]interface{})
	aps[apnsKeyMessage] = msg

	aps[apnsKeySound] = "default"
	if v, ok := opt[apnsKeySound]; ok {
		aps[apnsKeySound] = v
	}

	if v, ok := opt[apnsKeyBadge]; ok {
		aps[apnsKeyBadge] = v
	}

	message := make(map[string]interface{})
	message["aps"] = aps
	for k, v := range opt {
		switch k {
		case apnsKeySound:
			continue
		case apnsKeyBadge:
			continue
		default:
			message[k] = v
		}
	}

	payload, err := json.Marshal(message)
	if err != nil {
		log.Error("[SNS] error on json.Marshal", err.Error())
	}
	return string(payload)
}

// 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
}

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

	// 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())
	}

	var list []*Message
	if resp == nil || len(resp.Messages) == 0 {
		return list, err
	}

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

	for _, msg := range resp.Messages {
		list = append(list, NewMessage(msg))
	}

	return list, err
}

func (q *Query) AndLT(key string, val interface{}) {
	if _, ok := q.conditions[key]; ok {
		log.Error("[DynamoDB] only contain one condition per key, key="+key, val)
	}
	cond := newQueryCondition(key, val)
	cond.Condition = conditionLT
	q.conditions[key] = cond
}

// 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
}

// RegisterEndpoint creates endpoint(device) to application
func (svc *AmazonSNS) RegisterEndpoint(device, token string) (*SNSEndpoint, error) {
	app, err := svc.getApp(device)
	if err != nil {
		log.Error(err.Error(), token)
		return nil, err
	}

	return app.CreateEndpoint(token)
}

// 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
}

func (q *Query) AndBETWEEN(key string, from, to interface{}) {
	if _, ok := q.conditions[key]; ok {
		log.Error("[DynamoDB] only contain one condition per key, key="+key, from)
	}
	cond := newQueryCondition(key, from)
	cond.SubValue = to
	cond.Condition = conditionBETWEEN
	q.conditions[key] = cond
}

// 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
}

// 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
}

// RegisterEndpointWithUserData creates endpoint(device) and CustomUserData to application
func (svc *AmazonSNS) RegisterEndpointWithUserData(device, token, userData string) (*SNSEndpoint, error) {
	app, err := svc.getApp(device)
	if err != nil {
		log.Error(err.Error(), token)
		return nil, err
	}
	app.SetUserData(userData)
	return app.CreateEndpoint(token)
}

// 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 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
}

// 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
}

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 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
}