package com.twitter.mesos.scheduler.httphandlers;

import com.google.common.base.Predicates;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.google.inject.Inject;
import com.twitter.common.base.Closure;
import com.twitter.common.net.http.handlers.StringTemplateServlet;
import com.twitter.mesos.Tasks;
import com.twitter.mesos.gen.ScheduleStatus;
import static com.twitter.mesos.gen.ScheduleStatus.*;
import com.twitter.mesos.gen.TaskQuery;
import com.twitter.mesos.gen.TrackedTask;
import com.twitter.mesos.scheduler.SchedulerCore;
import org.antlr.stringtemplate.StringTemplate;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.util.Collections;
import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * HTTP interface to view information about a job in the mesos scheduler.
 *
 * @author wfarner
 */
public class SchedulerzJob extends StringTemplateServlet {

  private static final String USER_PARAM = "user";
  private static final String JOB_PARAM = "job";
  // TODO(wfarner): Allow filtering by task status.
  private static final String STATUS_FILTER_PARAM = "status";

  private static final Map<ScheduleStatus, Set<ScheduleStatus>> FILTER_MAP =
      ImmutableMap.<ScheduleStatus, Set<ScheduleStatus>>builder()
        .put(PENDING, Sets.newHashSet(PENDING))
        .put(RUNNING, Sets.newHashSet(STARTING, RUNNING))
        .put(FINISHED, Sets.newHashSet(KILLED, KILLED_BY_CLIENT, FINISHED))
        .put(FAILED, Sets.newHashSet(LOST, NOT_FOUND, FAILED))
      .build();

  @Inject private SchedulerCore scheduler;

  @Inject
  public SchedulerzJob(@CacheTemplates boolean cacheTemplates) {
    super("schedulerzjob", cacheTemplates);
  }

  @Override
  protected void doGet(final HttpServletRequest req, HttpServletResponse resp)
      throws ServletException, IOException {
    writeTemplate(resp, new Closure<StringTemplate>() {
      @Override public void execute(StringTemplate template) {
        String user = req.getParameter(USER_PARAM);
        if (user == null) {
          template.setAttribute("exception", "Please specify a user.");
          return;
        }
        String job = req.getParameter(JOB_PARAM);
        if (job == null) {
          template.setAttribute("exception", "Please specify a job.");
          return;
        }

        String filterArg = req.getParameter(STATUS_FILTER_PARAM);
        ScheduleStatus statusFilter = null;
        if (filterArg != null) {
          try {
            statusFilter = ScheduleStatus.valueOf(filterArg.toUpperCase());
          } catch (IllegalArgumentException e) {
            template.setAttribute("exception", "Invalid status type: " + filterArg);
          }
        }

        template.setAttribute("user", user);
        template.setAttribute("job", job);

        TaskQuery query = new TaskQuery()
            .setOwner(user)
            .setJobName(job);

        if (statusFilter != null) {
          query.setStatuses(FILTER_MAP.get(statusFilter));
          template.setAttribute("activeTasks",
              Lists.newArrayList(scheduler.getTasks(query)));
        } else {
          template.setAttribute("activeTasks",
              Lists.newArrayList(scheduler.getTasks(query, Tasks.ACTIVE_FILTER)));

          List<TrackedTask> completedTasks = Lists.newArrayList(scheduler.getTasks(query,
                  Predicates.not(Tasks.ACTIVE_FILTER)));
          Collections.sort(completedTasks, new Comparator<TrackedTask>() {
              @Override public int compare(TrackedTask taskA, TrackedTask taskB) {
                // Sort in reverse chronological order.
                return taskB.getTaskId() - taskA.getTaskId();
              }
          });

          template.setAttribute("completedTasks", completedTasks);
        }
      }
    });
  }
}

@@ -1735,16 +1735,339 @@
 ild();%0A%0A
+  private static final Comparator%3CTrackedTask%3E REVERSE_CHRON_COMPARATOR =%0A      new Comparator%3CTrackedTask%3E() %7B%0A          @Override public int compare(TrackedTask taskA, TrackedTask taskB) %7B%0A            // Sort in reverse chronological order.%0A            return taskB.getTaskId() - taskA.getTaskId();%0A          %7D%0A      %7D;%0A%0A
   @Injec
@@ -3416,24 +3416,63 @@
 Name(job);%0A%0A
+        List%3CTrackedTask%3E activeTasks;%0A
         if (
@@ -3568,58 +3568,21 @@
     
-template.setAttribute(%22activeTasks%22,%0A             
+activeTasks =
  Lis
@@ -3623,17 +3623,16 @@
 (query))
-)
 ;%0A      
@@ -3656,58 +3656,21 @@
     
-template.setAttribute(%22activeTasks%22,%0A             
+activeTasks =
  Lis
@@ -3732,19 +3732,17 @@
 FILTER))
-);%0A
+;
 %0A       
@@ -3838,20 +3838,16 @@
         
-  
-  
 Predicat
@@ -3924,122 +3924,101 @@
 ks, 
-new Comparator%3CTrackedTask%3E() %7B%0A              @Override public int compare(TrackedTask taskA, TrackedTask taskB) %7B
+REVERSE_CHRON_COMPARATOR);%0A          template.setAttribute(%22completedTasks%22, completedTasks);
 %0A   
@@ -4018,32 +4018,34 @@
 s);%0A        
+%7D%0A
         
 // Sort in r
@@ -4036,217 +4036,123 @@
     
-// Sort in reverse chronological order.%0A                return taskB.getTaskId() - taskA.getTaskId();%0A              %7D%0A          %7D);%0A%0A          template.setAttribute(%22completedTasks%22, completedTasks);%0A        %7D
+Collections.sort(activeTasks, REVERSE_CHRON_COMPARATOR);%0A        template.setAttribute(%22activeTasks%22, activeTasks);
 %0A   

Add sorting to filtered task view http://reviewboard.local.twitter.com/r/17473/