Restructure a little

db/schema/schema.go

@@ -1,9 +1,7 @@
 // Package schema tracks and migrates database schemas, and Migrate creates or upgrades a
 // store's schema inside a single transaction. On PostgreSQL, all stores share one database, so
 // versions live in a shared schema_version table keyed by store name. On SQLite, every store is
-// its own database file, so the version lives in the schemaVersion table keyed by id = 1 -- the
+// its own database file, so the version lives in the schemaVersion table keyed by id = 1.
-// exact layout pre-existing ntfy SQLite databases already use, which lets them migrate onto
-// this framework without an adoption step.
 package schema
 
 import (
@@ -17,36 +15,18 @@ import (
 // Queries by dialect; the version table is shared by all stores on Postgres, and per database
 // file on SQLite
 const (
-	postgresCreateVersionTableQuery = `CREATE TABLE IF NOT EXISTS schema_version (store TEXT PRIMARY KEY, version INT NOT NULL)`
-	postgresSelectVersionQuery      = `SELECT version FROM schema_version WHERE store = $1`
-	postgresUpsertVersionQuery      = `INSERT INTO schema_version (store, version) VALUES ($1, $2) ON CONFLICT (store) DO UPDATE SET version = EXCLUDED.version`
-	postgresAdvisoryLockQuery       = `SELECT pg_advisory_xact_lock($1)`
-
 	sqliteCreateVersionTableQuery = `CREATE TABLE IF NOT EXISTS schemaVersion (id INT PRIMARY KEY, version INT NOT NULL)`
 	sqliteSelectVersionQuery      = `SELECT version FROM schemaVersion WHERE id = 1`
 	sqliteUpsertVersionQuery      = `INSERT INTO schemaVersion (id, version) VALUES (1, ?) ON CONFLICT (id) DO UPDATE SET version = excluded.version`
+
+	postgresCreateVersionTableQuery = `CREATE TABLE IF NOT EXISTS schema_version (store TEXT PRIMARY KEY, version INT NOT NULL)`
+	postgresSelectVersionQuery      = `SELECT version FROM schema_version WHERE store = $1`
+	postgresUpsertVersionQuery      = `INSERT INTO schema_version (store, version) VALUES ($1, $2) ON CONFLICT (store) DO UPDATE SET version = EXCLUDED.version`
+	postgresAdvisoryLockQuery       = `SELECT pg_advisory_xact_lock($1)` // Transaction-scoped lock to avoid migration races
 )
 
-// Dialect selects the SQL flavor Migrate speaks to the version table.
+// Migrate creates or upgrades the named store's schema to targetVersion in one transaction, or
-type Dialect int
+// creates a new database using the "create" function.
-
-// Supported dialects; SQLite is the zero value
-const (
-	SQLite Dialect = iota
-	Postgres
-)
-
-// MigrateFunc applies one schema change inside the setup transaction: the initial creation of
-// a store's tables, or one step upgrading a store from version N to N+1. Migrations needing
-// config capture it via closure, e.g. func migrations(cacheDuration time.Duration) map[int]MigrateFunc.
-type MigrateFunc func(tx *sql.Tx) error
-
-// Migrate creates or upgrades the named store's schema to targetVersion in one transaction: a
-// fresh database gets the create func; an existing one is upgraded step by step through the
-// migrations map (keyed by the FROM version; always append, never insert in the middle); a
-// schema migrated by newer code is refused. A failed migration rolls back atomically. Caveats:
-// statements that refuse transaction blocks (Postgres CREATE INDEX CONCURRENTLY, VACUUM) cannot
-// go through Migrate, and DDL holds exclusive locks until commit, so keep migrations fast.
 func Migrate(db *sql.DB, dialect Dialect, store string, targetVersion int, create MigrateFunc, migrations map[int]MigrateFunc) error {
 	if dialect != Postgres && dialect != SQLite {
 		return fmt.Errorf("unsupported schema dialect %d", dialect)

db/schema/schema_test.go

@@ -84,10 +84,7 @@ func TestMigrate_ClosureCarriesConfig(t *testing.T) {
 	// params plumbing in the framework itself
 	migrationsFor := func(defaultName string) map[int]schema.MigrateFunc {
 		return map[int]schema.MigrateFunc{
-			1: func(tx *sql.Tx) error {
+			1: schema.AsMigrateFunc(fmt.Sprintf(`ALTER TABLE things ADD COLUMN nick TEXT NOT NULL DEFAULT '%s'`, defaultName)),
-				_, err := tx.Exec(fmt.Sprintf(`ALTER TABLE things ADD COLUMN nick TEXT NOT NULL DEFAULT '%s'`, defaultName))
-				return err
-			},
 		}
 	}
 	forEachDialect(t, func(t *testing.T, d *sql.DB, dialect schema.Dialect) {

db/schema/types.go

@@ -0,0 +1,25 @@
+package schema
+
+import "database/sql"
+
+// Dialect selects the SQL flavor Migrate speaks to the version table.
+type Dialect int
+
+// Supported dialects; SQLite is the zero value
+const (
+	SQLite Dialect = iota
+	Postgres
+)
+
+// MigrateFunc applies one schema change inside the setup transaction: the initial creation of
+// a store's tables, or one step upgrading a store from version N to N+1. Migrations needing
+// config capture it via closure, e.g. func migrations(cacheDuration time.Duration) map[int]MigrateFunc.
+type MigrateFunc func(tx *sql.Tx) error
+
+// AsMigrateFunc converts a simple query to a migration function
+func AsMigrateFunc(query string) MigrateFunc {
+	return func(tx *sql.Tx) error {
+		_, err := tx.Exec(query)
+		return err
+	}
+}

webpush/store.go

@@ -14,6 +14,8 @@ const (
 	subscriptionIDPrefix                     = "wps_"
 	subscriptionIDLength                     = 10
 	subscriptionEndpointLimitPerSubscriberIP = 10
+
+	schemaStore = "webpush"
 )
 
 // Errors returned by the store

webpush/store_postgres.go

@@ -1,39 +1,11 @@
 package webpush
 
 import (
-	"database/sql"
-	"fmt"
-
 	"heckel.io/ntfy/v2/db"
+	"heckel.io/ntfy/v2/db/schema"
 )
 
 const (
-	postgresCreateTablesQuery = `
-		CREATE TABLE IF NOT EXISTS webpush_subscription (
-			id TEXT PRIMARY KEY,
-			endpoint TEXT NOT NULL UNIQUE,
-			key_auth TEXT NOT NULL,
-			key_p256dh TEXT NOT NULL,
-			user_id TEXT NOT NULL,
-			subscriber_ip TEXT NOT NULL,
-			updated_at BIGINT NOT NULL,
-			warned_at BIGINT NOT NULL DEFAULT 0
-		);
-		CREATE INDEX IF NOT EXISTS idx_webpush_subscriber_ip ON webpush_subscription (subscriber_ip);
-		CREATE INDEX IF NOT EXISTS idx_webpush_updated_at ON webpush_subscription (updated_at);
-		CREATE INDEX IF NOT EXISTS idx_webpush_user_id ON webpush_subscription (user_id);
-		CREATE TABLE IF NOT EXISTS webpush_subscription_topic (
-			subscription_id TEXT NOT NULL REFERENCES webpush_subscription (id) ON DELETE CASCADE,
-			topic TEXT NOT NULL,
-			PRIMARY KEY (subscription_id, topic)
-		);
-		CREATE INDEX IF NOT EXISTS idx_webpush_topic ON webpush_subscription_topic (topic);
-		CREATE TABLE IF NOT EXISTS schema_version (
-			store TEXT PRIMARY KEY,
-			version INT NOT NULL
-		);
-	`
-
 	postgresSelectSubscriptionIDByEndpointQuery        = `SELECT id FROM webpush_subscription WHERE endpoint = $1`
 	postgresSelectSubscriptionCountBySubscriberIPQuery = `SELECT COUNT(*) FROM webpush_subscription WHERE subscriber_ip = $1`
 	postgresSelectSubscriptionsForTopicQuery           = `
@@ -66,16 +38,38 @@ const (
 	postgresDeleteSubscriptionTopicWithoutSubscriptionQuery = `DELETE FROM webpush_subscription_topic WHERE subscription_id NOT IN (SELECT id FROM webpush_subscription)`
 )
 
-// PostgreSQL schema management queries
+// Schema version and queries
 const (
-	pgCurrentSchemaVersion           = 1
+	postgresCurrentSchemaVersion = 1
-	postgresInsertSchemaVersionQuery = `INSERT INTO schema_version (store, version) VALUES ('webpush', $1)`
+)
-	postgresSelectSchemaVersionQuery = `SELECT version FROM schema_version WHERE store = 'webpush'`
+
+var (
+	postgresCreateTables = schema.AsMigrateFunc(`
+		CREATE TABLE IF NOT EXISTS webpush_subscription (
+			id TEXT PRIMARY KEY,
+			endpoint TEXT NOT NULL UNIQUE,
+			key_auth TEXT NOT NULL,
+			key_p256dh TEXT NOT NULL,
+			user_id TEXT NOT NULL,
+			subscriber_ip TEXT NOT NULL,
+			updated_at BIGINT NOT NULL,
+			warned_at BIGINT NOT NULL DEFAULT 0
+		);
+		CREATE INDEX IF NOT EXISTS idx_webpush_subscriber_ip ON webpush_subscription (subscriber_ip);
+		CREATE INDEX IF NOT EXISTS idx_webpush_updated_at ON webpush_subscription (updated_at);
+		CREATE INDEX IF NOT EXISTS idx_webpush_user_id ON webpush_subscription (user_id);
+		CREATE TABLE IF NOT EXISTS webpush_subscription_topic (
+			subscription_id TEXT NOT NULL REFERENCES webpush_subscription (id) ON DELETE CASCADE,
+			topic TEXT NOT NULL,
+			PRIMARY KEY (subscription_id, topic)
+		);
+		CREATE INDEX IF NOT EXISTS idx_webpush_topic ON webpush_subscription_topic (topic);
+	`)
 )
 
 // NewPostgresStore creates a new PostgreSQL-backed web push store using an existing database connection pool.
 func NewPostgresStore(d *db.DB) (*Store, error) {
-	if err := setupPostgres(d.Primary()); err != nil {
+	if err := schema.Migrate(d.Primary(), schema.Postgres, schemaStore, postgresCurrentSchemaVersion, postgresCreateTables, nil); err != nil {
 		return nil, err
 	}
 	return &Store{
@@ -97,27 +91,3 @@ func NewPostgresStore(d *db.DB) (*Store, error) {
 		},
 	}, nil
 }
-
-func setupPostgres(d *sql.DB) error {
-	var schemaVersion int
-	err := d.QueryRow(postgresSelectSchemaVersionQuery).Scan(&schemaVersion)
-	if err != nil {
-		return setupNewPostgres(d)
-	}
-	if schemaVersion > pgCurrentSchemaVersion {
-		return fmt.Errorf("unexpected schema version: version %d is higher than current version %d", schemaVersion, pgCurrentSchemaVersion)
-	}
-	return nil
-}
-
-func setupNewPostgres(d *sql.DB) error {
-	return db.ExecTx(d, func(tx *sql.Tx) error {
-		if _, err := tx.Exec(postgresCreateTablesQuery); err != nil {
-			return err
-		}
-		if _, err := tx.Exec(postgresInsertSchemaVersionQuery, pgCurrentSchemaVersion); err != nil {
-			return err
-		}
-		return nil
-	})
-}

webpush/store_sqlite.go

@@ -2,38 +2,13 @@ package webpush
 
 import (
 	"database/sql"
-	"fmt"
 
 	_ "github.com/mattn/go-sqlite3" // SQLite driver
 	"heckel.io/ntfy/v2/db"
+	"heckel.io/ntfy/v2/db/schema"
 )
 
 const (
-	sqliteCreateTablesQuery = `
-		CREATE TABLE IF NOT EXISTS subscription (
-			id TEXT PRIMARY KEY,
-			endpoint TEXT NOT NULL,
-			key_auth TEXT NOT NULL,
-			key_p256dh TEXT NOT NULL,
-			user_id TEXT NOT NULL,		
-			subscriber_ip TEXT NOT NULL,
-			updated_at INT NOT NULL,
-			warned_at INT NOT NULL DEFAULT 0
-		);
-		CREATE UNIQUE INDEX IF NOT EXISTS idx_endpoint ON subscription (endpoint);
-		CREATE INDEX IF NOT EXISTS idx_subscriber_ip ON subscription (subscriber_ip);
-		CREATE TABLE IF NOT EXISTS subscription_topic (
-			subscription_id TEXT NOT NULL,
-			topic TEXT NOT NULL,
-			PRIMARY KEY (subscription_id, topic),
-			FOREIGN KEY (subscription_id) REFERENCES subscription (id) ON DELETE CASCADE
-		);
-		CREATE INDEX IF NOT EXISTS idx_topic ON subscription_topic (topic);
-		CREATE TABLE IF NOT EXISTS schemaVersion (
-			id INT PRIMARY KEY,
-			version INT NOT NULL
-		);
-	`
 	sqliteBuiltinStartupQueries = `
 		PRAGMA foreign_keys = ON;
 	`
@@ -70,11 +45,33 @@ const (
 	sqliteDeleteSubscriptionTopicWithoutSubscriptionQuery = `DELETE FROM subscription_topic WHERE subscription_id NOT IN (SELECT id FROM subscription)`
 )
 
-// SQLite schema management queries
+// Schema version and queries
 const (
-	sqliteCurrentSchemaVersion     = 1
+	sqliteCurrentSchemaVersion = 1
-	sqliteInsertSchemaVersionQuery = `INSERT INTO schemaVersion VALUES (1, ?)`
+)
-	sqliteSelectSchemaVersionQuery = `SELECT version FROM schemaVersion WHERE id = 1`
+
+var (
+	sqliteCreateTables = schema.AsMigrateFunc(`
+		CREATE TABLE IF NOT EXISTS subscription (
+			id TEXT PRIMARY KEY,
+			endpoint TEXT NOT NULL,
+			key_auth TEXT NOT NULL,
+			key_p256dh TEXT NOT NULL,
+			user_id TEXT NOT NULL,
+			subscriber_ip TEXT NOT NULL,
+			updated_at INT NOT NULL,
+			warned_at INT NOT NULL DEFAULT 0
+		);
+		CREATE UNIQUE INDEX IF NOT EXISTS idx_endpoint ON subscription (endpoint);
+		CREATE INDEX IF NOT EXISTS idx_subscriber_ip ON subscription (subscriber_ip);
+		CREATE TABLE IF NOT EXISTS subscription_topic (
+			subscription_id TEXT NOT NULL,
+			topic TEXT NOT NULL,
+			PRIMARY KEY (subscription_id, topic),
+			FOREIGN KEY (subscription_id) REFERENCES subscription (id) ON DELETE CASCADE
+		);
+		CREATE INDEX IF NOT EXISTS idx_topic ON subscription_topic (topic);
+	`)
 )
 
 // NewSQLiteStore creates a new SQLite-backed web push store.
@@ -83,7 +80,7 @@ func NewSQLiteStore(filename, startupQueries string) (*Store, error) {
 	if err != nil {
 		return nil, err
 	}
-	if err := setupSQLite(d); err != nil {
+	if err := schema.Migrate(d, schema.SQLite, schemaStore, sqliteCurrentSchemaVersion, sqliteCreateTables, nil); err != nil {
 		return nil, err
 	}
 	if err := runSQLiteStartupQueries(d, startupQueries); err != nil {
@@ -109,28 +106,6 @@ func NewSQLiteStore(filename, startupQueries string) (*Store, error) {
 	}, nil
 }
 
-func setupSQLite(db *sql.DB) error {
-	var schemaVersion int
-	if err := db.QueryRow(sqliteSelectSchemaVersionQuery).Scan(&schemaVersion); err != nil {
-		return setupNewSQLite(db)
-	} else if schemaVersion > sqliteCurrentSchemaVersion {
-		return fmt.Errorf("unexpected schema version: version %d is higher than current version %d", schemaVersion, sqliteCurrentSchemaVersion)
-	}
-	return nil
-}
-
-func setupNewSQLite(sqlDB *sql.DB) error {
-	return db.ExecTx(sqlDB, func(tx *sql.Tx) error {
-		if _, err := tx.Exec(sqliteCreateTablesQuery); 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(startupQueries); err != nil {
 		return err

webpush/store_test.go

@@ -1,6 +1,7 @@
 package webpush_test
 
 import (
+	"database/sql"
 	"fmt"
 	"net/netip"
 	"path/filepath"
@@ -14,6 +15,82 @@ import (
 
 const testWebPushEndpoint = "https://updates.push.services.mozilla.com/wpush/v1/AAABBCCCDDEEEFFF"
 
+// Schema layout as written by ntfy releases before the db/schema framework; used to verify
+// that existing databases open cleanly without an adoption step
+const (
+	testPreFrameworkSQLiteSchema = `
+		CREATE TABLE subscription (
+			id TEXT PRIMARY KEY,
+			endpoint TEXT NOT NULL,
+			key_auth TEXT NOT NULL,
+			key_p256dh TEXT NOT NULL,
+			user_id TEXT NOT NULL,
+			subscriber_ip TEXT NOT NULL,
+			updated_at INT NOT NULL,
+			warned_at INT NOT NULL DEFAULT 0
+		);
+		CREATE UNIQUE INDEX idx_endpoint ON subscription (endpoint);
+		CREATE TABLE subscription_topic (
+			subscription_id TEXT NOT NULL,
+			topic TEXT NOT NULL,
+			PRIMARY KEY (subscription_id, topic)
+		);
+		CREATE TABLE schemaVersion (id INT PRIMARY KEY, version INT NOT NULL);
+		INSERT INTO schemaVersion VALUES (1, 1);
+	`
+	testPreFrameworkPostgresSchema = `
+		CREATE TABLE webpush_subscription (
+			id TEXT PRIMARY KEY,
+			endpoint TEXT NOT NULL UNIQUE,
+			key_auth TEXT NOT NULL,
+			key_p256dh TEXT NOT NULL,
+			user_id TEXT NOT NULL,
+			subscriber_ip TEXT NOT NULL,
+			updated_at BIGINT NOT NULL,
+			warned_at BIGINT NOT NULL DEFAULT 0
+		);
+		CREATE TABLE webpush_subscription_topic (
+			subscription_id TEXT NOT NULL REFERENCES webpush_subscription (id) ON DELETE CASCADE,
+			topic TEXT NOT NULL,
+			PRIMARY KEY (subscription_id, topic)
+		);
+		CREATE TABLE schema_version (store TEXT PRIMARY KEY, version INT NOT NULL);
+		INSERT INTO schema_version (store, version) VALUES ('webpush', 1);
+	`
+)
+
+func TestStoreSQLiteOpensExistingDatabase(t *testing.T) {
+	filename := filepath.Join(t.TempDir(), "webpush.db")
+	d, err := sql.Open("sqlite3", filename)
+	require.Nil(t, err)
+	_, err = d.Exec(testPreFrameworkSQLiteSchema)
+	require.Nil(t, err)
+	require.Nil(t, d.Close())
+	store, err := webpush.NewSQLiteStore(filename, "")
+	require.Nil(t, err)
+	defer store.Close()
+	requireStoreUsable(t, store)
+}
+
+func TestStorePostgresOpensExistingDatabase(t *testing.T) {
+	testDB := dbtest.CreateTestPostgres(t)
+	_, err := testDB.Exec(testPreFrameworkPostgresSchema)
+	require.Nil(t, err)
+	store, err := webpush.NewPostgresStore(testDB)
+	require.Nil(t, err)
+	requireStoreUsable(t, store)
+}
+
+func requireStoreUsable(t *testing.T, store *webpush.Store) {
+	t.Helper()
+	err := store.UpsertSubscription(testWebPushEndpoint, "auth-key", "p256dh-key", "u_1234", netip.MustParseAddr("1.2.3.4"), []string{"mytopic"})
+	require.Nil(t, err)
+	subs, err := store.SubscriptionsForTopic("mytopic")
+	require.Nil(t, err)
+	require.Len(t, subs, 1)
+	require.Equal(t, testWebPushEndpoint, subs[0].Endpoint)
+}
+
 func forEachBackend(t *testing.T, f func(t *testing.T, store *webpush.Store)) {
 	t.Run("sqlite", func(t *testing.T) {
 		store, err := webpush.NewSQLiteStore(filepath.Join(t.TempDir(), "webpush.db"), "")