fix: handle deleted jobs in scheduler BC (#712)

* fix: handle delered jobs in scheduler BC

* fix: handle delered jobs in scheduler BC

* fix: add better error mesages in airflow.go

Author: Mryashbhardwaj

ext/scheduler/airflow/airflow.go

@@ -83,12 +83,14 @@ func (s *Scheduler) DeployJobs(ctx context.Context, tenant tenant.Tenant, jobs [
 
 	bucket, err := s.bucketFac.New(spanCtx, tenant)
 	if err != nil {
-		return err
+		return errors.AddErrContext(err, EntityAirflow, "error in creating storage client instance")
 	}
 	defer bucket.Close()
 
-	bucket.WriteAll(spanCtx, filepath.Join(jobsDir, baseLibFileName), SharedLib, nil)
-
+	err = bucket.WriteAll(spanCtx, filepath.Join(jobsDir, baseLibFileName), SharedLib, nil)
+	if err != nil {
+		return errors.AddErrContext(err, EntityAirflow, "error in writing __lib.py file")
+	}
 	multiError := errors.NewMultiError("ErrorsInDeployJobs")
 	runner := parallel.NewRunner(parallel.WithTicket(concurrentTicketPerSec), parallel.WithLimit(concurrentLimit))
 	for _, job := range jobs {

internal/store/postgres/scheduler/job_repository.go

@@ -241,7 +241,7 @@ func FromRow(row pgx.Row) (*Job, error) {
 }
 
 func (j *JobRepository) GetJob(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName) (*scheduler.Job, error) {
-	getJobByNameAtProject := `SELECT ` + jobColumns + ` FROM job WHERE name = $1 AND project_name = $2`
+	getJobByNameAtProject := `SELECT ` + jobColumns + ` FROM job WHERE name = $1 AND project_name = $2 AND deleted_at IS NULL`
 	spec, err := FromRow(j.db.QueryRow(ctx, getJobByNameAtProject, jobName, projectName))
 	if err != nil {
 		return nil, err
@@ -250,7 +250,7 @@ func (j *JobRepository) GetJob(ctx context.Context, projectName tenant.ProjectNa
 }
 
 func (j *JobRepository) GetJobDetails(ctx context.Context, projectName tenant.ProjectName, jobName scheduler.JobName) (*scheduler.JobWithDetails, error) {
-	getJobByNameAtProject := `SELECT ` + jobColumns + ` FROM job WHERE name = $1 AND project_name = $2`
+	getJobByNameAtProject := `SELECT ` + jobColumns + ` FROM job WHERE name = $1 AND project_name = $2 AND deleted_at IS NULL`
 	spec, err := FromRow(j.db.QueryRow(ctx, getJobByNameAtProject, jobName, projectName))
 	if err != nil {
 		return nil, err
@@ -305,7 +305,7 @@ func (j *JobRepository) getJobsUpstreams(ctx context.Context, projectName tenant
 }
 
 func (j *JobRepository) GetAll(ctx context.Context, projectName tenant.ProjectName) ([]*scheduler.JobWithDetails, error) {
-	getJobByNameAtProject := `SELECT ` + jobColumns + ` FROM job WHERE project_name = $1`
+	getJobByNameAtProject := `SELECT ` + jobColumns + ` FROM job WHERE project_name = $1 AND deleted_at IS NULL`
 	rows, err := j.db.Query(ctx, getJobByNameAtProject, projectName)
 	if err != nil {
 		return nil, errors.Wrap(job.EntityJob, "error while getting all jobs", err)