ntfy/server/server_manager.go

package server

import (
	"git.zio.sh/astra/ntfy/v2/log"
	"git.zio.sh/astra/ntfy/v2/util"
	"strings"
)

func (s *Server) execManager() {
	// WARNING: Make sure to only selectively lock with the mutex, and be aware that this
	//          there is no mutex for the entire function.

	// Prune all the things
	s.pruneVisitors()
	s.pruneTokens()
	s.pruneAttachments()
	s.pruneMessages()
	s.pruneAndNotifyWebPushSubscriptions()

	// Message count per topic
	var messagesCached int
	messageCounts, err := s.messageCache.MessageCounts()
	if err != nil {
		log.Tag(tagManager).Err(err).Warn("Cannot get message counts")
		messageCounts = make(map[string]int) // Empty, so we can continue
	}
	for _, count := range messageCounts {
		messagesCached += count
	}

	// Remove subscriptions without subscribers
	var emptyTopics, subscribers int
	log.
		Tag(tagManager).
		Timing(func() {
			s.mu.Lock()
			defer s.mu.Unlock()
			for _, t := range s.topics {
				subs, lastAccess := t.Stats()
				ev := log.Tag(tagManager).With(t)
				if t.Stale() {
					if ev.IsTrace() {
						ev.Trace("- topic %s: Deleting stale topic (%d subscribers, accessed %s)", t.ID, subs, util.FormatTime(lastAccess))
					}
					emptyTopics++
					delete(s.topics, t.ID)
				} else {
					if ev.IsTrace() {
						ev.Trace("- topic %s: %d subscribers, accessed %s", t.ID, subs, util.FormatTime(lastAccess))
					}
					subscribers += subs
				}
			}
		}).
		Debug("Removed %d empty topic(s)", emptyTopics)

	// Mail stats
	var receivedMailTotal, receivedMailSuccess, receivedMailFailure int64
	if s.smtpServerBackend != nil {
		receivedMailTotal, receivedMailSuccess, receivedMailFailure = s.smtpServerBackend.Counts()
	}
	var sentMailTotal, sentMailSuccess, sentMailFailure int64
	if s.smtpSender != nil {
		sentMailTotal, sentMailSuccess, sentMailFailure = s.smtpSender.Counts()
	}

	// Users
	var usersCount int64
	if s.userManager != nil {
		usersCount, err = s.userManager.UsersCount()
		if err != nil {
			log.Tag(tagManager).Err(err).Warn("Error counting users")
		}
	}

	// Print stats
	s.mu.RLock()
	messagesCount, topicsCount, visitorsCount := s.messages, len(s.topics), len(s.visitors)
	s.mu.RUnlock()

	// Update stats
	s.updateAndWriteStats(messagesCount)

	// Log stats
	log.
		Tag(tagManager).
		Fields(log.Context{
			"messages_published":      messagesCount,
			"messages_cached":         messagesCached,
			"topics_active":           topicsCount,
			"subscribers":             subscribers,
			"visitors":                visitorsCount,
			"users":                   usersCount,
			"emails_received":         receivedMailTotal,
			"emails_received_success": receivedMailSuccess,
			"emails_received_failure": receivedMailFailure,
			"emails_sent":             sentMailTotal,
			"emails_sent_success":     sentMailSuccess,
			"emails_sent_failure":     sentMailFailure,
		}).
		Info("Server stats")
	mset(metricMessagesCached, messagesCached)
	mset(metricVisitors, visitorsCount)
	mset(metricUsers, usersCount)
	mset(metricSubscribers, subscribers)
	mset(metricTopics, topicsCount)
}

func (s *Server) pruneVisitors() {
	staleVisitors := 0
	log.
		Tag(tagManager).
		Timing(func() {
			s.mu.Lock()
			defer s.mu.Unlock()
			for ip, v := range s.visitors {
				if v.Stale() {
					log.Tag(tagManager).With(v).Trace("Deleting stale visitor")
					delete(s.visitors, ip)
					staleVisitors++
				}
			}
		}).
		Field("stale_visitors", staleVisitors).
		Debug("Deleted %d stale visitor(s)", staleVisitors)
}

func (s *Server) pruneTokens() {
	if s.userManager != nil {
		log.
			Tag(tagManager).
			Timing(func() {
				if err := s.userManager.RemoveExpiredTokens(); err != nil {
					log.Tag(tagManager).Err(err).Warn("Error expiring user tokens")
				}
				if err := s.userManager.RemoveDeletedUsers(); err != nil {
					log.Tag(tagManager).Err(err).Warn("Error deleting soft-deleted users")
				}
			}).
			Debug("Removed expired tokens and users")
	}
}

func (s *Server) pruneAttachments() {
	if s.fileCache == nil {
		return
	}
	log.
		Tag(tagManager).
		Timing(func() {
			ids, err := s.messageCache.AttachmentsExpired()
			if err != nil {
				log.Tag(tagManager).Err(err).Warn("Error retrieving expired attachments")
			} else if len(ids) > 0 {
				if log.Tag(tagManager).IsDebug() {
					log.Tag(tagManager).Debug("Deleting attachments %s", strings.Join(ids, ", "))
				}
				if err := s.fileCache.Remove(ids...); err != nil {
					log.Tag(tagManager).Err(err).Warn("Error deleting attachments")
				}
				if err := s.messageCache.MarkAttachmentsDeleted(ids...); err != nil {
					log.Tag(tagManager).Err(err).Warn("Error marking attachments deleted")
				}
			} else {
				log.Tag(tagManager).Debug("No expired attachments to delete")
			}
		}).
		Debug("Deleted expired attachments")
}

func (s *Server) pruneMessages() {
	log.
		Tag(tagManager).
		Timing(func() {
			expiredMessageIDs, err := s.messageCache.MessagesExpired()
			if err != nil {
				log.Tag(tagManager).Err(err).Warn("Error retrieving expired messages")
			} else if len(expiredMessageIDs) > 0 {
				if s.fileCache != nil {
					if err := s.fileCache.Remove(expiredMessageIDs...); err != nil {
						log.Tag(tagManager).Err(err).Warn("Error deleting attachments for expired messages")
					}
				}
				if err := s.messageCache.DeleteMessages(expiredMessageIDs...); err != nil {
					log.Tag(tagManager).Err(err).Warn("Error marking attachments deleted")
				}
			} else {
				log.Tag(tagManager).Debug("No expired messages to delete")
			}
		}).
		Debug("Pruned messages")
}