1
0
mirror of https://github.com/muety/wakapi.git synced 2023-08-10 21:12:56 +03:00
wakapi/services/aggregation.go

189 lines
5.5 KiB
Go
Raw Normal View History

package services
import (
"errors"
datastructure "github.com/duke-git/lancet/v2/datastructure/set"
"github.com/emvi/logbuch"
2022-12-01 17:31:19 +03:00
"github.com/muety/artifex/v2"
"github.com/muety/wakapi/config"
"sync"
"time"
2020-03-31 13:22:17 +03:00
"github.com/muety/wakapi/models"
)
2019-10-10 17:47:19 +03:00
const (
2020-10-09 22:37:16 +03:00
aggregateIntervalDays int = 1
2019-10-10 17:47:19 +03:00
)
2021-04-30 15:07:14 +03:00
var aggregationLock = sync.Mutex{}
type AggregationService struct {
config *config.Config
userService IUserService
summaryService ISummaryService
heartbeatService IHeartbeatService
inProgress datastructure.Set[string]
queueDefault *artifex.Dispatcher
queueWorkers *artifex.Dispatcher
}
func NewAggregationService(userService IUserService, summaryService ISummaryService, heartbeatService IHeartbeatService) *AggregationService {
return &AggregationService{
config: config.Get(),
userService: userService,
summaryService: summaryService,
heartbeatService: heartbeatService,
inProgress: datastructure.NewSet[string](),
queueDefault: config.GetDefaultQueue(),
queueWorkers: config.GetQueue(config.QueueProcessing),
}
}
type AggregationJob struct {
2019-10-10 17:47:19 +03:00
UserID string
From time.Time
To time.Time
}
2019-10-11 10:06:34 +03:00
// Schedule a job to (re-)generate summaries every day shortly after midnight
func (srv *AggregationService) Schedule() {
logbuch.Info("scheduling summary aggregation")
2019-10-10 17:47:19 +03:00
if _, err := srv.queueDefault.DispatchCron(func() {
if err := srv.AggregateSummaries(datastructure.NewSet[string]()); err != nil {
config.Log().Error("failed to generate summaries, %v", err)
2019-10-10 17:47:19 +03:00
}
}, srv.config.App.GetAggregationTimeCron()); err != nil {
config.Log().Error("failed to schedule summary generation, %v", err)
2019-10-10 17:47:19 +03:00
}
}
func (srv *AggregationService) AggregateSummaries(userIds datastructure.Set[string]) error {
if err := srv.lockUsers(userIds); err != nil {
return err
2019-10-10 17:47:19 +03:00
}
defer srv.unlockUsers(userIds)
2019-10-10 17:47:19 +03:00
logbuch.Info("generating summaries")
2019-10-11 10:06:34 +03:00
// Get a map from user ids to the time of their latest summary or nil if none exists yet
lastUserSummaryTimes, err := srv.summaryService.GetLatestByUser()
if err != nil {
2021-04-16 16:59:39 +03:00
config.Log().Error(err.Error())
2019-10-10 17:47:19 +03:00
return err
}
// Get a map from user ids to the time of their earliest heartbeats or nil if none exists yet
firstUserHeartbeatTimes, err := srv.heartbeatService.GetFirstByUsers()
if err != nil {
2021-04-16 16:59:39 +03:00
config.Log().Error(err.Error())
2019-10-10 17:47:19 +03:00
return err
}
// Build actual lookup table from it
firstUserHeartbeatLookup := make(map[string]models.CustomTime)
for _, e := range firstUserHeartbeatTimes {
firstUserHeartbeatLookup[e.User] = e.Time
}
// Dispatch summary generation jobs
jobs := make(chan *AggregationJob)
defer close(jobs)
go func() {
for jobRef := range jobs {
job := *jobRef
if err := srv.queueWorkers.Dispatch(func() {
srv.process(job)
}); err != nil {
config.Log().Error("failed to dispatch summary generation job for user '%s'", job.UserID)
}
}
}()
// Generate summary aggregation jobs
for _, e := range lastUserSummaryTimes {
if userIds != nil && !userIds.IsEmpty() && !userIds.Contain(e.User) {
continue
}
if e.Time.Valid() {
// Case 1: User has aggregated summaries already
// -> Spawn jobs to create summaries from their latest aggregation to now
generateUserJobs(e.User, e.Time.T(), jobs)
} else if t := firstUserHeartbeatLookup[e.User]; t.Valid() {
// Case 2: User has no aggregated summaries, yet, but has heartbeats
// -> Spawn jobs to create summaries from their first heartbeat to now
generateUserJobs(e.User, t.T(), jobs)
}
// Case 3: User doesn't have heartbeats at all
// -> Nothing to do
2019-10-10 17:47:19 +03:00
}
2019-10-10 17:47:19 +03:00
return nil
}
func (srv *AggregationService) process(job AggregationJob) {
if summary, err := srv.summaryService.Summarize(job.From, job.To, &models.User{ID: job.UserID}, nil); err != nil {
config.Log().Error("failed to generate summary (%v, %v, %s) - %v", job.From, job.To, job.UserID, err)
} else {
logbuch.Info("successfully generated summary (%v, %v, %s)", job.From, job.To, job.UserID)
if err := srv.summaryService.Insert(summary); err != nil {
config.Log().Error("failed to save summary (%v, %v, %s) - %v", summary.UserID, summary.FromTime, summary.ToTime, err)
}
}
}
func generateUserJobs(userId string, from time.Time, jobs chan<- *AggregationJob) {
var to time.Time
2019-10-10 17:47:19 +03:00
// Go to next day of either user's first heartbeat or latest aggregation
from = from.Add(-1 * time.Second)
from = time.Date(
from.Year(),
from.Month(),
from.Day()+aggregateIntervalDays,
0, 0, 0, 0,
from.Location(),
)
// Iteratively aggregate per-day summaries until end of yesterday is reached
end := getStartOfToday().Add(-1 * time.Second)
2019-10-10 17:47:19 +03:00
for from.Before(end) && to.Before(end) {
to = time.Date(
from.Year(),
from.Month(),
2020-03-09 19:30:23 +03:00
from.Day()+aggregateIntervalDays,
0, 0, 0, 0,
from.Location(),
)
2019-10-10 17:47:19 +03:00
jobs <- &AggregationJob{userId, from, to}
from = to
}
}
func (srv *AggregationService) lockUsers(userIds datastructure.Set[string]) error {
aggregationLock.Lock()
defer aggregationLock.Unlock()
for uid := range userIds {
if srv.inProgress.Contain(uid) {
return errors.New("aggregation already in progress for at least of the request users")
}
}
srv.inProgress = srv.inProgress.Union(userIds)
return nil
}
func (srv *AggregationService) unlockUsers(userIds datastructure.Set[string]) {
aggregationLock.Lock()
defer aggregationLock.Unlock()
for uid := range userIds {
srv.inProgress.Delete(uid)
}
}
2019-10-10 17:47:19 +03:00
func getStartOfToday() time.Time {
now := time.Now()
return time.Date(now.Year(), now.Month(), now.Day(), 0, 0, 0, 1, now.Location())
}