Route authorization query to read-only database replica to reduce primary database load

Support `postgresql://` and `postgres://` URLs

cmd/publish_test.go

@@ -2,9 +2,6 @@ package cmd
 
 import (
 	"fmt"
-	"github.com/stretchr/testify/require"
-	"heckel.io/ntfy/v2/test"
-	"heckel.io/ntfy/v2/util"
 	"net/http"
 	"net/http/httptest"
 	"os"
@@ -14,9 +11,14 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/stretchr/testify/require"
+	"heckel.io/ntfy/v2/test"
+	"heckel.io/ntfy/v2/util"
 )
 
 func TestCLI_Publish_Subscribe_Poll_Real_Server(t *testing.T) {
+	t.Skip("temporarily disabled") // FIXME
 	testMessage := util.RandomString(10)
 	app, _, _, _ := newTestApp()
 	require.Nil(t, app.Run([]string{"ntfy", "publish", "ntfytest", "ntfy unit test " + testMessage}))

cmd/serve.go

@@ -284,8 +284,8 @@ func execServe(c *cli.Context) error {
 	}
 
 	// Check values
-	if databaseURL != "" && !strings.HasPrefix(databaseURL, "postgres://") {
-		return errors.New("if database-url is set, it must start with postgres://")
+	if databaseURL != "" && !strings.HasPrefix(databaseURL, "postgres://") && !strings.HasPrefix(databaseURL, "postgresql://") {
+		return errors.New("if database-url is set, it must start with postgres:// or postgresql://")
 	} 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 len(databaseReplicaURLs) > 0 && databaseURL == "" {

docs/install.md

@@ -30,37 +30,37 @@ deb/rpm packages.
 
 === "x86_64/amd64"
     ```bash
-    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_amd64.tar.gz
-    tar zxvf ntfy_2.19.0_linux_amd64.tar.gz
-    sudo cp -a ntfy_2.19.0_linux_amd64/ntfy /usr/local/bin/ntfy
-    sudo mkdir /etc/ntfy && sudo cp ntfy_2.19.0_linux_amd64/{client,server}/*.yml /etc/ntfy
+    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_amd64.tar.gz
+    tar zxvf ntfy_2.19.1_linux_amd64.tar.gz
+    sudo cp -a ntfy_2.19.1_linux_amd64/ntfy /usr/local/bin/ntfy
+    sudo mkdir /etc/ntfy && sudo cp ntfy_2.19.1_linux_amd64/{client,server}/*.yml /etc/ntfy
     sudo ntfy serve
     ```
 
 === "armv6"
     ```bash
-    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_armv6.tar.gz
-    tar zxvf ntfy_2.19.0_linux_armv6.tar.gz
-    sudo cp -a ntfy_2.19.0_linux_armv6/ntfy /usr/bin/ntfy
-    sudo mkdir /etc/ntfy && sudo cp ntfy_2.19.0_linux_armv6/{client,server}/*.yml /etc/ntfy
+    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_armv6.tar.gz
+    tar zxvf ntfy_2.19.1_linux_armv6.tar.gz
+    sudo cp -a ntfy_2.19.1_linux_armv6/ntfy /usr/bin/ntfy
+    sudo mkdir /etc/ntfy && sudo cp ntfy_2.19.1_linux_armv6/{client,server}/*.yml /etc/ntfy
     sudo ntfy serve
     ```
 
 === "armv7/armhf"
     ```bash
-    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_armv7.tar.gz
-    tar zxvf ntfy_2.19.0_linux_armv7.tar.gz
-    sudo cp -a ntfy_2.19.0_linux_armv7/ntfy /usr/bin/ntfy
-    sudo mkdir /etc/ntfy && sudo cp ntfy_2.19.0_linux_armv7/{client,server}/*.yml /etc/ntfy
+    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_armv7.tar.gz
+    tar zxvf ntfy_2.19.1_linux_armv7.tar.gz
+    sudo cp -a ntfy_2.19.1_linux_armv7/ntfy /usr/bin/ntfy
+    sudo mkdir /etc/ntfy && sudo cp ntfy_2.19.1_linux_armv7/{client,server}/*.yml /etc/ntfy
     sudo ntfy serve
     ```
 
 === "arm64"
     ```bash
-    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_arm64.tar.gz
-    tar zxvf ntfy_2.19.0_linux_arm64.tar.gz
-    sudo cp -a ntfy_2.19.0_linux_arm64/ntfy /usr/bin/ntfy
-    sudo mkdir /etc/ntfy && sudo cp ntfy_2.19.0_linux_arm64/{client,server}/*.yml /etc/ntfy
+    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_arm64.tar.gz
+    tar zxvf ntfy_2.19.1_linux_arm64.tar.gz
+    sudo cp -a ntfy_2.19.1_linux_arm64/ntfy /usr/bin/ntfy
+    sudo mkdir /etc/ntfy && sudo cp ntfy_2.19.1_linux_arm64/{client,server}/*.yml /etc/ntfy
     sudo ntfy serve
     ```
 
@@ -116,7 +116,7 @@ Manually installing the .deb file:
 
 === "x86_64/amd64"
     ```bash
-    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_amd64.deb
+    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_amd64.deb
     sudo dpkg -i ntfy_*.deb
     sudo systemctl enable ntfy
     sudo systemctl start ntfy
@@ -124,7 +124,7 @@ Manually installing the .deb file:
 
 === "armv6"
     ```bash
-    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_armv6.deb
+    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_armv6.deb
     sudo dpkg -i ntfy_*.deb
     sudo systemctl enable ntfy
     sudo systemctl start ntfy
@@ -132,7 +132,7 @@ Manually installing the .deb file:
 
 === "armv7/armhf"
     ```bash
-    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_armv7.deb
+    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_armv7.deb
     sudo dpkg -i ntfy_*.deb
     sudo systemctl enable ntfy
     sudo systemctl start ntfy
@@ -140,7 +140,7 @@ Manually installing the .deb file:
 
 === "arm64"
     ```bash
-    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_arm64.deb
+    wget https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_arm64.deb
     sudo dpkg -i ntfy_*.deb
     sudo systemctl enable ntfy
     sudo systemctl start ntfy
@@ -150,28 +150,28 @@ Manually installing the .deb file:
 
 === "x86_64/amd64"
     ```bash
-    sudo rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_amd64.rpm
+    sudo rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_amd64.rpm
     sudo systemctl enable ntfy 
     sudo systemctl start ntfy
     ```
 
 === "armv6"
     ```bash
-    sudo rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_armv6.rpm
+    sudo rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_armv6.rpm
     sudo systemctl enable ntfy
     sudo systemctl start ntfy
     ```
 
 === "armv7/armhf"
     ```bash
-    sudo rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_armv7.rpm
+    sudo rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_armv7.rpm
     sudo systemctl enable ntfy 
     sudo systemctl start ntfy
     ```
 
 === "arm64"
     ```bash
-    sudo rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_linux_arm64.rpm
+    sudo rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_linux_arm64.rpm
     sudo systemctl enable ntfy 
     sudo systemctl start ntfy
     ```
@@ -213,18 +213,18 @@ pkg install go-ntfy
 
 ## macOS
 The [ntfy CLI](subscribe/cli.md) (`ntfy publish` and `ntfy subscribe` only) is supported on macOS as well. 
-To install, please [download the tarball](https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_darwin_all.tar.gz), 
+To install, please [download the tarball](https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_darwin_all.tar.gz), 
 extract it and place it somewhere in your `PATH` (e.g. `/usr/local/bin/ntfy`). 
 
 If run as `root`, ntfy will look for its config at `/etc/ntfy/client.yml`. For all other users, it'll look for it at 
 `~/Library/Application Support/ntfy/client.yml` (sample included in the tarball).
 
 ```bash
-curl -L https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_darwin_all.tar.gz > ntfy_2.19.0_darwin_all.tar.gz
-tar zxvf ntfy_2.19.0_darwin_all.tar.gz
-sudo cp -a ntfy_2.19.0_darwin_all/ntfy /usr/local/bin/ntfy
+curl -L https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_darwin_all.tar.gz > ntfy_2.19.1_darwin_all.tar.gz
+tar zxvf ntfy_2.19.1_darwin_all.tar.gz
+sudo cp -a ntfy_2.19.1_darwin_all/ntfy /usr/local/bin/ntfy
 mkdir ~/Library/Application\ Support/ntfy 
-cp ntfy_2.19.0_darwin_all/client/client.yml ~/Library/Application\ Support/ntfy/client.yml
+cp ntfy_2.19.1_darwin_all/client/client.yml ~/Library/Application\ Support/ntfy/client.yml
 ntfy --help
 ```
 
@@ -245,7 +245,7 @@ brew install ntfy
 The ntfy server and CLI are fully supported on Windows. You can run the ntfy server directly or as a Windows service.
 To install, you can either
 
-* [Download the latest ZIP](https://github.com/binwiederhier/ntfy/releases/download/v2.19.0/ntfy_2.19.0_windows_amd64.zip),
+* [Download the latest ZIP](https://github.com/binwiederhier/ntfy/releases/download/v2.19.1/ntfy_2.19.1_windows_amd64.zip),
 extract it and place the `ntfy.exe` binary somewhere in your `%Path%`. 
 * Or install ntfy from the [Scoop](https://scoop.sh) main repository via `scoop install ntfy`
 

docs/releases.md

@@ -6,13 +6,25 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release
 
 | Component        | Version | Release date |
 |------------------|---------|--------------|
-| ntfy server      | v2.19.0 | Mar 15, 2026 |
+| ntfy server      | v2.19.1 | Mar 15, 2026 |
 | ntfy Android app | v1.24.0 | Mar 5, 2026  |
 | ntfy iOS app     | v1.3    | Nov 26, 2023 |
 
 Please check out the release notes for [upcoming releases](#not-released-yet) below.
 
+## ntfy server v2.19.1
+Released March 15, 2026
+
+This is a bugfix release to avoid PostgreSQL insert failures due to invalid UTF-8 messages. It also fixes `database-url`
+validation incorrectly rejecting `postgresql://` connection strings.
+
+**Bug fixes + maintenance:**
+
+* Fix invalid UTF-8 in HTTP headers (e.g. Latin-1 encoded text) causing PostgreSQL insert failures and dropping entire message batches
+* Fix `database-url` validation rejecting `postgresql://` connection strings ([#1657](https://github.com/binwiederhier/ntfy/issues/1657)/[#1658](https://github.com/binwiederhier/ntfy/pull/1658))
+
 ## ntfy server v2.19.0
+Released March 15, 2026
 
 This is a fast-follow release that enables Postgres read replica support.
 
@@ -1775,4 +1787,8 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release
 
 ## Not released yet
 
-Nothing to see here.
+### ntfy server v2.20.x (UNRELEASED)
+
+**Bug fixes + maintenance:**
+
+* Route authorization query to read-only database replica to reduce primary database load

message/cache.go

@@ -125,16 +125,16 @@ func (c *Cache) addMessages(ms []*model.Message) error {
 			return model.ErrUnexpectedMessageType
 		}
 		published := m.Time <= time.Now().Unix()
-		tags := strings.Join(m.Tags, ",")
+		tags := util.SanitizeUTF8(strings.Join(m.Tags, ","))
 		var attachmentName, attachmentType, attachmentURL string
 		var attachmentSize, attachmentExpires int64
 		var attachmentDeleted bool
 		if m.Attachment != nil {
-			attachmentName = m.Attachment.Name
-			attachmentType = m.Attachment.Type
+			attachmentName = util.SanitizeUTF8(m.Attachment.Name)
+			attachmentType = util.SanitizeUTF8(m.Attachment.Type)
 			attachmentSize = m.Attachment.Size
 			attachmentExpires = m.Attachment.Expires
-			attachmentURL = m.Attachment.URL
+			attachmentURL = util.SanitizeUTF8(m.Attachment.URL)
 		}
 		var actionsStr string
 		if len(m.Actions) > 0 {
@@ -154,13 +154,13 @@ func (c *Cache) addMessages(ms []*model.Message) error {
 			m.Time,
 			m.Event,
 			m.Expires,
-			m.Topic,
-			m.Message,
-			m.Title,
+			util.SanitizeUTF8(m.Topic),
+			util.SanitizeUTF8(m.Message),
+			util.SanitizeUTF8(m.Title),
 			m.Priority,
 			tags,
-			m.Click,
-			m.Icon,
+			util.SanitizeUTF8(m.Click),
+			util.SanitizeUTF8(m.Icon),
 			actionsStr,
 			attachmentName,
 			attachmentType,
@@ -170,7 +170,7 @@ func (c *Cache) addMessages(ms []*model.Message) error {
 			attachmentDeleted, // Always zero
 			sender,
 			m.User,
-			m.ContentType,
+			util.SanitizeUTF8(m.ContentType),
 			m.Encoding,
 			published,
 		)

message/cache_test.go

@@ -827,3 +827,141 @@ func TestStore_MessageFieldRoundTrip(t *testing.T) {
 		require.Equal(t, `{"key":"value"}`, retrieved.Actions[1].Body)
 	})
 }
+
+func TestStore_AddMessage_InvalidUTF8(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, s *message.Cache) {
+		// 0xc9 0x43: Latin-1 "ÉC" — 0xc9 starts a 2-byte UTF-8 sequence but 0x43 ('C') is not a continuation byte
+		m := model.NewDefaultMessage("mytopic", "\xc9Cas du serveur")
+		require.Nil(t, s.AddMessage(m))
+		messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, 1, len(messages))
+		require.Equal(t, "\uFFFDCas du serveur", messages[0].Message)
+
+		// 0xae: Latin-1 "®" — isolated byte above 0x7F, not a valid UTF-8 start for single byte
+		m2 := model.NewDefaultMessage("mytopic", "Product\xae Pro")
+		require.Nil(t, s.AddMessage(m2))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "Product\uFFFD Pro", messages[1].Message)
+
+		// 0xe8 0x6d 0x65: Latin-1 "ème" — 0xe8 starts a 3-byte UTF-8 sequence but 0x6d ('m') is not a continuation byte
+		m3 := model.NewDefaultMessage("mytopic", "probl\xe8me critique")
+		require.Nil(t, s.AddMessage(m3))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "probl\uFFFDme critique", messages[2].Message)
+
+		// 0xb2: Latin-1 "²" — isolated byte in 0x80-0xBF range (UTF-8 continuation byte without lead)
+		m4 := model.NewDefaultMessage("mytopic", "CO\xb2 level high")
+		require.Nil(t, s.AddMessage(m4))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "CO\uFFFD level high", messages[3].Message)
+
+		// 0xe9 0x6d 0x61: Latin-1 "éma" — 0xe9 starts a 3-byte UTF-8 sequence but 0x6d ('m') is not a continuation byte
+		m5 := model.NewDefaultMessage("mytopic", "th\xe9matique")
+		require.Nil(t, s.AddMessage(m5))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "th\uFFFDmatique", messages[4].Message)
+
+		// 0xed 0x64 0x65: Latin-1 "íde" — 0xed starts a 3-byte UTF-8 sequence but 0x64 ('d') is not a continuation byte
+		m6 := model.NewDefaultMessage("mytopic", "vid\xed\x64eo surveillance")
+		require.Nil(t, s.AddMessage(m6))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "vid\uFFFDdeo surveillance", messages[5].Message)
+
+		// 0xf3 0x6e 0x3a 0x20: Latin-1 "ón: " — 0xf3 starts a 4-byte UTF-8 sequence but 0x6e ('n') is not a continuation byte
+		m7 := model.NewDefaultMessage("mytopic", "notificaci\xf3n: alerta")
+		require.Nil(t, s.AddMessage(m7))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "notificaci\uFFFDn: alerta", messages[6].Message)
+
+		// 0xb7: Latin-1 "·" — isolated continuation byte
+		m8 := model.NewDefaultMessage("mytopic", "item\xb7value")
+		require.Nil(t, s.AddMessage(m8))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "item\uFFFDvalue", messages[7].Message)
+
+		// 0xa8: Latin-1 "¨" — isolated continuation byte
+		m9 := model.NewDefaultMessage("mytopic", "na\xa8ve")
+		require.Nil(t, s.AddMessage(m9))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "na\uFFFDve", messages[8].Message)
+
+		// 0xdf 0x64: Latin-1 "ßd" — 0xdf starts a 2-byte UTF-8 sequence but 0x64 ('d') is not a continuation byte
+		m10 := model.NewDefaultMessage("mytopic", "gro\xdf\x64ruck")
+		require.Nil(t, s.AddMessage(m10))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "gro\uFFFDdruck", messages[9].Message)
+
+		// 0xe4 0x67 0x74: Latin-1 "ägt" — 0xe4 starts a 3-byte UTF-8 sequence but 0x67 ('g') is not a continuation byte
+		m11 := model.NewDefaultMessage("mytopic", "tr\xe4gt Last")
+		require.Nil(t, s.AddMessage(m11))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "tr\uFFFDgt Last", messages[10].Message)
+
+		// 0xe9 0x65 0x20: Latin-1 "ée " — 0xe9 starts a 3-byte UTF-8 sequence but 0x65 ('e') is not a continuation byte
+		m12 := model.NewDefaultMessage("mytopic", "journ\xe9\x65 termin\xe9\x65")
+		require.Nil(t, s.AddMessage(m12))
+		messages, err = s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, "journ\uFFFDe termin\uFFFDe", messages[11].Message)
+	})
+}
+
+func TestStore_AddMessage_NullByte(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, s *message.Cache) {
+		// 0x00: NUL byte — valid UTF-8 but rejected by PostgreSQL
+		m := model.NewDefaultMessage("mytopic", "hello\x00world")
+		require.Nil(t, s.AddMessage(m))
+
+		messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, 1, len(messages))
+		require.Equal(t, "helloworld", messages[0].Message)
+	})
+}
+
+func TestStore_AddMessage_InvalidUTF8InTitleAndTags(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, s *message.Cache) {
+		// Invalid UTF-8 can arrive via HTTP headers (Title, Tags) which bypass body validation
+		m := model.NewDefaultMessage("mytopic", "valid message")
+		m.Title = "\xc9clipse du syst\xe8me"
+		m.Tags = []string{"probl\xe8me", "syst\xe9me"}
+		m.Click = "https://example.com/\xae"
+		require.Nil(t, s.AddMessage(m))
+
+		messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, 1, len(messages))
+		require.Equal(t, "\uFFFDclipse du syst\uFFFDme", messages[0].Title)
+		require.Equal(t, "probl\uFFFDme", messages[0].Tags[0])
+		require.Equal(t, "syst\uFFFDme", messages[0].Tags[1])
+		require.Equal(t, "https://example.com/\uFFFD", messages[0].Click)
+	})
+}
+
+func TestStore_AddMessage_InvalidUTF8BatchDoesNotDropValidMessages(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, s *message.Cache) {
+		// Previously, a single invalid message would roll back the entire batch transaction.
+		// Sanitization ensures all messages in a batch are written successfully.
+		msgs := []*model.Message{
+			model.NewDefaultMessage("mytopic", "valid message 1"),
+			model.NewDefaultMessage("mytopic", "notificaci\xf3n: alerta"),
+			model.NewDefaultMessage("mytopic", "valid message 3"),
+		}
+		require.Nil(t, s.AddMessages(msgs))
+
+		messages, err := s.Messages("mytopic", model.SinceAllMessages, false)
+		require.Nil(t, err)
+		require.Equal(t, 3, len(messages))
+	})
+}

model/model.go

@@ -70,6 +70,26 @@ func (m *Message) Context() log.Context {
 	return fields
 }
 
+// SanitizeUTF8 replaces invalid UTF-8 sequences and strips NUL bytes from all user-supplied
+// string fields. This is called early in the publish path so that all downstream consumers
+// (Firebase, WebPush, SMTP, cache) receive clean UTF-8 strings.
+func (m *Message) SanitizeUTF8() {
+	m.Topic = util.SanitizeUTF8(m.Topic)
+	m.Message = util.SanitizeUTF8(m.Message)
+	m.Title = util.SanitizeUTF8(m.Title)
+	m.Click = util.SanitizeUTF8(m.Click)
+	m.Icon = util.SanitizeUTF8(m.Icon)
+	m.ContentType = util.SanitizeUTF8(m.ContentType)
+	for i, tag := range m.Tags {
+		m.Tags[i] = util.SanitizeUTF8(tag)
+	}
+	if m.Attachment != nil {
+		m.Attachment.Name = util.SanitizeUTF8(m.Attachment.Name)
+		m.Attachment.Type = util.SanitizeUTF8(m.Attachment.Type)
+		m.Attachment.URL = util.SanitizeUTF8(m.Attachment.URL)
+	}
+}
+
 // ForJSON returns a copy of the message suitable for JSON output.
 // It clears the SequenceID if it equals the ID to reduce redundancy.
 func (m *Message) ForJSON() *Message {

server/server.go

@@ -880,6 +880,7 @@ func (s *Server) handlePublishInternal(r *http.Request, v *visitor) (*model.Mess
 	if m.Message == "" {
 		m.Message = emptyMessageBody
 	}
+	m.SanitizeUTF8()
 	delayed := m.Time > time.Now().Unix()
 	ev := logvrm(v, r, m).
 		Tag(tagPublish).

server/server_test.go

@@ -4441,3 +4441,88 @@ func TestServer_HandleError_SkipsWriteHeaderOnHijackedConnection(t *testing.T) {
 		}
 	})
 }
+
+func TestServer_Publish_InvalidUTF8InBody(t *testing.T) {
+	// All byte sequences from production logs, sent as message body
+	tests := []struct {
+		name    string
+		body    string
+		message string
+	}{
+		{"0xc9_0x43", "\xc9Cas du serveur", "\uFFFDCas du serveur"},                      // Latin-1 "ÉC"
+		{"0xae", "Product\xae Pro", "Product\uFFFD Pro"},                                 // Latin-1 "®"
+		{"0xe8_0x6d_0x65", "probl\xe8me critique", "probl\uFFFDme critique"},             // Latin-1 "ème"
+		{"0xb2", "CO\xb2 level high", "CO\uFFFD level high"},                             // Latin-1 "²"
+		{"0xe9_0x6d_0x61", "th\xe9matique", "th\uFFFDmatique"},                           // Latin-1 "éma"
+		{"0xed_0x64_0x65", "vid\xed\x64eo surveillance", "vid\uFFFDdeo surveillance"},    // Latin-1 "íde"
+		{"0xf3_0x6e_0x3a_0x20", "notificaci\xf3n: alerta", "notificaci\uFFFDn: alerta"},  // Latin-1 "ón: "
+		{"0xb7", "item\xb7value", "item\uFFFDvalue"},                                     // Latin-1 "·"
+		{"0xa8", "na\xa8ve", "na\uFFFDve"},                                               // Latin-1 "¨"
+		{"0x00", "hello\x00world", "helloworld"},                                         // NUL byte
+		{"0xdf_0x64", "gro\xdf\x64ruck", "gro\uFFFDdruck"},                               // Latin-1 "ßd"
+		{"0xe4_0x67_0x74", "tr\xe4gt Last", "tr\uFFFDgt Last"},                           // Latin-1 "ägt"
+		{"0xe9_0x65_0x20", "journ\xe9\x65 termin\xe9\x65", "journ\uFFFDe termin\uFFFDe"}, // Latin-1 "ée"
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			s := newTestServer(t, newTestConfig(t, ""))
+
+			// Publish via x-message header (the most common path for invalid UTF-8 from HTTP headers)
+			response := request(t, s, "PUT", "/mytopic", "", map[string]string{
+				"X-Message": tc.body,
+			})
+			require.Equal(t, 200, response.Code)
+			msg := toMessage(t, response.Body.String())
+			require.Equal(t, tc.message, msg.Message)
+
+			// Verify it was stored in the cache correctly
+			response = request(t, s, "GET", "/mytopic/json?poll=1", "", nil)
+			require.Equal(t, 200, response.Code)
+			msg = toMessage(t, response.Body.String())
+			require.Equal(t, tc.message, msg.Message)
+		})
+	}
+}
+
+func TestServer_Publish_InvalidUTF8InTitle(t *testing.T) {
+	s := newTestServer(t, newTestConfig(t, ""))
+	response := request(t, s, "PUT", "/mytopic", "valid body", map[string]string{
+		"Title": "\xc9clipse du syst\xe8me",
+	})
+	require.Equal(t, 200, response.Code)
+	msg := toMessage(t, response.Body.String())
+	require.Equal(t, "\uFFFDclipse du syst\uFFFDme", msg.Title)
+	require.Equal(t, "valid body", msg.Message)
+}
+
+func TestServer_Publish_InvalidUTF8InTags(t *testing.T) {
+	s := newTestServer(t, newTestConfig(t, ""))
+	response := request(t, s, "PUT", "/mytopic", "valid body", map[string]string{
+		"Tags": "probl\xe8me,syst\xe9me",
+	})
+	require.Equal(t, 200, response.Code)
+	msg := toMessage(t, response.Body.String())
+	require.Equal(t, "probl\uFFFDme", msg.Tags[0])
+	require.Equal(t, "syst\uFFFDme", msg.Tags[1])
+}
+
+func TestServer_Publish_InvalidUTF8WithFirebase(t *testing.T) {
+	// Verify that sanitization happens before Firebase dispatch, so Firebase
+	// receives clean UTF-8 strings rather than invalid byte sequences
+	sender := newTestFirebaseSender(10)
+	s := newTestServer(t, newTestConfig(t, ""))
+	s.firebaseClient = newFirebaseClient(sender, &testAuther{Allow: true})
+
+	response := request(t, s, "PUT", "/mytopic", "", map[string]string{
+		"X-Message": "notificaci\xf3n: alerta",
+		"Title":     "\xc9clipse",
+		"Tags":      "probl\xe8me",
+	})
+	require.Equal(t, 200, response.Code)
+
+	time.Sleep(100 * time.Millisecond) // Firebase publishing happens asynchronously
+	require.Equal(t, 1, len(sender.Messages()))
+	require.Equal(t, "notificaci\uFFFDn: alerta", sender.Messages()[0].Data["message"])
+	require.Equal(t, "\uFFFDclipse", sender.Messages()[0].Data["title"])
+	require.Equal(t, "probl\uFFFDme", sender.Messages()[0].Data["tags"])
+}

tools/pgimport/main.go

@@ -65,12 +65,12 @@ const (
 			key TEXT PRIMARY KEY,
 			value BIGINT
 		);
-		INSERT INTO message_stats (key, value) VALUES ('messages', 0);
+		INSERT INTO message_stats (key, value) VALUES ('messages', 0) ON CONFLICT (key) DO NOTHING;
 		CREATE TABLE IF NOT EXISTS schema_version (
 			store TEXT PRIMARY KEY,
 			version INT NOT NULL
 		);
-		INSERT INTO schema_version (store, version) VALUES ('message', 14);
+		INSERT INTO schema_version (store, version) VALUES ('message', 14) ON CONFLICT (store) DO NOTHING;
 	`
 
 	// Initial PostgreSQL schema for user store (from user/manager_postgres_schema.go)
@@ -146,7 +146,7 @@ const (
 		INSERT INTO "user" (id, user_name, pass, role, sync_topic, provisioned, created)
 		VALUES ('` + everyoneID + `', '*', '', 'anonymous', '', false, EXTRACT(EPOCH FROM NOW())::BIGINT)
 		ON CONFLICT (id) DO NOTHING;
-		INSERT INTO schema_version (store, version) VALUES ('user', 6);
+		INSERT INTO schema_version (store, version) VALUES ('user', 6) ON CONFLICT (store) DO NOTHING;
 	`
 
 	// Initial PostgreSQL schema for web push store (from webpush/store_postgres.go)
@@ -174,7 +174,7 @@ const (
 			store TEXT PRIMARY KEY,
 			version INT NOT NULL
 		);
-		INSERT INTO schema_version (store, version) VALUES ('webpush', 1);
+		INSERT INTO schema_version (store, version) VALUES ('webpush', 1) ON CONFLICT (store) DO NOTHING;
 	`
 )
 
@@ -185,6 +185,7 @@ var flags = []cli.Flag{
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "auth-file", Aliases: []string{"auth_file"}, Usage: "SQLite user/auth database file path"}),
 	altsrc.NewStringFlag(&cli.StringFlag{Name: "web-push-file", Aliases: []string{"web_push_file"}, Usage: "SQLite web push database file path"}),
 	&cli.BoolFlag{Name: "create-schema", Usage: "create initial PostgreSQL schema before importing"},
+	&cli.BoolFlag{Name: "pre-import", Usage: "pre-import messages while ntfy is still running (only imports messages)"},
 }
 
 func main() {
@@ -207,10 +208,17 @@ func execImport(c *cli.Context) error {
 	cacheFile := c.String("cache-file")
 	authFile := c.String("auth-file")
 	webPushFile := c.String("web-push-file")
+	preImport := c.Bool("pre-import")
 
 	if databaseURL == "" {
 		return fmt.Errorf("database-url must be set (via --database-url or config file)")
 	}
+	if preImport {
+		if cacheFile == "" {
+			return fmt.Errorf("--cache-file must be set when using --pre-import")
+		}
+		return execPreImport(c, databaseURL, cacheFile)
+	}
 	if cacheFile == "" && authFile == "" && webPushFile == "" {
 		return fmt.Errorf("at least one of --cache-file, --auth-file, or --web-push-file must be set")
 	}
@@ -261,7 +269,8 @@ func execImport(c *cli.Context) error {
 		if err := verifySchemaVersion(pgDB, "message", expectedMessageSchemaVersion); err != nil {
 			return err
 		}
-		if err := importMessages(cacheFile, pgDB); err != nil {
+		sinceTime := maxMessageTime(pgDB)
+		if err := importMessages(cacheFile, pgDB, sinceTime); err != nil {
 			return fmt.Errorf("cannot import messages: %w", err)
 		}
 	}
@@ -300,6 +309,54 @@ func execImport(c *cli.Context) error {
 	return nil
 }
 
+func execPreImport(c *cli.Context, databaseURL, cacheFile string) error {
+	fmt.Println("pgimport - PRE-IMPORT mode (ntfy can keep running)")
+	fmt.Println()
+	fmt.Println("Source:")
+	printSource("  Cache file:    ", cacheFile)
+	fmt.Println()
+	fmt.Println("Target:")
+	fmt.Printf("  Database URL:  %s\n", maskPassword(databaseURL))
+	fmt.Println()
+	fmt.Println("This will pre-import messages into PostgreSQL while ntfy is still running.")
+	fmt.Println("After this completes, stop ntfy and run pgimport again without --pre-import")
+	fmt.Println("to import remaining messages, users, and web push subscriptions.")
+	fmt.Print("Continue? (y/n): ")
+
+	var answer string
+	fmt.Scanln(&answer)
+	if strings.TrimSpace(strings.ToLower(answer)) != "y" {
+		fmt.Println("Aborted.")
+		return nil
+	}
+	fmt.Println()
+
+	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") {
+		if err := createSchema(pgDB, cacheFile, "", ""); err != nil {
+			return fmt.Errorf("cannot create schema: %w", err)
+		}
+	}
+
+	if err := verifySchemaVersion(pgDB, "message", expectedMessageSchemaVersion); err != nil {
+		return err
+	}
+	if err := importMessages(cacheFile, pgDB, 0); err != nil {
+		return fmt.Errorf("cannot import messages: %w", err)
+	}
+
+	fmt.Println()
+	fmt.Println("Pre-import complete. Now stop ntfy and run pgimport again without --pre-import")
+	fmt.Println("to import any remaining messages, users, and web push subscriptions.")
+	return nil
+}
+
 func createSchema(pgDB *sql.DB, cacheFile, authFile, webPushFile string) error {
 	fmt.Println("Creating initial PostgreSQL schema ...")
 	// User schema must be created before message schema, because message_stats and
@@ -645,16 +702,41 @@ func importUserPhones(sqlDB, pgDB *sql.DB) (int, error) {
 
 // Message import
 
-func importMessages(sqliteFile string, pgDB *sql.DB) error {
+const preImportTimeDelta = 30 // seconds to subtract from max time to account for in-flight messages
+
+// maxMessageTime returns the maximum message time in PostgreSQL minus a small buffer,
+// or 0 if there are no messages yet. This is used after a --pre-import run to only
+// import messages that arrived since the pre-import.
+func maxMessageTime(pgDB *sql.DB) int64 {
+	var maxTime sql.NullInt64
+	if err := pgDB.QueryRow(`SELECT MAX(time) FROM message`).Scan(&maxTime); err != nil || !maxTime.Valid || maxTime.Int64 == 0 {
+		return 0
+	}
+	sinceTime := maxTime.Int64 - preImportTimeDelta
+	if sinceTime < 0 {
+		return 0
+	}
+	fmt.Printf("Pre-imported messages detected (max time: %d), importing delta (since time %d) ...\n", maxTime.Int64, sinceTime)
+	return sinceTime
+}
+
+func importMessages(sqliteFile string, pgDB *sql.DB, sinceTime int64) error {
 	sqlDB, err := openSQLite(sqliteFile)
 	if err != nil {
 		fmt.Printf("Skipping message import: %s\n", err)
 		return nil
 	}
 	defer sqlDB.Close()
-	fmt.Printf("Importing messages from %s ...\n", sqliteFile)
 
-	rows, err := sqlDB.Query(`SELECT mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, attachment_deleted, sender, user, content_type, encoding, published FROM messages`)
+	query := `SELECT mid, sequence_id, time, event, expires, topic, message, title, priority, tags, click, icon, actions, attachment_name, attachment_type, attachment_size, attachment_expires, attachment_url, attachment_deleted, sender, user, content_type, encoding, published FROM messages`
+	var rows *sql.Rows
+	if sinceTime > 0 {
+		fmt.Printf("Importing messages from %s (since time %d) ...\n", sqliteFile, sinceTime)
+		rows, err = sqlDB.Query(query+` WHERE time >= ?`, sinceTime)
+	} else {
+		fmt.Printf("Importing messages from %s ...\n", sqliteFile)
+		rows, err = sqlDB.Query(query)
+	}
 	if err != nil {
 		return fmt.Errorf("querying messages: %w", err)
 	}
@@ -837,7 +919,9 @@ func importWebPush(sqliteFile string, pgDB *sql.DB) error {
 }
 
 func toUTF8(s string) string {
-	return strings.ToValidUTF8(s, "\uFFFD")
+	s = strings.ToValidUTF8(s, "\uFFFD")
+	s = strings.ReplaceAll(s, "\x00", "")
+	return s
 }
 
 // Verification

user/manager.go

@@ -642,7 +642,7 @@ func (a *Manager) AllowReservation(username string, topic string) error {
 // - Furthermore, the query prioritizes more specific permissions (longer!) over more generic ones, e.g. "test*" > "*"
 // - It also prioritizes write permissions over read permissions
 func (a *Manager) authorizeTopicAccess(usernameOrEveryone, topic string) (read, write, found bool, err error) {
-	rows, err := a.db.Query(a.queries.selectTopicPerms, Everyone, usernameOrEveryone, topic)
+	rows, err := a.db.ReadOnly().Query(a.queries.selectTopicPerms, Everyone, usernameOrEveryone, topic)
 	if err != nil {
 		return false, false, false, err
 	}

util/util.go

@@ -17,6 +17,7 @@ import (
 	"strings"
 	"sync"
 	"time"
+	"unicode/utf8"
 
 	"github.com/gabriel-vasile/mimetype"
 	"golang.org/x/term"
@@ -434,3 +435,22 @@ func Int(v int) *int {
 func Time(v time.Time) *time.Time {
 	return &v
 }
+
+// SanitizeUTF8 ensures a string is safe to store in PostgreSQL by handling two cases:
+//
+//  1. Invalid UTF-8 sequences: Some clients send Latin-1/ISO-8859-1 encoded text (e.g. accented
+//     characters like é, ñ, ß) in HTTP headers or SMTP messages. Go treats these as raw bytes in
+//     strings, but PostgreSQL rejects them. Any invalid UTF-8 byte is replaced with the Unicode
+//     replacement character (U+FFFD, "<22>") so the message is still delivered rather than lost.
+//
+//  2. NUL bytes (0x00): These are valid in UTF-8 but PostgreSQL TEXT columns reject them.
+//     They are stripped entirely.
+func SanitizeUTF8(s string) string {
+	if !utf8.ValidString(s) {
+		s = strings.ToValidUTF8(s, "\xef\xbf\xbd") // U+FFFD
+	}
+	if strings.ContainsRune(s, 0) {
+		s = strings.ReplaceAll(s, "\x00", "")
+	}
+	return s
+}