1/*
  2Package db have some common utilities shared by db/l2db and db/historydb, the most relevant ones are:
  3- SQL connection utilities
  4- Managing the SQL schema: this is done using migration files placed under db/migrations. The files are executed by
  5order of the file name.
  6- Custom meddlers: used to easily transform struct <==> table
  7*/
  8package db
  9
 10import (
 11	"context"
 12	"database/sql"
 13	"fmt"
 14	"math/big"
 15	"reflect"
 16	"strings"
 17	"time"
 18
 19	"github.com/gobuffalo/packr/v2"
 20	"github.com/hermeznetwork/hermez-node/log"
 21	"github.com/hermeznetwork/tracerr"
 22	"github.com/jmoiron/sqlx"
 23
 24	//nolint:errcheck // driver for postgres DB
 25	_ "github.com/lib/pq"
 26	migrate "github.com/rubenv/sql-migrate"
 27	"github.com/russross/meddler"
 28	"golang.org/x/sync/semaphore"
 29)
 30
 31var migrations *migrate.PackrMigrationSource
 32
 33func init() {
 34	migrations = &migrate.PackrMigrationSource{
 35		Box: packr.New("hermez-db-migrations", "./migrations"),
 36	}
 37	ms, err := migrations.FindMigrations()
 38	if err != nil {
 39		panic(err)
 40	}
 41	if len(ms) == 0 {
 42		panic(fmt.Errorf("no SQL migrations found"))
 43	}
 44}
 45
 46// MigrationsUp runs the SQL migrations Up
 47func MigrationsUp(db *sql.DB) error {
 48	nMigrations, err := migrate.Exec(db, "postgres", migrations, migrate.Up)
 49	if err != nil {
 50		return tracerr.Wrap(err)
 51	}
 52	log.Info("successfully ran ", nMigrations, " migrations Up")
 53	return nil
 54}
 55
 56// MigrationsDown runs the SQL migrations Down
 57func MigrationsDown(db *sql.DB) error {
 58	nMigrations, err := migrate.Exec(db, "postgres", migrations, migrate.Down)
 59	if err != nil {
 60		return tracerr.Wrap(err)
 61	}
 62	log.Info("successfully ran ", nMigrations, " migrations Down")
 63	return nil
 64}
 65
 66// ConnectSQLDB connects to the SQL DB
 67func ConnectSQLDB(port int, host, user, password, name string) (*sqlx.DB, error) {
 68	// Init meddler
 69	initMeddler()
 70	meddler.Default = meddler.PostgreSQL
 71	// Stablish connection
 72	psqlconn := fmt.Sprintf(
 73		"host=%s port=%d user=%s password=%s dbname=%s sslmode=disable",
 74		host,
 75		port,
 76		user,
 77		password,
 78		name,
 79	)
 80	db, err := sqlx.Connect("postgres", psqlconn)
 81	if err != nil {
 82		return nil, tracerr.Wrap(err)
 83	}
 84	return db, nil
 85}
 86
 87// InitSQLDB runs migrations and registers meddlers
 88func InitSQLDB(port int, host, user, password, name string) (*sqlx.DB, error) {
 89	db, err := ConnectSQLDB(port, host, user, password, name)
 90	if err != nil {
 91		return nil, tracerr.Wrap(err)
 92	}
 93	// Run DB migrations
 94	if err := MigrationsUp(db.DB); err != nil {
 95		return nil, tracerr.Wrap(err)
 96	}
 97	return db, nil
 98}
 99
100// APIConnectionController is used to limit the SQL open connections used by the API
101type APIConnectionController struct {
102	smphr   *semaphore.Weighted
103	timeout time.Duration
104}
105
106// NewAPIConnectionController initialize APIConnectionController
107func NewAPIConnectionController(maxConnections int, timeout time.Duration) *APIConnectionController {
108	return &APIConnectionController{
109		smphr:   semaphore.NewWeighted(int64(maxConnections)),
110		timeout: timeout,
111	}
112}
113
114// Acquire reserves a SQL connection. If the connection is not acquired
115// within the timeout, the function will return an error
116func (acc *APIConnectionController) Acquire() (context.CancelFunc, error) {
117	ctx, cancel := context.WithTimeout(context.Background(), acc.timeout) //nolint:govet
118	return cancel, acc.smphr.Acquire(ctx, 1)
119}
120
121// Release frees a SQL connection
122func (acc *APIConnectionController) Release() {
123	acc.smphr.Release(1)
124}
125
126// initMeddler registers tags to be used to read/write from SQL DBs using meddler
127func initMeddler() {
128	meddler.Register("bigint", BigIntMeddler{})
129	meddler.Register("bigintnull", BigIntNullMeddler{})
130}
131
132// BulkInsert performs a bulk insert with a single statement into the specified table.  Example:
133// `db.BulkInsert(myDB, "INSERT INTO block (eth_block_num, timestamp, hash) VALUES %s", blocks[:])`
134// Note that all the columns must be specified in the query, and they must be
135// in the same order as in the table.
136// Note that the fields in the structs need to be defined in the same order as
137// in the table columns.
138func BulkInsert(db meddler.DB, q string, args interface{}) error {
139	arrayValue := reflect.ValueOf(args)
140	arrayLen := arrayValue.Len()
141	valueStrings := make([]string, 0, arrayLen)
142	var arglist = make([]interface{}, 0)
143	for i := 0; i < arrayLen; i++ {
144		arg := arrayValue.Index(i).Addr().Interface()
145		elemArglist, err := meddler.Default.Values(arg, true)
146		if err != nil {
147			return tracerr.Wrap(err)
148		}
149		arglist = append(arglist, elemArglist...)
150		value := "("
151		for j := 0; j < len(elemArglist); j++ {
152			value += fmt.Sprintf("$%d, ", i*len(elemArglist)+j+1)
153		}
154		value = value[:len(value)-2] + ")"
155		valueStrings = append(valueStrings, value)
156	}
157	stmt := fmt.Sprintf(q, strings.Join(valueStrings, ","))
158	_, err := db.Exec(stmt, arglist...)
159	return tracerr.Wrap(err)
160}
161
162// BigIntMeddler encodes or decodes the field value to or from JSON
163type BigIntMeddler struct{}
164
165// PreRead is called before a Scan operation for fields that have the BigIntMeddler
166func (b BigIntMeddler) PreRead(fieldAddr interface{}) (scanTarget interface{}, err error) {
167	// give a pointer to a byte buffer to grab the raw data
168	return new(string), nil
169}
170
171// PostRead is called after a Scan operation for fields that have the BigIntMeddler
172func (b BigIntMeddler) PostRead(fieldPtr, scanTarget interface{}) error {
173	ptr := scanTarget.(*string)
174	if ptr == nil {
175		return tracerr.Wrap(fmt.Errorf("BigIntMeddler.PostRead: nil pointer"))
176	}
177	field := fieldPtr.(**big.Int)
178	var ok bool
179	*field, ok = new(big.Int).SetString(*ptr, 10)
180	if !ok {
181		return tracerr.Wrap(fmt.Errorf("big.Int.SetString failed on \"%v\"", *ptr))
182	}
183	return nil
184}
185
186// PreWrite is called before an Insert or Update operation for fields that have the BigIntMeddler
187func (b BigIntMeddler) PreWrite(fieldPtr interface{}) (saveValue interface{}, err error) {
188	field := fieldPtr.(*big.Int)
189
190	return field.String(), nil
191}
192
193// BigIntNullMeddler encodes or decodes the field value to or from JSON
194type BigIntNullMeddler struct{}
195
196// PreRead is called before a Scan operation for fields that have the BigIntNullMeddler
197func (b BigIntNullMeddler) PreRead(fieldAddr interface{}) (scanTarget interface{}, err error) {
198	return &fieldAddr, nil
199}
200
201// PostRead is called after a Scan operation for fields that have the BigIntNullMeddler
202func (b BigIntNullMeddler) PostRead(fieldPtr, scanTarget interface{}) error {
203	field := fieldPtr.(**big.Int)
204	ptrPtr := scanTarget.(*interface{})
205	if *ptrPtr == nil {
206		// null column, so set target to be zero value
207		*field = nil
208		return nil
209	}
210	// not null
211	ptr := (*ptrPtr).([]byte)
212	if ptr == nil {
213		return tracerr.Wrap(fmt.Errorf("BigIntMeddler.PostRead: nil pointer"))
214	}
215	var ok bool
216	*field, ok = new(big.Int).SetString(string(ptr), 10)
217	if !ok {
218		return tracerr.Wrap(fmt.Errorf("big.Int.SetString failed on \"%v\"", string(ptr)))
219	}
220
221	return nil
222}
223
224// PreWrite is called before an Insert or Update operation for fields that have the BigIntNullMeddler
225func (b BigIntNullMeddler) PreWrite(fieldPtr interface{}) (saveValue interface{}, err error) {
226	field := fieldPtr.(*big.Int)
227	if field == nil {
228		return nil, nil
229	}
230	return field.String(), nil
231}
232
233// SliceToSlicePtrs converts any []Foo to []*Foo
234func SliceToSlicePtrs(slice interface{}) interface{} {
235	v := reflect.ValueOf(slice)
236	vLen := v.Len()
237	typ := v.Type().Elem()
238	res := reflect.MakeSlice(reflect.SliceOf(reflect.PtrTo(typ)), vLen, vLen)
239	for i := 0; i < vLen; i++ {
240		res.Index(i).Set(v.Index(i).Addr())
241	}
242	return res.Interface()
243}
244
245// SlicePtrsToSlice converts any []*Foo to []Foo
246func SlicePtrsToSlice(slice interface{}) interface{} {
247	v := reflect.ValueOf(slice)
248	vLen := v.Len()
249	typ := v.Type().Elem().Elem()
250	res := reflect.MakeSlice(reflect.SliceOf(typ), vLen, vLen)
251	for i := 0; i < vLen; i++ {
252		res.Index(i).Set(v.Index(i).Elem())
253	}
254	return res.Interface()
255}
256
257// Rollback an sql transaction, and log the error if it's not nil
258func Rollback(txn *sqlx.Tx) {
259	if err := txn.Rollback(); err != nil {
260		log.Errorw("Rollback", "err", err)
261	}
262}
263
264// RowsClose close the rows of an sql query, and log the errir if it's not nil
265func RowsClose(rows *sql.Rows) {
266	if err := rows.Close(); err != nil {
267		log.Errorw("rows.Close", "err", err)
268	}
269}