transport.go

package transport

import (
	"bytes"
	"context"
	"crypto/ecdsa"
	"database/sql"
	"encoding/hex"
	"fmt"
	"sync"
	"time"

	"github.com/google/uuid"
	"github.com/pkg/errors"
	"go.uber.org/zap"

	"github.com/ethereum/go-ethereum/common"
	"github.com/status-im/status-go/eth-node/crypto"
	"github.com/status-im/status-go/eth-node/types"
)

var (
	// ErrNoMailservers returned if there is no configured mailservers that can be used.
	ErrNoMailservers = errors.New("no configured mailservers")
)

type transportKeysManager struct {
	waku types.Waku

	// Identity of the current user.
	privateKey *ecdsa.PrivateKey

	passToSymKeyMutex sync.RWMutex
	passToSymKeyCache map[string]string
}

func (m *transportKeysManager) AddOrGetKeyPair(priv *ecdsa.PrivateKey) (string, error) {
	// caching is handled in waku
	return m.waku.AddKeyPair(priv)
}

func (m *transportKeysManager) AddOrGetSymKeyFromPassword(password string) (string, error) {
	m.passToSymKeyMutex.Lock()
	defer m.passToSymKeyMutex.Unlock()

	if val, ok := m.passToSymKeyCache[password]; ok {
		return val, nil
	}

	id, err := m.waku.AddSymKeyFromPassword(password)
	if err != nil {
		return id, err
	}

	m.passToSymKeyCache[password] = id

	return id, nil
}

func (m *transportKeysManager) RawSymKey(id string) ([]byte, error) {
	return m.waku.GetSymKey(id)
}

type Option func(*Transport) error

// Transport is a transport based on Whisper service.
type Transport struct {
	waku        types.Waku
	api         types.PublicWakuAPI // only PublicWakuAPI implements logic to send messages
	keysManager *transportKeysManager
	filters     *FiltersManager
	logger      *zap.Logger
	cache       *ProcessedMessageIDsCache

	mailservers      []string
	envelopesMonitor *EnvelopesMonitor
	quit             chan struct{}
}

// NewTransport returns a new Transport.
// TODO: leaving a chat should verify that for a given public key
//       there are no other chats. It may happen that we leave a private chat
//       but still have a public chat for a given public key.
func NewTransport(
	waku types.Waku,
	privateKey *ecdsa.PrivateKey,
	db *sql.DB,
	sqlitePersistenceTableName string,
	mailservers []string,
	envelopesMonitorConfig *EnvelopesMonitorConfig,
	logger *zap.Logger,
	opts ...Option,
) (*Transport, error) {
	filtersManager, err := NewFiltersManager(newSQLitePersistence(db, sqlitePersistenceTableName), waku, privateKey, logger)
	if err != nil {
		return nil, err
	}

	var envelopesMonitor *EnvelopesMonitor
	if envelopesMonitorConfig != nil {
		envelopesMonitor = NewEnvelopesMonitor(waku, *envelopesMonitorConfig)
		envelopesMonitor.Start()
	}

	var api types.PublicWhisperAPI
	if waku != nil {
		api = waku.PublicWakuAPI()
	}
	t := &Transport{
		waku:             waku,
		api:              api,
		cache:            NewProcessedMessageIDsCache(db),
		envelopesMonitor: envelopesMonitor,
		quit:             make(chan struct{}),
		keysManager: &transportKeysManager{
			waku:              waku,
			privateKey:        privateKey,
			passToSymKeyCache: make(map[string]string),
		},
		filters:     filtersManager,
		mailservers: mailservers,
		logger:      logger.With(zap.Namespace("Transport")),
	}

	for _, opt := range opts {
		if err := opt(t); err != nil {
			return nil, err
		}
	}

	t.cleanFiltersLoop()

	return t, nil
}

func (t *Transport) InitFilters(chatIDs []string, publicKeys []*ecdsa.PublicKey) ([]*Filter, error) {
	return t.filters.Init(chatIDs, publicKeys)
}

func (t *Transport) InitPublicFilters(chatIDs []string) ([]*Filter, error) {
	return t.filters.InitPublicFilters(chatIDs)
}

func (t *Transport) Filters() []*Filter {
	return t.filters.Filters()
}

func (t *Transport) FilterByChatID(chatID string) *Filter {
	return t.filters.FilterByChatID(chatID)
}

func (t *Transport) FiltersByIdentities(identities []string) []*Filter {
	return t.filters.FiltersByIdentities(identities)
}

func (t *Transport) LoadFilters(filters []*Filter) ([]*Filter, error) {
	return t.filters.InitWithFilters(filters)
}

func (t *Transport) InitCommunityFilters(pks []*ecdsa.PrivateKey) ([]*Filter, error) {
	return t.filters.InitCommunityFilters(pks)
}

func (t *Transport) RemoveFilters(filters []*Filter) error {
	return t.filters.Remove(filters...)
}

func (t *Transport) RemoveFilterByChatID(chatID string) (*Filter, error) {
	return t.filters.RemoveFilterByChatID(chatID)
}

func (t *Transport) ResetFilters() error {
	return t.filters.Reset()
}

func (t *Transport) ProcessNegotiatedSecret(secret types.NegotiatedSecret) (*Filter, error) {
	filter, err := t.filters.LoadNegotiated(secret)
	if err != nil {
		return nil, err
	}
	return filter, nil
}

func (t *Transport) JoinPublic(chatID string) (*Filter, error) {
	return t.filters.LoadPublic(chatID)
}

func (t *Transport) LeavePublic(chatID string) error {
	chat := t.filters.Filter(chatID)
	if chat != nil {
		return nil
	}
	return t.filters.Remove(chat)
}

func (t *Transport) JoinPrivate(publicKey *ecdsa.PublicKey) (*Filter, error) {
	return t.filters.LoadContactCode(publicKey)
}

func (t *Transport) JoinGroup(publicKeys []*ecdsa.PublicKey) ([]*Filter, error) {
	var filters []*Filter
	for _, pk := range publicKeys {
		f, err := t.filters.LoadContactCode(pk)
		if err != nil {
			return nil, err
		}
		filters = append(filters, f)

	}
	return filters, nil
}

func (t *Transport) GetStats() types.StatsSummary {
	return t.waku.GetStats()
}

func (t *Transport) RetrieveRawAll() (map[Filter][]*types.Message, error) {
	result := make(map[Filter][]*types.Message)
	logger := t.logger.With(zap.String("site", "retrieveRawAll"))

	for _, filter := range t.filters.Filters() {
		msgs, err := t.api.GetFilterMessages(filter.FilterID)
		if err != nil {
			logger.Warn("failed to fetch messages", zap.Error(err))
			continue
		}
		// Don't pull from filters we don't listen to
		if !filter.Listen {
			for _, msg := range msgs {
				t.waku.MarkP2PMessageAsProcessed(common.BytesToHash(msg.Hash))
			}
			continue
		}

		if len(msgs) == 0 {
			continue
		}

		ids := make([]string, len(msgs))
		for i := range msgs {
			id := types.EncodeHex(msgs[i].Hash)
			ids[i] = id
		}

		hits, err := t.cache.Hits(ids)
		if err != nil {
			logger.Error("failed to check messages exists", zap.Error(err))
			return nil, err
		}

		for i := range msgs {
			// Exclude anything that is a cache hit
			if !hits[types.EncodeHex(msgs[i].Hash)] {
				result[*filter] = append(result[*filter], msgs[i])
				logger.Debug("message not cached", zap.String("hash", types.EncodeHex(msgs[i].Hash)))
			} else {
				logger.Debug("message cached", zap.String("hash", types.EncodeHex(msgs[i].Hash)))
				t.waku.MarkP2PMessageAsProcessed(common.BytesToHash(msgs[i].Hash))
			}
		}

	}

	return result, nil
}

// SendPublic sends a new message using the Whisper service.
// For public filters, chat name is used as an ID as well as
// a topic.
func (t *Transport) SendPublic(ctx context.Context, newMessage *types.NewMessage, chatName string) ([]byte, error) {
	if err := t.addSig(newMessage); err != nil {
		return nil, err
	}

	filter, err := t.filters.LoadPublic(chatName)
	if err != nil {
		return nil, err
	}

	newMessage.SymKeyID = filter.SymKeyID
	newMessage.Topic = filter.Topic

	return t.api.Post(ctx, *newMessage)
}

func (t *Transport) SendPrivateWithSharedSecret(ctx context.Context, newMessage *types.NewMessage, publicKey *ecdsa.PublicKey, secret []byte) ([]byte, error) {
	if err := t.addSig(newMessage); err != nil {
		return nil, err
	}

	filter, err := t.filters.LoadNegotiated(types.NegotiatedSecret{
		PublicKey: publicKey,
		Key:       secret,
	})
	if err != nil {
		return nil, err
	}

	newMessage.SymKeyID = filter.SymKeyID
	newMessage.Topic = filter.Topic
	newMessage.PublicKey = nil

	return t.api.Post(ctx, *newMessage)
}

func (t *Transport) SendPrivateWithPartitioned(ctx context.Context, newMessage *types.NewMessage, publicKey *ecdsa.PublicKey) ([]byte, error) {
	if err := t.addSig(newMessage); err != nil {
		return nil, err
	}

	filter, err := t.filters.LoadPartitioned(publicKey, t.keysManager.privateKey, false)
	if err != nil {
		return nil, err
	}

	newMessage.Topic = filter.Topic
	newMessage.PublicKey = crypto.FromECDSAPub(publicKey)

	return t.api.Post(ctx, *newMessage)
}

func (t *Transport) SendPrivateOnPersonalTopic(ctx context.Context, newMessage *types.NewMessage, publicKey *ecdsa.PublicKey) ([]byte, error) {
	if err := t.addSig(newMessage); err != nil {
		return nil, err
	}

	filter, err := t.filters.LoadPersonal(publicKey, t.keysManager.privateKey, false)
	if err != nil {
		return nil, err
	}

	newMessage.Topic = filter.Topic
	newMessage.PublicKey = crypto.FromECDSAPub(publicKey)

	return t.api.Post(ctx, *newMessage)
}

func (t *Transport) PersonalTopicFilter() *Filter {
	return t.filters.PersonalTopicFilter()
}

func (t *Transport) LoadKeyFilters(key *ecdsa.PrivateKey) (*Filter, error) {
	return t.filters.LoadEphemeral(&key.PublicKey, key, true)
}

func (t *Transport) SendCommunityMessage(ctx context.Context, newMessage *types.NewMessage, publicKey *ecdsa.PublicKey) ([]byte, error) {
	if err := t.addSig(newMessage); err != nil {
		return nil, err
	}

	// We load the filter to make sure we can post on it
	filter, err := t.filters.LoadPublic(PubkeyToHex(publicKey)[2:])
	if err != nil {
		return nil, err
	}

	newMessage.Topic = filter.Topic
	newMessage.PublicKey = crypto.FromECDSAPub(publicKey)

	t.logger.Debug("SENDING message", zap.Binary("topic", filter.Topic[:]))

	return t.api.Post(ctx, *newMessage)
}

func (t *Transport) cleanFilters() error {
	return t.filters.RemoveNoListenFilters()
}

func (t *Transport) addSig(newMessage *types.NewMessage) error {
	sigID, err := t.keysManager.AddOrGetKeyPair(t.keysManager.privateKey)
	if err != nil {
		return err
	}
	newMessage.SigID = sigID
	return nil
}

func (t *Transport) Track(identifiers [][]byte, hash []byte, newMessage *types.NewMessage) {
	if t.envelopesMonitor != nil {
		t.envelopesMonitor.Add(identifiers, types.BytesToHash(hash), *newMessage)
	}
}

// GetCurrentTime returns the current unix timestamp in milliseconds
func (t *Transport) GetCurrentTime() uint64 {
	return uint64(t.waku.GetCurrentTime().UnixNano() / int64(time.Millisecond))
}

func (t *Transport) MaxMessageSize() uint32 {
	return t.waku.MaxMessageSize()
}

func (t *Transport) Stop() error {
	close(t.quit)
	if t.envelopesMonitor != nil {
		t.envelopesMonitor.Stop()
	}
	return nil
}

// cleanFiltersLoop cleans up the topic we create for the only purpose
// of sending messages.
// Whenever we send a message we also need to listen to that particular topic
// but in case of asymettric topics, we are not interested in listening to them.
// We therefore periodically clean them up so we don't receive unnecessary data.

func (t *Transport) cleanFiltersLoop() {

	ticker := time.NewTicker(5 * time.Minute)
	go func() {
		for {
			select {
			case <-t.quit:
				ticker.Stop()
				return
			case <-ticker.C:
				err := t.cleanFilters()
				if err != nil {
					t.logger.Error("failed to clean up topics", zap.Error(err))
				}
			}
		}
	}()
}

func (t *Transport) WakuVersion() uint {
	return t.waku.Version()
}

func (t *Transport) PeerCount() int {
	return t.waku.PeerCount()
}

func (t *Transport) Peers() map[string][]string {
	return t.waku.Peers()
}

func (t *Transport) createMessagesRequestV1(
	ctx context.Context,
	peerID []byte,
	from, to uint32,
	previousCursor []byte,
	topics []types.TopicType,
	waitForResponse bool,
) (cursor []byte, err error) {
	r := createMessagesRequest(from, to, previousCursor, nil, topics)

	events := make(chan types.EnvelopeEvent, 10)
	sub := t.waku.SubscribeEnvelopeEvents(events)
	defer sub.Unsubscribe()

	err = t.waku.SendMessagesRequest(peerID, r)
	if err != nil {
		return
	}

	if !waitForResponse {
		return
	}

	var resp *types.MailServerResponse
	resp, err = t.waitForRequestCompleted(ctx, r.ID, events)
	if err == nil && resp != nil && resp.Error != nil {
		err = resp.Error
	} else if err == nil && resp != nil {
		cursor = resp.Cursor
	}

	return
}

func (t *Transport) createMessagesRequestV2(
	peerID []byte,
	from, to uint32,
	previousStoreCursor *types.StoreRequestCursor,
	topics []types.TopicType,
) (storeCursor *types.StoreRequestCursor, err error) {
	r := createMessagesRequest(from, to, nil, previousStoreCursor, topics)
	storeCursor, err = t.waku.RequestStoreMessages(peerID, r)
	if err != nil {
		return
	}
	return
}

func (t *Transport) SendMessagesRequestForTopics(
	ctx context.Context,
	peerID []byte,
	from, to uint32,
	previousCursor []byte,
	previousStoreCursor *types.StoreRequestCursor,
	topics []types.TopicType,
	waitForResponse bool,
) (cursor []byte, storeCursor *types.StoreRequestCursor, err error) {
	switch t.waku.Version() {
	case 2:
		storeCursor, err = t.createMessagesRequestV2(peerID, from, to, previousStoreCursor, topics)
	case 1:
		cursor, err = t.createMessagesRequestV1(ctx, peerID, from, to, previousCursor, topics, waitForResponse)
	default:
		err = fmt.Errorf("unsupported version %d", t.waku.Version())
	}
	return
}

// RequestHistoricMessages requests historic messages for all registered filters.
func (t *Transport) SendMessagesRequest(
	ctx context.Context,
	peerID []byte,
	from, to uint32,
	previousCursor []byte,
	previousStoreCursor *types.StoreRequestCursor,
	waitForResponse bool,
) (cursor []byte, storeCursor *types.StoreRequestCursor, err error) {

	topics := make([]types.TopicType, len(t.Filters()))
	for _, f := range t.Filters() {
		topics = append(topics, f.Topic)
	}

	return t.SendMessagesRequestForTopics(ctx, peerID, from, to, previousCursor, previousStoreCursor, topics, waitForResponse)
}

func (t *Transport) SendMessagesRequestForFilter(
	ctx context.Context,
	peerID []byte,
	from, to uint32,
	previousCursor []byte,
	previousStoreCursor *types.StoreRequestCursor,
	filter *Filter,
	waitForResponse bool,
) (cursor []byte, storeCursor *types.StoreRequestCursor, err error) {

	topics := make([]types.TopicType, len(t.Filters()))
	topics = append(topics, filter.Topic)

	return t.SendMessagesRequestForTopics(ctx, peerID, from, to, previousCursor, previousStoreCursor, topics, waitForResponse)
}

func createMessagesRequest(from, to uint32, cursor []byte, storeCursor *types.StoreRequestCursor, topics []types.TopicType) types.MessagesRequest {
	aUUID := uuid.New()
	// uuid is 16 bytes, converted to hex it's 32 bytes as expected by types.MessagesRequest
	id := []byte(hex.EncodeToString(aUUID[:]))
	var topicBytes [][]byte
	for idx := range topics {
		topicBytes = append(topicBytes, topics[idx][:])
	}
	return types.MessagesRequest{
		ID:          id,
		From:        from,
		To:          to,
		Limit:       1000,
		Cursor:      cursor,
		Topics:      topicBytes,
		StoreCursor: storeCursor,
	}
}

func (t *Transport) waitForRequestCompleted(ctx context.Context, requestID []byte, events chan types.EnvelopeEvent) (*types.MailServerResponse, error) {
	for {
		select {
		case ev := <-events:
			if !bytes.Equal(ev.Hash.Bytes(), requestID) {
				continue
			}
			if ev.Event != types.EventMailServerRequestCompleted {
				continue
			}
			data, ok := ev.Data.(*types.MailServerResponse)
			if ok {
				return data, nil
			}
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
}

// ConfirmMessagesProcessed marks the messages as processed in the cache so
// they won't be passed to the next layer anymore
func (t *Transport) ConfirmMessagesProcessed(ids []string, timestamp uint64) error {
	return t.cache.Add(ids, timestamp)
}

// CleanMessagesProcessed clears the messages that are older than timestamp
func (t *Transport) CleanMessagesProcessed(timestamp uint64) error {
	return t.cache.Clean(timestamp)
}

func (t *Transport) SetEnvelopeEventsHandler(handler EnvelopeEventsHandler) error {
	if t.envelopesMonitor == nil {
		return errors.New("Current transport has no envelopes monitor")
	}
	t.envelopesMonitor.handler = handler
	return nil
}

func (t *Transport) ClearProcessedMessageIDsCache() error {
	return t.cache.Clear()
}

func (t *Transport) BloomFilter() []byte {
	return t.api.BloomFilter()
}

func PubkeyToHex(key *ecdsa.PublicKey) string {
	return types.EncodeHex(crypto.FromECDSAPub(key))
}

func (t *Transport) StartDiscV5() error {
	return t.waku.StartDiscV5()
}

func (t *Transport) StopDiscV5() error {
	return t.waku.StopDiscV5()
}

func (t *Transport) AddStorePeer(address string) (string, error) {
	return t.waku.AddStorePeer(address)
}

func (t *Transport) AddRelayPeer(address string) (string, error) {
	return t.waku.AddRelayPeer(address)
}

func (t *Transport) DialPeer(address string) error {
	return t.waku.DialPeer(address)
}

func (t *Transport) DialPeerByID(peerID string) error {
	return t.waku.DialPeerByID(peerID)
}

func (t *Transport) DropPeer(peerID string) error {
	return t.waku.DropPeer(peerID)
}

func (t *Transport) ProcessingP2PMessages() bool {
	return t.waku.ProcessingP2PMessages()
}

func (t *Transport) MarkP2PMessageAsProcessed(hash common.Hash) {
	t.waku.MarkP2PMessageAsProcessed(hash)
}

func (t *Transport) SubscribeToConnStatusChanges() (*types.ConnStatusSubscription, error) {
	return t.waku.SubscribeToConnStatusChanges()
}