fix: deadlock caused by not using open transaction during batch insert

fix: regenerate durations upon language mapping update
fix: minor ui

Author: muety

config/eventbus.go

@@ -8,18 +8,19 @@ type ApplicationEvent struct {
 }
 
 const (
-	TopicUser               = "user.*"
-	TopicHeartbeat          = "heartbeat.*"
-	TopicProjectLabel       = "project_label.*"
-	EventUserUpdate         = "user.update"
-	EventUserDelete         = "user.delete"
-	EventHeartbeatCreate    = "heartbeat.create"
-	EventProjectLabelCreate = "project_label.create"
-	EventProjectLabelDelete = "project_label.delete"
-	EventWakatimeFailure    = "wakatime.failure"
-	FieldPayload            = "payload"
-	FieldUser               = "user"
-	FieldUserId             = "user.id"
+	TopicUser                    = "user.*"
+	TopicHeartbeat               = "heartbeat.*"
+	TopicProjectLabel            = "project_label.*"
+	EventUserUpdate              = "user.update"
+	EventUserDelete              = "user.delete"
+	EventHeartbeatCreate         = "heartbeat.create"
+	EventProjectLabelCreate      = "project_label.create"
+	EventProjectLabelDelete      = "project_label.delete"
+	EventWakatimeFailure         = "wakatime.failure"
+	EventLanguageMappingsChanged = "language_mappings.changed"
+	FieldPayload                 = "payload"
+	FieldUser                    = "user"
+	FieldUserId                  = "user.id"
 )
 
 var eventHub *hub.Hub

coverage/coverage.out

@@ -144,6 +144,10 @@ func (u *User) HeartbeatsTimeout() time.Duration {
 	return DefaultHeartbeatsTimeout
 }
 
+func (u *User) HeartbeatsTimeoutMin() int {
+	return int(u.HeartbeatsTimeout() / time.Minute)
+}
+
 // WakaTimeURL returns the user's effective WakaTime URL, i.e. a custom one (which could also point to another Wakapi instance) or fallback if not specified otherwise.
 func (u *User) WakaTimeURL(fallback string) string {
 	if u.WakatimeApiUrl != "" {

models/user.go

@@ -51,7 +51,7 @@ func InsertBatchChunked[T any](data []T, model T, db *gorm.DB) error {
 	return db.Transaction(func(tx *gorm.DB) error {
 		chunks := slice.Chunk[T](data, chunkSize)
 		for _, chunk := range chunks {
-			if err := insertBatch[T](chunk, model, db); err != nil {
+			if err := insertBatch[T](chunk, model, tx); err != nil {
 				return err
 			}
 		}

repositories/base.go

@@ -56,6 +56,23 @@ func NewDurationService(durationRepository repositories.IDurationRepository, hea
 		}
 	}(&sub1)
 
+	sub2 := srv.eventBus.Subscribe(0, config.EventLanguageMappingsChanged)
+	go func(sub *hub.Subscription) {
+		for m := range sub.Receiver {
+			userId := m.Fields[config.FieldUserId].(string)
+			user, err := srv.userService.GetUserById(userId)
+			if err != nil {
+				config.Log().Error("user not found for regenerating durations after language mapping change", "user", userId)
+				continue
+			}
+
+			slog.Info("regenerating durations because language mappings were updated", "user", userId)
+			srv.queue.Dispatch(func() {
+				srv.Regenerate(user, true)
+			})
+		}
+	}(&sub2)
+
 	return srv
 }
 

services/duration.go

@@ -2,6 +2,7 @@ package services
 
 import (
 	"errors"
+	"github.com/leandro-lugaresi/hub"
 	"github.com/muety/wakapi/config"
 	"github.com/muety/wakapi/models"
 	"github.com/muety/wakapi/repositories"
@@ -12,12 +13,14 @@ import (
 type LanguageMappingService struct {
 	config     *config.Config
 	cache      *cache.Cache
+	eventBus   *hub.Hub
 	repository repositories.ILanguageMappingRepository
 }
 
 func NewLanguageMappingService(languageMappingsRepo repositories.ILanguageMappingRepository) *LanguageMappingService {
 	return &LanguageMappingService{
 		config:     config.Get(),
+		eventBus:   config.EventBus(),
 		repository: languageMappingsRepo,
 		cache:      cache.New(24*time.Hour, 24*time.Hour),
 	}
@@ -60,6 +63,7 @@ func (srv *LanguageMappingService) Create(mapping *models.LanguageMapping) (*mod
 	}
 
 	srv.cache.Delete(result.UserID)
+	srv.notifyUpdate(mapping)
 	return result, nil
 }
 
@@ -69,10 +73,19 @@ func (srv *LanguageMappingService) Delete(mapping *models.LanguageMapping) error
 	}
 	err := srv.repository.Delete(mapping.ID)
 	srv.cache.Delete(mapping.UserID)
+	srv.notifyUpdate(mapping)
 	return err
 }
 
 func (srv *LanguageMappingService) getServerMappings() map[string]string {
 	// https://dave.cheney.net/2017/04/30/if-a-map-isnt-a-reference-variable-what-is-it
 	return srv.config.App.GetCustomLanguages()
 }
+
+func (srv *LanguageMappingService) notifyUpdate(mapping *models.LanguageMapping) {
+	name := config.EventLanguageMappingsChanged
+	srv.eventBus.Publish(hub.Message{
+		Name:   name,
+		Fields: map[string]interface{}{config.FieldPayload: mapping, config.FieldUserId: mapping.UserID},
+	})
+}

services/language_mapping.go

@@ -464,9 +464,9 @@ <h3 class="inline-block font-semibold text-gray-300">Add Rule</h3>
                     <div class="flex-col w-full md:w-2/3 inline-block space-y-4">
                         <div class="flex justify-between items-center">
                             <div class="flex flex-col flex-grow gap-y-1">
-                                <label class="font-semibold text-gray-300" for="heartbeats_timeout">Timeout / offset (seconds)</label>
+                                <label class="font-semibold text-gray-300" for="heartbeats_timeout">Timeout / offset (minutes)</label>
                                 <div class="flex gap-x-2 items-center">
-                                    <input class="input-default" type="number" id="heartbeats_timeout" name="heartbeats_timeout" style="max-width: 100px;" placeholder="1" min="1" max="60" step="1" required value="{{ .User.HeartbeatsTimeoutSec }}">
+                                    <input class="input-default" type="number" id="heartbeats_timeout" name="heartbeats_timeout" style="max-width: 100px;" placeholder="1" min="1" max="60" step="1" required value="{{ .User.HeartbeatsTimeoutMin }}">
                                     <span class="text-gray-600 text-sm">(min. 1 min, max. 60 min)</span>
                                 </div>
                             </div>