mirror of
https://github.com/status-im/go-waku.git
synced 2025-01-28 14:34:49 +00:00
103 lines
2.3 KiB
Go
103 lines
2.3 KiB
Go
package postgres
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"fmt"
|
|
"time"
|
|
|
|
"github.com/golang-migrate/migrate/v4/database"
|
|
"github.com/golang-migrate/migrate/v4/database/pgx"
|
|
_ "github.com/jackc/pgx/v5/stdlib" // Blank import to register the postgres driver
|
|
"github.com/waku-org/go-waku/waku/persistence"
|
|
"github.com/waku-org/go-waku/waku/persistence/migrate"
|
|
"github.com/waku-org/go-waku/waku/persistence/postgres/migrations"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
func executeVacuum(db *sql.DB, logger *zap.Logger) error {
|
|
logger.Info("starting PostgreSQL database vacuuming")
|
|
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
|
|
errCh := make(chan error)
|
|
|
|
go func() {
|
|
defer cancel()
|
|
_, err := db.Exec("VACUUM FULL")
|
|
if err != nil {
|
|
errCh <- err
|
|
}
|
|
}()
|
|
|
|
t := time.NewTicker(2 * time.Minute)
|
|
defer t.Stop()
|
|
|
|
loop:
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
break loop
|
|
case err := <-errCh:
|
|
return err
|
|
case <-t.C:
|
|
logger.Info("still vacuuming...")
|
|
}
|
|
}
|
|
|
|
logger.Info("finished PostgreSQL database vacuuming")
|
|
return nil
|
|
}
|
|
|
|
// NewDB connects to postgres DB in the specified path
|
|
func NewDB(dburl string, shouldVacuum bool, logger *zap.Logger) (*sql.DB, error) {
|
|
db, err := sql.Open("pgx", dburl)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
if shouldVacuum {
|
|
err := executeVacuum(db, logger)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
return db, nil
|
|
}
|
|
|
|
func migrationDriver(db *sql.DB) (database.Driver, error) {
|
|
return pgx.WithInstance(db, &pgx.Config{
|
|
MigrationsTable: "gowaku_" + pgx.DefaultMigrationsTable,
|
|
})
|
|
}
|
|
|
|
// Migrations is the function used for DB migration with postgres driver
|
|
func Migrations(db *sql.DB) error {
|
|
migrationDriver, err := migrationDriver(db)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return migrate.Migrate(db, migrationDriver, migrations.AssetNames(), migrations.Asset)
|
|
}
|
|
|
|
// CreateTable creates the table that will persist the peers
|
|
func CreateTable(db *sql.DB, tableName string) error {
|
|
sqlStmt := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s (key TEXT NOT NULL UNIQUE, data BYTEA);", tableName)
|
|
_, err := db.Exec(sqlStmt)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// NewQueries creates a new SQL set of queries for the passed table
|
|
func NewQueries(tbl string, db *sql.DB) (*persistence.Queries, error) {
|
|
err := CreateTable(db, tbl)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return persistence.CreateQueries(tbl), nil
|
|
}
|