Share daily limits through the usage table, drop the midnight reset and the stats queue

The server builds the usage store on the user database (the shared pool on
Postgres, the auth-file on SQLite) and starts the limiter before serving. A
visitor without a user writes its message count from a tenth of the cap on,
everyone else from the first message; anonymous emails get the daily cap the
replenish interval implies. The account API reports reset_at. The user stats
columns are no longer read or written.

Claude-Session: https://claude.ai/code/session_01DeXuhSjq5Gta5qPJgSceFe

server/config.go

@@ -116,9 +116,6 @@ const (
 )
 
 var (
-	// DefaultVisitorStatsResetTime defines the time at which visitor stats are reset (wall clock only)
-	DefaultVisitorStatsResetTime = time.Date(0, 0, 0, 0, 0, 0, 0, time.UTC)
-
 	// DefaultDisallowedTopics defines the topics that are forbidden, because they are used elsewhere. This array can be
 	// extended using the server.yml config. If updated, also update in Android and web app.
 	DefaultDisallowedTopics = []string{"docs", "static", "file", "app", "metrics", "account", "settings", "signup", "login", "v1"}
@@ -215,7 +212,6 @@ type Config struct {
 	VisitorAccountCreationLimitReplenish time.Duration
 	VisitorAuthFailureLimitBurst         int
 	VisitorAuthFailureLimitReplenish     time.Duration
-	VisitorStatsResetTime                time.Time      // Time of the day at which to reset visitor stats
 	VisitorSubscriberRateLimiting        bool           // Enable subscriber-based rate limiting for UnifiedPush topics
 	VisitorPrefixBitsIPv4                int            // Number of bits for IPv4 rate limiting (default: 32)
 	VisitorPrefixBitsIPv6                int            // Number of bits for IPv6 rate limiting (default: 64)
@@ -335,7 +331,6 @@ func NewConfig() *Config {
 		VisitorAccountCreationLimitReplenish: DefaultVisitorAccountCreationLimitReplenish,
 		VisitorAuthFailureLimitBurst:         DefaultVisitorAuthFailureLimitBurst,
 		VisitorAuthFailureLimitReplenish:     DefaultVisitorAuthFailureLimitReplenish,
-		VisitorStatsResetTime:                DefaultVisitorStatsResetTime,
 		VisitorPrefixBitsIPv4:                DefaultVisitorPrefixBitsIPv4, // Default: use full IPv4 address
 		VisitorPrefixBitsIPv6:                DefaultVisitorPrefixBitsIPv6, // Default: use /64 for IPv6
 		BehindProxy:                          false,                        // If true, the server will trust the proxy client IP header to determine the client IP address

server/log.go

@@ -27,7 +27,6 @@ const (
 	tagStripe    = "stripe"
 	tagAccount   = "account"
 	tagManager   = "manager"
-	tagResetter  = "resetter"
 	tagWebsocket = "websocket"
 	tagMatrix    = "matrix"
 	tagWebPush   = "webpush"

server/server.go

@@ -301,7 +301,6 @@ func New(conf *Config) (*Server, error) {
 		messages:        messages,
 		messagesHistory: []int64{messages},
 		messagesFlushed: messages, // The loaded total is already persisted; only new publishes are deltas
-		limiter:         limit.New(&limit.Config{}),
 		stripe:          stripe,
 		streams:         newStreamRegistry(),
 	}
@@ -348,6 +347,9 @@ func New(conf *Config) (*Server, error) {
 		firebaseClient = newFirebaseClient(sender, auther)
 	}
 	s.userManager = userManager
+	s.limiter = limit.New(&limit.Config{
+		Store: createUsageStore(pool, userManager),
+	})
 	s.visitors = newVisitorEnv(conf, s.limiter, messageCache, userManager)
 	s.firebaseClient = firebaseClient
 	if pool != nil {
@@ -376,6 +378,9 @@ func New(conf *Config) (*Server, error) {
 		s.tail = tail
 		go s.runTail()
 	}
+	if err := s.limiter.Start(context.Background()); err != nil {
+		return nil, err
+	}
 	return s, nil
 }
 
@@ -392,6 +397,18 @@ func (s *Server) dbPoolStats() []*metrics.DBPool {
 	return pools
 }
 
+// createUsageStore returns the store of the usage table, which is part of the user database: the
+// shared pool on Postgres, the auth-file on SQLite. Without a user database, the daily counts are
+// kept in memory only.
+func createUsageStore(pool *db.DB, userManager *user.Manager) *limit.Store {
+	if pool != nil {
+		return limit.NewPostgresStore(pool)
+	} else if userManager != nil {
+		return limit.NewSQLiteStore(userManager.DB())
+	}
+	return nil
+}
+
 func createMessageCache(conf *Config, pool *db.DB) (*message.Cache, error) {
 	if conf.CacheDuration == 0 {
 		return message.NewNopStore()
@@ -532,7 +549,6 @@ func (s *Server) Run() error {
 	}
 	s.mu.Unlock()
 	go s.runManager()
-	go s.runStatsResetter()
 	go s.runDelayedSender()
 	go s.runFirebaseKeepaliver()
 
@@ -585,6 +601,7 @@ func (s *Server) stop() {
 	if s.tail != nil {
 		s.tail.Close() // Before the databases close under it
 	}
+	s.limiter.Close() // Writes the unflushed daily counts, before the databases close under it
 	s.closeDatabases()
 	if s.ban != nil {
 		s.ban.Close()
@@ -2136,36 +2153,6 @@ func (s *Server) runManager() {
 	}
 }
 
-// runStatsResetter runs once a day (usually midnight UTC) to reset all the visitor's message and
-// email counters. The stats are used to display the counters in the web app, as well as for rate limiting.
-func (s *Server) runStatsResetter() {
-	for {
-		runAt := util.NextOccurrenceUTC(s.config.VisitorStatsResetTime, time.Now())
-		timer := time.NewTimer(time.Until(runAt))
-		log.Tag(tagResetter).Debug("Waiting until %v to reset visitor stats", runAt)
-		select {
-		case <-timer.C:
-			log.Tag(tagResetter).Debug("Running stats resetter")
-			s.resetStats()
-		case <-s.closeChan:
-			log.Tag(tagResetter).Debug("Stopping stats resetter")
-			timer.Stop()
-			return
-		}
-	}
-}
-
-func (s *Server) resetStats() {
-	log.Info("Resetting all visitor stats (daily task)")
-	s.mu.Lock()
-	defer s.mu.Unlock() // Includes the database query to avoid races with other processes
-	if s.userManager != nil {
-		if err := s.userManager.ResetStats(); err != nil {
-			log.Tag(tagResetter).Warn("Failed to write to database: %s", err.Error())
-		}
-	}
-}
-
 func (s *Server) runFirebaseKeepaliver() {
 	if s.firebaseClient == nil {
 		return

server/server_account.go

@@ -96,6 +96,7 @@ func (s *Server) handleAccountGet(w http.ResponseWriter, r *http.Request, v *vis
 			ReservationsRemaining:        stats.ReservationsRemaining,
 			AttachmentTotalSize:          stats.AttachmentTotalSize,
 			AttachmentTotalSizeRemaining: stats.AttachmentTotalSizeRemaining,
+			ResetAt:                      stats.ResetAt.Unix(),
 		},
 	}
 	u := v.User()

server/server_account_test.go

@@ -1113,3 +1113,17 @@ func TestAccount_PhoneNumberAdd_WrongCode(t *testing.T) {
 		require.Equal(t, []string{"+12223334444"}, phoneNumbers)
 	})
 }
+
+func TestServer_AccountStatsResetAt(t *testing.T) {
+	// The account reports when the visitor's daily counts start over: the end of its key's own
+	// 24h period, not midnight UTC
+	s := newTestServer(t, newTestConfigWithAuthFile(t, ""))
+	response := request(t, s, "GET", "/v1/account", "", nil)
+	require.Equal(t, 200, response.Code)
+	account, err := util.UnmarshalJSON[apiAccountResponse](io.NopCloser(response.Body))
+	require.Nil(t, err)
+	resetAt := s.limiter.ResetAt(s.visitor(netip.MustParseAddr("9.9.9.9"), nil).key)
+	require.Equal(t, resetAt.Unix(), account.Stats.ResetAt)
+	require.Greater(t, account.Stats.ResetAt, time.Now().Unix())
+	require.LessOrEqual(t, account.Stats.ResetAt, time.Now().Add(24*time.Hour).Unix()+1)
+}

server/server_payments_test.go

@@ -361,8 +361,6 @@ func TestPayments_Checkout_Success_And_Increase_Rate_Limits(t *testing.T) {
 		require.Equal(t, payments.PriceRecurringInterval(""), u.Billing.StripeSubscriptionInterval)
 		require.Equal(t, int64(0), u.Billing.StripeSubscriptionPaidUntil.Unix())
 		require.Equal(t, int64(0), u.Billing.StripeSubscriptionCancelAt.Unix())
-		require.Equal(t, int64(0), u.Stats.Messages) // Messages and emails are not persisted for no-tier users!
-		require.Equal(t, int64(0), u.Stats.Emails)
 
 		// Simulate Stripe success return URL call (no user context)
 		rr = request(t, s, "GET", "/v1/account/billing/subscription/success/SOMETOKEN", "", nil)
@@ -378,8 +376,6 @@ func TestPayments_Checkout_Success_And_Increase_Rate_Limits(t *testing.T) {
 		require.Equal(t, payments.PriceRecurringInterval(stripe.PriceRecurringIntervalMonth), u.Billing.StripeSubscriptionInterval)
 		require.Equal(t, int64(123456789), u.Billing.StripeSubscriptionPaidUntil.Unix())
 		require.Equal(t, int64(0), u.Billing.StripeSubscriptionCancelAt.Unix())
-		require.Equal(t, int64(0), u.Stats.Messages)
-		require.Equal(t, int64(0), u.Stats.Emails)
 
 		// Now for the fun part: Verify that new rate limits are immediately applied
 		// This only tests the request limiter, which kicks in before the message limiter.

server/server_test.go

@@ -6123,3 +6123,70 @@ func TestServer_SubscriptionLimit_ReleasedWhenStreamCloses(t *testing.T) {
 	require.Equal(t, 200, rr.Code)
 	require.Empty(t, s.visitors.subscriptions.counts)
 }
+
+func TestServer_UsageSurvivesRestart(t *testing.T) {
+	// Daily counts are kept in the usage table of the auth-file: a restart continues where the old
+	// server stopped, except for keys without a user below the threshold (10% of the cap)
+	c := newTestConfigWithAuthFile(t, "")
+	c.VisitorMessageDailyLimit = 100 // Threshold 10
+	s := newTestServer(t, c)
+	require.Nil(t, s.userManager.AddTier(&user.Tier{
+		Code:         "test",
+		MessageLimit: 50,
+	}))
+	require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleUser, false))
+	require.Nil(t, s.userManager.ChangeTier("phil", "test"))
+	publish := func(s *Server, n int, headers map[string]string, ip string) {
+		for i := 0; i < n; i++ {
+			response := request(t, s, "PUT", "/mytopic", "test", headers, func(r *http.Request) {
+				r.RemoteAddr = ip + ":1234"
+			})
+			require.Equal(t, 200, response.Code)
+		}
+	}
+	auth := map[string]string{
+		"Authorization": util.BasicAuth("phil", "phil"),
+	}
+	publish(s, 3, auth, "9.9.9.9")
+	publish(s, 12, nil, "1.1.1.1")
+	publish(s, 5, nil, "2.2.2.2")
+	s.Stop()
+
+	s = newTestServer(t, c)
+	u, err := s.userManager.User("phil")
+	require.Nil(t, err)
+	require.Equal(t, int64(3), s.visitor(netip.MustParseAddr("9.9.9.9"), u).Stats().Messages)
+	require.Equal(t, int64(12), s.visitor(netip.MustParseAddr("1.1.1.1"), nil).Stats().Messages)
+	require.Equal(t, int64(0), s.visitor(netip.MustParseAddr("2.2.2.2"), nil).Stats().Messages)
+}
+
+func TestServer_AnonymousEmailDailyCapShared(t *testing.T) {
+	// Anonymous emails have a daily cap (24h / visitor-email-limit-replenish) shared by all nodes
+	// on one usage table, not one per node
+	forEachBackend(t, func(t *testing.T, databaseURL string) {
+		c := newTestConfigWithAuthFile(t, databaseURL)
+		c.VisitorEmailLimitBurst = 10
+		c.VisitorEmailLimitReplenish = 24 * time.Hour / 5 // Daily cap 5
+		a := newTestServer(t, c)
+		a.mailer = &testMailer{}
+		bConf := *c
+		bConf.CacheFile = filepath.Join(t.TempDir(), "cache.db")
+		b := newTestServer(t, &bConf)
+		b.mailer = &testMailer{}
+		email := map[string]string{
+			"Email": "test@email.com",
+		}
+		for i := 0; i < 3; i++ {
+			require.Equal(t, 200, request(t, a, "PUT", "/mytopic", "test", email).Code)
+		}
+		waitForWithMaxWait(t, 15*time.Second, func() bool {
+			return b.visitor(netip.MustParseAddr("9.9.9.9"), nil).Stats().Emails == 3
+		})
+		for i := 0; i < 2; i++ {
+			require.Equal(t, 200, request(t, b, "PUT", "/mytopic", "test", email).Code)
+		}
+		response := request(t, b, "PUT", "/mytopic", "test", email)
+		require.Equal(t, 429, response.Code)
+		require.Equal(t, 42902, toHTTPError(t, response.Body.String()).Code)
+	})
+}

server/types.go

@@ -309,6 +309,7 @@ type apiAccountStats struct {
 	ReservationsRemaining        int64 `json:"reservations_remaining"`
 	AttachmentTotalSize          int64 `json:"attachment_total_size"`
 	AttachmentTotalSizeRemaining int64 `json:"attachment_total_size_remaining"`
+	ResetAt                      int64 `json:"reset_at"` // Unix seconds: when the daily counts start over (see limit.Service.ResetAt)
 }
 
 type apiAccountReservation struct {

server/visitor.go

@@ -21,6 +21,10 @@ const (
 	// is gone. This number has to be very high to prevent e-mail abuse, but it doesn't really affect
 	// the other limits anyway, since they are replenished faster (typically).
 	visitorExpungeAfter = oneDay
+
+	// promoteRate is the fraction of the daily message cap (1/promoteRate) from which the count of
+	// a visitor without a user is written to the usage table; below it, a restart forgets it
+	promoteRate = 10
 )
 
 // visitor represents one request's API user: its identity (the IP, or the user) and the limits that
@@ -52,6 +56,13 @@ type visitorInfo struct {
 	Stats  *visitorStats
 }
 
+// visitorUsage is a key's daily counts (see visitor.Stats)
+type visitorUsage struct {
+	Messages int64
+	Emails   int64
+	Calls    int64
+}
+
 type visitorStats struct {
 	Messages                     int64
 	MessagesRemaining            int64
@@ -63,6 +74,7 @@ type visitorStats struct {
 	ReservationsRemaining        int64
 	AttachmentTotalSize          int64
 	AttachmentTotalSizeRemaining int64
+	ResetAt                      time.Time // When the daily counts start over
 }
 
 func newVisitorEnv(conf *Config, limiter *limit.Service, messageCache *message.Cache, userManager *user.Manager) *visitorEnv {
@@ -181,13 +193,7 @@ func (v *visitor) MessageAllowed() bool {
 	if v.limits.MessageLimit <= 0 {
 		return false // A daily limit of zero admits nothing; the limit service reads it as "no cap"
 	}
-	allowed := v.env.limiter.Allow(v.key, v.messageRule(), 1).OK
-	if allowed {
-		v.enqueueUserStats(&user.Stats{
-			Messages: 1,
-		})
-	}
-	return allowed
+	return v.env.limiter.Allow(v.key, v.messageRule(), 1).OK
 }
 
 func (v *visitor) EmailAllowed() bool {
@@ -196,35 +202,14 @@ func (v *visitor) EmailAllowed() bool {
 	} else if v.limits.Basis == vis.BasisTier && v.limits.EmailLimit <= 0 {
 		return false // No emails on the tier: the limit service reads a daily limit of zero as "no cap"
 	}
-	allowed := v.env.limiter.Allow(v.key, v.emailRule(), 1).OK
-	if allowed {
-		v.enqueueUserStats(&user.Stats{
-			Emails: 1,
-		})
-	}
-	return allowed
+	return v.env.limiter.Allow(v.key, v.emailRule(), 1).OK
 }
 
 func (v *visitor) CallAllowed() bool {
 	if v.limits.CallLimit <= 0 {
 		return false // A daily limit of zero admits nothing (e.g. a tier without calls, verified number or not)
 	}
-	allowed := v.env.limiter.Allow(v.key, v.callRule(), 1).OK
-	if allowed {
-		v.enqueueUserStats(&user.Stats{
-			Calls: 1,
-		})
-	}
-	return allowed
-}
-
-// enqueueUserStats hands what a tier user's request admitted to the user manager's write queue, which
-// adds it to the persisted stats (deltas add up across requests and cluster nodes)
-func (v *visitor) enqueueUserStats(delta *user.Stats) {
-	if v.env.userManager == nil || v.user == nil || v.user.Tier == nil {
-		return
-	}
-	v.env.userManager.EnqueueUserStats(v.user.ID, delta)
+	return v.env.limiter.Allow(v.key, v.callRule(), 1).OK
 }
 
 // SubscriptionAllowed takes one of the visitor key's subscription slots on this node, if one is free.
@@ -290,9 +275,10 @@ func (v *visitor) BandwidthLimiter() util.Limiter {
 	return v.env.limiter.Limiter(v.key, v.bandwidthRule())
 }
 
-func (v *visitor) Stats() *user.Stats {
+// Stats returns the key's daily counts in its current period
+func (v *visitor) Stats() *visitorUsage {
 	usage := v.env.limiter.Usage(v.key)
-	return &user.Stats{
+	return &visitorUsage{
 		Messages: usage.Count(limit.KindMessages),
 		Emails:   usage.Count(limit.KindEmails),
 		Calls:    usage.Count(limit.KindCalls),
@@ -338,25 +324,29 @@ func (v *visitor) requestRule() limit.Rule {
 	}
 }
 
+// messageRule is the daily message limit. The count of a visitor without a user is written to the
+// usage table only from a tenth of the cap on (most keys never get there, so their rows are
+// saved); a logged-in user's from the first message, so the account page agrees across nodes.
 func (v *visitor) messageRule() limit.Rule {
-	return limit.Rule{
+	rule := limit.Rule{
 		Kind:  limit.KindMessages,
 		Daily: v.limits.MessageLimit,
 	}
+	if v.user == nil {
+		rule.Threshold = v.limits.MessageLimit / promoteRate
+	}
+	return rule
 }
 
-// emailRule is the email bucket, and for a tier also its daily email limit. Without a tier, the
-// daily limit is only derived from the refill, for display, and is not a cap.
+// emailRule is the email bucket and the daily email limit; without a tier, the daily limit is
+// derived from the refill (24h / visitor-email-limit-replenish)
 func (v *visitor) emailRule() limit.Rule {
-	rule := limit.Rule{
+	return limit.Rule{
 		Kind:   limit.KindEmails,
 		Burst:  int64(v.limits.EmailLimitBurst),
 		Refill: v.limits.EmailLimitReplenish,
+		Daily:  v.limits.EmailLimit,
 	}
-	if v.limits.Basis == vis.BasisTier {
-		rule.Daily = v.limits.EmailLimit
-	}
-	return rule
 }
 
 func (v *visitor) callRule() limit.Rule {
@@ -459,6 +449,7 @@ func (v *visitor) infoLight() *visitorInfo {
 			EmailsRemaining:   zeroIfNegative(limits.EmailLimit - stats.Emails),
 			Calls:             stats.Calls,
 			CallsRemaining:    zeroIfNegative(limits.CallLimit - stats.Calls),
+			ResetAt:           v.env.limiter.ResetAt(v.key),
 		},
 	}
 }

server/visitor_test.go

@@ -15,40 +15,10 @@ import (
 	"heckel.io/ntfy/v2/util"
 )
 
-func TestVisitor_UserStatsDeltaFlush(t *testing.T) {
-	forEachBackend(t, func(t *testing.T, databaseURL string) {
-		c := newTestConfigWithAuthFile(t, databaseURL)
-		c.AuthStatsQueueWriterInterval = 300 * time.Millisecond
-		s := newTestServer(t, c)
-		require.Nil(t, s.userManager.AddTier(&user.Tier{
-			Code:         "test",
-			MessageLimit: 100,
-		}))
-		require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleUser, false))
-		require.Nil(t, s.userManager.ChangeTier("phil", "test"))
-
-		// Publish 3 messages; the visitor enqueues a stats delta after each publish
-		for i := 0; i < 3; i++ {
-			response := request(t, s, "PUT", "/mytopic", "test", map[string]string{
-				"Authorization": util.BasicAuth("phil", "phil"),
-			})
-			require.Equal(t, 200, response.Code)
-		}
-
-		// After two flush intervals, the persisted count must be exactly 3: per-publish
-		// enqueues must be deltas, not absolute snapshots (snapshots would add up to 1+2+3=6)
-		time.Sleep(time.Second)
-		u, err := s.userManager.User("phil")
-		require.Nil(t, err)
-		require.Equal(t, int64(3), u.Stats.Messages)
-	})
-}
-
 func TestVisitor_TierUserDoesNotGetAnonymousUsage(t *testing.T) {
 	// What a user's IP counted anonymously must not land on the account once the user (with a
 	// tier) shows up from that IP.
 	c := newTestConfigWithAuthFile(t, "")
-	c.AuthStatsQueueWriterInterval = 100 * time.Millisecond
 	s := newTestServer(t, c)
 	require.Nil(t, s.userManager.AddTier(&user.Tier{Code: "test", MessageLimit: 100}))
 	require.Nil(t, s.userManager.AddUser("phil", "phil", user.RoleUser, false))
@@ -59,12 +29,10 @@ func TestVisitor_TierUserDoesNotGetAnonymousUsage(t *testing.T) {
 	for i := 0; i < 3; i++ {
 		require.True(t, v.MessageAllowed())
 	}
-
-	require.True(t, s.visitor(netip.MustParseAddr("9.9.9.9"), u).RequestAllowed())
-	time.Sleep(500 * time.Millisecond)
-	u, err = s.userManager.User("phil")
-	require.Nil(t, err)
-	require.Equal(t, int64(0), u.Stats.Messages)
+	vu := s.visitor(netip.MustParseAddr("9.9.9.9"), u)
+	require.True(t, vu.RequestAllowed())
+	require.Equal(t, int64(0), vu.Stats().Messages)
+	require.Equal(t, int64(3), v.Stats().Messages)
 }
 
 func TestVisitor_BandwidthRefund(t *testing.T) {
@@ -379,10 +347,9 @@ func TestVisitor_AuthAndAccountLimitsIgnoreUser(t *testing.T) {
 
 func TestVisitor_SharedKeyCountsOnce(t *testing.T) {
 	// Two visitors with the same user key (the user from two addresses) share one count in the
-	// limiter, and each admitted unit is persisted once, so the user's stats are not double counted
+	// limiter, so the user's usage is not double counted
 	forEachBackend(t, func(t *testing.T, databaseURL string) {
 		c := newTestConfigWithAuthFile(t, databaseURL)
-		c.AuthStatsQueueWriterInterval = 100 * time.Millisecond
 		s := newTestServer(t, c)
 		require.Nil(t, s.userManager.AddTier(&user.Tier{
 			Code:         "test",
@@ -404,17 +371,6 @@ func TestVisitor_SharedKeyCountsOnce(t *testing.T) {
 		require.Equal(t, int64(5), v1.Stats().Messages)
 		require.Equal(t, int64(5), v2.Stats().Messages)
 
-		// Persisted once: 3 + 2, not 5 + 5
-		waitFor(t, func() bool {
-			u, err := s.userManager.User("phil")
-			require.Nil(t, err)
-			return u.Stats.Messages > 0
-		})
-		time.Sleep(300 * time.Millisecond) // Any double counting would have been flushed by now
-		u, err = s.userManager.User("phil")
-		require.Nil(t, err)
-		require.Equal(t, int64(5), u.Stats.Messages)
-
 		// The account sees the shared count, and the day's remaining messages are shared too
 		response := request(t, s, "GET", "/v1/account", "", map[string]string{
 			"Authorization": util.BasicAuth("phil", "phil"),
@@ -471,12 +427,10 @@ func TestVisitor_TopicCreationLimiterDisabledByZero(t *testing.T) {
 }
 
 func TestVisitor_TierChangeDoesNotReseedCounts(t *testing.T) {
-	// The persisted stats seed a key only on its first sight. On a tier change (and for a new
-	// visitor object of a known key), the live counts win over the user's stale database row,
-	// which lags behind by whatever the stats queue has not flushed yet.
+	// On a tier change (and for a new visitor object of a known key), the key keeps its live
+	// counts; nothing is seeded from the user's row.
 	forEachBackend(t, func(t *testing.T, databaseURL string) {
 		c := newTestConfigWithAuthFile(t, databaseURL)
-		c.AuthStatsQueueWriterInterval = time.Hour // Never flushed during the test: the row stays stale
 		s := newTestServer(t, c)
 		require.Nil(t, s.userManager.AddTier(&user.Tier{
 			Code:         "small",
@@ -498,7 +452,6 @@ func TestVisitor_TierChangeDoesNotReseedCounts(t *testing.T) {
 		require.Nil(t, s.userManager.ChangeTier("phil", "large"))
 		u, err = s.userManager.User("phil")
 		require.Nil(t, err)
-		require.Equal(t, int64(0), u.Stats.Messages) // Stale
 		v = s.visitor(ip, u) // The next request carries the new tier
 		require.Equal(t, int64(4), v.Stats().Messages)
 		for i := 0; i < 6; i++ {
@@ -701,15 +654,15 @@ func TestVisitor_RateVisitorLivenessIsTheSubscribers(t *testing.T) {
 	require.Nil(t, topics[0].RateVisitor())
 }
 
-func TestVisitor_EmailsWithoutTierHaveNoDailyCap(t *testing.T) {
-	// Without a tier, emails are limited by the email bucket alone, as before: the daily limit derived
-	// from the refill (24 with a refill of an hour) is not a cap, so a full burst of 30 can be sent
+func TestVisitor_EmailsWithoutTierHaveDailyCap(t *testing.T) {
+	// Without a tier, the daily limit derived from the refill (24 with a refill of an hour) is a
+	// cap, as for tiers: a burst of 30 is cut at 24
 	c := newTestConfig(t, "")
 	c.VisitorEmailLimitBurst = 30
 	c.VisitorEmailLimitReplenish = time.Hour
 	s := newTestServer(t, c)
 	v := s.visitor(netip.MustParseAddr("9.9.9.9"), nil)
-	for i := 0; i < 30; i++ {
+	for i := 0; i < 24; i++ {
 		require.True(t, v.EmailAllowed(), "email %d", i)
 	}
 	require.False(t, v.EmailAllowed())
@@ -727,3 +680,20 @@ func TestVisitor_EmailsWithoutTierAllowedWithSlowRefill(t *testing.T) {
 	require.True(t, v.EmailAllowed())
 	require.False(t, v.EmailAllowed())
 }
+
+func TestVisitor_LoggedInUserWrittenFromFirstMessage(t *testing.T) {
+	// A logged-in user's messages are written from the first one, even without a tier (on the IP
+	// key), so the account page agrees across nodes and restarts; anonymous ones only from the
+	// threshold on
+	c := newTestConfigWithAuthFile(t, "")
+	s := newTestServer(t, c)
+	require.Nil(t, s.userManager.AddUser("ben", "ben", user.RoleUser, false))
+	u, err := s.userManager.User("ben")
+	require.Nil(t, err)
+	require.True(t, s.visitor(netip.MustParseAddr("1.1.1.1"), u).MessageAllowed())
+	require.True(t, s.visitor(netip.MustParseAddr("2.2.2.2"), nil).MessageAllowed())
+	s.Stop()
+	s = newTestServer(t, c)
+	require.Equal(t, int64(1), s.visitor(netip.MustParseAddr("1.1.1.1"), u).Stats().Messages)
+	require.Equal(t, int64(0), s.visitor(netip.MustParseAddr("2.2.2.2"), nil).Stats().Messages)
+}

user/manager.go

@@ -38,7 +38,7 @@ const (
 
 // Default constants that may be overridden by configs
 const (
-	DefaultUserStatsQueueWriterInterval = 33 * time.Second
+	DefaultUserStatsQueueWriterInterval = 33 * time.Second // How often the token access stats are written; the daily counts are the limit service's (see package limit)
 	DefaultUserPasswordBcryptCost       = 10
 	DefaultAccessCacheEnabled           = false
 	DefaultAccessCacheReloadInterval    = 67 * time.Second
@@ -56,7 +56,6 @@ type Manager struct {
 	config      *Config
 	db          *db.DB
 	queries     queries
-	statsQueue  map[string]*Stats       // Accumulated stats deltas to be written to the database (UserID -> Stats)
 	tokenQueue  map[string]*TokenUpdate // "Queue" to asynchronously write token access stats to the database (Token ID -> TokenUpdate)
 	accessCache *accessCache            // In-memory snapshot of user_access; reloaded by accessWritten after every ACL mutation
 	quit        chan struct{}           // Closed by Close() to signal background goroutines to stop
@@ -81,7 +80,6 @@ func newManager(d *db.DB, queries queries, config *Config) (*Manager, error) {
 	manager := &Manager{
 		config:     config,
 		db:         d,
-		statsQueue: make(map[string]*Stats),
 		tokenQueue: make(map[string]*TokenUpdate),
 		quit:       make(chan struct{}),
 		queries:    queries,
@@ -422,33 +420,6 @@ func (a *Manager) checkReservationsLimitTx(tx *sql.Tx, username string, reservat
 	return nil
 }
 
-// ResetStats resets all user stats in the user database. This touches all users.
-func (a *Manager) ResetStats() error {
-	a.mu.Lock() // Includes database query to avoid races!
-	defer a.mu.Unlock()
-	if _, err := a.db.Exec(a.queries.updateUserStatsResetAll); err != nil {
-		return err
-	}
-	a.statsQueue = make(map[string]*Stats)
-	return nil
-}
-
-// EnqueueUserStats adds the given stats *delta* to a queue which writes out user stats
-// (messages, emails, ..) in batches at a regular interval. Deltas accumulate in the queue and
-// are flushed as SQL increments, so concurrent flushes from multiple nodes add up instead of
-// overwriting each other.
-func (a *Manager) EnqueueUserStats(userID string, delta *Stats) {
-	a.mu.Lock()
-	defer a.mu.Unlock()
-	if queued, ok := a.statsQueue[userID]; ok {
-		queued.Messages += delta.Messages
-		queued.Emails += delta.Emails
-		queued.Calls += delta.Calls
-	} else {
-		a.statsQueue[userID] = &Stats{Messages: delta.Messages, Emails: delta.Emails, Calls: delta.Calls}
-	}
-}
-
 func (a *Manager) asyncQueueWriteLoop(interval time.Duration) {
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
@@ -457,9 +428,6 @@ func (a *Manager) asyncQueueWriteLoop(interval time.Duration) {
 		case <-a.quit:
 			return
 		case <-ticker.C:
-			if err := a.writeUserStatsQueue(); err != nil {
-				log.Tag(tag).Err(err).Warn("Writing user stats queue failed")
-			}
 			if err := a.writeTokenUpdateQueue(); err != nil {
 				log.Tag(tag).Err(err).Warn("Writing token update queue failed")
 			}
@@ -467,37 +435,6 @@ func (a *Manager) asyncQueueWriteLoop(interval time.Duration) {
 	}
 }
 
-func (a *Manager) writeUserStatsQueue() error {
-	a.mu.Lock()
-	if len(a.statsQueue) == 0 {
-		a.mu.Unlock()
-		log.Tag(tag).Trace("No user stats updates to commit")
-		return nil
-	}
-	statsQueue := a.statsQueue
-	a.statsQueue = make(map[string]*Stats)
-	a.mu.Unlock()
-
-	return db.ExecTx(a.db, func(tx *sql.Tx) error {
-		log.Tag(tag).Debug("Writing user stats queue for %d user(s)", len(statsQueue))
-		for userID, update := range statsQueue {
-			log.
-				Tag(tag).
-				Fields(log.Context{
-					"user_id":        userID,
-					"messages_count": update.Messages,
-					"emails_count":   update.Emails,
-					"calls_count":    update.Calls,
-				}).
-				Trace("Updating stats for user %s", userID)
-			if _, err := tx.Exec(a.queries.updateUserStats, update.Messages, update.Emails, update.Calls, userID); err != nil {
-				return err
-			}
-		}
-		return nil
-	})
-}
-
 // User returns the user with the given username if it exists, or ErrUserNotFound otherwise
 func (a *Manager) User(username string) (*User, error) {
 	return a.userTx(a.db, username)
@@ -622,9 +559,8 @@ func (a *Manager) scanUser(rows *sql.Rows) (*User, error) {
 	var id, username, hash, role, prefs, syncTopic string
 	var provisioned bool
 	var stripeCustomerID, stripeSubscriptionID, stripeSubscriptionStatus, stripeSubscriptionInterval, stripeMonthlyPriceID, stripeYearlyPriceID, tierID, tierCode, tierName sql.NullString
-	var messages, emails, calls int64
 	var messagesLimit, messagesExpiryDuration, emailsLimit, callsLimit, reservationsLimit, attachmentFileSizeLimit, attachmentTotalSizeLimit, attachmentExpiryDuration, attachmentBandwidthLimit, stripeSubscriptionPaidUntil, stripeSubscriptionCancelAt, deleted sql.NullInt64
-	if err := rows.Scan(&id, &username, &hash, &role, &prefs, &syncTopic, &provisioned, &messages, &emails, &calls, &stripeCustomerID, &stripeSubscriptionID, &stripeSubscriptionStatus, &stripeSubscriptionInterval, &stripeSubscriptionPaidUntil, &stripeSubscriptionCancelAt, &deleted, &tierID, &tierCode, &tierName, &messagesLimit, &messagesExpiryDuration, &emailsLimit, &callsLimit, &reservationsLimit, &attachmentFileSizeLimit, &attachmentTotalSizeLimit, &attachmentExpiryDuration, &attachmentBandwidthLimit, &stripeMonthlyPriceID, &stripeYearlyPriceID); err != nil {
+	if err := rows.Scan(&id, &username, &hash, &role, &prefs, &syncTopic, &provisioned, &stripeCustomerID, &stripeSubscriptionID, &stripeSubscriptionStatus, &stripeSubscriptionInterval, &stripeSubscriptionPaidUntil, &stripeSubscriptionCancelAt, &deleted, &tierID, &tierCode, &tierName, &messagesLimit, &messagesExpiryDuration, &emailsLimit, &callsLimit, &reservationsLimit, &attachmentFileSizeLimit, &attachmentTotalSizeLimit, &attachmentExpiryDuration, &attachmentBandwidthLimit, &stripeMonthlyPriceID, &stripeYearlyPriceID); err != nil {
 		return nil, err
 	} else if err := rows.Err(); err != nil {
 		return nil, err
@@ -637,11 +573,6 @@ func (a *Manager) scanUser(rows *sql.Rows) (*User, error) {
 		Prefs:       &Prefs{},
 		SyncTopic:   syncTopic,
 		Provisioned: provisioned,
-		Stats: &Stats{
-			Messages: messages,
-			Emails:   emails,
-			Calls:    calls,
-		},
 		Billing: &Billing{
 			StripeCustomerID:            stripeCustomerID.String,                                            // May be empty
 			StripeSubscriptionID:        stripeSubscriptionID.String,                                        // May be empty

user/manager_postgres.go

@@ -9,7 +9,7 @@ import (
 const (
 	// User queries
 	postgresSelectUsersQuery = `
-		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM "user" u
 		LEFT JOIN tier t on t.id = u.tier_id
 		ORDER BY
@@ -20,19 +20,19 @@ const (
 			END, u.user_name
 	`
 	postgresSelectUserByIDQuery = `
-		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM "user" u
 		LEFT JOIN tier t on t.id = u.tier_id
 		WHERE u.id = $1
 	`
 	postgresSelectUserByNameQuery = `
-		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM "user" u
 		LEFT JOIN tier t on t.id = u.tier_id
 		WHERE user_name = $1
 	`
 	postgresSelectUserByNameOrPrimaryEmailQuery = `
-		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM "user" u
 		LEFT JOIN tier t on t.id = u.tier_id
 		WHERE u.user_name = $1
@@ -41,14 +41,14 @@ const (
 		LIMIT 1
 	`
 	postgresSelectUserByTokenQuery = `
-		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM "user" u
 		JOIN user_token tk on u.id = tk.user_id
 		LEFT JOIN tier t on t.id = u.tier_id
 		WHERE tk.token = $1 AND (tk.expires = 0 OR tk.expires >= $2)
 	`
 	postgresSelectUserByStripeCustomerIDQuery = `
-		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user_name, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, u.deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM "user" u
 		LEFT JOIN tier t on t.id = u.tier_id
 		WHERE u.stripe_customer_id = $1
@@ -70,8 +70,6 @@ const (
 	postgresUpdateUserRoleQuery           = `UPDATE "user" SET role = $1 WHERE user_name = $2`
 	postgresUpdateUserProvisionedQuery    = `UPDATE "user" SET provisioned = $1 WHERE user_name = $2`
 	postgresUpdateUserPrefsQuery          = `UPDATE "user" SET prefs = $1 WHERE id = $2`
-	postgresUpdateUserStatsQuery          = `UPDATE "user" SET stats_messages = stats_messages + $1, stats_emails = stats_emails + $2, stats_calls = stats_calls + $3 WHERE id = $4`
-	postgresUpdateUserStatsResetAllQuery  = `UPDATE "user" SET stats_messages = 0, stats_emails = 0, stats_calls = 0`
 	postgresUpdateUserTierQuery           = `UPDATE "user" SET tier_id = (SELECT id FROM tier WHERE code = $1) WHERE user_name = $2`
 	postgresUpdateUserDeletedQuery        = `UPDATE "user" SET deleted = $1 WHERE id = $2`
 	postgresDeleteUserQuery               = `DELETE FROM "user" WHERE user_name = $1`
@@ -266,8 +264,6 @@ var postgresQueries = queries{
 	updateUserRole:                 postgresUpdateUserRoleQuery,
 	updateUserProvisioned:          postgresUpdateUserProvisionedQuery,
 	updateUserPrefs:                postgresUpdateUserPrefsQuery,
-	updateUserStats:                postgresUpdateUserStatsQuery,
-	updateUserStatsResetAll:        postgresUpdateUserStatsResetAllQuery,
 	updateUserTier:                 postgresUpdateUserTierQuery,
 	updateUserDeleted:              postgresUpdateUserDeletedQuery,
 	deleteUser:                     postgresDeleteUserQuery,

user/manager_sqlite.go

@@ -15,7 +15,7 @@ import (
 const (
 	// User queries
 	sqliteSelectUsersQuery = `
-		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM user u
 		LEFT JOIN tier t on t.id = u.tier_id
 		ORDER BY
@@ -26,19 +26,19 @@ const (
 			END, u.user
 	`
 	sqliteSelectUserByIDQuery = `
-		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM user u
 		LEFT JOIN tier t on t.id = u.tier_id
 		WHERE u.id = ?
 	`
 	sqliteSelectUserByNameQuery = `
-		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM user u
 		LEFT JOIN tier t on t.id = u.tier_id
 		WHERE user = ?
 	`
 	sqliteSelectUserByNameOrPrimaryEmailQuery = `
-		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM user u
 		LEFT JOIN tier t on t.id = u.tier_id
 		WHERE u.user = ?
@@ -47,14 +47,14 @@ const (
 		LIMIT 1
 	`
 	sqliteSelectUserByTokenQuery = `
-		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM user u
 		JOIN user_token tk on u.id = tk.user_id
 		LEFT JOIN tier t on t.id = u.tier_id
 		WHERE tk.token = ? AND (tk.expires = 0 OR tk.expires >= ?)
 	`
 	sqliteSelectUserByStripeCustomerIDQuery = `
-		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stats_messages, u.stats_emails, u.stats_calls, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
+		SELECT u.id, u.user, u.pass, u.role, u.prefs, u.sync_topic, u.provisioned, u.stripe_customer_id, u.stripe_subscription_id, u.stripe_subscription_status, u.stripe_subscription_interval, u.stripe_subscription_paid_until, u.stripe_subscription_cancel_at, deleted, t.id, t.code, t.name, t.messages_limit, t.messages_expiry_duration, t.emails_limit, t.calls_limit, t.reservations_limit, t.attachment_file_size_limit, t.attachment_total_size_limit, t.attachment_expiry_duration, t.attachment_bandwidth_limit, t.stripe_monthly_price_id, t.stripe_yearly_price_id
 		FROM user u
 		LEFT JOIN tier t on t.id = u.tier_id
 		WHERE u.stripe_customer_id = ?
@@ -76,8 +76,6 @@ const (
 	sqliteUpdateUserRoleQuery           = `UPDATE user SET role = ? WHERE user = ?`
 	sqliteUpdateUserProvisionedQuery    = `UPDATE user SET provisioned = ? WHERE user = ?`
 	sqliteUpdateUserPrefsQuery          = `UPDATE user SET prefs = ? WHERE id = ?`
-	sqliteUpdateUserStatsQuery          = `UPDATE user SET stats_messages = stats_messages + ?, stats_emails = stats_emails + ?, stats_calls = stats_calls + ? WHERE id = ?`
-	sqliteUpdateUserStatsResetAllQuery  = `UPDATE user SET stats_messages = 0, stats_emails = 0, stats_calls = 0`
 	sqliteUpdateUserTierQuery           = `UPDATE user SET tier_id = (SELECT id FROM tier WHERE code = ?) WHERE user = ?`
 	sqliteUpdateUserDeletedQuery        = `UPDATE user SET deleted = ? WHERE id = ?`
 	sqliteDeleteUserQuery               = `DELETE FROM user WHERE user = ?`
@@ -264,8 +262,6 @@ var sqliteQueries = queries{
 	updateUserRole:                 sqliteUpdateUserRoleQuery,
 	updateUserProvisioned:          sqliteUpdateUserProvisionedQuery,
 	updateUserPrefs:                sqliteUpdateUserPrefsQuery,
-	updateUserStats:                sqliteUpdateUserStatsQuery,
-	updateUserStatsResetAll:        sqliteUpdateUserStatsResetAllQuery,
 	updateUserTier:                 sqliteUpdateUserTierQuery,
 	updateUserDeleted:              sqliteUpdateUserDeletedQuery,
 	deleteUser:                     sqliteDeleteUserQuery,

user/manager_test.go

@@ -798,54 +798,6 @@ func TestManager_Token_MaxCount_AutoDelete(t *testing.T) {
 	})
 }
 
-func TestManager_EnqueueStats_ResetStats(t *testing.T) {
-	forEachBackend(t, func(t *testing.T, newManager newManagerFunc) {
-		conf := &Config{
-			DefaultAccess:       PermissionReadWrite,
-			BcryptCost:          bcrypt.MinCost,
-			QueueWriterInterval: 1500 * time.Millisecond,
-		}
-		a := newTestManagerFromConfig(t, newManager, conf)
-		require.Nil(t, a.AddUser("ben", "ben", RoleUser, false))
-
-		// Baseline: No messages or emails
-		u, err := a.User("ben")
-		require.Nil(t, err)
-		require.Equal(t, int64(0), u.Stats.Messages)
-		require.Equal(t, int64(0), u.Stats.Emails)
-		a.EnqueueUserStats(u.ID, &Stats{
-			Messages: 11,
-			Emails:   2,
-		})
-
-		// Still no change, because it's queued asynchronously
-		u, err = a.User("ben")
-		require.Nil(t, err)
-		require.Equal(t, int64(0), u.Stats.Messages)
-		require.Equal(t, int64(0), u.Stats.Emails)
-
-		// After 2 seconds they should be persisted
-		time.Sleep(2 * time.Second)
-
-		u, err = a.User("ben")
-		require.Nil(t, err)
-		require.Equal(t, int64(11), u.Stats.Messages)
-		require.Equal(t, int64(2), u.Stats.Emails)
-
-		// Now reset stats (enqueued stats will be thrown out)
-		a.EnqueueUserStats(u.ID, &Stats{
-			Messages: 99,
-			Emails:   23,
-		})
-		require.Nil(t, a.ResetStats())
-
-		u, err = a.User("ben")
-		require.Nil(t, err)
-		require.Equal(t, int64(0), u.Stats.Messages)
-		require.Equal(t, int64(0), u.Stats.Emails)
-	})
-}
-
 func TestManager_EnqueueTokenUpdate(t *testing.T) {
 	forEachBackend(t, func(t *testing.T, newManager newManagerFunc) {
 		conf := &Config{
@@ -3019,41 +2971,6 @@ func TestStoreChangeBilling(t *testing.T) {
 	})
 }
 
-func TestStoreUpdateStats(t *testing.T) {
-	forEachStoreBackend(t, func(t *testing.T, manager *Manager) {
-		require.Nil(t, manager.AddUser("phil", "mypass", RoleUser, false))
-		u, err := manager.User("phil")
-		require.Nil(t, err)
-
-		manager.EnqueueUserStats(u.ID, &Stats{Messages: 42, Emails: 3, Calls: 1})
-		require.Nil(t, manager.writeUserStatsQueue())
-
-		u2, err := manager.User("phil")
-		require.Nil(t, err)
-		require.Equal(t, int64(42), u2.Stats.Messages)
-		require.Equal(t, int64(3), u2.Stats.Emails)
-		require.Equal(t, int64(1), u2.Stats.Calls)
-	})
-}
-
-func TestStoreResetStats(t *testing.T) {
-	forEachStoreBackend(t, func(t *testing.T, manager *Manager) {
-		require.Nil(t, manager.AddUser("phil", "mypass", RoleUser, false))
-		u, err := manager.User("phil")
-		require.Nil(t, err)
-
-		manager.EnqueueUserStats(u.ID, &Stats{Messages: 42, Emails: 3, Calls: 1})
-		require.Nil(t, manager.writeUserStatsQueue())
-		require.Nil(t, manager.ResetStats())
-
-		u2, err := manager.User("phil")
-		require.Nil(t, err)
-		require.Equal(t, int64(0), u2.Stats.Messages)
-		require.Equal(t, int64(0), u2.Stats.Emails)
-		require.Equal(t, int64(0), u2.Stats.Calls)
-	})
-}
-
 func TestStoreMarkUserRemoved(t *testing.T) {
 	forEachStoreBackend(t, func(t *testing.T, manager *Manager) {
 		require.Nil(t, manager.AddUser("phil", "mypass", RoleUser, false))
@@ -3696,43 +3613,6 @@ func TestManager_AccountReadsUsePrimary(t *testing.T) {
 	require.Len(t, pendingEmails, 1)
 }
 
-func TestManager_EnqueueStats_DeltasAccumulate(t *testing.T) {
-	forEachBackend(t, func(t *testing.T, newManager newManagerFunc) {
-		conf := &Config{
-			DefaultAccess:       PermissionReadWrite,
-			BcryptCost:          bcrypt.MinCost,
-			QueueWriterInterval: time.Hour, // Flushed manually below
-		}
-		a := newTestManagerFromConfig(t, newManager, conf)
-		require.Nil(t, a.AddUser("ben", "ben", RoleUser, false))
-		u, err := a.User("ben")
-		require.Nil(t, err)
-
-		// Two enqueues before a flush add up (deltas, not absolute snapshots)
-		a.EnqueueUserStats(u.ID, &Stats{Messages: 3, Emails: 1})
-		a.EnqueueUserStats(u.ID, &Stats{Messages: 2, Calls: 4})
-		require.Nil(t, a.writeUserStatsQueue())
-		u, err = a.User("ben")
-		require.Nil(t, err)
-		require.Equal(t, int64(5), u.Stats.Messages)
-		require.Equal(t, int64(1), u.Stats.Emails)
-		require.Equal(t, int64(4), u.Stats.Calls)
-
-		// A flush from a second manager on the same database (= another cluster node)
-		// adds to the existing stats instead of overwriting them
-		b := newManager(conf)
-		b.EnqueueUserStats(u.ID, &Stats{Messages: 10, Emails: 2})
-		require.Nil(t, b.writeUserStatsQueue())
-		a.EnqueueUserStats(u.ID, &Stats{Messages: 1})
-		require.Nil(t, a.writeUserStatsQueue())
-		u, err = a.User("ben")
-		require.Nil(t, err)
-		require.Equal(t, int64(16), u.Stats.Messages)
-		require.Equal(t, int64(3), u.Stats.Emails)
-		require.Equal(t, int64(4), u.Stats.Calls)
-	})
-}
-
 func TestMigrationFrom9To10SQLite(t *testing.T) {
 	filename := filepath.Join(t.TempDir(), "user.db")
 	a := newTestManagerFromFile(t, filename, "", PermissionDenyAll, bcrypt.MinCost, DefaultUserStatsQueueWriterInterval)

user/types.go

@@ -19,7 +19,6 @@ type User struct {
 	Role        Role
 	Prefs       *Prefs
 	Tier        *Tier
-	Stats       *Stats
 	Billing     *Billing
 	SyncTopic   string
 	Provisioned bool // Whether the user was provisioned by the config file
@@ -153,13 +152,6 @@ type NotificationPrefs struct {
 	DeleteAfter *int    `json:"delete_after,omitempty"`
 }
 
-// Stats is a struct holding daily user statistics
-type Stats struct {
-	Messages int64
-	Emails   int64
-	Calls    int64
-}
-
 // Billing is a struct holding a user's billing information
 type Billing struct {
 	StripeCustomerID            string
@@ -377,8 +369,6 @@ type queries struct {
 	updateUserRole                 string
 	updateUserProvisioned          string
 	updateUserPrefs                string
-	updateUserStats                string
-	updateUserStatsResetAll        string
 	updateUserTier                 string
 	updateUserDeleted              string
 	deleteUser                     string

util/time.go

@@ -24,18 +24,6 @@ func FormatTime(t time.Time) string {
 	return t.Format(timestampFormat)
 }
 
-// NextOccurrenceUTC takes a time of day (e.g. 9:00am), and returns the next occurrence
-// of that time from the current time (in UTC).
-func NextOccurrenceUTC(timeOfDay, base time.Time) time.Time {
-	hour, minute, seconds := timeOfDay.UTC().Clock()
-	now := base.UTC()
-	next := time.Date(now.Year(), now.Month(), now.Day(), hour, minute, seconds, 0, time.UTC)
-	if next.Before(now) {
-		next = next.AddDate(0, 0, 1)
-	}
-	return next
-}
-
 // ParseFutureTime parses a date/time string to a time.Time. It supports unix timestamps, durations
 // and natural language dates
 func ParseFutureTime(s string, now time.Time) (time.Time, error) {

util/time_test.go

@@ -11,26 +11,6 @@ var (
 	base = time.Date(2021, 12, 10, 10, 17, 23, 0, time.UTC)
 )
 
-func TestNextOccurrenceUTC_NextDate(t *testing.T) {
-	loc, err := time.LoadLocation("America/New_York")
-	require.Nil(t, err)
-
-	timeOfDay := time.Date(0, 0, 0, 0, 0, 0, 0, time.UTC) // Run at midnight UTC
-	nowInFairfieldCT := time.Date(2023, time.January, 10, 22, 19, 12, 0, loc)
-	nextRunTme := NextOccurrenceUTC(timeOfDay, nowInFairfieldCT)
-	require.Equal(t, time.Date(2023, time.January, 12, 0, 0, 0, 0, time.UTC), nextRunTme)
-}
-
-func TestNextOccurrenceUTC_SameDay(t *testing.T) {
-	loc, err := time.LoadLocation("America/New_York")
-	require.Nil(t, err)
-
-	timeOfDay := time.Date(0, 0, 0, 4, 0, 0, 0, time.UTC) // Run at 4am UTC
-	nowInFairfieldCT := time.Date(2023, time.January, 10, 22, 19, 12, 0, loc)
-	nextRunTme := NextOccurrenceUTC(timeOfDay, nowInFairfieldCT)
-	require.Equal(t, time.Date(2023, time.January, 11, 4, 0, 0, 0, time.UTC), nextRunTme)
-}
-
 func TestParseFutureTime_11am_FutureTime(t *testing.T) {
 	d, err := ParseFutureTime("11am", base)
 	require.Nil(t, err)