feat: add modular SQLC tools for DB definitions

Author: ezdac

rolling-shutter/go.mod

@@ -44,6 +44,7 @@ require (
 	golang.org/x/crypto v0.12.0
 	golang.org/x/sync v0.3.0
 	google.golang.org/protobuf v1.30.0
+	gopkg.in/yaml.v3 v3.0.1
 	gotest.tools v2.2.0+incompatible
 	gotest.tools/v3 v3.2.0
 )
@@ -226,7 +227,6 @@ require (
 	gopkg.in/ini.v1 v1.67.0 // indirect
 	gopkg.in/natefinch/npipe.v2 v2.0.0-20160621034901-c1b8fa8bdcce // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
-	gopkg.in/yaml.v3 v3.0.1 // indirect
 	lukechampine.com/blake3 v1.2.1 // indirect
 )
 

rolling-shutter/medley/db/connect.go

@@ -0,0 +1,43 @@
+package db
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog/log"
+
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/medley/service"
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/shdb"
+)
+
+func ValidateDBVersion(ctx context.Context, dbpool *pgxpool.Pool, role string) error {
+	err := dbpool.BeginFunc(
+		ctx,
+		func(tx pgx.Tx) error {
+			return ValidateDatabaseVersion(ctx, tx, role)
+		},
+	)
+	if err != nil {
+		return errors.Wrap(err, "database is used for a different role already, preventing overwrite")
+	}
+	return nil
+}
+
+// Connect to the database `url` from within a runner.Start() method
+// and create the pgxpool.Pool.
+func Connect(ctx context.Context, runner service.Runner, url, version string) (*pgxpool.Pool, error) {
+	dbpool, err := pgxpool.Connect(ctx, url)
+	if err != nil {
+		return nil, err
+	}
+	runner.Defer(dbpool.Close)
+
+	err = ValidateDBVersion(ctx, dbpool, version)
+	if err != nil {
+		return nil, err
+	}
+	shdb.AddConnectionInfo(log.Info(), dbpool).Msg("connected to database")
+	return dbpool, nil
+}

rolling-shutter/medley/db/db.sqlc.gen.go

@@ -0,0 +1,21 @@
+package db
+
+import (
+	"embed"
+
+	"github.com/rs/zerolog/log"
+)
+
+//go:generate sqlc generate --file sql/sqlc.yaml
+
+//go:embed sql
+var files embed.FS
+var metaDefinition Definition
+
+func init() {
+	var err error
+	metaDefinition, err = NewSQLCDefinition(files, "sql/", "meta", 1)
+	if err != nil {
+		log.Fatal().Err(err).Msg("failed to initialize DB metadata")
+	}
+}

rolling-shutter/medley/db/definition.go

@@ -0,0 +1,140 @@
+package db
+
+import (
+	"context"
+
+	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/pgxpool"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog/log"
+
+	"github.com/shutter-network/rolling-shutter/rolling-shutter/shdb"
+)
+
+// InitDB initializes an empty database with the all schema definitions as specified from
+// the passed in `Definition`'s Create() and Init() actions.
+// Additionally, a `role` will be written in the database's meta key/value store, that
+// pins the database to a specific role, e.g. "keyper-test" or "snapshot-keyper-production"
+// in order to prevent later usage of the database with commands that fulfill a different role.
+func InitDB(ctx context.Context, dbpool *pgxpool.Pool, role string, definition Definition) error {
+	err := dbpool.BeginFunc(WrapContext(ctx, definition.Validate))
+	if err == nil {
+		shdb.AddConnectionInfo(log.Info(), dbpool).Msg("database already exists")
+		return nil
+	} else if errors.Is(err, ErrValueMismatch) {
+		return err
+	}
+
+	err = dbpool.BeginFunc(WrapContext(ctx, definition.Create))
+	if err != nil {
+		return err
+	}
+
+	err = dbpool.BeginFunc(WrapContext(ctx, definition.Init))
+	if err != nil {
+		return err
+	}
+
+	// For the outer DB initialisation, also set the database version to
+	// the overall "role", so that e.g. a snapshot keyper database won't be
+	// used by another keyper implementation, no matter if the schemas
+	// are compatible
+	err = dbpool.BeginFunc(
+		ctx,
+		func(tx pgx.Tx) error {
+			return InsertDBVersion(ctx, tx, role)
+		},
+	)
+	if err != nil {
+		return err
+	}
+	shdb.AddConnectionInfo(log.Info(), dbpool).Msg("database initialized")
+	return nil
+}
+
+var _ Definition = AggregateDefinition{}
+
+// NewAggregateDefinition constructs a new AggregateDefinition instance
+// and implements the `Definition` interface.
+// The passed in `definitions` will be stored in the AggregateDefinition's
+// state and a call to the AggregateDefinition's methods will be dispatched to all
+// stored definitions.
+// Importantly, this constructor will unpack the internal definitions of any passed in,
+// wrapped AggregateDefinition. This means that the outer AggregateDefinition always stores
+// the flattened set of all underlying child-definitions.
+func NewAggregateDefinition(name string, definitions ...Definition) AggregateDefinition {
+	defs := map[Definition]bool{}
+	for _, def := range definitions {
+		nestedDefs, isWrapper := def.(AggregateDefinition)
+		if isWrapper {
+			for deff := range nestedDefs.defs {
+				defs[deff] = true
+			}
+		} else {
+			defs[def] = true
+		}
+	}
+	return AggregateDefinition{name: name, defs: defs}
+}
+
+type AggregateDefinition struct {
+	name string
+	defs map[Definition]bool
+}
+
+func (d AggregateDefinition) Name() string {
+	return d.name
+}
+
+func (d AggregateDefinition) Init(ctx context.Context, tx pgx.Tx) error {
+	for def := range d.defs {
+		err := def.Init(ctx, tx)
+		if err != nil {
+			return errors.Wrapf(err, "can't initialize DB for definition '%s'", def.Name())
+		}
+	}
+	return nil
+}
+
+func (d AggregateDefinition) Create(ctx context.Context, tx pgx.Tx) error {
+	err := metaDefinition.Create(ctx, tx)
+	if err != nil {
+		return errors.Wrap(err, "can't create DB for meta definition")
+	}
+	for def := range d.defs {
+		err := def.Create(ctx, tx)
+		if err != nil {
+			return errors.Wrapf(err, "can't create DB for definition '%s'", def.Name())
+		}
+	}
+	return nil
+}
+
+func (d AggregateDefinition) Validate(ctx context.Context, tx pgx.Tx) error {
+	for def := range d.defs {
+		err := def.Validate(ctx, tx)
+		if err != nil {
+			return errors.Wrapf(err, "validation error for DB '%s'", def.Name())
+		}
+	}
+	return nil
+}
+
+type Definition interface {
+	Name() string
+	Create(context.Context, pgx.Tx) error
+	Init(context.Context, pgx.Tx) error
+	Validate(context.Context, pgx.Tx) error
+}
+
+type Schema struct {
+	Version int
+	Name    string
+	Path    string
+}
+
+type Migration struct {
+	Version int
+	Path    string
+	Up      bool // up or down migration
+}

rolling-shutter/medley/db/definitions.go

@@ -0,0 +1,69 @@
+package db
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/jackc/pgx/v4"
+	"github.com/pkg/errors"
+	"github.com/rs/zerolog/log"
+)
+
+var (
+	ErrValueMismatch = errors.New("database has unexpected value")
+	ErrKeyNotFound   = errors.New("key does not exist")
+)
+
+var DatabaseVersionKey string = "database-version"
+
+// MakeSchemaVersionKey generates the version key for the metadb
+// that is used to check database schema compatibility.
+// `definitionName` is the name of the database definition that corresponds
+// to a database subset (one entry in an sqlc file).
+// `schemaName` corresponds to the individual *.sql files in the schema folder.
+func MakeSchemaVersionKey(definitionName, schemaName string) string {
+	return "schema-version-" + definitionName + "-" + schemaName
+}
+
+func InsertDBVersion(ctx context.Context, tx pgx.Tx, version string) error {
+	return insertMetaInf(ctx, tx, DatabaseVersionKey, version)
+}
+
+func InsertSchemaVersion(ctx context.Context, tx pgx.Tx, definitionName string, schema Schema) error {
+	return insertMetaInf(ctx, tx, MakeSchemaVersionKey(definitionName, schema.Name), fmt.Sprint(schema.Version))
+}
+
+func insertMetaInf(ctx context.Context, tx pgx.Tx, key, val string) error {
+	log.Info().Str("key", key).
+		Str("value", val).Msg("insert schema meta inf")
+	return New(tx).InsertMeta(ctx, InsertMetaParams{
+		Key:   key,
+		Value: val,
+	})
+}
+
+// ValidateSchemaVersion checks that the database schema is compatible.
+func ValidateSchemaVersion(ctx context.Context, tx pgx.Tx, definitionName string, schema Schema) error {
+	return expectMetaKeyVal(ctx, tx, MakeSchemaVersionKey(definitionName, schema.Name), fmt.Sprint(schema.Version))
+}
+
+func expectMetaKeyVal(ctx context.Context, tx pgx.Tx, key, val string) error {
+	haveVal, err := New(tx).GetMeta(ctx, key)
+	if err == pgx.ErrNoRows {
+		return errors.Wrapf(ErrKeyNotFound, "key: %s", key)
+	} else if err != nil {
+		return errors.Wrapf(err, "failed to get key '%s' from meta_inf table", key)
+	}
+	if haveVal != val {
+		return errors.Wrapf(ErrValueMismatch, "expected %s, have %s", val, haveVal)
+	}
+	return nil
+}
+
+// ValidateDatabaseVersion checks that the overall database is compatible.
+// This corresponds to the "role" of the database, e.g. a snapshot-keyper
+// might not be compatible with a snapshot test-keyper, even if the schema's
+// versions would match exactly.
+func ValidateDatabaseVersion(ctx context.Context, tx pgx.Tx, version string) error {
+	return expectMetaKeyVal(ctx, tx, DatabaseVersionKey, version)
+}

rolling-shutter/medley/db/meta.sqlc.gen.go

@@ -0,0 +1,5 @@
+-- name: InsertMeta :exec
+INSERT INTO meta_inf (key, value) VALUES ($1, $2);
+
+-- name: GetMeta :one
+SELECT value FROM meta_inf WHERE key = $1;