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