diff --git a/cmd/serve.go b/cmd/serve.go index 313ec835..1ac4f86e 100644 --- a/cmd/serve.go +++ b/cmd/serve.go @@ -40,6 +40,7 @@ var flagsServe = append( 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.NewStringSliceFlag(&cli.StringSliceFlag{Name: "database-replica-urls", Aliases: []string{"database_replica_urls"}, EnvVars: []string{"NTFY_DATABASE_REPLICA_URLS"}, Usage: "PostgreSQL read replica connection strings for offloading read queries"}), 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)"}), @@ -145,6 +146,7 @@ func execServe(c *cli.Context) error { certFile := c.String("cert-file") firebaseKeyFile := c.String("firebase-key-file") databaseURL := c.String("database-url") + databaseReplicaURLs := c.StringSlice("database-replica-urls") webPushPrivateKey := c.String("web-push-private-key") webPushPublicKey := c.String("web-push-public-key") webPushFile := c.String("web-push-file") @@ -282,7 +284,9 @@ func execServe(c *cli.Context) error { } // Check values - if databaseURL != "" && (authFile != "" || cacheFile != "" || webPushFile != "") { + if len(databaseReplicaURLs) > 0 && databaseURL == "" { + return errors.New("database-replica-urls can only be used if database-url is also set") + } else 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") @@ -502,6 +506,7 @@ func execServe(c *cli.Context) error { conf.MetricsListenHTTP = metricsListenHTTP conf.ProfileListenHTTP = profileListenHTTP conf.DatabaseURL = databaseURL + conf.DatabaseReplicaURLs = databaseReplicaURLs conf.WebPushPrivateKey = webPushPrivateKey conf.WebPushPublicKey = webPushPublicKey conf.WebPushFile = webPushFile diff --git a/cmd/user.go b/cmd/user.go index 1ffc3e6b..cd6cf795 100644 --- a/cmd/user.go +++ b/cmd/user.go @@ -11,6 +11,7 @@ import ( "github.com/urfave/cli/v2" "github.com/urfave/cli/v2/altsrc" + "heckel.io/ntfy/v2/db" "heckel.io/ntfy/v2/db/pg" "heckel.io/ntfy/v2/server" "heckel.io/ntfy/v2/user" @@ -379,11 +380,11 @@ func createUserManager(c *cli.Context) (*user.Manager, error) { QueueWriterInterval: user.DefaultUserStatsQueueWriterInterval, } if databaseURL != "" { - pool, dbErr := pg.Open(databaseURL) + host, dbErr := pg.Open(databaseURL) if dbErr != nil { return nil, dbErr } - return user.NewPostgresManager(pool, authConfig) + return user.NewPostgresManager(db.New(host, nil), authConfig) } 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") diff --git a/db/db.go b/db/db.go index 00ae91f4..6586e763 100644 --- a/db/db.go +++ b/db/db.go @@ -1,38 +1,137 @@ package db import ( + "context" "database/sql" + "sync/atomic" + "time" + + "heckel.io/ntfy/v2/log" ) -// ExecTx executes a function within a database transaction. If the function returns an error, -// the transaction is rolled back. Otherwise, the transaction is committed. -func ExecTx(db *sql.DB, f func(tx *sql.Tx) error) error { - tx, err := db.Begin() - if err != nil { - return err - } - defer tx.Rollback() - if err := f(tx); err != nil { - return err - } - return tx.Commit() +const ( + tag = "db" + replicaHealthCheckInitialDelay = 5 * time.Second + replicaHealthCheckInterval = 30 * time.Second + replicaHealthCheckTimeout = 10 * time.Second +) + +// DB wraps a primary *sql.DB and optional read replicas. All standard query/exec methods +// delegate to the primary. The ReadOnly() method returns a *sql.DB from a healthy replica +// (round-robin), falling back to the primary if no replicas are configured or all are unhealthy. +type DB struct { + primary *Host + replicas []*Host + counter atomic.Uint64 + cancel context.CancelFunc } -// QueryTx executes a function within a database transaction and returns the result. If the function -// returns an error, the transaction is rolled back. Otherwise, the transaction is committed. -func QueryTx[T any](db *sql.DB, f func(tx *sql.Tx) (T, error)) (T, error) { - tx, err := db.Begin() - if err != nil { - var zero T - return zero, err +// New creates a new DB that wraps the given primary and optional replica connections. +// If replicas is nil or empty, ReadOnly() simply returns the primary. +// Replicas start unhealthy and are checked immediately by a background goroutine. +func New(primary *Host, replicas []*Host) *DB { + ctx, cancel := context.WithCancel(context.Background()) + d := &DB{ + primary: primary, + replicas: replicas, + cancel: cancel, } - defer tx.Rollback() - t, err := f(tx) - if err != nil { - return t, err + if len(d.replicas) > 0 { + go d.healthCheckLoop(ctx) + } + return d +} + +// Query delegates to the primary database. +func (d *DB) Query(query string, args ...any) (*sql.Rows, error) { + return d.primary.DB.Query(query, args...) +} + +// QueryRow delegates to the primary database. +func (d *DB) QueryRow(query string, args ...any) *sql.Row { + return d.primary.DB.QueryRow(query, args...) +} + +// Exec delegates to the primary database. +func (d *DB) Exec(query string, args ...any) (sql.Result, error) { + return d.primary.DB.Exec(query, args...) +} + +// Begin delegates to the primary database. +func (d *DB) Begin() (*sql.Tx, error) { + return d.primary.DB.Begin() +} + +// Ping delegates to the primary database. +func (d *DB) Ping() error { + return d.primary.DB.Ping() +} + +// Primary returns the underlying primary *sql.DB. This is only intended for +// one-time schema setup during store initialization, not for regular queries. +func (d *DB) Primary() *sql.DB { + return d.primary.DB +} + +// ReadOnly returns a *sql.DB suitable for read-only queries. It round-robins across healthy +// replicas. If all replicas are unhealthy or none are configured, the primary is returned. +func (d *DB) ReadOnly() *sql.DB { + if len(d.replicas) == 0 { + return d.primary.DB + } + n := len(d.replicas) + start := int(d.counter.Add(1) - 1) + for i := 0; i < n; i++ { + r := d.replicas[(start+i)%n] + if r.healthy.Load() { + return r.DB + } + } + return d.primary.DB +} + +// Close closes the primary database and all replicas, and stops the health-check goroutine. +func (d *DB) Close() error { + d.cancel() + for _, r := range d.replicas { + r.DB.Close() + } + return d.primary.DB.Close() +} + +// healthCheckLoop checks replicas immediately, then periodically on a ticker. +func (d *DB) healthCheckLoop(ctx context.Context) { + select { + case <-ctx.Done(): + return + case <-time.After(replicaHealthCheckInitialDelay): + d.checkReplicas(ctx) + } + for { + select { + case <-ctx.Done(): + return + case <-time.After(replicaHealthCheckInterval): + d.checkReplicas(ctx) + } + } +} + +// checkReplicas pings each replica with a timeout and updates its health status. +func (d *DB) checkReplicas(ctx context.Context) { + for _, r := range d.replicas { + wasHealthy := r.healthy.Load() + pingCtx, cancel := context.WithTimeout(ctx, replicaHealthCheckTimeout) + err := r.DB.PingContext(pingCtx) + cancel() + if err != nil { + r.healthy.Store(false) + log.Tag(tag).Error("Database replica %s is unhealthy: %s", r.Addr, err) + } else { + r.healthy.Store(true) + if !wasHealthy { + log.Tag(tag).Info("Database replica %s is healthy", r.Addr) + } + } } - if err := tx.Commit(); err != nil { - return t, err - } - return t, nil } diff --git a/db/pg/pg.go b/db/pg/pg.go index 228c167f..3b034736 100644 --- a/db/pg/pg.go +++ b/db/pg/pg.go @@ -9,22 +9,34 @@ import ( "time" _ "github.com/jackc/pgx/v5/stdlib" // PostgreSQL driver + + "heckel.io/ntfy/v2/db" ) -const ( - paramMaxOpenConns = "pool_max_conns" - paramMaxIdleConns = "pool_max_idle_conns" - paramConnMaxLifetime = "pool_conn_max_lifetime" - paramConnMaxIdleTime = "pool_conn_max_idle_time" +// Open opens a PostgreSQL connection pool for a primary database. It pings the database +// to verify connectivity before returning. +func Open(dsn string) (*db.Host, error) { + d, err := open(dsn) + if err != nil { + return nil, fmt.Errorf("failed to open database: %w", err) + } + if err := d.DB.Ping(); err != nil { + return nil, fmt.Errorf("database ping failed on %v: %w", d.Addr, err) + } + return d, nil +} - defaultMaxOpenConns = 10 -) +// OpenReplica opens a PostgreSQL connection pool for a read replica. Unlike Open, it does +// not ping the database, since replicas are health-checked in the background by db.DB. +func OpenReplica(dsn string) (*db.Host, error) { + return open(dsn) +} -// Open opens a PostgreSQL database connection pool from a DSN string. It supports custom +// open 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 Open(dsn string) (*sql.DB, error) { +func open(dsn string) (*db.Host, error) { u, err := url.Parse(dsn) if err != nil { return nil, fmt.Errorf("invalid database URL: %w", err) @@ -36,41 +48,41 @@ func Open(dsn string) (*sql.DB, error) { return nil, fmt.Errorf("invalid database URL scheme %q, must be \"postgres\" or \"postgresql\" (URL: %s)", u.Scheme, censorPassword(u)) } q := u.Query() - maxOpenConns, err := extractIntParam(q, paramMaxOpenConns, defaultMaxOpenConns) + maxOpenConns, err := extractIntParam(q, "pool_max_conns", 10) if err != nil { return nil, err } - maxIdleConns, err := extractIntParam(q, paramMaxIdleConns, 0) + maxIdleConns, err := extractIntParam(q, "pool_max_idle_conns", 0) if err != nil { return nil, err } - connMaxLifetime, err := extractDurationParam(q, paramConnMaxLifetime, 0) + connMaxLifetime, err := extractDurationParam(q, "pool_conn_max_lifetime", 0) if err != nil { return nil, err } - connMaxIdleTime, err := extractDurationParam(q, paramConnMaxIdleTime, 0) + connMaxIdleTime, err := extractDurationParam(q, "pool_conn_max_idle_time", 0) if err != nil { return nil, err } u.RawQuery = q.Encode() - db, err := sql.Open("pgx", u.String()) + d, err := sql.Open("pgx", u.String()) if err != nil { return nil, err } - db.SetMaxOpenConns(maxOpenConns) + d.SetMaxOpenConns(maxOpenConns) if maxIdleConns > 0 { - db.SetMaxIdleConns(maxIdleConns) + d.SetMaxIdleConns(maxIdleConns) } if connMaxLifetime > 0 { - db.SetConnMaxLifetime(connMaxLifetime) + d.SetConnMaxLifetime(connMaxLifetime) } if connMaxIdleTime > 0 { - db.SetConnMaxIdleTime(connMaxIdleTime) + d.SetConnMaxIdleTime(connMaxIdleTime) } - if err := db.Ping(); err != nil { - return nil, fmt.Errorf("database ping failed (URL: %s): %w", censorPassword(u), err) - } - return db, nil + return &db.Host{ + Addr: u.Host, + DB: d, + }, nil } func extractIntParam(q url.Values, key string, defaultValue int) (int, error) { diff --git a/db/test/test.go b/db/test/test.go index 36c3fc86..8d3f329b 100644 --- a/db/test/test.go +++ b/db/test/test.go @@ -1,13 +1,13 @@ package dbtest import ( - "database/sql" "fmt" "net/url" "os" "testing" "github.com/stretchr/testify/require" + "heckel.io/ntfy/v2/db" "heckel.io/ntfy/v2/db/pg" "heckel.io/ntfy/v2/util" ) @@ -30,34 +30,35 @@ func CreateTestPostgresSchema(t *testing.T) string { q.Set("pool_max_conns", testPoolMaxConns) u.RawQuery = q.Encode() dsn = u.String() - setupDB, err := pg.Open(dsn) + setupHost, err := pg.Open(dsn) require.Nil(t, err) - _, err = setupDB.Exec(fmt.Sprintf("CREATE SCHEMA %s", schema)) + _, err = setupHost.DB.Exec(fmt.Sprintf("CREATE SCHEMA %s", schema)) require.Nil(t, err) - require.Nil(t, setupDB.Close()) + require.Nil(t, setupHost.DB.Close()) q.Set("search_path", schema) u.RawQuery = q.Encode() schemaDSN := u.String() t.Cleanup(func() { - cleanDB, err := pg.Open(dsn) + cleanHost, err := pg.Open(dsn) if err == nil { - cleanDB.Exec(fmt.Sprintf("DROP SCHEMA %s CASCADE", schema)) - cleanDB.Close() + cleanHost.DB.Exec(fmt.Sprintf("DROP SCHEMA %s CASCADE", schema)) + cleanHost.DB.Close() } }) return schemaDSN } -// CreateTestPostgres creates a temporary PostgreSQL schema and returns an open *sql.DB connection to it. +// CreateTestPostgres creates a temporary PostgreSQL schema and returns an open *db.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 { +func CreateTestPostgres(t *testing.T) *db.DB { t.Helper() schemaDSN := CreateTestPostgresSchema(t) - testDB, err := pg.Open(schemaDSN) + testHost, err := pg.Open(schemaDSN) require.Nil(t, err) + d := db.New(testHost, nil) t.Cleanup(func() { - testDB.Close() + d.Close() }) - return testDB + return d } diff --git a/db/types.go b/db/types.go new file mode 100644 index 00000000..534d6168 --- /dev/null +++ b/db/types.go @@ -0,0 +1,19 @@ +package db + +import ( + "database/sql" + "sync/atomic" +) + +// Beginner is an interface for types that can begin a database transaction. +// Both *sql.DB and *DB implement this. +type Beginner interface { + Begin() (*sql.Tx, error) +} + +// Host pairs a *sql.DB with the host:port it was opened against. +type Host struct { + Addr string // "host:port" + DB *sql.DB + healthy atomic.Bool +} diff --git a/db/util.go b/db/util.go new file mode 100644 index 00000000..4621cb38 --- /dev/null +++ b/db/util.go @@ -0,0 +1,36 @@ +package db + +import "database/sql" + +// ExecTx executes a function within a database transaction. If the function returns an error, +// the transaction is rolled back. Otherwise, the transaction is committed. +func ExecTx(db Beginner, f func(tx *sql.Tx) error) error { + tx, err := db.Begin() + if err != nil { + return err + } + defer tx.Rollback() + if err := f(tx); err != nil { + return err + } + return tx.Commit() +} + +// QueryTx executes a function within a database transaction and returns the result. If the function +// returns an error, the transaction is rolled back. Otherwise, the transaction is committed. +func QueryTx[T any](db Beginner, f func(tx *sql.Tx) (T, error)) (T, error) { + tx, err := db.Begin() + if err != nil { + var zero T + return zero, err + } + defer tx.Rollback() + t, err := f(tx) + if err != nil { + return t, err + } + if err := tx.Commit(); err != nil { + return t, err + } + return t, nil +} diff --git a/docs/config.md b/docs/config.md index a202cd95..8a2f965e 100644 --- a/docs/config.md +++ b/docs/config.md @@ -149,11 +149,7 @@ no external dependencies: ### PostgreSQL (EXPERIMENTAL) As an alternative, you can configure ntfy to use PostgreSQL for **all** database-backed stores by setting the -`database-url` option to a PostgreSQL connection string: - -```yaml -database-url: "postgres://user:pass@host:5432/ntfy" -``` +`database-url` option to a PostgreSQL connection string. When `database-url` is set, ntfy will use PostgreSQL for the [message cache](#message-cache), [access control](#access-control), and [web push](#web-push) subscriptions instead of SQLite. The `cache-file`, @@ -165,11 +161,44 @@ topics. To restrict access, set `auth-default-access` to `deny-all` (see [access You can also set this via the environment variable `NTFY_DATABASE_URL` or the command line flag `--database-url`. +To offload read-heavy queries from the primary database, you can optionally configure one or more read replicas +using the `database-replica-urls` option. When configured, non-critical read-only queries (e.g. fetching messages, checking access permissions, etc) +are distributed across the replicas using round-robin, while all writes and correctness-critical reads continue to go +to the primary. If a replica becomes unhealthy, ntfy automatically falls back to the primary until the replica recovers. +You can also set this via the environment variable `NTFY_DATABASE_REPLICA_URLS` (comma-separated) or the command line +flag `--database-replica-urls`. + +Examples: + +=== "Simple" + ```yaml + database-url: "postgres://user:pass@host:5432/ntfy" + ``` + +=== "With SSL and pool tuning" + ```yaml + database-url: "postgres://user:pass@host:5432/ntfy?sslmode=require&pool_max_conns=50&pool_conn_max_idle_time=5m" + ``` + +=== "With CA certificate" + ```yaml + database-url: "postgres://user:pass@host:25060/ntfy?sslmode=require&sslrootcert=/etc/ntfy/db-ca-cert.pem&pool_max_conns=30" + ``` + +=== "With read replicas" + ```yaml + database-url: "postgres://user:pass@primary:5432/ntfy?sslmode=require&sslrootcert=/etc/ntfy/db-ca-cert.pem&pool_max_conns=30" + database-replica-urls: + - "postgres://user:pass@replica1:5432/ntfy?sslmode=require&sslrootcert=/etc/ntfy/db-ca-cert.pem&pool_max_conns=30" + - "postgres://user:pass@replica2:5432/ntfy?sslmode=require&sslrootcert=/etc/ntfy/db-ca-cert.pem&pool_max_conns=30" + ``` + The database URL supports the standard [PostgreSQL connection parameters](https://www.postgresql.org/docs/current/libpq-connect.html#LIBPQ-PARAMKEYWORDS) as query parameters, such as `sslmode`, `connect_timeout`, `sslcert`, `sslkey`, `sslrootcert`, and `application_name`. See the [pgx driver documentation](https://pkg.go.dev/github.com/jackc/pgx/v5) for the full list of supported parameters. -In addition, ntfy supports the following custom query parameters to tune the connection pool: +In addition, ntfy supports the following custom query parameters to tune the connection pool (these apply to both +the primary and replica URLs): | Parameter | Default | Description | |---------------------------|---------|----------------------------------------------------------------------------------| @@ -178,11 +207,6 @@ In addition, ntfy supports the following custom query parameters to tune the con | `pool_conn_max_lifetime` | - | Maximum amount of time a connection may be reused (Go duration, e.g. `5m`, `1h`) | | `pool_conn_max_idle_time` | - | Maximum amount of time a connection may be idle (Go duration, e.g. `30s`, `5m`) | -Example: - -```yaml -database-url: "postgres://user:pass@host:5432/ntfy?sslmode=require&pool_max_conns=50&pool_conn_max_idle_time=5m" -``` ## Message cache If desired, ntfy can temporarily keep notifications in an in-memory or an on-disk cache. Caching messages for a short period @@ -1819,6 +1843,7 @@ variable before running the `ntfy` command (e.g. `export NTFY_LISTEN_HTTP=:80`). | `cert-file` | `NTFY_CERT_FILE` | *filename* | - | HTTPS/TLS certificate file, only used if `listen-https` is set. | | `firebase-key-file` | `NTFY_FIREBASE_KEY_FILE` | *filename* | - | If set, also publish messages to a Firebase Cloud Messaging (FCM) topic for your app. This is optional and only required to save battery when using the Android app. See [Firebase (FCM)](#firebase-fcm). | | `database-url` | `NTFY_DATABASE_URL` | *string (connection URL)* | - | PostgreSQL connection string (e.g. `postgres://user:pass@host:5432/ntfy`). If set, uses PostgreSQL for all database-backed stores (message cache, user manager, web push) instead of SQLite. See [database options](#database-options). | +| `database-replica-urls` | `NTFY_DATABASE_REPLICA_URLS` | *list of strings (connection URLs)* | - | PostgreSQL read replica connection strings. Non-critical read-only queries are distributed across replicas (round-robin) with automatic fallback to primary. Requires `database-url`. See [read replicas](#read-replicas). | | `cache-file` | `NTFY_CACHE_FILE` | *filename* | - | If set, messages are cached in a local SQLite database instead of only in-memory. This allows for service restarts without losing messages in support of the since= parameter. See [message cache](#message-cache). | | `cache-duration` | `NTFY_CACHE_DURATION` | *duration* | 12h | Duration for which messages will be buffered before they are deleted. This is required to support the `since=...` and `poll=1` parameter. Set this to `0` to disable the cache entirely. | | `cache-startup-queries` | `NTFY_CACHE_STARTUP_QUERIES` | *string (SQL queries)* | - | SQL queries to run during database startup; this is useful for tuning and [enabling WAL mode](#message-cache) | diff --git a/docs/releases.md b/docs/releases.md index 15018d88..6ca2b86f 100644 --- a/docs/releases.md +++ b/docs/releases.md @@ -1757,6 +1757,10 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release ### ntfy server v2.19.x (UNRELEASED) +**Features:** + +* Support PostgreSQL read replicas for offloading non-critical read queries via `database-replica-urls` config option + **Bug fixes + maintenance:** * Web: Throttle notification sound in web app to play at most once every 2 seconds (similar to [#1550](https://github.com/binwiederhier/ntfy/issues/1550), thanks to [@jlaffaye](https://github.com/jlaffaye) for reporting) diff --git a/go.mod b/go.mod index 6dd9384f..c073d6aa 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,7 @@ go 1.25.0 require ( cloud.google.com/go/firestore v1.21.0 // indirect - cloud.google.com/go/storage v1.60.0 // indirect + cloud.google.com/go/storage v1.61.1 // indirect github.com/BurntSushi/toml v1.6.0 // indirect github.com/cpuguy83/go-md2man/v2 v2.0.7 // indirect github.com/emersion/go-smtp v0.18.0 @@ -14,12 +14,12 @@ require ( github.com/olebedev/when v1.1.0 github.com/stretchr/testify v1.11.1 github.com/urfave/cli/v2 v2.27.7 - golang.org/x/crypto v0.48.0 - golang.org/x/oauth2 v0.35.0 // indirect - golang.org/x/sync v0.19.0 - golang.org/x/term v0.40.0 - golang.org/x/time v0.14.0 - google.golang.org/api v0.269.0 + golang.org/x/crypto v0.49.0 + golang.org/x/oauth2 v0.36.0 // indirect + golang.org/x/sync v0.20.0 + golang.org/x/term v0.41.0 + golang.org/x/time v0.15.0 + google.golang.org/api v0.271.0 gopkg.in/yaml.v2 v2.4.0 ) @@ -34,14 +34,14 @@ require ( github.com/microcosm-cc/bluemonday v1.0.27 github.com/prometheus/client_golang v1.23.2 github.com/stripe/stripe-go/v74 v74.30.0 - golang.org/x/sys v0.41.0 - golang.org/x/text v0.34.0 + golang.org/x/sys v0.42.0 + golang.org/x/text v0.35.0 ) require ( cel.dev/expr v0.25.1 // indirect cloud.google.com/go v0.123.0 // indirect - cloud.google.com/go/auth v0.18.2 // indirect + cloud.google.com/go/auth v0.18.3-0.20260310051336-87cdcc9f7568 // indirect cloud.google.com/go/auth/oauth2adapt v0.2.8 // indirect cloud.google.com/go/compute/metadata v0.9.0 // indirect cloud.google.com/go/iam v1.5.3 // indirect @@ -70,7 +70,7 @@ require ( github.com/google/s2a-go v0.1.9 // indirect github.com/google/uuid v1.6.0 // indirect github.com/googleapis/enterprise-certificate-proxy v0.3.14 // indirect - github.com/googleapis/gax-go/v2 v2.17.0 // indirect + github.com/googleapis/gax-go/v2 v2.18.0 // indirect github.com/gorilla/css v1.0.1 // indirect github.com/jackc/pgpassfile v1.0.0 // indirect github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 // indirect @@ -86,20 +86,20 @@ require ( github.com/stretchr/objx v0.5.2 // indirect github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342 // indirect go.opentelemetry.io/auto/sdk v1.2.1 // indirect - go.opentelemetry.io/contrib/detectors/gcp v1.41.0 // indirect - go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.66.0 // indirect - go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.66.0 // indirect - go.opentelemetry.io/otel v1.41.0 // indirect - go.opentelemetry.io/otel/metric v1.41.0 // indirect - go.opentelemetry.io/otel/sdk v1.41.0 // indirect - go.opentelemetry.io/otel/sdk/metric v1.41.0 // indirect - go.opentelemetry.io/otel/trace v1.41.0 // indirect - go.yaml.in/yaml/v2 v2.4.3 // indirect - golang.org/x/net v0.51.0 // indirect + go.opentelemetry.io/contrib/detectors/gcp v1.42.0 // indirect + go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 // indirect + go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 // indirect + go.opentelemetry.io/otel v1.42.0 // indirect + go.opentelemetry.io/otel/metric v1.42.0 // indirect + go.opentelemetry.io/otel/sdk v1.42.0 // indirect + go.opentelemetry.io/otel/sdk/metric v1.42.0 // indirect + go.opentelemetry.io/otel/trace v1.42.0 // indirect + go.yaml.in/yaml/v2 v2.4.4 // indirect + golang.org/x/net v0.52.0 // indirect google.golang.org/appengine/v2 v2.0.6 // indirect - google.golang.org/genproto v0.0.0-20260226221140-a57be14db171 // indirect - google.golang.org/genproto/googleapis/api v0.0.0-20260226221140-a57be14db171 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 // indirect + google.golang.org/genproto v0.0.0-20260311181403-84a4fc48630c // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260311181403-84a4fc48630c // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260311181403-84a4fc48630c // indirect google.golang.org/grpc v1.79.2 // indirect google.golang.org/protobuf v1.36.11 // indirect gopkg.in/yaml.v3 v3.0.1 // indirect diff --git a/go.sum b/go.sum index c39fc3a1..1c6eada9 100644 --- a/go.sum +++ b/go.sum @@ -2,8 +2,8 @@ cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4= cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= cloud.google.com/go v0.123.0 h1:2NAUJwPR47q+E35uaJeYoNhuNEM9kM8SjgRgdeOJUSE= cloud.google.com/go v0.123.0/go.mod h1:xBoMV08QcqUGuPW65Qfm1o9Y4zKZBpGS+7bImXLTAZU= -cloud.google.com/go/auth v0.18.2 h1:+Nbt5Ev0xEqxlNjd6c+yYUeosQ5TtEUaNcN/3FozlaM= -cloud.google.com/go/auth v0.18.2/go.mod h1:xD+oY7gcahcu7G2SG2DsBerfFxgPAJz17zz2joOFF3M= +cloud.google.com/go/auth v0.18.3-0.20260310051336-87cdcc9f7568 h1:PJt3KrySfZkKdcEV2wlyNkfAPbMZGjtnv5oLrT4tWPg= +cloud.google.com/go/auth v0.18.3-0.20260310051336-87cdcc9f7568/go.mod h1:/Tt0rLCp4FHXEBtdyYqvIZPcJzbpJ/fmqtgIaXseDK4= cloud.google.com/go/auth/oauth2adapt v0.2.8 h1:keo8NaayQZ6wimpNSmW5OPc283g65QNIiLpZnkHRbnc= cloud.google.com/go/auth/oauth2adapt v0.2.8/go.mod h1:XQ9y31RkqZCcwJWNSx2Xvric3RrU88hAYYbjDWYDL+c= cloud.google.com/go/compute/metadata v0.9.0 h1:pDUj4QMoPejqq20dK0Pg2N4yG9zIkYGdBtwLoEkH9Zs= @@ -18,8 +18,8 @@ cloud.google.com/go/longrunning v0.8.0 h1:LiKK77J3bx5gDLi4SMViHixjD2ohlkwBi+mKA7 cloud.google.com/go/longrunning v0.8.0/go.mod h1:UmErU2Onzi+fKDg2gR7dusz11Pe26aknR4kHmJJqIfk= cloud.google.com/go/monitoring v1.24.3 h1:dde+gMNc0UhPZD1Azu6at2e79bfdztVDS5lvhOdsgaE= cloud.google.com/go/monitoring v1.24.3/go.mod h1:nYP6W0tm3N9H/bOw8am7t62YTzZY+zUeQ+Bi6+2eonI= -cloud.google.com/go/storage v1.60.0 h1:oBfZrSOCimggVNz9Y/bXY35uUcts7OViubeddTTVzQ8= -cloud.google.com/go/storage v1.60.0/go.mod h1:q+5196hXfejkctrnx+VYU8RKQr/L3c0cBIlrjmiAKE0= +cloud.google.com/go/storage v1.61.1 h1:VELCSvZKiSw0AS1k3so5mKGy3CB7bTCYD8EHhTF42bY= +cloud.google.com/go/storage v1.61.1/go.mod h1:k30/hwYfd0M8aULYbPkQLgNf+SFcdjlRHvLMXggw18E= cloud.google.com/go/trace v1.11.7 h1:kDNDX8JkaAG3R2nq1lIdkb7FCSi1rCmsEtKVsty7p+U= cloud.google.com/go/trace v1.11.7/go.mod h1:TNn9d5V3fQVf6s4SCveVMIBS2LJUqo73GACmq/Tky0s= firebase.google.com/go/v4 v4.19.0 h1:f5NMlC2YHFsncz00c2+ecBr+ZYlRMhKIhj1z8Iz0lD8= @@ -98,8 +98,8 @@ github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= github.com/googleapis/enterprise-certificate-proxy v0.3.14 h1:yh8ncqsbUY4shRD5dA6RlzjJaT4hi3kII+zYw8wmLb8= github.com/googleapis/enterprise-certificate-proxy v0.3.14/go.mod h1:vqVt9yG9480NtzREnTlmGSBmFrA+bzb0yl0TxoBQXOg= -github.com/googleapis/gax-go/v2 v2.17.0 h1:RksgfBpxqff0EZkDWYuz9q/uWsTVz+kf43LsZ1J6SMc= -github.com/googleapis/gax-go/v2 v2.17.0/go.mod h1:mzaqghpQp4JDh3HvADwrat+6M3MOIDp5YKHhb9PAgDY= +github.com/googleapis/gax-go/v2 v2.18.0 h1:jxP5Uuo3bxm3M6gGtV94P4lliVetoCB4Wk2x8QA86LI= +github.com/googleapis/gax-go/v2 v2.18.0/go.mod h1:uSzZN4a356eRG985CzJ3WfbFSpqkLTjsnhWGJR6EwrE= github.com/gorilla/css v1.0.1 h1:ntNaBIghp6JmvWnxbZKANoLyuXTPZ4cAMlo6RyhlbO8= github.com/gorilla/css v1.0.1/go.mod h1:BvnYkspnSzMmwRK+b8/xgNPLiIuNZr6vbZBTPQ2A3b0= github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg= @@ -165,36 +165,36 @@ github.com/xrash/smetrics v0.0.0-20250705151800-55b8f293f342/go.mod h1:Ohn+xnUBi github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY= go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= -go.opentelemetry.io/contrib/detectors/gcp v1.41.0 h1:MBzEwqhroF0JK0DpTVYWDxsenxm6L4PqOEfA90uZ5AA= -go.opentelemetry.io/contrib/detectors/gcp v1.41.0/go.mod h1:5pSDD0v0t2HqUmPC5cBBc+nLQO4dLYWnzBNheXLBLgs= -go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.66.0 h1:w/o339tDd6Qtu3+ytwt+/jon2yjAs3Ot8Xq8pelfhSo= -go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.66.0/go.mod h1:pdhNtM9C4H5fRdrnwO7NjxzQWhKSSxCHk/KluVqDVC0= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.66.0 h1:PnV4kVnw0zOmwwFkAzCN5O07fw1YOIQor120zrh0AVo= -go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.66.0/go.mod h1:ofAwF4uinaf8SXdVzzbL4OsxJ3VfeEg3f/F6CeF49/Y= -go.opentelemetry.io/otel v1.41.0 h1:YlEwVsGAlCvczDILpUXpIpPSL/VPugt7zHThEMLce1c= -go.opentelemetry.io/otel v1.41.0/go.mod h1:Yt4UwgEKeT05QbLwbyHXEwhnjxNO6D8L5PQP51/46dE= -go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.39.0 h1:5gn2urDL/FBnK8OkCfD1j3/ER79rUuTYmCvlXBKeYL8= -go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.39.0/go.mod h1:0fBG6ZJxhqByfFZDwSwpZGzJU671HkwpWaNe2t4VUPI= -go.opentelemetry.io/otel/metric v1.41.0 h1:rFnDcs4gRzBcsO9tS8LCpgR0dxg4aaxWlJxCno7JlTQ= -go.opentelemetry.io/otel/metric v1.41.0/go.mod h1:xPvCwd9pU0VN8tPZYzDZV/BMj9CM9vs00GuBjeKhJps= -go.opentelemetry.io/otel/sdk v1.41.0 h1:YPIEXKmiAwkGl3Gu1huk1aYWwtpRLeskpV+wPisxBp8= -go.opentelemetry.io/otel/sdk v1.41.0/go.mod h1:ahFdU0G5y8IxglBf0QBJXgSe7agzjE4GiTJ6HT9ud90= -go.opentelemetry.io/otel/sdk/metric v1.41.0 h1:siZQIYBAUd1rlIWQT2uCxWJxcCO7q3TriaMlf08rXw8= -go.opentelemetry.io/otel/sdk/metric v1.41.0/go.mod h1:HNBuSvT7ROaGtGI50ArdRLUnvRTRGniSUZbxiWxSO8Y= -go.opentelemetry.io/otel/trace v1.41.0 h1:Vbk2co6bhj8L59ZJ6/xFTskY+tGAbOnCtQGVVa9TIN0= -go.opentelemetry.io/otel/trace v1.41.0/go.mod h1:U1NU4ULCoxeDKc09yCWdWe+3QoyweJcISEVa1RBzOis= +go.opentelemetry.io/contrib/detectors/gcp v1.42.0 h1:kpt2PEJuOuqYkPcktfJqWWDjTEd/FNgrxcniL7kQrXQ= +go.opentelemetry.io/contrib/detectors/gcp v1.42.0/go.mod h1:W9zQ439utxymRrXsUOzZbFX4JhLxXU4+ZnCt8GG7yA8= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0 h1:yI1/OhfEPy7J9eoa6Sj051C7n5dvpj0QX8g4sRchg04= +go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc v0.67.0/go.mod h1:NoUCKYWK+3ecatC4HjkRktREheMeEtrXoQxrqYFeHSc= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0 h1:OyrsyzuttWTSur2qN/Lm0m2a8yqyIjUVBZcxFPuXq2o= +go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.67.0/go.mod h1:C2NGBr+kAB4bk3xtMXfZ94gqFDtg/GkI7e9zqGh5Beg= +go.opentelemetry.io/otel v1.42.0 h1:lSQGzTgVR3+sgJDAU/7/ZMjN9Z+vUip7leaqBKy4sho= +go.opentelemetry.io/otel v1.42.0/go.mod h1:lJNsdRMxCUIWuMlVJWzecSMuNjE7dOYyWlqOXWkdqCc= +go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.40.0 h1:ZrPRak/kS4xI3AVXy8F7pipuDXmDsrO8Lg+yQjBLjw0= +go.opentelemetry.io/otel/exporters/stdout/stdoutmetric v1.40.0/go.mod h1:3y6kQCWztq6hyW8Z9YxQDDm0Je9AJoFar2G0yDcmhRk= +go.opentelemetry.io/otel/metric v1.42.0 h1:2jXG+3oZLNXEPfNmnpxKDeZsFI5o4J+nz6xUlaFdF/4= +go.opentelemetry.io/otel/metric v1.42.0/go.mod h1:RlUN/7vTU7Ao/diDkEpQpnz3/92J9ko05BIwxYa2SSI= +go.opentelemetry.io/otel/sdk v1.42.0 h1:LyC8+jqk6UJwdrI/8VydAq/hvkFKNHZVIWuslJXYsDo= +go.opentelemetry.io/otel/sdk v1.42.0/go.mod h1:rGHCAxd9DAph0joO4W6OPwxjNTYWghRWmkHuGbayMts= +go.opentelemetry.io/otel/sdk/metric v1.42.0 h1:D/1QR46Clz6ajyZ3G8SgNlTJKBdGp84q9RKCAZ3YGuA= +go.opentelemetry.io/otel/sdk/metric v1.42.0/go.mod h1:Ua6AAlDKdZ7tdvaQKfSmnFTdHx37+J4ba8MwVCYM5hc= +go.opentelemetry.io/otel/trace v1.42.0 h1:OUCgIPt+mzOnaUTpOQcBiM/PLQ/Op7oq6g4LenLmOYY= +go.opentelemetry.io/otel/trace v1.42.0/go.mod h1:f3K9S+IFqnumBkKhRJMeaZeNk9epyhnCmQh/EysQCdc= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= -go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0= -go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8= +go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= +go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc= golang.org/x/crypto v0.13.0/go.mod h1:y6Z2r+Rw4iayiXXAIxJIDAJ1zMW4yaTpebo8fPOliYc= golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDfU= golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8= golang.org/x/crypto v0.31.0/go.mod h1:kDsLvtWBEx7MV9tJOj9bnXsPbxwJQ6csT/x4KIN4Ssk= -golang.org/x/crypto v0.48.0 h1:/VRzVqiRSggnhY7gNRxPauEQ5Drw9haKdM0jqfcCFts= -golang.org/x/crypto v0.48.0/go.mod h1:r0kV5h3qnFPlQnBSrULhlsRfryS2pmewsg+XfMgkVos= +golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4= +golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA= golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4= golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= golang.org/x/mod v0.12.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs= @@ -209,10 +209,10 @@ golang.org/x/net v0.10.0/go.mod h1:0qNGK6F8kojg2nk9dLZ2mShWaEBan6FAoqfSigmmuDg= golang.org/x/net v0.15.0/go.mod h1:idbUs1IY1+zTqbi8yxTbhexhEEk5ur9LInksu6HrEpk= golang.org/x/net v0.21.0/go.mod h1:bIjVDfnllIU7BJ2DNgfnXvpSvtn8VRwhlsaeUTyUS44= golang.org/x/net v0.25.0/go.mod h1:JkAGAh7GEvH74S6FOH42FLoXpXbE/aqXSrIQjXgsiwM= -golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= -golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= -golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ= -golang.org/x/oauth2 v0.35.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= +golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0= +golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw= +golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs= +golang.org/x/oauth2 v0.36.0/go.mod h1:YDBUJMTkDnJS+A4BP4eZBjCqtokkg1hODuPjwiGPO7Q= golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= @@ -220,8 +220,8 @@ golang.org/x/sync v0.3.0/go.mod h1:FU7BRWz2tNW+3quACPkgCx/L+uEAv1htQ0V83Z9Rj+Y= golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sync v0.10.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= -golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= -golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= @@ -234,8 +234,8 @@ golang.org/x/sys v0.12.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.17.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.20.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.28.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= -golang.org/x/sys v0.41.0 h1:Ivj+2Cp/ylzLiEU89QhWblYnOE9zerudt9Ftecq2C6k= -golang.org/x/sys v0.41.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= +golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= golang.org/x/telemetry v0.0.0-20240228155512-f48c80bd79b2/go.mod h1:TeRTkGYfJXctD9OcfyVLyj2J3IxLnKwHJR8f4D8a3YE= golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8= @@ -245,8 +245,8 @@ golang.org/x/term v0.12.0/go.mod h1:owVbMEjm3cBLCHdkQu9b1opXd4ETQWc3BhuQGKgXgvU= golang.org/x/term v0.17.0/go.mod h1:lLRBjIVuehSbZlaOtGMbcMncT+aqLLLmKrsjNrUguwk= golang.org/x/term v0.20.0/go.mod h1:8UkIAJTvZgivsXaD6/pH6U9ecQzZ45awqEOzuCvwpFY= golang.org/x/term v0.27.0/go.mod h1:iMsnZpn0cago0GOrHO2+Y7u7JPn5AylBrcoWkElMTSM= -golang.org/x/term v0.40.0 h1:36e4zGLqU4yhjlmxEaagx2KuYbJq3EwY8K943ZsHcvg= -golang.org/x/term v0.40.0/go.mod h1:w2P8uVp06p2iyKKuvXIm7N/y0UCRt3UfJTfZ7oOpglM= +golang.org/x/term v0.41.0 h1:QCgPso/Q3RTJx2Th4bDLqML4W6iJiaXFq2/ftQF13YU= +golang.org/x/term v0.41.0/go.mod h1:3pfBgksrReYfZ5lvYM0kSO0LIkAl4Yl2bXOkKP7Ec2A= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= @@ -258,10 +258,10 @@ golang.org/x/text v0.13.0/go.mod h1:TvPlkZtksWOMsz7fbANvkp4WM8x/WCo/om8BMLbz+aE= golang.org/x/text v0.14.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/text v0.15.0/go.mod h1:18ZOQIKpY8NJVqYksKHtTdi31H5itFRjB5/qKTNYzSU= golang.org/x/text v0.21.0/go.mod h1:4IBbMaMmOPCJ8SecivzSH54+73PCFmPWxNTLm+vZkEQ= -golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk= -golang.org/x/text v0.34.0/go.mod h1:homfLqTYRFyVYemLBFl5GgL/DWEiH5wcsQ5gSh1yziA= -golang.org/x/time v0.14.0 h1:MRx4UaLrDotUKUdCIqzPC48t1Y9hANFKIRpNx+Te8PI= -golang.org/x/time v0.14.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= +golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= +golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA= +golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U= +golang.org/x/time v0.15.0/go.mod h1:Y4YMaQmXwGQZoFaVFk4YpCt4FLQMYKZe9oeV/f4MSno= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc= @@ -272,16 +272,16 @@ golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8T golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= -google.golang.org/api v0.269.0 h1:qDrTOxKUQ/P0MveH6a7vZ+DNHxJQjtGm/uvdbdGXCQg= -google.golang.org/api v0.269.0/go.mod h1:N8Wpcu23Tlccl0zSHEkcAZQKDLdquxK+l9r2LkwAauE= +google.golang.org/api v0.271.0 h1:cIPN4qcUc61jlh7oXu6pwOQqbJW2GqYh5PS6rB2C/JY= +google.golang.org/api v0.271.0/go.mod h1:CGT29bhwkbF+i11qkRUJb2KMKqcJ1hdFceEIRd9u64Q= google.golang.org/appengine/v2 v2.0.6 h1:LvPZLGuchSBslPBp+LAhihBeGSiRh1myRoYK4NtuBIw= google.golang.org/appengine/v2 v2.0.6/go.mod h1:WoEXGoXNfa0mLvaH5sV3ZSGXwVmy8yf7Z1JKf3J3wLI= -google.golang.org/genproto v0.0.0-20260226221140-a57be14db171 h1:RxhCsti413yL0IjU9dVvuTbCISo8gs3RW1jPMStck+4= -google.golang.org/genproto v0.0.0-20260226221140-a57be14db171/go.mod h1:uhvzakVEqAuXU3TC2JCsxIRe5f77l+JySE3EqPoMyqM= -google.golang.org/genproto/googleapis/api v0.0.0-20260226221140-a57be14db171 h1:tu/dtnW1o3wfaxCOjSLn5IRX4YDcJrtlpzYkhHhGaC4= -google.golang.org/genproto/googleapis/api v0.0.0-20260226221140-a57be14db171/go.mod h1:M5krXqk4GhBKvB596udGL3UyjL4I1+cTbK0orROM9ng= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171 h1:ggcbiqK8WWh6l1dnltU4BgWGIGo+EVYxCaAPih/zQXQ= -google.golang.org/genproto/googleapis/rpc v0.0.0-20260226221140-a57be14db171/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/genproto v0.0.0-20260311181403-84a4fc48630c h1:ZhFDeBMmFc/4g8/GwxnJ4rzB3O4GwQVNr+8Mh7Y5z4g= +google.golang.org/genproto v0.0.0-20260311181403-84a4fc48630c/go.mod h1:hf4r/rBuzaTkLUWRO03771Xvcs6P5hwdQK3UUEJjqo0= +google.golang.org/genproto/googleapis/api v0.0.0-20260311181403-84a4fc48630c h1:OyQPd6I3pN/9gDxz6L13kYGJgqkpdrAohJRBeXyxlgI= +google.golang.org/genproto/googleapis/api v0.0.0-20260311181403-84a4fc48630c/go.mod h1:X2gu9Qwng7Nn009s/r3RUxqkzQNqOrAy79bluY7ojIg= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260311181403-84a4fc48630c h1:xgCzyF2LFIO/0X2UAoVRiXKU5Xg6VjToG4i2/ecSswk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260311181403-84a4fc48630c/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= google.golang.org/grpc v1.79.2 h1:fRMD94s2tITpyJGtBBn7MkMseNpOZU8ZxgC3MMBaXRU= google.golang.org/grpc v1.79.2/go.mod h1:KmT0Kjez+0dde/v2j9vzwoAScgEPx/Bw1CYChhHLrHQ= google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw= diff --git a/message/cache.go b/message/cache.go index 3b12af3e..b123fba4 100644 --- a/message/cache.go +++ b/message/cache.go @@ -50,14 +50,14 @@ type queries struct { // Cache stores published messages type Cache struct { - db *sql.DB + db *db.DB queue *util.BatchingQueue[*model.Message] nop bool mu *sync.Mutex // nil for PostgreSQL (concurrent writes supported), set for SQLite (single writer) queries queries } -func newCache(db *sql.DB, queries queries, mu *sync.Mutex, batchSize int, batchTimeout time.Duration, nop bool) *Cache { +func newCache(db *db.DB, queries queries, mu *sync.Mutex, batchSize int, batchTimeout time.Duration, nop bool) *Cache { var queue *util.BatchingQueue[*model.Message] if batchSize > 0 || batchTimeout > 0 { queue = util.NewBatchingQueue[*model.Message](batchSize, batchTimeout) @@ -201,10 +201,11 @@ func (c *Cache) Messages(topic string, since model.SinceMarker, scheduled bool) func (c *Cache) messagesSinceTime(topic string, since model.SinceMarker, scheduled bool) ([]*model.Message, error) { var rows *sql.Rows var err error + rdb := c.db.ReadOnly() if scheduled { - rows, err = c.db.Query(c.queries.selectMessagesSinceTimeScheduled, topic, since.Time().Unix()) + rows, err = rdb.Query(c.queries.selectMessagesSinceTimeScheduled, topic, since.Time().Unix()) } else { - rows, err = c.db.Query(c.queries.selectMessagesSinceTime, topic, since.Time().Unix()) + rows, err = rdb.Query(c.queries.selectMessagesSinceTime, topic, since.Time().Unix()) } if err != nil { return nil, err @@ -215,10 +216,11 @@ func (c *Cache) messagesSinceTime(topic string, since model.SinceMarker, schedul func (c *Cache) messagesSinceID(topic string, since model.SinceMarker, scheduled bool) ([]*model.Message, error) { var rows *sql.Rows var err error + rdb := c.db.ReadOnly() if scheduled { - rows, err = c.db.Query(c.queries.selectMessagesSinceIDScheduled, topic, since.ID()) + rows, err = rdb.Query(c.queries.selectMessagesSinceIDScheduled, topic, since.ID()) } else { - rows, err = c.db.Query(c.queries.selectMessagesSinceID, topic, since.ID()) + rows, err = rdb.Query(c.queries.selectMessagesSinceID, topic, since.ID()) } if err != nil { return nil, err @@ -227,7 +229,7 @@ func (c *Cache) messagesSinceID(topic string, since model.SinceMarker, scheduled } func (c *Cache) messagesLatest(topic string) ([]*model.Message, error) { - rows, err := c.db.Query(c.queries.selectMessagesLatest, topic) + rows, err := c.db.ReadOnly().Query(c.queries.selectMessagesLatest, topic) if err != nil { return nil, err } @@ -266,7 +268,7 @@ func (c *Cache) MessagesExpired() ([]string, error) { // Message returns the message with the given ID, or ErrMessageNotFound if not found func (c *Cache) Message(id string) (*model.Message, error) { - rows, err := c.db.Query(c.queries.selectMessagesByID, id) + rows, err := c.db.ReadOnly().Query(c.queries.selectMessagesByID, id) if err != nil { return nil, err } @@ -295,7 +297,7 @@ func (c *Cache) MarkPublished(m *model.Message) error { // MessagesCount returns the total number of messages in the cache func (c *Cache) MessagesCount() (int, error) { - rows, err := c.db.Query(c.queries.selectMessagesCount) + rows, err := c.db.ReadOnly().Query(c.queries.selectMessagesCount) if err != nil { return 0, err } @@ -312,7 +314,7 @@ func (c *Cache) MessagesCount() (int, error) { // Topics returns a list of all topics with messages in the cache func (c *Cache) Topics() ([]string, error) { - rows, err := c.db.Query(c.queries.selectTopics) + rows, err := c.db.ReadOnly().Query(c.queries.selectTopics) if err != nil { return nil, err } @@ -426,7 +428,7 @@ func (c *Cache) MarkAttachmentsDeleted(ids ...string) error { // AttachmentBytesUsedBySender returns the total size of active attachments sent by the given sender func (c *Cache) AttachmentBytesUsedBySender(sender string) (int64, error) { - rows, err := c.db.Query(c.queries.selectAttachmentsSizeBySender, sender, time.Now().Unix()) + rows, err := c.db.ReadOnly().Query(c.queries.selectAttachmentsSizeBySender, sender, time.Now().Unix()) if err != nil { return 0, err } @@ -435,7 +437,7 @@ func (c *Cache) AttachmentBytesUsedBySender(sender string) (int64, error) { // AttachmentBytesUsedByUser returns the total size of active attachments for the given user func (c *Cache) AttachmentBytesUsedByUser(userID string) (int64, error) { - rows, err := c.db.Query(c.queries.selectAttachmentsSizeByUserID, userID, time.Now().Unix()) + rows, err := c.db.ReadOnly().Query(c.queries.selectAttachmentsSizeByUserID, userID, time.Now().Unix()) if err != nil { return 0, err } @@ -466,7 +468,7 @@ func (c *Cache) UpdateStats(messages int64) error { // Stats returns the total message count statistic func (c *Cache) Stats() (messages int64, err error) { - rows, err := c.db.Query(c.queries.selectStats) + rows, err := c.db.ReadOnly().Query(c.queries.selectStats) if err != nil { return 0, err } diff --git a/message/cache_postgres.go b/message/cache_postgres.go index 0146f409..ba162da2 100644 --- a/message/cache_postgres.go +++ b/message/cache_postgres.go @@ -1,8 +1,9 @@ package message import ( - "database/sql" "time" + + "heckel.io/ntfy/v2/db" ) // PostgreSQL runtime query constants @@ -102,9 +103,9 @@ var postgresQueries = queries{ } // NewPostgresStore creates a new PostgreSQL-backed message cache store using an existing database connection pool. -func NewPostgresStore(db *sql.DB, batchSize int, batchTimeout time.Duration) (*Cache, error) { - if err := setupPostgres(db); err != nil { +func NewPostgresStore(d *db.DB, batchSize int, batchTimeout time.Duration) (*Cache, error) { + if err := setupPostgres(d.Primary()); err != nil { return nil, err } - return newCache(db, postgresQueries, nil, batchSize, batchTimeout, false), nil + return newCache(d, postgresQueries, nil, batchSize, batchTimeout, false), nil } diff --git a/message/cache_sqlite.go b/message/cache_sqlite.go index f9d8605e..a36aba0e 100644 --- a/message/cache_sqlite.go +++ b/message/cache_sqlite.go @@ -8,6 +8,7 @@ import ( "time" _ "github.com/mattn/go-sqlite3" // SQLite driver + "heckel.io/ntfy/v2/db" "heckel.io/ntfy/v2/util" ) @@ -110,14 +111,14 @@ func NewSQLiteStore(filename, startupQueries string, cacheDuration time.Duration if !util.FileExists(parentDir) { return nil, fmt.Errorf("cache database directory %s does not exist or is not accessible", parentDir) } - db, err := sql.Open("sqlite3", filename) + d, err := sql.Open("sqlite3", filename) if err != nil { return nil, err } - if err := setupSQLite(db, startupQueries, cacheDuration); err != nil { + if err := setupSQLite(d, startupQueries, cacheDuration); err != nil { return nil, err } - return newCache(db, sqliteQueries, &sync.Mutex{}, batchSize, batchTimeout, nop), nil + return newCache(db.New(&db.Host{DB: d}, nil), sqliteQueries, &sync.Mutex{}, batchSize, batchTimeout, nop), nil } // NewMemStore creates an in-memory cache diff --git a/server/config.go b/server/config.go index 786f0d78..8ead312c 100644 --- a/server/config.go +++ b/server/config.go @@ -95,7 +95,8 @@ type Config struct { ListenUnixMode fs.FileMode KeyFile string CertFile string - DatabaseURL string // PostgreSQL connection string (e.g. "postgres://user:pass@host:5432/ntfy") + DatabaseURL string // PostgreSQL connection string (e.g. "postgres://user:pass@host:5432/ntfy") + DatabaseReplicaURLs []string // PostgreSQL read replica connection strings FirebaseKeyFile string CacheFile string CacheDuration time.Duration diff --git a/server/server.go b/server/server.go index 329b0ab5..24c712bd 100644 --- a/server/server.go +++ b/server/server.go @@ -4,7 +4,6 @@ import ( "bytes" "context" "crypto/sha256" - "database/sql" "embed" "encoding/base64" "encoding/json" @@ -33,6 +32,7 @@ import ( "github.com/prometheus/client_golang/prometheus/promhttp" "golang.org/x/sync/errgroup" "gopkg.in/yaml.v2" + "heckel.io/ntfy/v2/db" "heckel.io/ntfy/v2/db/pg" "heckel.io/ntfy/v2/log" "heckel.io/ntfy/v2/message" @@ -47,7 +47,7 @@ import ( // Server is the main server, providing the UI and API for ntfy type Server struct { config *Config - db *sql.DB // Shared PostgreSQL connection pool, nil when using SQLite + db *db.DB // Shared PostgreSQL connection pool (with optional replicas), nil when using SQLite httpServer *http.Server httpsServer *http.Server httpMetricsServer *http.Server @@ -179,13 +179,26 @@ func New(conf *Config) (*Server, error) { stripe = newStripeAPI() } // Open shared PostgreSQL connection pool if configured - var pool *sql.DB + var pool *db.DB if conf.DatabaseURL != "" { - var err error - pool, err = pg.Open(conf.DatabaseURL) + primary, err := pg.Open(conf.DatabaseURL) if err != nil { return nil, err } + var replicas []*db.Host + for _, replicaURL := range conf.DatabaseReplicaURLs { + r, err := pg.OpenReplica(replicaURL) + if err != nil { + // Close already-opened replicas before returning + for _, opened := range replicas { + opened.DB.Close() + } + primary.DB.Close() + return nil, fmt.Errorf("failed to open database replica: %w", err) + } + replicas = append(replicas, r) + } + pool = db.New(primary, replicas) } messageCache, err := createMessageCache(conf, pool) if err != nil { @@ -277,7 +290,7 @@ func New(conf *Config) (*Server, error) { return s, nil } -func createMessageCache(conf *Config, pool *sql.DB) (*message.Cache, error) { +func createMessageCache(conf *Config, pool *db.DB) (*message.Cache, error) { if conf.CacheDuration == 0 { return message.NewNopStore() } else if pool != nil { diff --git a/test/server.go b/test/server.go index 5398cf9e..21e3af78 100644 --- a/test/server.go +++ b/test/server.go @@ -3,7 +3,7 @@ package test import ( "fmt" "heckel.io/ntfy/v2/server" - "math/rand" + "net" "net/http" "path/filepath" "testing" @@ -16,7 +16,7 @@ func StartServer(t *testing.T) (*server.Server, int) { // StartServerWithConfig starts a server.Server with a random port and waits for the server to be up func StartServerWithConfig(t *testing.T, conf *server.Config) (*server.Server, int) { - port := 10000 + rand.Intn(30000) + port := findAvailablePort(t) conf.ListenHTTP = fmt.Sprintf(":%d", port) conf.AttachmentCacheDir = t.TempDir() conf.CacheFile = filepath.Join(t.TempDir(), "cache.db") @@ -33,6 +33,17 @@ func StartServerWithConfig(t *testing.T, conf *server.Config) (*server.Server, i return s, port } +// findAvailablePort asks the OS for a free port by binding to :0 +func findAvailablePort(t *testing.T) int { + listener, err := net.Listen("tcp", ":0") + if err != nil { + t.Fatal(err) + } + port := listener.Addr().(*net.TCPAddr).Port + listener.Close() + return port +} + // StopServer stops the test server and waits for the port to be down func StopServer(t *testing.T, s *server.Server, port int) { s.Stop() diff --git a/tools/pgimport/main.go b/tools/pgimport/main.go index cbc171dd..3ba5273e 100644 --- a/tools/pgimport/main.go +++ b/tools/pgimport/main.go @@ -236,10 +236,11 @@ func execImport(c *cli.Context) error { } fmt.Println() - pgDB, err := pg.Open(databaseURL) + pgHost, err := pg.Open(databaseURL) if err != nil { return fmt.Errorf("cannot connect to PostgreSQL: %w", err) } + pgDB := pgHost.DB defer pgDB.Close() if c.Bool("create-schema") { diff --git a/user/manager.go b/user/manager.go index 0ee6a6e1..bc1a13d3 100644 --- a/user/manager.go +++ b/user/manager.go @@ -49,7 +49,7 @@ var ( // Manager handles user authentication, authorization, and management type Manager struct { config *Config - db *sql.DB + db *db.DB queries queries statsQueue map[string]*Stats // "Queue" to asynchronously write user stats to the database (UserID -> Stats) tokenQueue map[string]*TokenUpdate // "Queue" to asynchronously write token access stats to the database (Token ID -> TokenUpdate) @@ -58,7 +58,7 @@ type Manager struct { var _ Auther = (*Manager)(nil) -func newManager(db *sql.DB, queries queries, config *Config) (*Manager, error) { +func newManager(d *db.DB, queries queries, config *Config) (*Manager, error) { if config.BcryptCost <= 0 { config.BcryptCost = DefaultUserPasswordBcryptCost } @@ -67,7 +67,7 @@ func newManager(db *sql.DB, queries queries, config *Config) (*Manager, error) { } manager := &Manager{ config: config, - db: db, + db: d, statsQueue: make(map[string]*Stats), tokenQueue: make(map[string]*TokenUpdate), queries: queries, @@ -415,7 +415,7 @@ func (a *Manager) userByToken(token string) (*User, error) { // UserByStripeCustomer returns the user with the given Stripe customer ID if it exists, or ErrUserNotFound otherwise func (a *Manager) UserByStripeCustomer(customerID string) (*User, error) { - rows, err := a.db.Query(a.queries.selectUserByStripeCustomerID, customerID) + rows, err := a.db.ReadOnly().Query(a.queries.selectUserByStripeCustomerID, customerID) if err != nil { return nil, err } @@ -425,7 +425,7 @@ func (a *Manager) UserByStripeCustomer(customerID string) (*User, error) { // Users returns a list of users. It loads all users in a single query // rather than one query per user to avoid N+1 performance issues. func (a *Manager) Users() ([]*User, error) { - rows, err := a.db.Query(a.queries.selectUsers) + rows, err := a.db.ReadOnly().Query(a.queries.selectUsers) if err != nil { return nil, err } @@ -434,7 +434,7 @@ func (a *Manager) Users() ([]*User, error) { // UsersCount returns the number of users in the database func (a *Manager) UsersCount() (int64, error) { - rows, err := a.db.Query(a.queries.selectUserCount) + rows, err := a.db.ReadOnly().Query(a.queries.selectUserCount) if err != nil { return 0, err } @@ -660,7 +660,7 @@ func (a *Manager) authorizeTopicAccess(usernameOrEveryone, topic string) (read, // AllGrants returns all user-specific access control entries, mapped to their respective user IDs func (a *Manager) AllGrants() (map[string][]Grant, error) { - rows, err := a.db.Query(a.queries.selectUserAllAccess) + rows, err := a.db.ReadOnly().Query(a.queries.selectUserAllAccess) if err != nil { return nil, err } @@ -688,7 +688,7 @@ func (a *Manager) AllGrants() (map[string][]Grant, error) { // Grants returns all user-specific access control entries func (a *Manager) Grants(username string) ([]Grant, error) { - rows, err := a.db.Query(a.queries.selectUserAccess, username) + rows, err := a.db.ReadOnly().Query(a.queries.selectUserAccess, username) if err != nil { return nil, err } @@ -753,7 +753,7 @@ func (a *Manager) RemoveReservations(username string, topics ...string) error { // Reservations returns all user-owned topics, and the associated everyone-access func (a *Manager) Reservations(username string) ([]Reservation, error) { - rows, err := a.db.Query(a.queries.selectUserReservations, Everyone, username) + rows, err := a.db.ReadOnly().Query(a.queries.selectUserReservations, Everyone, username) if err != nil { return nil, err } @@ -796,7 +796,7 @@ func (a *Manager) HasReservation(username, topic string) (bool, error) { // ReservationsCount returns the number of reservations owned by this user func (a *Manager) ReservationsCount(username string) (int64, error) { - rows, err := a.db.Query(a.queries.selectUserReservationsCount, username) + rows, err := a.db.ReadOnly().Query(a.queries.selectUserReservationsCount, username) if err != nil { return 0, err } @@ -962,7 +962,7 @@ func (a *Manager) canChangeToken(userID, token string) error { // Token returns a specific token for a user func (a *Manager) Token(userID, token string) (*Token, error) { - rows, err := a.db.Query(a.queries.selectToken, userID, token) + rows, err := a.db.ReadOnly().Query(a.queries.selectToken, userID, token) if err != nil { return nil, err } @@ -972,7 +972,7 @@ func (a *Manager) Token(userID, token string) (*Token, error) { // Tokens returns all existing tokens for the user with the given user ID func (a *Manager) Tokens(userID string) ([]*Token, error) { - rows, err := a.db.Query(a.queries.selectTokens, userID) + rows, err := a.db.ReadOnly().Query(a.queries.selectTokens, userID) if err != nil { return nil, err } @@ -991,7 +991,7 @@ func (a *Manager) Tokens(userID string) ([]*Token, error) { } func (a *Manager) allProvisionedTokens() ([]*Token, error) { - rows, err := a.db.Query(a.queries.selectAllProvisionedTokens) + rows, err := a.db.ReadOnly().Query(a.queries.selectAllProvisionedTokens) if err != nil { return nil, err } @@ -1114,7 +1114,7 @@ func (a *Manager) RemoveTier(code string) error { // Tiers returns a list of all Tier structs func (a *Manager) Tiers() ([]*Tier, error) { - rows, err := a.db.Query(a.queries.selectTiers) + rows, err := a.db.ReadOnly().Query(a.queries.selectTiers) if err != nil { return nil, err } @@ -1134,7 +1134,7 @@ func (a *Manager) Tiers() ([]*Tier, error) { // Tier returns a Tier based on the code, or ErrTierNotFound if it does not exist func (a *Manager) Tier(code string) (*Tier, error) { - rows, err := a.db.Query(a.queries.selectTierByCode, code) + rows, err := a.db.ReadOnly().Query(a.queries.selectTierByCode, code) if err != nil { return nil, err } @@ -1144,7 +1144,7 @@ func (a *Manager) Tier(code string) (*Tier, error) { // TierByStripePrice returns a Tier based on the Stripe price ID, or ErrTierNotFound if it does not exist func (a *Manager) TierByStripePrice(priceID string) (*Tier, error) { - rows, err := a.db.Query(a.queries.selectTierByPriceID, priceID, priceID) + rows, err := a.db.ReadOnly().Query(a.queries.selectTierByPriceID, priceID, priceID) if err != nil { return nil, err } @@ -1185,7 +1185,7 @@ func (a *Manager) readTier(rows *sql.Rows) (*Tier, error) { // PhoneNumbers returns all phone numbers for the user with the given user ID func (a *Manager) PhoneNumbers(userID string) ([]string, error) { - rows, err := a.db.Query(a.queries.selectPhoneNumbers, userID) + rows, err := a.db.ReadOnly().Query(a.queries.selectPhoneNumbers, userID) if err != nil { return nil, err } diff --git a/user/manager_postgres.go b/user/manager_postgres.go index 7138ae2c..77c35ece 100644 --- a/user/manager_postgres.go +++ b/user/manager_postgres.go @@ -1,7 +1,7 @@ package user import ( - "database/sql" + "heckel.io/ntfy/v2/db" ) // PostgreSQL queries @@ -278,9 +278,9 @@ var postgresQueries = queries{ } // NewPostgresManager creates a new Manager backed by a PostgreSQL database -func NewPostgresManager(db *sql.DB, config *Config) (*Manager, error) { - if err := setupPostgres(db); err != nil { +func NewPostgresManager(d *db.DB, config *Config) (*Manager, error) { + if err := setupPostgres(d.Primary()); err != nil { return nil, err } - return newManager(db, postgresQueries, config) + return newManager(d, postgresQueries, config) } diff --git a/user/manager_sqlite.go b/user/manager_sqlite.go index b4068599..e92c6349 100644 --- a/user/manager_sqlite.go +++ b/user/manager_sqlite.go @@ -7,6 +7,7 @@ import ( _ "github.com/mattn/go-sqlite3" // SQLite driver + "heckel.io/ntfy/v2/db" "heckel.io/ntfy/v2/util" ) @@ -280,15 +281,15 @@ func NewSQLiteManager(filename, startupQueries string, config *Config) (*Manager if !util.FileExists(parentDir) { return nil, fmt.Errorf("user database directory %s does not exist or is not accessible", parentDir) } - db, err := sql.Open("sqlite3", filename) + d, err := sql.Open("sqlite3", filename) if err != nil { return nil, err } - if err := setupSQLite(db); err != nil { + if err := setupSQLite(d); err != nil { return nil, err } - if err := runSQLiteStartupQueries(db, startupQueries); err != nil { + if err := runSQLiteStartupQueries(d, startupQueries); err != nil { return nil, err } - return newManager(db, sqliteQueries, config) + return newManager(db.New(&db.Host{DB: d}, nil), sqliteQueries, config) } diff --git a/user/manager_test.go b/user/manager_test.go index 53cae1d1..3e023909 100644 --- a/user/manager_test.go +++ b/user/manager_test.go @@ -11,6 +11,7 @@ import ( "github.com/stretchr/testify/require" "golang.org/x/crypto/bcrypt" + "heckel.io/ntfy/v2/db" "heckel.io/ntfy/v2/db/pg" dbtest "heckel.io/ntfy/v2/db/test" "heckel.io/ntfy/v2/util" @@ -36,9 +37,9 @@ func forEachBackend(t *testing.T, f func(t *testing.T, newManager newManagerFunc t.Run("postgres", func(t *testing.T) { schemaDSN := dbtest.CreateTestPostgresSchema(t) f(t, func(config *Config) *Manager { - pool, err := pg.Open(schemaDSN) + host, err := pg.Open(schemaDSN) require.Nil(t, err) - a, err := NewPostgresManager(pool, config) + a, err := NewPostgresManager(db.New(host, nil), config) require.Nil(t, err) return a }) @@ -1734,8 +1735,8 @@ func TestMigrationFrom4(t *testing.T) { require.Nil(t, a.Authorize(nil, "up", PermissionRead)) // % matches 0 or more characters } -func checkSchemaVersion(t *testing.T, db *sql.DB) { - rows, err := db.Query(`SELECT version FROM schemaVersion`) +func checkSchemaVersion(t *testing.T, d *db.DB) { + rows, err := d.Query(`SELECT version FROM schemaVersion`) require.Nil(t, err) require.True(t, rows.Next()) @@ -1771,7 +1772,7 @@ func newTestManagerFromConfig(t *testing.T, newManager newManagerFunc, conf *Con return a } -func testDB(a *Manager) *sql.DB { +func testDB(a *Manager) *db.DB { return a.db } diff --git a/web/package-lock.json b/web/package-lock.json index e775138f..bec8660f 100644 --- a/web/package-lock.json +++ b/web/package-lock.json @@ -4324,9 +4324,9 @@ } }, "node_modules/es-iterator-helpers": { - "version": "1.2.2", - "resolved": "https://registry.npmjs.org/es-iterator-helpers/-/es-iterator-helpers-1.2.2.tgz", - "integrity": "sha512-BrUQ0cPTB/IwXj23HtwHjS9n7O4h9FX94b4xc5zlTHxeLgTAdzYUDyy6KdExAl9lbN5rtfe44xpjpmj9grxs5w==", + "version": "1.3.0", + "resolved": "https://registry.npmjs.org/es-iterator-helpers/-/es-iterator-helpers-1.3.0.tgz", + "integrity": "sha512-04cg8iJFDOxWcYlu0GFFWgs7vtaEPCmr5w1nrj9V3z3axu/48HCMwK6VMp45Zh3ZB+xLP1ifbJfrq86+1ypKKQ==", "dev": true, "license": "MIT", "dependencies": { @@ -4345,6 +4345,7 @@ "has-symbols": "^1.1.0", "internal-slot": "^1.1.0", "iterator.prototype": "^1.1.5", + "math-intrinsics": "^1.1.0", "safe-array-concat": "^1.1.3" }, "engines": { @@ -5065,9 +5066,9 @@ } }, "node_modules/flatted": { - "version": "3.3.4", - "resolved": "https://registry.npmjs.org/flatted/-/flatted-3.3.4.tgz", - "integrity": "sha512-3+mMldrTAPdta5kjX2G2J7iX4zxtnwpdA8Tr2ZSjkyPSanvbZAcy6flmtnXbEybHrDcU9641lxrMfFuUxVz9vA==", + "version": "3.4.1", + "resolved": "https://registry.npmjs.org/flatted/-/flatted-3.4.1.tgz", + "integrity": "sha512-IxfVbRFVlV8V/yRaGzk0UVIcsKKHMSfYw66T/u4nTwlWteQePsxe//LjudR1AMX4tZW3WFCh3Zqa/sjlqpbURQ==", "dev": true, "license": "ISC" }, diff --git a/webpush/store.go b/webpush/store.go index 9a93a074..02b7552e 100644 --- a/webpush/store.go +++ b/webpush/store.go @@ -24,7 +24,7 @@ var ( // Store holds the database connection and queries for web push subscriptions. type Store struct { - db *sql.DB + db *db.DB queries queries } @@ -83,7 +83,7 @@ func (s *Store) UpsertSubscription(endpoint string, auth, p256dh, userID string, // SubscriptionsForTopic returns all subscriptions for the given topic. func (s *Store) SubscriptionsForTopic(topic string) ([]*Subscription, error) { - rows, err := s.db.Query(s.queries.selectSubscriptionsForTopic, topic) + rows, err := s.db.ReadOnly().Query(s.queries.selectSubscriptionsForTopic, topic) if err != nil { return nil, err } @@ -93,7 +93,7 @@ func (s *Store) SubscriptionsForTopic(topic string) ([]*Subscription, error) { // SubscriptionsExpiring returns all subscriptions that have not been updated for a given time period. func (s *Store) SubscriptionsExpiring(warnAfter time.Duration) ([]*Subscription, error) { - rows, err := s.db.Query(s.queries.selectSubscriptionsExpiringSoon, time.Now().Add(-warnAfter).Unix()) + rows, err := s.db.ReadOnly().Query(s.queries.selectSubscriptionsExpiringSoon, time.Now().Add(-warnAfter).Unix()) if err != nil { return nil, err } diff --git a/webpush/store_postgres.go b/webpush/store_postgres.go index ec541d37..1c9adf0a 100644 --- a/webpush/store_postgres.go +++ b/webpush/store_postgres.go @@ -73,12 +73,12 @@ const ( ) // NewPostgresStore creates a new PostgreSQL-backed web push store using an existing database connection pool. -func NewPostgresStore(db *sql.DB) (*Store, error) { - if err := setupPostgres(db); err != nil { +func NewPostgresStore(d *db.DB) (*Store, error) { + if err := setupPostgres(d.Primary()); err != nil { return nil, err } return &Store{ - db: db, + db: d, queries: queries{ selectSubscriptionIDByEndpoint: postgresSelectSubscriptionIDByEndpointQuery, selectSubscriptionCountBySubscriberIP: postgresSelectSubscriptionCountBySubscriberIPQuery, @@ -97,11 +97,11 @@ func NewPostgresStore(db *sql.DB) (*Store, error) { }, nil } -func setupPostgres(db *sql.DB) error { +func setupPostgres(d *sql.DB) error { var schemaVersion int - err := db.QueryRow(postgresSelectSchemaVersionQuery).Scan(&schemaVersion) + err := d.QueryRow(postgresSelectSchemaVersionQuery).Scan(&schemaVersion) if err != nil { - return setupNewPostgres(db) + return setupNewPostgres(d) } if schemaVersion > pgCurrentSchemaVersion { return fmt.Errorf("unexpected schema version: version %d is higher than current version %d", schemaVersion, pgCurrentSchemaVersion) @@ -109,8 +109,8 @@ func setupPostgres(db *sql.DB) error { return nil } -func setupNewPostgres(sqlDB *sql.DB) error { - return db.ExecTx(sqlDB, func(tx *sql.Tx) error { +func setupNewPostgres(d *sql.DB) error { + return db.ExecTx(d, func(tx *sql.Tx) error { if _, err := tx.Exec(postgresCreateTablesQuery); err != nil { return err } diff --git a/webpush/store_sqlite.go b/webpush/store_sqlite.go index 4ef78140..fcf49fcf 100644 --- a/webpush/store_sqlite.go +++ b/webpush/store_sqlite.go @@ -79,18 +79,18 @@ const ( // NewSQLiteStore creates a new SQLite-backed web push store. func NewSQLiteStore(filename, startupQueries string) (*Store, error) { - db, err := sql.Open("sqlite3", filename) + d, err := sql.Open("sqlite3", filename) if err != nil { return nil, err } - if err := setupSQLite(db); err != nil { + if err := setupSQLite(d); err != nil { return nil, err } - if err := runSQLiteStartupQueries(db, startupQueries); err != nil { + if err := runSQLiteStartupQueries(d, startupQueries); err != nil { return nil, err } return &Store{ - db: db, + db: db.New(&db.Host{DB: d}, nil), queries: queries{ selectSubscriptionIDByEndpoint: sqliteSelectSubscriptionIDByEndpointQuery, selectSubscriptionCountBySubscriberIP: sqliteSelectSubscriptionCountBySubscriberIPQuery,