Added dbschema

db.go

@@ -3,6 +3,7 @@ package main
 import (
 	// External
 	"github.com/jackc/pgx/v5/pgxpool"
+	"smon/dbschema"
 
 	// Standard
 	"context"
@@ -14,6 +15,8 @@ var (
 )
 
 func initDb(host string, port int, name, username, password string) (err error) {
+	logger.Info("database", "host", host, "port", port, "name", name, "username", username)
+
 	// The main database connection is configured.
 	connString := fmt.Sprintf(
 		"host=%s port=%d user=%s password=%s dbname=%s sslmode=disable",
@@ -27,17 +30,28 @@ func initDb(host string, port int, name, username, password string) (err error)
 
 	// The schema upgrader keeps track of the SQL schema version and
 	// runs SQL scripts to upgrade it if necessary.
-	/*
-	upgrader := dbschema.NewUpgrader("_jsonstore")
+	upgrader := dbschema.NewUpgrader()
 	upgrader.SetLogCallback(sqlLogCallback)
 	upgrader.SetSqlCallback(sqlSourceCallback)
 
-	_, err = upgrader.AddDatabaseInstance(db, config.Database.Db)
+	_, err = upgrader.AddDatabaseInstance(db, config.Database.Name)
 	if err != nil {
 		return
 	}
 
 	err = upgrader.Run()
-	*/
 	return
 }
+
+func sqlLogCallback(op, msg string) {
+	logger.Info("dbschema", "op", op, "message", msg)
+}
+
+func sqlSourceCallback(dbName string, version int) ([]byte, bool) {
+	fname := fmt.Sprintf("sql/%05d.sql", version)
+	data, err := sqlFS.ReadFile(fname)
+	if err != nil {
+		return []byte{}, false
+	}
+	return data, true
+}

dbschema/database.go

@@ -0,0 +1,38 @@
+package dbschema
+
+import (
+	// External
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	// Standard
+	"context"
+	"fmt"
+)
+
+func newDatabase(host string, port int, dbName, user, pass string) (dbase Database, err error) {// {{{
+	dbase.Host = host
+	dbase.Port = port
+	dbase.DbName = dbName
+	dbase.Username = user
+	dbase.Password = pass
+
+	dbase.db, err = pgxpool.New(context.Background(), dbase.sqlConnString())
+	return
+}// }}}
+func databaseFromInstance(db *pgxpool.Pool) (dbase Database, err error) {
+	dbase.db = db
+	return
+}
+
+func (dbase Database) sqlConnString() string {// {{{
+	return fmt.Sprintf(
+		"host=%s port=%d user=%s password=%s dbname=%s sslmode=disable",
+		dbase.Host,
+		dbase.Port,
+		dbase.Username,
+		dbase.Password,
+		dbase.DbName,
+	)
+}// }}}
+
+// vim: foldmethod=marker

dbschema/pkg.go

@@ -0,0 +1,198 @@
+package dbschema
+
+import (
+	// External
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgconn"
+	"github.com/jackc/pgx/v5/pgxpool"
+
+	// Standard
+	"context"
+	"fmt"
+)
+
+func defaultCallback(topic, msg string) { // {{{
+	fmt.Printf("[%s] %s\n", topic, msg)
+} // }}}
+
+// NewUpgrader creates an upgrader with an empty list of databases.
+func NewUpgrader(schema ...string) (upgrader Upgrader) { // {{{
+	// Using a variadic function for backward compatibility.
+	if len(schema) > 0 {
+		upgrader.schema = schema[0]
+	} else {
+		upgrader.schema = "_db"
+	}
+
+	upgrader.logCallback = defaultCallback
+	upgrader.databases = map[string]Database{}
+	return
+} // }}}
+
+// SetLogCallback allows to set a callback for custom logging.
+func (upgrader *Upgrader) SetLogCallback(callback func(string, string)) { // {{{
+	upgrader.logCallback = callback
+} // }}}
+// SetSqlCallback is required for providing the SQL schema updates.
+func (upgrader *Upgrader) SetSqlCallback(callback func(string, int) ([]byte, bool)) { // {{{
+	upgrader.sqlCallback = callback
+} // }}}
+// Version returns the current dbschema version for the given database name.
+func (upgrader *Upgrader) Version(dbName string) (version int, err error) { // {{{
+	dbase, found := upgrader.databases[dbName]
+	if !found {
+		err = fmt.Errorf("Database %s not previously added to the upgrader", dbName)
+		return
+	}
+
+	version, err = dbase.Version()
+	return
+} // }}}
+
+func (dbase Database) createSchemaTable() (err error) { // {{{
+	dbase.upgrader.logCallback("create", fmt.Sprintf("%s, %s.schema", dbase.DbName, dbase.upgrader.schema))
+	_, err = dbase.db.Exec(context.Background(), `CREATE SCHEMA "` + dbase.upgrader.schema + `"`)
+
+	// Error code 42P06 "duplicate_schema" is an OK error,
+	// table can still be missing and created.
+	pqErr, _ := err.(*pgconn.PgError)
+	if pqErr != nil && pqErr.Code != "42P06" {
+		return
+	}
+
+	_, err = dbase.db.Exec(
+		context.Background(),
+		`CREATE TABLE "` + dbase.upgrader.schema + `"."schema" (
+			version  int4      NOT NULL,
+			updated  timestamp NOT NULL DEFAULT NOW(),
+
+			CONSTRAINT schema_pk PRIMARY KEY (version)
+		)`,
+	)
+	return
+} // }}}
+func (dbase Database) appendSchemaVersion(version int) (err error) { // {{{
+	_, err = dbase.db.Exec(context.Background(), `INSERT INTO `+dbase.upgrader.schema+`.schema(version) VALUES($1)`, version)
+	return
+} // }}}
+
+func (dbase Database) verifySchemaTable() (err error) { // {{{
+	var rows pgx.Rows
+	if rows, err = dbase.db.Query(
+		context.Background(),
+		`SELECT EXISTS (
+			SELECT FROM pg_catalog.pg_class c
+			JOIN   pg_catalog.pg_namespace n ON n.oid = c.relnamespace
+			WHERE  n.nspname = '` + dbase.upgrader.schema + `'
+			AND    c.relname = 'schema'
+		)`,
+	); err != nil {
+		return
+	}
+	defer rows.Close()
+	var exists bool
+	rows.Next()
+	if err = rows.Scan(&exists); err != nil {
+		return
+	}
+
+	if exists {
+		return
+	}
+	err = dbase.createSchemaTable()
+	return
+} // }}}
+func (dbase Database) verifySchemaEntry() (err error) { // {{{
+	var version int
+	var row pgx.Row
+	row = dbase.db.QueryRow(context.Background(), `SELECT version FROM `+dbase.upgrader.schema+`.schema LIMIT 1`)
+
+	err = row.Scan(&version)
+	if err == pgx.ErrNoRows {
+		dbase.upgrader.logCallback("initiate version", dbase.DbName)
+		err = dbase.appendSchemaVersion(0)
+	}
+
+	return
+} // }}}
+func (dbase Database) Version() (version int, err error) { // {{{
+	var rows pgx.Rows
+	rows, err = dbase.db.Query(
+		context.Background(),
+		`SELECT version FROM ` + dbase.upgrader.schema + `.schema ORDER BY version DESC LIMIT 1`,
+	)
+	if err != nil {
+		return
+	}
+	defer rows.Close()
+
+	if rows.Next() {
+		err = rows.Scan(&version)
+	} else {
+		err = fmt.Errorf(`Database "%s" is missing an entry in `+dbase.upgrader.schema+`.schema`, dbase.DbName)
+	}
+	return
+} // }}}
+
+// AddDatabase sets a database up for the Run() function with verifying/creating the _db.schema table.
+func (upgrader Upgrader) AddDatabase(host string, port int, dbName, user, pass string) (db Database, err error) { // {{{
+	if db, err = newDatabase(host, port, dbName, user, pass); err != nil {
+		return
+	}
+	db.upgrader = &upgrader
+
+	upgrader.databases[dbName] = db
+
+	if err = db.verifySchemaTable(); err != nil {
+		return
+	}
+
+	err = db.verifySchemaEntry()
+	return
+} // }}}
+func (upgrader Upgrader) AddDatabaseInstance(sqlDB *pgxpool.Pool, dbName string) (db Database, err error) { // {{{
+	db, err = databaseFromInstance(sqlDB)
+
+	db.upgrader = &upgrader
+
+	upgrader.databases[dbName] = db
+
+	if err = db.verifySchemaTable(); err != nil {
+		return
+	}
+
+	err = db.verifySchemaEntry()
+	return
+} // }}}
+
+// Run executes the actual schema updates until there are no more available.
+func (upgrader Upgrader) Run() (err error) { // {{{
+	var version int
+
+	for dbName, dbase := range upgrader.databases {
+		version, err = dbase.Version()
+		if err != nil {
+			return
+		}
+		upgrader.logCallback("version", fmt.Sprintf("%s.%s: %d", dbName, upgrader.schema, version))
+
+		for {
+			version++
+			sql, found := upgrader.sqlCallback(dbName, version)
+			if !found {
+				break
+			}
+
+			upgrader.logCallback("exec", fmt.Sprintf("%s.%s: %d", dbName, upgrader.schema, version))
+			if _, err = dbase.db.Exec(context.Background(), string(sql)); err != nil {
+				return
+			}
+			if err = dbase.appendSchemaVersion(version); err != nil {
+				return
+			}
+		}
+	}
+	return
+} // }}}
+
+// vim: foldmethod=marker

dbschema/schema.go

@@ -0,0 +1,46 @@
+/*
+Package dbschema is used to keep the SQL schema up to date.
+
+	func sqlProvider(dbName string, version int) (sql []byte, found bool) {
+		// read an SQL file and return the contents
+		return
+	}
+
+	upgrader := dbschema.NewUpgrader()
+	upgrader.SetSqlCallback(sqlProvider)
+
+	if err = upgrader.AddDatabase("127.0.0.1", 5432, "foo", "postgres", "password"); err != nil {
+		panic(err)
+	}
+
+	if err = upgrader.Run(); err != nil {
+		panic(err)
+	}
+*/
+package dbschema
+
+import (
+	// External
+	"github.com/jackc/pgx/v5/pgxpool"
+)
+
+// An upgrader verifies the schema for one or more databases and upgrades them if possible.
+type Upgrader struct {
+	schema      string
+	databases   map[string]Database
+	logCallback func(string, string)
+	sqlCallback func(string, int) ([]byte, bool)
+}
+
+type Database struct {
+	Host     string
+	Port     int
+	DbName   string
+	Username string
+	Password string
+
+	db       *pgxpool.Pool
+	upgrader *Upgrader
+}
+
+// vim: foldmethod=marker