Add Postgres support

.github/workflows/release.yaml

@@ -6,6 +6,22 @@ on:
 jobs:
   release:
     runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:17
+        env:
+          POSTGRES_USER: ntfy
+          POSTGRES_PASSWORD: ntfy
+          POSTGRES_DB: ntfy_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U ntfy"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    env:
+      NTFY_TEST_DATABASE_URL: "postgres://ntfy:ntfy@localhost:5432/ntfy_test?sslmode=disable"
     steps:
       - name: Checkout code
         uses: actions/checkout@v3

.github/workflows/test.yaml

@@ -3,6 +3,22 @@ on: [ push, pull_request ]
 jobs:
   test:
     runs-on: ubuntu-latest
+    services:
+      postgres:
+        image: postgres:17
+        env:
+          POSTGRES_USER: ntfy
+          POSTGRES_PASSWORD: ntfy
+          POSTGRES_DB: ntfy_test
+        ports:
+          - 5432:5432
+        options: >-
+          --health-cmd "pg_isready -U ntfy"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    env:
+      NTFY_TEST_DATABASE_URL: "postgres://ntfy:ntfy@localhost:5432/ntfy_test?sslmode=disable"
     steps:
       - name: Checkout code
         uses: actions/checkout@v3
@@ -23,7 +39,7 @@ jobs:
       - name: Build web app (required for tests)
         run: make web
       - name: Run tests, formatting, vetting and linting
-        run: make check
+        run: make checkv
       - name: Run coverage
         run: make coverage
       - name: Upload coverage to codecov.io

Makefile

@@ -265,11 +265,13 @@ cli-build-results:
 
 check: test web-fmt-check fmt-check vet web-lint lint staticcheck
 
+checkv: testv web-fmt-check fmt-check vet web-lint lint staticcheck
+
 test: .PHONY
-	go test $(shell go list ./... | grep -vE 'ntfy/(test|examples|tools)')
+	go test -parallel 3 $(shell go list ./... | grep -vE 'ntfy/(test|examples|tools)')
 
 testv: .PHONY
-	go test -v $(shell go list ./... | grep -vE 'ntfy/(test|examples|tools)')
+	go test -v -parallel 3 $(shell go list ./... | grep -vE 'ntfy/(test|examples|tools)')
 
 race: .PHONY
 	go test -v -race $(shell go list ./... | grep -vE 'ntfy/(test|examples|tools)')

cmd/serve.go

@@ -39,6 +39,7 @@ var flagsServe = append(
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "key-file", Aliases: []string{"key_file", "K"}, EnvVars: []string{"NTFY_KEY_FILE"}, Usage: "private key file, if listen-https is set"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "cert-file", Aliases: []string{"cert_file", "E"}, EnvVars: []string{"NTFY_CERT_FILE"}, Usage: "certificate file, if listen-https is set"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "firebase-key-file", Aliases: []string{"firebase_key_file", "F"}, EnvVars: []string{"NTFY_FIREBASE_KEY_FILE"}, Usage: "Firebase credentials file; if set additionally publish to FCM topic"}),
+	altsrc.NewStringFlag(&cli.StringFlag{Name: "database-url", Aliases: []string{"database_url"}, EnvVars: []string{"NTFY_DATABASE_URL"}, Usage: "PostgreSQL connection string for database-backed stores (e.g. postgres://user:pass@host:5432/ntfy)"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "cache-file", Aliases: []string{"cache_file", "C"}, EnvVars: []string{"NTFY_CACHE_FILE"}, Usage: "cache file used for message caching"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "cache-duration", Aliases: []string{"cache_duration", "b"}, EnvVars: []string{"NTFY_CACHE_DURATION"}, Value: util.FormatDuration(server.DefaultCacheDuration), Usage: "buffer messages for this time to allow `since` requests"}),
 	altsrc.NewIntFlag(&cli.IntFlag{Name: "cache-batch-size", Aliases: []string{"cache_batch_size"}, EnvVars: []string{"NTFY_BATCH_SIZE"}, Usage: "max size of messages to batch together when writing to message cache (if zero, writes are synchronous)"}),
@@ -143,6 +144,7 @@ func execServe(c *cli.Context) error {
 	keyFile := c.String("key-file")
 	certFile := c.String("cert-file")
 	firebaseKeyFile := c.String("firebase-key-file")
+	databaseURL := c.String("database-url")
 	webPushPrivateKey := c.String("web-push-private-key")
 	webPushPublicKey := c.String("web-push-public-key")
 	webPushFile := c.String("web-push-file")
@@ -280,12 +282,14 @@ func execServe(c *cli.Context) error {
 	}
 
 	// Check values
-	if firebaseKeyFile != "" && !util.FileExists(firebaseKeyFile) {
+	if databaseURL != "" && (authFile != "" || cacheFile != "" || webPushFile != "") {
+		return errors.New("if database-url is set, auth-file, cache-file, and web-push-file must not be set")
+	} else if firebaseKeyFile != "" && !util.FileExists(firebaseKeyFile) {
 		return errors.New("if set, FCM key file must exist")
 	} else if firebaseKeyFile != "" && !server.FirebaseAvailable {
 		return errors.New("cannot set firebase-key-file, support for Firebase is not available (nofirebase)")
-	} else if webPushPublicKey != "" && (webPushPrivateKey == "" || webPushFile == "" || webPushEmailAddress == "" || baseURL == "") {
-		return errors.New("if web push is enabled, web-push-private-key, web-push-public-key, web-push-file, web-push-email-address, and base-url should be set. run 'ntfy webpush keys' to generate keys")
+	} else if webPushPublicKey != "" && (webPushPrivateKey == "" || (webPushFile == "" && databaseURL == "") || webPushEmailAddress == "" || baseURL == "") {
+		return errors.New("if web push is enabled, web-push-private-key, web-push-public-key, web-push-file (or database-url), web-push-email-address, and base-url should be set. run 'ntfy webpush keys' to generate keys")
 	} else if keepaliveInterval < 5*time.Second {
 		return errors.New("keepalive interval cannot be lower than five seconds")
 	} else if managerInterval < 5*time.Second {
@@ -321,8 +325,8 @@ func execServe(c *cli.Context) error {
 		return errors.New("if upstream-base-url is set, base-url must also be set")
 	} else if upstreamBaseURL != "" && baseURL != "" && baseURL == upstreamBaseURL {
 		return errors.New("base-url and upstream-base-url cannot be identical, you'll likely want to set upstream-base-url to https://ntfy.sh, see https://ntfy.sh/docs/config/#ios-instant-notifications")
-	} else if authFile == "" && (enableSignup || enableLogin || requireLogin || enableReservations || stripeSecretKey != "") {
-		return errors.New("cannot set enable-signup, enable-login, require-login, enable-reserve-topics, or stripe-secret-key if auth-file is not set")
+	} else if authFile == "" && databaseURL == "" && (enableSignup || enableLogin || requireLogin || enableReservations || stripeSecretKey != "") {
+		return errors.New("cannot set enable-signup, enable-login, require-login, enable-reserve-topics, or stripe-secret-key if auth-file or database-url is not set")
 	} else if enableSignup && !enableLogin {
 		return errors.New("cannot set enable-signup without also setting enable-login")
 	} else if requireLogin && !enableLogin {
@@ -331,8 +335,8 @@ func execServe(c *cli.Context) error {
 		return errors.New("cannot set stripe-secret-key or stripe-webhook-key, support for payments is not available in this build (nopayments)")
 	} else if stripeSecretKey != "" && (stripeWebhookKey == "" || baseURL == "") {
 		return errors.New("if stripe-secret-key is set, stripe-webhook-key and base-url must also be set")
-	} else if twilioAccount != "" && (twilioAuthToken == "" || twilioPhoneNumber == "" || twilioVerifyService == "" || baseURL == "" || authFile == "") {
-		return errors.New("if twilio-account is set, twilio-auth-token, twilio-phone-number, twilio-verify-service, base-url, and auth-file must also be set")
+	} else if twilioAccount != "" && (twilioAuthToken == "" || twilioPhoneNumber == "" || twilioVerifyService == "" || baseURL == "" || (authFile == "" && databaseURL == "")) {
+		return errors.New("if twilio-account is set, twilio-auth-token, twilio-phone-number, twilio-verify-service, base-url, and auth-file (or database-url) must also be set")
 	} else if messageSizeLimit > server.DefaultMessageSizeLimit {
 		log.Warn("message-size-limit is greater than 4K, this is not recommended and largely untested, and may lead to issues with some clients")
 		if messageSizeLimit > 5*1024*1024 {
@@ -412,6 +416,15 @@ func execServe(c *cli.Context) error {
 		payments.Setup(stripeSecretKey)
 	}
 
+	// Parse Twilio template
+	var twilioCallFormatTemplate *template.Template
+	if twilioCallFormat != "" {
+		twilioCallFormatTemplate, err = template.New("").Parse(twilioCallFormat)
+		if err != nil {
+			return fmt.Errorf("failed to parse twilio-call-format template: %w", err)
+		}
+	}
+
 	// Add default forbidden topics
 	disallowedTopics = append(disallowedTopics, server.DefaultDisallowedTopics...)
 
@@ -459,13 +472,7 @@ func execServe(c *cli.Context) error {
 	conf.TwilioAuthToken = twilioAuthToken
 	conf.TwilioPhoneNumber = twilioPhoneNumber
 	conf.TwilioVerifyService = twilioVerifyService
-	if twilioCallFormat != "" {
-		tmpl, err := template.New("twiml").Parse(twilioCallFormat)
-		if err != nil {
-			return fmt.Errorf("failed to parse twilio-call-format template: %w", err)
-		}
-		conf.TwilioCallFormat = tmpl
-	}
+	conf.TwilioCallFormat = twilioCallFormatTemplate
 	conf.MessageSizeLimit = int(messageSizeLimit)
 	conf.MessageDelayMax = messageDelayLimit
 	conf.TotalTopicLimit = totalTopicLimit
@@ -494,6 +501,7 @@ func execServe(c *cli.Context) error {
 	conf.EnableMetrics = enableMetrics
 	conf.MetricsListenHTTP = metricsListenHTTP
 	conf.ProfileListenHTTP = profileListenHTTP
+	conf.DatabaseURL = databaseURL
 	conf.WebPushPrivateKey = webPushPrivateKey
 	conf.WebPushPublicKey = webPushPublicKey
 	conf.WebPushFile = webPushFile

cmd/user.go

@@ -6,13 +6,14 @@ import (
 	"crypto/subtle"
 	"errors"
 	"fmt"
-	"heckel.io/ntfy/v2/server"
-	"heckel.io/ntfy/v2/user"
 	"os"
 	"strings"
 
 	"github.com/urfave/cli/v2"
 	"github.com/urfave/cli/v2/altsrc"
+	"heckel.io/ntfy/v2/db"
+	"heckel.io/ntfy/v2/server"
+	"heckel.io/ntfy/v2/user"
 	"heckel.io/ntfy/v2/util"
 )
 
@@ -29,6 +30,7 @@ var flagsUser = append(
 	&cli.StringFlag{Name: "config", Aliases: []string{"c"}, EnvVars: []string{"NTFY_CONFIG_FILE"}, Value: server.DefaultConfigFile, DefaultText: server.DefaultConfigFile, Usage: "config file"},
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "auth-file", Aliases: []string{"auth_file", "H"}, EnvVars: []string{"NTFY_AUTH_FILE"}, Usage: "auth database file used for access control"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "auth-default-access", Aliases: []string{"auth_default_access", "p"}, EnvVars: []string{"NTFY_AUTH_DEFAULT_ACCESS"}, Value: "read-write", Usage: "default permissions if no matching entries in the auth database are found"}),
+	altsrc.NewStringFlag(&cli.StringFlag{Name: "database-url", Aliases: []string{"database_url"}, EnvVars: []string{"NTFY_DATABASE_URL"}, Usage: "PostgreSQL connection string for database-backed stores"}),
 )
 
 var cmdUser = &cli.Command{
@@ -365,24 +367,36 @@ func createUserManager(c *cli.Context) (*user.Manager, error) {
 	authFile := c.String("auth-file")
 	authStartupQueries := c.String("auth-startup-queries")
 	authDefaultAccess := c.String("auth-default-access")
-	if authFile == "" {
-		return nil, errors.New("option auth-file not set; auth is unconfigured for this server")
-	} else if !util.FileExists(authFile) {
-		return nil, errors.New("auth-file does not exist; please start the server at least once to create it")
-	}
+	databaseURL := c.String("database-url")
 	authDefault, err := user.ParsePermission(authDefaultAccess)
 	if err != nil {
 		return nil, errors.New("if set, auth-default-access must start set to 'read-write', 'read-only', 'write-only' or 'deny-all'")
 	}
 	authConfig := &user.Config{
-		Filename:            authFile,
-		StartupQueries:      authStartupQueries,
 		DefaultAccess:       authDefault,
 		ProvisionEnabled:    false, // Hack: Do not re-provision users on manager initialization
 		BcryptCost:          user.DefaultUserPasswordBcryptCost,
 		QueueWriterInterval: user.DefaultUserStatsQueueWriterInterval,
 	}
-	return user.NewManager(authConfig)
+	var store user.Store
+	if databaseURL != "" {
+		pool, dbErr := db.OpenPostgres(databaseURL)
+		if dbErr != nil {
+			return nil, dbErr
+		}
+		store, err = user.NewPostgresStore(pool)
+	} else if authFile != "" {
+		if !util.FileExists(authFile) {
+			return nil, errors.New("auth-file does not exist; please start the server at least once to create it")
+		}
+		store, err = user.NewSQLiteStore(authFile, authStartupQueries)
+	} else {
+		return nil, errors.New("option database-url or auth-file not set; auth is unconfigured for this server")
+	}
+	if err != nil {
+		return nil, err
+	}
+	return user.NewManager(store, authConfig)
 }
 
 func readPasswordAndConfirm(c *cli.Context) (string, error) {

db/db.go

@@ -0,0 +1,93 @@
+package db
+
+import (
+	"database/sql"
+	"fmt"
+	"net/url"
+	"strconv"
+	"time"
+
+	_ "github.com/jackc/pgx/v5/stdlib" // PostgreSQL driver
+)
+
+const (
+	paramMaxOpenConns    = "pool_max_conns"
+	paramMaxIdleConns    = "pool_max_idle_conns"
+	paramConnMaxLifetime = "pool_conn_max_lifetime"
+	paramConnMaxIdleTime = "pool_conn_max_idle_time"
+
+	defaultMaxOpenConns = 10
+)
+
+// OpenPostgres opens a PostgreSQL database connection pool from a DSN string. It supports custom
+// query parameters for pool configuration: pool_max_conns (default 10), pool_max_idle_conns,
+// pool_conn_max_lifetime, and pool_conn_max_idle_time. These parameters are stripped from
+// the DSN before passing it to the driver.
+func OpenPostgres(dsn string) (*sql.DB, error) {
+	u, err := url.Parse(dsn)
+	if err != nil {
+		return nil, fmt.Errorf("invalid database URL: %w", err)
+	}
+	q := u.Query()
+	maxOpenConns, err := extractIntParam(q, paramMaxOpenConns, defaultMaxOpenConns)
+	if err != nil {
+		return nil, err
+	}
+	maxIdleConns, err := extractIntParam(q, paramMaxIdleConns, 0)
+	if err != nil {
+		return nil, err
+	}
+	connMaxLifetime, err := extractDurationParam(q, paramConnMaxLifetime, 0)
+	if err != nil {
+		return nil, err
+	}
+	connMaxIdleTime, err := extractDurationParam(q, paramConnMaxIdleTime, 0)
+	if err != nil {
+		return nil, err
+	}
+	u.RawQuery = q.Encode()
+	db, err := sql.Open("pgx", u.String())
+	if err != nil {
+		return nil, err
+	}
+	db.SetMaxOpenConns(maxOpenConns)
+	if maxIdleConns > 0 {
+		db.SetMaxIdleConns(maxIdleConns)
+	}
+	if connMaxLifetime > 0 {
+		db.SetConnMaxLifetime(connMaxLifetime)
+	}
+	if connMaxIdleTime > 0 {
+		db.SetConnMaxIdleTime(connMaxIdleTime)
+	}
+	if err := db.Ping(); err != nil {
+		return nil, fmt.Errorf("ping failed: %w", err)
+	}
+	return db, nil
+}
+
+func extractIntParam(q url.Values, key string, defaultValue int) (int, error) {
+	s := q.Get(key)
+	if s == "" {
+		return defaultValue, nil
+	}
+	q.Del(key)
+	v, err := strconv.Atoi(s)
+	if err != nil {
+		return 0, fmt.Errorf("invalid %s value %q: %w", key, s, err)
+	}
+	return v, nil
+}
+
+func extractDurationParam(q url.Values, key string, defaultValue time.Duration) (time.Duration, error) {
+	s := q.Get(key)
+	if s == "" {
+		return defaultValue, nil
+	}
+	q.Del(key)
+	d, err := time.ParseDuration(s)
+	if err != nil {
+		return 0, fmt.Errorf("invalid %s value %q: %w", key, s, err)
+	}
+	return d, nil
+}

db/test/test.go

@@ -0,0 +1,63 @@
+package dbtest
+
+import (
+	"database/sql"
+	"fmt"
+	"net/url"
+	"os"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"heckel.io/ntfy/v2/db"
+	"heckel.io/ntfy/v2/util"
+)
+
+const testPoolMaxConns = "2"
+
+// CreateTestPostgresSchema creates a temporary PostgreSQL schema and returns the DSN pointing to it.
+// It registers a cleanup function to drop the schema when the test finishes.
+// If NTFY_TEST_DATABASE_URL is not set, the test is skipped.
+func CreateTestPostgresSchema(t *testing.T) string {
+	t.Helper()
+	dsn := os.Getenv("NTFY_TEST_DATABASE_URL")
+	if dsn == "" {
+		t.Skip("NTFY_TEST_DATABASE_URL not set")
+	}
+	schema := fmt.Sprintf("test_%s", util.RandomString(10))
+	u, err := url.Parse(dsn)
+	require.Nil(t, err)
+	q := u.Query()
+	q.Set("pool_max_conns", testPoolMaxConns)
+	u.RawQuery = q.Encode()
+	dsn = u.String()
+	setupDB, err := db.OpenPostgres(dsn)
+	require.Nil(t, err)
+	_, err = setupDB.Exec(fmt.Sprintf("CREATE SCHEMA %s", schema))
+	require.Nil(t, err)
+	require.Nil(t, setupDB.Close())
+	q.Set("search_path", schema)
+	u.RawQuery = q.Encode()
+	schemaDSN := u.String()
+	t.Cleanup(func() {
+		cleanDB, err := db.OpenPostgres(dsn)
+		if err == nil {
+			cleanDB.Exec(fmt.Sprintf("DROP SCHEMA %s CASCADE", schema))
+			cleanDB.Close()
+		}
+	})
+	return schemaDSN
+}
+
+// CreateTestPostgres creates a temporary PostgreSQL schema and returns an open *sql.DB connection to it.
+// It registers cleanup functions to close the DB and drop the schema when the test finishes.
+// If NTFY_TEST_DATABASE_URL is not set, the test is skipped.
+func CreateTestPostgres(t *testing.T) *sql.DB {
+	t.Helper()
+	schemaDSN := CreateTestPostgresSchema(t)
+	testDB, err := db.OpenPostgres(schemaDSN)
+	require.Nil(t, err)
+	t.Cleanup(func() {
+		testDB.Close()
+	})
+	return testDB
+}