func (db *DB) ZUnionStore(destKey []byte, srcKeys [][]byte, weights []int64, aggregate byte) (int64, error) {

	var destMap = map[string]int64{}
	aggregateFunc := getAggregateFunc(aggregate)
	if aggregateFunc == nil {
		return 0, errInvalidAggregate
	}
	if len(srcKeys) < 1 {
		return 0, errInvalidSrcKeyNum
	}
	if weights != nil {
		if len(srcKeys) != len(weights) {
			return 0, errInvalidWeightNum
		}
	} else {
		weights = make([]int64, len(srcKeys))
		for i := 0; i < len(weights); i++ {
			weights[i] = 1
		}
	}

	for i, key := range srcKeys {
		scorePairs, err := db.ZRange(key, 0, -1)
		if err != nil {
			return 0, err
		}
		for _, pair := range scorePairs {
			if score, ok := destMap[hack.String(pair.Member)]; !ok {
				destMap[hack.String(pair.Member)] = pair.Score * weights[i]
			} else {
				destMap[hack.String(pair.Member)] = aggregateFunc(score, pair.Score*weights[i])
			}
		}
	}

	t := db.zsetBatch
	t.Lock()
	defer t.Unlock()

	db.zDelete(t, destKey)

	for member, score := range destMap {
		if err := checkZSetKMSize(destKey, []byte(member)); err != nil {
			return 0, err
		}

		if _, err := db.zSetItem(t, destKey, score, []byte(member)); err != nil {
			return 0, err
		}
	}

	var n = int64(len(destMap))
	sk := db.zEncodeSizeKey(destKey)
	t.Put(sk, PutInt64(n))

	if err := t.Commit(); err != nil {
		return 0, err
	}
	return n, nil
}

func (w *httpWriter) writeFVPairArray(lst []ledis.FVPair) {
	m := make(map[string]string)
	for _, elem := range lst {
		m[hack.String(elem.Field)] = hack.String(elem.Value)
	}
	w.genericWrite(m)
}

// XSELECT db THEN command
func (c *respClient) handleXSelectCmd() error {
	if len(c.args) <= 2 {
		// invalid command format
		return fmt.Errorf("invalid format for XSELECT, must XSELECT db THEN your command")
	}

	if hack.String(upperSlice(c.args[1])) != "THEN" {
		// invalid command format, just resturn here
		return fmt.Errorf("invalid format for XSELECT, must XSELECT db THEN your command")
	}

	index, err := strconv.Atoi(hack.String(c.args[0]))
	if err != nil {
		return fmt.Errorf("invalid db for XSELECT, err %v", err)
	}

	db, err := c.app.ldb.Select(index)
	if err != nil {
		return fmt.Errorf("invalid db for XSELECT, err %v", err)
	}

	c.db = db

	c.cmd = hack.String(lowerSlice(c.args[2]))
	c.args = c.args[3:]

	return nil
}

//inner command, only for replication
//REPLCONF <option> <value> <option> <value> ...
func replconfCommand(c *client) error {
	args := c.args
	if len(args)%2 != 0 {
		return ErrCmdParams
	}

	if !c.app.ldb.ReplicationUsed() {
		return ledis.ErrRplNotSupport
	}

	//now only support "listening-port"
	for i := 0; i < len(args); i += 2 {
		switch strings.ToLower(hack.String(args[i])) {
		case "listening-port":
			var host string
			var err error
			if _, err = num.ParseUint16(hack.String(args[i+1])); err != nil {
				return err
			}
			if host, _, err = net.SplitHostPort(c.remoteAddr); err != nil {
				return err
			} else {
				c.slaveListeningAddr = net.JoinHostPort(host, hack.String(args[i+1]))
			}

			c.app.addSlave(c)
		default:
			return ErrSyntax
		}
	}

	c.resp.writeStatus(OK)
	return nil
}

func zparseRange(c *client, a1 []byte, a2 []byte) (start int, stop int, err error) {
	if start, err = strconv.Atoi(hack.String(a1)); err != nil {
		return
	}

	if stop, err = strconv.Atoi(hack.String(a2)); err != nil {
		return
	}

	return
}

func (db *DB) sInterGeneric(keys ...[]byte) ([][]byte, error) {
	destMap := make(map[string]bool)

	members, err := db.SMembers(keys[0])
	if err != nil {
		return nil, err
	}

	for _, m := range members {
		destMap[hack.String(m)] = true
	}

	for _, key := range keys[1:] {
		if err := checkKeySize(key); err != nil {
			return nil, err
		}

		members, err := db.SMembers(key)
		if err != nil {
			return nil, err
		} else if len(members) == 0 {
			return nil, err
		}

		tempMap := make(map[string]bool)
		for _, member := range members {
			if err := checkKeySize(member); err != nil {
				return nil, err
			}
			if _, ok := destMap[hack.String(member)]; ok {
				tempMap[hack.String(member)] = true //mark this item as selected
			}
		}
		destMap = tempMap //reduce the size of the result set
		if len(destMap) == 0 {
			return nil, nil
		}
	}

	slice := make([][]byte, len(destMap))
	idx := 0
	for k, v := range destMap {
		if !v {
			continue
		}

		slice[idx] = []byte(k)
		idx++
	}

	return slice, nil

}

func decodeString(data []byte, length int) (v string, n int) {
	if length < 256 {
		length = int(data[0])

		n = int(length) + 1
		v = hack.String(data[1:n])
	} else {
		length = int(binary.LittleEndian.Uint16(data[0:]))
		n = length + 2
		v = hack.String(data[2:n])
	}

	return
}

func (w *luaWriter) writeFVPairArray(lst []ledis.FVPair) {
	if lst == nil {
		w.l.PushBoolean(false)
		return
	}

	w.l.CreateTable(len(lst)*2, 0)
	for i, v := range lst {
		w.l.PushString(hack.String(v.Field))
		w.l.RawSeti(-2, 2*i+1)

		w.l.PushString(hack.String(v.Value))
		w.l.RawSeti(-2, 2*i+2)
	}
}

func (l *lBlockKeys) signal(key []byte, num int) {
	l.Lock()
	defer l.Unlock()

	s := hack.String(key)
	chs, ok := l.keys[s]
	if !ok {
		return
	}

	var n *list.Element

	i := 0
	for e := chs.Front(); e != nil && i < num; e = n {
		ch := e.Value.(lbKeyCh)
		n = e.Next()
		select {
		case ch <- key:
			chs.Remove(e)
			i++
		default:
			//waiter unwait
			chs.Remove(e)
		}
	}

	if chs.Len() == 0 {
		delete(l.keys, s)
	}
}

func scriptLoadCommand(c *client) error {
	s := c.app.script
	l := s.l

	if len(c.args) != 2 {
		return ErrCmdParams
	}

	h := sha1.Sum(c.args[1])
	key := hex.EncodeToString(h[0:20])

	if r := l.LoadString(hack.String(c.args[1])); r != 0 {
		err := fmt.Errorf("%s", l.ToString(-1))
		l.Pop(1)
		return err
	} else {
		l.PushValue(-1)
		l.SetGlobal(key)

		s.chunks[key] = struct{}{}
	}

	c.resp.writeBulk(hack.Slice(key))
	return nil
}

func scriptCommand(c *client) error {
	s := c.app.script
	l := s.l

	s.Lock()

	base := l.GetTop()

	defer func() {
		l.SetTop(base)
		s.Unlock()
	}()

	args := c.args

	if len(args) < 1 {
		return ErrCmdParams
	}

	switch strings.ToLower(hack.String(args[0])) {
	case "load":
		return scriptLoadCommand(c)
	case "exists":
		return scriptExistsCommand(c)
	case "flush":
		return scriptFlushCommand(c)
	default:
		return fmt.Errorf("invalid script %s", args[0])
	}

	return nil
}

func (w *luaWriter) writeBulk(b []byte) {
	if b == nil {
		w.l.PushBoolean(false)
	} else {
		w.l.PushString(hack.String(b))
	}
}

func convertBytesToString(b []byte) interface{} {
	if b == nil {
		return nil
	} else {
		return hack.String(b)
	}
}

func (w *httpWriter) writeScorePairArray(lst []ledis.ScorePair, withScores bool) {
	var arr []string
	if withScores {
		arr = make([]string, 2*len(lst))
		for i, data := range lst {
			arr[2*i] = hack.String(data.Member)
			arr[2*i+1] = strconv.FormatInt(data.Score, 10)
		}
	} else {
		arr = make([]string, len(lst))
		for i, data := range lst {
			arr[i] = hack.String(data.Member)
		}
	}
	w.genericWrite(arr)
}

func (w *httpWriter) writeBulk(b []byte) {
	if b == nil {
		w.genericWrite(nil)
	} else {
		w.genericWrite(hack.String(b))
	}
}

func (db *DB) sUnionGeneric(keys ...[]byte) ([][]byte, error) {
	dstMap := make(map[string]bool)

	for _, key := range keys {
		if err := checkKeySize(key); err != nil {
			return nil, err
		}

		members, err := db.SMembers(key)
		if err != nil {
			return nil, err
		}

		for _, member := range members {
			dstMap[hack.String(member)] = true
		}
	}

	slice := make([][]byte, len(dstMap))
	idx := 0
	for k, v := range dstMap {
		if !v {
			continue
		}
		slice[idx] = []byte(k)
		idx++
	}

	return slice, nil
}

func selectCommand(c *client) error {
	if len(c.args) != 1 {
		return ErrCmdParams
	}

	if index, err := strconv.Atoi(hack.String(c.args[0])); err != nil {
		return err
	} else {
		// if c.db.IsInMulti() {
		// 	if err := c.script.Select(index); err != nil {
		// 		return err
		// 	} else {
		// 		c.db = c.script.DB
		// 	}
		// } else {
		// 	if db, err := c.ldb.Select(index); err != nil {
		// 		return err
		// 	} else {
		// 		c.db = db
		// 	}
		// }

		if db, err := c.ldb.Select(index); err != nil {
			return err
		} else {
			c.db = db
		}

		c.resp.writeStatus(OK)
	}

	return nil
}

func listAdaptor(db *DB) *adaptor {
	adp := new(adaptor)
	adp.showIdent = func() string {
		return "list-adptor"
	}

	adp.set = func(k []byte, v []byte) (int64, error) {
		eles := make([][]byte, 0)
		for i := 0; i < 3; i++ {
			e := []byte(hack.String(v) + fmt.Sprintf("_%d", i))
			eles = append(eles, e)
		}

		if n, err := db.LPush(k, eles...); err != nil {
			return 0, err
		} else {
			return n, nil
		}
	}

	adp.exists = func(k []byte) (int64, error) {
		if llen, err := db.LLen(k); err != nil || llen <= 0 {
			return 0, err
		} else {
			return 1, nil
		}
	}

	adp.del = db.LClear
	adp.expire = db.LExpire
	adp.expireAt = db.LExpireAt
	adp.ttl = db.LTTL

	return adp
}

func zrangeGeneric(c *client, reverse bool) error {
	args := c.args
	if len(args) < 3 {
		return ErrCmdParams
	}

	key := args[0]

	start, stop, err := zparseRange(c, args[1], args[2])
	if err != nil {
		return ErrValue
	}

	args = args[3:]
	var withScores bool = false

	if len(args) > 0 {
		if len(args) != 1 {
			return ErrCmdParams
		}
		if strings.ToLower(hack.String(args[0])) == "withscores" {
			withScores = true
		} else {
			return ErrSyntax
		}
	}

	if datas, err := c.db.ZRangeGeneric(key, start, stop, reverse); err != nil {
		return err
	} else {
		c.resp.writeScorePairArray(datas, withScores)
	}
	return nil
}

func (d *jsonBinaryDecoder) decodeOpaque(data []byte) interface{} {
	if d.isDataShort(data, 1) {
		return nil
	}

	tp := data[0]
	data = data[1:]

	l, n := d.decodeVariableLength(data)

	if d.isDataShort(data, int(l)+n) {
		return nil
	}

	data = data[n : int(l)+n]

	switch tp {
	case MYSQL_TYPE_NEWDECIMAL:
		return d.decodeDecimal(data)
	case MYSQL_TYPE_TIME:
		return d.decodeTime(data)
	case MYSQL_TYPE_DATE, MYSQL_TYPE_DATETIME, MYSQL_TYPE_TIMESTAMP:
		return d.decodeDateTime(data)
	default:
		return hack.String(data)
	}

	return nil
}