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