TSPS-123 Refactor to follow DSP's Async API best practices

service/src/main/java/bio/terra/pipelines/app/StartupInitializer.java

@@ -2,7 +2,7 @@
 
 import bio.terra.common.migrate.LiquibaseMigrator;
 import bio.terra.pipelines.app.configuration.internal.TspsDatabaseConfiguration;
-import bio.terra.pipelines.dependencies.stairway.StairwayJobService;
+import bio.terra.pipelines.dependencies.stairway.JobService;
 import org.springframework.context.ApplicationContext;
 
 public final class StartupInitializer {
@@ -11,7 +11,7 @@ public final class StartupInitializer {
   public static void initialize(ApplicationContext applicationContext) {
     // Initialize the Terra Scientific Pipelines Service library
     LiquibaseMigrator migrateService = applicationContext.getBean(LiquibaseMigrator.class);
-    StairwayJobService stairwayJobService = applicationContext.getBean(StairwayJobService.class);
+    JobService jobService = applicationContext.getBean(JobService.class);
     TspsDatabaseConfiguration tspsDatabaseConfiguration =
         applicationContext.getBean(TspsDatabaseConfiguration.class);
 
@@ -21,7 +21,7 @@ public static void initialize(ApplicationContext applicationContext) {
       migrateService.upgrade(CHANGELOG_PATH, tspsDatabaseConfiguration.getDataSource());
     }
 
-    // The StairwayJobService initialization also handles Stairway initialization.
-    stairwayJobService.initialize();
+    // The JobService initialization also handles Stairway initialization.
+    jobService.initialize();
   }
 }

service/src/main/java/bio/terra/pipelines/app/common/MetricsUtils.java

@@ -1,5 +1,6 @@
 package bio.terra.pipelines.app.common;
 
+import bio.terra.pipelines.common.utils.PipelinesEnum;
 import io.micrometer.core.instrument.Metrics;
 import lombok.AccessLevel;
 import lombok.NoArgsConstructor;
@@ -20,9 +21,10 @@ public class MetricsUtils {
    *
    * @param pipelineId - pipelineId that was run e.g. "imputation"
    */
-  public static void incrementPipelineRun(String pipelineId) {
+  public static void incrementPipelineRun(PipelinesEnum pipelineId) {
     Metrics.globalRegistry
-        .counter(String.format("%s.pipeline.run.count", NAMESPACE), PIPELINE_TAG, pipelineId)
+        .counter(
+            String.format("%s.pipeline.run.count", NAMESPACE), PIPELINE_TAG, pipelineId.getValue())
         .increment();
   }
 }

service/src/main/java/bio/terra/pipelines/app/controller/GlobalExceptionHandler.java

@@ -32,7 +32,6 @@ public ResponseEntity<ApiErrorReport> errorReportHandler(ErrorReportException ex
   }
 
   // -- validation exceptions - we don't control the exception raised
-  // TODO add JobNotFoundException method here - see TSPS-9
   @ExceptionHandler({
     MethodArgumentNotValidException.class,
     MethodArgumentTypeMismatchException.class,

service/src/main/java/bio/terra/pipelines/app/controller/JobApiUtils.java

@@ -0,0 +1,133 @@
+package bio.terra.pipelines.app.controller;
+
+import bio.terra.common.exception.ErrorReportException;
+import bio.terra.pipelines.dependencies.stairway.JobMapKeys;
+import bio.terra.pipelines.dependencies.stairway.exception.InvalidResultStateException;
+import bio.terra.pipelines.dependencies.stairway.model.EnumeratedJob;
+import bio.terra.pipelines.dependencies.stairway.model.EnumeratedJobs;
+import bio.terra.pipelines.generated.model.ApiErrorReport;
+import bio.terra.pipelines.generated.model.ApiGetJobsResponse;
+import bio.terra.pipelines.generated.model.ApiJobReport;
+import bio.terra.stairway.FlightMap;
+import bio.terra.stairway.FlightState;
+import bio.terra.stairway.FlightStatus;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.List;
+import org.springframework.http.HttpStatus;
+import org.springframework.stereotype.Component;
+
+@Component
+public class JobApiUtils {
+
+  JobApiUtils() {}
+
+  public static ApiGetJobsResponse mapEnumeratedJobsToApi(EnumeratedJobs enumeratedJobs) {
+
+    // Convert the result to API-speak
+    List<ApiJobReport> apiJobList = new ArrayList<>();
+    for (EnumeratedJob enumeratedJob : enumeratedJobs.getResults()) {
+      ApiJobReport jobReport = mapFlightStateToApiJobReport(enumeratedJob.getFlightState());
+      apiJobList.add(jobReport);
+    }
+
+    return new ApiGetJobsResponse()
+        .pageToken(enumeratedJobs.getPageToken())
+        .totalResults(enumeratedJobs.getTotalResults())
+        .results(apiJobList);
+  }
+
+  public static ApiJobReport mapFlightStateToApiJobReport(FlightState flightState) {
+    FlightMap inputParameters = flightState.getInputParameters();
+    String description = inputParameters.get(JobMapKeys.DESCRIPTION.getKeyName(), String.class);
+    FlightStatus flightStatus = flightState.getFlightStatus();
+    String submittedDate = flightState.getSubmitted().toString();
+    ApiJobReport.StatusEnum jobStatus = mapFlightStatusToApi(flightStatus);
+
+    String completedDate = null;
+    HttpStatus statusCode = HttpStatus.ACCEPTED;
+
+    if (jobStatus != ApiJobReport.StatusEnum.RUNNING) {
+      // If the job is completed, the JobReport should include a result code indicating success or
+      // failure. For failed jobs, this code is the error code. For successful jobs, this is the
+      // code specified by the flight if present, or a default of 200 if not.
+      completedDate =
+          flightState
+              .getCompleted()
+              .map(Instant::toString)
+              .orElseThrow(
+                  () -> new InvalidResultStateException("No completed time for completed flight"));
+      switch (jobStatus) {
+        case FAILED -> {
+          int errorCode =
+              flightState
+                  .getException()
+                  .map(e -> buildApiErrorReport(e).getStatusCode())
+                  .orElseThrow(
+                      () ->
+                          new InvalidResultStateException(
+                              String.format(
+                                  "Flight %s failed with no exception reported",
+                                  flightState.getFlightId())));
+          statusCode = HttpStatus.valueOf(errorCode);
+        }
+        case SUCCEEDED -> {
+          FlightMap resultMap =
+              flightState.getResultMap().orElseThrow(InvalidResultStateException::noResultMap);
+          statusCode = resultMap.get(JobMapKeys.STATUS_CODE.getKeyName(), HttpStatus.class);
+          if (statusCode == null) {
+            statusCode = HttpStatus.OK;
+          }
+        }
+        default -> throw new IllegalStateException(
+            "Cannot get status code of flight in unknown state " + jobStatus);
+      }
+    }
+
+    return new ApiJobReport()
+        .id(flightState.getFlightId())
+        .description(description)
+        .status(jobStatus)
+        .statusCode(statusCode.value())
+        .submitted(submittedDate)
+        .completed(completedDate)
+        .resultURL(resultUrlFromFlightState(flightState));
+  }
+
+  private static ApiJobReport.StatusEnum mapFlightStatusToApi(FlightStatus flightStatus) {
+    switch (flightStatus) {
+      case RUNNING, QUEUED, WAITING, READY, READY_TO_RESTART:
+        return ApiJobReport.StatusEnum.RUNNING;
+      case SUCCESS:
+        return ApiJobReport.StatusEnum.SUCCEEDED;
+      case ERROR, FATAL:
+        return ApiJobReport.StatusEnum.FAILED;
+      default:
+        return ApiJobReport.StatusEnum.FAILED;
+    }
+  }
+
+  public static ApiErrorReport buildApiErrorReport(Exception exception) {
+    if (exception instanceof ErrorReportException errorReport) {
+      return new ApiErrorReport()
+          .message(errorReport.getMessage())
+          .statusCode(errorReport.getStatusCode().value())
+          .causes(errorReport.getCauses());
+    } else {
+      return new ApiErrorReport()
+          .message(exception.getMessage())
+          .statusCode(HttpStatus.INTERNAL_SERVER_ERROR.value())
+          .causes(null);
+    }
+  }
+
+  private static String resultUrlFromFlightState(FlightState flightState) {
+    String resultPath =
+        flightState.getInputParameters().get(JobMapKeys.RESULT_PATH.getKeyName(), String.class);
+    if (resultPath == null) {
+      resultPath = "";
+    }
+    // TSPS-135 will implement the GET result endpoint, at which point this path should be created
+    return resultPath;
+  }
+}

service/src/main/java/bio/terra/pipelines/app/controller/JobsApiController.java

@@ -1,19 +1,14 @@
 package bio.terra.pipelines.app.controller;
 
-import bio.terra.common.exception.ApiException;
 import bio.terra.common.iam.SamUser;
 import bio.terra.common.iam.SamUserFactory;
-import bio.terra.pipelines.app.common.MetricsUtils;
 import bio.terra.pipelines.app.configuration.external.SamConfiguration;
-import bio.terra.pipelines.db.entities.Job;
-import bio.terra.pipelines.db.exception.PipelineNotFoundException;
+import bio.terra.pipelines.dependencies.stairway.JobService;
+import bio.terra.pipelines.dependencies.stairway.model.EnumeratedJobs;
 import bio.terra.pipelines.generated.api.JobsApi;
 import bio.terra.pipelines.generated.model.*;
-import bio.terra.pipelines.service.ImputationService;
-import bio.terra.pipelines.service.JobsService;
-import bio.terra.pipelines.service.PipelinesService;
+import bio.terra.stairway.FlightState;
 import io.swagger.annotations.Api;
-import java.util.List;
 import java.util.UUID;
 import javax.servlet.http.HttpServletRequest;
 import org.slf4j.Logger;
@@ -23,7 +18,6 @@
 import org.springframework.http.ResponseEntity;
 import org.springframework.stereotype.Controller;
 import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.RequestBody;
 
 /** Jobs controller */
 @Controller
@@ -32,24 +26,18 @@ public class JobsApiController implements JobsApi {
   private final SamConfiguration samConfiguration;
   private final SamUserFactory samUserFactory;
   private final HttpServletRequest request;
-  private final JobsService jobsService;
-  private final PipelinesService pipelinesService;
-  private final ImputationService imputationService;
+  private final JobService jobService;
 
   @Autowired
   public JobsApiController(
       SamConfiguration samConfiguration,
       SamUserFactory samUserFactory,
       HttpServletRequest request,
-      JobsService jobsService,
-      PipelinesService pipelinesService,
-      ImputationService imputationService) {
+      JobService jobService) {
     this.samConfiguration = samConfiguration;
     this.samUserFactory = samUserFactory;
     this.request = request;
-    this.jobsService = jobsService;
-    this.pipelinesService = pipelinesService;
-    this.imputationService = imputationService;
+    this.jobService = jobService;
   }
 
   private static final Logger logger = LoggerFactory.getLogger(JobsApiController.class);
@@ -61,93 +49,22 @@ private SamUser getAuthenticatedInfo() {
   // -- Jobs --
 
   @Override
-  public ResponseEntity<ApiPostJobResponse> createJob(
-      @PathVariable("pipelineId") String pipelineId, @RequestBody ApiPostJobRequestBody body) {
+  public ResponseEntity<ApiJobReport> getJob(@PathVariable("jobId") UUID jobId) {
     final SamUser userRequest = getAuthenticatedInfo();
     String userId = userRequest.getSubjectId();
-    String pipelineVersion = body.getPipelineVersion();
-    Object pipelineInputs = body.getPipelineInputs();
-
-    if (!pipelinesService.pipelineExists(pipelineId)) {
-      throw new PipelineNotFoundException(
-          String.format("Requested pipeline %s not found.", pipelineId));
-    }
-
-    logger.info(
-        "Creating {} pipeline job (version {}) for {} user {} with inputs {}",
-        pipelineId,
-        pipelineVersion,
-        userRequest.getEmail(),
-        userId,
-        pipelineInputs);
-
-    // TODO assuming we will write outputs back to source workspace, we will need to check user
-    // permissions for write access to the workspace - explore interceptors
-
-    UUID createdJobUuid =
-        jobsService.createJob(userId, pipelineId, pipelineVersion, pipelineInputs);
-    if (createdJobUuid == null) {
-      logger.error("New {} pipeline job creation failed.", pipelineId);
-      throw new ApiException("An internal error occurred.");
-    }
-
-    // eventually we'll expand this out to kick off the imputation pipeline flight but for
-    // now this is good enough.
-    imputationService.queryForWorkspaceApps();
-
-    ApiPostJobResponse createdJobResponse = new ApiPostJobResponse();
-    createdJobResponse.setJobId(createdJobUuid.toString());
-    logger.info("Created {} job {}", pipelineId, createdJobUuid);
-    MetricsUtils.incrementPipelineRun(pipelineId);
-
-    return new ResponseEntity<>(createdJobResponse, HttpStatus.OK);
-  }
-
-  @Override
-  public ResponseEntity<ApiGetJobResponse> getJob(
-      @PathVariable("pipelineId") String pipelineId, @PathVariable("jobId") UUID jobId) {
-    final SamUser userRequest = getAuthenticatedInfo();
-    String userId = userRequest.getSubjectId();
-    Job job = jobsService.getJob(userId, pipelineId, jobId);
-    ApiGetJobResponse result = jobToApi(job);
-
+    logger.info("Retrieving jobId {} for userId {}", jobId, userId);
+    FlightState flightState = jobService.retrieveJob(jobId, userId);
+    ApiJobReport result = JobApiUtils.mapFlightStateToApiJobReport(flightState);
     return new ResponseEntity<>(result, HttpStatus.OK);
   }
 
   @Override
-  public ResponseEntity<ApiGetJobsResponse> getJobs(@PathVariable("pipelineId") String pipelineId) {
+  public ResponseEntity<ApiGetJobsResponse> getAllJobs(Integer limit, String pageToken) {
     final SamUser userRequest = getAuthenticatedInfo();
     String userId = userRequest.getSubjectId();
-    List<Job> jobList = jobsService.getJobs(userId, pipelineId);
-    ApiGetJobsResponse result = jobsToApi(jobList);
-
+    logger.info("Retrieving all jobs for userId {}", userId);
+    EnumeratedJobs enumeratedJobs = jobService.enumerateJobs(userId, limit, pageToken, null);
+    ApiGetJobsResponse result = JobApiUtils.mapEnumeratedJobsToApi(enumeratedJobs);
     return new ResponseEntity<>(result, HttpStatus.OK);
   }
-
-  static ApiGetJobResponse jobToApi(Job job) {
-    ApiGetJobResponse apiGetJobResponse =
-        new ApiGetJobResponse()
-            .jobId(job.getJobId().toString())
-            .userId(job.getUserId())
-            .pipelineId(job.getPipelineId())
-            .pipelineVersion(job.getPipelineVersion())
-            .timeSubmitted(job.getTimeSubmitted().toString())
-            .status(job.getStatus());
-    if (job.getTimeCompleted() != null) {
-      apiGetJobResponse.setTimeCompleted(job.getTimeCompleted().toString());
-    }
-    return apiGetJobResponse;
-  }
-
-  static ApiGetJobsResponse jobsToApi(List<Job> jobList) {
-    ApiGetJobsResponse apiResult = new ApiGetJobsResponse();
-
-    for (Job job : jobList) {
-      var apiJob = jobToApi(job);
-
-      apiResult.add(apiJob);
-    }
-
-    return apiResult;
-  }
 }