class RepairRunResource {

  @POST
  public Response addRepairRun(
      @Context UriInfo uriInfo,
      @QueryParam("clusterName") Optional<String> clusterName,
      @QueryParam("keyspace") Optional<String> keyspace,
      @QueryParam("tables") Optional<String> tableNamesParam,
      @QueryParam("owner") Optional<String> owner,
      @QueryParam("cause") Optional<String> cause,
      @QueryParam("segmentCount") Optional<Integer> segmentCountPerNode,
      @QueryParam("repairParallelism") Optional<String> repairParallelism,
      @QueryParam("intensity") Optional<String> intensityStr,
      @QueryParam("incrementalRepair") Optional<String> incrementalRepairStr,
      @QueryParam("nodes") Optional<String> nodesToRepairParam,
      @QueryParam("datacenters") Optional<String> datacentersToRepairParam,
      @QueryParam("blacklistedTables") Optional<String> blacklistedTableNamesParam,
      @QueryParam("repairThreadCount") Optional<Integer> repairThreadCountParam) {

    try {
      final Response possibleFailedResponse
          = RepairRunResource.checkRequestForAddRepair(
          context,
          clusterName,
          keyspace,
          tableNamesParam,
          owner,
          segmentCountPerNode,
          repairParallelism,
          intensityStr,
          incrementalRepairStr,
          nodesToRepairParam,
          datacentersToRepairParam,
          blacklistedTableNamesParam,
          repairThreadCountParam);

      if (null != possibleFailedResponse) {
        return possibleFailedResponse;
      }

      Double intensity;
      if (intensityStr.isPresent()) {
        intensity = Double.parseDouble(intensityStr.get());
      } else {
        intensity = context.config.getRepairIntensity();
        LOG.debug("no intensity given, so using default value: {}", intensity);
      }

      boolean incrementalRepair;
      if (incrementalRepairStr.isPresent()) {
        incrementalRepair = Boolean.parseBoolean(incrementalRepairStr.get());
      } else {
        incrementalRepair = context.config.getIncrementalRepair();
        LOG.debug("no incremental repair given, so using default value: {}", incrementalRepair);
      }

      int segments = context.config.getSegmentCountPerNode();
      if (!incrementalRepair) {
        if (segmentCountPerNode.isPresent()) {
          LOG.debug(
              "using given segment count {} instead of configured value {}",
              segmentCountPerNode.get(),
              context.config.getSegmentCount());
          segments = segmentCountPerNode.get();
        }
      } else {
        // hijack the segment count in case of incremental repair
        segments = -1;
      }

      final Cluster cluster = context.storage.getCluster(Cluster.toSymbolicName(clusterName.get()));
      Set<String> tableNames;
      try {
        tableNames = repairRunService.getTableNamesBasedOnParam(cluster, keyspace.get(), tableNamesParam);
      } catch (IllegalArgumentException ex) {
        LOG.error(ex.getMessage(), ex);
        return Response.status(Response.Status.NOT_FOUND).entity(ex.getMessage()).build();
      }

      Set<String> blacklistedTableNames;
      try {
        blacklistedTableNames
            = repairRunService.getTableNamesBasedOnParam(cluster, keyspace.get(), blacklistedTableNamesParam);
      } catch (IllegalArgumentException ex) {
        LOG.error(ex.getMessage(), ex);
        return Response.status(Response.Status.NOT_FOUND).entity(ex.getMessage()).build();
      }

      final Set<String> nodesToRepair;
      try {
        nodesToRepair = repairRunService.getNodesToRepairBasedOnParam(cluster, nodesToRepairParam);
      } catch (IllegalArgumentException ex) {
        LOG.error(ex.getMessage(), ex);
        return Response.status(Response.Status.NOT_FOUND).entity(ex.getMessage()).build();
      }

      final Set<String> datacentersToRepair;
      try {
        datacentersToRepair = RepairRunService
            .getDatacentersToRepairBasedOnParam(cluster, datacentersToRepairParam);

      } catch (IllegalArgumentException ex) {
        LOG.error(ex.getMessage(), ex);
        return Response.status(Response.Status.NOT_FOUND).entity(ex.getMessage()).build();
      }


      RepairUnit.Builder builder = RepairUnit.builder()
              .clusterName(cluster.getName())
              .keyspaceName(keyspace.get())
              .columnFamilies(tableNames)
              .incrementalRepair(incrementalRepair)
              .nodes(nodesToRepair)
              .datacenters(datacentersToRepair)
              .blacklistedTables(blacklistedTableNames)
              .repairThreadCount(repairThreadCountParam.orElse(context.config.getRepairThreadCount()));

      final RepairUnit theRepairUnit = repairUnitService.getOrCreateRepairUnit(cluster, builder);

      if (theRepairUnit.getIncrementalRepair() != incrementalRepair) {
        String msg = String.format(
            "A repair unit %s already exist for the same cluster/keyspace/tables"
                + " but with a different incremental repair value. Requested value %s | Existing value: %s",
            theRepairUnit.getId(),
            incrementalRepair,
            theRepairUnit.getIncrementalRepair());

        return Response.status(Response.Status.BAD_REQUEST).entity(msg).build();
      }

      RepairParallelism parallelism = context.config.getRepairParallelism();
      if (repairParallelism.isPresent()) {
        LOG.debug(
            "using given repair parallelism {} instead of configured value {}",
            repairParallelism.get(),
            context.config.getRepairParallelism());

        parallelism = RepairParallelism.valueOf(repairParallelism.get().toUpperCase());
      }

      if (incrementalRepair) {
        parallelism = RepairParallelism.PARALLEL;
      }

      final RepairRun newRepairRun = repairRunService.registerRepairRun(
              cluster,
              theRepairUnit,
              cause,
              owner.get(),
              0,
              segments,
              parallelism,
              intensity);

      return Response.created(buildRepairRunUri(uriInfo, newRepairRun))
          .entity(new RepairRunStatus(newRepairRun, theRepairUnit, 0))
          .build();

    } catch (ReaperException e) {
      LOG.error(e.getMessage(), e);
      return Response.serverError().entity(e.getMessage()).build();
    }
  }

  public  RepairRunResource(AppContext context);

}