1package l2db
  2
  3import (
  4	"math/big"
  5	"time"
  6
  7	ethCommon "github.com/ethereum/go-ethereum/common"
  8	"github.com/hermeznetwork/hermez-node/common"
  9	"github.com/hermeznetwork/hermez-node/db"
 10	"github.com/hermeznetwork/hermez-node/log"
 11	"github.com/jmoiron/sqlx"
 12
 13	//nolint:errcheck // driver for postgres DB
 14	_ "github.com/lib/pq"
 15	"github.com/russross/meddler"
 16)
 17
 18// TODO(Edu): Check DB consistency while there's concurrent use from Coordinator/TxSelector & API
 19
 20// L2DB stores L2 txs and authorization registers received by the coordinator and keeps them until they are no longer relevant
 21// due to them being forged or invalid after a safety period
 22type L2DB struct {
 23	db           *sqlx.DB
 24	safetyPeriod common.BatchNum
 25	ttl          time.Duration
 26	maxTxs       uint32
 27}
 28
 29// NewL2DB creates a L2DB.
 30// To create it, it's needed db connection, safety period expressed in batches,
 31// maxTxs that the DB should have and TTL (time to live) for pending txs.
 32func NewL2DB(db *sqlx.DB, safetyPeriod common.BatchNum, maxTxs uint32, TTL time.Duration) *L2DB {
 33	return &L2DB{
 34		db:           db,
 35		safetyPeriod: safetyPeriod,
 36		ttl:          TTL,
 37		maxTxs:       maxTxs,
 38	}
 39}
 40
 41// DB returns a pointer to the L2DB.db. This method should be used only for
 42// internal testing purposes.
 43func (l2db *L2DB) DB() *sqlx.DB {
 44	return l2db.db
 45}
 46
 47// AddAccountCreationAuth inserts an account creation authorization into the DB
 48func (l2db *L2DB) AddAccountCreationAuth(auth *common.AccountCreationAuth) error {
 49	// return meddler.Insert(l2db.db, "account_creation_auth", auth)
 50	_, err := l2db.db.Exec(
 51		`INSERT INTO account_creation_auth (eth_addr, bjj, signature)
 52		VALUES ($1, $2, $3);`,
 53		auth.EthAddr, auth.BJJ, auth.Signature,
 54	)
 55	return err
 56}
 57
 58// GetAccountCreationAuth returns an account creation authorization from the DB
 59func (l2db *L2DB) GetAccountCreationAuth(addr ethCommon.Address) (*common.AccountCreationAuth, error) {
 60	auth := new(common.AccountCreationAuth)
 61	return auth, meddler.QueryRow(
 62		l2db.db, auth,
 63		"SELECT * FROM account_creation_auth WHERE eth_addr = $1;",
 64		addr,
 65	)
 66}
 67
 68// GetAccountCreationAuthAPI returns an account creation authorization from the DB
 69func (l2db *L2DB) GetAccountCreationAuthAPI(addr ethCommon.Address) (*AccountCreationAuthAPI, error) {
 70	auth := new(AccountCreationAuthAPI)
 71	return auth, meddler.QueryRow(
 72		l2db.db, auth,
 73		"SELECT * FROM account_creation_auth WHERE eth_addr = $1;",
 74		addr,
 75	)
 76}
 77
 78// AddTx inserts a tx to the pool
 79func (l2db *L2DB) AddTx(tx *PoolL2TxWrite) error {
 80	return meddler.Insert(l2db.db, "tx_pool", tx)
 81}
 82
 83// AddTxTest inserts a tx into the L2DB. This is useful for test purposes,
 84// but in production txs will only be inserted through the API
 85func (l2db *L2DB) AddTxTest(tx *common.PoolL2Tx) error {
 86	// transform tx from *common.PoolL2Tx to PoolL2TxWrite
 87	insertTx := &PoolL2TxWrite{
 88		TxID:      tx.TxID,
 89		FromIdx:   tx.FromIdx,
 90		ToBJJ:     tx.ToBJJ,
 91		TokenID:   tx.TokenID,
 92		Amount:    tx.Amount,
 93		Fee:       tx.Fee,
 94		Nonce:     tx.Nonce,
 95		State:     common.PoolL2TxStatePending,
 96		Signature: tx.Signature,
 97		RqToBJJ:   tx.RqToBJJ,
 98		RqAmount:  tx.RqAmount,
 99		Type:      tx.Type,
100	}
101	if tx.ToIdx != 0 {
102		insertTx.ToIdx = &tx.ToIdx
103	}
104	nilAddr := ethCommon.BigToAddress(big.NewInt(0))
105	if tx.ToEthAddr != nilAddr {
106		insertTx.ToEthAddr = &tx.ToEthAddr
107	}
108	if tx.RqFromIdx != 0 {
109		insertTx.RqFromIdx = &tx.RqFromIdx
110	}
111	if tx.RqToIdx != 0 { // if true, all Rq... fields must be different to nil
112		insertTx.RqToIdx = &tx.RqToIdx
113		insertTx.RqTokenID = &tx.RqTokenID
114		insertTx.RqFee = &tx.RqFee
115		insertTx.RqNonce = &tx.RqNonce
116	}
117	if tx.RqToEthAddr != nilAddr {
118		insertTx.RqToEthAddr = &tx.RqToEthAddr
119	}
120	f := new(big.Float).SetInt(tx.Amount)
121	amountF, _ := f.Float64()
122	insertTx.AmountFloat = amountF
123	// insert tx
124	return meddler.Insert(l2db.db, "tx_pool", insertTx)
125}
126
127// selectPoolTxAPI select part of queries to get PoolL2TxRead
128const selectPoolTxAPI = `SELECT  tx_pool.tx_id, hez_idx(tx_pool.from_idx, token.symbol) AS from_idx, tx_pool.from_eth_addr, 
129tx_pool.from_bjj, hez_idx(tx_pool.to_idx, token.symbol) AS to_idx, tx_pool.to_eth_addr, 
130tx_pool.to_bjj, tx_pool.token_id, tx_pool.amount, tx_pool.fee, tx_pool.nonce, 
131tx_pool.state, tx_pool.signature, tx_pool.timestamp, tx_pool.batch_num, hez_idx(tx_pool.rq_from_idx, token.symbol) AS rq_from_idx, 
132hez_idx(tx_pool.rq_to_idx, token.symbol) AS rq_to_idx, tx_pool.rq_to_eth_addr, tx_pool.rq_to_bjj, tx_pool.rq_token_id, tx_pool.rq_amount, 
133tx_pool.rq_fee, tx_pool.rq_nonce, tx_pool.tx_type, 
134token.item_id AS token_item_id, token.eth_block_num, token.eth_addr, token.name, token.symbol, token.decimals, token.usd, token.usd_update 
135FROM tx_pool INNER JOIN token ON tx_pool.token_id = token.token_id `
136
137// selectPoolTxCommon select part of queries to get common.PoolL2Tx
138const selectPoolTxCommon = `SELECT  tx_pool.tx_id, from_idx, to_idx, tx_pool.to_eth_addr, 
139tx_pool.to_bjj, tx_pool.token_id, tx_pool.amount, tx_pool.fee, tx_pool.nonce, 
140tx_pool.state, tx_pool.signature, tx_pool.timestamp, rq_from_idx, 
141rq_to_idx, tx_pool.rq_to_eth_addr, tx_pool.rq_to_bjj, tx_pool.rq_token_id, tx_pool.rq_amount, 
142tx_pool.rq_fee, tx_pool.rq_nonce, tx_pool.tx_type, 
143fee_percentage(tx_pool.fee::NUMERIC) * token.usd * tx_pool.amount_f AS fee_usd, token.usd_update  
144FROM tx_pool INNER JOIN token ON tx_pool.token_id = token.token_id `
145
146// GetTx  return the specified Tx in common.PoolL2Tx format
147func (l2db *L2DB) GetTx(txID common.TxID) (*common.PoolL2Tx, error) {
148	tx := new(common.PoolL2Tx)
149	return tx, meddler.QueryRow(
150		l2db.db, tx,
151		selectPoolTxCommon+"WHERE tx_id = $1;",
152		txID,
153	)
154}
155
156// GetTxAPI return the specified Tx in PoolTxAPI format
157func (l2db *L2DB) GetTxAPI(txID common.TxID) (*PoolTxAPI, error) {
158	tx := new(PoolTxAPI)
159	return tx, meddler.QueryRow(
160		l2db.db, tx,
161		selectPoolTxAPI+"WHERE tx_id = $1;",
162		txID,
163	)
164}
165
166// GetPendingTxs return all the pending txs of the L2DB, that have a non NULL AbsoluteFee
167func (l2db *L2DB) GetPendingTxs() ([]common.PoolL2Tx, error) {
168	var txs []*common.PoolL2Tx
169	err := meddler.QueryAll(
170		l2db.db, &txs,
171		selectPoolTxCommon+"WHERE state = $1",
172		common.PoolL2TxStatePending,
173	)
174	return db.SlicePtrsToSlice(txs).([]common.PoolL2Tx), err
175}
176
177// StartForging updates the state of the transactions that will begin the forging process.
178// The state of the txs referenced by txIDs will be changed from Pending -> Forging
179func (l2db *L2DB) StartForging(txIDs []common.TxID, batchNum common.BatchNum) error {
180	query, args, err := sqlx.In(
181		`UPDATE tx_pool
182		SET state = ?, batch_num = ?
183		WHERE state = ? AND tx_id IN (?);`,
184		common.PoolL2TxStateForging,
185		batchNum,
186		common.PoolL2TxStatePending,
187		txIDs,
188	)
189	if err != nil {
190		return err
191	}
192	query = l2db.db.Rebind(query)
193	_, err = l2db.db.Exec(query, args...)
194	return err
195}
196
197// DoneForging updates the state of the transactions that have been forged
198// so the state of the txs referenced by txIDs will be changed from Forging -> Forged
199func (l2db *L2DB) DoneForging(txIDs []common.TxID, batchNum common.BatchNum) error {
200	query, args, err := sqlx.In(
201		`UPDATE tx_pool
202		SET state = ?, batch_num = ?
203		WHERE state = ? AND tx_id IN (?);`,
204		common.PoolL2TxStateForged,
205		batchNum,
206		common.PoolL2TxStateForging,
207		txIDs,
208	)
209	if err != nil {
210		return err
211	}
212	query = l2db.db.Rebind(query)
213	_, err = l2db.db.Exec(query, args...)
214	return err
215}
216
217// InvalidateTxs updates the state of the transactions that are invalid.
218// The state of the txs referenced by txIDs will be changed from * -> Invalid
219func (l2db *L2DB) InvalidateTxs(txIDs []common.TxID, batchNum common.BatchNum) error {
220	query, args, err := sqlx.In(
221		`UPDATE tx_pool
222		SET state = ?, batch_num = ?
223		WHERE tx_id IN (?);`,
224		common.PoolL2TxStateInvalid,
225		batchNum,
226		txIDs,
227	)
228	if err != nil {
229		return err
230	}
231	query = l2db.db.Rebind(query)
232	_, err = l2db.db.Exec(query, args...)
233	return err
234}
235
236// CheckNonces invalidate txs with nonces that are smaller or equal than their respective accounts nonces.
237// The state of the affected txs will be changed from Pending -> Invalid
238func (l2db *L2DB) CheckNonces(updatedAccounts []common.Account, batchNum common.BatchNum) (err error) {
239	txn, err := l2db.db.Begin()
240	if err != nil {
241		return err
242	}
243	defer func() {
244		// Rollback the transaction if there was an error.
245		if err != nil {
246			errRollback := txn.Rollback()
247			if errRollback != nil {
248				log.Errorw("Rollback", "err", errRollback)
249			}
250		}
251	}()
252	for i := 0; i < len(updatedAccounts); i++ {
253		_, err = txn.Exec(
254			`UPDATE tx_pool
255			SET state = $1, batch_num = $2
256			WHERE state = $3 AND from_idx = $4 AND nonce <= $5;`,
257			common.PoolL2TxStateInvalid,
258			batchNum,
259			common.PoolL2TxStatePending,
260			updatedAccounts[i].Idx,
261			updatedAccounts[i].Nonce,
262		)
263		if err != nil {
264			return err
265		}
266	}
267	return txn.Commit()
268}
269
270// Reorg updates the state of txs that were updated in a batch that has been discarted due to a blockchain reorg.
271// The state of the affected txs can change form Forged -> Pending or from Invalid -> Pending
272func (l2db *L2DB) Reorg(lastValidBatch common.BatchNum) error {
273	_, err := l2db.db.Exec(
274		`UPDATE tx_pool SET batch_num = NULL, state = $1 
275		WHERE (state = $2 OR state = $3) AND batch_num > $4`,
276		common.PoolL2TxStatePending,
277		common.PoolL2TxStateForged,
278		common.PoolL2TxStateInvalid,
279		lastValidBatch,
280	)
281	return err
282}
283
284// Purge deletes transactions that have been forged or marked as invalid for longer than the safety period
285// it also deletes txs that has been in the L2DB for longer than the ttl if maxTxs has been exceeded
286func (l2db *L2DB) Purge(currentBatchNum common.BatchNum) (err error) {
287	txn, err := l2db.db.Begin()
288	if err != nil {
289		return err
290	}
291	defer func() {
292		// Rollback the transaction if there was an error.
293		if err != nil {
294			errRollback := txn.Rollback()
295			if errRollback != nil {
296				log.Errorw("Rollback", "err", errRollback)
297			}
298		}
299	}()
300	// Delete pending txs that have been in the pool after the TTL if maxTxs is reached
301	now := time.Now().UTC().Unix()
302	_, err = txn.Exec(
303		`DELETE FROM tx_pool WHERE (SELECT count(*) FROM tx_pool) > $1 AND timestamp < $2`,
304		l2db.maxTxs,
305		time.Unix(now-int64(l2db.ttl.Seconds()), 0),
306	)
307	if err != nil {
308		return err
309	}
310	// Delete txs that have been marked as forged / invalid after the safety period
311	_, err = txn.Exec(
312		`DELETE FROM tx_pool 
313		WHERE batch_num < $1 AND (state = $2 OR state = $3)`,
314		currentBatchNum-l2db.safetyPeriod,
315		common.PoolL2TxStateForged,
316		common.PoolL2TxStateInvalid,
317	)
318	if err != nil {
319		return err
320	}
321	return txn.Commit()
322}