// Broadcast sends a message to all peers with that have the hash in their keyspace.
func (s *Server) Broadcast(hash *uint64, msg *protocol.Message) error {
	alreadySentTo := make(map[uint64]bool)
	if msg.Gossip {
		for _, to := range msg.SentTo {
			alreadySentTo[to] = true
		}
	}
	sentTo := []uint64{murmur3.Sum64([]byte(s.LocalPeer().Id))}
	var toPeers []*Conn
	for _, peer := range s.Peers {
		peerHash := murmur3.Sum64([]byte(peer.Peer.Id))
		if (hash == nil || peer.Peer.GetKeyspace().Includes(*hash)) && !alreadySentTo[peerHash] {
			sentTo = append(sentTo, peerHash)
			toPeers = append(toPeers, peer)
		}
	}
	if msg.Gossip {
		msg.SentTo = append(msg.SentTo, sentTo...)
	}
	for _, peer := range toPeers {
		s.Printf("Broadcasting to %s", peer.Peer.Id)
		if err := peer.Send(msg); err != nil {
			return err
		}
	}
	return nil
}

func (ds *InMemDS) Create(parent, id []byte, inode uint64, name string, attr *pb.Attr, isdir bool) (string, *pb.Attr, error) {
	ds.Lock()
	defer ds.Unlock()
	p := murmur3.Sum64(parent)
	if _, exists := ds.nodes[p].entries[name]; exists {
		return "", &pb.Attr{}, nil
	}
	entry := &Entry{
		path:   name,
		inode:  inode,
		isdir:  isdir,
		attr:   attr,
		xattrs: make(map[string][]byte),
		blocks: 0,
	}
	if isdir {
		entry.entries = make(map[string]uint64)
		entry.ientries = make(map[uint64]string)
	}
	i := murmur3.Sum64(id)
	ds.nodes[i] = entry
	ds.nodes[p].entries[name] = i
	ds.nodes[p].ientries[i] = name
	atomic.AddUint64(&ds.nodes[p].nodeCount, 1)
	return name, attr, nil
}

func (s *Store) getShard(key string) *_shard {
	s.mutex.RLock()
	num := int(murmur3.Sum64([]byte(key))>>1) % s.shardsCount
	shard := s.shards[num]
	s.mutex.RUnlock()
	return shard
}

// LocalKeyspace returns the keyspace that the local node represents.
func (s *Server) LocalKeyspace() *protocol.Keyspace {
	center := murmur3.Sum64([]byte(s.LocalID()))
	return &protocol.Keyspace{
		Start: center - math.MaxUint64/4,
		End:   center + math.MaxUint64/4,
	}
}

func (ds *InMemDS) SetAttr(id []byte, attr *pb.Attr, v uint32) (*pb.Attr, error) {
	ds.Lock()
	defer ds.Unlock()
	valid := fuse.SetattrValid(v)
	if entry, ok := ds.nodes[murmur3.Sum64(id)]; ok {
		if valid.Mode() {
			entry.attr.Mode = attr.Mode
		}
		if valid.Size() {
			if attr.Size == 0 {
				entry.blocks = 0
				entry.lastblock = 0
			}
			entry.attr.Size = attr.Size
		}
		if valid.Mtime() {
			entry.attr.Mtime = attr.Mtime
		}
		if valid.Atime() {
			entry.attr.Atime = attr.Atime
		}
		if valid.Uid() {
			entry.attr.Uid = attr.Uid
		}
		if valid.Gid() {
			entry.attr.Gid = attr.Gid
		}
		return entry.attr, nil
	}
	return &pb.Attr{}, nil
}

func NewInMemDS() *InMemDS {
	ds := &InMemDS{
		nodes: make(map[uint64]*Entry),
	}
	n := &Entry{
		path:     "/",
		inode:    1,
		isdir:    true,
		entries:  make(map[string]uint64),
		ientries: make(map[uint64]string),
	}
	ts := time.Now().Unix()
	n.attr = &pb.Attr{
		Inode:  n.inode,
		Atime:  ts,
		Mtime:  ts,
		Ctime:  ts,
		Crtime: ts,
		Mode:   uint32(os.ModeDir | 0775),
		Uid:    1001, // TODO: need to config default user/group id
		Gid:    1001,
	}
	ds.nodes[murmur3.Sum64(GetID([]byte("1"), n.attr.Inode, 0))] = n
	return ds
}

// signAndInsertTriples signs a set of triples with the server's key and then inserts them into the graph.
func (s *server) signAndInsertTriples(triples []*protocol.Triple, key *crypto.PrivateKey) error {
	hashes := make(map[uint64][]*protocol.Triple)
	unix := time.Now().Unix()
	for _, triple := range triples {
		if err := key.SignTriple(triple); err != nil {
			return err
		}
		triple.Created = unix
		hash := murmur3.Sum64([]byte(triple.Subj))
		hashes[hash] = append(hashes[hash], triple)
	}

	for hash, triples := range hashes {
		msg := &protocol.Message{
			Message: &protocol.Message_InsertTriples{
				InsertTriples: &protocol.InsertTriples{
					Triples: triples,
				}},
			Gossip: true,
		}
		currentKeyspace := s.network.LocalPeer().Keyspace.Includes(hash)
		if err := s.network.Broadcast(&hash, msg); currentKeyspace && err == network.ErrNoRecipients {
		} else if err != nil {
			return err
		}
		if currentKeyspace {
			s.ts.Insert(triples)
		}
	}
	return nil
}

func verifyShardDist(t *testing.T, client *ConsistentHashRes, nShard int, n int) {
	sdMax := float64(6)
	result := make(map[string]int)

	fmt.Println("========================")
	for i := 0; i < n; i++ {
		id := murmur3.Sum64(uuid.NewV4().Bytes())
		info, ok := client.Get(fmt.Sprintf("%d", id))
		assert.True(t, ok, "should get shard info")
		if _, b := result[info]; b == true {
			result[info]++
		} else {
			result[info] = 1
		}
	}

	avg := float64(100.0) / float64(nShard)
	var sum float64
	for key, val := range result {
		fmt.Printf("%s, count = %d\n", key, val)
		sum += math.Pow(((100.0 * float64(val) / float64(n)) - avg), 2)
	}

	sd := math.Sqrt(sum / float64(nShard))

	fmt.Printf("average: %.3f%% standard deviation: %.3f%%\n", avg, sd)
	if sd > sdMax {
		assert.Fail(t, fmt.Sprintf("standard deviation is too high %v", sd))
	}
}

func (ds *InMemDS) GetAttr(id []byte) (*pb.Attr, error) {
	ds.RLock()
	defer ds.RUnlock()
	if entry, ok := ds.nodes[murmur3.Sum64(id)]; ok {
		return entry.attr, nil
	}
	return &pb.Attr{}, nil
}

func (ds *InMemDS) Getxattr(id []byte, name string) (*pb.GetxattrResponse, error) {
	ds.RLock()
	defer ds.RUnlock()
	if xattr, ok := ds.nodes[murmur3.Sum64(id)].xattrs[name]; ok {
		return &pb.GetxattrResponse{Xattr: xattr}, nil
	}
	return &pb.GetxattrResponse{}, nil
}

func (ds *InMemDS) Removexattr(id []byte, name string) (*pb.RemovexattrResponse, error) {
	ds.Lock()
	defer ds.Unlock()
	if entry, ok := ds.nodes[murmur3.Sum64(id)]; ok {
		delete(entry.xattrs, name)
	}
	return &pb.RemovexattrResponse{}, nil
}

func (ds *InMemDS) Setxattr(id []byte, name string, value []byte) (*pb.SetxattrResponse, error) {
	ds.Lock()
	defer ds.Unlock()
	if entry, ok := ds.nodes[murmur3.Sum64(id)]; ok {
		entry.xattrs[name] = value
	}
	return &pb.SetxattrResponse{}, nil
}

// subjInKeyspace appends an increasing number to the prefix until it finds a
// string that will hash into the given keyspace.
func subjInKeyspace(keyspace *protocol.Keyspace, prefix string) string {
	subj := prefix
	i := 1
	for !keyspace.Includes(murmur3.Sum64([]byte(subj))) {
		subj = prefix + strconv.Itoa(i)
		i++
	}
	return subj
}

func (ds *InMemDS) Rename(oldParent, newParent []byte, oldName, newName string) (*pb.RenameResponse, error) {
	ds.Lock()
	defer ds.Unlock()
	p := murmur3.Sum64(oldParent)
	if id, ok := ds.nodes[p].entries[oldName]; ok {
		// remove old
		delete(ds.nodes[p].entries, oldName)
		delete(ds.nodes[p].ientries, id)
		atomic.AddUint64(&ds.nodes[p].nodeCount, ^uint64(0)) // -1
		// add new
		ds.nodes[id].path = newName
		np := murmur3.Sum64(newParent)
		ds.nodes[np].entries[newName] = id
		ds.nodes[np].ientries[id] = newName
		atomic.AddUint64(&ds.nodes[np].nodeCount, 1)
	}
	return &pb.RenameResponse{}, nil
}

func (ds *InMemDS) Lookup(parent []byte, name string) (string, *pb.Attr, error) {
	ds.RLock()
	defer ds.RUnlock()
	id, ok := ds.nodes[murmur3.Sum64(parent)].entries[name]
	if !ok {
		return "", &pb.Attr{}, nil
	}
	entry := ds.nodes[id]
	return entry.path, entry.attr, nil
}

func (s *Server) LocalPeer() *protocol.Peer {
	id := fmt.Sprintf("%s:%d", s.IP, s.Port)
	center := murmur3.Sum64([]byte(id))
	keyspace := &protocol.Keyspace{
		Start: center - math.MaxUint64/4,
		End:   center + math.MaxUint64/4,
	}
	return &protocol.Peer{
		Id:       id,
		Keyspace: keyspace,
	}
}

func main() {
	nodeMap = make(map[uint64]string)
	FirstNode := "http://localhost:3000/"
	SecondNode := "http://localhost:3001/"
	ThirdNode := "http://localhost:3002/"

	keys = append(keys, murmur3.Sum64([]byte(FirstNode)))
	keys = append(keys, murmur3.Sum64([]byte(SecondNode)))
	keys = append(keys, murmur3.Sum64([]byte(ThirdNode)))

	keys.Sort()
	fmt.Println("Keys array is : ", keys)

	for _, element := range keys {
		switch element {

		case murmur3.Sum64([]byte(FirstNode)):
			nodeMap[element] = FirstNode
		case murmur3.Sum64([]byte(SecondNode)):
			nodeMap[element] = SecondNode
		case murmur3.Sum64([]byte(ThirdNode)):
			nodeMap[element] = ThirdNode
		}

	}

	mux := routes.New()
	mux.Put("/keys/:keyID/:value", PutValue)
	mux.Get("/keys/:keyID", RetrieveValue)

	http.Handle("/", mux)
	http.ListenAndServe(":8080", nil)

}

func (self *Node) Get(key *string, data *[]byte) error {
	var found bool
	self.RLock()
	if *data, found = self.data[*key]; found {
		self.RUnlock()
		return nil
	}
	self.RUnlock()
	closestpeer := self.Peers.ClosestPeer(murmur3.Sum64([]byte(*key)) >> 16)
	log.Println(self.Key(), closestpeer.Key())
	var err error
	*data, err = self.Peers.Get(closestpeer, *key)
	return err
}

func (ds *InMemDS) Remove(parent []byte, name string) (int32, error) {
	ds.Lock()
	defer ds.Unlock()
	p := murmur3.Sum64(parent)
	id, ok := ds.nodes[p].entries[name]
	if !ok {
		return 1, nil
	}
	delete(ds.nodes, id)
	delete(ds.nodes[p].entries, name)
	delete(ds.nodes[p].ientries, id)
	atomic.AddUint64(&ds.nodes[p].nodeCount, ^uint64(0)) // -1
	return 0, nil
}

func (ds *InMemDS) Listxattr(id []byte) (*pb.ListxattrResponse, error) {
	ds.RLock()
	defer ds.RUnlock()
	resp := &pb.ListxattrResponse{}
	if entry, ok := ds.nodes[murmur3.Sum64(id)]; ok {
		names := ""
		for name := range entry.xattrs {
			names += name
			names += "\x00"
		}
		resp.Xattr = []byte(names)
	}
	return resp, nil
}