-
Notifications
You must be signed in to change notification settings - Fork 1
/
ethereum_rail.go
70 lines (59 loc) · 2.26 KB
/
ethereum_rail.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
package server
import (
"github.com/diamnet/go/services/bifrost/database"
"github.com/diamnet/go/services/bifrost/ethereum"
"github.com/diamnet/go/services/bifrost/queue"
"github.com/diamnet/go/services/bifrost/sse"
"github.com/diamnet/go/support/errors"
"github.com/diamnet/go/support/log"
)
// onNewEthereumTransaction checks if transaction is valid and adds it to
// the transactions queue for DiamNetAccountConfigurator to consume.
//
// Transaction added to transactions queue should be in a format described in
// queue.Transaction (especialy amounts). Pooling service should not have to deal with any
// conversions.
func (s *Server) onNewEthereumTransaction(transaction ethereum.Transaction) error {
localLog := s.log.WithFields(log.F{"transaction": transaction, "rail": "ethereum"})
localLog.Debug("Processing transaction")
// Let's check if tx is valid first.
// Check if value is above minimum required
if transaction.ValueWei.Cmp(s.minimumValueWei) < 0 {
localLog.Debug("Value is below minimum required amount, skipping")
return nil
}
addressAssociation, err := s.Database.GetAssociationByChainAddress(database.ChainEthereum, transaction.To)
if err != nil {
return errors.Wrap(err, "Error getting association")
}
if addressAssociation == nil {
localLog.Debug("Associated address not found, skipping")
return nil
}
// Add transaction as processing.
processed, err := s.Database.AddProcessedTransaction(database.ChainEthereum, transaction.Hash, transaction.To)
if err != nil {
return err
}
if processed {
localLog.Debug("Transaction already processed, skipping")
return nil
}
// Add tx to the processing queue
queueTx := queue.Transaction{
TransactionID: transaction.Hash,
AssetCode: queue.AssetCodeETH,
// Amount in the base unit of currency.
Amount: transaction.ValueToDiamNet(),
DiamNetPublicKey: addressAssociation.DiamNetPublicKey,
}
err = s.TransactionsQueue.QueueAdd(queueTx)
if err != nil {
return errors.Wrap(err, "Error adding transaction to the processing queue")
}
localLog.Info("Transaction added to transaction queue")
// Broadcast event to address stream
s.SSEServer.BroadcastEvent(transaction.To, sse.TransactionReceivedAddressEvent, nil)
localLog.Info("Transaction processed successfully")
return nil
}