chore: support replay with new file format (#3834)

app/apphandlers/embeddedAppHandler.go

@@ -330,8 +330,14 @@ func (a *embeddedApp) StartRudderCore(ctx context.Context, options *app.Options)
 		if err != nil {
 			return fmt.Errorf("could not setup replayDB: %w", err)
 		}
-		defer replayDB.TearDown()
-		a.app.Features().Replay.Setup(ctx, &replayDB, gatewayDB, routerDB, batchRouterDB)
+		replay, err := a.app.Features().Replay.Setup(ctx, config, &replayDB, gatewayDB, routerDB, batchRouterDB)
+		if err != nil {
+			return err
+		}
+		if err := replay.Start(); err != nil {
+			return fmt.Errorf("could not start replay: %w", err)
+		}
+		defer func() { _ = replay.Stop() }()
 	}
 
 	g.Go(func() error {

app/features.go

@@ -5,7 +5,9 @@ package app
 import (
 	"context"
 
+	"github.com/rudderlabs/rudder-go-kit/config"
 	backendconfig "github.com/rudderlabs/rudder-server/backend-config"
+	"github.com/rudderlabs/rudder-server/enterprise/replay"
 	"github.com/rudderlabs/rudder-server/jobsdb"
 	"github.com/rudderlabs/rudder-server/utils/types"
 )
@@ -40,7 +42,7 @@ Replay Feature
 
 // ReplayFeature handles inserting of failed jobs into respective gw/rt jobsdb
 type ReplayFeature interface {
-	Setup(ctx context.Context, replayDB, gwDB, routerDB, batchRouterDB *jobsdb.Handle)
+	Setup(ctx context.Context, config *config.Config, replayDB, gwDB, routerDB, batchRouterDB *jobsdb.Handle) (replay.Replay, error)
 }
 
 // ReplayFeatureSetup is a function that initializes a Replay feature