refactor(api): remove deprecated workers

The old `api/workers` has been removed in favor of the new worker
package (#3979). The workers' initialization has also been moved to the
`startServer` function to ensure proper shutdown within the router.

api/go.mod

@@ -28,6 +28,7 @@ require (
 	github.com/Azure/go-ansiterm v0.0.0-20230124172434-306776ec8161 // indirect
 	github.com/Microsoft/go-winio v0.6.2 // indirect
 	github.com/Microsoft/hcsshim v0.12.2 // indirect
+	github.com/adhocore/gronx v1.8.1 // indirect
 	github.com/andybalholm/brotli v1.1.0 // indirect
 	github.com/cenkalti/backoff/v4 v4.3.0 // indirect
 	github.com/cespare/xxhash/v2 v2.2.0 // indirect

api/go.sum

@@ -8,6 +8,8 @@ github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERo
 github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
 github.com/Microsoft/hcsshim v0.12.2 h1:AcXy+yfRvrx20g9v7qYaJv5Rh+8GaHOS6b8G6Wx/nKs=
 github.com/Microsoft/hcsshim v0.12.2/go.mod h1:RZV12pcHCXQ42XnlQ3pz6FZfmrC1C+R4gaOHhRNML1g=
+github.com/adhocore/gronx v1.8.1 h1:F2mLTG5sB11z7vplwD4iydz3YCEjstSfYmCrdSm3t6A=
+github.com/adhocore/gronx v1.8.1/go.mod h1:7oUY1WAU8rEJWmAxXR2DN0JaO4gi9khSgKjiRypqteg=
 github.com/andybalholm/brotli v1.0.1/go.mod h1:loMXtMfwqflxFJPmdbJO0a3KNoPuLBgiu3qAvBg8x/Y=
 github.com/andybalholm/brotli v1.1.0 h1:eLKJA0d02Lf0mVpIDgYnqXcUn0GqVmEFny3VuID1U3M=
 github.com/andybalholm/brotli v1.1.0/go.mod h1:sms7XGricyQI9K10gOSf56VKKWS4oLer58Q+mhRPtnY=
@@ -267,6 +269,8 @@ github.com/testcontainers/testcontainers-go v0.32.0 h1:ug1aK08L3gCHdhknlTTwWjPHP
 github.com/testcontainers/testcontainers-go v0.32.0/go.mod h1:CRHrzHLQhlXUsa5gXjTOfqIEJcrK5+xMDmBr/WMI88E=
 github.com/testcontainers/testcontainers-go/modules/mongodb v0.32.0 h1:DvmvHV1irfNIVBhixeTAcoaWCvmdkoNQxRmZisqic4E=
 github.com/testcontainers/testcontainers-go/modules/mongodb v0.32.0/go.mod h1:z0ZvM2V2iThZGrzEN6sddJpvnGhJd6O1O0FTFoZXmpk=
+github.com/testcontainers/testcontainers-go/modules/redis v0.32.0 h1:HW5Qo9qfLi5iwfS7cbXwG6qe8ybXGePcgGPEmVlVDlo=
+github.com/testcontainers/testcontainers-go/modules/redis v0.32.0/go.mod h1:5kltdxVKZG0aP1iegeqKz4K8HHyP0wbkW5o84qLyMjY=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI=
 github.com/tklauser/go-sysconf v0.3.13 h1:GBUpcahXSpR2xN01jhkNAbTLRk2Yzgggk8IM08lq3r4=

api/server.go

@@ -13,10 +13,11 @@ import (
 	"github.com/shellhub-io/shellhub/api/store"
 	"github.com/shellhub-io/shellhub/api/store/mongo"
 	"github.com/shellhub-io/shellhub/api/store/mongo/options"
-	"github.com/shellhub-io/shellhub/api/workers"
 	requests "github.com/shellhub-io/shellhub/pkg/api/internalclient"
 	storecache "github.com/shellhub-io/shellhub/pkg/cache"
 	"github.com/shellhub-io/shellhub/pkg/geoip"
+	workerlib "github.com/shellhub-io/shellhub/pkg/worker"
+	"github.com/shellhub-io/shellhub/pkg/worker/asynq"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 )
@@ -61,13 +62,6 @@ var serverCmd = &cobra.Command{
 
 		log.Info("Connected to MongoDB")
 
-		worker, err := workers.New(store)
-		if err != nil {
-			log.WithError(err).Warn("Failed to create workers.")
-		}
-
-		worker.Start(ctx)
-
 		go func() {
 			sig := <-sigs
 
@@ -99,9 +93,28 @@ type config struct {
 	GeoIP               bool   `env:"GEOIP,default=false"`
 	GeoIPMaxMindLicense string `env:"MAXMIND_LICENSE,default="`
 	// Session record cleanup worker schedule
-	SessionRecordCleanupSchedule string `env:"SESSION_RECORD_CLEANUP_SCHEDULE,default=@daily"`
 	// Sentry DSN.
 	SentryDSN string `env:"SENTRY_DSN,default="`
+	// AsynqGroupMaxDelay is the maximum duration to wait before processing a group of tasks.
+	//
+	// Its time unit is second.
+	//
+	// Check [https://github.com/hibiken/asynq/wiki/Task-aggregation] for more information.
+	AsynqGroupMaxDelay int `env:"ASYNQ_GROUP_MAX_DELAY,default=1"`
+	// AsynqGroupGracePeriod is the grace period has configurable upper bound: you can set a maximum aggregation delay, after which Asynq server
+	// will aggregate the tasks regardless of the remaining grace period.
+	///
+	// Its time unit is second.
+	//
+	// Check [https://github.com/hibiken/asynq/wiki/Task-aggregation] for more information.
+	AsynqGroupGracePeriod int64 `env:"ASYNQ_GROUP_GRACE_PERIOD,default=1"`
+	// AsynqGroupMaxSize is the maximum number of tasks that can be aggregated together. If that number is reached, Asynq
+	// server will aggregate the tasks immediately.
+	//
+	// Check [https://github.com/hibiken/asynq/wiki/Task-aggregation] for more information.
+	AsynqGroupMaxSize             int    `env:"ASYNQ_GROUP_MAX_SIZE,default=500"`
+	SessionRecordCleanupSchedule  string `env:"SESSION_RECORD_CLEANUP_SCHEDULE,default=@daily"`
+	SessionRecordCleanupRetention int    `env:"RECORD_RETENTION,default=0"`
 }
 
 // startSentry initializes the Sentry client.
@@ -165,13 +178,25 @@ func startServer(ctx context.Context, cfg *config, store store.Store, cache stor
 		routerOptions = append(routerOptions, routes.WithReporter(reporter))
 	}
 
+	worker := asynq.NewServer(cfg.RedisURI, asynq.BatchConfig(cfg.AsynqGroupMaxSize, cfg.AsynqGroupMaxDelay, int(cfg.AsynqGroupGracePeriod)))
+	worker.HandleTask(services.TaskDevicesHeartbeat, service.DevicesHeartbeat(), asynq.BatchTask())
+	if cfg.SessionRecordCleanupRetention > 0 {
+		worker.HandleCron(workerlib.CronSpec(cfg.SessionRecordCleanupSchedule), service.CleanupSessions(cfg.SessionRecordCleanupRetention))
+	}
+
+	if err := worker.Start(); err != nil {
+		log.WithError(err).
+			Fatal("failed to start the worker")
+	}
+
 	router := routes.NewRouter(service, routerOptions...)
 
 	go func() {
 		<-ctx.Done()
 
 		log.Debug("Closing HTTP server due context cancellation")
 
+		worker.Shutdown()
 		router.Close()
 	}()