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