// MinimumCoveringPeers returns a set of peers that minimizes overlap. This is similar to the Set Covering Problem and is NP-hard.
// This is a greedy algorithm. While the keyspace is not entirely covered, scan through all peers and pick the peer that will add the most to the set while still having the start in the selected set.
// TODO(wiz): Make this more optimal.
// TODO(wiz): achieve n-redundancy
func (s *Server) MinimumCoveringPeers() []*Conn {
	usedPeers := make(map[string]bool)
	var peers []*Conn
	var keyspace *protocol.Keyspace
	for i := 0; i < len(s.Peers) && !keyspace.Maxed(); i++ {
		var bestPeer *Conn
		var increase uint64
		// By definition, ranging through peer map will go in random order.
	Peers:
		for id, conn := range s.Peers {
			if conn == nil || conn.Peer == nil || usedPeers[id] {
				continue
			}
			peer := conn.Peer
			if keyspace == nil {
				peers = append(peers, conn)
				keyspace = peer.Keyspace
				break Peers
			}
			incr := keySpaceIncrease(keyspace, peer.Keyspace)
			if incr > increase {
				increase = incr
				bestPeer = conn
			}
		}
		if bestPeer != nil {
			peers = append(peers, bestPeer)
			keyspace = keyspace.Union(bestPeer.Peer.Keyspace)
			usedPeers[bestPeer.Peer.Id] = true
			// break?
		}
	}
	return peers
}

// keySpaceIncrease calculates the increase in keyspace if b was to be unioned.
func keySpaceIncrease(a, b *protocol.Keyspace) uint64 {
	unionMag := a.Union(b).Mag()
	aMag := a.Mag()
	if unionMag > aMag {
		return unionMag - aMag
	}
	return 0
}

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

// Bloom returns a ScalableBloomFilter containing all the triples the current node has in the optional keyspace.
func (ts *TripleStore) Bloom(keyspace *protocol.Keyspace) (*boom.ScalableBloomFilter, error) {
	filter := boom.NewDefaultScalableBloomFilter(BloomFalsePositiveRate)

	results, errs := ts.EachTripleBatch(DefaultTripleBatchSize)
	for triples := range results {
		for _, triple := range triples {
			if keyspace != nil {
				hash := murmur3.Sum64([]byte(triple.Subj))
				if !keyspace.Includes(hash) {
					continue
				}
			}
			data, err := triple.Marshal()
			if err != nil {
				return nil, err
			}
			filter.Add(data)
		}
	}
	for err := range errs {
		return nil, err
	}
	return filter, nil
}