From 7fb1d257404420b9f51802fc1e5718656e7f69f4 Mon Sep 17 00:00:00 2001 From: binwiederhier Date: Mon, 27 Jul 2026 23:48:43 +0200 Subject: [PATCH] Restructure a little --- db/schema/schema.go | 36 ++++------------ db/schema/schema_test.go | 5 +-- db/schema/types.go | 25 ++++++++++++ webpush/store.go | 2 + webpush/store_postgres.go | 86 +++++++++++++-------------------------- webpush/store_sqlite.go | 81 +++++++++++++----------------------- webpush/store_test.go | 77 +++++++++++++++++++++++++++++++++++ 7 files changed, 169 insertions(+), 143 deletions(-) create mode 100644 db/schema/types.go diff --git a/db/schema/schema.go b/db/schema/schema.go index f4f97b7c..ce4587f7 100644 --- a/db/schema/schema.go +++ b/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 -// exact layout pre-existing ntfy SQLite databases already use, which lets them migrate onto -// this framework without an adoption step. +// its own database file, so the version lives in the schemaVersion table keyed by id = 1. 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. -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 - -// 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. +// Migrate creates or upgrades the named store's schema to targetVersion in one transaction, or +// creates a new database using the "create" function. 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) diff --git a/db/schema/schema_test.go b/db/schema/schema_test.go index 9d848e94..28e6c376 100644 --- a/db/schema/schema_test.go +++ b/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 { - _, err := tx.Exec(fmt.Sprintf(`ALTER TABLE things ADD COLUMN nick TEXT NOT NULL DEFAULT '%s'`, defaultName)) - return err - }, + 1: schema.AsMigrateFunc(fmt.Sprintf(`ALTER TABLE things ADD COLUMN nick TEXT NOT NULL DEFAULT '%s'`, defaultName)), } } forEachDialect(t, func(t *testing.T, d *sql.DB, dialect schema.Dialect) { diff --git a/db/schema/types.go b/db/schema/types.go new file mode 100644 index 00000000..430a02cd --- /dev/null +++ b/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 + } +} diff --git a/webpush/store.go b/webpush/store.go index 1a9825f5..a05fee6c 100644 --- a/webpush/store.go +++ b/webpush/store.go @@ -14,6 +14,8 @@ const ( subscriptionIDPrefix = "wps_" subscriptionIDLength = 10 subscriptionEndpointLimitPerSubscriberIP = 10 + + schemaStore = "webpush" ) // Errors returned by the store diff --git a/webpush/store_postgres.go b/webpush/store_postgres.go index 84168d89..dbb67faa 100644 --- a/webpush/store_postgres.go +++ b/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 - postgresInsertSchemaVersionQuery = `INSERT INTO schema_version (store, version) VALUES ('webpush', $1)` - postgresSelectSchemaVersionQuery = `SELECT version FROM schema_version WHERE store = 'webpush'` + postgresCurrentSchemaVersion = 1 +) + +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 - }) -} diff --git a/webpush/store_sqlite.go b/webpush/store_sqlite.go index e03a43b1..6a5f9683 100644 --- a/webpush/store_sqlite.go +++ b/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 - sqliteInsertSchemaVersionQuery = `INSERT INTO schemaVersion VALUES (1, ?)` - sqliteSelectSchemaVersionQuery = `SELECT version FROM schemaVersion WHERE id = 1` + sqliteCurrentSchemaVersion = 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 diff --git a/webpush/store_test.go b/webpush/store_test.go index 348f9998..2323de56 100644 --- a/webpush/store_test.go +++ b/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"), "")