package de.otto.edison.jobs.service;

import de.otto.edison.jobs.domain.JobInfo;
import de.otto.edison.jobs.repository.JobRepository;
import org.slf4j.Logger;
import org.slf4j.MDC;

import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.slf4j.LoggerFactory.getLogger;

public final class JobRunner {

    private static final Logger LOG = getLogger(JobRunner.class);
    public static final long PING_PERIOD = 1l;

    private volatile JobInfo jobInfo;
    private final JobRepository jobRepository;
    private final ScheduledExecutorService executorService;
    private ScheduledFuture<?> pingJob;

    private JobRunner(final JobInfo jobInfo,
                      final JobRepository jobRepository,
                      final ScheduledExecutorService executorService) {
        this.jobInfo = jobInfo;
        this.jobRepository = jobRepository;
        this.executorService = executorService;
    }

    public static JobRunner newJobRunner(final JobInfo job,
                                         final JobRepository jobRepository,
                                         final ScheduledExecutorService executorService) {
        final JobRunner jobRunner = new JobRunner(job, jobRepository, executorService);
        jobRepository.createOrUpdate(jobRunner.jobInfo);
        return jobRunner;
    }

    public void start(final JobRunnable runnable) {
        start();
        try {
            runnable.execute(jobInfo);
        } catch (final RuntimeException e) {
            error(e);
        } finally {
            stop();
        }
    }

    private void start() {
        synchronized (this) {
            pingJob = executorService.scheduleAtFixedRate(this::ping, PING_PERIOD, PING_PERIOD, SECONDS);

            final String jobId = jobInfo.getJobUri().toString();
            MDC.put("job_id", jobId.substring(jobId.lastIndexOf('/') + 1));
            MDC.put("job_type", jobInfo.getJobType());
            LOG.info("[started]");
        }
    }

    private void ping() {
        try {
            synchronized (this) {
                if (jobRepository.findStatus(jobInfo.getJobUri()).equals(JobInfo.JobStatus.DEAD)) {
                    jobInfo.dead();
                }
                jobInfo.ping();
                jobRepository.createOrUpdate(jobInfo);
            }
        } catch (Exception e) {
            assert !jobInfo.isStopped();
            LOG.error("Fatal error in ping job for" + jobInfo.getJobType() + " (" + jobInfo.getJobUri() + ")", e.getMessage());
        }
    }

    private void error(final Exception e) {
        synchronized (this) {
            assert !jobInfo.isStopped();
            jobInfo.error(e.getMessage());
            jobRepository.createOrUpdate(jobInfo);
            LOG.error("Fatal error in job " + jobInfo.getJobType() + " (" + jobInfo.getJobUri() + ")", e);
        }
    }

    private void stop() {
        synchronized (this) {
            pingJob.cancel(false);

            assert !jobInfo.isStopped();
            try {
                LOG.info("stopped job {}", jobInfo);
                jobInfo.stop();
                jobRepository.createOrUpdate(jobInfo);
            } finally {
                MDC.clear();
            }
        }
    }

}

@@ -1674,24 +1674,37 @@
     private 
+synchronized 
 void start()
@@ -1710,42 +1710,8 @@
 ) %7B%0A
-        synchronized (this) %7B%0A    
     
@@ -1813,28 +1813,24 @@
 );%0A%0A        
-    
 final String
@@ -1866,28 +1866,24 @@
 toString();%0A
-    
         MDC.
@@ -1946,28 +1946,24 @@
 ));%0A        
-  
-  
 MDC.put(%22job
@@ -1993,36 +1993,32 @@
 ype());%0A        
-    
 LOG.info(%22%5Bstart
@@ -2024,26 +2024,16 @@
 ted%5D%22);%0A
-        %7D%0A
     %7D%0A%0A 
@@ -2039,24 +2039,37 @@
     private 
+synchronized 
 void ping() 
@@ -2088,46 +2088,8 @@
 y %7B%0A
-            synchronized (this) %7B%0A    
     
@@ -2188,36 +2188,32 @@
                 
-    
 jobInfo.dead();%0A
@@ -2208,28 +2208,24 @@
 nfo.dead();%0A
-    
             
@@ -2230,36 +2230,32 @@
   %7D%0A            
-    
 jobInfo.ping();%0A
@@ -2258,36 +2258,32 @@
 ();%0A            
-   
- 
 jobRepository.cr
@@ -2305,30 +2305,16 @@
 bInfo);%0A
-            %7D%0A
         
@@ -2490,29 +2490,16 @@
 + %22)%22, e
-.getMessage()
 );%0A     
@@ -2518,24 +2518,37 @@
     private 
+synchronized 
 void error(f
@@ -2571,42 +2571,8 @@
 ) %7B%0A
-        synchronized (this) %7B%0A    
     
@@ -2596,36 +2596,32 @@
 fo.isStopped();%0A
-    
         jobInfo.
@@ -2643,36 +2643,32 @@
 age());%0A        
-  
-  
 jobRepository.cr
@@ -2682,36 +2682,32 @@
 pdate(jobInfo);%0A
-    
         LOG.erro
@@ -2789,34 +2789,24 @@
  + %22)%22, e);%0A
-        %7D%0A
     %7D%0A%0A    p
@@ -2812,16 +2812,29 @@
 private 
+synchronized 
 void sto
@@ -2843,42 +2843,8 @@
 ) %7B%0A
-        synchronized (this) %7B%0A    
     
@@ -2867,28 +2867,24 @@
 el(false);%0A%0A
-    
         asse
@@ -2912,38 +2912,30 @@
 ();%0A        
-    
 try %7B%0A
-    
             
@@ -2975,36 +2975,32 @@
 o);%0A            
-  
-  
 jobInfo.stop();%0A
@@ -2995,28 +2995,24 @@
 nfo.stop();%0A
-    
             
@@ -3054,28 +3054,24 @@
 o);%0A        
-   
- 
 %7D finally %7B%0A
@@ -3062,36 +3062,32 @@
     %7D finally %7B%0A
-    
             MDC.
@@ -3091,38 +3091,24 @@
 DC.clear();%0A
-            %7D%0A
         %7D%0A  

refactor ping job and add some error logging