// hanleTCPConn handle a long live tcp connection.
func handleTCPConn(conn net.Conn, rc chan *bufio.Reader) {
	addr := conn.RemoteAddr().String()
	log.Debug("<%s> handleTcpConn routine start", addr)
	rd := newBufioReader(rc, conn)
	if args, err := parseCmd(rd); err == nil {
		// return buffer bufio.Reader
		putBufioReader(rc, rd)
		switch args[0] {
		case "sub":
			SubscribeTCPHandle(conn, args[1:])
			break
		default:
			conn.Write(ParamReply)
			log.Warn("<%s> unknown cmd \"%s\"", addr, args[0])
			break
		}
	} else {
		// return buffer bufio.Reader
		putBufioReader(rc, rd)
		log.Error("<%s> parseCmd() error(%v)", addr, err)
	}
	// close the connection
	if err := conn.Close(); err != nil {
		log.Error("<%s> conn.Close() error(%v)", addr, err)
	}
	log.Debug("<%s> handleTcpConn routine stop", addr)
	return
}

// HandleWrite start a goroutine get msg from chan, then send to the conn.
func (c *Connection) HandleWrite(key string) {
	go func() {
		var (
			n   int
			err error
		)
		log.Debug("user_key: \"%s\" HandleWrite goroutine start", key)
		for {
			msg, ok := <-c.Buf
			if !ok {
				log.Debug("user_key: \"%s\" HandleWrite goroutine stop", key)
				return
			}
			if c.Proto == WebsocketProto {
				// raw
				n, err = c.Conn.Write(msg)
			} else if c.Proto == TCPProto {
				// redis protocol
				msg = []byte(fmt.Sprintf("$%d\r\n%s\r\n", len(msg), string(msg)))
				n, err = c.Conn.Write(msg)
			} else {
				log.Error("unknown connection protocol: %d", c.Proto)
				panic(ErrConnProto)
			}
			// update stat
			if err != nil {
				log.Error("user_key: \"%s\" conn.Write() error(%v)", key, err)
				MsgStat.IncrFailed(1)
			} else {
				log.Debug("user_key: \"%s\" write \r\n========%s(%d)========", key, string(msg), n)
				MsgStat.IncrSucceed(1)
			}
		}
	}()
}

// Migrate migrate portion of connections which don't belong to this comet.
func (l *ChannelList) Migrate(nw map[string]int) (err error) {
	migrate := false
	// check new/update node
	for k, v := range nw {
		weight, ok := nodeWeightMap[k]
		// not found or weight change
		if !ok || weight != v {
			migrate = true
			break
		}
	}
	// check del node
	if !migrate {
		for k, _ := range nodeWeightMap {
			// node deleted
			if _, ok := nw[k]; !ok {
				migrate = true
				break
			}
		}
	}
	if !migrate {
		return
	}
	// init ketama
	ring := ketama.NewRing(ketama.Base)
	for node, weight := range nw {
		ring.AddNode(node, weight)
	}
	ring.Bake()
	// atomic update
	nodeWeightMap = nw
	CometRing = ring
	// get all the channel lock
	channels := []Channel{}
	for i, c := range l.Channels {
		c.Lock()
		for k, v := range c.Data {
			hn := ring.Hash(k)
			if hn != Conf.ZookeeperCometNode {
				channels = append(channels, v)
				delete(c.Data, k)
				log.Debug("migrate delete channel key \"%s\"", k)
			}
		}
		c.Unlock()
		log.Debug("migrate channel bucket:%d finished", i)
	}
	// close all the migrate channels
	log.Info("close all the migrate channels")
	for _, channel := range channels {
		if err := channel.Close(); err != nil {
			log.Error("channel.Close() error(%v)", err)
			continue
		}
	}
	log.Info("close all the migrate channels finished")
	return
}

func tcpListen(bind string) {
	addr, err := net.ResolveTCPAddr("tcp", bind)
	if err != nil {
		log.Error("net.ResolveTCPAddr(\"tcp\"), %s) error(%v)", bind, err)
		panic(err)
	}
	l, err := net.ListenTCP("tcp", addr)
	if err != nil {
		log.Error("net.ListenTCP(\"tcp4\", \"%s\") error(%v)", bind, err)
		panic(err)
	}
	// free the listener resource
	defer func() {
		log.Info("tcp addr: \"%s\" close", bind)
		if err := l.Close(); err != nil {
			log.Error("listener.Close() error(%v)", err)
		}
	}()
	// init reader buffer instance
	rb := newtcpBufCache()
	for {
		log.Debug("start accept")
		conn, err := l.AcceptTCP()
		if err != nil {
			log.Error("listener.AcceptTCP() error(%v)", err)
			continue
		}
		if err = conn.SetKeepAlive(Conf.TCPKeepalive); err != nil {
			log.Error("conn.SetKeepAlive() error(%v)", err)
			conn.Close()
			continue
		}
		if err = conn.SetReadBuffer(Conf.RcvbufSize); err != nil {
			log.Error("conn.SetReadBuffer(%d) error(%v)", Conf.RcvbufSize, err)
			conn.Close()
			continue
		}
		if err = conn.SetWriteBuffer(Conf.SndbufSize); err != nil {
			log.Error("conn.SetWriteBuffer(%d) error(%v)", Conf.SndbufSize, err)
			conn.Close()
			continue
		}
		// first packet must sent by client in specified seconds
		if err = conn.SetReadDeadline(time.Now().Add(fitstPacketTimedoutSec)); err != nil {
			log.Error("conn.SetReadDeadLine() error(%v)", err)
			conn.Close()
			continue
		}
		rc := rb.Get()
		// one connection one routine
		go handleTCPConn(conn, rc)
		log.Debug("accept finished")
	}
}

// validate check the key is belong to this comet.
func (l *ChannelList) validate(key string) error {
	if len(nodeWeightMap) == 0 {
		log.Debug("no node found")
		return ErrChannelKey
	}
	node := CometRing.Hash(key)
	log.Debug("match node:%s hash node:%s", Conf.ZookeeperCometNode, node)
	if Conf.ZookeeperCometNode != node {
		log.Warn("user_key:\"%s\" node:%s not match this node:%s", key, node, Conf.ZookeeperCometNode)
		return ErrChannelKey
	}
	return nil
}

// GetPrivate rpc interface get user private message.
func (r *MessageRPC) GetPrivate(m *myrpc.MessageGetPrivateArgs, rw *myrpc.MessageGetResp) error {
	log.Debug("messageRPC.GetPrivate key:\"%s\" mid:\"%d\"", m.Key, m.MsgId)
	if m == nil || m.Key == "" || m.MsgId < 0 {
		return myrpc.ErrParam
	}
	msgs, err := UseStorage.GetPrivate(m.Key, m.MsgId)
	if err != nil {
		log.Error("UseStorage.GetPrivate(\"%s\", %d) error(%v)", m.Key, m.MsgId, err)
		return err
	}
	rw.Msgs = msgs
	log.Debug("UserStorage.GetPrivate(\"%s\", %d) ok", m.Key, m.MsgId)
	return nil
}

// RegisterTmp create a ephemeral node, and watch it, if node droped then send a SIGQUIT to self.
func RegisterTemp(conn *zk.Conn, fpath string, data []byte) error {
	tpath, err := conn.Create(path.Join(fpath)+"/", data, zk.FlagEphemeral|zk.FlagSequence, zk.WorldACL(zk.PermAll))
	if err != nil {
		log.Error("conn.Create(\"%s\", \"%s\", zk.FlagEphemeral|zk.FlagSequence) error(%v)", fpath, string(data), err)
		return err
	}
	log.Debug("create a zookeeper node:%s", tpath)
	// watch self
	go func() {
		for {
			log.Info("zk path: \"%s\" set a watch", tpath)
			exist, _, watch, err := conn.ExistsW(tpath)
			if err != nil {
				log.Error("zk.ExistsW(\"%s\") error(%v)", tpath, err)
				log.Warn("zk path: \"%s\" set watch failed, kill itself", tpath)
				killSelf()
				return
			}
			if !exist {
				log.Warn("zk path: \"%s\" not exist, kill itself", tpath)
				killSelf()
				return
			}
			event := <-watch
			log.Info("zk path: \"%s\" receive a event %v", tpath, event)
		}
	}()
	return nil
}

// bucket return a channelBucket use murmurhash3.
func (l *ChannelList) Bucket(key string) *ChannelBucket {
	h := hash.NewMurmur3C()
	h.Write([]byte(key))
	idx := uint(h.Sum32()) & uint(Conf.ChannelBucket-1)
	log.Debug("user_key:\"%s\" hit channel bucket index:%d", key, idx)
	return l.Channels[idx]
}

// newTCPBufCache return a new tcpBuf cache.
func newtcpBufCache() *tcpBufCache {
	inst := make([]chan *bufio.Reader, 0, Conf.BufioInstance)
	log.Debug("create %d read buffer instance", Conf.BufioInstance)
	for i := 0; i < Conf.BufioInstance; i++ {
		inst = append(inst, make(chan *bufio.Reader, Conf.BufioNum))
	}
	return &tcpBufCache{instance: inst, round: 0}
}

// getConn get the connection of matching with key using ketama hashing.
func (s *RedisStorage) getConn(key string) redis.Conn {
	if len(s.pool) == 0 {
		return nil
	}
	node := s.ring.Hash(key)
	log.Debug("user_key: \"%s\" hit redis node: \"%s\"", key, node)
	return s.getConnByNode(node)
}

// GetComet get the node infomation under the node.
func GetComet(key string) *CometNodeInfo {
	if cometRing == nil || len(cometNodeInfoMap) == 0 {
		return nil
	}
	node := cometRing.Hash(key)
	log.Debug("cometHash hits \"%s\"", node)
	return cometNodeInfoMap[node]
}

// SavePrivate rpc interface save user private message.
func (r *MessageRPC) SavePrivate(m *myrpc.MessageSavePrivateArgs, ret *int) error {
	if m == nil || m.Msg == nil || m.MsgId < 0 {
		return myrpc.ErrParam
	}
	if err := UseStorage.SavePrivate(m.Key, m.Msg, m.MsgId, m.Expire); err != nil {
		log.Error("UseStorage.SavePrivate(\"%s\", \"%s\", %d, %d) error(%v)", m.Key, string(m.Msg), m.MsgId, m.Expire, err)
		return err
	}
	log.Debug("UseStorage.SavePrivate(\"%s\", \"%s\", %d, %d) ok", m.Key, string(m.Msg), m.MsgId, m.Expire)
	return nil
}

// notify every Comet node to migrate
func notifyMigrate(conn *zk.Conn, migrateLockPath, znode, key string, update bool, nodeWeightMap map[string]int) (err error) {
	// try lock
	if _, err = conn.Create(migrateLockPath, []byte("1"), zk.FlagEphemeral, zk.WorldACL(zk.PermAll)); err != nil {
		log.Error("conn.Create(\"/gopush-migrate-lock\", \"1\", zk.FlagEphemeral) error(%v)", err)
		return
	}
	// call comet migrate rpc
	wg := &sync.WaitGroup{}
	wg.Add(len(cometNodeInfoMap))
	for node, nodeInfo := range cometNodeInfoMap {
		go func(n string, info *CometNodeInfo) {
			if info.Rpc == nil {
				log.Error("notify migrate failed, no rpc found, node:%s", n)
				wg.Done()
				return
			}
			r := info.Rpc.Get()
			if r == nil {
				log.Error("notify migrate failed, no rpc found, node:%s", n)
				wg.Done()
				return
			}
			reply := 0
			args := &CometMigrateArgs{Nodes: nodeWeightMap}
			if err = r.Call(CometServiceMigrate, args, &reply); err != nil {
				log.Error("rpc.Call(\"%s\") error(%v)", CometServiceMigrate, err)
				wg.Done()
				return
			}
			log.Debug("notify node:%s migrate succeed", n)
			wg.Done()
		}(node, nodeInfo)
	}
	wg.Wait()
	// update znode info
	if update {
		var data []byte
		data, err = json.Marshal(cometNodeInfoMap[key])
		if err != nil {
			log.Error("json.Marshal() node:%s error(%v)", key, err)
			return
		}
		if _, err = conn.Set(znode, data, -1); err != nil {
			log.Error("conn.Set(\"%s\",\"%s\",\"-1\") error(%v)", znode, string(data), err)
			return
		}
	}

	// release lock
	if err = conn.Delete(migrateLockPath, -1); err != nil {
		log.Error("conn.Delete(\"%s\") error(%v)", migrateLockPath, err)
	}
	return
}

// DelPrivate rpc interface delete user private message.
func (r *MessageRPC) DelPrivate(key string, ret *int) error {
	if key == "" {
		return myrpc.ErrParam
	}
	if err := UseStorage.DelPrivate(key); err != nil {
		log.Error("UserStorage.DelPrivate(\"%s\") error(%v)", key, err)
		return err
	}
	log.Debug("UserStorage.DelPrivate(\"%s\") ok", key)
	return nil
}

// watchMessageRoot watch the message root path.
func watchMessageRoot(conn *zk.Conn, fpath string, ch chan *MessageNodeEvent) error {
	for {
		nodes, watch, err := myzk.GetNodesW(conn, fpath)
		if err == myzk.ErrNodeNotExist {
			log.Warn("zk don't have node \"%s\", retry in %d second", fpath, waitNodeDelay)
			time.Sleep(waitNodeDelaySecond)
			continue
		} else if err == myzk.ErrNoChild {
			log.Warn("zk don't have any children in \"%s\", retry in %d second", fpath, waitNodeDelay)
			// all child died, kick all the nodes
			for _, client := range MessageRPC.Clients {
				log.Debug("node: \"%s\" send del node event", client.Addr)
				ch <- &MessageNodeEvent{Event: eventNodeDel, Key: &WeightRpc{Addr: client.Addr, Weight: client.Weight}}
			}
			time.Sleep(waitNodeDelaySecond)
			continue
		} else if err != nil {
			log.Error("getNodes error(%v), retry in %d second", err, waitNodeDelay)
			time.Sleep(waitNodeDelaySecond)
			continue
		}
		nodesMap := map[string]bool{}
		// handle new add nodes
		for _, node := range nodes {
			data, _, err := conn.Get(path.Join(fpath, node))
			if err != nil {
				log.Error("zk.Get(\"%s\") error(%v)", path.Join(fpath, node), err)
				continue
			}
			// parse message node info
			nodeInfo := &MessageNodeInfo{}
			if err := json.Unmarshal(data, nodeInfo); err != nil {
				log.Error("json.Unmarshal(\"%s\", nodeInfo) error(%v)", string(data), err)
				continue
			}
			for _, addr := range nodeInfo.Rpc {
				// if not exists in old map then trigger a add event
				if _, ok := MessageRPC.Clients[addr]; !ok {
					ch <- &MessageNodeEvent{Event: eventNodeAdd, Key: &WeightRpc{Addr: addr, Weight: nodeInfo.Weight}}
				}
				nodesMap[addr] = true
			}
		}
		// handle delete nodes
		for _, client := range MessageRPC.Clients {
			if _, ok := nodesMap[client.Addr]; !ok {
				ch <- &MessageNodeEvent{Event: eventNodeDel, Key: client}
			}
		}
		// blocking wait node changed
		event := <-watch
		log.Info("zk path: \"%s\" receive a event %v", fpath, event)
	}
}

// SavePrivates rpc interface save user private messages.
func (r *MessageRPC) SavePrivates(m *myrpc.MessageSavePrivatesArgs, rw *myrpc.MessageSavePrivatesResp) error {
	if m == nil || m.Msg == nil || m.MsgId < 0 {
		return myrpc.ErrParam
	}
	fkeys, err := UseStorage.SavePrivates(m.Keys, m.Msg, m.MsgId, m.Expire)
	if err != nil {
		log.Error("UseStorage.SavePrivates(\"%v\", \"%s\", %d, %d) error(%v)", m.Keys, string(m.Msg), m.MsgId, m.Expire, err)
	}
	rw.FKeys = fkeys
	log.Debug("UseStorage.SavePrivates(\"%v\", \"%s\", %d, %d) ok fkeys len(%d)", m.Keys, string(m.Msg), m.MsgId, m.Expire, len(fkeys))
	return nil
}

// registerCometNode get infomation of comet node
func registerCometNode(conn *zk.Conn, node, fpath string, retry, ping time.Duration, startPing bool) (info *CometNodeInfo, err error) {
	// get current node info from zookeeper
	fpath = path.Join(fpath, node)
	data, _, err := conn.Get(fpath)
	if err != nil {
		log.Error("zk.Get(\"%s\") error(%v)", fpath, err)
		return
	}
	info = &CometNodeInfo{}
	if err = json.Unmarshal(data, info); err != nil {
		log.Error("json.Unmarshal(\"%s\", nodeData) error(%v)", string(data), err)
		return
	}
	if len(info.RpcAddr) == 0 {
		log.Error("zk nodes: \"%s\" don't have rpc addr", fpath)
		err = ErrCometRPC
		return
	}
	// get old node info for finding the old rpc connection
	oldInfo := cometNodeInfoMap[node]
	// init comet rpc
	clients := make(map[string]*WeightRpc, len(info.RpcAddr))
	for _, addr := range info.RpcAddr {
		var (
			r *rpc.Client
		)
		if oldInfo != nil && oldInfo.Rpc != nil {
			if wr, ok := oldInfo.Rpc.Clients[addr]; ok && wr.Client != nil {
				// reuse the rpc connection must let old client = nil, avoid reclose rpc.
				oldInfo.Rpc.Clients[addr].Client = nil
				r = wr.Client
			}
		}
		if r == nil {
			if r, err = rpc.Dial("tcp", addr); err != nil {
				log.Error("rpc.Dial(\"%s\") error(%v)", addr, err)
				return
			}
			log.Debug("node:%s addr:%s rpc reconnect", node, addr)
		}
		clients[addr] = &WeightRpc{Weight: 1, Addr: addr, Client: r}
	}
	// comet rpc use rand load balance
	lb, err := NewRandLB(clients, cometService, retry, ping, startPing)
	if err != nil {
		log.Error("NewRandLR() error(%v)", err)
		return
	}
	info.Rpc = lb
	log.Info("zk path: \"%s\" register nodes: \"%s\"", fpath, node)
	return
}

// NewChannelList create a new channel bucket set.
func NewChannelList() *ChannelList {
	l := &ChannelList{Channels: []*ChannelBucket{}}
	// split hashmap to many bucket
	log.Debug("create %d ChannelBucket", Conf.ChannelBucket)
	for i := 0; i < Conf.ChannelBucket; i++ {
		c := &ChannelBucket{
			Data:  map[string]Channel{},
			mutex: &sync.Mutex{},
		}
		l.Channels = append(l.Channels, c)
	}
	return l
}

// retPWrite marshal the result and write to client(post).
func retPWrite(w http.ResponseWriter, r *http.Request, res map[string]interface{}, body *string, start time.Time) {
	data, err := json.Marshal(res)
	if err != nil {
		log.Error("json.Marshal(\"%v\") error(%v)", res, err)
		return
	}
	dataStr := string(data)
	if n, err := w.Write([]byte(dataStr)); err != nil {
		log.Error("w.Write(\"%s\") error(%v)", dataStr, err)
	} else {
		log.Debug("w.Write(\"%s\") write %d bytes", dataStr, n)
	}
	log.Info("req: \"%s\", post: \"%s\", res:\"%s\", ip:\"%s\", time:\"%fs\"", r.URL.String(), *body, dataStr, r.RemoteAddr, time.Now().Sub(start).Seconds())
}

// Connect connect to zookeeper, and start a goroutine log the event.
func Connect(addr []string, timeout time.Duration) (*zk.Conn, error) {
	conn, session, err := zk.Connect(addr, timeout)
	if err != nil {
		log.Error("zk.Connect(\"%v\", %d) error(%v)", addr, timeout, err)
		return nil, err
	}
	go func() {
		for {
			event := <-session
			log.Debug("zookeeper get a event: %s", event.State.String())
		}
	}()
	return conn, nil
}