feat: per-user heartbeats count metrics

config.default.yml

@@ -10,7 +10,7 @@ server:
 
 app:
   aggregation_time: '02:15' # time at which to run daily aggregation batch jobs
-  counting_time: '05:15'    # time at which to run daily job to count total hours tracked in the system
+  inactive_days: 7          # time of previous days within a user must have logged in to be considered active
   custom_languages:
     vue: Vue
     jsx: JSX

config/config.go

@@ -34,7 +34,7 @@ const (
 	SimpleDateFormat     = "2006-01-02"
 	SimpleDateTimeFormat = "2006-01-02 15:04:05"
 
-	ErrInternalServerError = "internal server error"
+	ErrInternalServerError = "500 internal server error"
 )
 
 const (
@@ -54,6 +54,7 @@ type appConfig struct {
 	AggregationTime  string                       `yaml:"aggregation_time" default:"02:15" env:"WAKAPI_AGGREGATION_TIME"`
 	ImportBackoffMin int                          `yaml:"import_backoff_min" default:"5" env:"WAKAPI_IMPORT_BACKOFF_MIN"`
 	ImportBatchSize  int                          `yaml:"import_batch_size" default:"100" env:"WAKAPI_IMPORT_BATCH_SIZE"`
+	InactiveDays     int                          `yaml:"inactive_days" default:"7" env:"WAKAPI_INACTIVE_DAYS"`
 	CustomLanguages  map[string]string            `yaml:"custom_languages"`
 	Colors           map[string]map[string]string `yaml:"-"`
 }

models/summary.go

@@ -92,6 +92,10 @@ func (s *Summary) MappedItems() map[uint8]*SummaryItems {
 	}
 }
 
+func (s *Summary) ItemsByType(summaryType uint8) *SummaryItems {
+	return s.MappedItems()[summaryType]
+}
+
 /* Augments the summary in a way that at least one item is present for every type.
 If a summary has zero items for a given type, but one or more for any of the other types,
 the total summary duration can be derived from those and inserted as a dummy-item with key "unknown"

repositories/repositories.go

@@ -49,6 +49,7 @@ type IUserRepository interface {
 	GetById(string) (*models.User, error)
 	GetByApiKey(string) (*models.User, error)
 	GetAll() ([]*models.User, error)
+	GetByLoggedInAfter(time.Time) ([]*models.User, error)
 	Count() (int64, error)
 	InsertOrGet(*models.User) (*models.User, bool, error)
 	Update(*models.User) (*models.User, error)

repositories/user.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"github.com/muety/wakapi/models"
 	"gorm.io/gorm"
+	"time"
 )
 
 type UserRepository struct {
@@ -40,6 +41,16 @@ func (r *UserRepository) GetAll() ([]*models.User, error) {
 	return users, nil
 }
 
+func (r *UserRepository) GetByLoggedInAfter(t time.Time) ([]*models.User, error) {
+	var users []*models.User
+	if err := r.db.
+		Where("last_logged_in_at >= ?", t).
+		Find(&users).Error; err != nil {
+		return nil, err
+	}
+	return users, nil
+}
+
 func (r *UserRepository) Count() (int64, error) {
 	var count int64
 	if err := r.db.

routes/api/metrics.go

@@ -1,6 +1,7 @@
 package api
 
 import (
+	"errors"
 	"github.com/emvi/logbuch"
 	"github.com/gorilla/mux"
 	conf "github.com/muety/wakapi/config"
@@ -10,15 +11,16 @@ import (
 	mm "github.com/muety/wakapi/models/metrics"
 	"github.com/muety/wakapi/services"
 	"github.com/muety/wakapi/utils"
+	"go.uber.org/atomic"
 	"net/http"
 	"sort"
-	"strconv"
 	"time"
 )
 
 const (
 	MetricsPrefix = "wakatime"
 
+	DescHeartbeats       = "Total number of tracked heartbeats."
 	DescAllTime          = "Total seconds (all time)."
 	DescTotal            = "Total seconds."
 	DescEditors          = "Total seconds for each editor."
@@ -27,9 +29,10 @@ const (
 	DescOperatingSystems = "Total seconds for each operating system."
 	DescMachines         = "Total seconds for each machine."
 
-	DescAdminTotalTime       = "Total seconds (all users, all time)"
+	DescAdminTotalTime       = "Total seconds (all users, all time)."
-	DescAdminTotalHeartbeats = "Total number of tracked heartbeats (all users, all time)"
+	DescAdminTotalHeartbeats = "Total number of tracked heartbeats (all time)."
-	DescAdminTotalUser       = "Total number of registered users"
+	DescAdminTotalUsers      = "Total number of registered users."
+	DescAdminActiveUsers     = "Number of active users."
 )
 
 type MetricsHandler struct {
@@ -65,26 +68,61 @@ func (h *MetricsHandler) RegisterRoutes(router *mux.Router) {
 }
 
 func (h *MetricsHandler) Get(w http.ResponseWriter, r *http.Request) {
-	var metrics mm.Metrics
+	reqUser := r.Context().Value(models.UserKey).(*models.User)
-
+	if reqUser == nil {
-	user := r.Context().Value(models.UserKey).(*models.User)
-	if user == nil {
 		w.WriteHeader(http.StatusUnauthorized)
 		return
 	}
 
+	var metrics mm.Metrics
+
+	if userMetrics, err := h.getUserMetrics(reqUser); err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		w.Write([]byte(conf.ErrInternalServerError))
+	} else {
+		for _, m := range *userMetrics {
+			metrics = append(metrics, m)
+		}
+	}
+
+	if reqUser.IsAdmin {
+		if adminMetrics, err := h.getAdminMetrics(reqUser); err != nil {
+			w.WriteHeader(http.StatusInternalServerError)
+			w.Write([]byte(conf.ErrInternalServerError))
+		} else {
+			for _, m := range *adminMetrics {
+				metrics = append(metrics, m)
+			}
+		}
+	}
+
+	sort.Sort(metrics)
+
+	w.Header().Set("content-type", "text/plain; charset=utf-8")
+	w.Write([]byte(metrics.Print()))
+}
+
+func (h *MetricsHandler) getUserMetrics(user *models.User) (*mm.Metrics, error) {
+	var metrics mm.Metrics
+
 	summaryAllTime, err := h.summarySrvc.Aliased(time.Time{}, time.Now(), user, h.summarySrvc.Retrieve)
 	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
+		logbuch.Error("failed to retrieve all time summary for user '%s' for metric", user.ID)
-		return
+		return nil, err
 	}
 
 	from, to := utils.MustResolveIntervalRaw("today")
 
 	summaryToday, err := h.summarySrvc.Aliased(from, to, user, h.summarySrvc.Retrieve)
 	if err != nil {
-		w.WriteHeader(http.StatusInternalServerError)
+		logbuch.Error("failed to retrieve today's summary for user '%s' for metric", user.ID)
-		return
+		return nil, err
+	}
+
+	heartbeatCount, err := h.heartbeatSrvc.CountByUser(user)
+	if err != nil {
+		logbuch.Error("failed to count heartbeats for user '%s' for metric", user.ID)
+		return nil, err
 	}
 
 	// User Metrics
@@ -103,6 +141,13 @@ func (h *MetricsHandler) Get(w http.ResponseWriter, r *http.Request) {
 		Labels: []mm.Label{},
 	})
 
+	metrics = append(metrics, &mm.CounterMetric{
+		Name:   MetricsPrefix + "_heartbeats_total",
+		Desc:   DescHeartbeats,
+		Value:  int(heartbeatCount),
+		Labels: []mm.Label{},
+	})
+
 	for _, p := range summaryToday.Projects {
 		metrics = append(metrics, &mm.CounterMetric{
 			Name:   MetricsPrefix + "_project_seconds_total",
@@ -148,61 +193,79 @@ func (h *MetricsHandler) Get(w http.ResponseWriter, r *http.Request) {
 		})
 	}
 
-	// Admin metrics
+	return &metrics, nil
+}
 
-	if user.IsAdmin {
+func (h *MetricsHandler) getAdminMetrics(user *models.User) (*mm.Metrics, error) {
-		var (
+	var metrics mm.Metrics
-			totalSeconds    int
-			totalUsers      int
-			totalHeartbeats int
-		)
 
-		if t, err := h.keyValueSrvc.GetString(conf.KeyLatestTotalTime); err == nil && t != nil && t.Value != "" {
+	if !user.IsAdmin {
-			if d, err := time.ParseDuration(t.Value); err == nil {
+		return nil, errors.New("unauthorized")
-				totalSeconds = int(d.Seconds())
+	}
+
+	var totalSeconds int
+	if t, err := h.keyValueSrvc.GetString(conf.KeyLatestTotalTime); err == nil && t != nil && t.Value != "" {
+		if d, err := time.ParseDuration(t.Value); err == nil {
+			totalSeconds = int(d.Seconds())
+		}
+	}
+
+	totalUsers, _ := h.userSrvc.Count()
+
+	activeUsers, err := h.userSrvc.GetActive()
+	if err != nil {
+		logbuch.Error("failed to retrieve active users for metric", err)
+		return nil, err
+	}
+
+	metrics = append(metrics, &mm.CounterMetric{
+		Name:   MetricsPrefix + "_admin_seconds_total",
+		Desc:   DescAdminTotalTime,
+		Value:  totalSeconds,
+		Labels: []mm.Label{},
+	})
+
+	metrics = append(metrics, &mm.CounterMetric{
+		Name:   MetricsPrefix + "_admin_users_total",
+		Desc:   DescAdminTotalUsers,
+		Value:  int(totalUsers),
+		Labels: []mm.Label{},
+	})
+
+	metrics = append(metrics, &mm.CounterMetric{
+		Name:   MetricsPrefix + "_admin_users_active_total",
+		Desc:   DescAdminActiveUsers,
+		Value:  len(activeUsers),
+		Labels: []mm.Label{},
+	})
+
+	// Count per-user heartbeats
+	type userCount struct {
+		user  string
+		count int64
+	}
+
+	i := atomic.NewUint32(uint32(len(activeUsers)))
+	c := make(chan *userCount, len(activeUsers))
+
+	for _, u := range activeUsers {
+		go func(u *models.User) {
+			count, _ := h.heartbeatSrvc.CountByUser(u)
+			c <- &userCount{user: u.ID, count: count}
+			if i.Dec() == 0 {
+				close(c)
 			}
-		}
+		}(u)
-
+	}
-		if t, err := h.keyValueSrvc.GetString(conf.KeyLatestTotalUsers); err == nil && t != nil && t.Value != "" {
-			if d, err := strconv.Atoi(t.Value); err == nil {
-				totalUsers = d
-			}
-		}
-
-		if t, err := h.keyValueSrvc.GetString(conf.KeyLatestTotalUsers); err == nil && t != nil && t.Value != "" {
-			if d, err := strconv.Atoi(t.Value); err == nil {
-				totalUsers = d
-			}
-		}
-
-		if t, err := h.heartbeatSrvc.Count(); err == nil {
-			totalHeartbeats = int(t)
-		}
-
-		metrics = append(metrics, &mm.CounterMetric{
-			Name:   MetricsPrefix + "_admin_seconds_total",
-			Desc:   DescAdminTotalTime,
-			Value:  totalSeconds,
-			Labels: []mm.Label{},
-		})
-
-		metrics = append(metrics, &mm.CounterMetric{
-			Name:   MetricsPrefix + "_admin_users_total",
-			Desc:   DescAdminTotalUser,
-			Value:  totalUsers,
-			Labels: []mm.Label{},
-		})
 
+	for uc := range c {
 		metrics = append(metrics, &mm.CounterMetric{
 			Name:   MetricsPrefix + "_admin_heartbeats_total",
 			Desc:   DescAdminTotalHeartbeats,
-			Value:  totalHeartbeats,
+			Value:  int(uc.count),
-			Labels: []mm.Label{},
+			Labels: []mm.Label{{Key: "user", Value: uc.user}},
 		})
 	}
 
-	sort.Sort(metrics)
+	return &metrics, nil
-
-	w.Header().Set("content-type", "text/plain; charset=utf-8")
-	w.Write([]byte(metrics.Print()))
 }

services/services.go

@@ -64,6 +64,7 @@ type IUserService interface {
 	GetUserById(string) (*models.User, error)
 	GetUserByKey(string) (*models.User, error)
 	GetAll() ([]*models.User, error)
+	GetActive() ([]*models.User, error)
 	Count() (int64, error)
 	CreateOrGet(*models.Signup, bool) (*models.User, bool, error)
 	Update(*models.User) (*models.User, error)

services/user.go

@@ -56,6 +56,12 @@ func (srv *UserService) GetAll() ([]*models.User, error) {
 	return srv.repository.GetAll()
 }
 
+func (srv *UserService) GetActive() ([]*models.User, error) {
+	// a user is considered active if she has logged in to the web interface at least once within the last x days
+	minDate := time.Now().Add(-24 * time.Hour * time.Duration(srv.Config.App.InactiveDays))
+	return srv.repository.GetByLoggedInAfter(minDate)
+}
+
 func (srv *UserService) Count() (int64, error) {
 	return srv.repository.Count()
 }