feat: migrate probes to redis

cmd/swat4master/application/application.go

@@ -9,8 +9,8 @@ import (
 	"github.com/sergeii/swat4master/internal/core/repositories"
 	"github.com/sergeii/swat4master/internal/metrics"
 	"github.com/sergeii/swat4master/internal/persistence/memory/instances"
-	"github.com/sergeii/swat4master/internal/persistence/memory/probes"
 	"github.com/sergeii/swat4master/internal/persistence/memory/servers"
+	"github.com/sergeii/swat4master/internal/persistence/redis/probes"
 	"github.com/sergeii/swat4master/internal/validation"
 )
 

cmd/swat4master/config/config.go

@@ -14,6 +14,8 @@ type Config struct {
 	LogLevel  string
 	LogOutput string
 
+	RedisURL string
+
 	ReporterListenAddr string
 	ReporterBufferSize int
 
@@ -63,6 +65,10 @@ func Provide() Config {
 		&cfg.LogOutput, "log.output", "console",
 		"Output format of log messages. Available options: console, stdout, json",
 	)
+	flag.StringVar(
+		&cfg.RedisURL, "redis.url", "redis://localhost:6379",
+		"URL to connect to the Redis server",
+	)
 	flag.StringVar(
 		&cfg.ReporterListenAddr, "reporter.address", ":27900",
 		"Address to listen on for the reporter service",
@@ -152,7 +158,7 @@ func Provide() Config {
 		"Determines how many times a failed revival probe is retried",
 	)
 	flag.DurationVar(
-		&cfg.ProbePollSchedule, "probe.schedule", time.Millisecond*50,
+		&cfg.ProbePollSchedule, "probe.schedule", time.Millisecond*250,
 		"Defines how often the discovery queue is checked for new probes",
 	)
 	flag.DurationVar(

cmd/swat4master/main.go

@@ -19,13 +19,15 @@ import (
 	"github.com/sergeii/swat4master/cmd/swat4master/modules/refresher"
 	"github.com/sergeii/swat4master/cmd/swat4master/modules/reporter"
 	"github.com/sergeii/swat4master/cmd/swat4master/modules/reviver"
+	"github.com/sergeii/swat4master/cmd/swat4master/persistence"
 
 	"github.com/sergeii/swat4master/cmd/swat4master/application"
 )
 
 func main() {
 	app := fx.New(
 		fx.Provide(config.Provide),
+		fx.Provide(persistence.Provide),
 		application.Module,
 		refresher.Module,
 		reviver.Module,

cmd/swat4master/persistence/persistence.go

@@ -0,0 +1,37 @@
+package persistence
+
+import (
+	"context"
+
+	"github.com/redis/go-redis/v9"
+	"go.uber.org/fx"
+
+	"github.com/sergeii/swat4master/cmd/swat4master/config"
+)
+
+type Persistence struct {
+	fx.Out
+
+	RedisClient *redis.Client
+}
+
+func Provide(cfg config.Config, lc fx.Lifecycle) (Persistence, error) {
+	opts, err := redis.ParseURL(cfg.RedisURL)
+	if err != nil {
+		return Persistence{}, err
+	}
+
+	redisClient := redis.NewClient(opts)
+
+	lc.Append(fx.Hook{
+		OnStop: func(_ context.Context) error {
+			return redisClient.Close()
+		},
+	})
+
+	persistence := Persistence{
+		RedisClient: redisClient,
+	}
+
+	return persistence, nil
+}

go.mod

@@ -3,30 +3,36 @@ module github.com/sergeii/swat4master
 go 1.23
 
 require (
+	github.com/alicebob/miniredis/v2 v2.34.0
 	github.com/gin-gonic/gin v1.10.0
 	github.com/go-playground/validator/v10 v10.23.0
+	github.com/google/uuid v1.6.0
 	github.com/gosimple/slug v1.15.0
 	github.com/jonboulle/clockwork v0.5.0
 	github.com/prometheus/client_golang v1.20.5
 	github.com/prometheus/client_model v0.6.1
 	github.com/prometheus/common v0.61.0
+	github.com/redis/go-redis/v9 v9.7.0
 	github.com/rs/zerolog v1.33.0
 	github.com/stretchr/testify v1.10.0
 	github.com/swaggo/files v1.0.1
 	github.com/swaggo/gin-swagger v1.6.0
+	github.com/swaggo/swag v1.16.3
 	go.uber.org/fx v1.23.0
 	golang.org/x/text v0.21.0
 )
 
 require (
 	github.com/KyleBanks/depth v1.2.1 // indirect
+	github.com/alicebob/gopher-json v0.0.0-20230218143504-906a9b012302 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/bytedance/sonic v1.11.9 // indirect
 	github.com/bytedance/sonic/loader v0.1.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/cloudwego/base64x v0.1.4 // indirect
 	github.com/cloudwego/iasm v0.2.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
 	github.com/gabriel-vasile/mimetype v1.4.4 // indirect
 	github.com/gin-contrib/sse v0.1.0 // indirect
 	github.com/go-openapi/jsonpointer v0.21.0 // indirect
@@ -53,9 +59,9 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
 	github.com/stretchr/objx v0.5.2 // indirect
-	github.com/swaggo/swag v1.16.3 // indirect
 	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
 	github.com/ugorji/go/codec v1.2.12 // indirect
+	github.com/yuin/gopher-lua v1.1.1 // indirect
 	go.uber.org/dig v1.18.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
 	go.uber.org/zap v1.27.0 // indirect

go.sum

@@ -1,7 +1,15 @@
 github.com/KyleBanks/depth v1.2.1 h1:5h8fQADFrWtarTdtDudMmGsC7GPbOAu6RVB3ffsVFHc=
 github.com/KyleBanks/depth v1.2.1/go.mod h1:jzSb9d0L43HxTQfT+oSA1EEp2q+ne2uh6XgeJcm8brE=
+github.com/alicebob/gopher-json v0.0.0-20230218143504-906a9b012302 h1:uvdUDbHQHO85qeSydJtItA4T55Pw6BtAejd0APRJOCE=
+github.com/alicebob/gopher-json v0.0.0-20230218143504-906a9b012302/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
+github.com/alicebob/miniredis/v2 v2.34.0 h1:mBFWMaJSNL9RwdGRyEDoAAv8OQc5UlEhLDQggTglU/0=
+github.com/alicebob/miniredis/v2 v2.34.0/go.mod h1:kWShP4b58T1CW0Y5dViCd5ztzrDqRWqM3nksiyXk5s8=
 github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
 github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
+github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
+github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
+github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
+github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/bytedance/sonic v1.11.9 h1:LFHENlIY/SLzDWverzdOvgMztTxcfcF+cqNsz9pK5zg=
 github.com/bytedance/sonic v1.11.9/go.mod h1:LysEHSvpvDySVdC2f87zGWf6CIKJcAvqab1ZaiQtds4=
 github.com/bytedance/sonic/loader v0.1.1 h1:c+e5Pt1k/cy5wMveRDyk2X4B9hF4g7an8N3zCYjJFNM=
@@ -16,6 +24,8 @@ github.com/coreos/go-systemd/v22 v22.5.0/go.mod h1:Y58oyj3AT4RCenI/lSvhwexgC+NSV
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
+github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
 github.com/gabriel-vasile/mimetype v1.4.4 h1:QjV6pZ7/XZ7ryI2KuyeEDE8wnh7fHP9YnQy+R0LnH8I=
 github.com/gabriel-vasile/mimetype v1.4.4/go.mod h1:JwLei5XPtWdGiMFB5Pjle1oEeoSeEuJfJE+TtfvdB/s=
 github.com/gin-contrib/gzip v0.0.6 h1:NjcunTcGAj5CO1gn4N8jHOSIeRFHIbn51z6K+xaN4d4=
@@ -46,6 +56,8 @@ github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5x
 github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI=
 github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
 github.com/gosimple/slug v1.15.0 h1:wRZHsRrRcs6b0XnxMUBM6WK1U1Vg5B0R7VkIf1Xzobo=
 github.com/gosimple/slug v1.15.0/go.mod h1:UiRaFH+GEilHstLUmcBgWcI42viBN7mAb818JrYOeFQ=
 github.com/gosimple/unidecode v1.0.1 h1:hZzFTMMqSswvf0LBJZCZgThIZrpDHFXux9KeGmn6T/o=
@@ -98,6 +110,8 @@ github.com/prometheus/common v0.61.0 h1:3gv/GThfX0cV2lpO7gkTUwZru38mxevy90Bj8YFS
 github.com/prometheus/common v0.61.0/go.mod h1:zr29OCN/2BsJRaFwG8QOBr41D6kkchKbpeNH7pAjb/s=
 github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
 github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
+github.com/redis/go-redis/v9 v9.7.0 h1:HhLSs+B6O021gwzl+locl0zEDnyNkxMtf/Z3NNBMa9E=
+github.com/redis/go-redis/v9 v9.7.0/go.mod h1:f6zhXITC7JUJIlPEiBOTXxJgPLdZcA93GewI7inzyWw=
 github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M=
 github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA=
 github.com/rs/xid v1.5.0/go.mod h1:trrq9SKmegXys3aeAKXMUTdJsYXVwGY3RLcfgqegfbg=
@@ -128,6 +142,8 @@ github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2
 github.com/ugorji/go/codec v1.2.12 h1:9LC83zGrHhuUA9l16C9AHXAqEV/2wBQ4nkvumAE65EE=
 github.com/ugorji/go/codec v1.2.12/go.mod h1:UNopzCgEMSXjBc6AOMqYvWC1ktqTAfzJZUZgYf6w6lg=
 github.com/yuin/goldmark v1.4.13/go.mod h1:6yULJ656Px+3vBD8DxQVa3kxgyrAnzto9xy5taEt/CY=
+github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M=
+github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw=
 go.uber.org/dig v1.18.0 h1:imUL1UiY0Mg4bqbFfsRQO5G4CGRBec/ZujWTvSVp3pw=
 go.uber.org/dig v1.18.0/go.mod h1:Us0rSJiThwCv2GteUN0Q7OKvU7n5J4dxZ9JKUXozFdE=
 go.uber.org/fx v1.23.0 h1:lIr/gYWQGfTwGcSXWXu4vP5Ws6iqnNEIY+F/aFzCKTg=

internal/core/entities/addr/addr.go

@@ -9,8 +9,8 @@ import (
 )
 
 type Addr struct {
-	IP   [4]byte
-	Port int
+	IP   [4]byte `json:"ip"`
+	Port int     `json:"port"`
 }
 
 var Blank Addr // nolint: gochecknoglobals

internal/core/entities/probe/probe.go

@@ -27,11 +27,11 @@ func (goal Goal) String() string {
 var NC = time.Time{} // no constraint
 
 type Probe struct {
-	Addr       addr.Addr
-	Port       int
-	Goal       Goal
-	Retries    int
-	MaxRetries int
+	Addr       addr.Addr `json:"addr"`
+	Port       int       `json:"port"`
+	Goal       Goal      `json:"goal"`
+	Retries    int       `json:"retries"`
+	MaxRetries int       `json:"max_retries"`
 }
 
 var Blank Probe // nolint: gochecknoglobals

internal/core/entities/server/server_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/sergeii/swat4master/internal/core/entities/details"
 	ds "github.com/sergeii/swat4master/internal/core/entities/discovery/status"
 	"github.com/sergeii/swat4master/internal/core/entities/server"
+	"github.com/sergeii/swat4master/internal/testutils/factories/infofactory"
 )
 
 func TestServer_New(t *testing.T) {
@@ -161,14 +162,16 @@ func TestServer_InfoIsUpdated(t *testing.T) {
 	svr := server.MustNew(net.ParseIP("1.1.1.1"), 10480, 10481)
 	assert.Equal(t, "", svr.Info.Hostname)
 
-	newInfo := details.MustNewInfoFromParams(map[string]string{
-		"hostname":    "Swat4 Server",
-		"hostport":    "10480",
-		"mapname":     "A-Bomb Nightclub",
-		"gamever":     "1.1",
-		"gamevariant": "SWAT 4",
-		"gametype":    "Barricaded Suspects",
-	})
+	newInfo := infofactory.Build(infofactory.WithFields(
+		infofactory.F{
+			"hostname":    "Swat4 Server",
+			"hostport":    "10480",
+			"mapname":     "A-Bomb Nightclub",
+			"gamever":     "1.1",
+			"gamevariant": "SWAT 4",
+			"gametype":    "Barricaded Suspects",
+		},
+	))
 	svr.UpdateInfo(newInfo, time.Now())
 
 	updatedInfo := svr.Info

internal/core/repositories/probe.go

@@ -8,19 +8,15 @@ import (
 	"github.com/sergeii/swat4master/internal/core/entities/probe"
 )
 
-var (
-	ErrProbeQueueIsEmpty = errors.New("queue is empty")
-	ErrProbeIsNotReady   = errors.New("queue has waiting probes")
-	ErrProbeHasExpired   = errors.New("probe has expired")
-)
+var ErrProbeQueueIsEmpty = errors.New("queue is empty")
 
 var NC = time.Time{} // no constraint
 
 type ProbeRepository interface {
 	Add(context.Context, probe.Probe) error
 	AddBetween(context.Context, probe.Probe, time.Time, time.Time) error
 	Pop(context.Context) (probe.Probe, error)
-	PopAny(context.Context) (probe.Probe, error)
+	Peek(context.Context) (probe.Probe, error)
 	PopMany(context.Context, int) ([]probe.Probe, int, error)
 	Count(context.Context) (int, error)
 }

internal/core/usecases/addserver/addserver_test.go

@@ -17,7 +17,7 @@ import (
 	"github.com/sergeii/swat4master/internal/core/repositories"
 	"github.com/sergeii/swat4master/internal/core/usecases/addserver"
 	"github.com/sergeii/swat4master/internal/metrics"
-	"github.com/sergeii/swat4master/internal/testutils/factories"
+	"github.com/sergeii/swat4master/internal/testutils/factories/serverfactory"
 )
 
 type MockServerRepository struct {
@@ -133,9 +133,9 @@ func TestAddServerUseCase_ServerExists(t *testing.T) {
 			logger := zerolog.Nop()
 			collector := metrics.New()
 
-			svr := factories.BuildServer(
-				factories.WithAddress("1.1.1.1", 10480),
-				factories.WithDiscoveryStatus(tt.status),
+			svr := serverfactory.Build(
+				serverfactory.WithAddress("1.1.1.1", 10480),
+				serverfactory.WithDiscoveryStatus(tt.status),
 			)
 
 			serverRepo := new(MockServerRepository)
@@ -200,7 +200,7 @@ func TestAddServerUseCase_ServerDoesNotExist(t *testing.T) {
 	logger := zerolog.Nop()
 	collector := metrics.New()
 
-	newSvr := factories.BuildServer(factories.WithAddress("1.1.1.1", 10480))
+	newSvr := serverfactory.Build(serverfactory.WithAddress("1.1.1.1", 10480))
 
 	serverRepo := new(MockServerRepository)
 	serverRepo.On("Get", ctx, newSvr.Addr).Return(server.Blank, repositories.ErrServerNotFound)

internal/core/usecases/cleanservers/cleanservers_test.go

@@ -15,7 +15,7 @@ import (
 	"github.com/sergeii/swat4master/internal/core/entities/server"
 	"github.com/sergeii/swat4master/internal/core/repositories"
 	"github.com/sergeii/swat4master/internal/core/usecases/cleanservers"
-	"github.com/sergeii/swat4master/internal/testutils/factories"
+	"github.com/sergeii/swat4master/internal/testutils/factories/serverfactory"
 )
 
 type MockServerRepository struct {
@@ -62,8 +62,8 @@ func TestCleanServersUseCase_Success(t *testing.T) {
 	until := time.Now().Add(-24 * time.Hour) // Example time filter
 
 	outdatedServers := []server.Server{
-		factories.BuildRandomServer(),
-		factories.BuildRandomServer(),
+		serverfactory.BuildRandom(),
+		serverfactory.BuildRandom(),
 	}
 
 	serverRepo := new(MockServerRepository)
@@ -132,9 +132,9 @@ func TestCleanServersUseCase_RemoveErrors(t *testing.T) {
 
 	until := time.Now().Add(-24 * time.Hour) // Example time filter
 
-	svr1 := factories.BuildRandomServer()
-	svr2 := factories.BuildRandomServer()
-	svr3 := factories.BuildRandomServer()
+	svr1 := serverfactory.BuildRandom()
+	svr2 := serverfactory.BuildRandom()
+	svr3 := serverfactory.BuildRandom()
 	outdatedServers := []server.Server{svr1, svr2, svr3}
 
 	serverRepo := new(MockServerRepository)

internal/core/usecases/getserver/getserver_test.go

@@ -12,7 +12,7 @@ import (
 	"github.com/sergeii/swat4master/internal/core/entities/server"
 	"github.com/sergeii/swat4master/internal/core/repositories"
 	"github.com/sergeii/swat4master/internal/core/usecases/getserver"
-	"github.com/sergeii/swat4master/internal/testutils/factories"
+	"github.com/sergeii/swat4master/internal/testutils/factories/serverfactory"
 )
 
 type MockServerRepository struct {
@@ -28,7 +28,7 @@ func (m *MockServerRepository) Get(ctx context.Context, addr addr.Addr) (server.
 func TestGetServerUseCase_OK(t *testing.T) {
 	ctx := context.TODO()
 
-	svr := factories.BuildServer(factories.WithDiscoveryStatus(ds.Details))
+	svr := serverfactory.Build(serverfactory.WithDiscoveryStatus(ds.Details))
 
 	mockRepo := new(MockServerRepository)
 	mockRepo.On("Get", ctx, svr.Addr).Return(svr, nil)
@@ -100,9 +100,9 @@ func TestGetServerUseCase_ValidateStatus(t *testing.T) {
 		t.Run(tt.name, func(t *testing.T) {
 			ctx := context.TODO()
 
-			svr := factories.BuildServer(
-				factories.WithAddress("1.1.1.1", 10480),
-				factories.WithDiscoveryStatus(tt.status),
+			svr := serverfactory.Build(
+				serverfactory.WithAddress("1.1.1.1", 10480),
+				serverfactory.WithDiscoveryStatus(tt.status),
 			)
 
 			mockRepo := new(MockServerRepository)