diff --git a/server/config.go b/server/config.go index 67b24a08..33728b85 100644 --- a/server/config.go +++ b/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 diff --git a/server/log.go b/server/log.go index 69c8fcca..8fbeab31 100644 --- a/server/log.go +++ b/server/log.go @@ -27,7 +27,6 @@ const ( tagStripe = "stripe" tagAccount = "account" tagManager = "manager" - tagResetter = "resetter" tagWebsocket = "websocket" tagMatrix = "matrix" tagWebPush = "webpush" diff --git a/server/server.go b/server/server.go index 12781736..43627cbf 100644 --- a/server/server.go +++ b/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 diff --git a/server/server_account.go b/server/server_account.go index 24f542ea..61616cc3 100644 --- a/server/server_account.go +++ b/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() diff --git a/server/server_account_test.go b/server/server_account_test.go index 9ef9f245..39528166 100644 --- a/server/server_account_test.go +++ b/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) +} diff --git a/server/server_payments_test.go b/server/server_payments_test.go index 64bad107..e1df26d7 100644 --- a/server/server_payments_test.go +++ b/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. diff --git a/server/server_test.go b/server/server_test.go index 20aeb6e7..d5d01048 100644 --- a/server/server_test.go +++ b/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) + }) +} diff --git a/server/types.go b/server/types.go index a779db3f..c4b20e05 100644 --- a/server/types.go +++ b/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 { diff --git a/server/visitor.go b/server/visitor.go index 980742f4..7ea9e571 100644 --- a/server/visitor.go +++ b/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), }, } } diff --git a/server/visitor_test.go b/server/visitor_test.go index b1c4b72c..a89a6c4b 100644 --- a/server/visitor_test.go +++ b/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,8 +452,7 @@ 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 + 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++ { require.True(t, v.MessageAllowed()) @@ -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) +} diff --git a/user/manager.go b/user/manager.go index 86b93c9e..fba9e74c 100644 --- a/user/manager.go +++ b/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 diff --git a/user/manager_postgres.go b/user/manager_postgres.go index 6587dac5..645495f4 100644 --- a/user/manager_postgres.go +++ b/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, diff --git a/user/manager_sqlite.go b/user/manager_sqlite.go index c92627f6..21289794 100644 --- a/user/manager_sqlite.go +++ b/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, diff --git a/user/manager_test.go b/user/manager_test.go index 023bb7a5..de8f7edd 100644 --- a/user/manager_test.go +++ b/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) diff --git a/user/types.go b/user/types.go index 490442e2..a69041fe 100644 --- a/user/types.go +++ b/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 diff --git a/util/time.go b/util/time.go index 4583e7e2..ab46367e 100644 --- a/util/time.go +++ b/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) { diff --git a/util/time_test.go b/util/time_test.go index 8e49614d..6602723b 100644 --- a/util/time_test.go +++ b/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)