Fix delayed message sending from authenticated users, closes #679

docs/releases.md

@@ -1135,3 +1135,9 @@ and the [ntfy Android app](https://github.com/binwiederhier/ntfy-android/release
 **Additional languages:**
 
 * Swedish (thanks to [@hellbown](https://hosted.weblate.org/user/hellbown/))
+
+## ntfy server v2.3.0 (UNRELEASED)
+
+**Bug fixes + maintenance:**
+
+* Fix delayed message sending from authenticated users ([#679](https://github.com/binwiederhier/ntfy/issues/679))

server/server.go

@@ -1541,7 +1541,7 @@ func (s *Server) sendDelayedMessages() error {
 	for _, m := range messages {
 		var u *user.User
 		if s.userManager != nil && m.User != "" {
-			u, err = s.userManager.User(m.User)
+			u, err = s.userManager.UserByID(m.User)
 			if err != nil {
 				log.With(m).Err(err).Warn("Error sending delayed message")
 				continue

server/server_test.go

@@ -327,13 +327,10 @@ func TestServer_PublishNoCache(t *testing.T) {
 
 func TestServer_PublishAt(t *testing.T) {
 	t.Parallel()
-	c := newTestConfig(t)
+	s := newTestServer(t, newTestConfig(t))
-	c.MinDelay = time.Second
-	c.DelayedSenderInterval = 100 * time.Millisecond
-	s := newTestServer(t, c)
 
 	response := request(t, s, "PUT", "/mytopic", "a message", map[string]string{
-		"In": "1s",
+		"In": "1h",
 	})
 	require.Equal(t, 200, response.Code)
 
@@ -341,22 +338,62 @@ func TestServer_PublishAt(t *testing.T) {
 	messages := toMessages(t, response.Body.String())
 	require.Equal(t, 0, len(messages))
 
-	time.Sleep(time.Second)
+	// Update message time to the past
-	require.Nil(t, s.sendDelayedMessages())
+	fakeTime := time.Now().Add(-10 * time.Second).Unix()
+	_, err := s.messageCache.db.Exec(`UPDATE messages SET time=?`, fakeTime)
+	require.Nil(t, err)
 
+	// Trigger delayed message sending
+	require.Nil(t, s.sendDelayedMessages())
 	response = request(t, s, "GET", "/mytopic/json?poll=1", "", nil)
 	messages = toMessages(t, response.Body.String())
 	require.Equal(t, 1, len(messages))
 	require.Equal(t, "a message", messages[0].Message)
 	require.Equal(t, netip.Addr{}, messages[0].Sender) // Never return the sender!
 
-	messages, err := s.messageCache.Messages("mytopic", sinceAllMessages, true)
+	messages, err = s.messageCache.Messages("mytopic", sinceAllMessages, true)
 	require.Nil(t, err)
 	require.Equal(t, 1, len(messages))
 	require.Equal(t, "a message", messages[0].Message)
 	require.Equal(t, "9.9.9.9", messages[0].Sender.String()) // It's stored in the DB though!
 }
 
+func TestServer_PublishAt_FromUser(t *testing.T) {
+	t.Parallel()
+	s := newTestServer(t, newTestConfigWithAuthFile(t))
+
+	require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleAdmin))
+	response := request(t, s, "PUT", "/mytopic", "a message", map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+		"In":            "1h",
+	})
+	require.Equal(t, 200, response.Code)
+
+	// Message doesn't show up immediately
+	response = request(t, s, "GET", "/mytopic/json?poll=1", "", nil)
+	messages := toMessages(t, response.Body.String())
+	require.Equal(t, 0, len(messages))
+
+	// Update message time to the past
+	fakeTime := time.Now().Add(-10 * time.Second).Unix()
+	_, err := s.messageCache.db.Exec(`UPDATE messages SET time=?`, fakeTime)
+	require.Nil(t, err)
+
+	// Trigger delayed message sending
+	require.Nil(t, s.sendDelayedMessages())
+	response = request(t, s, "GET", "/mytopic/json?poll=1", "", nil)
+	messages = toMessages(t, response.Body.String())
+	require.Equal(t, 1, len(messages))
+	require.Equal(t, fakeTime, messages[0].Time)
+	require.Equal(t, "a message", messages[0].Message)
+
+	messages, err = s.messageCache.Messages("mytopic", sinceAllMessages, true)
+	require.Nil(t, err)
+	require.Equal(t, 1, len(messages))
+	require.Equal(t, "a message", messages[0].Message)
+	require.True(t, strings.HasPrefix(messages[0].User, "u_"))
+}
+
 func TestServer_PublishAt_Expires(t *testing.T) {
 	s := newTestServer(t, newTestConfig(t))