ARCO-184: read and process callbacks from db on callbacker start

cmd/arc/services/callbacker.go

@@ -79,6 +79,13 @@ func StartCallbacker(logger *slog.Logger, appConfig *config.ArcConfig) (func(),
 	}
 
 	dispatcher = callbacker.NewCallbackDispatcher(sender, store, config.Pause)
+	logger.Info("Init callback dispatcher, add to processing abandoned callbacks")
+	err = dispatcher.Init()
+	if err != nil {
+		stopFn()
+		return nil, fmt.Errorf("failed to init callback dispatcher, couldn't process all abandoned callbacks: %v", err)
+	}
+
 	server = callbacker.NewServer(dispatcher, callbacker.WithLogger(logger.With(slog.String("module", "server"))))
 	err = server.Serve(config.ListenAddr, appConfig.GrpcMessageSize, appConfig.PrometheusEndpoint)
 	if err != nil {
@@ -91,6 +98,7 @@ func StartCallbacker(logger *slog.Logger, appConfig *config.ArcConfig) (func(),
 		stopFn()
 		return nil, fmt.Errorf("failed to start health server: %v", err)
 	}
+
 	logger.Info("Ready to work")
 	return stopFn, nil
 }

internal/callbacker/dispatcher.go

@@ -50,7 +50,7 @@ func (d *CallbackDispatcher) Init() error {
 
 	for {
 		callbacks, err := d.s.PopMany(ctx, bundleSize)
-		if err != nil || len(callbacks) > 0 {
+		if err != nil || len(callbacks) == 0 {
 			return err
 		}
 

internal/callbacker/dispatcher_test.go

@@ -3,6 +3,7 @@ package callbacker
 import (
 	"context"
 	"fmt"
+	"math"
 	"sync"
 	"testing"
 	"time"
@@ -91,6 +92,57 @@ func Test_CallbackDispatcher(t *testing.T) {
 	}
 }
 
+func Test_CallbackDispatcher_Init(t *testing.T) {
+	tcs := []struct {
+		name                 string
+		danglingCallbacksNum int
+	}{
+		{
+			name:                 "no dangling callbacks",
+			danglingCallbacksNum: 0,
+		},
+		{
+			name:                 "callbacks to process on init",
+			danglingCallbacksNum: 259,
+		},
+	}
+
+	for _, tc := range tcs {
+		t.Run(tc.name, func(t *testing.T) {
+			// given
+			var danglingCallbacks []*store.CallbackData
+			for range tc.danglingCallbacksNum {
+				danglingCallbacks = append(danglingCallbacks, &store.CallbackData{})
+			}
+
+			cMq := &CallbackerIMock{
+				SendFunc: func(url, token string, callback *Callback) {},
+			}
+
+			sMq := &mocks.CallbackerStoreMock{
+				PopManyFunc: func(ctx context.Context, limit int) ([]*store.CallbackData, error) {
+					limit = int(math.Min(float64(len(danglingCallbacks)), float64(limit)))
+
+					r := danglingCallbacks[:limit]
+					danglingCallbacks = danglingCallbacks[limit:]
+
+					return r, nil
+				},
+			}
+
+			sut := NewCallbackDispatcher(cMq, sMq, 0)
+
+			// when
+			err := sut.Init()
+			time.Sleep(50 * time.Millisecond)
+
+			// then
+			require.NoError(t, err)
+			require.Equal(t, tc.danglingCallbacksNum, len(cMq.SendCalls()))
+		})
+	}
+}
+
 func Test_sendManager(t *testing.T) {
 	tcs := []struct {
 		name         string