Merge branch 'binwiederhier:main' into main

server/errors.go

@@ -142,6 +142,7 @@ var (
 	errHTTPBadRequestTemplateFileNotFound            = &errHTTP{40047, http.StatusBadRequest, "invalid request: template file not found", "https://ntfy.sh/docs/publish/#message-templating", nil}
 	errHTTPBadRequestTemplateFileInvalid             = &errHTTP{40048, http.StatusBadRequest, "invalid request: template file invalid", "https://ntfy.sh/docs/publish/#message-templating", nil}
 	errHTTPBadRequestSequenceIDInvalid               = &errHTTP{40049, http.StatusBadRequest, "invalid request: sequence ID invalid", "https://ntfy.sh/docs/publish/#updating-deleting-notifications", nil}
+	errHTTPBadRequestEmailAddressInvalid             = &errHTTP{40050, http.StatusBadRequest, "invalid request: invalid e-mail address", "https://ntfy.sh/docs/publish/#e-mail-notifications", nil}
 	errHTTPNotFound                                  = &errHTTP{40401, http.StatusNotFound, "page not found", "", nil}
 	errHTTPUnauthorized                              = &errHTTP{40101, http.StatusUnauthorized, "unauthorized", "https://ntfy.sh/docs/publish/#authentication", nil}
 	errHTTPForbidden                                 = &errHTTP{40301, http.StatusForbidden, "forbidden", "https://ntfy.sh/docs/publish/#authentication", nil}

server/file_cache.go

@@ -1,128 +0,0 @@
-package server
-
-import (
-	"errors"
-	"fmt"
-	"heckel.io/ntfy/v2/log"
-	"heckel.io/ntfy/v2/model"
-	"heckel.io/ntfy/v2/util"
-	"io"
-	"os"
-	"path/filepath"
-	"regexp"
-	"sync"
-)
-
-var (
-	fileIDRegex      = regexp.MustCompile(fmt.Sprintf(`^[-_A-Za-z0-9]{%d}$`, model.MessageIDLength))
-	errInvalidFileID = errors.New("invalid file ID")
-	errFileExists    = errors.New("file exists")
-)
-
-type fileCache struct {
-	dir              string
-	totalSizeCurrent int64
-	totalSizeLimit   int64
-	mu               sync.Mutex
-}
-
-func newFileCache(dir string, totalSizeLimit int64) (*fileCache, error) {
-	if err := os.MkdirAll(dir, 0700); err != nil {
-		return nil, err
-	}
-	size, err := dirSize(dir)
-	if err != nil {
-		return nil, err
-	}
-	return &fileCache{
-		dir:              dir,
-		totalSizeCurrent: size,
-		totalSizeLimit:   totalSizeLimit,
-	}, nil
-}
-
-func (c *fileCache) Write(id string, in io.Reader, limiters ...util.Limiter) (int64, error) {
-	if !fileIDRegex.MatchString(id) {
-		return 0, errInvalidFileID
-	}
-	log.Tag(tagFileCache).Field("message_id", id).Debug("Writing attachment")
-	file := filepath.Join(c.dir, id)
-	if _, err := os.Stat(file); err == nil {
-		return 0, errFileExists
-	}
-	f, err := os.OpenFile(file, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0600)
-	if err != nil {
-		return 0, err
-	}
-	defer f.Close()
-	limiters = append(limiters, util.NewFixedLimiter(c.Remaining()))
-	limitWriter := util.NewLimitWriter(f, limiters...)
-	size, err := io.Copy(limitWriter, in)
-	if err != nil {
-		os.Remove(file)
-		return 0, err
-	}
-	if err := f.Close(); err != nil {
-		os.Remove(file)
-		return 0, err
-	}
-	c.mu.Lock()
-	c.totalSizeCurrent += size
-	mset(metricAttachmentsTotalSize, c.totalSizeCurrent)
-	c.mu.Unlock()
-	return size, nil
-}
-
-func (c *fileCache) Remove(ids ...string) error {
-	for _, id := range ids {
-		if !fileIDRegex.MatchString(id) {
-			return errInvalidFileID
-		}
-		log.Tag(tagFileCache).Field("message_id", id).Debug("Deleting attachment")
-		file := filepath.Join(c.dir, id)
-		if err := os.Remove(file); err != nil {
-			log.Tag(tagFileCache).Field("message_id", id).Err(err).Debug("Error deleting attachment")
-		}
-	}
-	size, err := dirSize(c.dir)
-	if err != nil {
-		return err
-	}
-	c.mu.Lock()
-	c.totalSizeCurrent = size
-	c.mu.Unlock()
-	mset(metricAttachmentsTotalSize, size)
-	return nil
-}
-
-func (c *fileCache) Size() int64 {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	return c.totalSizeCurrent
-}
-
-func (c *fileCache) Remaining() int64 {
-	c.mu.Lock()
-	defer c.mu.Unlock()
-	remaining := c.totalSizeLimit - c.totalSizeCurrent
-	if remaining < 0 {
-		return 0
-	}
-	return remaining
-}
-
-func dirSize(dir string) (int64, error) {
-	entries, err := os.ReadDir(dir)
-	if err != nil {
-		return 0, err
-	}
-	var size int64
-	for _, e := range entries {
-		info, err := e.Info()
-		if err != nil {
-			return 0, err
-		}
-		size += info.Size()
-	}
-	return size, nil
-}

server/file_cache_test.go

@@ -1,76 +0,0 @@
-package server
-
-import (
-	"bytes"
-	"fmt"
-	"github.com/stretchr/testify/require"
-	"heckel.io/ntfy/v2/util"
-	"os"
-	"strings"
-	"testing"
-)
-
-var (
-	oneKilobyteArray = make([]byte, 1024)
-)
-
-func TestFileCache_Write_Success(t *testing.T) {
-	dir, c := newTestFileCache(t)
-	size, err := c.Write("abcdefghijkl", strings.NewReader("normal file"), util.NewFixedLimiter(999))
-	require.Nil(t, err)
-	require.Equal(t, int64(11), size)
-	require.Equal(t, "normal file", readFile(t, dir+"/abcdefghijkl"))
-	require.Equal(t, int64(11), c.Size())
-	require.Equal(t, int64(10229), c.Remaining())
-}
-
-func TestFileCache_Write_Remove_Success(t *testing.T) {
-	dir, c := newTestFileCache(t) // max = 10k (10240), each = 1k (1024)
-	for i := 0; i < 10; i++ {     // 10x999 = 9990
-		size, err := c.Write(fmt.Sprintf("abcdefghijk%d", i), bytes.NewReader(make([]byte, 999)))
-		require.Nil(t, err)
-		require.Equal(t, int64(999), size)
-	}
-	require.Equal(t, int64(9990), c.Size())
-	require.Equal(t, int64(250), c.Remaining())
-	require.FileExists(t, dir+"/abcdefghijk1")
-	require.FileExists(t, dir+"/abcdefghijk5")
-
-	require.Nil(t, c.Remove("abcdefghijk1", "abcdefghijk5"))
-	require.NoFileExists(t, dir+"/abcdefghijk1")
-	require.NoFileExists(t, dir+"/abcdefghijk5")
-	require.Equal(t, int64(7992), c.Size())
-	require.Equal(t, int64(2248), c.Remaining())
-}
-
-func TestFileCache_Write_FailedTotalSizeLimit(t *testing.T) {
-	dir, c := newTestFileCache(t)
-	for i := 0; i < 10; i++ {
-		size, err := c.Write(fmt.Sprintf("abcdefghijk%d", i), bytes.NewReader(oneKilobyteArray))
-		require.Nil(t, err)
-		require.Equal(t, int64(1024), size)
-	}
-	_, err := c.Write("abcdefghijkX", bytes.NewReader(oneKilobyteArray))
-	require.Equal(t, util.ErrLimitReached, err)
-	require.NoFileExists(t, dir+"/abcdefghijkX")
-}
-
-func TestFileCache_Write_FailedAdditionalLimiter(t *testing.T) {
-	dir, c := newTestFileCache(t)
-	_, err := c.Write("abcdefghijkl", bytes.NewReader(make([]byte, 1001)), util.NewFixedLimiter(1000))
-	require.Equal(t, util.ErrLimitReached, err)
-	require.NoFileExists(t, dir+"/abcdefghijkl")
-}
-
-func newTestFileCache(t *testing.T) (dir string, cache *fileCache) {
-	dir = t.TempDir()
-	cache, err := newFileCache(dir, 10*1024)
-	require.Nil(t, err)
-	return dir, cache
-}
-
-func readFile(t *testing.T, f string) string {
-	b, err := os.ReadFile(f)
-	require.Nil(t, err)
-	return string(b)
-}

server/log.go

@@ -24,7 +24,6 @@ const (
 	tagSMTP         = "smtp"  // Receive email
 	tagEmail        = "email" // Send email
 	tagTwilio       = "twilio"
-	tagFileCache    = "file_cache"
 	tagMessageCache = "message_cache"
 	tagStripe       = "stripe"
 	tagAccount      = "account"
@@ -36,7 +35,7 @@ const (
 )
 
 var (
-	normalErrorCodes       = []int{http.StatusNotFound, http.StatusBadRequest, http.StatusTooManyRequests, http.StatusUnauthorized, http.StatusForbidden, http.StatusInsufficientStorage}
+	normalErrorCodes       = []int{http.StatusNotFound, http.StatusBadRequest, http.StatusTooManyRequests, http.StatusUnauthorized, http.StatusForbidden, http.StatusInsufficientStorage, http.StatusRequestEntityTooLarge}
 	rateLimitingErrorCodes = []int{http.StatusTooManyRequests, http.StatusRequestEntityTooLarge}
 )
 

server/server.go

@@ -32,6 +32,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"golang.org/x/sync/errgroup"
 	"gopkg.in/yaml.v2"
+	"heckel.io/ntfy/v2/attachment"
 	"heckel.io/ntfy/v2/db"
 	"heckel.io/ntfy/v2/db/pg"
 	"heckel.io/ntfy/v2/log"
@@ -64,7 +65,7 @@ type Server struct {
 	userManager       *user.Manager                       // Might be nil!
 	messageCache      *message.Cache                      // Database that stores the messages
 	webPush           *webpush.Store                      // Database that stores web push subscriptions
-	fileCache         *fileCache                          // File system based cache that stores attachments
+	attachment        *attachment.Store                   // Attachment store (file system or S3)
 	stripe            stripeAPI                           // Stripe API, can be replaced with a mock
 	priceCache        *util.LookupCache[map[string]int64] // Stripe price ID -> price as cents (USD implied!)
 	metricsHandler    http.Handler                        // Handles /metrics if enable-metrics set, and listen-metrics-http not set
@@ -122,6 +123,7 @@ var (
 	fileRegex                                            = regexp.MustCompile(`^/file/([-_A-Za-z0-9]{1,64})(?:\.[A-Za-z0-9]{1,16})?$`)
 	urlRegex                                             = regexp.MustCompile(`^https?://`)
 	phoneNumberRegex                                     = regexp.MustCompile(`^\+\d{1,100}$`)
+	emailAddressRegex                                    = regexp.MustCompile(`^[^\s,;]+@[^\s,;]+$`)
 
 	//go:embed site
 	webFs       embed.FS
@@ -227,12 +229,9 @@ func New(conf *Config) (*Server, error) {
 	if err != nil {
 		return nil, err
 	}
-	var fileCache *fileCache
-	if conf.AttachmentCacheDir != "" {
-		fileCache, err = newFileCache(conf.AttachmentCacheDir, conf.AttachmentTotalSizeLimit)
-		if err != nil {
-			return nil, err
-		}
+	attachmentStore, err := createAttachmentStore(conf, messageCache)
+	if err != nil {
+		return nil, err
 	}
 	var userManager *user.Manager
 	if conf.AuthFile != "" || pool != nil {
@@ -276,7 +275,7 @@ func New(conf *Config) (*Server, error) {
 		db:              pool,
 		messageCache:    messageCache,
 		webPush:         wp,
-		fileCache:       fileCache,
+		attachment:      attachmentStore,
 		firebaseClient:  firebaseClient,
 		smtpSender:      mailer,
 		topics:          topics,
@@ -301,6 +300,15 @@ func createMessageCache(conf *Config, pool *db.DB) (*message.Cache, error) {
 	return message.NewMemStore()
 }
 
+func createAttachmentStore(conf *Config, messageCache *message.Cache) (*attachment.Store, error) {
+	if strings.HasPrefix(conf.AttachmentCacheDir, "s3://") {
+		return attachment.NewS3Store(conf.AttachmentCacheDir, conf.AttachmentTotalSizeLimit, messageCache.AttachmentsWithSizes)
+	} else if conf.AttachmentCacheDir != "" {
+		return attachment.NewFileStore(conf.AttachmentCacheDir, conf.AttachmentTotalSizeLimit, messageCache.AttachmentsWithSizes)
+	}
+	return nil, nil
+}
+
 // Run executes the main server. It listens on HTTP (+ HTTPS, if configured), and starts
 // a manager go routine to print stats and prune messages.
 func (s *Server) Run() error {
@@ -421,6 +429,9 @@ func (s *Server) Stop() {
 	if s.smtpServer != nil {
 		s.smtpServer.Close()
 	}
+	if s.attachment != nil {
+		s.attachment.Close()
+	}
 	s.closeDatabases()
 	close(s.closeChan)
 }
@@ -595,7 +606,7 @@ func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request, v *visit
 		return s.ensureWebEnabled(s.handleStatic)(w, r, v)
 	} else if r.Method == http.MethodGet && docsRegex.MatchString(r.URL.Path) {
 		return s.ensureWebEnabled(s.handleDocs)(w, r, v)
-	} else if (r.Method == http.MethodGet || r.Method == http.MethodHead) && fileRegex.MatchString(r.URL.Path) && s.config.AttachmentCacheDir != "" {
+	} else if (r.Method == http.MethodGet || r.Method == http.MethodHead) && fileRegex.MatchString(r.URL.Path) && s.attachment != nil {
 		return s.limitRequests(s.handleFile)(w, r, v)
 	} else if r.Method == http.MethodOptions {
 		return s.limitRequests(s.handleOptions)(w, r, v) // Should work even if the web app is not enabled, see #598
@@ -752,7 +763,7 @@ func (s *Server) handleStats(w http.ResponseWriter, _ *http.Request, _ *visitor)
 // Before streaming the file to a client, it locates uploader (m.Sender or m.User) in the message cache, so it
 // can associate the download bandwidth with the uploader.
 func (s *Server) handleFile(w http.ResponseWriter, r *http.Request, v *visitor) error {
-	if s.config.AttachmentCacheDir == "" {
+	if s.attachment == nil {
 		return errHTTPInternalError
 	}
 	matches := fileRegex.FindStringSubmatch(r.URL.Path)
@@ -760,16 +771,16 @@ func (s *Server) handleFile(w http.ResponseWriter, r *http.Request, v *visitor)
 		return errHTTPInternalErrorInvalidPath
 	}
 	messageID := matches[1]
-	file := filepath.Join(s.config.AttachmentCacheDir, messageID)
-	stat, err := os.Stat(file)
+	reader, size, err := s.attachment.Read(messageID)
 	if err != nil {
 		return errHTTPNotFound.Fields(log.Context{
 			"message_id":    messageID,
-			"error_context": "filesystem",
+			"error_context": "attachment_store",
 		})
 	}
+	defer reader.Close()
 	w.Header().Set("Access-Control-Allow-Origin", s.config.AccessControlAllowOrigin) // CORS, allow cross-origin requests
-	w.Header().Set("Content-Length", fmt.Sprintf("%d", stat.Size()))
+	w.Header().Set("Content-Length", fmt.Sprintf("%d", size))
 	if r.Method == http.MethodHead {
 		return nil
 	}
@@ -805,19 +816,14 @@ func (s *Server) handleFile(w http.ResponseWriter, r *http.Request, v *visitor)
 	} else if m.Sender.IsValid() {
 		bandwidthVisitor = s.visitor(m.Sender, nil)
 	}
-	if !bandwidthVisitor.BandwidthAllowed(stat.Size()) {
+	if !bandwidthVisitor.BandwidthAllowed(size) {
 		return errHTTPTooManyRequestsLimitAttachmentBandwidth.With(m)
 	}
 	// Actually send file
-	f, err := os.Open(file)
-	if err != nil {
-		return err
-	}
-	defer f.Close()
 	if m.Attachment.Name != "" {
 		w.Header().Set("Content-Disposition", "attachment; filename="+strconv.Quote(m.Attachment.Name))
 	}
-	_, err = io.Copy(util.NewContentTypeWriter(w, r.URL.Path), f)
+	_, err = io.Copy(util.NewContentTypeWriter(w, r.URL.Path), reader)
 	return err
 }
 
@@ -926,8 +932,8 @@ func (s *Server) handlePublishInternal(r *http.Request, v *visitor) (*model.Mess
 			return nil, err
 		}
 		// Delete attachment files for deleted scheduled messages
-		if s.fileCache != nil && len(deletedIDs) > 0 {
-			if err := s.fileCache.Remove(deletedIDs...); err != nil {
+		if s.attachment != nil && len(deletedIDs) > 0 {
+			if err := s.attachment.Remove(deletedIDs...); err != nil {
 				logvrm(v, r, m).Tag(tagPublish).Err(err).Warn("Error removing attachments for deleted scheduled messages")
 			}
 		}
@@ -1033,8 +1039,8 @@ func (s *Server) handleActionMessage(w http.ResponseWriter, r *http.Request, v *
 			return err
 		}
 		// Delete attachment files for deleted scheduled messages
-		if s.fileCache != nil && len(deletedIDs) > 0 {
-			if err := s.fileCache.Remove(deletedIDs...); err != nil {
+		if s.attachment != nil && len(deletedIDs) > 0 {
+			if err := s.attachment.Remove(deletedIDs...); err != nil {
 				logvrm(v, r, m).Tag(tagPublish).Err(err).Warn("Error removing attachments for deleted scheduled messages")
 			}
 		}
@@ -1163,6 +1169,9 @@ func (s *Server) parsePublishParams(r *http.Request, m *model.Message) (cache bo
 		m.Icon = icon
 	}
 	email = readParam(r, "x-email", "x-e-mail", "email", "e-mail", "mail", "e")
+	if email != "" && !emailAddressRegex.MatchString(email) {
+		return false, false, "", "", "", false, "", errHTTPBadRequestEmailAddressInvalid
+	}
 	if s.smtpSender == nil && email != "" {
 		return false, false, "", "", "", false, "", errHTTPBadRequestEmailDisabled
 	}
@@ -1409,7 +1418,7 @@ func (s *Server) renderTemplate(name, tpl, source string) (string, error) {
 }
 
 func (s *Server) handleBodyAsAttachment(r *http.Request, v *visitor, m *model.Message, body *util.PeekedReadCloser) error {
-	if s.fileCache == nil || s.config.BaseURL == "" || s.config.AttachmentCacheDir == "" {
+	if s.attachment == nil || s.config.BaseURL == "" {
 		return errHTTPBadRequestAttachmentsDisallowed.With(m)
 	}
 	vinfo, err := v.Info()
@@ -1420,16 +1429,13 @@ func (s *Server) handleBodyAsAttachment(r *http.Request, v *visitor, m *model.Me
 	if m.Time > attachmentExpiry {
 		return errHTTPBadRequestAttachmentsExpiryBeforeDelivery.With(m)
 	}
-	contentLengthStr := r.Header.Get("Content-Length")
-	if contentLengthStr != "" { // Early "do-not-trust" check, hard limit see below
-		contentLength, err := strconv.ParseInt(contentLengthStr, 10, 64)
-		if err == nil && (contentLength > vinfo.Stats.AttachmentTotalSizeRemaining || contentLength > vinfo.Limits.AttachmentFileSizeLimit) {
-			return errHTTPEntityTooLargeAttachment.With(m).Fields(log.Context{
-				"message_content_length":          contentLength,
-				"attachment_total_size_remaining": vinfo.Stats.AttachmentTotalSizeRemaining,
-				"attachment_file_size_limit":      vinfo.Limits.AttachmentFileSizeLimit,
-			})
-		}
+	// Early "do-not-trust" check, hard limit see below
+	if r.ContentLength > 0 && (r.ContentLength > vinfo.Stats.AttachmentTotalSizeRemaining || r.ContentLength > vinfo.Limits.AttachmentFileSizeLimit) {
+		return errHTTPEntityTooLargeAttachment.With(m).Fields(log.Context{
+			"message_content_length":          r.ContentLength,
+			"attachment_total_size_remaining": vinfo.Stats.AttachmentTotalSizeRemaining,
+			"attachment_file_size_limit":      vinfo.Limits.AttachmentFileSizeLimit,
+		})
 	}
 	if m.Attachment == nil {
 		m.Attachment = &model.Attachment{}
@@ -1449,7 +1455,7 @@ func (s *Server) handleBodyAsAttachment(r *http.Request, v *visitor, m *model.Me
 		util.NewFixedLimiter(vinfo.Limits.AttachmentFileSizeLimit),
 		util.NewFixedLimiter(vinfo.Stats.AttachmentTotalSizeRemaining),
 	}
-	m.Attachment.Size, err = s.fileCache.Write(m.ID, body, limiters...)
+	m.Attachment.Size, err = s.attachment.Write(m.ID, body, r.ContentLength, limiters...)
 	if errors.Is(err, util.ErrLimitReached) {
 		return errHTTPEntityTooLargeAttachment.With(m)
 	} else if err != nil {

server/server.yml

@@ -153,7 +153,8 @@
 # If enabled, clients can attach files to notifications as attachments. Minimum settings to enable attachments
 # are "attachment-cache-dir" and "base-url".
 #
-# - attachment-cache-dir is the cache directory for attached files
+# - attachment-cache-dir is the cache directory for attached files, or an S3 URL for object storage
+#   e.g. /var/cache/ntfy/attachments, or s3://ACCESS_KEY:SECRET_KEY@bucket/prefix?region=us-east-1&endpoint=https://...
 # - attachment-total-size-limit is the limit of the on-disk attachment cache directory (total size)
 # - attachment-file-size-limit is the per-file attachment size limit (e.g. 300k, 2M, 100M)
 # - attachment-expiry-duration is the duration after which uploaded attachments will be deleted (e.g. 3h, 20h)

server/server_account.go

@@ -3,14 +3,15 @@ package server
 import (
 	"encoding/json"
 	"errors"
-	"heckel.io/ntfy/v2/log"
-	"heckel.io/ntfy/v2/model"
-	"heckel.io/ntfy/v2/user"
-	"heckel.io/ntfy/v2/util"
 	"net/http"
 	"net/netip"
 	"strings"
 	"time"
+
+	"heckel.io/ntfy/v2/log"
+	"heckel.io/ntfy/v2/model"
+	"heckel.io/ntfy/v2/user"
+	"heckel.io/ntfy/v2/util"
 )
 
 const (
@@ -455,21 +456,8 @@ func (s *Server) handleAccountReservationAdd(w http.ResponseWriter, r *http.Requ
 		return errHTTPUnauthorized
 	} else if err := s.userManager.AllowReservation(u.Name, req.Topic); err != nil {
 		return errHTTPConflictTopicReserved
-	} else if u.IsUser() {
-		hasReservation, err := s.userManager.HasReservation(u.Name, req.Topic)
-		if err != nil {
-			return err
-		}
-		if !hasReservation {
-			reservations, err := s.userManager.ReservationsCount(u.Name)
-			if err != nil {
-				return err
-			} else if reservations >= u.Tier.ReservationLimit {
-				return errHTTPTooManyRequestsLimitReservations
-			}
-		}
 	}
-	// Actually add the reservation
+	// Actually add the reservation (with limit check inside the transaction to avoid races)
 	logvr(v, r).
 		Tag(tagAccount).
 		Fields(log.Context{
@@ -477,7 +465,14 @@ func (s *Server) handleAccountReservationAdd(w http.ResponseWriter, r *http.Requ
 			"everyone": everyone.String(),
 		}).
 		Debug("Adding topic reservation")
-	if err := s.userManager.AddReservation(u.Name, req.Topic, everyone); err != nil {
+	var limit int64
+	if u.IsUser() && u.Tier != nil {
+		limit = u.Tier.ReservationLimit
+	}
+	if err := s.userManager.AddReservation(u.Name, req.Topic, everyone, limit); err != nil {
+		if errors.Is(err, user.ErrTooManyReservations) {
+			return errHTTPTooManyRequestsLimitReservations
+		}
 		return err
 	}
 	// Kill existing subscribers
@@ -530,22 +525,15 @@ func (s *Server) handleAccountReservationDelete(w http.ResponseWriter, r *http.R
 // and marks associated messages for the topics as deleted. This also eventually deletes attachments.
 // The process relies on the manager to perform the actual deletions (see runManager).
 func (s *Server) maybeRemoveMessagesAndExcessReservations(r *http.Request, v *visitor, u *user.User, reservationsLimit int64) error {
-	reservations, err := s.userManager.Reservations(u.Name)
+	removedTopics, err := s.userManager.RemoveExcessReservations(u.Name, reservationsLimit)
 	if err != nil {
 		return err
-	} else if int64(len(reservations)) <= reservationsLimit {
+	} else if len(removedTopics) == 0 {
 		logvr(v, r).Tag(tagAccount).Debug("No excess reservations to remove")
 		return nil
 	}
-	topics := make([]string, 0)
-	for i := int64(len(reservations)) - 1; i >= reservationsLimit; i-- {
-		topics = append(topics, reservations[i].Topic)
-	}
-	logvr(v, r).Tag(tagAccount).Info("Removing excess reservations for topics %s", strings.Join(topics, ", "))
-	if err := s.userManager.RemoveReservations(u.Name, topics...); err != nil {
-		return err
-	}
-	if err := s.messageCache.ExpireMessages(topics...); err != nil {
+	logvr(v, r).Tag(tagAccount).Info("Removed excess topic reservations, now removing messages for topics %s", strings.Join(removedTopics, ", "))
+	if err := s.messageCache.ExpireMessages(removedTopics...); err != nil {
 		return err
 	}
 	go s.pruneMessages()

server/server_account_test.go

@@ -503,7 +503,7 @@ func TestAccount_Reservation_AddAdminSuccess(t *testing.T) {
 		}))
 		require.Nil(t, s.userManager.AddUser("noadmin1", "pass", user.RoleUser, false))
 		require.Nil(t, s.userManager.ChangeTier("noadmin1", "pro"))
-		require.Nil(t, s.userManager.AddReservation("noadmin1", "mytopic", user.PermissionDenyAll))
+		require.Nil(t, s.userManager.AddReservation("noadmin1", "mytopic", user.PermissionDenyAll, 0))
 
 		require.Nil(t, s.userManager.AddUser("noadmin2", "pass", user.RoleUser, false))
 		require.Nil(t, s.userManager.ChangeTier("noadmin2", "pro"))

server/server_manager.go

@@ -99,6 +99,9 @@ func (s *Server) execManager() {
 	mset(metricUsers, usersCount)
 	mset(metricSubscribers, subscribers)
 	mset(metricTopics, topicsCount)
+	if s.attachment != nil {
+		mset(metricAttachmentsTotalSize, s.attachment.Size())
+	}
 }
 
 func (s *Server) pruneVisitors() {
@@ -137,7 +140,7 @@ func (s *Server) pruneTokens() {
 }
 
 func (s *Server) pruneAttachments() {
-	if s.fileCache == nil {
+	if s.attachment == nil {
 		return
 	}
 	log.
@@ -150,7 +153,7 @@ func (s *Server) pruneAttachments() {
 				if log.Tag(tagManager).IsDebug() {
 					log.Tag(tagManager).Debug("Deleting attachments %s", strings.Join(ids, ", "))
 				}
-				if err := s.fileCache.Remove(ids...); err != nil {
+				if err := s.attachment.Remove(ids...); err != nil {
 					log.Tag(tagManager).Err(err).Warn("Error deleting attachments")
 				}
 				if err := s.messageCache.MarkAttachmentsDeleted(ids...); err != nil {
@@ -171,8 +174,8 @@ func (s *Server) pruneMessages() {
 			if err != nil {
 				log.Tag(tagManager).Err(err).Warn("Error retrieving expired messages")
 			} else if len(expiredMessageIDs) > 0 {
-				if s.fileCache != nil {
-					if err := s.fileCache.Remove(expiredMessageIDs...); err != nil {
+				if s.attachment != nil {
+					if err := s.attachment.Remove(expiredMessageIDs...); err != nil {
 						log.Tag(tagManager).Err(err).Warn("Error deleting attachments for expired messages")
 					}
 				}

server/server_payments_test.go

@@ -478,8 +478,8 @@ func TestPayments_Webhook_Subscription_Updated_Downgrade_From_PastDue_To_Active(
 		}))
 		require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleUser, false))
 		require.Nil(t, s.userManager.ChangeTier("phil", "pro"))
-		require.Nil(t, s.userManager.AddReservation("phil", "atopic", user.PermissionDenyAll))
-		require.Nil(t, s.userManager.AddReservation("phil", "ztopic", user.PermissionDenyAll))
+		require.Nil(t, s.userManager.AddReservation("phil", "atopic", user.PermissionDenyAll, 0))
+		require.Nil(t, s.userManager.AddReservation("phil", "ztopic", user.PermissionDenyAll, 0))
 
 		// Add billing details
 		u, err := s.userManager.User("phil")
@@ -589,7 +589,7 @@ func TestPayments_Webhook_Subscription_Deleted(t *testing.T) {
 		}))
 		require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleUser, false))
 		require.Nil(t, s.userManager.ChangeTier("phil", "pro"))
-		require.Nil(t, s.userManager.AddReservation("phil", "atopic", user.PermissionDenyAll))
+		require.Nil(t, s.userManager.AddReservation("phil", "atopic", user.PermissionDenyAll, 0))
 
 		// Add billing details
 		u, err := s.userManager.User("phil")

server/server_test.go

@@ -1543,6 +1543,30 @@ func TestServer_PublishEmailNoMailer_Fail(t *testing.T) {
 	})
 }
 
+func TestServer_PublishEmailAddressInvalid(t *testing.T) {
+	forEachBackend(t, func(t *testing.T, databaseURL string) {
+		s := newTestServer(t, newTestConfig(t, databaseURL))
+		s.smtpSender = &testMailer{}
+		addresses := []string{
+			"test@example.com, other@example.com",
+			"invalidaddress",
+			"@nope",
+			"nope@",
+		}
+		for _, email := range addresses {
+			response := request(t, s, "PUT", "/mytopic", "fail", map[string]string{
+				"E-Mail": email,
+			})
+			require.Equal(t, 400, response.Code, "expected 400 for email: %s", email)
+		}
+		// Valid address should succeed
+		response := request(t, s, "PUT", "/mytopic", "success", map[string]string{
+			"E-Mail": "test@example.com",
+		})
+		require.Equal(t, 200, response.Code)
+	})
+}
+
 func TestServer_PublishAndExpungeTopicAfter16Hours(t *testing.T) {
 	forEachBackend(t, func(t *testing.T, databaseURL string) {
 		t.Parallel()
@@ -2121,7 +2145,7 @@ func TestServer_PublishAttachmentShortWithFilename(t *testing.T) {
 		require.Equal(t, "myfile.txt", msg.Attachment.Name)
 		require.Equal(t, "text/plain; charset=utf-8", msg.Attachment.Type)
 		require.Equal(t, int64(21), msg.Attachment.Size)
-		require.GreaterOrEqual(t, msg.Attachment.Expires, time.Now().Add(3*time.Hour).Unix())
+		require.GreaterOrEqual(t, msg.Attachment.Expires, time.Now().Add(3*time.Hour).Unix()-1)
 		require.Contains(t, msg.Attachment.URL, "http://127.0.0.1:12345/file/")
 		require.Equal(t, netip.Addr{}, msg.Sender) // Should never be returned
 		require.FileExists(t, filepath.Join(s.config.AttachmentCacheDir, msg.ID))
@@ -2194,8 +2218,8 @@ func TestServer_PublishAttachmentTooLargeContentLength(t *testing.T) {
 	forEachBackend(t, func(t *testing.T, databaseURL string) {
 		content := util.RandomString(5000) // > 4096
 		s := newTestServer(t, newTestConfig(t, databaseURL))
-		response := request(t, s, "PUT", "/mytopic", content, map[string]string{
-			"Content-Length": "20000000",
+		response := request(t, s, "PUT", "/mytopic", content, nil, func(r *http.Request) {
+			r.ContentLength = 20000000
 		})
 		err := toHTTPError(t, response.Body.String())
 		require.Equal(t, 413, response.Code)

server/server_webpush_test.go

@@ -235,13 +235,12 @@ func TestServer_WebPush_Publish_RemoveOnError(t *testing.T) {
 
 		request(t, s, "POST", "/test-topic", "web push test", nil)
 
-		waitFor(t, func() bool {
-			return received.Load()
-		})
-
 		// Receiving the 410 should've caused the publisher to expire all subscriptions on the endpoint
-
-		requireSubscriptionCount(t, s, "test-topic", 0)
+		waitFor(t, func() bool {
+			subs, err := s.webPush.SubscriptionsForTopic("test-topic")
+			require.Nil(t, err)
+			return len(subs) == 0
+		})
 		requireSubscriptionCount(t, s, "test-topic-abc", 0)
 	})
 }