class Scheduler {

  public CompletableFuture<Void> createJob(@NonNull DyneinJobSpec jobSpec) {
    CompletableFuture<Void> ret = new CompletableFuture<>();
    log.info("Create new job {}", jobSpec);
    String serializedJobSpec = jobSpecTransformer.serializeJobSpec(jobSpec);

    switch (jobSpec.getSchedulePolicy().getType()) {
      case IMMEDIATE:
        {
          ret =
              asyncClient
                  .add(serializedJobSpec, jobSpec.getQueueName())
                  .whenComplete(
                      (it, ex) -> {
                        if (ex == null) {
                          log.info(
                              "DyneinJob {} triggered, enqueue to queue {}, type: IMMEDIATE",
                              jobSpec.getJobToken(),
                              jobSpec.getQueueName());
                          metrics.dispatchJob(jobSpec.getQueueName());
                        } else {
                          log.error(
                              "Failed to trigger DyneinJob {} to queue {}, type: IMMEDIATE",
                              jobSpec.getJobToken(),
                              jobSpec.getQueueName(),
                              ex);
                        }
                      });
          break;
        }
      case SCHEDULED:
        {
          ret =
              asyncClient
                  .add(serializedJobSpec, inboundQueueName)
                  .whenComplete(
                      (it, ex) -> {
                        if (ex == null) {
                          log.info(
                              "Added job spec {} to inbound job queue {}, with token {}",
                              jobSpec.getName(),
                              inboundQueueName,
                              jobSpec.getJobToken());
                          metrics.enqueueToInboundQueue(jobSpec);
                        } else {
                          log.error(
                              "Exception when adding job spec {} to inbound job queue {}, with token {}",
                              jobSpec.getName(),
                              inboundQueueName,
                              jobSpec.getJobToken(),
                              ex);
                        }
                      });
          break;
        }
      case SQS_DELAYED:
        {
          long delayMs = TimeUtils.getDelayMillis(jobSpec.getSchedulePolicy(), clock);

          if (delayMs > TimeUnit.MINUTES.toMillis(15)) {
            ret.completeExceptionally(
                new IllegalArgumentException(
                    "Delay "
                        + delayMs
                        + "ms is longer than 15 minutes, cannot be scheduled with SQS_DELAYED"));
          } else {
            ret =
                asyncClient
                    .add(
                        serializedJobSpec,
                        jobSpec.getQueueName(),
                        (int) TimeUnit.MILLISECONDS.toSeconds(delayMs))
                    .whenComplete(
                        (it, ex) -> {
                          if (ex == null) {
                            log.info(
                                "DyneinJob {} triggered, enqueue to queue {}, type: SQS_DELAYED",
                                jobSpec.getJobToken(),
                                jobSpec.getQueueName());
                            metrics.dispatchJob(jobSpec.getQueueName());
                          } else {
                            log.error(
                                "Failed to trigger DyneinJob {} to queue {}, type: SQS_DELAYED",
                                jobSpec.getJobToken(),
                                jobSpec.getQueueName(),
                                ex);
                          }
                        });
          }
          break;
        }
      default:
        ret.completeExceptionally(
            new UnsupportedOperationException("Unsupported Job Schedule Policy Type"));
    }
    return ret;
  }

}