GlcoinCore/lightning-terminal
1
0
Fork 0
mirror of https://github.com/lightninglabs/lightning-terminal.git synced 2026-08-13 12:33:36 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

multi: remove unused db code

Browse source 
As we've now switched over to using sqldb v2 for most of the db objects,
we can remove a lot of deprecated code that's no longer used in the litd
project. This commit removes that code.
This commit is contained in:
Viktor Torstensson 2025-07-24 01:46:27 +02:00
parent 297d203db2
commit 03f4261714
No known key found for this signature in database
GPG key ID: 961CC8259AE675D4
5 changed files with 0 additions and 1026 deletions
Download patch file Download diff file Expand all files Collapse all files

306
db/interfaces.go
View file

@ -1,311 +1,5 @@
package db
import (
"context"
"database/sql"
"math"
prand "math/rand"
"time"
"github.com/lightninglabs/lightning-terminal/db/sqlc"
"github.com/lightningnetwork/lnd/sqldb/v2"
)
var (
// DefaultStoreTimeout is the default timeout used for any interaction
// with the storage/database.
DefaultStoreTimeout = time.Second * 10
)
const (
// DefaultNumTxRetries is the default number of times we'll retry a
// transaction if it fails with an error that permits transaction
// repetition.
DefaultNumTxRetries = 10
// DefaultInitialRetryDelay is the default initial delay between
// retries. This will be used to generate a random delay between -50%
// and +50% of this value, so 20 to 60 milliseconds. The retry will be
// doubled after each attempt until we reach DefaultMaxRetryDelay. We
// start with a random value to avoid multiple goroutines that are
// created at the same time to effectively retry at the same time.
DefaultInitialRetryDelay = time.Millisecond * 40
// DefaultMaxRetryDelay is the default maximum delay between retries.
DefaultMaxRetryDelay = time.Second * 3
)
// TxOptions represents a set of options one can use to control what type of
// database transaction is created. Transaction can wither be read or write.
type TxOptions interface {
// ReadOnly returns true if the transaction should be read only.
ReadOnly() bool
}
// BatchedTx is a generic interface that represents the ability to execute
// several operations to a given storage interface in a single atomic
// transaction. Typically, Q here will be some subset of the main sqlc.Querier
// interface allowing it to only depend on the routines it needs to implement
// any additional business logic.
type BatchedTx[Q any] interface {
// ExecTx will execute the passed txBody, operating upon generic
// parameter Q (usually a storage interface) in a single transaction.
// The set of TxOptions are passed in in order to allow the caller to
// specify if a transaction should be read-only and optionally what
// type of concurrency control should be used.
ExecTx(ctx context.Context, txOptions TxOptions,
txBody func(Q) error) error
// Backend returns the type of the database backend used.
Backend() sqldb.BackendType
}
// Tx represents a database transaction that can be committed or rolled back.
type Tx interface {
// Commit commits the database transaction, an error should be returned
// if the commit isn't possible.
Commit() error
// Rollback rolls back an incomplete database transaction.
// Transactions that were able to be committed can still call this as a
// noop.
Rollback() error
}
// QueryCreator is a generic function that's used to create a Querier, which is
// a type of interface that implements storage related methods from a database
// transaction. This will be used to instantiate an object callers can use to
// apply multiple modifications to an object interface in a single atomic
// transaction.
type QueryCreator[Q any] func(*sql.Tx) Q
// BatchedQuerier is a generic interface that allows callers to create a new
// database transaction based on an abstract type that implements the TxOptions
// interface.
type BatchedQuerier interface {
// Querier is the underlying query source, this is in place so we can
// pass a BatchedQuerier implementation directly into objects that
// create a batched version of the normal methods they need.
sqlc.Querier
// CustomQueries is the set of custom queries that we have manually
// defined in addition to the ones generated by sqlc.
sqlc.CustomQueries
// BeginTx creates a new database transaction given the set of
// transaction options.
BeginTx(ctx context.Context, options TxOptions) (*sql.Tx, error)
}
// txExecutorOptions is a struct that holds the options for the transaction
// executor. This can be used to do things like retry a transaction due to an
// error a certain amount of times.
type txExecutorOptions struct {
numRetries int
initialRetryDelay time.Duration
maxRetryDelay time.Duration
}
// defaultTxExecutorOptions returns the default options for the transaction
// executor.
func defaultTxExecutorOptions() *txExecutorOptions {
return &txExecutorOptions{
numRetries: DefaultNumTxRetries,
initialRetryDelay: DefaultInitialRetryDelay,
maxRetryDelay: DefaultMaxRetryDelay,
}
}
// randRetryDelay returns a random retry delay between -50% and +50%
// of the configured delay that is doubled for each attempt and capped at a max
// value.
func (t *txExecutorOptions) randRetryDelay(attempt int) time.Duration {
halfDelay := t.initialRetryDelay / 2
randDelay := prand.Int63n(int64(t.initialRetryDelay)) //nolint:gosec
// 50% plus 0%-100% gives us the range of 50%-150%.
initialDelay := halfDelay + time.Duration(randDelay)
// If this is the first attempt, we just return the initial delay.
if attempt == 0 {
return initialDelay
}
// For each subsequent delay, we double the initial delay. This still
// gives us a somewhat random delay, but it still increases with each
// attempt. If we double something n times, that's the same as
// multiplying the value with 2^n. We limit the power to 32 to avoid
// overflows.
factor := time.Duration(math.Pow(2, math.Min(float64(attempt), 32)))
actualDelay := initialDelay * factor
// Cap the delay at the maximum configured value.
if actualDelay > t.maxRetryDelay {
return t.maxRetryDelay
}
return actualDelay
}
// TxExecutorOption is a functional option that allows us to pass in optional
// argument when creating the executor.
type TxExecutorOption func(*txExecutorOptions)
// WithTxRetries is a functional option that allows us to specify the number of
// times a transaction should be retried if it fails with a repeatable error.
func WithTxRetries(numRetries int) TxExecutorOption {
return func(o *txExecutorOptions) {
o.numRetries = numRetries
}
}
// WithTxRetryDelay is a functional option that allows us to specify the delay
// to wait before a transaction is retried.
func WithTxRetryDelay(delay time.Duration) TxExecutorOption {
return func(o *txExecutorOptions) {
o.initialRetryDelay = delay
}
}
// TransactionExecutor is a generic struct that abstracts away from the type of
// query a type needs to run under a database transaction, and also the set of
// options for that transaction. The QueryCreator is used to create a query
// given a database transaction created by the BatchedQuerier.
type TransactionExecutor[Query any] struct {
BatchedQuerier
createQuery QueryCreator[Query]
opts *txExecutorOptions
}
// NewTransactionExecutor creates a new instance of a TransactionExecutor given
// a Querier query object and a concrete type for the type of transactions the
// Querier understands.
func NewTransactionExecutor[Querier any](db BatchedQuerier,
createQuery QueryCreator[Querier],
opts ...TxExecutorOption) *TransactionExecutor[Querier] {
txOpts := defaultTxExecutorOptions()
for _, optFunc := range opts {
optFunc(txOpts)
}
return &TransactionExecutor[Querier]{
BatchedQuerier: db,
createQuery: createQuery,
opts: txOpts,
}
}
// ExecTx is a wrapper for txBody to abstract the creation and commit of a db
// transaction. The db transaction is embedded in a `*Queries` that txBody
// needs to use when executing each one of the queries that need to be applied
// atomically. This can be used by other storage interfaces to parameterize the
// type of query and options run, in order to have access to batched operations
// related to a storage object.
func (t *TransactionExecutor[Q]) ExecTx(ctx context.Context,
txOptions TxOptions, txBody func(Q) error) error {
waitBeforeRetry := func(attemptNumber int) {
retryDelay := t.opts.randRetryDelay(attemptNumber)
log.Tracef("Retrying transaction due to tx serialization or "+
"deadlock error, attempt_number=%v, delay=%v",
attemptNumber, retryDelay)
// Before we try again, we'll wait with a random backoff based
// on the retry delay.
time.Sleep(retryDelay)
}
for i := 0; i < t.opts.numRetries; i++ {
// Create the db transaction.
tx, err := t.BatchedQuerier.BeginTx(ctx, txOptions)
if err != nil {
dbErr := MapSQLError(err)
if IsSerializationOrDeadlockError(dbErr) {
// Nothing to roll back here, since we didn't
// even get a transaction yet.
waitBeforeRetry(i)
continue
}
return dbErr
}
// Rollback is safe to call even if the tx is already closed,
// so if the tx commits successfully, this is a no-op.
defer func() {
_ = tx.Rollback()
}()
if err := txBody(t.createQuery(tx)); err != nil {
dbErr := MapSQLError(err)
if IsSerializationOrDeadlockError(dbErr) {
// Roll back the transaction, then pop back up
// to try once again.
_ = tx.Rollback()
waitBeforeRetry(i)
continue
}
return dbErr
}
// Commit transaction.
if err = tx.Commit(); err != nil {
dbErr := MapSQLError(err)
if IsSerializationOrDeadlockError(dbErr) {
// Roll back the transaction, then pop back up
// to try once again.
_ = tx.Rollback()
waitBeforeRetry(i)
continue
}
return dbErr
}
return nil
}
// If we get to this point, then we weren't able to successfully commit
// a tx given the max number of retries.
return ErrRetriesExceeded
}
// Backend returns the type of the database backend used.
func (t *TransactionExecutor[Q]) Backend() sqldb.BackendType {
return t.BatchedQuerier.Backend()
}
// BaseDB is the base database struct that each implementation can embed to
// gain some common functionality.
type BaseDB struct {
*sql.DB
*sqlc.Queries
}
// BeginTx wraps the normal sql specific BeginTx method with the TxOptions
// interface. This interface is then mapped to the concrete sql tx options
// struct.
func (s *BaseDB) BeginTx(ctx context.Context, opts TxOptions) (*sql.Tx, error) {
sqlOptions := sql.TxOptions{
ReadOnly: opts.ReadOnly(),
Isolation: sql.LevelSerializable,
}
return s.DB.BeginTx(ctx, &sqlOptions)
}
// Backend returns the type of the database backend used.
func (s *BaseDB) Backend() sqldb.BackendType {
return s.Queries.Backend()
}
// QueriesTxOptions defines the set of db txn options the SQLQueries
// understands.
type QueriesTxOptions struct {

271
db/migrations.go
View file

@ -1,21 +1,5 @@
package db
import (
"bytes"
"errors"
"fmt"
"io"
"io/fs"
"net/http"
"strings"
"github.com/btcsuite/btclog/v2"
"github.com/golang-migrate/migrate/v4"
"github.com/golang-migrate/migrate/v4/database"
"github.com/golang-migrate/migrate/v4/source/httpfs"
"github.com/lightninglabs/taproot-assets/fn"
)
const (
// LatestMigrationVersion is the latest migration version of the
// database. This is used to implement downgrade protection for the
@ -24,258 +8,3 @@ const (
// NOTE: This MUST be updated when a new migration is added.
LatestMigrationVersion = 5
)
// MigrationTarget is a functional option that can be passed to applyMigrations
// to specify a target version to migrate to. `currentDbVersion` is the current
// (migration) version of the database, or None if unknown.
// `maxMigrationVersion` is the maximum migration version known to the driver,
// or None if unknown.
type MigrationTarget func(mig *migrate.Migrate,
currentDbVersion int, maxMigrationVersion uint) error
var (
// TargetLatest is a MigrationTarget that migrates to the latest
// version available.
TargetLatest = func(mig *migrate.Migrate, _ int, _ uint) error {
return mig.Up()
}
// TargetVersion is a MigrationTarget that migrates to the given
// version.
TargetVersion = func(version uint) MigrationTarget {
return func(mig *migrate.Migrate, _ int, _ uint) error {
return mig.Migrate(version)
}
}
)
var (
// ErrMigrationDowngrade is returned when a database downgrade is
// detected.
ErrMigrationDowngrade = errors.New("database downgrade detected")
)
// migrationOption is a functional option that can be passed to migrate related
// methods to modify their behavior.
type migrateOptions struct {
latestVersion fn.Option[uint]
}
// defaultMigrateOptions returns a new migrateOptions instance with default
// settings.
func defaultMigrateOptions() *migrateOptions {
return &migrateOptions{}
}
// MigrateOpt is a functional option that can be passed to migrate related
// methods to modify behavior.
type MigrateOpt func(*migrateOptions)
// WithLatestVersion allows callers to override the default latest version
// setting.
func WithLatestVersion(version uint) MigrateOpt {
return func(o *migrateOptions) {
o.latestVersion = fn.Some(version)
}
}
// migrationLogger is a logger that wraps the passed btclog.Logger so it can be
// used to log migrations.
type migrationLogger struct {
log btclog.Logger
}
// Printf is like fmt.Printf. We map this to the target logger based on the
// current log level.
func (m *migrationLogger) Printf(format string, v ...interface{}) {
// Trim trailing newlines from the format.
format = strings.TrimRight(format, "\n")
switch m.log.Level() {
case btclog.LevelTrace:
m.log.Tracef(format, v...)
case btclog.LevelDebug:
m.log.Debugf(format, v...)
case btclog.LevelInfo:
m.log.Infof(format, v...)
case btclog.LevelWarn:
m.log.Warnf(format, v...)
case btclog.LevelError:
m.log.Errorf(format, v...)
case btclog.LevelCritical:
m.log.Criticalf(format, v...)
case btclog.LevelOff:
}
}
// Verbose should return true when verbose logging output is wanted
func (m *migrationLogger) Verbose() bool {
return m.log.Level() <= btclog.LevelDebug
}
// applyMigrations executes database migration files found in the given file
// system under the given path, using the passed database driver and database
// name, up to or down to the given target version.
func applyMigrations(fs fs.FS, driver database.Driver, path, dbName string,
targetVersion MigrationTarget, opts *migrateOptions) error {
// With the migrate instance open, we'll create a new migration source
// using the embedded file system stored in sqlSchemas. The library
// we're using can't handle a raw file system interface, so we wrap it
// in this intermediate layer.
migrateFileServer, err := httpfs.New(http.FS(fs), path)
if err != nil {
return err
}
// Finally, we'll run the migration with our driver above based on the
// open DB, and also the migration source stored in the file system
// above.
sqlMigrate, err := migrate.NewWithInstance(
"migrations", migrateFileServer, dbName, driver,
)
if err != nil {
return err
}
migrationVersion, _, _ := sqlMigrate.Version()
// As the down migrations may end up *dropping* data, we want to
// prevent that without explicit accounting.
latestVersion := opts.latestVersion.UnwrapOr(LatestMigrationVersion)
if migrationVersion > latestVersion {
return fmt.Errorf("%w: database version is newer than the "+
"latest migration version, preventing downgrade: "+
"db_version=%v, latest_migration_version=%v",
ErrMigrationDowngrade, migrationVersion, latestVersion)
}
// Report the current version of the database before the migration.
currentDbVersion, _, err := driver.Version()
if err != nil {
return fmt.Errorf("unable to get current db version: %w", err)
}
log.Infof("Attempting to apply migration(s) "+
"(current_db_version=%v, latest_migration_version=%v)",
currentDbVersion, latestVersion)
// Apply our local logger to the migration instance.
sqlMigrate.Log = &migrationLogger{log}
// Execute the migration based on the target given.
err = targetVersion(sqlMigrate, currentDbVersion, latestVersion)
if err != nil && !errors.Is(err, migrate.ErrNoChange) {
return err
}
// Report the current version of the database after the migration.
currentDbVersion, _, err = driver.Version()
if err != nil {
return fmt.Errorf("unable to get current db version: %w", err)
}
log.Infof("Database version after migration: %v", currentDbVersion)
return nil
}
// replacerFS is an implementation of a fs.FS virtual file system that wraps an
// existing file system but does a search-and-replace operation on each file
// when it is opened.
type replacerFS struct {
parentFS fs.FS
replaces map[string]string
}
// A compile-time assertion to make sure replacerFS implements the fs.FS
// interface.
var _ fs.FS = (*replacerFS)(nil)
// newReplacerFS creates a new replacer file system, wrapping the given parent
// virtual file system. Each file within the file system is undergoing a
// search-and-replace operation when it is opened, using the given map where the
// key denotes the search term and the value the term to replace each occurrence
// with.
func newReplacerFS(parent fs.FS, replaces map[string]string) *replacerFS {
return &replacerFS{
parentFS: parent,
replaces: replaces,
}
}
// Open opens a file in the virtual file system.
//
// NOTE: This is part of the fs.FS interface.
func (t *replacerFS) Open(name string) (fs.File, error) {
f, err := t.parentFS.Open(name)
if err != nil {
return nil, err
}
stat, err := f.Stat()
if err != nil {
return nil, err
}
if stat.IsDir() {
return f, err
}
return newReplacerFile(f, t.replaces)
}
type replacerFile struct {
parentFile fs.File
buf bytes.Buffer
}
// A compile-time assertion to make sure replacerFile implements the fs.File
// interface.
var _ fs.File = (*replacerFile)(nil)
func newReplacerFile(parent fs.File, replaces map[string]string) (*replacerFile,
error) {
content, err := io.ReadAll(parent)
if err != nil {
return nil, err
}
contentStr := string(content)
for from, to := range replaces {
contentStr = strings.ReplaceAll(contentStr, from, to)
}
var buf bytes.Buffer
_, err = buf.WriteString(contentStr)
if err != nil {
return nil, err
}
return &replacerFile{
parentFile: parent,
buf: buf,
}, nil
}
// Stat returns statistics/info about the file.
//
// NOTE: This is part of the fs.File interface.
func (t *replacerFile) Stat() (fs.FileInfo, error) {
return t.parentFile.Stat()
}
// Read reads as many bytes as possible from the file into the given slice.
//
// NOTE: This is part of the fs.File interface.
func (t *replacerFile) Read(bytes []byte) (int, error) {
return t.buf.Read(bytes)
}
// Close closes the underlying file.
//
// NOTE: This is part of the fs.File interface.
func (t *replacerFile) Close() error {
// We already fully read and then closed the file when creating this
// instance, so there's nothing to do for us here.
return nil
}

154
db/postgres.go
View file

@ -1,27 +1,16 @@
package db
import (
"database/sql"
"fmt"
"testing"
"time"
postgres_migrate "github.com/golang-migrate/migrate/v4/database/postgres"
_ "github.com/golang-migrate/migrate/v4/source/file"
"github.com/lightninglabs/lightning-terminal/db/sqlc"
"github.com/lightningnetwork/lnd/sqldb/v2"
"github.com/stretchr/testify/require"
)
const (
dsnTemplate = "postgres://%v:%v@%v:%d/%v?sslmode=%v"
// defaultMaxIdleConns is the number of permitted idle connections.
defaultMaxIdleConns = 6
// defaultConnMaxIdleTime is the amount of time a connection can be
// idle before it is closed.
defaultConnMaxIdleTime = 5 * time.Minute
)
var (
@ -31,16 +20,6 @@ var (
// fully executed yet. So this time needs to be chosen correctly to be
// longer than the longest expected individual test run time.
DefaultPostgresFixtureLifetime = 60 * time.Minute
// postgresSchemaReplacements is a map of schema strings that need to be
// replaced for postgres. This is needed because we write the schemas
// to work with sqlite primarily, and postgres has some differences.
postgresSchemaReplacements = map[string]string{
"BLOB": "BYTEA",
"INTEGER PRIMARY KEY": "BIGSERIAL PRIMARY KEY",
"TIMESTAMP": "TIMESTAMP WITHOUT TIME ZONE",
"UNHEX": "DECODE",
}
)
// PostgresConfig holds the postgres database configuration.
@ -77,94 +56,6 @@ func (s *PostgresConfig) DSN(hidePassword bool) string {
s.DBName, sslMode)
}
// PostgresStore is a database store implementation that uses a Postgres
// backend.
type PostgresStore struct {
cfg *PostgresConfig
*BaseDB
}
// NewPostgresStore creates a new store that is backed by a Postgres database
// backend.
func NewPostgresStore(cfg *PostgresConfig) (*PostgresStore, error) {
log.Infof("Using SQL database '%s'", cfg.DSN(true))
rawDb, err := sql.Open("pgx", cfg.DSN(false))
if err != nil {
return nil, err
}
maxConns := defaultMaxConns
if cfg.MaxOpenConnections > 0 {
maxConns = cfg.MaxOpenConnections
}
maxIdleConns := defaultMaxIdleConns
if cfg.MaxIdleConnections > 0 {
maxIdleConns = cfg.MaxIdleConnections
}
connMaxLifetime := defaultConnMaxLifetime
if cfg.ConnMaxLifetime > 0 {
connMaxLifetime = cfg.ConnMaxLifetime
}
connMaxIdleTime := defaultConnMaxIdleTime
if cfg.ConnMaxIdleTime > 0 {
connMaxIdleTime = cfg.ConnMaxIdleTime
}
rawDb.SetMaxOpenConns(maxConns)
rawDb.SetMaxIdleConns(maxIdleConns)
rawDb.SetConnMaxLifetime(connMaxLifetime)
rawDb.SetConnMaxIdleTime(connMaxIdleTime)
queries := sqlc.NewPostgres(rawDb)
s := &PostgresStore{
cfg: cfg,
BaseDB: &BaseDB{
DB: rawDb,
Queries: queries,
},
}
// Now that the database is open, populate the database with our set of
// schemas based on our embedded in-memory file system.
if !cfg.SkipMigrations {
if err := s.ExecuteMigrations(TargetLatest); err != nil {
return nil, fmt.Errorf("error executing migrations: "+
"%w", err)
}
}
return s, nil
}
// ExecuteMigrations runs migrations for the Postgres database, depending on the
// target given, either all migrations or up to a given version.
func (s *PostgresStore) ExecuteMigrations(target MigrationTarget,
optFuncs ...MigrateOpt) error {
opts := defaultMigrateOptions()
for _, optFunc := range optFuncs {
optFunc(opts)
}
driver, err := postgres_migrate.WithInstance(
s.DB, &postgres_migrate.Config{},
)
if err != nil {
return fmt.Errorf("error creating postgres migration: %w", err)
}
postgresFS := newReplacerFS(sqlSchemas, postgresSchemaReplacements)
return applyMigrations(
postgresFS, driver, "sqlc/migrations", s.cfg.DBName, target,
opts,
)
}
// NewTestPostgresV2DB is a helper function that creates a Postgres database for
// testing, using the sqldb v2 package's definition of the PostgresStore.
func NewTestPostgresV2DB(t *testing.T) *sqldb.PostgresStore {
@ -179,48 +70,3 @@ func NewTestPostgresV2DB(t *testing.T) *sqldb.PostgresStore {
return sqldb.NewTestPostgresDB(t, sqlFixture, LitdMigrationStreams)
}
// NewTestPostgresDB is a helper function that creates a Postgres database for
// testing, using the litd db package's definition of the PostgresStore.
//
// TODO(viktor): remove this once the sqldb v2 package is implemented in
// all of litd's packages.
func NewTestPostgresDB(t *testing.T) *PostgresStore {
t.Helper()
t.Logf("Creating new Postgres DB for testing")
sqlFixture := NewTestPgFixture(t, DefaultPostgresFixtureLifetime, true)
store, err := NewPostgresStore(sqlFixture.GetConfig())
require.NoError(t, err)
t.Cleanup(func() {
sqlFixture.TearDown(t)
})
return store
}
// NewTestPostgresDBWithVersion is a helper function that creates a Postgres
// database for testing and migrates it to the given version.
func NewTestPostgresDBWithVersion(t *testing.T, version uint) *PostgresStore {
t.Helper()
t.Logf("Creating new Postgres DB for testing, migrating to version %d",
version)
sqlFixture := NewTestPgFixture(t, DefaultPostgresFixtureLifetime, true)
storeCfg := sqlFixture.GetConfig()
storeCfg.SkipMigrations = true
store, err := NewPostgresStore(storeCfg)
require.NoError(t, err)
err = store.ExecuteMigrations(TargetVersion(version))
require.NoError(t, err)
t.Cleanup(func() {
sqlFixture.TearDown(t)
})
return store
}

10
db/sqlc/db_custom.go
View file

@ -26,16 +26,6 @@ func (q *Queries) Backend() sqldb.BackendType {
return wtx.backendType
}
// NewSqlite creates a new Queries instance for a SQLite database.
func NewSqlite(db DBTX) *Queries {
return &Queries{db: &wrappedTX{db, sqldb.BackendTypeSqlite}}
}
// NewPostgres creates a new Queries instance for a Postgres database.
func NewPostgres(db DBTX) *Queries {
return &Queries{db: &wrappedTX{db, sqldb.BackendTypePostgres}}
}
// NewForType creates a new Queries instance for the given database type.
func NewForType(db DBTX, typ sqldb.BackendType) *Queries {
return &Queries{db: &wrappedTX{db, typ}}

285
db/sqlite.go
View file

@ -1,49 +1,9 @@
package db
import (
"database/sql"
"fmt"
"net/url"
"path/filepath"
"testing"
"time"
"github.com/golang-migrate/migrate/v4"
sqlite_migrate "github.com/golang-migrate/migrate/v4/database/sqlite"
"github.com/lightninglabs/lightning-terminal/db/sqlc"
"github.com/stretchr/testify/require"
_ "modernc.org/sqlite" // Register relevant drivers.
)
const (
// sqliteOptionPrefix is the string prefix sqlite uses to set various
// options. This is used in the following format:
// * sqliteOptionPrefix || option_name = option_value.
sqliteOptionPrefix = "_pragma"
// sqliteTxLockImmediate is a dsn option used to ensure that write
// transactions are started immediately.
sqliteTxLockImmediate = "_txlock=immediate"
// defaultMaxConns is the number of permitted active and idle
// connections. We want to limit this so it isn't unlimited. We use the
// same value for the number of idle connections as, this can speed up
// queries given a new connection doesn't need to be established each
// time.
defaultMaxConns = 25
// defaultConnMaxLifetime is the maximum amount of time a connection can
// be reused for before it is closed.
defaultConnMaxLifetime = 10 * time.Minute
)
var (
// sqliteSchemaReplacements is a map of schema strings that need to be
// replaced for sqlite. There currently aren't any replacements, because
// the SQL files are written with SQLite compatibility in mind.
sqliteSchemaReplacements = map[string]string{}
)
// SqliteConfig holds all the config arguments needed to interact with our
// sqlite DB.
//
@ -61,248 +21,3 @@ type SqliteConfig struct {
// found.
DatabaseFileName string `long:"dbfile" description:"The full path to the database."`
}
// SqliteStore is a sqlite3 based database for the Taproot Asset daemon.
type SqliteStore struct {
cfg *SqliteConfig
*BaseDB
}
// NewSqliteStore attempts to open a new sqlite database based on the passed
// config.
func NewSqliteStore(cfg *SqliteConfig) (*SqliteStore, error) {
// The set of pragma options are accepted using query options. For now
// we only want to ensure that foreign key constraints are properly
// enforced.
pragmaOptions := []struct {
name string
value string
}{
{
name: "foreign_keys",
value: "on",
},
{
name: "journal_mode",
value: "WAL",
},
{
name: "busy_timeout",
value: "5000",
},
{
// With the WAL mode, this ensures that we also do an
// extra WAL sync after each transaction. The normal
// sync mode skips this and gives better performance,
// but risks durability.
name: "synchronous",
value: "full",
},
{
// This is used to ensure proper durability for users
// running on Mac OS. It uses the correct fsync system
// call to ensure items are fully flushed to disk.
name: "fullfsync",
value: "true",
},
}
sqliteOptions := make(url.Values)
for _, option := range pragmaOptions {
sqliteOptions.Add(
sqliteOptionPrefix,
fmt.Sprintf("%v=%v", option.name, option.value),
)
}
// Construct the DSN which is just the database file name, appended
// with the series of pragma options as a query URL string. For more
// details on the formatting here, see the modernc.org/sqlite docs:
// https://pkg.go.dev/modernc.org/sqlite#Driver.Open.
dsn := fmt.Sprintf(
"%v?%v&%v", cfg.DatabaseFileName, sqliteOptions.Encode(),
sqliteTxLockImmediate,
)
db, err := sql.Open("sqlite", dsn)
if err != nil {
return nil, err
}
db.SetMaxOpenConns(defaultMaxConns)
db.SetMaxIdleConns(defaultMaxConns)
db.SetConnMaxLifetime(defaultConnMaxLifetime)
queries := sqlc.NewSqlite(db)
s := &SqliteStore{
cfg: cfg,
BaseDB: &BaseDB{
DB: db,
Queries: queries,
},
}
// Now that the database is open, populate the database with our set of
// schemas based on our embedded in-memory file system.
if !cfg.SkipMigrations {
if err := s.ExecuteMigrations(s.backupAndMigrate); err != nil {
return nil, fmt.Errorf("error executing migrations: "+
"%w", err)
}
}
return s, nil
}
// backupSqliteDatabase creates a backup of the given SQLite database.
func backupSqliteDatabase(srcDB *sql.DB, dbFullFilePath string) error {
if srcDB == nil {
return fmt.Errorf("backup source database is nil")
}
// Create a database backup file full path from the given source
// database full file path.
//
// Get the current time and format it as a Unix timestamp in
// nanoseconds.
timestamp := time.Now().UnixNano()
// Add the timestamp to the backup name.
backupFullFilePath := fmt.Sprintf(
"%s.%d.backup", dbFullFilePath, timestamp,
)
log.Infof("Creating backup of database file: %v -> %v",
dbFullFilePath, backupFullFilePath)
// Create the database backup.
vacuumIntoQuery := "VACUUM INTO ?;"
stmt, err := srcDB.Prepare(vacuumIntoQuery)
if err != nil {
return err
}
defer stmt.Close()
_, err = stmt.Exec(backupFullFilePath)
if err != nil {
return err
}
return nil
}
// backupAndMigrate is a helper function that creates a database backup before
// initiating the migration, and then migrates the database to the latest
// version.
func (s *SqliteStore) backupAndMigrate(mig *migrate.Migrate,
currentDbVersion int, maxMigrationVersion uint) error {
// Determine if a database migration is necessary given the current
// database version and the maximum migration version.
versionUpgradePending := currentDbVersion < int(maxMigrationVersion)
if !versionUpgradePending {
log.Infof("Current database version is up-to-date, skipping "+
"migration attempt and backup creation "+
"(current_db_version=%v, max_migration_version=%v)",
currentDbVersion, maxMigrationVersion)
return nil
}
// At this point, we know that a database migration is necessary.
// Create a backup of the database before starting the migration.
if !s.cfg.SkipMigrationDbBackup {
log.Infof("Creating database backup (before applying " +
"migration(s))")
err := backupSqliteDatabase(s.DB, s.cfg.DatabaseFileName)
if err != nil {
return err
}
} else {
log.Infof("Skipping database backup creation before applying " +
"migration(s)")
}
log.Infof("Applying migrations to database")
return mig.Up()
}
// ExecuteMigrations runs migrations for the sqlite database, depending on the
// target given, either all migrations or up to a given version.
func (s *SqliteStore) ExecuteMigrations(target MigrationTarget,
optFuncs ...MigrateOpt) error {
opts := defaultMigrateOptions()
for _, optFunc := range optFuncs {
optFunc(opts)
}
driver, err := sqlite_migrate.WithInstance(
s.DB, &sqlite_migrate.Config{},
)
if err != nil {
return fmt.Errorf("error creating sqlite migration: %w", err)
}
sqliteFS := newReplacerFS(sqlSchemas, sqliteSchemaReplacements)
return applyMigrations(
sqliteFS, driver, "sqlc/migrations", "sqlite", target, opts,
)
}
// NewTestSqliteDB is a helper function that creates an SQLite database for
// testing.
func NewTestSqliteDB(t *testing.T) *SqliteStore {
t.Helper()
// TODO(roasbeef): if we pass :memory: for the file name, then we get
// an in mem version to speed up tests
dbPath := filepath.Join(t.TempDir(), "tmp.db")
t.Logf("Creating new SQLite DB handle for testing: %s", dbPath)
return NewTestSqliteDbHandleFromPath(t, dbPath)
}
// NewTestSqliteDbHandleFromPath is a helper function that creates a SQLite
// database handle given a database file path.
func NewTestSqliteDbHandleFromPath(t *testing.T, dbPath string) *SqliteStore {
t.Helper()
sqlDB, err := NewSqliteStore(&SqliteConfig{
DatabaseFileName: dbPath,
SkipMigrations: false,
})
require.NoError(t, err)
t.Cleanup(func() {
require.NoError(t, sqlDB.DB.Close())
})
return sqlDB
}
// NewTestSqliteDBWithVersion is a helper function that creates an SQLite
// database for testing and migrates it to the given version.
func NewTestSqliteDBWithVersion(t *testing.T, version uint) *SqliteStore {
t.Helper()
t.Logf("Creating new SQLite DB for testing, migrating to version %d",
version)
// TODO(roasbeef): if we pass :memory: for the file name, then we get
// an in mem version to speed up tests
dbFileName := filepath.Join(t.TempDir(), "tmp.db")
sqlDB, err := NewSqliteStore(&SqliteConfig{
DatabaseFileName: dbFileName,
SkipMigrations: true,
})
require.NoError(t, err)
err = sqlDB.ExecuteMigrations(TargetVersion(version))
require.NoError(t, err)
t.Cleanup(func() {
require.NoError(t, sqlDB.DB.Close())
})
return sqlDB
}