application/mov/database/mov_store.go

   1 package database
   2
   3 import (
   4         "encoding/binary"
   5         "encoding/json"
   6         "errors"
   7         "math"
   8
   9         "github.com/vapor/application/mov/common"
  10         dbm "github.com/vapor/database/leveldb"
  11         "github.com/vapor/protocol/bc"
  12         "github.com/vapor/protocol/bc/types"
  13 )
  14
  15 // MovStore is the interface for mov's persistent storage
  16 type MovStore interface {
  17         GetMovDatabaseState() (*common.MovDatabaseState, error)
  18         InitDBState(height uint64, hash *bc.Hash) error
  19         ListOrders(orderAfter *common.Order) ([]*common.Order, error)
  20         ListTradePairsWithStart(fromAssetIDAfter, toAssetIDAfter *bc.AssetID) ([]*common.TradePair, error)
  21         ProcessOrders(addOrders []*common.Order, delOrders []*common.Order, blockHeader *types.BlockHeader) error
  22 }
  23
  24 const (
  25         order byte = iota + 1
  26         tradePair
  27         matchStatus
  28
  29         fromAssetIDPos = 0
  30         toAssetIDPos   = 1
  31         assetIDLen     = 32
  32         rateByteLen    = 8
  33
  34         tradePairsNum = 1024
  35         ordersNum     = 4096
  36 )
  37
  38 var (
  39         movStore         = []byte("MOV:")
  40         ordersPrefix     = append(movStore, order)
  41         tradePairsPrefix = append(movStore, tradePair)
  42         bestMatchStore   = append(movStore, matchStatus)
  43 )
  44
  45 func calcOrderKey(fromAssetID, toAssetID *bc.AssetID, utxoHash *bc.Hash, rate float64) []byte {
  46         buf := make([]byte, 8)
  47         binary.BigEndian.PutUint64(buf, math.Float64bits(rate))
  48         key := append(ordersPrefix, fromAssetID.Bytes()...)
  49         key = append(key, toAssetID.Bytes()...)
  50         key = append(key, buf...)
  51         return append(key, utxoHash.Bytes()...)
  52 }
  53
  54 func calcTradePairKey(fromAssetID, toAssetID *bc.AssetID) []byte {
  55         key := append(tradePairsPrefix, fromAssetID.Bytes()...)
  56         return append(key, toAssetID.Bytes()...)
  57 }
  58
  59 func getAssetIDFromTradePairKey(key []byte, posIndex int) *bc.AssetID {
  60         b := [32]byte{}
  61         pos := len(tradePairsPrefix) + assetIDLen*posIndex
  62         copy(b[:], key[pos:pos+assetIDLen])
  63         assetID := bc.NewAssetID(b)
  64         return &assetID
  65 }
  66
  67 func getRateFromOrderKey(key []byte) float64 {
  68         ratePos := len(ordersPrefix) + assetIDLen*2
  69         return math.Float64frombits(binary.BigEndian.Uint64(key[ratePos : ratePos+rateByteLen]))
  70 }
  71
  72 type tradePairData struct {
  73         Count int
  74 }
  75
  76 // LevelDBMovStore is the LevelDB implementation for MovStore
  77 type LevelDBMovStore struct {
  78         db dbm.DB
  79 }
  80
  81 // NewLevelDBMovStore create a new LevelDBMovStore object
  82 func NewLevelDBMovStore(db dbm.DB) *LevelDBMovStore {
  83         return &LevelDBMovStore{db: db}
  84 }
  85
  86 // GetMovDatabaseState return the current DB's image status
  87 func (m *LevelDBMovStore) GetMovDatabaseState() (*common.MovDatabaseState, error) {
  88         if value := m.db.Get(bestMatchStore); value != nil {
  89                 state := &common.MovDatabaseState{}
  90                 return state, json.Unmarshal(value, state)
  91         }
  92
  93         return nil, errors.New("don't find state of mov-database")
  94 }
  95
  96 // InitDBState set the DB's image status
  97 func (m *LevelDBMovStore) InitDBState(height uint64, hash *bc.Hash) error {
  98         state := &common.MovDatabaseState{Height: height, Hash: hash}
  99         value, err := json.Marshal(state)
 100         if err != nil {
 101                 return err
 102         }
 103
 104         m.db.Set(bestMatchStore, value)
 105         return nil
 106 }
 107
 108 // ListOrders return n orders after the input order
 109 func (m *LevelDBMovStore) ListOrders(orderAfter *common.Order) ([]*common.Order, error) {
 110         if orderAfter.FromAssetID == nil || orderAfter.ToAssetID == nil {
 111                 return nil, errors.New("assetID is nil")
 112         }
 113
 114         orderPrefix := append(ordersPrefix, orderAfter.FromAssetID.Bytes()...)
 115         orderPrefix = append(orderPrefix, orderAfter.ToAssetID.Bytes()...)
 116
 117         var startKey []byte
 118
 119         if orderAfter.Rate > 0 {
 120                 startKey = calcOrderKey(orderAfter.FromAssetID, orderAfter.ToAssetID, orderAfter.UTXOHash(), orderAfter.Rate)
 121         }
 122
 123         itr := m.db.IteratorPrefixWithStart(orderPrefix, startKey, false)
 124         defer itr.Release()
 125
 126         var orders []*common.Order
 127         for txNum := 0; txNum < ordersNum && itr.Next(); txNum++ {
 128                 movUtxo := &common.MovUtxo{}
 129                 if err := json.Unmarshal(itr.Value(), movUtxo); err != nil {
 130                         return nil, err
 131                 }
 132
 133                 orders = append(orders, &common.Order{
 134                         FromAssetID: orderAfter.FromAssetID,
 135                         ToAssetID:   orderAfter.ToAssetID,
 136                         Rate:        getRateFromOrderKey(itr.Key()),
 137                         Utxo:        movUtxo,
 138                 })
 139         }
 140         return orders, nil
 141 }
 142
 143 // ListTradePairsWithStart return n trade pairs after the input trade pair
 144 func (m *LevelDBMovStore) ListTradePairsWithStart(fromAssetIDAfter, toAssetIDAfter *bc.AssetID) ([]*common.TradePair, error) {
 145         var startKey []byte
 146         if fromAssetIDAfter != nil && toAssetIDAfter != nil {
 147                 startKey = calcTradePairKey(fromAssetIDAfter, toAssetIDAfter)
 148         }
 149
 150         itr := m.db.IteratorPrefixWithStart(tradePairsPrefix, startKey, false)
 151         defer itr.Release()
 152
 153         var tradePairs []*common.TradePair
 154         for txNum := 0; txNum < tradePairsNum && itr.Next(); txNum++ {
 155                 key := itr.Key()
 156                 fromAssetID := getAssetIDFromTradePairKey(key, fromAssetIDPos)
 157                 toAssetID := getAssetIDFromTradePairKey(key, toAssetIDPos)
 158
 159                 tradePairData := &tradePairData{}
 160                 if err := json.Unmarshal(itr.Value(), tradePairData); err != nil {
 161                         return nil, err
 162                 }
 163
 164                 tradePairs = append(tradePairs, &common.TradePair{FromAssetID: fromAssetID, ToAssetID: toAssetID, Count: tradePairData.Count})
 165         }
 166         return tradePairs, nil
 167 }
 168
 169 // ProcessOrders update the DB's image by add new orders, delete the used order
 170 func (m *LevelDBMovStore) ProcessOrders(addOrders []*common.Order, delOrders []*common.Order, blockHeader *types.BlockHeader) error {
 171         if err := m.checkMovDatabaseState(blockHeader); err != nil {
 172                 return err
 173         }
 174
 175         batch := m.db.NewBatch()
 176         tradePairsCnt := make(map[string]*common.TradePair)
 177         if err := m.addOrders(batch, addOrders, tradePairsCnt); err != nil {
 178                 return err
 179         }
 180
 181         m.deleteOrders(batch, delOrders, tradePairsCnt)
 182         if err := m.updateTradePairs(batch, tradePairsCnt); err != nil {
 183                 return err
 184         }
 185
 186         state, err := m.calcNextDatabaseState(blockHeader)
 187         if err != nil {
 188                 return err
 189         }
 190
 191         if err := m.saveMovDatabaseState(batch, state); err != nil {
 192                 return err
 193         }
 194
 195         batch.Write()
 196         return nil
 197 }
 198
 199 func (m *LevelDBMovStore) addOrders(batch dbm.Batch, orders []*common.Order, tradePairsCnt map[string]*common.TradePair) error {
 200         for _, order := range orders {
 201                 data, err := json.Marshal(order.Utxo)
 202                 if err != nil {
 203                         return err
 204                 }
 205
 206                 key := calcOrderKey(order.FromAssetID, order.ToAssetID, order.UTXOHash(), order.Rate)
 207                 batch.Set(key, data)
 208
 209                 tradePair := &common.TradePair{
 210                         FromAssetID: order.FromAssetID,
 211                         ToAssetID:   order.ToAssetID,
 212                 }
 213                 if _, ok := tradePairsCnt[tradePair.Key()]; !ok {
 214                         tradePairsCnt[tradePair.Key()] = tradePair
 215                 }
 216                 tradePairsCnt[tradePair.Key()].Count++
 217         }
 218         return nil
 219 }
 220
 221 func (m *LevelDBMovStore) calcNextDatabaseState(blockHeader *types.BlockHeader) (*common.MovDatabaseState, error) {
 222         hash := blockHeader.Hash()
 223         height := blockHeader.Height
 224
 225         state, err := m.GetMovDatabaseState()
 226         if err != nil {
 227                 return nil, err
 228         }
 229
 230         if *state.Hash == hash {
 231                 hash = blockHeader.PreviousBlockHash
 232                 height = blockHeader.Height - 1
 233         }
 234
 235         return &common.MovDatabaseState{Height: height, Hash: &hash}, nil
 236 }
 237
 238 func (m *LevelDBMovStore) checkMovDatabaseState(header *types.BlockHeader) error {
 239         state, err := m.GetMovDatabaseState()
 240         if err != nil {
 241                 return err
 242         }
 243
 244         if (*state.Hash == header.PreviousBlockHash && (state.Height+1) == header.Height) || *state.Hash == header.Hash() {
 245                 return nil
 246         }
 247
 248         return errors.New("the status of the block is inconsistent with that of mov-database")
 249 }
 250
 251 func (m *LevelDBMovStore) deleteOrders(batch dbm.Batch, orders []*common.Order, tradePairsCnt map[string]*common.TradePair) {
 252         for _, order := range orders {
 253                 key := calcOrderKey(order.FromAssetID, order.ToAssetID, order.UTXOHash(), order.Rate)
 254                 batch.Delete(key)
 255
 256                 tradePair := &common.TradePair{
 257                         FromAssetID: order.FromAssetID,
 258                         ToAssetID:   order.ToAssetID,
 259                 }
 260                 if _, ok := tradePairsCnt[tradePair.Key()]; !ok {
 261                         tradePairsCnt[tradePair.Key()] = tradePair
 262                 }
 263                 tradePairsCnt[tradePair.Key()].Count--
 264         }
 265 }
 266
 267 func (m *LevelDBMovStore) saveMovDatabaseState(batch dbm.Batch, state *common.MovDatabaseState) error {
 268         value, err := json.Marshal(state)
 269         if err != nil {
 270                 return err
 271         }
 272
 273         batch.Set(bestMatchStore, value)
 274         return nil
 275 }
 276
 277 func (m *LevelDBMovStore) updateTradePairs(batch dbm.Batch, tradePairs map[string]*common.TradePair) error {
 278         for _, v := range tradePairs {
 279                 key := calcTradePairKey(v.FromAssetID, v.ToAssetID)
 280                 tradePairData := &tradePairData{}
 281                 if value := m.db.Get(key); value != nil {
 282                         if err := json.Unmarshal(value, tradePairData); err != nil {
 283                                 return err
 284                         }
 285                 }
 286
 287                 if tradePairData.Count += v.Count; tradePairData.Count < 0 {
 288                         return errors.New("negative trade count")
 289                 }
 290
 291                 if tradePairData.Count > 0 {
 292                         value, err := json.Marshal(tradePairData)
 293                         if err != nil {
 294                                 return err
 295                         }
 296
 297                         batch.Set(key, value)
 298                 } else {
 299                         batch.Delete(key)
 300                 }
 301         }
 302         return nil
 303 }