package io.teknek.nibiru;

import java.util.List;

import io.teknek.nibiru.personality.ColumnFamilyPersonality;
import io.teknek.nibiru.personality.KeyValuePersonality;
import io.teknek.nibiru.transport.Message;
import io.teknek.nibiru.transport.Response;

public class Coordinator {

  private static final String SYSTEM_KEYSPACE = "system";  
  private final Server server;
  private Destination destinationLocal;
  
  public Coordinator(Server server){
    this.server = server;
  }
  
  public void init(){
    destinationLocal = new Destination();
    destinationLocal.setDestinationId(server.getServerId().getU().toString());
  }
  
  public Response handle(Message message) {
    if (SYSTEM_KEYSPACE.equals(message.getKeyspace())) {
      return null;
    }
    Keyspace keyspace = server.getKeyspaces().get(message.getKeyspace());
    ColumnFamily columnFamily = keyspace.getColumnFamilies().get(message.getColumnFamily());
    List<Destination> destinations = keyspace.getKeyspaceMetadata().getRouter()
            .routesTo(message, server.getServerId(), keyspace);
    long timeoutInMs = determineTimeout(columnFamily, message);
    Keyspace ks = server.getKeyspaces().get(message.getKeyspace());
    if (ks == null){
      throw new RuntimeException(message.getKeyspace() + " is not found");
    }
    ColumnFamily cf = ks.getColumnFamilies().get(message.getColumnFamily());
    if (cf == null){
      throw new RuntimeException(message.getColumnFamily() + " is not found");
    }
    if (destinations.contains(destinationLocal)) {
      if (ColumnFamilyPersonality.COLUMN_FAMILY_PERSONALITY.equals(message.getRequestPersonality())) {
        return handleColumnFamilyPersonality(message, keyspace, columnFamily);
      } else if (KeyValuePersonality.KEY_VALUE_PERSONALITY.equals(message.getRequestPersonality())) { 
        return handleKeyValuePersonality(message, keyspace, columnFamily);
      } else {
        throw new UnsupportedOperationException(message.getRequestPersonality());
      }
    } else {
      throw new UnsupportedOperationException("We can not route messages. Yet!");
    }
  }
  
  private static long determineTimeout(ColumnFamily columnFamily, Message message){
    if (message.getPayload().containsKey("timeout")){
      return ((Number) message.getPayload().get("timeout")).longValue();
    } else {
      return columnFamily.getColumnFamilyMetadata().getOperationTimeoutInMs();
    }
  }
  
  private Response handleKeyValuePersonality(Message message, Keyspace ks, ColumnFamily cf){
    if (cf instanceof KeyValuePersonality){
      KeyValuePersonality personality = (KeyValuePersonality) cf;
      if (message.getPayload().get("type").equals("get")){
        String s = personality.get((String) message.getPayload().get("key"));
        Response r = new Response();
        r.put("payload", s);
        return r;
      } else if (message.getPayload().get("type").equals("put")){
        personality.put((String) message.getPayload().get("key"), 
                (String) message.getPayload().get("value"));
        return new Response();
      } else {
        throw new RuntimeException("Does not support this type of message");
      }
    }
    return null;
  }
  
  private Response handleColumnFamilyPersonality(Message message, Keyspace ks, ColumnFamily cf){
    if (cf instanceof ColumnFamilyPersonality){
      ColumnFamilyPersonality personality = (ColumnFamilyPersonality) cf;
      if (message.getPayload().get("type").equals("get")){
        Val v = personality.get(
                (String)message.getPayload().get("rowkey"),
                (String) message.getPayload().get("column"));
        Response r = new Response();
        r.put("payload", v);
        return r;
      } else if (message.getPayload().get("type").equals("put")) {
        Long l = ((Long) message.getPayload().get("ttl"));
        if (l == null){
          personality.put(
                (String) message.getPayload().get("rowkey"),
                (String) message.getPayload().get("column"),
                (String) message.getPayload().get("value"),
                ((Number) message.getPayload().get("time")).longValue());
          return new Response();
        } else {
          personality.put(
                  (String) message.getPayload().get("rowkey"),
                  (String) message.getPayload().get("column"),
                  (String) message.getPayload().get("value"),
                  ((Number) message.getPayload().get("time")).longValue(), l);
          return new Response();
        }
      } else if (message.getPayload().get("type").equals("delete")) { 
        personality.delete(
                (String) message.getPayload().get("rowkey"),
                (String) message.getPayload().get("column"),
                ((Number) message.getPayload().get("time")).longValue());
        return new Response();
      } else {
        throw new RuntimeException("Does not support this type of message");
      }
      
    } else {
      throw new RuntimeException("Does not support this personality");
    }
  }
}

@@ -754,24 +754,25 @@
 null;%0A    %7D%0A
+%0A
     Keyspace
@@ -841,657 +841,528 @@
     
-ColumnFamily columnFamily = keyspace.getColumnFamilies().get(message.getColumnFamily());%0A    List%3CDestination%3E destinations = keyspace.getKeyspaceMetadata().getRouter()%0A            .routesTo(message, server.getServerId(), keyspace);%0A    long timeoutInMs = determineTimeout(columnFamily, message);%0A    Keyspace ks = server.getKeyspaces().get(message.getKeyspace());%0A    if (ks == null)%7B%0A      throw new RuntimeException(message.getKeyspace() + %22 is not found%22);%0A    %7D%0A    ColumnFamily cf = ks.getColumnFamilies().get(message.getColumnFamily());%0A    if (cf == null)%7B%0A      throw new RuntimeException(message.getColumnFamily() + %22 is not found%22);%0A    %7D
+if (keyspace == null)%7B%0A      throw new RuntimeException(message.getKeyspace() + %22 is not found%22);%0A    %7D%0A    ColumnFamily columnFamily = keyspace.getColumnFamilies().get(message.getColumnFamily());%0A    if (columnFamily == null)%7B%0A      throw new RuntimeException(message.getColumnFamily() + %22 is not found%22);%0A    %7D%0A    List%3CDestination%3E destinations = keyspace.getKeyspaceMetadata().getRouter()%0A            .routesTo(message, server.getServerId(), keyspace);%0A    long timeoutInMs = determineTimeout(columnFamily, message);
 %0A   

Remove redundant code