blockchain/wallet/indexer.go

   1 package wallet
   2
   3 import (
   4         "encoding/json"
   5         "fmt"
   6
   7         log "github.com/sirupsen/logrus"
   8         "github.com/tendermint/tmlibs/db"
   9
  10         "github.com/bytom/blockchain/account"
  11         "github.com/bytom/blockchain/query"
  12         "github.com/bytom/crypto/sha3pool"
  13         "github.com/bytom/errors"
  14         "github.com/bytom/protocol/bc"
  15         "github.com/bytom/protocol/bc/legacy"
  16 )
  17
  18 type rawOutput struct {
  19         OutputID bc.Hash
  20         bc.AssetAmount
  21         ControlProgram []byte
  22         txHash         bc.Hash
  23         outputIndex    uint32
  24         sourceID       bc.Hash
  25         sourcePos      uint64
  26         refData        bc.Hash
  27 }
  28
  29 type accountOutput struct {
  30         rawOutput
  31         AccountID string
  32         Address   string
  33         keyIndex  uint64
  34         change    bool
  35 }
  36
  37 const (
  38         //TxPrefix is wallet database transactions prefix
  39         TxPrefix = "TXS:"
  40         //TxIndex is wallet database tx index prefix
  41         TxIndex = "TID:"
  42 )
  43
  44 func formatKey(blockHeight uint64, position uint32) string {
  45         return fmt.Sprintf("%016x%08x", blockHeight, position)
  46 }
  47
  48 func calcAnnotatedKey(formatKey string) []byte {
  49         return []byte(TxPrefix + formatKey)
  50 }
  51
  52 func calcDeleteKey(blockHeight uint64) []byte {
  53         return []byte(fmt.Sprintf("%s%016x", TxPrefix, blockHeight))
  54 }
  55
  56 func calcTxIndexKey(txID string) []byte {
  57         return []byte(TxIndex + txID)
  58 }
  59
  60 //deleteTransaction delete transactions when orphan block rollback
  61 func (w *Wallet) deleteTransactions(batch db.Batch, height uint64) {
  62         tmpTx := query.AnnotatedTx{}
  63
  64         txIter := w.DB.IteratorPrefix(calcDeleteKey(height))
  65         defer txIter.Release()
  66
  67         for txIter.Next() {
  68                 if err := json.Unmarshal(txIter.Value(), &tmpTx); err == nil {
  69                         //delete index
  70                         batch.Delete(calcTxIndexKey(tmpTx.ID.String()))
  71                 }
  72
  73                 batch.Delete(txIter.Key())
  74         }
  75 }
  76
  77 //ReverseAccountUTXOs process the invalid blocks when orphan block rollback
  78 func (w *Wallet) reverseAccountUTXOs(batch db.Batch, b *legacy.Block) {
  79         var err error
  80
  81         //unknow how many spent and retire outputs
  82         reverseOuts := make([]*rawOutput, 0)
  83
  84         //handle spent UTXOs
  85         for _, tx := range b.Transactions {
  86                 for _, inpID := range tx.Tx.InputIDs {
  87                         //spend and retire
  88                         sp, err := tx.Spend(inpID)
  89                         if err != nil {
  90                                 continue
  91                         }
  92
  93                         resOut, ok := tx.Entries[*sp.SpentOutputId].(*bc.Output)
  94                         if !ok {
  95                                 continue
  96                         }
  97
  98                         out := &rawOutput{
  99                                 OutputID:       *sp.SpentOutputId,
 100                                 AssetAmount:    *resOut.Source.Value,
 101                                 ControlProgram: resOut.ControlProgram.Code,
 102                                 txHash:         tx.ID,
 103                                 sourceID:       *resOut.Source.Ref,
 104                                 sourcePos:      resOut.Source.Position,
 105                                 refData:        *resOut.Data,
 106                         }
 107                         reverseOuts = append(reverseOuts, out)
 108                 }
 109         }
 110
 111         accOuts := loadAccountInfo(reverseOuts, w)
 112         if err = upsertConfirmedAccountOutputs(accOuts, batch); err != nil {
 113                 log.WithField("err", err).Error("reversing account spent and retire outputs")
 114                 return
 115         }
 116
 117         //handle new UTXOs
 118         for _, tx := range b.Transactions {
 119                 for j := range tx.Outputs {
 120                         resOutID := tx.ResultIds[j]
 121                         if _, ok := tx.Entries[*resOutID].(*bc.Output); !ok {
 122                                 //retirement
 123                                 continue
 124                         }
 125                         //delete new UTXOs
 126                         batch.Delete(account.UTXOKey(*resOutID))
 127                 }
 128         }
 129 }
 130
 131 //indexTransactions saves all annotated transactions to the database.
 132 func (w *Wallet) indexTransactions(batch db.Batch, b *legacy.Block) error {
 133         annotatedTxs := filterAccountTxs(b, w)
 134         annotateTxsAsset(annotatedTxs, w.DB)
 135         annotateTxsAccount(annotatedTxs, w.DB)
 136
 137         for _, tx := range annotatedTxs {
 138                 rawTx, err := json.Marshal(tx)
 139                 if err != nil {
 140                         return errors.Wrap(err, "inserting annotated_txs to db")
 141                 }
 142
 143                 batch.Set(calcAnnotatedKey(formatKey(b.Height, uint32(tx.Position))), rawTx)
 144                 batch.Set(calcTxIndexKey(tx.ID.String()), []byte(formatKey(b.Height, uint32(tx.Position))))
 145         }
 146         return nil
 147 }
 148
 149 //buildAccountUTXOs process valid blocks to build account unspent outputs db
 150 func (w *Wallet) buildAccountUTXOs(batch db.Batch, b *legacy.Block) {
 151         var err error
 152
 153         //handle spent UTXOs
 154         delOutputIDs := prevoutDBKeys(b.Transactions...)
 155         for _, delOutputID := range delOutputIDs {
 156                 batch.Delete(account.UTXOKey(delOutputID))
 157         }
 158
 159         //handle new UTXOs
 160         outs := make([]*rawOutput, 0, len(b.Transactions))
 161         for _, tx := range b.Transactions {
 162                 for j, out := range tx.Outputs {
 163                         resOutID := tx.ResultIds[j]
 164                         resOut, ok := tx.Entries[*resOutID].(*bc.Output)
 165                         if !ok {
 166                                 continue
 167                         }
 168                         out := &rawOutput{
 169                                 OutputID:       *tx.OutputID(j),
 170                                 AssetAmount:    out.AssetAmount,
 171                                 ControlProgram: out.ControlProgram,
 172                                 txHash:         tx.ID,
 173                                 outputIndex:    uint32(j),
 174                                 sourceID:       *resOut.Source.Ref,
 175                                 sourcePos:      resOut.Source.Position,
 176                                 refData:        *resOut.Data,
 177                         }
 178                         outs = append(outs, out)
 179                 }
 180         }
 181         accOuts := loadAccountInfo(outs, w)
 182
 183         if err = upsertConfirmedAccountOutputs(accOuts, batch); err != nil {
 184                 log.WithField("err", err).Error("building new account outputs")
 185                 return
 186         }
 187 }
 188
 189 func prevoutDBKeys(txs ...*legacy.Tx) (outputIDs []bc.Hash) {
 190         for _, tx := range txs {
 191                 for _, inpID := range tx.Tx.InputIDs {
 192                         if sp, err := tx.Spend(inpID); err == nil {
 193                                 outputIDs = append(outputIDs, *sp.SpentOutputId)
 194                         }
 195                 }
 196         }
 197         return
 198 }
 199
 200 // loadAccountInfo turns a set of output IDs into a set of
 201 // outputs by adding account annotations.  Outputs that can't be
 202 // annotated are excluded from the result.
 203 func loadAccountInfo(outs []*rawOutput, w *Wallet) []*accountOutput {
 204         outsByScript := make(map[string][]*rawOutput, len(outs))
 205         for _, out := range outs {
 206                 scriptStr := string(out.ControlProgram)
 207                 outsByScript[scriptStr] = append(outsByScript[scriptStr], out)
 208         }
 209
 210         result := make([]*accountOutput, 0, len(outs))
 211         cp := account.CtrlProgram{}
 212
 213         var hash [32]byte
 214         for s := range outsByScript {
 215                 sha3pool.Sum256(hash[:], []byte(s))
 216                 bytes := w.DB.Get(account.CPKey(hash))
 217                 if bytes == nil {
 218                         continue
 219                 }
 220
 221                 err := json.Unmarshal(bytes, &cp)
 222                 if err != nil {
 223                         continue
 224                 }
 225
 226                 isExist := w.DB.Get(account.Key(cp.AccountID))
 227                 if isExist == nil {
 228                         continue
 229                 }
 230
 231                 for _, out := range outsByScript[s] {
 232                         newOut := &accountOutput{
 233                                 rawOutput: *out,
 234                                 AccountID: cp.AccountID,
 235                                 Address:   cp.Address,
 236                                 keyIndex:  cp.KeyIndex,
 237                                 change:    cp.Change,
 238                         }
 239                         result = append(result, newOut)
 240                 }
 241         }
 242
 243         return result
 244 }
 245
 246 // upsertConfirmedAccountOutputs records the account data for confirmed utxos.
 247 // If the account utxo already exists (because it's from a local tx), the
 248 // block confirmation data will in the row will be updated.
 249 func upsertConfirmedAccountOutputs(outs []*accountOutput, batch db.Batch) error {
 250         var u *account.UTXO
 251
 252         for _, out := range outs {
 253                 u = &account.UTXO{
 254                         OutputID:            out.OutputID,
 255                         SourceID:            out.sourceID,
 256                         AssetID:             *out.AssetId,
 257                         Amount:              out.Amount,
 258                         SourcePos:           out.sourcePos,
 259                         ControlProgram:      out.ControlProgram,
 260                         RefDataHash:         out.refData,
 261                         ControlProgramIndex: out.keyIndex,
 262                         AccountID:           out.AccountID,
 263                         Address:             out.Address,
 264                 }
 265
 266                 data, err := json.Marshal(u)
 267                 if err != nil {
 268                         return errors.Wrap(err, "failed marshal accountutxo")
 269                 }
 270                 batch.Set(account.UTXOKey(out.OutputID), data)
 271         }
 272         return nil
 273 }
 274
 275 // filt related and build the fully annotated transactions.
 276 func filterAccountTxs(b *legacy.Block, w *Wallet) []*query.AnnotatedTx {
 277         annotatedTxs := make([]*query.AnnotatedTx, 0, len(b.Transactions))
 278         for pos, tx := range b.Transactions {
 279                 local := false
 280                 for _, v := range tx.Outputs {
 281                         var hash [32]byte
 282
 283                         sha3pool.Sum256(hash[:], v.ControlProgram)
 284                         if bytes := w.DB.Get(account.CPKey(hash)); bytes != nil {
 285                                 annotatedTxs = append(annotatedTxs, buildAnnotatedTransaction(tx, b, uint32(pos)))
 286                                 local = true
 287                                 break
 288                         }
 289                 }
 290
 291                 if local == true {
 292                         continue
 293                 }
 294
 295                 for _, v := range tx.Inputs {
 296                         outid, err := v.SpentOutputID()
 297                         if err != nil {
 298                                 continue
 299                         }
 300                         if bytes := w.DB.Get(account.UTXOKey(outid)); bytes != nil {
 301                                 annotatedTxs = append(annotatedTxs, buildAnnotatedTransaction(tx, b, uint32(pos)))
 302                                 break
 303                         }
 304                 }
 305         }
 306
 307         return annotatedTxs
 308 }
 309
 310 //GetTransactionsByTxID get account txs by account tx ID
 311 func (w *Wallet) GetTransactionsByTxID(txID string) ([]query.AnnotatedTx, error) {
 312         annotatedTx := query.AnnotatedTx{}
 313         annotatedTxs := make([]query.AnnotatedTx, 0)
 314         formatKey := ""
 315
 316         if txID != "" {
 317                 rawFormatKey := w.DB.Get(calcTxIndexKey(txID))
 318                 if rawFormatKey == nil {
 319                         return nil, fmt.Errorf("No transaction(txid=%s)", txID)
 320                 }
 321                 formatKey = string(rawFormatKey)
 322         }
 323
 324         txIter := w.DB.IteratorPrefix([]byte(TxPrefix + formatKey))
 325         defer txIter.Release()
 326         for txIter.Next() {
 327                 if err := json.Unmarshal(txIter.Value(), &annotatedTx); err != nil {
 328                         return nil, err
 329                 }
 330                 annotatedTxs = append(annotatedTxs, annotatedTx)
 331         }
 332
 333         return annotatedTxs, nil
 334 }
 335
 336 func findTransactionsByAccount(annotatedTx query.AnnotatedTx, accountID string) bool {
 337         for _, input := range annotatedTx.Inputs {
 338                 if input.AccountID == accountID {
 339                         return true
 340                 }
 341         }
 342
 343         for _, output := range annotatedTx.Outputs {
 344                 if output.AccountID == accountID {
 345                         return true
 346                 }
 347         }
 348
 349         return false
 350 }
 351
 352 //GetTransactionsByAccountID get account txs by account ID
 353 func (w *Wallet) GetTransactionsByAccountID(accountID string) ([]query.AnnotatedTx, error) {
 354         annotatedTxs := make([]query.AnnotatedTx, 0)
 355
 356         txIter := w.DB.IteratorPrefix([]byte(TxPrefix))
 357         defer txIter.Release()
 358         for txIter.Next() {
 359                 annotatedTx := query.AnnotatedTx{}
 360                 if err := json.Unmarshal(txIter.Value(), &annotatedTx); err != nil {
 361                         return nil, err
 362                 }
 363
 364                 if findTransactionsByAccount(annotatedTx, accountID) {
 365                         annotatedTxs = append(annotatedTxs, annotatedTx)
 366                 }
 367         }
 368
 369         return annotatedTxs, nil
 370 }
 371
 372 //GetAccountUTXOs return all account unspent outputs
 373 func (w *Wallet) GetAccountUTXOs(id string) ([]account.UTXO, error) {
 374         accountUTXO := account.UTXO{}
 375         accountUTXOs := make([]account.UTXO, 0)
 376
 377         accountUTXOIter := w.DB.IteratorPrefix([]byte(account.UTXOPreFix + id))
 378         defer accountUTXOIter.Release()
 379         for accountUTXOIter.Next() {
 380                 if err := json.Unmarshal(accountUTXOIter.Value(), &accountUTXO); err != nil {
 381                         hashKey := accountUTXOIter.Key()[len(account.UTXOPreFix):]
 382                         log.WithField("UTXO hash", string(hashKey)).Warn("get account UTXO")
 383                         continue
 384                 }
 385
 386                 accountUTXOs = append(accountUTXOs, accountUTXO)
 387         }
 388
 389         return accountUTXOs, nil
 390 }