refactor: remove job repository

cli/server.go

@@ -132,16 +132,10 @@ func runServer(ctx context.Context, cfg *Config) error {
 		return fmt.Errorf("create new lineage repository: %w", err)
 	}
 
-	jobRepository, err := postgres.NewJobRepository(pgClient)
-	if err != nil {
-		return fmt.Errorf("create new job repository: %w", err)
-	}
-
 	wrkr, err := initAssetWorker(ctx, workermanager.Deps{
 		Config:        cfg.Worker,
 		DiscoveryRepo: discoveryRepository,
 		AssetRepo:     assetRepository,
-		JobRepo:       jobRepository,
 		Logger:        logger,
 	})
 	if err != nil {

cli/worker.go

@@ -78,17 +78,11 @@ func runWorker(ctx context.Context, cfg *Config) error {
 		return fmt.Errorf("create new asset repository: %w", err)
 	}
 
-	jobRepository, err := postgres.NewJobRepository(pgClient)
-	if err != nil {
-		return fmt.Errorf("create new job repository: %w", err)
-	}
-
 	mgr, err := workermanager.New(ctx, workermanager.Deps{
 		Config: cfg.Worker,
 		DiscoveryRepo: elasticsearch.NewDiscoveryRepository(esClient, logger, cfg.Elasticsearch.RequestTimeout,
 			strings.Split(cfg.ColSearchExclusionKeywords, ",")),
 		AssetRepo: assetRepository,
-		JobRepo:   jobRepository,
 		Logger:    logger,
 	})
 	if err != nil {

internal/store/elasticsearch/discovery_repository.go

@@ -85,7 +85,7 @@ func (repo *DiscoveryRepository) SyncAssets(ctx context.Context, indexName strin
 		return nil, err
 	}
 
-	err = repo.updateAlias(ctx, backupIndexName, "universe")
+	err = repo.updateAlias(ctx, backupIndexName, defaultSearchIndex)
 	if err != nil {
 		return nil, err
 	}
@@ -101,7 +101,7 @@ func (repo *DiscoveryRepository) SyncAssets(ctx context.Context, indexName strin
 	}
 
 	cleanup := func() error {
-		err = repo.updateAlias(ctx, indexName, "universe")
+		err = repo.updateAlias(ctx, indexName, defaultSearchIndex)
 		if err != nil {
 			return err
 		}

internal/workermanager/discovery_worker.go

@@ -75,7 +75,7 @@ func (m *Manager) SyncAssets(ctx context.Context, job worker.JobSpec) error {
 	const batchSize = 1000
 	service := string(job.Payload)
 
-	jobs, err := m.jobRepo.GetSyncJobsByService(ctx, service)
+	jobs, err := m.worker.GetSyncJobsByService(ctx, service)
 	if err != nil {
 		return fmt.Errorf("sync asset: get sync jobs by service: %w", err)
 	}

internal/workermanager/in_situ_worker.go

@@ -3,22 +3,20 @@ package workermanager
 import (
 	"context"
 	"fmt"
-	"time"
+	"sync"
 
 	"github.com/goto/compass/core/asset"
-	"github.com/goto/compass/core/job"
 )
 
 type InSituWorker struct {
 	discoveryRepo DiscoveryRepository
-	jobRepo       job.Repository
 	assetRepo     asset.Repository
+	mutex         sync.Mutex
 }
 
 func NewInSituWorker(deps Deps) *InSituWorker {
 	return &InSituWorker{
 		discoveryRepo: deps.DiscoveryRepo,
-		jobRepo:       deps.JobRepo,
 		assetRepo:     deps.AssetRepo,
 	}
 }
@@ -40,22 +38,9 @@ func (m *InSituWorker) EnqueueDeleteAssetJob(ctx context.Context, urn string) er
 
 func (m *InSituWorker) EnqueueSyncAssetJob(ctx context.Context, service string) error {
 	const batchSize = 1000
-	jobs, err := m.jobRepo.GetSyncJobsByService(ctx, service)
-	if err != nil {
-		return fmt.Errorf("sync asset: get sync jobs by service: %w", err)
-	}
-
-	if len(jobs) > 0 {
-		return nil // mark job as done if there's earlier job with same service
-	}
 
-	jobID, err := m.jobRepo.Insert(ctx, jobSyncAsset, ([]byte)(service), time.Now().UTC())
-	if err != nil {
-		return fmt.Errorf("sync asset: insert job queue: %w", err)
-	}
-	defer func() {
-		_ = m.jobRepo.Delete(ctx, jobID)
-	}()
+	m.mutex.Lock()
+	defer m.mutex.Unlock()
 
 	cleanup, err := m.discoveryRepo.SyncAssets(ctx, service)
 	if err != nil {

internal/workermanager/worker_manager.go

@@ -9,7 +9,6 @@ import (
 	"time"
 
 	"github.com/goto/compass/core/asset"
-	"github.com/goto/compass/core/job"
 	"github.com/goto/compass/pkg/worker"
 	"github.com/goto/compass/pkg/worker/pgq"
 	"github.com/goto/compass/pkg/worker/workermw"
@@ -26,7 +25,6 @@ type Manager struct {
 	jobManagerPort int
 	discoveryRepo  DiscoveryRepository
 	assetRepo      asset.Repository
-	jobRepo        job.Repository
 	logger         log.Logger
 }
 
@@ -36,6 +34,7 @@ type Worker interface {
 	Register(typ string, h worker.JobHandler) error
 	Run(ctx context.Context) error
 	Enqueue(ctx context.Context, jobs ...worker.JobSpec) error
+	GetSyncJobsByService(ctx context.Context, service string) ([]worker.Job, error)
 }
 
 type Config struct {
@@ -51,7 +50,6 @@ type Deps struct {
 	Config        Config
 	DiscoveryRepo DiscoveryRepository
 	AssetRepo     asset.Repository
-	JobRepo       job.Repository
 	Logger        log.Logger
 }
 
@@ -78,7 +76,6 @@ func New(ctx context.Context, deps Deps) (*Manager, error) {
 		jobManagerPort: cfg.JobManagerPort,
 		discoveryRepo:  deps.DiscoveryRepo,
 		assetRepo:      deps.AssetRepo,
-		jobRepo:        deps.JobRepo,
 		logger:         deps.Logger,
 	}, nil
 }

pkg/worker/job_processor.go

@@ -21,6 +21,9 @@ type JobProcessor interface {
 	// Stats returns the job statistics by job type. It includes the number of
 	// active and dead jobs.
 	Stats(ctx context.Context) ([]JobTypeStats, error)
+
+	//
+	GetSyncJobsByService(ctx context.Context, service string) ([]Job, error)
 }
 
 // JobTypeStats is the statistics for the job type with number of active and