// Copyright 2014 The go-ethereum Authors
// This file is part of the go-ethereum library.
//
// The go-ethereum library is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// The go-ethereum library is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Lesser General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with the go-ethereum library. If not, see <http://www.gnu.org/licenses/>.

package whisper5

import (
	"crypto/ecdsa"
	"sync"
	"time"

	"github.com/ethereum/go-ethereum/common"
	"github.com/ethereum/go-ethereum/crypto"
	"github.com/ethereum/go-ethereum/logger"
	"github.com/ethereum/go-ethereum/logger/glog"
	"github.com/ethereum/go-ethereum/p2p"
	"github.com/ethereum/go-ethereum/rpc"

	set "gopkg.in/fatih/set.v0"
)

// Whisper represents a dark communication interface through the Ethereum
// network, using its very own P2P communication layer.
type Whisper struct {
	protocol p2p.Protocol
	filters  *Filters

	privateKeys map[string]*ecdsa.PrivateKey
	//topicKeys   map[TopicType][]byte // todo: move to the filter. this is not suitable because of possible collisions

	envelopes   map[common.Hash]*Envelope        // Pool of messages currently tracked by this node
	messages    map[common.Hash]*ReceivedMessage // Pool of successfully decrypted messages
	expirations map[uint32]*set.SetNonTS         // Message expiration pool (TODO: something lighter)
	poolMu      sync.RWMutex                     // Mutex to sync the message and expiration pools

	peers  map[*peer]struct{} // Set of currently active peers
	peerMu sync.RWMutex       // Mutex to sync the active peer set

	quit chan struct{}
}

// New creates a Whisper client ready to communicate through the Ethereum P2P network.
func NewWhisper() *Whisper {
	whisper := &Whisper{
		//filters:     NewFilters(),
		privateKeys: make(map[string]*ecdsa.PrivateKey),
		//topicKeys:   make(map[TopicType][]byte),
		envelopes:   make(map[common.Hash]*Envelope),
		messages:    make(map[common.Hash]*ReceivedMessage),
		expirations: make(map[uint32]*set.SetNonTS),
		peers:       make(map[*peer]struct{}),
		quit:        make(chan struct{}),
	}
	whisper.filters = NewFilters(whisper)
	whisper.filters.Start()

	// p2p whisper sub protocol handler
	whisper.protocol = p2p.Protocol{
		Name:    protocolName,
		Version: uint(protocolVersion),
		Length:  2,
		Run:     whisper.handlePeer,
	}

	return whisper
}

// APIs returns the RPC descriptors the Whisper implementation offers
func (s *Whisper) APIs() []rpc.API {
	return []rpc.API{
		{
			Namespace: protocolName,
			Version:   protocolVersionStr,
			Service:   NewPublicWhisperAPI(s),
			Public:    true,
		},
	}
}

// Protocols returns the whisper sub-protocols ran by this particular client.
func (self *Whisper) Protocols() []p2p.Protocol {
	return []p2p.Protocol{self.protocol}
}

// Version returns the whisper sub-protocols version number.
func (self *Whisper) Version() uint {
	return self.protocol.Version
}

// todo: review. maybe we need to delete these identity-related functions, since the key moved to Filter
// NewIdentity generates a new cryptographic identity for the client, and injects
// it into the known identities for message decryption.
func (self *Whisper) NewIdentity() *ecdsa.PrivateKey {
	key, err := crypto.GenerateKey()
	if err != nil {
		panic(err)
	}
	self.privateKeys[string(crypto.FromECDSAPub(&key.PublicKey))] = key
	return key
}

// HasIdentity checks if the the whisper node is configured with the private key
// of the specified public pair.
func (self *Whisper) HasIdentity(key *ecdsa.PublicKey) bool {
	return self.privateKeys[string(crypto.FromECDSAPub(key))] != nil
}

// GetIdentity retrieves the private key of the specified public identity.
func (self *Whisper) GetIdentity(key *ecdsa.PublicKey) *ecdsa.PrivateKey {
	return self.privateKeys[string(crypto.FromECDSAPub(key))]
}

// Watch installs a new message handler to run in case a matching packet arrives
// from the whisper network.
func (self *Whisper) Watch(f *Filter) int {
	return self.filters.Install(f)
}

// Unwatch removes an installed message handler.
func (self *Whisper) Unwatch(id int) {
	self.filters.Uninstall(id)
}

// Send injects a message into the whisper send queue, to be distributed in the
// network in the coming cycles.
func (self *Whisper) Send(envelope *Envelope) error {
	return self.add(envelope)
}

// Start implements node.Service, starting the background data propagation thread
// of the Whisper protocol.
func (self *Whisper) Start(*p2p.Server) error {
	glog.V(logger.Info).Infoln("Whisper started")
	go self.update()
	return nil
}

// Stop implements node.Service, stopping the background data propagation thread
// of the Whisper protocol.
func (self *Whisper) Stop() error {
	close(self.quit)
	glog.V(logger.Info).Infoln("Whisper stopped")
	return nil
}

// handlePeer is called by the underlying P2P layer when the whisper sub-protocol
// connection is negotiated.
func (self *Whisper) handlePeer(peer *p2p.Peer, rw p2p.MsgReadWriter) error {
	// Create the new peer and start tracking it
	whisperPeer := newPeer(self, peer, rw)

	self.peerMu.Lock()
	self.peers[whisperPeer] = struct{}{}
	self.peerMu.Unlock()

	defer func() {
		self.peerMu.Lock()
		delete(self.peers, whisperPeer)
		self.peerMu.Unlock()
	}()

	// Run the peer handshake and state updates
	if err := whisperPeer.handshake(); err != nil {
		return err
	}
	whisperPeer.start()
	defer whisperPeer.stop()

	// Read and process inbound messages directly to merge into client-global state
	for {
		// Fetch the next packet and decode the contained envelopes
		packet, err := rw.ReadMsg()
		if err != nil {
			return err
		}
		var envelopes []*Envelope
		if err := packet.Decode(&envelopes); err != nil {
			glog.V(logger.Info).Infof("%v: failed to decode envelope: %v", peer, err)
			continue
		}
		// Inject all envelopes into the internal pool
		for _, envelope := range envelopes {
			if err := self.add(envelope); err != nil {
				// TODO Punish peer here. Invalid envelope.
				glog.V(logger.Debug).Infof("%v: failed to pool envelope: %v", peer, err)
			}
			whisperPeer.mark(envelope)
		}
	}
}

// add inserts a new envelope into the message pool to be distributed within the
// whisper network. It also inserts the envelope into the expiration pool at the
// appropriate time-stamp.
func (self *Whisper) add(envelope *Envelope) error {
	self.poolMu.Lock()
	defer self.poolMu.Unlock()

	// short circuit when a received envelope has already expired
	if envelope.Expiry < uint32(time.Now().Unix()) {
		return nil
	}

	// Insert the message into the tracked pool
	hash := envelope.Hash()
	if _, ok := self.envelopes[hash]; ok {
		glog.V(logger.Detail).Infof("whisper envelope already cached: %x\n", envelope)
		return nil
	}
	self.envelopes[hash] = envelope

	// Insert the message into the expiration pool for later removal
	if self.expirations[envelope.Expiry] == nil {
		self.expirations[envelope.Expiry] = set.NewNonTS()
	}
	if !self.expirations[envelope.Expiry].Has(hash) {
		self.expirations[envelope.Expiry].Add(hash)

		// Notify the local node of a message arrival
		go self.postEvent(envelope)
	}
	glog.V(logger.Detail).Infof("cached whisper envelope %x\n", envelope)
	return nil
}

// postEvent delivers the message to the watchers.
func (self *Whisper) postEvent(envelope *Envelope) {
	self.filters.Notify(envelope)
}

// update loops until the lifetime of the whisper node, updating its internal
// state by expiring stale messages from the pool.
func (self *Whisper) update() {
	// Start a ticker to check for expirations
	expire := time.NewTicker(expirationCycle)

	// Repeat updates until termination is requested
	for {
		select {
		case <-expire.C:
			self.expire()

		case <-self.quit:
			return
		}
	}
}

// expire iterates over all the expiration timestamps, removing all stale
// messages from the pools.
func (self *Whisper) expire() {
	self.poolMu.Lock()
	defer self.poolMu.Unlock()

	now := uint32(time.Now().Unix())
	for then, hashSet := range self.expirations {
		// Short circuit if a future time
		if then > now {
			continue
		}
		// Dump all expired messages and remove timestamp
		hashSet.Each(func(v interface{}) bool {
			delete(self.envelopes, v.(common.Hash))
			delete(self.messages, v.(common.Hash))
			return true
		})
		self.expirations[then].Clear()
	}
}

// envelopes retrieves all the messages currently pooled by the node.
func (self *Whisper) Envelopes() []*Envelope {
	self.poolMu.RLock()
	defer self.poolMu.RUnlock()

	all := make([]*Envelope, 0, len(self.envelopes))
	for _, envelope := range self.envelopes {
		all = append(all, envelope)
	}
	return all
}

// Messages retrieves all the currently pooled messages matching a filter id.
func (self *Whisper) Messages(id int) []*ReceivedMessage {
	self.poolMu.RLock()
	defer self.poolMu.RUnlock()

	result := make([]*ReceivedMessage, 0)
	if filter := self.filters.Get(id); filter != nil {
		for _, msg := range self.messages {
			if filter.MatchMessage(msg) {
				result = append(result, msg)
			}
		}
	}
	return result
}

func (self *Whisper) addDecryptedMessage(msg *ReceivedMessage) {
	self.poolMu.Lock()
	defer self.poolMu.Unlock()

	self.messages[msg.EnvelopeHash] = msg
}