blockchain/account/indexer.go

   1 package account
   2
   3 import (
   4         "context"
   5         "encoding/json"
   6         "time"
   7
   8         "github.com/bytom/blockchain/query"
   9         "github.com/bytom/blockchain/signers"
  10         "github.com/bytom/crypto/sha3pool"
  11         "github.com/bytom/errors"
  12         "github.com/bytom/protocol/bc"
  13         "github.com/bytom/protocol/bc/legacy"
  14
  15         chainjson "github.com/bytom/encoding/json"
  16 )
  17
  18 const (
  19         // PinName is used to identify the pin associated with
  20         // the account indexer block processor.
  21         PinName = "account"
  22         // ExpirePinName is used to identify the pin associated
  23         // with the account control program expiration processor.
  24         ExpirePinName = "expire-control-programs"
  25         // DeleteSpentsPinName is used to identify the pin associated
  26         // with the processor that deletes spent account UTXOs.
  27         DeleteSpentsPinName = "delete-account-spents"
  28 )
  29
  30 type AccountUTXOs struct {
  31         OutputID  []byte
  32         AssetID   []byte
  33         Amount    uint64
  34         AccountID string
  35         CpIndex   uint64
  36         Program   []byte
  37         InBlock   uint64
  38         SourceID  []byte
  39         SourcePos uint64
  40         RefData   []byte
  41         Change    bool
  42 }
  43
  44 var emptyJSONObject = json.RawMessage(`{}`)
  45
  46 // A Saver is responsible for saving an annotated account object.
  47 // for indexing and retrieval.
  48 // If the Core is configured not to provide search services,
  49 // SaveAnnotatedAccount can be a no-op.
  50 type Saver interface {
  51         SaveAnnotatedAccount(context.Context, *query.AnnotatedAccount) error
  52 }
  53
  54 func Annotated(a *Account) (*query.AnnotatedAccount, error) {
  55         aa := &query.AnnotatedAccount{
  56                 ID:     a.ID,
  57                 Alias:  a.Alias,
  58                 Quorum: a.Quorum,
  59                 Tags:   &emptyJSONObject,
  60         }
  61
  62         tags, err := json.Marshal(a.Tags)
  63         if err != nil {
  64                 return nil, err
  65         }
  66         if len(tags) > 0 {
  67                 rawTags := json.RawMessage(tags)
  68                 aa.Tags = &rawTags
  69         }
  70
  71         path := signers.Path(a.Signer, signers.AccountKeySpace)
  72         var jsonPath []chainjson.HexBytes
  73         for _, p := range path {
  74                 jsonPath = append(jsonPath, p)
  75         }
  76         for _, xpub := range a.XPubs {
  77                 aa.Keys = append(aa.Keys, &query.AccountKey{
  78                         RootXPub:              xpub,
  79                         AccountXPub:           xpub.Derive(path),
  80                         AccountDerivationPath: jsonPath,
  81                 })
  82         }
  83         return aa, nil
  84 }
  85
  86 func (m *Manager) indexAnnotatedAccount(ctx context.Context, a *Account) error {
  87         if m.indexer == nil {
  88                 return nil
  89         }
  90         aa, err := Annotated(a)
  91         if err != nil {
  92                 return err
  93         }
  94         return m.indexer.SaveAnnotatedAccount(ctx, aa)
  95 }
  96
  97 type rawOutput struct {
  98         OutputID bc.Hash
  99         bc.AssetAmount
 100         ControlProgram []byte
 101         txHash         bc.Hash
 102         outputIndex    uint32
 103         sourceID       bc.Hash
 104         sourcePos      uint64
 105         refData        bc.Hash
 106 }
 107
 108 type accountOutput struct {
 109         rawOutput
 110         AccountID string
 111         keyIndex  uint64
 112         change    bool
 113 }
 114
 115 func (m *Manager) ProcessBlocks(ctx context.Context) {
 116         if m.pinStore == nil {
 117                 return
 118         }
 119
 120         go m.pinStore.ProcessBlocks(ctx, m.chain, DeleteSpentsPinName, func(ctx context.Context, b *legacy.Block) error {
 121                 <-m.pinStore.PinWaiter(PinName, b.Height)
 122                 return m.deleteSpentOutputs(ctx, b)
 123         })
 124         m.pinStore.ProcessBlocks(ctx, m.chain, PinName, m.indexAccountUTXOs)
 125
 126 }
 127
 128 func (m *Manager) deleteSpentOutputs(ctx context.Context, b *legacy.Block) error {
 129         // Delete consumed account UTXOs.
 130         delOutputIDs := prevoutDBKeys(b.Transactions...)
 131         for _, delOutputID := range delOutputIDs {
 132                 m.pinStore.DB.Delete(json.RawMessage("acu" + string(delOutputID.Bytes())))
 133         }
 134
 135         return errors.Wrap(nil, "deleting spent account utxos")
 136 }
 137
 138 func (m *Manager) indexAccountUTXOs(ctx context.Context, b *legacy.Block) error {
 139         // Upsert any UTXOs belonging to accounts managed by this Core.
 140         outs := make([]*rawOutput, 0, len(b.Transactions))
 141         blockPositions := make(map[bc.Hash]uint32, len(b.Transactions))
 142         for i, tx := range b.Transactions {
 143                 blockPositions[tx.ID] = uint32(i)
 144                 for j, out := range tx.Outputs {
 145                         resOutID := tx.ResultIds[j]
 146                         resOut, ok := tx.Entries[*resOutID].(*bc.Output)
 147                         if !ok {
 148                                 continue
 149                         }
 150                         out := &rawOutput{
 151                                 OutputID:       *tx.OutputID(j),
 152                                 AssetAmount:    out.AssetAmount,
 153                                 ControlProgram: out.ControlProgram,
 154                                 txHash:         tx.ID,
 155                                 outputIndex:    uint32(j),
 156                                 sourceID:       *resOut.Source.Ref,
 157                                 sourcePos:      resOut.Source.Position,
 158                                 refData:        *resOut.Data,
 159                         }
 160                         outs = append(outs, out)
 161                 }
 162         }
 163         accOuts := m.loadAccountInfo(ctx, outs)
 164
 165         err := m.upsertConfirmedAccountOutputs(ctx, accOuts, blockPositions, b)
 166         return errors.Wrap(err, "upserting confirmed account utxos")
 167 }
 168
 169 func prevoutDBKeys(txs ...*legacy.Tx) (outputIDs []bc.Hash) {
 170         for _, tx := range txs {
 171                 for _, inpID := range tx.Tx.InputIDs {
 172                         if sp, err := tx.Spend(inpID); err == nil {
 173                                 outputIDs = append(outputIDs, *sp.SpentOutputId)
 174                         }
 175                 }
 176         }
 177         return
 178 }
 179
 180 // loadAccountInfo turns a set of output IDs into a set of
 181 // outputs by adding account annotations.  Outputs that can't be
 182 // annotated are excluded from the result.
 183 func (m *Manager) loadAccountInfo(ctx context.Context, outs []*rawOutput) []*accountOutput {
 184         outsByScript := make(map[string][]*rawOutput, len(outs))
 185         for _, out := range outs {
 186                 scriptStr := string(out.ControlProgram)
 187                 outsByScript[scriptStr] = append(outsByScript[scriptStr], out)
 188         }
 189
 190         result := make([]*accountOutput, 0, len(outs))
 191         cp := struct {
 192                 AccountID      string
 193                 KeyIndex       uint64
 194                 ControlProgram []byte
 195                 Change         bool
 196                 ExpiresAt      time.Time
 197         }{}
 198
 199         var b32 [32]byte
 200         for s := range outsByScript {
 201                 sha3pool.Sum256(b32[:], []byte(s))
 202                 bytes := m.db.Get(json.RawMessage("acp" + string(b32[:])))
 203                 if len(bytes) == 0 {
 204                         continue
 205                 }
 206
 207                 err := json.Unmarshal(bytes, &cp)
 208                 if err != nil {
 209                         continue
 210                 }
 211
 212                 //filte the accounts which exists in accountdb with wallet enabled
 213                 isExist := m.db.Get(json.RawMessage(cp.AccountID))
 214                 if len(isExist) == 0 {
 215                         continue
 216                 }
 217
 218                 for _, out := range outsByScript[s] {
 219                         newOut := &accountOutput{
 220                                 rawOutput: *out,
 221                                 AccountID: cp.AccountID,
 222                                 keyIndex:  cp.KeyIndex,
 223                                 change:    cp.Change,
 224                         }
 225                         result = append(result, newOut)
 226                 }
 227         }
 228
 229         return result
 230 }
 231
 232 // upsertConfirmedAccountOutputs records the account data for confirmed utxos.
 233 // If the account utxo already exists (because it's from a local tx), the
 234 // block confirmation data will in the row will be updated.
 235 func (m *Manager) upsertConfirmedAccountOutputs(ctx context.Context,
 236         outs []*accountOutput,
 237         pos map[bc.Hash]uint32,
 238         block *legacy.Block) error {
 239
 240         var au *AccountUTXOs
 241         for _, out := range outs {
 242                 au = &AccountUTXOs{OutputID: out.OutputID.Bytes(),
 243                         AssetID:   out.AssetId.Bytes(),
 244                         Amount:    out.Amount,
 245                         AccountID: out.AccountID,
 246                         CpIndex:   out.keyIndex,
 247                         Program:   out.ControlProgram,
 248                         InBlock:   block.Height,
 249                         SourceID:  out.sourceID.Bytes(),
 250                         SourcePos: out.sourcePos,
 251                         RefData:   out.refData.Bytes(),
 252                         Change:    out.change}
 253
 254                 accountutxo, err := json.Marshal(au)
 255                 if err != nil {
 256                         return errors.Wrap(err, "failed marshal accountutxo")
 257                 }
 258
 259                 if len(accountutxo) > 0 {
 260                         m.pinStore.DB.Set(json.RawMessage("acu"+string(au.OutputID)), accountutxo)
 261                 }
 262
 263         }
 264
 265         return nil
 266 }