func (m *Manager) GetReplay(ctx context.Context, replayUUID uuid.UUID) (models.ReplaySpec, error) {
	return m.replaySpecRepoFac.New().GetByID(ctx, replayUUID)
}
GetReplay using UUID
package job
import (
	"context"
	"sync"
	"sync/atomic"
	"time"
	"github.com/google/uuid"
	"github.com/odpf/optimus/core/tree"
	"github.com/odpf/optimus/models"
	"github.com/odpf/optimus/store"
	"github.com/odpf/optimus/utils"
	"github.com/odpf/salt/log"
	"github.com/pkg/errors"
	"github.com/robfig/cron/v3"
)
var (
	ErrRequestQueueFull = errors.New("request queue is full")
	ErrConflictedJobRun = errors.New("conflicted job run found")
	ReplayRunTimeout = "long running replay timeout"
	TimestampLogFormat = "2006-01-02T15:04:05+00:00"
	schedulerBatchSize = 100
	replayListWindow = -3 * 30 * 24 * time.Hour
)
const (
	syncInterval = "@every 5m"
)
type ReplayManagerConfig struct {
	NumWorkers    int
	WorkerTimeout time.Duration
	RunTimeout    time.Duration
}
type ReplayWorkerFactory interface {
	New() ReplayWorker
}
type ReplaySyncer interface {
	Sync(context.Context, time.Duration) error
}
type ReplayValidator interface {
	Validate(context.Context, store.ReplaySpecRepository, models.ReplayRequest, *tree.TreeNode) error
}
type Manager struct {
	wg sync.WaitGroup
	uuidProvider utils.UUIDProvider
	config       ReplayManagerConfig
	requestQ chan models.ReplayRequest
	replayWorkerFactory ReplayWorkerFactory
	replaySpecRepoFac   ReplaySpecRepoFactory
	scheduler           models.SchedulerUnit
	replayValidator     ReplayValidator
	replaySyncer        ReplaySyncer
	syncerScheduler     *cron.Cron
	l                   log.Logger
	workerCapacity int32
}
func (m *Manager) Replay(ctx context.Context, reqInput models.ReplayRequest) (string, error) {
	replaySpecRepo := m.replaySpecRepoFac.New()
	replayTree, err := prepareReplayExecutionTree(reqInput)
	if err != nil {
		return "", err
	}
	if err := m.replayValidator.Validate(ctx, replaySpecRepo, reqInput, replayTree); err != nil {
		return "", err
	}
	if reqInput.ID, err = m.uuidProvider.NewUUID(); err != nil {
		return "", err
	}
	replay := models.ReplaySpec{
		ID:            reqInput.ID,
		Job:           reqInput.Job,
		StartDate:     reqInput.Start,
		EndDate:       reqInput.End,
		Status:        models.ReplayStatusAccepted,
		ExecutionTree: replayTree,
	}
	if err = replaySpecRepo.Insert(ctx, &replay); err != nil {
		return "", err
	}
	select {
	case m.requestQ <- reqInput:
		return reqInput.ID.String(), nil
	default:
		_ = replaySpecRepo.UpdateStatus(ctx, reqInput.ID, models.ReplayStatusCancelled, models.ReplayMessage{
			Type:    models.ReplayStatusCancelled,
			Message: ErrRequestQueueFull.Error(),
		})
		return "", ErrRequestQueueFull
	}
}
func (m *Manager) spawnServiceWorker(worker ReplayWorker) {
	defer m.wg.Done()
	atomic.AddInt32(&m.workerCapacity, 1)
	for reqInput := range m.requestQ {
		atomic.AddInt32(&m.workerCapacity, -1)
		m.l.Info("worker picked up the request", "request id", reqInput.ID)
		ctx, cancelCtx := context.WithTimeout(context.Background(), m.config.WorkerTimeout)
		if err := worker.Process(ctx, reqInput); err != nil {
			m.l.Error("worker failed to process", "error", err)
		}
		cancelCtx()
		atomic.AddInt32(&m.workerCapacity, 1)
	}
}
func (m *Manager) SchedulerSyncer() {
	m.l.Debug("start synchronizing replays...")
	ctx, cancelCtx := context.WithTimeout(context.Background(), m.config.WorkerTimeout)
	defer cancelCtx()
	if err := m.replaySyncer.Sync(ctx, m.config.RunTimeout); err != nil {
		m.l.Error("syncer failed to process", "error", err)
	}
	m.l.Debug("replays synced")
}