trade.go

package daos

import (
	"strconv"
	"time"

	"github.com/ethereum/go-ethereum/common"
	"github.com/globalsign/mgo"
	"github.com/globalsign/mgo/bson"
	"github.com/tomochain/tomox-sdk/app"
	"github.com/tomochain/tomox-sdk/types"
)

// TradeDao contains:
// collectionName: MongoDB collection name
// dbName: name of mongodb to interact with
type TradeDao struct {
	collectionName string
	dbName         string
}

// NewTradeDao returns a new instance of TradeDao.
func NewTradeDao() *TradeDao {
	dbName := app.Config.DBName
	collection := "trades"
	i3 := mgo.Index{
		Key: []string{"createdAt"},
	}

	db.Session.DB(dbName).C(collection).EnsureIndex(i3)
	return &TradeDao{collection, dbName}
}

func (dao *TradeDao) GetCollection() *mgo.Collection {
	return db.GetCollection(dao.dbName, dao.collectionName)
}

func (dao *TradeDao) Watch() (*mgo.ChangeStream, *mgo.Session, error) {
	return db.Watch(dao.dbName, dao.collectionName, mgo.ChangeStreamOptions{
		FullDocument:   mgo.UpdateLookup,
		MaxAwaitTimeMS: 500,
		BatchSize:      1000,
	})
}

// Create function performs the DB insertion task for trade collection
// It accepts 1 or more trades as input.
// All the trades are inserted in one query itself.
func (dao *TradeDao) Create(trades ...*types.Trade) error {
	y := make([]interface{}, len(trades))

	for _, trade := range trades {
		trade.ID = bson.NewObjectId()
		trade.CreatedAt = time.Now()
		trade.UpdatedAt = time.Now()
		y = append(y, trade)
	}

	err := db.Create(dao.dbName, dao.collectionName, y...)
	if err != nil {
		logger.Error(err)
		return err
	}

	return nil
}

func (dao *TradeDao) DeleteByHashes(hashes ...common.Hash) error {
	err := db.RemoveAll(dao.dbName, dao.collectionName, bson.M{"hash": bson.M{"$in": hashes}})
	if err != nil {
		logger.Error(err)
		return err
	}

	return nil
}

func (dao *TradeDao) Delete(trades ...*types.Trade) error {
	hashes := []common.Hash{}
	for _, t := range trades {
		hashes = append(hashes, t.Hash)
	}

	err := db.RemoveAll(dao.dbName, dao.collectionName, bson.M{"hash": bson.M{"$in": hashes}})
	if err != nil {
		logger.Error(err)
		return err
	}

	return nil
}

func (dao *TradeDao) Update(trade *types.Trade) error {
	trade.UpdatedAt = time.Now()
	err := db.Update(dao.dbName, dao.collectionName, bson.M{"_id": trade.ID}, trade)
	if err != nil {
		logger.Error(err)
		return err
	}

	return nil
}

func (dao *TradeDao) Upsert(id bson.ObjectId, t *types.Trade) error {
	t.UpdatedAt = time.Now()

	_, err := db.Upsert(dao.dbName, dao.collectionName, bson.M{"_id": id}, t)
	if err != nil {
		logger.Error(err)
		return err
	}

	return nil
}

func (dao *TradeDao) UpsertByHash(h common.Hash, t *types.Trade) error {
	t.UpdatedAt = time.Now()

	_, err := db.Upsert(dao.dbName, dao.collectionName, bson.M{"hash": h}, t)
	if err != nil {
		logger.Error(err)
		return err
	}

	return nil
}

func (dao *TradeDao) FindAndModify(h common.Hash, t *types.Trade) (*types.Trade, error) {
	t.UpdatedAt = time.Now()
	query := bson.M{"hash": h.Hex()}
	updated := &types.Trade{}
	change := mgo.Change{
		Update:    types.TradeBSONUpdate{t},
		Upsert:    true,
		Remove:    false,
		ReturnNew: true,
	}

	err := db.FindAndModify(dao.dbName, dao.collectionName, query, change, &updated)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return updated, nil
}

// UpdateByHash updates the fields that can be normally updated in a structure. For a
// complete update, use the Update or UpdateAllByHash function
func (dao *TradeDao) UpdateByHash(h common.Hash, t *types.Trade) error {
	t.UpdatedAt = time.Now()
	query := bson.M{"hash": h.Hex()}
	update := bson.M{"$set": bson.M{
		"pricepoint":     t.PricePoint.String(),
		"amount":         t.Amount.String(),
		"txHash":         t.TxHash.String(),
		"takerOrderHash": t.TakerOrderHash.String(),
		"makerOrderHash": t.MakerOrderHash.String(),
		"updatedAt":      t.UpdatedAt,
	}}

	err := db.Update(dao.dbName, dao.collectionName, query, update)
	if err != nil {
		logger.Error(err)
		return err
	}

	return nil
}

// GetAll function fetches all the trades in mongodb
func (dao *TradeDao) GetAll() ([]types.Trade, error) {
	var response []types.Trade
	err := db.Get(dao.dbName, dao.collectionName, bson.M{}, 0, 0, &response)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return response, nil
}

// Aggregate function calls the aggregate pipeline of mongodb
func (dao *TradeDao) Aggregate(q []bson.M) ([]*types.Tick, error) {
	var res []*types.Tick

	err := db.Aggregate(dao.dbName, dao.collectionName, q, &res)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

// GetByPairName fetches all the trades corresponding to a particular pair name.
func (dao *TradeDao) GetByPairName(name string) ([]*types.Trade, error) {
	var res []*types.Trade
	q := bson.M{"pairName": bson.RegEx{
		Pattern: name,
		Options: "i",
	}}

	err := db.Get(dao.dbName, dao.collectionName, q, 0, 0, &res)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

// GetByHash fetches the first record that matches a certain hash
func (dao *TradeDao) GetByHash(h common.Hash) (*types.Trade, error) {
	q := bson.M{"hash": h.Hex()}

	res := []*types.Trade{}
	err := db.Get(dao.dbName, dao.collectionName, q, 0, 0, &res)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res[0], nil
}

// GetByOrderHash fetches the first trade record which matches a certain order hash
func (dao *TradeDao) GetByMakerOrderHash(h common.Hash) ([]*types.Trade, error) {
	q := bson.M{"makerOrderHash": h.Hex()}

	res := []*types.Trade{}
	err := db.Get(dao.dbName, dao.collectionName, q, 0, 0, &res)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

func (dao *TradeDao) GetByTakerOrderHash(h common.Hash) ([]*types.Trade, error) {
	q := bson.M{"takerOrderHash": h.Hex()}

	res := []*types.Trade{}
	err := db.Get(dao.dbName, dao.collectionName, q, 0, 0, &res)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

func (dao *TradeDao) GetByOrderHashes(hashes []common.Hash) ([]*types.Trade, error) {
	hexes := []string{}
	for _, h := range hashes {
		hexes = append(hexes, h.Hex())
	}

	q := bson.M{"makerOrderHash": bson.M{"$in": hexes}}
	res := []*types.Trade{}

	err := db.Get(dao.dbName, dao.collectionName, q, 0, 0, &res)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

func (dao *TradeDao) GetSortedTrades(bt, qt common.Address, from, to int64, n int) ([]*types.Trade, error) {
	res := make([]*types.Trade, 0)

	var q bson.M

	if from == 0 || to == 0 {
		q = bson.M{
			"baseToken":  bt.Hex(),
			"quoteToken": qt.Hex(),
		}
	} else {
		q = bson.M{
			"baseToken":  bt.Hex(),
			"quoteToken": qt.Hex(),
			"createdAt": bson.M{
				"$gte": strconv.FormatInt(from, 10),
				"$lt":  strconv.FormatInt(to, 10),
			},
		}
	}

	sort := []string{"-createdAt"}

	err := db.GetAndSort(dao.dbName, dao.collectionName, q, sort, 0, n, &res)

	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

func (dao *TradeDao) GetNTradesByPairAddress(bt, qt common.Address, n int) ([]*types.Trade, error) {
	res, err := dao.GetTradesByPairAddress(bt, qt, n)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

func (dao *TradeDao) GetAllTradesByPairAddress(bt, qt common.Address) ([]*types.Trade, error) {
	res, err := dao.GetTradesByPairAddress(bt, qt, 0)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

// GetByPairAddress fetches all the trades corresponding to a particular pair token address.
func (dao *TradeDao) GetTradesByPairAddress(bt, qt common.Address, n int) ([]*types.Trade, error) {
	var res []*types.Trade

	q := bson.M{"baseToken": bt.Hex(), "quoteToken": qt.Hex()}
	err := db.Get(dao.dbName, dao.collectionName, q, 0, n, &res)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

func (dao *TradeDao) GetSortedTradesByUserAddress(a, bt, qt common.Address, from, to int64, limit ...int) ([]*types.Trade, error) {
	if limit == nil {
		limit = []int{types.DefaultLimit}
	}

	var res []*types.Trade
	var q bson.M

	if (bt == common.Address{} || qt == common.Address{}) {
		if from == 0 || to == 0 {
			q = bson.M{
				"$or": []bson.M{
					{"maker": a.Hex()},
					{"taker": a.Hex()},
				},
			}
		} else {
			q = bson.M{
				"createdAt": bson.M{
					"$gte": strconv.FormatInt(from, 10),
					"$lt":  strconv.FormatInt(to, 10),
				},
				"$or": []bson.M{
					{"maker": a.Hex()},
					{"taker": a.Hex()},
				},
			}
		}
	} else {
		if from == 0 || to == 0 {
			q = bson.M{
				"baseToken":  bt.Hex(),
				"quoteToken": qt.Hex(),
				"$or": []bson.M{
					{"maker": a.Hex()},
					{"taker": a.Hex()},
				},
			}
		} else {
			q = bson.M{
				"baseToken":  bt.Hex(),
				"quoteToken": qt.Hex(),
				"createdAt": bson.M{
					"$gte": strconv.FormatInt(from, 10),
					"$lt":  strconv.FormatInt(to, 10),
				},
				"$or": []bson.M{
					{"maker": a.Hex()},
					{"taker": a.Hex()},
				},
			}
		}
	}

	sort := []string{"-createdAt"}

	err := db.GetAndSort(dao.dbName, dao.collectionName, q, sort, 0, limit[0], &res)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

// GetByUserAddress fetches all the trades corresponding to a particular user address.
func (dao *TradeDao) GetByUserAddress(a common.Address) ([]*types.Trade, error) {
	var res []*types.Trade
	q := bson.M{"$or": []bson.M{{"maker": a.Hex()}, {"taker": a.Hex()}}}

	err := db.Get(dao.dbName, dao.collectionName, q, 0, 0, &res)
	if err != nil {
		logger.Error(err)
		return nil, err
	}

	return res, nil
}

func (dao *TradeDao) GetLatestTrade(bt, qt common.Address) (*types.Trade, error) {
	res, err := dao.GetSortedTrades(bt, qt, 0, 0, 1)

	if err != nil {
		logger.Error(err)
		return nil, err
	}

	if len(res) > 0 {
		return res[0], nil
	}

	return nil, nil
}

func (dao *TradeDao) UpdateTradeStatus(h common.Hash, status string) error {
	query := bson.M{"hash": h.Hex()}
	update := bson.M{"$set": bson.M{"status": status}}

	err := db.Update(dao.dbName, dao.collectionName, query, update)
	if err != nil {
		logger.Error(err)
		return err
	}

	return nil
}

func (dao *TradeDao) UpdateTradeStatuses(status string, hashes ...common.Hash) ([]*types.Trade, error) {
	hexes := []string{}
	for _, h := range hashes {
		hexes = append(hexes, h.Hex())
	}

	query := bson.M{"hash": bson.M{"$in": hexes}}
	update := bson.M{
		"$set": bson.M{
			"updatedAt": time.Now(),
			"status":    status,
		},
	}

	err := db.UpdateAll(dao.dbName, dao.collectionName, query, update)
	if err != nil {
		logger.Error(err)
		return nil, nil
	}

	trades := []*types.Trade{}
	err = db.Get(dao.dbName, dao.collectionName, query, 0, 0, &trades)
	if err != nil {
		logger.Error(err)
		return nil, nil
	}

	return trades, nil
}

func (dao *TradeDao) UpdateTradeStatusesByOrderHashes(status string, hashes ...common.Hash) ([]*types.Trade, error) {
	hexes := []string{}
	for _, h := range hashes {
		hexes = append(hexes, h.Hex())
	}

	query := bson.M{"makerOrderHash": bson.M{"$in": hexes}}
	update := bson.M{
		"$set": bson.M{
			"updatedAt": time.Now(),
			"status":    status,
		},
	}

	err := db.UpdateAll(dao.dbName, dao.collectionName, query, update)
	if err != nil {
		logger.Error(err)
		return nil, nil
	}

	trades := []*types.Trade{}
	err = db.Get(dao.dbName, dao.collectionName, query, 0, 0, &trades)
	if err != nil {
		logger.Error(err)
		return nil, nil
	}

	return trades, nil
}

func (dao *TradeDao) UpdateTradeStatusesByHashes(status string, hashes ...common.Hash) ([]*types.Trade, error) {
	hexes := []string{}
	for _, h := range hashes {
		hexes = append(hexes, h.Hex())
	}

	query := bson.M{"hash": bson.M{"$in": hexes}}
	update := bson.M{
		"$set": bson.M{
			"updatedAt": time.Now(),
			"status":    status,
		},
	}

	err := db.UpdateAll(dao.dbName, dao.collectionName, query, update)
	if err != nil {
		logger.Error(err)
		return nil, nil
	}

	trades := []*types.Trade{}
	err = db.Get(dao.dbName, dao.collectionName, query, 0, 0, &trades)
	if err != nil {
		logger.Error(err)
		return nil, nil
	}

	return trades, nil
}

// Drop drops all the order documents in the current database
func (dao *TradeDao) Drop() {
	db.DropCollection(dao.dbName, dao.collectionName)
}

// GetTrades filter trade
func (dao *TradeDao) GetTrades(tradeSpec *types.TradeSpec, sortedBy []string, pageOffset int, pageSize int) (*types.TradeRes, error) {
	q := bson.M{}

	if tradeSpec.DateFrom != 0 || tradeSpec.DateTo != 0 {
		dateFilter := bson.M{}
		if tradeSpec.DateFrom != 0 {
			dateFilter["$gte"] = time.Unix(tradeSpec.DateFrom, 0)
		}
		if tradeSpec.DateTo != 0 {
			dateFilter["$lt"] = time.Unix(tradeSpec.DateTo, 0)
		}
		q["createdAt"] = dateFilter
	}
	if tradeSpec.BaseToken != "" {
		q["baseToken"] = tradeSpec.BaseToken
	}
	if tradeSpec.QuoteToken != "" {
		q["quoteToken"] = tradeSpec.QuoteToken
	}

	var res types.TradeRes
	trades := []*types.Trade{}
	c, err := db.GetEx(dao.dbName, dao.collectionName, q, sortedBy, pageOffset, pageSize, &trades)
	if err != nil {
		logger.Error(err)
		return nil, err
	}
	res.Total = c
	res.Trades = trades
	return &res, nil
}

// GetTradeByTime get range trade
func (dao *TradeDao) GetTradeByTime(dateFrom, dateTo int64, pageOffset int, pageSize int) ([]*types.Trade, error) {
	q := bson.M{}

	dateFilter := bson.M{}
	dateFilter["$gte"] = time.Unix(dateFrom, 0)
	dateFilter["$lt"] = time.Unix(dateTo, 0)
	q["createdAt"] = dateFilter

	trades := []*types.Trade{}
	_, err := db.GetEx(dao.dbName, dao.collectionName, q, []string{"-createdAt"}, pageOffset, pageSize, &trades)
	if err != nil {
		logger.Error(err)
		return nil, err
	}
	return trades, nil
}

// GetTradesUserHistory get trade by user address
func (dao *TradeDao) GetTradesUserHistory(a common.Address, tradeSpec *types.TradeSpec, sortedBy []string, pageOffset int, pageSize int) (*types.TradeRes, error) {

	q := bson.M{}
	if tradeSpec.DateFrom != 0 || tradeSpec.DateTo != 0 {
		dateFilter := bson.M{}
		if tradeSpec.DateFrom != 0 {
			dateFilter["$gte"] = time.Unix(tradeSpec.DateFrom, 0)
		}
		if tradeSpec.DateTo != 0 {
			dateFilter["$lt"] = time.Unix(tradeSpec.DateTo, 0)
		}
		q["createdAt"] = dateFilter
	}
	q["$and"] = []bson.M{
		{
			"$or": []bson.M{
				{"maker": a.Hex()},
				{"taker": a.Hex()},
			},
		},
		{
			"$or": []bson.M{
				{"makerExchange": tradeSpec.RelayerAddress.Hex()},
				{"takerExchange": tradeSpec.RelayerAddress.Hex()},
			},
		},
	}
	if tradeSpec.BaseToken != "" {
		q["baseToken"] = tradeSpec.BaseToken
	}
	if tradeSpec.QuoteToken != "" {
		q["quoteToken"] = tradeSpec.QuoteToken
	}
	var res types.TradeRes
	trades := []*types.Trade{}
	c, err := db.GetEx(dao.dbName, dao.collectionName, q, sortedBy, pageOffset, pageSize, &trades)
	if err != nil {
		logger.Error(err)
		return nil, err
	}
	res.Total = c
	res.Trades = trades
	return &res, nil
}