Use schema/ package in user/ package

user/manager.go

@@ -33,6 +33,7 @@ const (
 	tokenLength                     = 32
 	tokenMaxCount                   = 60 // Only keep this many tokens in the table per user
 	tag                             = "user_manager"
+	schemaStore                     = "user" // Store name in the schema_version table (see db/schema)
 )
 
 // Default constants that may be overridden by configs

user/manager_postgres.go

@@ -5,6 +5,7 @@ import (
 	"strings"
 
 	"heckel.io/ntfy/v2/db"
+	"heckel.io/ntfy/v2/db/schema"
 )
 
 // PostgreSQL queries
@@ -348,7 +349,7 @@ var postgresQueries = queries{
 
 // NewPostgresManager creates a new Manager backed by a PostgreSQL database
 func NewPostgresManager(d *db.DB, config *Config) (*Manager, error) {
-	if err := setupPostgres(d.Primary()); err != nil {
+	if err := schema.Migrate(d.Primary(), schema.Postgres, schemaStore, postgresCurrentSchemaVersion, postgresCreateTables, postgresMigrations); err != nil {
 		return nil, err
 	}
 	return newManager(d, postgresQueries, config)

user/manager_postgres_schema.go

@@ -1,8 +1,7 @@
 package user
 
 import (
-	"database/sql"
+	"heckel.io/ntfy/v2/db/schema"
-	"fmt"
 )
 
 // Initial PostgreSQL schema
@@ -90,21 +89,14 @@ const (
 			PRIMARY KEY (token_hash)
 		);
 		CREATE INDEX idx_magic_link_user_kind ON user_magic_link (user_id, kind);
-		CREATE TABLE IF NOT EXISTS schema_version (
-			store TEXT PRIMARY KEY,
-			version INT NOT NULL
-		);
 		INSERT INTO "user" (id, user_name, pass, role, sync_topic, provisioned, created)
 		VALUES ('` + everyoneID + `', '*', '', 'anonymous', '', false, EXTRACT(EPOCH FROM NOW())::BIGINT)
 		ON CONFLICT (id) DO NOTHING;
 	`
 )
 
-// Schema table management queries for Postgres
 const (
 	postgresCurrentSchemaVersion = 8
-	postgresSelectSchemaVersionQuery = `SELECT version FROM schema_version WHERE store = 'user'`
-	postgresInsertSchemaVersionQuery = `INSERT INTO schema_version (store, version) VALUES ('user', $1)`
 )
 
 const (
@@ -133,62 +125,15 @@ const (
 		);
 		CREATE INDEX idx_magic_link_user_kind ON user_magic_link (user_id, kind);
 	`
-	postgresUpdateSchemaVersionQuery = `UPDATE schema_version SET version = $1 WHERE store = 'user'`
 )
 
-var postgresMigrations = map[int]func(db *sql.DB) error{
+var (
-	6: postgresMigrateFrom6,
+	postgresCreateTables = schema.AsMigrateFunc(postgresCreateTablesQueries)
-	7: postgresMigrateFrom7,
-}
 
-func setupPostgres(db *sql.DB) error {
+	// postgresMigrations maps a schema version to the migration upgrading it to the next
-	var schemaVersion int
+	// version. Always append migrations at the end, never insert in the middle.
-	err := db.QueryRow(postgresSelectSchemaVersionQuery).Scan(&schemaVersion)
+	postgresMigrations = map[int]schema.MigrateFunc{
-	if err != nil {
+		6: schema.AsMigrateFunc(postgresMigrate6To7UpdateQueries),
-		return setupNewPostgres(db)
+		7: schema.AsMigrateFunc(postgresMigrate7To8UpdateQueries),
 	}
-	if schemaVersion == postgresCurrentSchemaVersion {
+)
-		return nil
-	} else if schemaVersion > postgresCurrentSchemaVersion {
-		return fmt.Errorf("unexpected schema version: version %d is higher than current version %d", schemaVersion, postgresCurrentSchemaVersion)
-	}
-	for i := schemaVersion; i < postgresCurrentSchemaVersion; i++ {
-		fn, ok := postgresMigrations[i]
-		if !ok {
-			return fmt.Errorf("cannot find migration step from schema version %d to %d", i, i+1)
-		} else if err := fn(db); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func postgresMigrateFrom6(db *sql.DB) error {
-	if _, err := db.Exec(postgresMigrate6To7UpdateQueries); err != nil {
-		return err
-	}
-	if _, err := db.Exec(postgresUpdateSchemaVersionQuery, 7); err != nil {
-		return err
-	}
-	return nil
-}
-
-func postgresMigrateFrom7(db *sql.DB) error {
-	if _, err := db.Exec(postgresMigrate7To8UpdateQueries); err != nil {
-		return err
-	}
-	if _, err := db.Exec(postgresUpdateSchemaVersionQuery, 8); err != nil {
-		return err
-	}
-	return nil
-}
-
-func setupNewPostgres(db *sql.DB) error {
-	if _, err := db.Exec(postgresCreateTablesQueries); err != nil {
-		return err
-	}
-	if _, err := db.Exec(postgresInsertSchemaVersionQuery, postgresCurrentSchemaVersion); err != nil {
-		return err
-	}
-	return nil
-}

user/manager_sqlite.go

@@ -9,6 +9,7 @@ import (
 	_ "github.com/mattn/go-sqlite3" // SQLite driver
 
 	"heckel.io/ntfy/v2/db"
+	"heckel.io/ntfy/v2/db/schema"
 	"heckel.io/ntfy/v2/util"
 )
 
@@ -351,14 +352,17 @@ func NewSQLiteManager(filename, startupQueries string, config *Config) (*Manager
 	// Open with case-sensitive LIKE. ACL topic matching is done via LIKE (see
 	// selectTopicPerms), and SQLite's LIKE is case-insensitive for ASCII by
 	// default -- without this, an ACL rule for "secret" would also match a
-	// request for "SECRET", which is a security iisue. PostgreSQL's LIKE is
+	// request for "SECRET", which is a security issue. PostgreSQL's LIKE is
 	// already case-sensitive, so this only affects SQLite. The pragma is
 	// applied to every pooled connection by the driver.
 	d, err := sql.Open("sqlite3", fmt.Sprintf("%s?_case_sensitive_like=on", filename))
 	if err != nil {
 		return nil, err
 	}
-	if err := setupSQLite(d); err != nil {
+	// Migrations must run before the startup queries: the 5 -> 6 table rebuilds rely on
+	// foreign keys being OFF, which is only guaranteed on fresh connections (the foreign_keys
+	// pragma is enabled as part of the builtin startup queries below)
+	if err := schema.Migrate(d, schema.SQLite, schemaStore, sqliteCurrentSchemaVersion, sqliteCreateTables, sqliteMigrations); err != nil {
 		return nil, err
 	}
 	if err := runSQLiteStartupQueries(d, startupQueries); err != nil {

user/manager_sqlite_schema.go

@@ -2,10 +2,8 @@ package user
 
 import (
 	"database/sql"
-	"fmt"
 
-	"heckel.io/ntfy/v2/db"
+	"heckel.io/ntfy/v2/db/schema"
-	"heckel.io/ntfy/v2/log"
 	"heckel.io/ntfy/v2/util"
 )
 
@@ -105,10 +103,6 @@ const (
 			FOREIGN KEY (user_id) REFERENCES user (id) ON DELETE CASCADE
 		);
 		CREATE INDEX idx_magic_link_user_kind ON user_magic_link (user_id, kind);
-		CREATE TABLE IF NOT EXISTS schemaVersion (
-			id INT PRIMARY KEY,
-			version INT NOT NULL
-		);
 		INSERT INTO user (id, user, pass, role, sync_topic, provisioned, created)
 		VALUES ('` + everyoneID + `', '*', '', 'anonymous', '', false, UNIXEPOCH())
 		ON CONFLICT (id) DO NOTHING;
@@ -119,12 +113,8 @@ const (
 	sqliteBuiltinStartupQueries = `PRAGMA foreign_keys = ON;`
 )
 
-// Schema version table management for SQLite
 const (
 	sqliteCurrentSchemaVersion = 8
-	sqliteInsertSchemaVersionQuery = `INSERT INTO schemaVersion VALUES (1, ?)`
-	sqliteUpdateSchemaVersionQuery = `UPDATE schemaVersion SET version = ? WHERE id = 1`
-	sqliteSelectSchemaVersionQuery = `SELECT version FROM schemaVersion WHERE id = 1`
 )
 
 // Schema migrations for SQLite
@@ -190,10 +180,6 @@ const (
 			PRIMARY KEY (user_id, token),
 			FOREIGN KEY (user_id) REFERENCES user (id) ON DELETE CASCADE
 		);
-		CREATE TABLE IF NOT EXISTS schemaVersion (
-			id INT PRIMARY KEY,
-			version INT NOT NULL
-		);
 		INSERT INTO user (id, user, pass, role, sync_topic, created)
 		VALUES ('u_everyone', '*', '', 'anonymous', '', UNIXEPOCH())
 		ON CONFLICT (id) DO NOTHING;
@@ -270,10 +256,11 @@ const (
 		CREATE INDEX idx_magic_link_user_kind ON user_magic_link (user_id, kind);
 	`
 
-	// 5 -> 6
+	// 5 -> 6. The table rebuilds below rely on foreign keys being OFF (otherwise RENAME
+	// would rewrite the childrens' REFERENCES clauses to point at the _old tables). This is
+	// guaranteed because migrations run on fresh connections, before the startup queries
+	// enable the foreign_keys pragma; see NewSQLiteManager.
 	sqliteMigrate5To6UpdateQueries = `
-		PRAGMA foreign_keys=off;
-
 		-- Alter user table: Add provisioned column
 		ALTER TABLE user RENAME TO user_old;
 		CREATE TABLE IF NOT EXISTS user (
@@ -359,57 +346,25 @@ const (
 		CREATE UNIQUE INDEX idx_user_stripe_customer_id ON user (stripe_customer_id);
 		CREATE UNIQUE INDEX idx_user_stripe_subscription_id ON user (stripe_subscription_id);
 		CREATE UNIQUE INDEX idx_user_token ON user_token (token);
-
-		-- Re-enable foreign keys
-		PRAGMA foreign_keys=on;
 	`
 )
 
 var (
-	sqliteMigrations = map[int]func(db *sql.DB) error{
+	sqliteCreateTables = schema.AsMigrateFunc(sqliteCreateTablesQueries)
+
+	// sqliteMigrations maps a schema version to the migration upgrading it to the next
+	// version. Always append migrations at the end, never insert in the middle.
+	sqliteMigrations = map[int]schema.MigrateFunc{
 		1: sqliteMigrateFrom1,
-		2: sqliteMigrateFrom2,
+		2: schema.AsMigrateFunc(sqliteMigrate2To3UpdateQueries),
-		3: sqliteMigrateFrom3,
+		3: schema.AsMigrateFunc(sqliteMigrate3To4UpdateQueries),
-		4: sqliteMigrateFrom4,
+		4: schema.AsMigrateFunc(sqliteMigrate4To5UpdateQueries),
-		5: sqliteMigrateFrom5,
+		5: schema.AsMigrateFunc(sqliteMigrate5To6UpdateQueries),
-		6: sqliteMigrateFrom6,
+		6: schema.AsMigrateFunc(sqliteMigrate6To7UpdateQueries),
-		7: sqliteMigrateFrom7,
+		7: schema.AsMigrateFunc(sqliteMigrate7To8UpdateQueries),
 	}
 )
 
-func setupSQLite(db *sql.DB) error {
-	var schemaVersion int
-	if err := db.QueryRow(sqliteSelectSchemaVersionQuery).Scan(&schemaVersion); err != nil {
-		return setupNewSQLite(db)
-	}
-	if schemaVersion == sqliteCurrentSchemaVersion {
-		return nil
-	} else if schemaVersion > sqliteCurrentSchemaVersion {
-		return fmt.Errorf("unexpected schema version: version %d is higher than current version %d", schemaVersion, sqliteCurrentSchemaVersion)
-	}
-	for i := schemaVersion; i < sqliteCurrentSchemaVersion; i++ {
-		fn, ok := sqliteMigrations[i]
-		if !ok {
-			return fmt.Errorf("cannot find migration step from schema version %d to %d", i, i+1)
-		} else if err := fn(db); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
-func setupNewSQLite(sqlDB *sql.DB) error {
-	return db.ExecTx(sqlDB, func(tx *sql.Tx) error {
-		if _, err := tx.Exec(sqliteCreateTablesQueries); err != nil {
-			return err
-		}
-		if _, err := tx.Exec(sqliteInsertSchemaVersionQuery, sqliteCurrentSchemaVersion); err != nil {
-			return err
-		}
-		return nil
-	})
-}
-
 func runSQLiteStartupQueries(db *sql.DB, startupQueries string) error {
 	if _, err := db.Exec(sqliteBuiltinStartupQueries); err != nil {
 		return err
@@ -422,9 +377,7 @@ func runSQLiteStartupQueries(db *sql.DB, startupQueries string) error {
 	return nil
 }
 
-func sqliteMigrateFrom1(sqlDB *sql.DB) error {
+func sqliteMigrateFrom1(tx *sql.Tx) error {
-	log.Tag(tag).Info("Migrating user database schema: from 1 to 2")
-	return db.ExecTx(sqlDB, func(tx *sql.Tx) error {
 	// Rename user -> user_old, and create new tables
 	if _, err := tx.Exec(sqliteMigrate1To2CreateTablesQueries); err != nil {
 		return err
@@ -454,90 +407,6 @@ func sqliteMigrateFrom1(sqlDB *sql.DB) error {
 		}
 	}
 	// Migrate old "access" table to "user_access" and drop "access" and "user_old"
-		if _, err := tx.Exec(sqliteMigrate1To2InsertFromOldTablesAndDropNoTxQuery); err != nil {
+	_, err = tx.Exec(sqliteMigrate1To2InsertFromOldTablesAndDropNoTxQuery)
 	return err
-		}
-		if _, err := tx.Exec(sqliteUpdateSchemaVersionQuery, 2); err != nil {
-			return err
-		}
-		return nil
-	})
-}
-
-func sqliteMigrateFrom2(sqlDB *sql.DB) error {
-	log.Tag(tag).Info("Migrating user database schema: from 2 to 3")
-	return db.ExecTx(sqlDB, func(tx *sql.Tx) error {
-		if _, err := tx.Exec(sqliteMigrate2To3UpdateQueries); err != nil {
-			return err
-		}
-		if _, err := tx.Exec(sqliteUpdateSchemaVersionQuery, 3); err != nil {
-			return err
-		}
-		return nil
-	})
-}
-
-func sqliteMigrateFrom3(sqlDB *sql.DB) error {
-	log.Tag(tag).Info("Migrating user database schema: from 3 to 4")
-	return db.ExecTx(sqlDB, func(tx *sql.Tx) error {
-		if _, err := tx.Exec(sqliteMigrate3To4UpdateQueries); err != nil {
-			return err
-		}
-		if _, err := tx.Exec(sqliteUpdateSchemaVersionQuery, 4); err != nil {
-			return err
-		}
-		return nil
-	})
-}
-
-func sqliteMigrateFrom4(sqlDB *sql.DB) error {
-	log.Tag(tag).Info("Migrating user database schema: from 4 to 5")
-	return db.ExecTx(sqlDB, func(tx *sql.Tx) error {
-		if _, err := tx.Exec(sqliteMigrate4To5UpdateQueries); err != nil {
-			return err
-		}
-		if _, err := tx.Exec(sqliteUpdateSchemaVersionQuery, 5); err != nil {
-			return err
-		}
-		return nil
-	})
-}
-
-func sqliteMigrateFrom5(sqlDB *sql.DB) error {
-	log.Tag(tag).Info("Migrating user database schema: from 5 to 6")
-	return db.ExecTx(sqlDB, func(tx *sql.Tx) error {
-		if _, err := tx.Exec(sqliteMigrate5To6UpdateQueries); err != nil {
-			return err
-		}
-		if _, err := tx.Exec(sqliteUpdateSchemaVersionQuery, 6); err != nil {
-			return err
-		}
-		return nil
-	})
-}
-
-func sqliteMigrateFrom6(sqlDB *sql.DB) error {
-	log.Tag(tag).Info("Migrating user database schema: from 6 to 7")
-	return db.ExecTx(sqlDB, func(tx *sql.Tx) error {
-		if _, err := tx.Exec(sqliteMigrate6To7UpdateQueries); err != nil {
-			return err
-		}
-		if _, err := tx.Exec(sqliteUpdateSchemaVersionQuery, 7); err != nil {
-			return err
-		}
-		return nil
-	})
-}
-
-func sqliteMigrateFrom7(sqlDB *sql.DB) error {
-	log.Tag(tag).Info("Migrating user database schema: from 7 to 8")
-	return db.ExecTx(sqlDB, func(tx *sql.Tx) error {
-		if _, err := tx.Exec(sqliteMigrate7To8UpdateQueries); err != nil {
-			return err
-		}
-		if _, err := tx.Exec(sqliteUpdateSchemaVersionQuery, 8); err != nil {
-			return err
-		}
-		return nil
-	})
 }

user/manager_test.go

@@ -14,6 +14,7 @@ import (
 	"golang.org/x/crypto/bcrypt"
 	"heckel.io/ntfy/v2/db"
 	"heckel.io/ntfy/v2/db/pg"
+	"heckel.io/ntfy/v2/db/schema"
 	dbtest "heckel.io/ntfy/v2/db/test"
 	"heckel.io/ntfy/v2/util"
 )
@@ -3352,7 +3353,7 @@ func TestManager_Emails_PrimaryFlagAndHelpers(t *testing.T) {
 func openReplicaTestSQLite(t *testing.T, filename string) *sql.DB {
 	d, err := sql.Open("sqlite3", filename+"?_case_sensitive_like=on")
 	require.Nil(t, err)
-	require.Nil(t, setupSQLite(d))
+	require.Nil(t, schema.Migrate(d, schema.SQLite, schemaStore, sqliteCurrentSchemaVersion, sqliteCreateTables, sqliteMigrations))
 	return d
 }