driver.go

/*
 * Copyright 2018 The CovenantSQL Authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package client

import (
	"context"
	"database/sql"
	"database/sql/driver"
	"fmt"
	"math/rand"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	"github.com/pkg/errors"

	bp "github.com/CovenantSQL/CovenantSQL/blockproducer"
	"github.com/CovenantSQL/CovenantSQL/blockproducer/interfaces"
	"github.com/CovenantSQL/CovenantSQL/conf"
	"github.com/CovenantSQL/CovenantSQL/crypto"
	"github.com/CovenantSQL/CovenantSQL/crypto/asymmetric"
	"github.com/CovenantSQL/CovenantSQL/crypto/hash"
	"github.com/CovenantSQL/CovenantSQL/crypto/kms"
	"github.com/CovenantSQL/CovenantSQL/proto"
	"github.com/CovenantSQL/CovenantSQL/route"
	rpc "github.com/CovenantSQL/CovenantSQL/rpc/mux"
	"github.com/CovenantSQL/CovenantSQL/types"
	"github.com/CovenantSQL/CovenantSQL/utils"
	"github.com/CovenantSQL/CovenantSQL/utils/log"
)

const (
	// DBScheme defines the dsn scheme.
	DBScheme = "covenantsql"
	// DBSchemeAlias defines the alias dsn scheme.
	DBSchemeAlias = "cql"
	// DefaultGasPrice defines the default gas price for new created database.
	DefaultGasPrice = 1
	// DefaultAdvancePayment defines the default advance payment for new created database.
	DefaultAdvancePayment = 20000000
)

var (
	// PeersUpdateInterval defines peers list refresh interval for client.
	PeersUpdateInterval = time.Second * 5

	driverInitialized uint32
	peersUpdaterRunning uint32
	peerList sync.Map // map[proto.DatabaseID]*proto.Peers
	connIDLock sync.Mutex
	connIDAvail []uint64
	globalSeqNo uint64
	randSource = rand.New(rand.NewSource(time.Now().UnixNano()))

	// DefaultConfigFile is the default path of config file
	DefaultConfigFile = "~/.cql/config.yaml"
)

func init() {
	d := new(covenantSQLDriver)
	sql.Register(DBScheme, d)
	sql.Register(DBSchemeAlias, d)
	log.Debug("CovenantSQL driver registered.")
}

// covenantSQLDriver implements sql.Driver interface.
type covenantSQLDriver struct {
}

// Open returns new db connection.
func (d *covenantSQLDriver) Open(dsn string) (conn driver.Conn, err error) {
	var cfg *Config
	if cfg, err = ParseDSN(dsn); err != nil {
		return
	}

	if atomic.LoadUint32(&driverInitialized) == 0 {
		err = defaultInit()
		if err != nil && err != ErrAlreadyInitialized {
			return
		}
	}

	return newConn(cfg)
}

// ResourceMeta defines new database resources requirement descriptions.
type ResourceMeta struct {
	// copied fields from types.ResourceMeta
	TargetMiners []proto.AccountAddress `json:"target-miners,omitempty"` // designated miners
	Node uint16 `json:"node,omitempty"` // reserved node count
	Space uint64 `json:"space,omitempty"` // reserved storage space in bytes
	Memory uint64 `json:"memory,omitempty"` // reserved memory in bytes
	LoadAvgPerCPU float64 `json:"load-avg-per-cpu,omitempty"` // max loadAvg15 per CPU
	EncryptionKey string `json:"encrypt-key,omitempty"` // encryption key for database instance
	UseEventualConsistency bool `json:"eventual-consistency,omitempty"` // use eventual consistency replication if enabled
	ConsistencyLevel float64 `json:"consistency-level,omitempty"` // customized strong consistency level
	IsolationLevel int `json:"isolation-level,omitempty"` // customized isolation level

	GasPrice uint64 `json:"gas-price"` // customized gas price
	AdvancePayment uint64 `json:"advance-payment"` // customized advance payment
}

func defaultInit() (err error) {
	configFile := utils.HomeDirExpand(DefaultConfigFile)
	if configFile == DefaultConfigFile {
		//System not support ~ dir, need Init manually.
		log.Debugf("Could not find CovenantSQL default config location: %v", configFile)
		return ErrNotInitialized
	}

	log.Debugf("Using CovenantSQL default config location: %v", configFile)
	return Init(configFile, []byte(""))
}

// Init defines init process for client.
func Init(configFile string, masterKey []byte) (err error) {
	if !atomic.CompareAndSwapUint32(&driverInitialized, 0, 1) {
		err = ErrAlreadyInitialized
		return
	}

	// load config
	if conf.GConf, err = conf.LoadConfig(configFile); err != nil {
		return
	}

	route.InitKMS(conf.GConf.PubKeyStoreFile)
	if err = kms.InitLocalKeyPair(conf.GConf.PrivateKeyFile, masterKey); err != nil {
		return
	}

	// ping block producer to register node
	if err = registerNode(); err != nil {
		return
	}

	// run peers updater
	if err = runPeerListUpdater(); err != nil {
		return
	}

	return
}

// Create sends create database operation to block producer.
func Create(meta ResourceMeta) (txHash hash.Hash, dsn string, err error) {
	if atomic.LoadUint32(&driverInitialized) == 0 {
		err = ErrNotInitialized
		return
	}

	var (
		nonceReq = new(types.NextAccountNonceReq)
		nonceResp = new(types.NextAccountNonceResp)
		req = new(types.AddTxReq)
		resp = new(types.AddTxResp)
		privateKey *asymmetric.PrivateKey
		clientAddr proto.AccountAddress
	)
	if privateKey, err = kms.GetLocalPrivateKey(); err != nil {
		err = errors.Wrap(err, "get local private key failed")
		return
	}
	if clientAddr, err = crypto.PubKeyHash(privateKey.PubKey()); err != nil {
		err = errors.Wrap(err, "get local account address failed")
		return
	}
	// allocate nonce
	nonceReq.Addr = clientAddr

	if err = requestBP(route.MCCNextAccountNonce, nonceReq, nonceResp); err != nil {
		err = errors.Wrap(err, "allocate create database transaction nonce failed")
		return
	}

	if meta.GasPrice == 0 {
		meta.GasPrice = DefaultGasPrice
	}
	if meta.AdvancePayment == 0 {
		meta.AdvancePayment = DefaultAdvancePayment
	}

	req.TTL = 1
	req.Tx = types.NewCreateDatabase(&types.CreateDatabaseHeader{
		Owner: clientAddr,
		ResourceMeta: types.ResourceMeta{
			TargetMiners: meta.TargetMiners,
			Node: meta.Node,
			Space: meta.Space,
			Memory: meta.Memory,
			LoadAvgPerCPU: meta.LoadAvgPerCPU,
			EncryptionKey: meta.EncryptionKey,
			UseEventualConsistency: meta.UseEventualConsistency,
			ConsistencyLevel: meta.ConsistencyLevel,
			IsolationLevel: meta.IsolationLevel,
		},
		GasPrice: meta.GasPrice,
		AdvancePayment: meta.AdvancePayment,
		TokenType: types.Particle,
		Nonce: nonceResp.Nonce,
	})

	if err = req.Tx.Sign(privateKey); err != nil {
		err = errors.Wrap(err, "sign request failed")
		return
	}

	if err = requestBP(route.MCCAddTx, req, resp); err != nil {
		err = errors.Wrap(err, "call create database transaction failed")
		return
	}

	txHash = req.Tx.Hash()
	cfg := NewConfig()
	cfg.DatabaseID = string(proto.FromAccountAndNonce(clientAddr, uint32(nonceResp.Nonce)))
	dsn = cfg.FormatDSN()

	return
}

// WaitDBCreation waits for database creation complete.
func WaitDBCreation(ctx context.Context, dsn string) (err error) {
	dsnCfg, err := ParseDSN(dsn)
	if err != nil {
		return
	}

	db, err := sql.Open("covenantsql", dsn)
	defer db.Close()
	if err != nil {
		return
	}

	// wait for creation
	err = WaitBPDatabaseCreation(ctx, proto.DatabaseID(dsnCfg.DatabaseID), db, 3*time.Second)
	return
}

// WaitBPDatabaseCreation waits for database creation complete.
func WaitBPDatabaseCreation(
	ctx context.Context,
	dbID proto.DatabaseID,
	db *sql.DB,
	period time.Duration,
) (err error) {
	var (
		ticker = time.NewTicker(period)
		req = &types.QuerySQLChainProfileReq{
			DBID: dbID,
		}
		count = 0
	)
	defer ticker.Stop()
	defer fmt.Printf("\n")
	for {
		select {
		case <-ticker.C:
			count++

			if err = rpc.RequestBP(
				route.MCCQuerySQLChainProfile.String(), req, nil,
			); err != nil {
				if !strings.Contains(err.Error(), bp.ErrDatabaseNotFound.Error()) {
					// err != nil && err != ErrDatabaseNotFound (unexpected error)
					return
				}
			} else {
				// err == nil (creation done on BP): try to use database connection
				if db == nil {
					return
				}
				if _, err = db.ExecContext(ctx, "SHOW TABLES"); err == nil {
					// err == nil (connect to Miner OK)
					return
				}
			}
			fmt.Printf("\rQuerying SQLChain Profile %vs", count*int(period.Seconds()))
		case <-ctx.Done():
			if err != nil {
				return errors.Wrapf(ctx.Err(), "last error: %s", err.Error())
			}
			return ctx.Err()
		}
	}
}

// Drop sends drop database operation to block producer.
func Drop(dsn string) (txHash hash.Hash, err error) {
	if atomic.LoadUint32(&driverInitialized) == 0 {
		err = ErrNotInitialized
		return
	}

	var cfg *Config
	if cfg, err = ParseDSN(dsn); err != nil {
		return
	}

	peerList.Delete(cfg.DatabaseID)

	//TODO(laodouya) currently not supported
	//err = errors.New("drop db current not support")

	return
}

// GetTokenBalance get the token balance of current account.
func GetTokenBalance(tt types.TokenType) (balance uint64, err error) {
	if atomic.LoadUint32(&driverInitialized) == 0 {
		err = ErrNotInitialized
		return
	}

	req := new(types.QueryAccountTokenBalanceReq)
	resp := new(types.QueryAccountTokenBalanceResp)

	var pubKey *asymmetric.PublicKey
	if pubKey, err = kms.GetLocalPublicKey(); err != nil {
		return
	}

	if req.Addr, err = crypto.PubKeyHash(pubKey); err != nil {
		return
	}
	req.TokenType = tt

	if err = requestBP(route.MCCQueryAccountTokenBalance, req, resp); err == nil {
		if !resp.OK {
			err = ErrNoSuchTokenBalance
			return
		}
		balance = resp.Balance
	}

	return
}

// UpdatePermission sends UpdatePermission transaction to chain.
func UpdatePermission(targetUser proto.AccountAddress,
	targetChain proto.AccountAddress, perm *types.UserPermission) (txHash hash.Hash, err error) {
	if atomic.LoadUint32(&driverInitialized) == 0 {
		err = ErrNotInitialized
		return
	}

	var (
		pubKey *asymmetric.PublicKey
		privKey *asymmetric.PrivateKey
		addr proto.AccountAddress
		nonce interfaces.AccountNonce
	)
	if pubKey, err = kms.GetLocalPublicKey(); err != nil {
		return
	}
	if privKey, err = kms.GetLocalPrivateKey(); err != nil {
		return
	}
	if addr, err = crypto.PubKeyHash(pubKey); err != nil {
		return
	}

	nonce, err = getNonce(addr)
	if err != nil {
		return
	}

	up := types.NewUpdatePermission(&types.UpdatePermissionHeader{
		TargetSQLChain: targetChain,
		TargetUser: targetUser,
		Permission: perm,
		Nonce: nonce,
	})
	err = up.Sign(privKey)
	if err != nil {
		log.WithError(err).Warning("sign failed")
		return
	}
	addTxReq := new(types.AddTxReq)
	addTxResp := new(types.AddTxResp)
	addTxReq.Tx = up
	err = requestBP(route.MCCAddTx, addTxReq, addTxResp)
	if err != nil {
		log.WithError(err).Warning("send tx failed")
		return
	}

	txHash = up.Hash()
	return
}

// TransferToken send Transfer transaction to chain.
func TransferToken(targetUser proto.AccountAddress, amount uint64, tokenType types.TokenType) (
	txHash hash.Hash, err error,
) {
	if atomic.LoadUint32(&driverInitialized) == 0 {
		err = ErrNotInitialized
		return
	}

	var (
		pubKey *asymmetric.PublicKey
		privKey *asymmetric.PrivateKey
		addr proto.AccountAddress
		nonce interfaces.AccountNonce
	)
	if pubKey, err = kms.GetLocalPublicKey(); err != nil {
		return
	}
	if privKey, err = kms.GetLocalPrivateKey(); err != nil {
		return
	}
	if addr, err = crypto.PubKeyHash(pubKey); err != nil {
		return
	}

	nonce, err = getNonce(addr)
	if err != nil {
		return
	}

	tran := types.NewTransfer(&types.TransferHeader{
		Sender: addr,
		Receiver: targetUser,
		Amount: amount,
		TokenType: tokenType,
		Nonce: nonce,
	})
	err = tran.Sign(privKey)
	if err != nil {
		log.WithError(err).Warning("sign failed")
		return
	}
	addTxReq := new(types.AddTxReq)
	addTxResp := new(types.AddTxResp)
	addTxReq.Tx = tran
	err = requestBP(route.MCCAddTx, addTxReq, addTxResp)
	if err != nil {
		log.WithError(err).Warning("send tx failed")
		return
	}

	txHash = tran.Hash()
	return
}

// WaitTxConfirmation waits for the transaction with target hash txHash to be confirmed. It also
// returns if any error occurs or a final state is returned from BP.
func WaitTxConfirmation(
	ctx context.Context, txHash hash.Hash) (state interfaces.TransactionState, err error,
) {
	var (
		ticker = time.NewTicker(1 * time.Second)
		method = route.MCCQueryTxState
		req = &types.QueryTxStateReq{Hash: txHash}
		resp = &types.QueryTxStateResp{}
		count = 0
	)
	defer ticker.Stop()
	defer fmt.Printf("\n")
	for {
		if err = requestBP(method, req, resp); err != nil {
			err = errors.Wrapf(err, "failed to call %s", method)
			return
		}

		state = resp.State

		count++
		fmt.Printf("\rWaiting blockproducers confirmation %vs, state: %v\033[0K", count, state)
		log.WithFields(log.Fields{
			"tx_hash": txHash,
			"tx_state": state,
		}).Debug("waiting for tx confirmation")

		switch state {
		case interfaces.TransactionStatePending:
		case interfaces.TransactionStatePacked:
		case interfaces.TransactionStateConfirmed,
			interfaces.TransactionStateExpired,
			interfaces.TransactionStateNotFound:
			return
		default:
			err = errors.Errorf("unknown transaction state %d", state)
			return
		}

		select {
		case <-ticker.C:
		case <-ctx.Done():
			err = ctx.Err()
			return
		}
	}
}

func getNonce(addr proto.AccountAddress) (nonce interfaces.AccountNonce, err error) {
	nonceReq := new(types.NextAccountNonceReq)
	nonceResp := new(types.NextAccountNonceResp)
	nonceReq.Addr = addr
	err = requestBP(route.MCCNextAccountNonce, nonceReq, nonceResp)
	if err != nil {
		log.WithError(err).Warning("get nonce failed")
		return
	}
	nonce = nonceResp.Nonce
	return
}

func requestBP(method route.RemoteFunc, request interface{}, response interface{}) (err error) {
	var bpNodeID proto.NodeID
	if bpNodeID, err = rpc.GetCurrentBP(); err != nil {
		return
	}

	return rpc.NewCaller().CallNode(bpNodeID, method.String(), request, response)
}

func registerNode() (err error) {
	var nodeID proto.NodeID

	if nodeID, err = kms.GetLocalNodeID(); err != nil {
		return
	}

	var nodeInfo *proto.Node
	if nodeInfo, err = kms.GetNodeInfo(nodeID); err != nil {
		return
	}

	if nodeInfo.Role != proto.Leader && nodeInfo.Role != proto.Follower {
		log.Infof("Self register to blockproducer: %v", conf.GConf.BP.NodeID)
		err = rpc.PingBP(nodeInfo, conf.GConf.BP.NodeID)
	}

	return
}

func runPeerListUpdater() (err error) {
	var privKey *asymmetric.PrivateKey
	if privKey, err = kms.GetLocalPrivateKey(); err != nil {
		return
	}

	if !atomic.CompareAndSwapUint32(&peersUpdaterRunning, 0, 1) {
		return
	}

	go func() {
		for {
			if atomic.LoadUint32(&peersUpdaterRunning) == 0 {
				return
			}

			var wg sync.WaitGroup

			peerList.Range(func(rawDBID, _ interface{}) bool {
				dbID := rawDBID.(proto.DatabaseID)

				wg.Add(1)
				go func(dbID proto.DatabaseID) {
					defer wg.Done()
					var err error

					if _, err = getPeers(dbID, privKey); err != nil {
						log.WithField("db", dbID).
							WithError(err).
							Debug("update peers failed")

						// TODO(xq262144), better rpc remote error judgement
						if strings.Contains(err.Error(), bp.ErrNoSuchDatabase.Error()) {
							log.WithField("db", dbID).
								Warning("database no longer exists, stopping peers update")
							peerList.Delete(dbID)
						}
					}
				}(dbID)

				return true
			})

			wg.Wait()

			time.Sleep(PeersUpdateInterval)
		}
	}()

	return
}

func stopPeersUpdater() {
	atomic.StoreUint32(&peersUpdaterRunning, 0)
}

func cacheGetPeers(dbID proto.DatabaseID, privKey *asymmetric.PrivateKey) (peers *proto.Peers, err error) {
	var ok bool
	var rawPeers interface{}
	var cacheHit bool

	defer func() {
		log.WithFields(log.Fields{
			"db": dbID,
			"hit": cacheHit,
		}).WithError(err).Debug("cache get peers for database")
	}()

	if rawPeers, ok = peerList.Load(dbID); ok {
		if peers, ok = rawPeers.(*proto.Peers); ok {
			cacheHit = true
			return
		}
	}

	// get peers using non-cache method
	return getPeers(dbID, privKey)
}

func getPeers(dbID proto.DatabaseID, privKey *asymmetric.PrivateKey) (peers *proto.Peers, err error) {
	defer func() {
		log.WithFields(log.Fields{
			"db": dbID,
			"peers": peers,
		}).WithError(err).Debug("get peers for database")
	}()

	profileReq := &types.QuerySQLChainProfileReq{}
	profileResp := &types.QuerySQLChainProfileResp{}
	profileReq.DBID = dbID
	err = rpc.RequestBP(route.MCCQuerySQLChainProfile.String(), profileReq, profileResp)
	if err != nil {
		err = errors.Wrap(err, "get sqlchain profile failed in getPeers")
		return
	}

	nodeIDs := make([]proto.NodeID, len(profileResp.Profile.Miners))
	if len(profileResp.Profile.Miners) <= 0 {
		err = errors.Wrap(ErrInvalidProfile, "unexpected error in getPeers")
		return
	}
	for i, mi := range profileResp.Profile.Miners {
		nodeIDs[i] = mi.NodeID
	}
	peers = &proto.Peers{
		PeersHeader: proto.PeersHeader{
			Leader: nodeIDs[0],
			Servers: nodeIDs[:],
		},
	}
	err = peers.Sign(privKey)
	if err != nil {
		err = errors.Wrap(err, "sign peers failed in getPeers")
		return
	}

	// set peers in the updater cache
	peerList.Store(dbID, peers)

	return
}

func allocateConnAndSeq() (connID uint64, seqNo uint64) {
	connIDLock.Lock()
	defer connIDLock.Unlock()

	if len(connIDAvail) == 0 {
		// generate one
		connID = randSource.Uint64()
		seqNo = atomic.AddUint64(&globalSeqNo, 1)
		return
	}

	// pop one conn
	connID = connIDAvail[0]
	connIDAvail = connIDAvail[1:]
	seqNo = atomic.AddUint64(&globalSeqNo, 1)

	return
}

func putBackConn(connID uint64) {
	connIDLock.Lock()
	defer connIDLock.Unlock()

	connIDAvail = append(connIDAvail, connID)
}