/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing,
 * software distributed under the License is distributed on an
 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 * KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations
 * under the License.
 */

package op

import (
	"encoding/binary"
	"fmt"
	"math"
	"time"

	"github.com/apache/incubator-pegasus/go-client/idl/base"
)

func validateHashKey(hashKey []byte) error {
	if hashKey == nil {
		return fmt.Errorf("InvalidParameter: hashkey must not be nil")
	}
	if len(hashKey) == 0 {
		return fmt.Errorf("InvalidParameter: hashkey must not be empty")
	}
	if len(hashKey) > math.MaxUint16 {
		return fmt.Errorf("InvalidParameter: length of hashkey (%d) must be less than %d", len(hashKey), math.MaxUint16)
	}
	return nil
}

func validateValue(value []byte) error {
	if value == nil {
		return fmt.Errorf("InvalidParameter: value must not be nil")
	}
	return nil
}

func validateValues(values [][]byte) error {
	if values == nil {
		return fmt.Errorf("InvalidParameter: values must not be nil")
	}
	if len(values) == 0 {
		return fmt.Errorf("InvalidParameter: values must not be empty")
	}
	for i, value := range values {
		if value == nil {
			return fmt.Errorf("InvalidParameter: values[%d] must not be nil", i)
		}
	}
	return nil
}

func validateSortKey(sortKey []byte) error {
	if sortKey == nil {
		return fmt.Errorf("InvalidParameter: sortkey must not be nil")
	}
	return nil
}

func validateSortKeys(sortKeys [][]byte) error {
	if sortKeys == nil {
		return fmt.Errorf("InvalidParameter: sortkeys must not be nil")
	}
	if len(sortKeys) == 0 {
		return fmt.Errorf("InvalidParameter: sortkeys must not be empty")
	}
	for i, sortKey := range sortKeys {
		if sortKey == nil {
			return fmt.Errorf("InvalidParameter: sortkeys[%d] must not be nil", i)
		}
	}
	return nil
}

func validateTTL(TTL time.Duration) error {
	if TTL < 0 {
		return fmt.Errorf("InvalidParameter: TTL[%d] must be greater than 0", TTL)
	}
	return nil
}

func encodeHashKeySortKey(hashKey []byte, sortKey []byte) *base.Blob {
	hashKeyLen := len(hashKey)
	sortKeyLen := len(sortKey)

	blob := &base.Blob{
		Data: make([]byte, 2+hashKeyLen+sortKeyLen),
	}

	binary.BigEndian.PutUint16(blob.Data, uint16(hashKeyLen))

	if hashKeyLen > 0 {
		copy(blob.Data[2:], hashKey)
	}

	if sortKeyLen > 0 {
		copy(blob.Data[2+hashKeyLen:], sortKey)
	}

	return blob
}

func expireTsSeconds(ttl time.Duration) int32 {
	if ttl == 0 {
		return 0
	}
	// 1451606400 means seconds since 2016.01.01-00:00:00 GMT
	return int32(ttl.Seconds()) + int32(time.Now().Unix()-1451606400)
}

type rpcResponse interface {
	GetError() int32
}

func wrapRPCFailure(resp rpcResponse, err error) error {
	if err != nil {
		return err
	}
	err = base.NewRocksDBErrFromInt(resp.GetError())
	if err != nil {
		return err
	}
	return nil
}
