package edu.berkeley.ground.ingest;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

import org.apache.avro.generic.GenericRecord;
import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.methods.PostMethod;
import org.codehaus.jackson.JsonNode;
import org.codehaus.jackson.map.ObjectMapper;

import edu.berkeley.ground.api.models.Node;
import edu.berkeley.ground.api.models.NodeVersion;
import edu.berkeley.ground.api.models.Tag;
import edu.berkeley.ground.api.versions.GroundType;
import gobblin.util.ConfigUtils;
import gobblin.writer.DataWriter;

import java.net.URLEncoder;

import com.typesafe.config.Config;


public class GroundWriter<D> implements DataWriter<GenericRecord> {

  HttpClient client = new HttpClient();
  private final String groundServerAddress;

  public GroundWriter(Config config) {

    this.groundServerAddress = ConfigUtils.getString(config, GroundWriterConfigurationKeys.GROUND_SERVER_ADDRESS, GroundWriterConfigurationKeys.DEFAULT_GROUND_SERVER_ADDRESS);

  }

  public void write(GenericRecord record) throws IOException {

    //extracting the metadata from the GenericRecord and storing it in the field tags
    Map<String, Tag> tags = new HashMap<String, Tag>();
    GroundType typeString = GroundType.STRING;
    GroundType typeInt = GroundType.INTEGER;
    Tag timeCreated = new Tag(" ", "timeCreated", record.get("timeCreated"), typeInt);
    Tag length = new Tag(" ", "fileLength", record.get("length"), typeInt);
    Tag modificationTime = new Tag(" ", "modificationTime", record.get("modificationTime"), typeInt);
    Tag owner = new Tag(" ", "owner", record.get("owner"), typeString);
    tags.put("timeCreated", timeCreated);
    tags.put("length", length);
    tags.put("modificationTime", modificationTime);
    tags.put("owner", owner);

    //extracting the name of the file to create the Node
    String name = record.get("name").toString();
    node(name, tags);

  }

  //method to create a Node given the name
  public void node(String name, Map<String, Tag> tags) throws IOException {

    Node node = new Node("id", name);
    ObjectMapper mapper = new ObjectMapper();
    String jsonString = mapper.writeValueAsString(node);
    String uri = groundServerAddress + name;
    String encodedUri = groundServerAddress + "nodes/" + URLEncoder.encode(name, "UTF-8");
    PostMethod post = new PostMethod(encodedUri);
    post.setRequestHeader("Content-type", "application/json");
    post.setRequestBody(jsonString);
    int statusCode = client.executeMethod(post);

    //getting the nodeId of the node created
    String text = post.getResponseBodyAsString();
    ObjectMapper objectMapper = new ObjectMapper();
    JsonNode jsonNode = objectMapper.readValue(text, JsonNode.class);
    JsonNode nodeId = jsonNode.get("id");
    String id = nodeId.asText();

    //creating a NodeVersion using the nodeId and the metadata stored in tags
    Map<String, String> refParameters = new HashMap<String, String>();
    nodeVersion("id", tags, null, null, refParameters, id);

  }

  //method to create the NodeVersion given the nodeId and the tags
  public void nodeVersion(String id, Map<String, Tag> tags, String structureVersionId, String reference, Map<String, String> referenceParameters, String nodeId) throws IOException {

    NodeVersion nodeVersion = new NodeVersion(id, tags, structureVersionId, reference, referenceParameters, nodeId);
    ObjectMapper mapper = new ObjectMapper();
    String jsonString = mapper.writeValueAsString(nodeVersion);
    String uri = groundServerAddress + "nodes/versions";
    PostMethod post = new PostMethod(uri);
    post.setRequestHeader("Content-type", "application/json");
    post.setRequestBody(jsonString);
    int statusCode = client.executeMethod(post);

  }

  @Override
  public void close() throws IOException {
  }

  @Override
  public long bytesWritten() throws IOException {

    return 0;

  }

  @Override
  public void cleanup() throws IOException {
  }

  @Override
  public void commit() throws IOException {
  }

  @Override
  public long recordsWritten() {

    return 0;

  }

}

@@ -1372,19 +1372,18 @@
 new Tag(
-%22 %22
+-1
 , %22timeC
@@ -1453,19 +1453,18 @@
 new Tag(
-%22 %22
+-1
 , %22fileL
@@ -1538,19 +1538,18 @@
 new Tag(
-%22 %22
+-1
 , %22modif
@@ -1632,11 +1632,10 @@
 Tag(
-%22 %22
+-1
 , %22o
@@ -2111,20 +2111,18 @@
 ew Node(
-%22id%22
+-1
 , name);
@@ -2824,20 +2824,18 @@
 %22);%0A    
-Stri
+lo
 ng id = 
@@ -2847,12 +2847,12 @@
 d.as
-Text
+Long
 ();%0A
@@ -3021,20 +3021,18 @@
 ion(
-%22id%22
+-1
 , tags, 
 null
@@ -3027,20 +3027,18 @@
 , tags, 
-null
+-1
 , null, 
@@ -3156,20 +3156,18 @@
 Version(
-Stri
+lo
 ng id, M
@@ -3188,20 +3188,18 @@
 %3E tags, 
-Stri
+lo
 ng struc
@@ -3272,20 +3272,18 @@
 meters, 
-Stri
+lo
 ng nodeI

fix ingest plugin