package com.coding4people.mosquitoreport.api.indexers;

import java.io.ByteArrayInputStream;
import java.nio.charset.StandardCharsets;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

import javax.annotation.PostConstruct;
import javax.inject.Inject;
import javax.ws.rs.InternalServerErrorException;

import com.amazonaws.services.cloudsearchdomain.AmazonCloudSearchDomain;
import com.amazonaws.services.cloudsearchdomain.AmazonCloudSearchDomainClient;
import com.amazonaws.services.cloudsearchdomain.model.SearchRequest;
import com.amazonaws.services.cloudsearchdomain.model.UploadDocumentsRequest;
import com.amazonaws.services.cloudsearchv2.AmazonCloudSearch;
import com.amazonaws.services.cloudsearchv2.model.DescribeDomainsRequest;
import com.amazonaws.services.cloudsearchv2.model.DomainStatus;
import com.amazonaws.services.cloudsearchv2.model.ServiceEndpoint;
import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBTable;
import com.coding4people.mosquitoreport.api.Env;
import com.coding4people.mosquitoreport.api.models.WithGuid;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;

abstract public class Indexer<T extends WithGuid> {
    @Inject
    AmazonCloudSearch amazonCloudSearch;

    @Inject
    Env env;

    AmazonCloudSearchDomain domain;

    abstract protected Class<T> getType();

    public void index(T item) {
        this.index(Arrays.asList(item));
    }

    // TODO send to SQS in order to handle it asynchronously
    public void index(List<T> items) {
        try {
            String json = new ObjectMapper().writeValueAsString(items.stream().map(item -> {
                Map<String, Object> map = new HashMap<>();
                map.put("type", "add");
                map.put("id", item.getGuid());
                map.put("fields", item);
                return map;
            }).collect(Collectors.toList()));
            
            domain.uploadDocuments(new UploadDocumentsRequest().withContentType("application/json")
                    .withContentLength(Long.valueOf(json.length()))
                    .withDocuments(new ByteArrayInputStream(json.getBytes(StandardCharsets.UTF_8))));
        } catch (JsonProcessingException e) {
            // TODO log indexer errors
            e.printStackTrace();
        }
    }

    public Object search(String query) {
        return domain.search(new SearchRequest().withQuery(query));
    }

    @PostConstruct
    protected void postConstruct() {
        DescribeDomainsRequest describeDomainsRequest = new DescribeDomainsRequest().withDomainNames(getDomainName());
        List<DomainStatus> list = amazonCloudSearch.describeDomains(describeDomainsRequest).getDomainStatusList();

        if (list.isEmpty()) {
            throw new InternalServerErrorException("Could not find CloudSearch domain: " + getDomainName());
        }

        ServiceEndpoint searchService = list.get(0).getSearchService();

        if (searchService.getEndpoint() == null) {
            throw new InternalServerErrorException("Could not find CloudSearch domain: " + getDomainName());
        }

        domain = new AmazonCloudSearchDomainClient();
        domain.setEndpoint(searchService.getEndpoint());
    }

    protected String getDomainName() {
        return getPrefix() + getType().getAnnotation(DynamoDBTable.class).tableName();
    }

    protected String getPrefix() {
        return Optional.ofNullable(env.get("MOSQUITO_REPORT_CLOUDSEARCH_DOMAIN_PREFIX")).orElse("localhost") + "-";
    }
}

@@ -251,16 +251,138 @@
 tional;%0A
+import java.util.concurrent.Callable;%0Aimport java.util.concurrent.ExecutorService;%0Aimport java.util.concurrent.Executors;%0A
 import j
@@ -473,24 +473,55 @@
 ect.Inject;%0A
+import javax.inject.Singleton;%0A
 import javax
@@ -1416,16 +1416,27 @@
 apper;%0A%0A
+@Singleton%0A
 abstract
@@ -1557,16 +1557,117 @@
 Env env;
+%0A    %0A    //TODO create factory in order to inject thread configuration%0A    ExecutorService executor;
 %0A%0A    Am
@@ -1886,33 +1886,356 @@
 sly%0A    
-public void index
+// TODO implement recover and fallback%0A    public void index(List%3CT%3E items) %7B%0A        executor.submit(new Callable%3CVoid%3E() %7B%0A            @Override%0A            public Void call() throws Exception %7B%0A                indexSync(items);%0A                %0A                return null;%0A            %7D%0A        %7D);%0A    %7D%0A    %0A    private void indexSync
 (List%3CT%3E
@@ -3066,16 +3066,73 @@
 ing 
-query) %7B
+latlonnw, String latlonse) %7B%0A        //TODO avoid query injection
 %0A   
@@ -3187,21 +3187,69 @@
 thQuery(
-query
+%22fq=location:%5B'%22 + latlonnw + %22','%22 + latlonse + %22'%5D%22
 ));%0A    
@@ -3303,24 +3303,114 @@
 nstruct() %7B%0A
+        //TODO get from env%0A        executor = Executors.newFixedThreadPool(10);%0A        %0A
         Desc

Index async