package org.gbif.crawler.dwca.downloader;

import org.gbif.api.model.crawler.CrawlJob;
import org.gbif.api.model.crawler.FinishReason;
import org.gbif.api.model.crawler.ProcessState;
import org.gbif.common.messaging.api.MessagePublisher;
import org.gbif.common.messaging.api.messages.DwcaDownloadFinishedMessage;
import org.gbif.crawler.common.CrawlConsumer;
import org.gbif.crawler.constants.CrawlerNodePaths;
import org.gbif.utils.HttpUtil;

import java.io.File;
import java.io.IOException;
import java.util.Date;
import java.util.UUID;

import com.yammer.metrics.Metrics;
import com.yammer.metrics.core.Counter;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.http.HttpStatus;
import org.apache.http.StatusLine;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.gbif.crawler.common.ZookeeperUtils.createOrUpdate;
import static org.gbif.crawler.common.ZookeeperUtils.updateCounter;
import static org.gbif.crawler.common.ZookeeperUtils.updateDate;
import static org.gbif.crawler.constants.CrawlerNodePaths.FINISHED_REASON;
import static org.gbif.crawler.constants.CrawlerNodePaths.PAGES_CRAWLED;
import static org.gbif.crawler.constants.CrawlerNodePaths.PROCESS_STATE_CHECKLIST;
import static org.gbif.crawler.constants.CrawlerNodePaths.PROCESS_STATE_OCCURRENCE;

/**
 * Consumer of the crawler queue that runs the actual dwc archive download and emits a DwcaDownloadFinishedMessage
 * when done.
 */
public class DwcaCrawlConsumer extends CrawlConsumer {

  private static final Logger LOG = LoggerFactory.getLogger(DwcaCrawlConsumer.class);
  public static final String DWCA_SUFFIX = ".dwca";

  private final File archiveRepository;
  private final Counter startedDownloads = Metrics.newCounter(DownloaderService.class, "startedDownloads");
  private final Counter failedDownloads = Metrics.newCounter(DownloaderService.class, "failedDownloads");
  private final Counter notModified = Metrics.newCounter(DownloaderService.class, "notModified");
  private final HttpUtil client = new HttpUtil(HttpUtil.newMultithreadedClient(10 * 60 * 1000, 20, 2));
  private final RetryPolicy retryPolicy = new ExponentialBackoffRetry(100, 10);

  public DwcaCrawlConsumer(CuratorFramework curator, MessagePublisher publisher, File archiveRepository) {
    super(curator, publisher);
    this.archiveRepository = archiveRepository;
    if (!archiveRepository.exists() && !archiveRepository.isDirectory()) {
      throw new IllegalArgumentException(
        "Archive repository needs to be an existing directory: " + archiveRepository.getAbsolutePath());
    }
    if (!archiveRepository.canWrite()) {
      throw new IllegalArgumentException(
        "Archive repository directory not writable: " + archiveRepository.getAbsolutePath());
    }
  }

  @Override
  protected void crawl(UUID datasetKey, CrawlJob crawlJob) throws Exception {
    // The started crawl is declared in zookeeper by putting the date into
    // /crawls/UUID/startedCrawling
    updateDate(curator, datasetKey, CrawlerNodePaths.STARTED_CRAWLING);
    startedDownloads.inc();

    // we keep the compressed file forever and use it to retrieve the last modified for conditional gets
    final File localFile = new File(archiveRepository, datasetKey + DWCA_SUFFIX);

    try {
      LOG.debug("Start download of dwc archive from {} to {}", crawlJob.getTargetUrl(), localFile);
      StatusLine status = client.downloadIfModifiedSince(crawlJob.getTargetUrl().toURL(), localFile);

      if (status.getStatusCode() == HttpStatus.SC_NOT_MODIFIED) {
        notModified(datasetKey);

      } else if (HttpUtil.success(status)) {
        success(datasetKey, crawlJob);

      } else {
        failedDownloads.inc();
        throw new IllegalStateException("HTTP " + status.getStatusCode() +
                                        ". Failed to download DwC-A for dataset " + datasetKey +
                                        " from " + crawlJob.getTargetUrl());
      }

    } catch (IOException e) {
      failedDownloads.inc();
      LOG.error("Failed to download dwc archive for dataset [{}] from [{}]", crawlJob.getDatasetKey(),
        crawlJob.getTargetUrl(), e);
      throw new RuntimeException(e);
    }

    // finished crawl
    updateDate(curator, datasetKey, CrawlerNodePaths.FINISHED_CRAWLING);
  }

  private void notModified(UUID datasetKey) {
    notModified.inc();
    LOG.info("DwC-A for dataset [{}] not modified. Crawl finished", datasetKey);
    // If the archive wasn't modified we are done processing so we need to update ZooKeeper to reflect this
    createOrUpdate(curator, datasetKey, FINISHED_REASON, FinishReason.NOT_MODIFIED);
    // we dont know the kind of dataset so we just put both states to finish
    createOrUpdate(curator, datasetKey, PROCESS_STATE_OCCURRENCE, ProcessState.FINISHED);
    createOrUpdate(curator, datasetKey, PROCESS_STATE_CHECKLIST, ProcessState.FINISHED);
  }

  private void success(UUID datasetKey, CrawlJob crawlJob) {
    LOG.info("Successfully downloaded new DwC-A for dataset [{}]", datasetKey);
    updateCounter(curator, datasetKey, PAGES_CRAWLED, 1l);
    // send download success message
    try {
      DwcaDownloadFinishedMessage finishedMessage =
        new DwcaDownloadFinishedMessage(crawlJob.getDatasetKey(), crawlJob.getTargetUrl(), crawlJob.getAttempt(),
          new Date(), true);
      publisher.send(finishedMessage);
    } catch (IOException e) {
      LOG.error("Failed to send download finished message for crawl [{}]", crawlJob.getDatasetKey(), e);
    }
    // The crawl finished normally, processing still to run
    createOrUpdate(curator, datasetKey, FINISHED_REASON, FinishReason.NORMAL);
  }
}

@@ -3788,38 +3788,35 @@
       failed
-Downloads.inc(
+(datasetKey
 );%0A        t
@@ -3884,48 +3884,8 @@
 () +
-%0A                                       
  %22. 
@@ -3926,24 +3926,11 @@
 et %22
- + datasetKey +%0A
+%0A  
     
@@ -3963,17 +3963,30 @@
         
- 
++ datasetKey +
  %22 from 
@@ -4058,37 +4058,8 @@
 ) %7B%0A
-      failedDownloads.inc();%0A
     
@@ -4190,24 +4190,50 @@
 tUrl(), e);%0A
+      failed(datasetKey);%0A
       throw 
@@ -4257,23 +4257,35 @@
 ion(e);%0A
+%0A
     %7D
-%0A%0A
+ finally %7B%0A  
     // f
@@ -4298,16 +4298,18 @@
 d crawl%0A
+  
     upda
@@ -4373,16 +4373,429 @@
 WLING);%0A
+    %7D%0A  %7D%0A%0A  private void failed(UUID datasetKey) %7B%0A    failedDownloads.inc();%0A    createOrUpdate(curator, datasetKey, FINISHED_REASON, FinishReason.ABORT);%0A    // we dont know the kind of dataset so we just put both states to finish%0A    createOrUpdate(curator, datasetKey, PROCESS_STATE_OCCURRENCE, ProcessState.FINISHED);%0A    createOrUpdate(curator, datasetKey, PROCESS_STATE_CHECKLIST, ProcessState.FINISHED);%0A
   %7D%0A%0A  p

Mark failed downloads as finished