feat: keyed queue

bot.go

@@ -5,10 +5,10 @@ import (
 	"path"
 	"sync"
 
-	"github.com/Gasoid/mergebot/config"
-	"github.com/Gasoid/mergebot/handlers"
-	"github.com/Gasoid/mergebot/logger"
-	"github.com/Gasoid/mergebot/webhook"
+	"github.com/Gasoid/merge-bot/config"
+	"github.com/Gasoid/merge-bot/handlers"
+	"github.com/Gasoid/merge-bot/logger"
+	"github.com/Gasoid/merge-bot/webhook"
 
 	"net/http"
 
@@ -63,7 +63,7 @@ func start() {
 	}
 
 	e.GET(HealthyEndpoint, healthcheck)
-	e.POST("/mergebot/webhook/:provider/", Handler)
+	e.POST("/merge-bot/webhook/:provider/", Handler)
 
 	if tlsEnabled {
 		tmpDir := path.Join(os.TempDir(), "tls", ".cache")

bot_test.go

@@ -7,7 +7,7 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/Gasoid/mergebot/webhook"
+	"github.com/Gasoid/merge-bot/webhook"
 
 	"github.com/labstack/echo/v4"
 	"github.com/stretchr/testify/assert"
@@ -107,7 +107,7 @@ func TestHandler(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			e := echo.New()
-			req := httptest.NewRequest(http.MethodPost, "/mergebot/webhook/"+tt.provider+"/", strings.NewReader(tt.body))
+			req := httptest.NewRequest(http.MethodPost, "/merge-bot/webhook/"+tt.provider+"/", strings.NewReader(tt.body))
 			req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
 			rec := httptest.NewRecorder()
 			c := e.NewContext(req, rec)
@@ -139,7 +139,7 @@ func TestHandlerConcurrency(t *testing.T) {
 	// Test multiple concurrent requests
 	for i := 0; i < numRequests; i++ {
 		go func() {
-			req := httptest.NewRequest(http.MethodPost, "/mergebot/webhook/concurrent/", strings.NewReader(`{}`))
+			req := httptest.NewRequest(http.MethodPost, "/merge-bot/webhook/concurrent/", strings.NewReader(`{}`))
 			req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
 			rec := httptest.NewRecorder()
 			c := e.NewContext(req, rec)
@@ -175,7 +175,7 @@ func TestHandlerWithDifferentMethods(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.method, func(t *testing.T) {
 			e := echo.New()
-			req := httptest.NewRequest(tt.method, "/mergebot/webhook/methodtest/", strings.NewReader(`{}`))
+			req := httptest.NewRequest(tt.method, "/merge-bot/webhook/methodtest/", strings.NewReader(`{}`))
 			req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
 			rec := httptest.NewRecorder()
 			c := e.NewContext(req, rec)
@@ -194,7 +194,7 @@ func TestHandlerWithEmptyBody(t *testing.T) {
 	defer cleanup()
 
 	e := echo.New()
-	req := httptest.NewRequest(http.MethodPost, "/mergebot/webhook/emptybody/", bytes.NewReader([]byte{}))
+	req := httptest.NewRequest(http.MethodPost, "/merge-bot/webhook/emptybody/", bytes.NewReader([]byte{}))
 	req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
 	rec := httptest.NewRecorder()
 	c := e.NewContext(req, rec)

commands.go

@@ -3,9 +3,9 @@ package main
 import (
 	"fmt"
 
-	"github.com/Gasoid/mergebot/handlers"
-	"github.com/Gasoid/mergebot/logger"
-	"github.com/Gasoid/mergebot/webhook"
+	"github.com/Gasoid/merge-bot/handlers"
+	"github.com/Gasoid/merge-bot/logger"
+	"github.com/Gasoid/merge-bot/webhook"
 )
 
 func init() {

commands_test.go

@@ -3,8 +3,8 @@ package main
 import (
 	"testing"
 
-	"github.com/Gasoid/mergebot/handlers"
-	"github.com/Gasoid/mergebot/webhook"
+	"github.com/Gasoid/merge-bot/handlers"
+	"github.com/Gasoid/merge-bot/webhook"
 	"github.com/stretchr/testify/assert"
 )
 

go.mod

@@ -1,4 +1,4 @@
-module github.com/Gasoid/mergebot
+module github.com/Gasoid/merge-bot
 
 go 1.24.1
 

handlers/gitlab/gitlab.go

@@ -6,9 +6,9 @@ import (
 	"net/http"
 	"slices"
 
-	"github.com/Gasoid/mergebot/config"
-	"github.com/Gasoid/mergebot/handlers"
-	"github.com/Gasoid/mergebot/logger"
+	"github.com/Gasoid/merge-bot/config"
+	"github.com/Gasoid/merge-bot/handlers"
+	"github.com/Gasoid/merge-bot/logger"
 	gitlab "gitlab.com/gitlab-org/api/client-go"
 
 	"github.com/dustin/go-humanize"

handlers/merge.go

@@ -5,7 +5,7 @@ import (
 	"net/url"
 	"os"
 
-	"github.com/Gasoid/mergebot/logger"
+	"github.com/Gasoid/merge-bot/logger"
 
 	"github.com/ldez/go-git-cmd-wrapper/v2/checkout"
 	"github.com/ldez/go-git-cmd-wrapper/v2/clone"

handlers/request.go

@@ -6,7 +6,8 @@ import (
 	"html/template"
 	"strings"
 
-	"github.com/Gasoid/mergebot/logger"
+	"github.com/Gasoid/merge-bot/logger"
+	"github.com/Gasoid/merge-bot/semaphore"
 
 	"gopkg.in/yaml.v3"
 )
@@ -18,6 +19,11 @@ const (
 	staleLabelColor      = "#cccccc"
 )
 
+var (
+	deleteStaleBranches = semaphore.NewKeyedSemaphore(1)
+	updateBranch        = semaphore.NewKeyedSemaphore(2)
+)
+
 type Request struct {
 	provider RequestProvider
 	info     *MrInfo
@@ -126,15 +132,22 @@ func (r *Request) Greetings() error {
 }
 
 func (r *Request) DeleteStaleBranches() error {
-	if r.config.StaleBranchesDeletion.Enabled {
+
+	if !r.config.StaleBranchesDeletion.Enabled {
+		return nil
+	}
+
+	deleteStaleBranches.Add(fmt.Sprintf("clean_stale_merge_requests_%d", r.info.ProjectId), func() {
 		if err := r.cleanStaleMergeRequests(); err != nil {
-			return err
+			logger.Info("cleanStaleMergeRequests", "err", err)
 		}
+	})
 
+	deleteStaleBranches.Add(fmt.Sprintf("clean_stale_branches_%d", r.info.ProjectId), func() {
 		if err := r.cleanStaleBranches(); err != nil {
-			return err
+			logger.Info("cleanStaleBranches", "err", err)
 		}
-	}
+	})
 
 	return nil
 }
@@ -171,9 +184,11 @@ func (r Request) UpdateBranches() error {
 	}
 
 	for _, mr := range listMr {
-		if err := r.provider.UpdateFromMaster(r.info.ProjectId, mr.Id); err != nil {
-			return err
-		}
+		updateBranch.Add(fmt.Sprintf("update_branch_%d_%d", r.info.ProjectId, mr.Id), func() {
+			if err := r.provider.UpdateFromMaster(r.info.ProjectId, mr.Id); err != nil {
+				logger.Info("UpdateFromMaster", "err", err)
+			}
+		})
 	}
 
 	return nil

handlers/stalebranches.go

@@ -2,14 +2,9 @@ package handlers
 
 import (
 	"fmt"
-	"sync"
 	"time"
 
-	"github.com/Gasoid/mergebot/logger"
-)
-
-var (
-	cleanStaleBranchesLock sync.Mutex
+	"github.com/Gasoid/merge-bot/logger"
 )
 
 type StaleBranch struct {
@@ -18,9 +13,6 @@ type StaleBranch struct {
 }
 
 func (r Request) cleanStaleBranches() error {
-	cleanStaleBranchesLock.Lock()
-	defer cleanStaleBranchesLock.Unlock()
-
 	logger.Debug("deletion of stale branches has been run")
 
 	candidates, err := r.provider.ListBranches(r.info.ProjectId, r.config.StaleBranchesDeletion.BatchSize)

handlers/stalemergerequests.go

@@ -3,14 +3,9 @@ package handlers
 import (
 	"fmt"
 	"slices"
-	"sync"
 	"time"
 )
 
-var (
-	cleanStaleMergeRquestsLock sync.Mutex
-)
-
 type MR struct {
 	Id          int
 	Branch      string
@@ -19,9 +14,6 @@ type MR struct {
 }
 
 func (r Request) cleanStaleMergeRequests() error {
-	cleanStaleMergeRquestsLock.Lock()
-	defer cleanStaleMergeRquestsLock.Unlock()
-
 	days := r.config.StaleBranchesDeletion.Days
 	coolDays := r.config.StaleBranchesDeletion.WaitDays
 	now := time.Now()

integration_test.go

@@ -7,8 +7,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/Gasoid/mergebot/handlers"
-	"github.com/Gasoid/mergebot/webhook"
+	"github.com/Gasoid/merge-bot/handlers"
+	"github.com/Gasoid/merge-bot/webhook"
 	"github.com/labstack/echo/v4"
 	"github.com/stretchr/testify/assert"
 )
@@ -77,7 +77,7 @@ func TestIntegrationWebhookFlow(t *testing.T) {
 	})
 
 	e := echo.New()
-	req := httptest.NewRequest(http.MethodPost, "/mergebot/webhook/integration/", strings.NewReader(`{}`))
+	req := httptest.NewRequest(http.MethodPost, "/merge-bot/webhook/integration/", strings.NewReader(`{}`))
 	req.Header.Set(echo.HeaderContentType, echo.MIMEApplicationJSON)
 	rec := httptest.NewRecorder()
 	c := e.NewContext(req, rec)

logger/logger.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"log/slog"
 
-	"github.com/Gasoid/mergebot/config"
+	"github.com/Gasoid/merge-bot/config"
 
 	"github.com/getsentry/sentry-go"
 	sentryslog "github.com/getsentry/sentry-go/slog"

main.go

@@ -1,10 +1,10 @@
 package main
 
 import (
-	"github.com/Gasoid/mergebot/config"
-	_ "github.com/Gasoid/mergebot/handlers/gitlab"
-	"github.com/Gasoid/mergebot/logger"
-	_ "github.com/Gasoid/mergebot/webhook/gitlab"
+	"github.com/Gasoid/merge-bot/config"
+	_ "github.com/Gasoid/merge-bot/handlers/gitlab"
+	"github.com/Gasoid/merge-bot/logger"
+	_ "github.com/Gasoid/merge-bot/webhook/gitlab"
 )
 
 func main() {

semaphore/semaphore.go

@@ -0,0 +1,78 @@
+package semaphore
+
+import (
+	"fmt"
+	"sync"
+)
+
+type keyState struct {
+	mu      sync.RWMutex
+	running int
+}
+
+// counting semaphore
+type KeyedSemaphore struct {
+	counters  map[string]*keyState
+	maxPerKey int
+	mu        sync.RWMutex
+}
+
+func NewKeyedSemaphore(maxPerKey int) *KeyedSemaphore {
+	return &KeyedSemaphore{
+		counters:  map[string]*keyState{},
+		maxPerKey: maxPerKey,
+	}
+}
+
+func (s *KeyedSemaphore) Print() {
+	s.mu.RLock()
+	defer s.mu.RUnlock()
+
+	fmt.Printf("length: %d", len(s.counters))
+
+	for k := range s.counters {
+		fmt.Printf("key: %s", k)
+	}
+}
+
+func (s *KeyedSemaphore) Add(key string, task func()) bool {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if state, ok := s.counters[key]; ok {
+		if state.running == s.maxPerKey {
+			return false
+		}
+		state.running++
+	} else {
+		s.counters[key] = &keyState{
+			running: 1,
+		}
+	}
+
+	go s.run(key, s.counters[key], task)
+	return true
+}
+
+func (s *KeyedSemaphore) clean(key string) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
+	if state, ok := s.counters[key]; ok {
+		if state.running == 0 {
+			delete(s.counters, key)
+		}
+	}
+}
+
+func (s *KeyedSemaphore) run(key string, state *keyState, task func()) {
+	state.mu.Lock()
+	defer state.mu.Unlock()
+	defer s.clean(key)
+
+	defer func() {
+		state.running--
+	}()
+
+	task()
+}

webhook/gitlab/gitlab.go

@@ -5,8 +5,8 @@ import (
 	"net/http"
 	"strings"
 
-	"github.com/Gasoid/mergebot/logger"
-	"github.com/Gasoid/mergebot/webhook"
+	"github.com/Gasoid/merge-bot/logger"
+	"github.com/Gasoid/merge-bot/webhook"
 	gitlab "gitlab.com/gitlab-org/api/client-go"
 )