1package l2db
  2
  3import (
  4	"fmt"
  5	"time"
  6
  7	ethCommon "github.com/ethereum/go-ethereum/common"
  8	"github.com/gobuffalo/packr/v2"
  9	"github.com/hermeznetwork/hermez-node/common"
 10	"github.com/hermeznetwork/hermez-node/db"
 11	"github.com/jmoiron/sqlx"
 12
 13	//nolint:errcheck // driver for postgres DB
 14	_ "github.com/lib/pq"
 15	migrate "github.com/rubenv/sql-migrate"
 16	"github.com/russross/meddler"
 17)
 18
 19// TODO(Edu): Check DB consistency while there's concurrent use from Coordinator/TxSelector & API
 20
 21// L2DB stores L2 txs and authorization registers received by the coordinator and keeps them until they are no longer relevant
 22// due to them being forged or invalid after a safety period
 23type L2DB struct {
 24	db           *sqlx.DB
 25	safetyPeriod common.BatchNum
 26	ttl          time.Duration
 27	maxTxs       uint32
 28}
 29
 30// NewL2DB creates a L2DB.
 31// To create it, it's needed postgres configuration, safety period expressed in batches,
 32// maxTxs that the DB should have and TTL (time to live) for pending txs.
 33func NewL2DB(
 34	port int, host, user, password, dbname string,
 35	safetyPeriod common.BatchNum,
 36	maxTxs uint32,
 37	TTL time.Duration,
 38) (*L2DB, error) {
 39	// init meddler
 40	db.InitMeddler()
 41	meddler.Default = meddler.PostgreSQL
 42	// Stablish DB connection
 43	psqlconn := fmt.Sprintf("host=%s port=%d user=%s password=%s dbname=%s sslmode=disable", host, port, user, password, dbname)
 44	db, err := sqlx.Connect("postgres", psqlconn)
 45	if err != nil {
 46		return nil, err
 47	}
 48
 49	// Run DB migrations
 50	migrations := &migrate.PackrMigrationSource{
 51		Box: packr.New("history-migrations", "./migrations"),
 52	}
 53	if _, err := migrate.Exec(db.DB, "postgres", migrations, migrate.Up); err != nil {
 54		return nil, err
 55	}
 56
 57	return &L2DB{
 58		db:           db,
 59		safetyPeriod: safetyPeriod,
 60		ttl:          TTL,
 61		maxTxs:       maxTxs,
 62	}, nil
 63}
 64
 65// DB returns a pointer to the L2DB.db. This method should be used only for
 66// internal testing purposes.
 67func (l2db *L2DB) DB() *sqlx.DB {
 68	return l2db.db
 69}
 70
 71// AddAccountCreationAuth inserts an account creation authorization into the DB
 72func (l2db *L2DB) AddAccountCreationAuth(auth *common.AccountCreationAuth) error {
 73	return meddler.Insert(l2db.db, "account_creation_auth", auth)
 74}
 75
 76// GetAccountCreationAuth returns an account creation authorization into the DB
 77func (l2db *L2DB) GetAccountCreationAuth(addr ethCommon.Address) (*common.AccountCreationAuth, error) {
 78	auth := new(common.AccountCreationAuth)
 79	return auth, meddler.QueryRow(
 80		l2db.db, auth,
 81		"SELECT * FROM account_creation_auth WHERE eth_addr = $1;",
 82		addr,
 83	)
 84}
 85
 86// AddTx inserts a tx into the L2DB
 87func (l2db *L2DB) AddTx(tx *common.PoolL2Tx) error {
 88	return meddler.Insert(l2db.db, "tx_pool", tx)
 89}
 90
 91// GetTx return the specified Tx
 92func (l2db *L2DB) GetTx(txID common.TxID) (*common.PoolL2Tx, error) {
 93	tx := new(common.PoolL2Tx)
 94	return tx, meddler.QueryRow(
 95		l2db.db, tx,
 96		"SELECT * FROM tx_pool WHERE tx_id = $1;",
 97		txID,
 98	)
 99}
100
101// GetPendingTxs return all the pending txs of the L2DB
102func (l2db *L2DB) GetPendingTxs() ([]*common.PoolL2Tx, error) {
103	var txs []*common.PoolL2Tx
104	err := meddler.QueryAll(
105		l2db.db, &txs,
106		"SELECT * FROM tx_pool WHERE state = $1",
107		common.PoolL2TxStatePending,
108	)
109	return txs, err
110}
111
112// StartForging updates the state of the transactions that will begin the forging process.
113// The state of the txs referenced by txIDs will be changed from Pending -> Forging
114func (l2db *L2DB) StartForging(txIDs []common.TxID, batchNum common.BatchNum) error {
115	query, args, err := sqlx.In(
116		`UPDATE tx_pool
117		SET state = ?, batch_num = ?
118		WHERE state = ? AND tx_id IN (?);`,
119		common.PoolL2TxStateForging,
120		batchNum,
121		common.PoolL2TxStatePending,
122		txIDs,
123	)
124	if err != nil {
125		return err
126	}
127	query = l2db.db.Rebind(query)
128	_, err = l2db.db.Exec(query, args...)
129	return err
130}
131
132// DoneForging updates the state of the transactions that have been forged
133// so the state of the txs referenced by txIDs will be changed from Forging -> Forged
134func (l2db *L2DB) DoneForging(txIDs []common.TxID, batchNum common.BatchNum) error {
135	query, args, err := sqlx.In(
136		`UPDATE tx_pool
137		SET state = ?, batch_num = ?
138		WHERE state = ? AND tx_id IN (?);`,
139		common.PoolL2TxStateForged,
140		batchNum,
141		common.PoolL2TxStateForging,
142		txIDs,
143	)
144	if err != nil {
145		return err
146	}
147	query = l2db.db.Rebind(query)
148	_, err = l2db.db.Exec(query, args...)
149	return err
150}
151
152// InvalidateTxs updates the state of the transactions that are invalid.
153// The state of the txs referenced by txIDs will be changed from * -> Invalid
154func (l2db *L2DB) InvalidateTxs(txIDs []common.TxID, batchNum common.BatchNum) error {
155	query, args, err := sqlx.In(
156		`UPDATE tx_pool
157		SET state = ?, batch_num = ?
158		WHERE tx_id IN (?);`,
159		common.PoolL2TxStateInvalid,
160		batchNum,
161		txIDs,
162	)
163	if err != nil {
164		return err
165	}
166	query = l2db.db.Rebind(query)
167	_, err = l2db.db.Exec(query, args...)
168	return err
169}
170
171// CheckNonces invalidate txs with nonces that are smaller or equal than their respective accounts nonces.
172// The state of the affected txs will be changed from Pending -> Invalid
173func (l2db *L2DB) CheckNonces(updatedAccounts []common.Account, batchNum common.BatchNum) error {
174	txn, err := l2db.db.Begin()
175	if err != nil {
176		return err
177	}
178	defer func() {
179		// Rollback the transaction if there was an error.
180		if err != nil {
181			err = txn.Rollback()
182		}
183	}()
184	for i := 0; i < len(updatedAccounts); i++ {
185		_, err = txn.Exec(
186			`UPDATE tx_pool
187			SET state = $1, batch_num = $2
188			WHERE state = $3 AND from_idx = $4 AND nonce <= $5;`,
189			common.PoolL2TxStateInvalid,
190			batchNum,
191			common.PoolL2TxStatePending,
192			updatedAccounts[i].Idx,
193			updatedAccounts[i].Nonce,
194		)
195		if err != nil {
196			return err
197		}
198	}
199	return txn.Commit()
200}
201
202// UpdateTxValue updates the absolute fee and value of txs given a token list that include their price in USD
203func (l2db *L2DB) UpdateTxValue(tokens []common.Token) error {
204	// WARNING: this is very slow and should be optimized
205	txn, err := l2db.db.Begin()
206	if err != nil {
207		return err
208	}
209	defer func() {
210		// Rollback the transaction if there was an error.
211		if err != nil {
212			err = txn.Rollback()
213		}
214	}()
215	now := time.Now()
216	for i := 0; i < len(tokens); i++ {
217		_, err = txn.Exec(
218			`UPDATE tx_pool
219			SET usd_update = $1, value_usd = amount_f * $2, fee_usd = $2 * amount_f * CASE
220				WHEN fee = 0 THEN 0	
221				WHEN fee >= 1 AND fee <= 32 THEN POWER(10,-24+(fee::float/2))	
222				WHEN fee >= 33 AND fee <= 223 THEN POWER(10,-8+(0.041666666666667*(fee::float-32)))	
223				WHEN fee >= 224 AND fee <= 255 THEN POWER(10,fee-224) END
224			WHERE token_id = $3;`,
225			now,
226			tokens[i].USD,
227			tokens[i].TokenID,
228		)
229		if err != nil {
230			return err
231		}
232	}
233	return txn.Commit()
234}
235
236// Reorg updates the state of txs that were updated in a batch that has been discarted due to a blockchain reorg.
237// The state of the affected txs can change form Forged -> Pending or from Invalid -> Pending
238func (l2db *L2DB) Reorg(lastValidBatch common.BatchNum) error {
239	_, err := l2db.db.Exec(
240		`UPDATE tx_pool SET batch_num = NULL, state = $1 
241		WHERE (state = $2 OR state = $3) AND batch_num > $4`,
242		common.PoolL2TxStatePending,
243		common.PoolL2TxStateForged,
244		common.PoolL2TxStateInvalid,
245		lastValidBatch,
246	)
247	return err
248}
249
250// Purge deletes transactions that have been forged or marked as invalid for longer than the safety period
251// it also deletes txs that has been in the L2DB for longer than the ttl if maxTxs has been exceeded
252func (l2db *L2DB) Purge(currentBatchNum common.BatchNum) error {
253	txn, err := l2db.db.Begin()
254	if err != nil {
255		return err
256	}
257	defer func() {
258		// Rollback the transaction if there was an error.
259		if err != nil {
260			err = txn.Rollback()
261		}
262	}()
263	// Delete pending txs that have been in the pool after the TTL if maxTxs is reached
264	now := time.Now().UTC().Unix()
265	_, err = txn.Exec(
266		`DELETE FROM tx_pool WHERE (SELECT count(*) FROM tx_pool) > $1 AND timestamp < $2`,
267		l2db.maxTxs,
268		time.Unix(now-int64(l2db.ttl.Seconds()), 0),
269	)
270	if err != nil {
271		return err
272	}
273	// Delete txs that have been marked as forged / invalid after the safety period
274	_, err = txn.Exec(
275		`DELETE FROM tx_pool 
276		WHERE batch_num < $1 AND (state = $2 OR state = $3)`,
277		currentBatchNum-l2db.safetyPeriod,
278		common.PoolL2TxStateForged,
279		common.PoolL2TxStateInvalid,
280	)
281	if err != nil {
282		return err
283	}
284	return txn.Commit()
285}
286
287// Close frees the resources used by the L2DB
288func (l2db *L2DB) Close() error {
289	return l2db.db.Close()
290}