JakeHillion/storj
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

multinode/multinodedb: add db migration

Browse Source 
Change-Id: Ied1a7f3b951a25ab7a8edc25902c0479d2d08e87
This commit is contained in:
Yaroslav Vorobiov 2021-04-28 04:30:01 +03:00
parent 910eec8eee
commit ea7fbdf843
16 changed files with 368 additions and 936 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
cmd/multinode/main.go
View File

@ -36,11 +36,6 @@ var (
Short: "Run the multinode dashboard", Short: "Run the multinode dashboard",
RunE: cmdRun, RunE: cmdRun,
} }
createSchemaCmd = &cobra.Command{
Use: "create-schema",
Short: "Create schemas for multinode dashboard databases",
RunE: cmdCreateSchema,
}
setupCmd = &cobra.Command{ setupCmd = &cobra.Command{
Use: "setup", Use: "setup",
Short: "Create config files", Short: "Create config files",
@ -67,10 +62,8 @@ func init() {
rootCmd.AddCommand(setupCmd) rootCmd.AddCommand(setupCmd)
rootCmd.AddCommand(runCmd) rootCmd.AddCommand(runCmd)
rootCmd.AddCommand(createSchemaCmd)
process.Bind(runCmd, &runCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir)) process.Bind(runCmd, &runCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir))
process.Bind(createSchemaCmd, &runCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir))
process.Bind(setupCmd, &setupCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir), cfgstruct.SetupMode()) process.Bind(setupCmd, &setupCfg, defaults, cfgstruct.ConfDir(confDir), cfgstruct.IdentityDir(identityDir), cfgstruct.SetupMode())
} }
@ -112,6 +105,9 @@ func cmdRun(cmd *cobra.Command, args []string) (err error) {
defer func() { defer func() {
err = errs.Combine(err, db.Close()) err = errs.Combine(err, db.Close())
}() }()
if err := db.MigrateToLatest(ctx); err != nil {
return err
}
peer, err := multinode.New(log, identity, runCfg.Config, db) peer, err := multinode.New(log, identity, runCfg.Config, db)
if err != nil { if err != nil {
@ -122,23 +118,3 @@ func cmdRun(cmd *cobra.Command, args []string) (err error) {
closeError := peer.Close() closeError := peer.Close()
return errs.Combine(runError, closeError) return errs.Combine(runError, closeError)
} }
func cmdCreateSchema(cmd *cobra.Command, args []string) (err error) {
ctx, _ := process.Ctx(cmd)
log := zap.L()
db, err := multinodedb.Open(ctx, log.Named("db"), runCfg.Database)
if err != nil {
return errs.New("error connecting to master database on multinode: %+v", err)
}
defer func() {
err = errs.Combine(err, db.Close())
}()
err = db.CreateSchema(ctx)
if err != nil {
return errs.New("error creating database schemas for multinode dashboard db: %+v", err)
}
return nil
}

40
multinode/console/members.go
View File

@ -1,40 +0,0 @@
// Copyright (C) 2020 Storj Labs, Inc.
// See LICENSE for copying information.
package console
import (
"context"
"github.com/zeebo/errs"
"storj.io/common/uuid"
)
// Members exposes needed by MND MembersDB functionality.
//
// architecture: Database
type Members interface {
// Invite will create empty row in membersDB.
Invite(ctx context.Context, member Member) error
// Update updates all updatable fields of member.
Update(ctx context.Context, member Member) error
// Remove deletes member from membersDB.
Remove(ctx context.Context, id uuid.UUID) error
// GetByEmail will return member with specified email.
GetByEmail(ctx context.Context, email string) (Member, error)
// GetByID will return member with specified id.
GetByID(ctx context.Context, id uuid.UUID) (Member, error)
}
// ErrNoMember is a special error type that indicates about absence of member in MembersDB.
var ErrNoMember = errs.Class("no such member")
// Member represents some person that is invited to the MND by node owner.
// Member will have configurable access privileges that will define which functions and which nodes are available for him.
type Member struct {
ID uuid.UUID
Email string
Name string
PasswordHash []byte
}

64
multinode/console/members_test.go
View File

@ -1,64 +0,0 @@
// Copyright (C) 2020 Storj Labs, Inc.
// See LICENSE for copying information.
package console_test
import (
"testing"
"github.com/stretchr/testify/require"
"github.com/zeebo/assert"
"storj.io/common/testcontext"
"storj.io/common/uuid"
"storj.io/storj/multinode"
"storj.io/storj/multinode/console"
"storj.io/storj/multinode/multinodedb/multinodedbtest"
)
func TestMembersDB(t *testing.T) {
multinodedbtest.Run(t, func(ctx *testcontext.Context, t *testing.T, db multinode.DB) {
members := db.Members()
memberID, err := uuid.New()
require.NoError(t, err)
memberBob := console.Member{
ID: memberID,
Email: "mail@example.com",
Name: "Bob",
PasswordHash: []byte{0},
}
err = members.Invite(ctx, memberBob)
assert.NoError(t, err)
memberToCheck, err := members.GetByEmail(ctx, memberBob.Email)
assert.NoError(t, err)
assert.Equal(t, memberToCheck.Email, memberBob.Email)
assert.Equal(t, memberToCheck.Name, memberBob.Name)
assert.Equal(t, memberToCheck.Email, memberBob.Email)
memberBob.Name = "Alice"
err = members.Update(ctx, memberBob)
assert.NoError(t, err)
memberAlice, err := members.GetByID(ctx, memberToCheck.ID)
assert.NoError(t, err)
assert.Equal(t, memberToCheck.Email, memberAlice.Email)
assert.Equal(t, memberToCheck.Name, memberAlice.Name)
assert.Equal(t, memberToCheck.Email, memberAlice.Email)
assert.Equal(t, memberToCheck.ID, memberAlice.ID)
err = members.Remove(ctx, memberAlice.ID)
assert.NoError(t, err)
_, err = members.GetByID(ctx, memberToCheck.ID)
assert.Error(t, err)
assert.Equal(t, true, console.ErrNoMember.Has(err))
_, err = members.GetByEmail(ctx, memberToCheck.Email)
assert.Error(t, err)
assert.Equal(t, true, console.ErrNoMember.Has(err))
})
}

61
multinode/multinodedb/database.go
View File

@ -13,15 +13,16 @@ import (
"storj.io/private/dbutil" "storj.io/private/dbutil"
"storj.io/private/dbutil/pgutil" "storj.io/private/dbutil/pgutil"
"storj.io/private/tagsql"
"storj.io/storj/multinode" "storj.io/storj/multinode"
"storj.io/storj/multinode/console"
"storj.io/storj/multinode/multinodedb/dbx" "storj.io/storj/multinode/multinodedb/dbx"
"storj.io/storj/multinode/nodes" "storj.io/storj/multinode/nodes"
"storj.io/storj/private/migrate"
) )
var ( var (
// ensures that multinodeDB implements multinode.DB. // ensures that multinodeDB implements multinode.DB.
_ multinode.DB = (*multinodeDB)(nil) _ multinode.DB = (*DB)(nil)
mon = monkit.Package() mon = monkit.Package()
@ -29,28 +30,29 @@ var (
Error = errs.Class("multinodedb") Error = errs.Class("multinodedb")
) )
// multinodeDB combines access to different database tables with a record // DB combines access to different database tables with a record
// of the db driver, db implementation, and db source URL. // of the db driver, db implementation, and db source URL.
// Implementation of multinode.DB interface. // Implementation of multinode.DB interface.
// //
// architecture: Master Database // architecture: Master Database
type multinodeDB struct { type DB struct {
*dbx.DB *dbx.DB
log *zap.Logger log *zap.Logger
driver string driver string
impl dbutil.Implementation source string
source string implementation dbutil.Implementation
migrationDB tagsql.DB
} }
// Open creates instance of database supports postgres. // Open creates instance of database supports postgres.
func Open(ctx context.Context, log *zap.Logger, databaseURL string) (multinode.DB, error) { func Open(ctx context.Context, log *zap.Logger, databaseURL string) (*DB, error) {
driver, source, impl, err := dbutil.SplitConnStr(databaseURL) driver, source, implementation, err := dbutil.SplitConnStr(databaseURL)
if err != nil { if err != nil {
return nil, err return nil, err
} }
switch impl { switch implementation {
case dbutil.SQLite3: case dbutil.SQLite3:
source = sqlite3SetDefaultOptions(source) source = sqlite3SetDefaultOptions(source)
case dbutil.Postgres: case dbutil.Postgres:
@ -67,41 +69,44 @@ func Open(ctx context.Context, log *zap.Logger, databaseURL string) (multinode.D
return nil, Error.New("failed opening database via DBX at %q: %v", return nil, Error.New("failed opening database via DBX at %q: %v",
source, err) source, err)
} }
log.Debug("Connected to:", zap.String("db source", source)) log.Debug("Connected to:", zap.String("db source", source))
dbutil.Configure(ctx, dbxDB.DB, "multinodedb", mon) dbutil.Configure(ctx, dbxDB.DB, "multinodedb", mon)
core := &multinodeDB{ core := &DB{
DB: dbxDB, DB: dbxDB,
log: log, log: log,
driver: driver, driver: driver,
impl: impl, implementation: implementation,
source: source, source: source,
} }
core.migrationDB = core
return core, nil return core, nil
} }
// Nodes returns nodes database. // Nodes returns nodes database.
func (db *multinodeDB) Nodes() nodes.DB { func (db *DB) Nodes() nodes.DB {
return &nodesdb{ return &nodesdb{
methods: db, methods: db,
} }
} }
// Members returns members database. // MigrateToLatest migrates db to the latest version.
func (db *multinodeDB) Members() console.Members { func (db DB) MigrateToLatest(ctx context.Context) error {
return &members{ var migration *migrate.Migration
methods: db,
}
}
// CreateSchema creates schema. switch db.implementation {
func (db *multinodeDB) CreateSchema(ctx context.Context) error { case dbutil.SQLite3:
_, err := db.ExecContext(ctx, db.DB.Schema()) migration = db.SQLite3Migration()
return err case dbutil.Postgres:
migration = db.PostgresMigration()
default:
return migrate.Create(ctx, "database", db.DB)
}
return migration.Run(ctx, db.log)
} }
// sqlite3SetDefaultOptions sets default options for disk-based db with URI filename source string // sqlite3SetDefaultOptions sets default options for disk-based db with URI filename source string

2
multinode/multinodedb/dbx/gen.sh
View File

@ -1,9 +1,9 @@
#!/bin/sh #!/bin/sh
set -euo pipefail set -euo pipefail
dbx schema -d pgx -d sqlite3 multinodedb.dbx . dbx schema -d pgx -d sqlite3 multinodedb.dbx .
dbx golang -d pgx -d sqlite3 -p dbx -t templates multinodedb.dbx . dbx golang -d pgx -d sqlite3 -p dbx -t templates multinodedb.dbx .
( printf '%s\n' '//lint:file-ignore U1000,ST1012 generated file'; cat multinodedb.dbx.go ) > multinodedb.dbx.go.tmp && mv multinodedb.dbx.go.tmp multinodedb.dbx.go ( printf '%s\n' '//lint:file-ignore U1000,ST1012 generated file'; cat multinodedb.dbx.go ) > multinodedb.dbx.go.tmp && mv multinodedb.dbx.go.tmp multinodedb.dbx.go
gofmt -r "*sql.Tx -> tagsql.Tx" -w multinodedb.dbx.go gofmt -r "*sql.Tx -> tagsql.Tx" -w multinodedb.dbx.go
gofmt -r "*sql.Rows -> tagsql.Rows" -w multinodedb.dbx.go gofmt -r "*sql.Rows -> tagsql.Rows" -w multinodedb.dbx.go

25
multinode/multinodedb/dbx/multinodedb.dbx
View File

@ -24,28 +24,3 @@ update node (
where node.id = ? where node.id = ?
noreturn noreturn
) )
model member (
key id
field id blob
field email text ( updatable )
field name text ( updatable )
field password_hash blob ( updatable )
field created_at timestamp ( autoinsert )
)
create member ( )
delete member ( where member.id = ? )
update member ( where member.id = ? )
read one (
select member
where member.email = ?
)
read one (
select member
where member.id = ?
)

614
multinode/multinodedb/dbx/multinodedb.dbx.go
View File

@ -275,15 +275,7 @@ func newpgx(db *DB) *pgxDB {
} }
func (obj *pgxDB) Schema() string { func (obj *pgxDB) Schema() string {
return `CREATE TABLE members ( return `CREATE TABLE nodes (
id bytea NOT NULL,
email text NOT NULL,
name text NOT NULL,
password_hash bytea NOT NULL,
created_at timestamp with time zone NOT NULL,
PRIMARY KEY ( id )
);
CREATE TABLE nodes (
id bytea NOT NULL, id bytea NOT NULL,
name text NOT NULL, name text NOT NULL,
public_address text NOT NULL, public_address text NOT NULL,
@ -353,15 +345,7 @@ func newsqlite3(db *DB) *sqlite3DB {
} }
func (obj *sqlite3DB) Schema() string { func (obj *sqlite3DB) Schema() string {
return `CREATE TABLE members ( return `CREATE TABLE nodes (
id BLOB NOT NULL,
email TEXT NOT NULL,
name TEXT NOT NULL,
password_hash BLOB NOT NULL,
created_at TIMESTAMP NOT NULL,
PRIMARY KEY ( id )
);
CREATE TABLE nodes (
id BLOB NOT NULL, id BLOB NOT NULL,
name TEXT NOT NULL, name TEXT NOT NULL,
public_address TEXT NOT NULL, public_address TEXT NOT NULL,
@ -430,117 +414,6 @@ nextval:
fmt.Fprint(f, "]") fmt.Fprint(f, "]")
} }
type Member struct {
Id []byte
Email string
Name string
PasswordHash []byte
CreatedAt time.Time
}
func (Member) _Table() string { return "members" }
type Member_Update_Fields struct {
Email Member_Email_Field
Name Member_Name_Field
PasswordHash Member_PasswordHash_Field
}
type Member_Id_Field struct {
_set bool
_null bool
_value []byte
}
func Member_Id(v []byte) Member_Id_Field {
return Member_Id_Field{_set: true, _value: v}
}
func (f Member_Id_Field) value() interface{} {
if !f._set || f._null {
return nil
}
return f._value
}
func (Member_Id_Field) _Column() string { return "id" }
type Member_Email_Field struct {
_set bool
_null bool
_value string
}
func Member_Email(v string) Member_Email_Field {
return Member_Email_Field{_set: true, _value: v}
}
func (f Member_Email_Field) value() interface{} {
if !f._set || f._null {
return nil
}
return f._value
}
func (Member_Email_Field) _Column() string { return "email" }
type Member_Name_Field struct {
_set bool
_null bool
_value string
}
func Member_Name(v string) Member_Name_Field {
return Member_Name_Field{_set: true, _value: v}
}
func (f Member_Name_Field) value() interface{} {
if !f._set || f._null {
return nil
}
return f._value
}
func (Member_Name_Field) _Column() string { return "name" }
type Member_PasswordHash_Field struct {
_set bool
_null bool
_value []byte
}
func Member_PasswordHash(v []byte) Member_PasswordHash_Field {
return Member_PasswordHash_Field{_set: true, _value: v}
}
func (f Member_PasswordHash_Field) value() interface{} {
if !f._set || f._null {
return nil
}
return f._value
}
func (Member_PasswordHash_Field) _Column() string { return "password_hash" }
type Member_CreatedAt_Field struct {
_set bool
_null bool
_value time.Time
}
func Member_CreatedAt(v time.Time) Member_CreatedAt_Field {
return Member_CreatedAt_Field{_set: true, _value: v}
}
func (f Member_CreatedAt_Field) value() interface{} {
if !f._set || f._null {
return nil
}
return f._value
}
func (Member_CreatedAt_Field) _Column() string { return "created_at" }
type Node struct { type Node struct {
Id []byte Id []byte
Name string Name string
@ -1079,38 +952,6 @@ func (obj *pgxImpl) Create_Node(ctx context.Context,
} }
func (obj *pgxImpl) Create_Member(ctx context.Context,
member_id Member_Id_Field,
member_email Member_Email_Field,
member_name Member_Name_Field,
member_password_hash Member_PasswordHash_Field) (
member *Member, err error) {
defer mon.Task()(&ctx)(&err)
__now := obj.db.Hooks.Now().UTC()
__id_val := member_id.value()
__email_val := member_email.value()
__name_val := member_name.value()
__password_hash_val := member_password_hash.value()
__created_at_val := __now
var __embed_stmt = __sqlbundle_Literal("INSERT INTO members ( id, email, name, password_hash, created_at ) VALUES ( ?, ?, ?, ?, ? ) RETURNING members.id, members.email, members.name, members.password_hash, members.created_at")
var __values []interface{}
__values = append(__values, __id_val, __email_val, __name_val, __password_hash_val, __created_at_val)
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, __values...)
member = &Member{}
err = obj.driver.QueryRowContext(ctx, __stmt, __values...).Scan(&member.Id, &member.Email, &member.Name, &member.PasswordHash, &member.CreatedAt)
if err != nil {
return nil, obj.makeErr(err)
}
return member, nil
}
func (obj *pgxImpl) Get_Node_By_Id(ctx context.Context, func (obj *pgxImpl) Get_Node_By_Id(ctx context.Context,
node_id Node_Id_Field) ( node_id Node_Id_Field) (
node *Node, err error) { node *Node, err error) {
@ -1165,72 +1006,6 @@ func (obj *pgxImpl) All_Node(ctx context.Context) (
} }
func (obj *pgxImpl) Get_Member_By_Email(ctx context.Context,
member_email Member_Email_Field) (
member *Member, err error) {
defer mon.Task()(&ctx)(&err)
var __embed_stmt = __sqlbundle_Literal("SELECT members.id, members.email, members.name, members.password_hash, members.created_at FROM members WHERE members.email = ? LIMIT 2")
var __values []interface{}
__values = append(__values, member_email.value())
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, __values...)
__rows, err := obj.driver.QueryContext(ctx, __stmt, __values...)
if err != nil {
return nil, obj.makeErr(err)
}
defer __rows.Close()
if !__rows.Next() {
if err := __rows.Err(); err != nil {
return nil, obj.makeErr(err)
}
return nil, makeErr(sql.ErrNoRows)
}
member = &Member{}
err = __rows.Scan(&member.Id, &member.Email, &member.Name, &member.PasswordHash, &member.CreatedAt)
if err != nil {
return nil, obj.makeErr(err)
}
if __rows.Next() {
return nil, tooManyRows("Member_By_Email")
}
if err := __rows.Err(); err != nil {
return nil, obj.makeErr(err)
}
return member, nil
}
func (obj *pgxImpl) Get_Member_By_Id(ctx context.Context,
member_id Member_Id_Field) (
member *Member, err error) {
defer mon.Task()(&ctx)(&err)
var __embed_stmt = __sqlbundle_Literal("SELECT members.id, members.email, members.name, members.password_hash, members.created_at FROM members WHERE members.id = ?")
var __values []interface{}
__values = append(__values, member_id.value())
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, __values...)
member = &Member{}
err = obj.driver.QueryRowContext(ctx, __stmt, __values...).Scan(&member.Id, &member.Email, &member.Name, &member.PasswordHash, &member.CreatedAt)
if err != nil {
return (*Member)(nil), obj.makeErr(err)
}
return member, nil
}
func (obj *pgxImpl) Update_Node_By_Id(ctx context.Context, func (obj *pgxImpl) Update_Node_By_Id(ctx context.Context,
node_id Node_Id_Field, node_id Node_Id_Field,
update Node_Update_Fields) ( update Node_Update_Fields) (
@ -1309,57 +1084,6 @@ func (obj *pgxImpl) UpdateNoReturn_Node_By_Id(ctx context.Context,
return nil return nil
} }
func (obj *pgxImpl) Update_Member_By_Id(ctx context.Context,
member_id Member_Id_Field,
update Member_Update_Fields) (
member *Member, err error) {
defer mon.Task()(&ctx)(&err)
var __sets = &__sqlbundle_Hole{}
var __embed_stmt = __sqlbundle_Literals{Join: "", SQLs: []__sqlbundle_SQL{__sqlbundle_Literal("UPDATE members SET "), __sets, __sqlbundle_Literal(" WHERE members.id = ? RETURNING members.id, members.email, members.name, members.password_hash, members.created_at")}}
__sets_sql := __sqlbundle_Literals{Join: ", "}
var __values []interface{}
var __args []interface{}
if update.Email._set {
__values = append(__values, update.Email.value())
__sets_sql.SQLs = append(__sets_sql.SQLs, __sqlbundle_Literal("email = ?"))
}
if update.Name._set {
__values = append(__values, update.Name.value())
__sets_sql.SQLs = append(__sets_sql.SQLs, __sqlbundle_Literal("name = ?"))
}
if update.PasswordHash._set {
__values = append(__values, update.PasswordHash.value())
__sets_sql.SQLs = append(__sets_sql.SQLs, __sqlbundle_Literal("password_hash = ?"))
}
if len(__sets_sql.SQLs) == 0 {
return nil, emptyUpdate()
}
__args = append(__args, member_id.value())
__values = append(__values, __args...)
__sets.SQL = __sets_sql
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, __values...)
member = &Member{}
err = obj.driver.QueryRowContext(ctx, __stmt, __values...).Scan(&member.Id, &member.Email, &member.Name, &member.PasswordHash, &member.CreatedAt)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, obj.makeErr(err)
}
return member, nil
}
func (obj *pgxImpl) Delete_Node_By_Id(ctx context.Context, func (obj *pgxImpl) Delete_Node_By_Id(ctx context.Context,
node_id Node_Id_Field) ( node_id Node_Id_Field) (
deleted bool, err error) { deleted bool, err error) {
@ -1387,33 +1111,6 @@ func (obj *pgxImpl) Delete_Node_By_Id(ctx context.Context,
} }
func (obj *pgxImpl) Delete_Member_By_Id(ctx context.Context,
member_id Member_Id_Field) (
deleted bool, err error) {
defer mon.Task()(&ctx)(&err)
var __embed_stmt = __sqlbundle_Literal("DELETE FROM members WHERE members.id = ?")
var __values []interface{}
__values = append(__values, member_id.value())
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, __values...)
__res, err := obj.driver.ExecContext(ctx, __stmt, __values...)
if err != nil {
return false, obj.makeErr(err)
}
__count, err := __res.RowsAffected()
if err != nil {
return false, obj.makeErr(err)
}
return __count > 0, nil
}
func (impl pgxImpl) isConstraintError(err error) ( func (impl pgxImpl) isConstraintError(err error) (
constraint string, ok bool) { constraint string, ok bool) {
if e, ok := err.(*pgconn.PgError); ok { if e, ok := err.(*pgconn.PgError); ok {
@ -1433,16 +1130,6 @@ func (obj *pgxImpl) deleteAll(ctx context.Context) (count int64, err error) {
return 0, obj.makeErr(err) return 0, obj.makeErr(err)
} }
__count, err = __res.RowsAffected()
if err != nil {
return 0, obj.makeErr(err)
}
count += __count
__res, err = obj.driver.ExecContext(ctx, "DELETE FROM members;")
if err != nil {
return 0, obj.makeErr(err)
}
__count, err = __res.RowsAffected() __count, err = __res.RowsAffected()
if err != nil { if err != nil {
return 0, obj.makeErr(err) return 0, obj.makeErr(err)
@ -1485,41 +1172,6 @@ func (obj *sqlite3Impl) Create_Node(ctx context.Context,
} }
func (obj *sqlite3Impl) Create_Member(ctx context.Context,
member_id Member_Id_Field,
member_email Member_Email_Field,
member_name Member_Name_Field,
member_password_hash Member_PasswordHash_Field) (
member *Member, err error) {
defer mon.Task()(&ctx)(&err)
__now := obj.db.Hooks.Now().UTC()
__id_val := member_id.value()
__email_val := member_email.value()
__name_val := member_name.value()
__password_hash_val := member_password_hash.value()
__created_at_val := __now
var __embed_stmt = __sqlbundle_Literal("INSERT INTO members ( id, email, name, password_hash, created_at ) VALUES ( ?, ?, ?, ?, ? )")
var __values []interface{}
__values = append(__values, __id_val, __email_val, __name_val, __password_hash_val, __created_at_val)
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, __values...)
__res, err := obj.driver.ExecContext(ctx, __stmt, __values...)
if err != nil {
return nil, obj.makeErr(err)
}
__pk, err := __res.LastInsertId()
if err != nil {
return nil, obj.makeErr(err)
}
return obj.getLastMember(ctx, __pk)
}
func (obj *sqlite3Impl) Get_Node_By_Id(ctx context.Context, func (obj *sqlite3Impl) Get_Node_By_Id(ctx context.Context,
node_id Node_Id_Field) ( node_id Node_Id_Field) (
node *Node, err error) { node *Node, err error) {
@ -1574,72 +1226,6 @@ func (obj *sqlite3Impl) All_Node(ctx context.Context) (
} }
func (obj *sqlite3Impl) Get_Member_By_Email(ctx context.Context,
member_email Member_Email_Field) (
member *Member, err error) {
defer mon.Task()(&ctx)(&err)
var __embed_stmt = __sqlbundle_Literal("SELECT members.id, members.email, members.name, members.password_hash, members.created_at FROM members WHERE members.email = ? LIMIT 2")
var __values []interface{}
__values = append(__values, member_email.value())
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, __values...)
__rows, err := obj.driver.QueryContext(ctx, __stmt, __values...)
if err != nil {
return nil, obj.makeErr(err)
}
defer __rows.Close()
if !__rows.Next() {
if err := __rows.Err(); err != nil {
return nil, obj.makeErr(err)
}
return nil, makeErr(sql.ErrNoRows)
}
member = &Member{}
err = __rows.Scan(&member.Id, &member.Email, &member.Name, &member.PasswordHash, &member.CreatedAt)
if err != nil {
return nil, obj.makeErr(err)
}
if __rows.Next() {
return nil, tooManyRows("Member_By_Email")
}
if err := __rows.Err(); err != nil {
return nil, obj.makeErr(err)
}
return member, nil
}
func (obj *sqlite3Impl) Get_Member_By_Id(ctx context.Context,
member_id Member_Id_Field) (
member *Member, err error) {
defer mon.Task()(&ctx)(&err)
var __embed_stmt = __sqlbundle_Literal("SELECT members.id, members.email, members.name, members.password_hash, members.created_at FROM members WHERE members.id = ?")
var __values []interface{}
__values = append(__values, member_id.value())
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, __values...)
member = &Member{}
err = obj.driver.QueryRowContext(ctx, __stmt, __values...).Scan(&member.Id, &member.Email, &member.Name, &member.PasswordHash, &member.CreatedAt)
if err != nil {
return (*Member)(nil), obj.makeErr(err)
}
return member, nil
}
func (obj *sqlite3Impl) Update_Node_By_Id(ctx context.Context, func (obj *sqlite3Impl) Update_Node_By_Id(ctx context.Context,
node_id Node_Id_Field, node_id Node_Id_Field,
update Node_Update_Fields) ( update Node_Update_Fields) (
@ -1728,67 +1314,6 @@ func (obj *sqlite3Impl) UpdateNoReturn_Node_By_Id(ctx context.Context,
return nil return nil
} }
func (obj *sqlite3Impl) Update_Member_By_Id(ctx context.Context,
member_id Member_Id_Field,
update Member_Update_Fields) (
member *Member, err error) {
defer mon.Task()(&ctx)(&err)
var __sets = &__sqlbundle_Hole{}
var __embed_stmt = __sqlbundle_Literals{Join: "", SQLs: []__sqlbundle_SQL{__sqlbundle_Literal("UPDATE members SET "), __sets, __sqlbundle_Literal(" WHERE members.id = ?")}}
__sets_sql := __sqlbundle_Literals{Join: ", "}
var __values []interface{}
var __args []interface{}
if update.Email._set {
__values = append(__values, update.Email.value())
__sets_sql.SQLs = append(__sets_sql.SQLs, __sqlbundle_Literal("email = ?"))
}
if update.Name._set {
__values = append(__values, update.Name.value())
__sets_sql.SQLs = append(__sets_sql.SQLs, __sqlbundle_Literal("name = ?"))
}
if update.PasswordHash._set {
__values = append(__values, update.PasswordHash.value())
__sets_sql.SQLs = append(__sets_sql.SQLs, __sqlbundle_Literal("password_hash = ?"))
}
if len(__sets_sql.SQLs) == 0 {
return nil, emptyUpdate()
}
__args = append(__args, member_id.value())
__values = append(__values, __args...)
__sets.SQL = __sets_sql
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, __values...)
member = &Member{}
_, err = obj.driver.ExecContext(ctx, __stmt, __values...)
if err != nil {
return nil, obj.makeErr(err)
}
var __embed_stmt_get = __sqlbundle_Literal("SELECT members.id, members.email, members.name, members.password_hash, members.created_at FROM members WHERE members.id = ?")
var __stmt_get = __sqlbundle_Render(obj.dialect, __embed_stmt_get)
obj.logStmt("(IMPLIED) "+__stmt_get, __args...)
err = obj.driver.QueryRowContext(ctx, __stmt_get, __args...).Scan(&member.Id, &member.Email, &member.Name, &member.PasswordHash, &member.CreatedAt)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, obj.makeErr(err)
}
return member, nil
}
func (obj *sqlite3Impl) Delete_Node_By_Id(ctx context.Context, func (obj *sqlite3Impl) Delete_Node_By_Id(ctx context.Context,
node_id Node_Id_Field) ( node_id Node_Id_Field) (
deleted bool, err error) { deleted bool, err error) {
@ -1816,33 +1341,6 @@ func (obj *sqlite3Impl) Delete_Node_By_Id(ctx context.Context,
} }
func (obj *sqlite3Impl) Delete_Member_By_Id(ctx context.Context,
member_id Member_Id_Field) (
deleted bool, err error) {
defer mon.Task()(&ctx)(&err)
var __embed_stmt = __sqlbundle_Literal("DELETE FROM members WHERE members.id = ?")
var __values []interface{}
__values = append(__values, member_id.value())
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, __values...)
__res, err := obj.driver.ExecContext(ctx, __stmt, __values...)
if err != nil {
return false, obj.makeErr(err)
}
__count, err := __res.RowsAffected()
if err != nil {
return false, obj.makeErr(err)
}
return __count > 0, nil
}
func (obj *sqlite3Impl) getLastNode(ctx context.Context, func (obj *sqlite3Impl) getLastNode(ctx context.Context,
pk int64) ( pk int64) (
node *Node, err error) { node *Node, err error) {
@ -1862,25 +1360,6 @@ func (obj *sqlite3Impl) getLastNode(ctx context.Context,
} }
func (obj *sqlite3Impl) getLastMember(ctx context.Context,
pk int64) (
member *Member, err error) {
defer mon.Task()(&ctx)(&err)
var __embed_stmt = __sqlbundle_Literal("SELECT members.id, members.email, members.name, members.password_hash, members.created_at FROM members WHERE _rowid_ = ?")
var __stmt = __sqlbundle_Render(obj.dialect, __embed_stmt)
obj.logStmt(__stmt, pk)
member = &Member{}
err = obj.driver.QueryRowContext(ctx, __stmt, pk).Scan(&member.Id, &member.Email, &member.Name, &member.PasswordHash, &member.CreatedAt)
if err != nil {
return (*Member)(nil), obj.makeErr(err)
}
return member, nil
}
func (impl sqlite3Impl) isConstraintError(err error) ( func (impl sqlite3Impl) isConstraintError(err error) (
constraint string, ok bool) { constraint string, ok bool) {
if e, ok := err.(sqlite3.Error); ok { if e, ok := err.(sqlite3.Error); ok {
@ -1905,16 +1384,6 @@ func (obj *sqlite3Impl) deleteAll(ctx context.Context) (count int64, err error)
return 0, obj.makeErr(err) return 0, obj.makeErr(err)
} }
__count, err = __res.RowsAffected()
if err != nil {
return 0, obj.makeErr(err)
}
count += __count
__res, err = obj.driver.ExecContext(ctx, "DELETE FROM members;")
if err != nil {
return 0, obj.makeErr(err)
}
__count, err = __res.RowsAffected() __count, err = __res.RowsAffected()
if err != nil { if err != nil {
return 0, obj.makeErr(err) return 0, obj.makeErr(err)
@ -1976,20 +1445,6 @@ func (rx *Rx) All_Node(ctx context.Context) (
return tx.All_Node(ctx) return tx.All_Node(ctx)
} }
func (rx *Rx) Create_Member(ctx context.Context,
member_id Member_Id_Field,
member_email Member_Email_Field,
member_name Member_Name_Field,
member_password_hash Member_PasswordHash_Field) (
member *Member, err error) {
var tx *Tx
if tx, err = rx.getTx(ctx); err != nil {
return
}
return tx.Create_Member(ctx, member_id, member_email, member_name, member_password_hash)
}
func (rx *Rx) Create_Node(ctx context.Context, func (rx *Rx) Create_Node(ctx context.Context,
node_id Node_Id_Field, node_id Node_Id_Field,
node_name Node_Name_Field, node_name Node_Name_Field,
@ -2004,16 +1459,6 @@ func (rx *Rx) Create_Node(ctx context.Context,
} }
func (rx *Rx) Delete_Member_By_Id(ctx context.Context,
member_id Member_Id_Field) (
deleted bool, err error) {
var tx *Tx
if tx, err = rx.getTx(ctx); err != nil {
return
}
return tx.Delete_Member_By_Id(ctx, member_id)
}
func (rx *Rx) Delete_Node_By_Id(ctx context.Context, func (rx *Rx) Delete_Node_By_Id(ctx context.Context,
node_id Node_Id_Field) ( node_id Node_Id_Field) (
deleted bool, err error) { deleted bool, err error) {
@ -2024,26 +1469,6 @@ func (rx *Rx) Delete_Node_By_Id(ctx context.Context,
return tx.Delete_Node_By_Id(ctx, node_id) return tx.Delete_Node_By_Id(ctx, node_id)
} }
func (rx *Rx) Get_Member_By_Email(ctx context.Context,
member_email Member_Email_Field) (
member *Member, err error) {
var tx *Tx
if tx, err = rx.getTx(ctx); err != nil {
return
}
return tx.Get_Member_By_Email(ctx, member_email)
}
func (rx *Rx) Get_Member_By_Id(ctx context.Context,
member_id Member_Id_Field) (
member *Member, err error) {
var tx *Tx
if tx, err = rx.getTx(ctx); err != nil {
return
}
return tx.Get_Member_By_Id(ctx, member_id)
}
func (rx *Rx) Get_Node_By_Id(ctx context.Context, func (rx *Rx) Get_Node_By_Id(ctx context.Context,
node_id Node_Id_Field) ( node_id Node_Id_Field) (
node *Node, err error) { node *Node, err error) {
@ -2065,17 +1490,6 @@ func (rx *Rx) UpdateNoReturn_Node_By_Id(ctx context.Context,
return tx.UpdateNoReturn_Node_By_Id(ctx, node_id, update) return tx.UpdateNoReturn_Node_By_Id(ctx, node_id, update)
} }
func (rx *Rx) Update_Member_By_Id(ctx context.Context,
member_id Member_Id_Field,
update Member_Update_Fields) (
member *Member, err error) {
var tx *Tx
if tx, err = rx.getTx(ctx); err != nil {
return
}
return tx.Update_Member_By_Id(ctx, member_id, update)
}
func (rx *Rx) Update_Node_By_Id(ctx context.Context, func (rx *Rx) Update_Node_By_Id(ctx context.Context,
node_id Node_Id_Field, node_id Node_Id_Field,
update Node_Update_Fields) ( update Node_Update_Fields) (
@ -2091,13 +1505,6 @@ type Methods interface {
All_Node(ctx context.Context) ( All_Node(ctx context.Context) (
rows []*Node, err error) rows []*Node, err error)
Create_Member(ctx context.Context,
member_id Member_Id_Field,
member_email Member_Email_Field,
member_name Member_Name_Field,
member_password_hash Member_PasswordHash_Field) (
member *Member, err error)
Create_Node(ctx context.Context, Create_Node(ctx context.Context,
node_id Node_Id_Field, node_id Node_Id_Field,
node_name Node_Name_Field, node_name Node_Name_Field,
@ -2105,22 +1512,10 @@ type Methods interface {
node_api_secret Node_ApiSecret_Field) ( node_api_secret Node_ApiSecret_Field) (
node *Node, err error) node *Node, err error)
Delete_Member_By_Id(ctx context.Context,
member_id Member_Id_Field) (
deleted bool, err error)
Delete_Node_By_Id(ctx context.Context, Delete_Node_By_Id(ctx context.Context,
node_id Node_Id_Field) ( node_id Node_Id_Field) (
deleted bool, err error) deleted bool, err error)
Get_Member_By_Email(ctx context.Context,
member_email Member_Email_Field) (
member *Member, err error)
Get_Member_By_Id(ctx context.Context,
member_id Member_Id_Field) (
member *Member, err error)
Get_Node_By_Id(ctx context.Context, Get_Node_By_Id(ctx context.Context,
node_id Node_Id_Field) ( node_id Node_Id_Field) (
node *Node, err error) node *Node, err error)
@ -2130,11 +1525,6 @@ type Methods interface {
update Node_Update_Fields) ( update Node_Update_Fields) (
err error) err error)
Update_Member_By_Id(ctx context.Context,
member_id Member_Id_Field,
update Member_Update_Fields) (
member *Member, err error)
Update_Node_By_Id(ctx context.Context, Update_Node_By_Id(ctx context.Context,
node_id Node_Id_Field, node_id Node_Id_Field,
update Node_Update_Fields) ( update Node_Update_Fields) (

8
multinode/multinodedb/dbx/multinodedb.dbx.pgx.sql
View File

@ -1,13 +1,5 @@
-- AUTOGENERATED BY storj.io/dbx -- AUTOGENERATED BY storj.io/dbx
-- DO NOT EDIT -- DO NOT EDIT
CREATE TABLE members (
id bytea NOT NULL,
email text NOT NULL,
name text NOT NULL,
password_hash bytea NOT NULL,
created_at timestamp with time zone NOT NULL,
PRIMARY KEY ( id )
);
CREATE TABLE nodes ( CREATE TABLE nodes (
id bytea NOT NULL, id bytea NOT NULL,
name text NOT NULL, name text NOT NULL,

8
multinode/multinodedb/dbx/multinodedb.dbx.sqlite3.sql
View File

@ -1,13 +1,5 @@
-- AUTOGENERATED BY storj.io/dbx -- AUTOGENERATED BY storj.io/dbx
-- DO NOT EDIT -- DO NOT EDIT
CREATE TABLE members (
id BLOB NOT NULL,
email TEXT NOT NULL,
name TEXT NOT NULL,
password_hash BLOB NOT NULL,
created_at TIMESTAMP NOT NULL,
PRIMARY KEY ( id )
);
CREATE TABLE nodes ( CREATE TABLE nodes (
id BLOB NOT NULL, id BLOB NOT NULL,
name TEXT NOT NULL, name TEXT NOT NULL,

117
multinode/multinodedb/members.go
View File

@ -1,117 +0,0 @@
// Copyright (C) 2020 Storj Labs, Inc.
// See LICENSE for copying information.
package multinodedb
import (
"context"
"database/sql"
"errors"
"github.com/zeebo/errs"
"storj.io/common/uuid"
"storj.io/storj/multinode/console"
"storj.io/storj/multinode/multinodedb/dbx"
)
// MembersDBError indicates about internal MembersDB error.
var MembersDBError = errs.Class("MembersDB")
// ensures that members implements console.Members.
var _ console.Members = (*members)(nil)
// members exposes needed by MND MembersDB functionality.
// dbx implementation of console.Members.
//
// architecture: Database
type members struct {
methods dbx.Methods
}
// Invite will create empty row in membersDB.
func (m *members) Invite(ctx context.Context, member console.Member) (err error) {
defer mon.Task()(&ctx)(&err)
id, err := uuid.New()
if err != nil {
return MembersDBError.Wrap(err)
}
_, err = m.methods.Create_Member(ctx, dbx.Member_Id(id[:]), dbx.Member_Email(member.Email), dbx.Member_Name(member.Name), dbx.Member_PasswordHash(member.PasswordHash))
return MembersDBError.Wrap(err)
}
// Update updates all updatable fields of member.
func (m *members) Update(ctx context.Context, member console.Member) (err error) {
defer mon.Task()(&ctx)(&err)
_, err = m.methods.Update_Member_By_Id(ctx, dbx.Member_Id(member.ID[:]), dbx.Member_Update_Fields{
Email: dbx.Member_Email(member.Email),
Name: dbx.Member_Name(member.Name),
PasswordHash: dbx.Member_PasswordHash(member.PasswordHash),
})
return MembersDBError.Wrap(err)
}
// Remove deletes member from membersDB.
func (m *members) Remove(ctx context.Context, id uuid.UUID) (err error) {
defer mon.Task()(&ctx)(&err)
_, err = m.methods.Delete_Member_By_Id(ctx, dbx.Member_Id(id[:]))
return MembersDBError.Wrap(err)
}
// GetByEmail will return member with specified email.
func (m *members) GetByEmail(ctx context.Context, email string) (_ console.Member, err error) {
defer mon.Task()(&ctx)(&err)
memberDbx, err := m.methods.Get_Member_By_Email(ctx, dbx.Member_Email(email))
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return console.Member{}, console.ErrNoMember.Wrap(err)
}
return console.Member{}, MembersDBError.Wrap(err)
}
member, err := fromDBXMember(memberDbx)
return member, MembersDBError.Wrap(err)
}
// GetByID will return member with specified id.
func (m *members) GetByID(ctx context.Context, id uuid.UUID) (_ console.Member, err error) {
defer mon.Task()(&ctx)(&err)
memberDbx, err := m.methods.Get_Member_By_Id(ctx, dbx.Member_Id(id[:]))
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return console.Member{}, console.ErrNoMember.Wrap(err)
}
return console.Member{}, MembersDBError.Wrap(err)
}
member, err := fromDBXMember(memberDbx)
return member, MembersDBError.Wrap(err)
}
// fromDBXMember converts dbx.Member to console.Member.
func fromDBXMember(member *dbx.Member) (_ console.Member, err error) {
id, err := uuid.FromBytes(member.Id)
if err != nil {
return console.Member{}, err
}
result := console.Member{
ID: id,
Email: member.Email,
Name: member.Name,
PasswordHash: member.PasswordHash,
}
return result, nil
}

54
multinode/multinodedb/migrate.go Normal file
View File

@ -0,0 +1,54 @@
// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.
package multinodedb
import (
"storj.io/storj/private/migrate"
)
// SQLite3Migration returns steps needed for migrating sqlite3 database.
func (db *DB) SQLite3Migration() *migrate.Migration {
return &migrate.Migration{
Table: "versions",
Steps: []*migrate.Step{
{
DB: &db.migrationDB,
Description: "Initial setup",
Version: 0,
Action: migrate.SQL{
`CREATE TABLE nodes (
id BLOB NOT NULL,
name TEXT NOT NULL,
public_address TEXT NOT NULL,
api_secret BLOB NOT NULL,
PRIMARY KEY ( id )
); `,
},
},
},
}
}
// PostgresMigration returns steps needed for migrating postgres database.
func (db *DB) PostgresMigration() *migrate.Migration {
return &migrate.Migration{
Table: "versions",
Steps: []*migrate.Step{
{
DB: &db.migrationDB,
Description: "Initial setup",
Version: 0,
Action: migrate.SQL{
`CREATE TABLE nodes (
id bytea NOT NULL,
name text NOT NULL,
public_address text NOT NULL,
api_secret bytea NOT NULL,
PRIMARY KEY ( id )
);`,
},
},
},
}
}

242
multinode/multinodedb/migrate_test.go Normal file
View File

@ -0,0 +1,242 @@
// Copyright (C) 2021 Storj Labs, Inc.
// See LICENSE for copying information.
package multinodedb_test
import (
"context"
"fmt"
"io/ioutil"
"path/filepath"
"strconv"
"strings"
"testing"
"time"
"github.com/stretchr/testify/require"
"github.com/zeebo/errs"
"go.uber.org/zap/zaptest"
"storj.io/common/testcontext"
"storj.io/private/dbutil/dbschema"
"storj.io/private/dbutil/pgtest"
"storj.io/private/dbutil/pgutil"
"storj.io/private/dbutil/sqliteutil"
"storj.io/private/dbutil/tempdb"
"storj.io/storj/multinode/multinodedb"
)
func TestMigrateSQLite3(t *testing.T) {
ctx := testcontext.NewWithTimeout(t, 8*time.Minute)
defer ctx.Cleanup()
log := zaptest.NewLogger(t)
dbURL := "sqlite3://file::memory:"
db, err := multinodedb.Open(ctx, log, dbURL)
require.NoError(t, err)
defer func() {
require.NoError(t, db.Close())
}()
// get snapshots
// find all sqlite3 sql files
matches, err := filepath.Glob("testdata/sqlite3.*")
require.NoError(t, err)
snapshots := new(dbschema.Snapshots)
snapshots.List = make([]*dbschema.Snapshot, len(matches))
for i, match := range matches {
version := parseTestdataVersion(match, "sqlite3")
require.True(t, version >= 0, "invalid testdata file %q: %v", match, err)
scriptData, err := ioutil.ReadFile(match)
require.NoError(t, err, "could not read testdata file for version %d: %v", version, err)
// exec per snapshot??
snapshot, err := sqliteutil.LoadSnapshotFromSQL(ctx, string(scriptData))
require.NoError(t, err)
snapshot.Version = version
snapshots.List[i] = snapshot
}
snapshots.Sort()
// get latest schema
schema, err := sqliteutil.LoadSchemaFromSQL(ctx, db.Schema())
require.NoError(t, err)
var finalSchema *dbschema.Schema
migration := db.SQLite3Migration()
for i, step := range migration.Steps {
tag := fmt.Sprintf("#%d - v%d", i, step.Version)
expected, ok := snapshots.FindVersion(step.Version)
require.True(t, ok)
err = migration.TargetVersion(step.Version).Run(ctx, log)
require.NoError(t, err)
if newData := expected.LookupSection(dbschema.NewData); newData != "" {
_, err = db.ExecContext(ctx, newData)
require.NoError(t, err)
}
currentSchema, err := sqliteutil.QuerySchema(ctx, db)
require.NoError(t, err)
currentSchema.DropTable("versions")
currentData, err := sqliteutil.QueryData(ctx, db, currentSchema)
require.NoError(t, err)
require.Equal(t, expected.Schema, currentSchema, tag)
require.Equal(t, expected.Data, currentData, tag)
finalSchema = currentSchema
}
// verify that we also match the dbx version
require.Equal(t, schema, finalSchema, "result of all migration scripts did not match dbx schema")
}
func TestMigratePostgres(t *testing.T) {
ctx := testcontext.NewWithTimeout(t, 8*time.Minute)
defer ctx.Cleanup()
log := zaptest.NewLogger(t)
connStr := pgtest.PickPostgres(t)
// create tempDB
tempDB, err := tempdb.OpenUnique(ctx, connStr, "migrate")
require.NoError(t, err)
defer func() {
require.NoError(t, tempDB.Close())
}()
db, err := multinodedb.Open(ctx, log, tempDB.ConnStr)
require.NoError(t, err)
defer func() {
require.NoError(t, db.Close())
}()
// get snapshots
// find all postgres sql files
matches, err := filepath.Glob("testdata/postgres.*")
require.NoError(t, err)
snapshots := new(dbschema.Snapshots)
snapshots.List = make([]*dbschema.Snapshot, len(matches))
for i, match := range matches {
version := parseTestdataVersion(match, "postgres")
require.True(t, version >= 0, "invalid testdata file %q: %v", match, err)
scriptData, err := ioutil.ReadFile(match)
require.NoError(t, err, "could not read testdata file for version %d: %v", version, err)
snapshot, err := loadSnapshotFromSQLPostgres(ctx, connStr, string(scriptData))
require.NoError(t, err)
snapshot.Version = version
snapshots.List[i] = snapshot
}
snapshots.Sort()
// get latest schema
schema, err := loadSchemaFromSQLPostgres(ctx, connStr, db.Schema())
require.NoError(t, err)
var finalSchema *dbschema.Schema
migration := db.PostgresMigration()
for i, step := range migration.Steps {
tag := fmt.Sprintf("#%d - v%d", i, step.Version)
expected, ok := snapshots.FindVersion(step.Version)
require.True(t, ok)
err = migration.TargetVersion(step.Version).Run(ctx, log)
require.NoError(t, err)
if newData := expected.LookupSection(dbschema.NewData); newData != "" {
_, err = db.ExecContext(ctx, newData)
require.NoError(t, err)
}
currentSchema, err := pgutil.QuerySchema(ctx, db)
require.NoError(t, err)
currentSchema.DropTable("versions")
currentData, err := pgutil.QueryData(ctx, db, currentSchema)
require.NoError(t, err)
require.Equal(t, expected.Schema, currentSchema, tag)
require.Equal(t, expected.Data, currentData, tag)
finalSchema = currentSchema
}
// verify that we also match the dbx version
require.Equal(t, schema, finalSchema, "result of all migration scripts did not match dbx schema")
}
func parseTestdataVersion(path string, impl string) int {
path = filepath.ToSlash(strings.ToLower(path))
path = strings.TrimPrefix(path, "testdata/"+impl+".v")
path = strings.TrimSuffix(path, ".sql")
v, err := strconv.Atoi(path)
if err != nil {
return -1
}
return v
}
// loadSnapshotFromSQLPostgres inserts script into connstr and loads snapshot for postgres db.
func loadSnapshotFromSQLPostgres(ctx context.Context, connstr, script string) (_ *dbschema.Snapshot, err error) {
db, err := tempdb.OpenUnique(ctx, connstr, "load-schema")
if err != nil {
return nil, err
}
defer func() { err = errs.Combine(err, db.Close()) }()
sections := dbschema.NewSections(script)
_, err = db.ExecContext(ctx, sections.LookupSection(dbschema.Main))
if err != nil {
return nil, err
}
_, err = db.ExecContext(ctx, sections.LookupSection(dbschema.MainData))
if err != nil {
return nil, err
}
_, err = db.ExecContext(ctx, sections.LookupSection(dbschema.NewData))
if err != nil {
return nil, err
}
snapshot, err := pgutil.QuerySnapshot(ctx, db)
if err != nil {
return nil, err
}
snapshot.Sections = sections
return snapshot, nil
}
// loadSnapshotFromSQLPostgres inserts script into connstr and loads schema for postgres db.
func loadSchemaFromSQLPostgres(ctx context.Context, connstr, script string) (_ *dbschema.Schema, err error) {
db, err := tempdb.OpenUnique(ctx, connstr, "load-schema")
if err != nil {
return nil, err
}
defer func() { err = errs.Combine(err, db.Close()) }()
_, err = db.ExecContext(ctx, script)
if err != nil {
return nil, err
}
return pgutil.QuerySchema(ctx, db)
}

2
multinode/multinodedb/multinodedbtest/run.go
View File

@ -144,7 +144,7 @@ func Run(t *testing.T, test func(ctx *testcontext.Context, t *testing.T, db mult
defer ctx.Check(db.Close) defer ctx.Check(db.Close)
err = db.CreateSchema(ctx) err = db.MigrateToLatest(ctx)
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }

15
multinode/multinodedb/testdata/postgres.v0.sql vendored Normal file
View File

@ -0,0 +1,15 @@
-- AUTOGENERATED BY storj.io/dbx
-- DO NOT EDIT
CREATE TABLE nodes (
id bytea NOT NULL,
name text NOT NULL,
public_address text NOT NULL,
api_secret bytea NOT NULL,
PRIMARY KEY ( id )
);
-- MAIN DATA --
-- NEW DATA --
INSERT INTO nodes (id, name, public_address, api_secret) VALUES (E'\\006\\223\\250R\\221\\005\\365\\377v>0\\266\\365\\216\\255?\\347\\244\\371?2\\264\\262\\230\\007<\\001\\262\\263\\237\\247n', 'node_name', '127.0.0.1:13000', E'\\153\\313\\233\\074\\327\\177\\136\\070\\346\\001');

15
multinode/multinodedb/testdata/sqlite3.v0.sql vendored Normal file
View File

@ -0,0 +1,15 @@
-- AUTOGENERATED BY storj.io/dbx
-- DO NOT EDIT
CREATE TABLE nodes (
id BLOB NOT NULL,
name TEXT NOT NULL,
public_address TEXT NOT NULL,
api_secret BLOB NOT NULL,
PRIMARY KEY ( id )
);
-- MAIN DATA --
-- NEW DATA --
INSERT INTO nodes (id, name, public_address, api_secret) VALUES (X'2b3a5863a41f25408a8f5348839d7a1361dbd886d75786bb139a8ca0bdf41000', 'node_name', '127.0.0.1:13000', X'62180593328b8ff3c9f97565fdfd305d');

7
multinode/peer.go
View File

@ -15,7 +15,6 @@ import (
"storj.io/common/peertls/tlsopts" "storj.io/common/peertls/tlsopts"
"storj.io/common/rpc" "storj.io/common/rpc"
"storj.io/private/debug" "storj.io/private/debug"
"storj.io/storj/multinode/console"
"storj.io/storj/multinode/console/server" "storj.io/storj/multinode/console/server"
"storj.io/storj/multinode/nodes" "storj.io/storj/multinode/nodes"
"storj.io/storj/multinode/payouts" "storj.io/storj/multinode/payouts"
@ -32,13 +31,11 @@ var (
type DB interface { type DB interface {
// Nodes returns nodes database. // Nodes returns nodes database.
Nodes() nodes.DB Nodes() nodes.DB
// Members returns members database.
Members() console.Members
// MigrateToLatest initializes the database.
MigrateToLatest(ctx context.Context) error
// Close closes the database. // Close closes the database.
Close() error Close() error
// CreateSchema creates schema.
CreateSchema(ctx context.Context) error
} }
// Config is all the configuration parameters for a Multinode Dashboard. // Config is all the configuration parameters for a Multinode Dashboard.