From 244a9bc06dc008c6b99ccfce54cb1e43a9b0ef19 Mon Sep 17 00:00:00 2001 From: binwiederhier Date: Wed, 29 Jul 2026 06:23:24 +0200 Subject: [PATCH] Use schema/ package in user/ package --- user/manager.go | 1 + user/manager_postgres.go | 3 +- user/manager_postgres_schema.go | 75 ++--------- user/manager_sqlite.go | 8 +- user/manager_sqlite_schema.go | 221 +++++++------------------------- user/manager_test.go | 3 +- 6 files changed, 66 insertions(+), 245 deletions(-) diff --git a/user/manager.go b/user/manager.go index da9190a6..4cfba6f5 100644 --- a/user/manager.go +++ b/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 diff --git a/user/manager_postgres.go b/user/manager_postgres.go index 0fa2604c..c949ba68 100644 --- a/user/manager_postgres.go +++ b/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) diff --git a/user/manager_postgres_schema.go b/user/manager_postgres_schema.go index a855dc66..1038fb38 100644 --- a/user/manager_postgres_schema.go +++ b/user/manager_postgres_schema.go @@ -1,8 +1,7 @@ package user import ( - "database/sql" - "fmt" + "heckel.io/ntfy/v2/db/schema" ) // 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)` + postgresCurrentSchemaVersion = 8 ) 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{ - 6: postgresMigrateFrom6, - 7: postgresMigrateFrom7, -} +var ( + postgresCreateTables = schema.AsMigrateFunc(postgresCreateTablesQueries) -func setupPostgres(db *sql.DB) error { - var schemaVersion int - err := db.QueryRow(postgresSelectSchemaVersionQuery).Scan(&schemaVersion) - if err != nil { - return setupNewPostgres(db) + // postgresMigrations maps a schema version to the migration upgrading it to the next + // version. Always append migrations at the end, never insert in the middle. + postgresMigrations = map[int]schema.MigrateFunc{ + 6: schema.AsMigrateFunc(postgresMigrate6To7UpdateQueries), + 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 -} +) diff --git a/user/manager_sqlite.go b/user/manager_sqlite.go index f016a4e2..e7e1c0be 100644 --- a/user/manager_sqlite.go +++ b/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 { diff --git a/user/manager_sqlite_schema.go b/user/manager_sqlite_schema.go index 5b389627..b64c6f31 100644 --- a/user/manager_sqlite_schema.go +++ b/user/manager_sqlite_schema.go @@ -2,10 +2,8 @@ package user import ( "database/sql" - "fmt" - "heckel.io/ntfy/v2/db" - "heckel.io/ntfy/v2/log" + "heckel.io/ntfy/v2/db/schema" "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` + sqliteCurrentSchemaVersion = 8 ) // 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, - 3: sqliteMigrateFrom3, - 4: sqliteMigrateFrom4, - 5: sqliteMigrateFrom5, - 6: sqliteMigrateFrom6, - 7: sqliteMigrateFrom7, + 2: schema.AsMigrateFunc(sqliteMigrate2To3UpdateQueries), + 3: schema.AsMigrateFunc(sqliteMigrate3To4UpdateQueries), + 4: schema.AsMigrateFunc(sqliteMigrate4To5UpdateQueries), + 5: schema.AsMigrateFunc(sqliteMigrate5To6UpdateQueries), + 6: schema.AsMigrateFunc(sqliteMigrate6To7UpdateQueries), + 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,122 +377,36 @@ func runSQLiteStartupQueries(db *sql.DB, startupQueries string) error { return nil } -func sqliteMigrateFrom1(sqlDB *sql.DB) 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 { +func sqliteMigrateFrom1(tx *sql.Tx) error { + // Rename user -> user_old, and create new tables + if _, err := tx.Exec(sqliteMigrate1To2CreateTablesQueries); err != nil { + return err + } + // Insert users from user_old into new user table, with ID and sync_topic + rows, err := tx.Query(sqliteMigrate1To2SelectAllOldUsernamesNoTxQuery) + if err != nil { + return err + } + defer rows.Close() + usernames := make([]string, 0) + for rows.Next() { + var username string + if err := rows.Scan(&username); err != nil { return err } - // Insert users from user_old into new user table, with ID and sync_topic - rows, err := tx.Query(sqliteMigrate1To2SelectAllOldUsernamesNoTxQuery) - if err != nil { + usernames = append(usernames, username) + } + if err := rows.Close(); err != nil { + return err + } + for _, username := range usernames { + userID := util.RandomStringPrefix(userIDPrefix, userIDLength) + syncTopic := util.RandomStringPrefix(syncTopicPrefix, syncTopicLength) + if _, err := tx.Exec(sqliteMigrate1To2InsertUserNoTxQuery, userID, syncTopic, username); err != nil { return err } - defer rows.Close() - usernames := make([]string, 0) - for rows.Next() { - var username string - if err := rows.Scan(&username); err != nil { - return err - } - usernames = append(usernames, username) - } - if err := rows.Close(); err != nil { - return err - } - for _, username := range usernames { - userID := util.RandomStringPrefix(userIDPrefix, userIDLength) - syncTopic := util.RandomStringPrefix(syncTopicPrefix, syncTopicLength) - if _, err := tx.Exec(sqliteMigrate1To2InsertUserNoTxQuery, userID, syncTopic, username); err != nil { - return err - } - } - // Migrate old "access" table to "user_access" and drop "access" and "user_old" - if _, err := tx.Exec(sqliteMigrate1To2InsertFromOldTablesAndDropNoTxQuery); err != nil { - 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 - }) + } + // Migrate old "access" table to "user_access" and drop "access" and "user_old" + _, err = tx.Exec(sqliteMigrate1To2InsertFromOldTablesAndDropNoTxQuery) + return err } diff --git a/user/manager_test.go b/user/manager_test.go index d2fa3b9e..a3104c79 100644 --- a/user/manager_test.go +++ b/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 }