package org.atlasapi.system.bootstrap.workers;

import static com.google.common.base.Preconditions.checkNotNull;

import java.util.concurrent.ExecutionException;

import org.atlasapi.content.Content;
import org.atlasapi.content.ContentResolver;
import org.atlasapi.content.ContentWriter;
import org.atlasapi.entity.Id;
import org.atlasapi.entity.util.MissingResourceException;
import org.atlasapi.entity.util.Resolved;
import org.atlasapi.entity.util.WriteResult;
import org.atlasapi.messaging.ResourceUpdatedMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
import com.google.common.base.Throwables;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.Futures;
import com.metabroadcast.common.queue.RecoverableException;
import com.metabroadcast.common.queue.Worker;
import com.netflix.astyanax.connectionpool.exceptions.OperationTimeoutException;

public class ContentBootstrapWorker implements Worker<ResourceUpdatedMessage> {

    private static final Logger LOG = LoggerFactory.getLogger(ContentBootstrapWorker.class);

    private final ContentResolver contentResolver;
    private final ContentWriter writer;
    private final Timer messagesTimer;

    public ContentBootstrapWorker(ContentResolver contentResolver, ContentWriter writer, MetricRegistry metricsRegistry) {
        this.contentResolver = checkNotNull(contentResolver);
        this.writer = checkNotNull(writer);
        this.messagesTimer = (metricsRegistry != null ? checkNotNull(metricsRegistry.timer("ContentBootstrapWorker")) : null);
    }

    @Override
    public void process(ResourceUpdatedMessage message) throws RecoverableException {
        Id contentId = message.getUpdatedResource().getId();
        LOG.debug("Processing message on id {}, message: {}", contentId, message);

        Timer.Context time = messagesTimer.time();
        Resolved<Content> content = resolveContent(contentId);

        try {
            WriteResult<Content, Content> result =
                    writer.writeContent(content.getResources().first().get());
            LOG.debug("Bootstrapped content {}", result.toString());
            time.stop();
        } catch (Exception e) {
            String errorMsg = "Failed to bootstrap content " + message.getUpdatedResource();
            LOG.error(errorMsg, e);

            if (isRecoverable(e)) {
                throw new RecoverableException(errorMsg, e);
            }
            throw Throwables.propagate(e);
        }
    }

    private Resolved<Content> resolveContent(Id contentId) {
        try {
            return Futures.get(
                    contentResolver.resolveIds(ImmutableList.of(contentId)),
                    ExecutionException.class
            );
        } catch (Exception e) {
            throw Throwables.propagate(e);
        }
    }

    private boolean isRecoverable(Exception e) {
        // OperationTimeoutException could be caused by a C* tombstone issue that is
        // currently non-recoverable
        return Throwables.getCausalChain(e).stream()
                .noneMatch(
                        cause -> MissingResourceException.class.isInstance(cause)
                                || OperationTimeoutException.class.isInstance(cause)
                );
    }
}

@@ -316,66 +316,8 @@
 Id;%0A
-import org.atlasapi.entity.util.MissingResourceException;%0A
 impo
@@ -354,16 +354,16 @@
 solved;%0A
+
 import o
@@ -839,89 +839,8 @@
 ker;
-%0Aimport com.netflix.astyanax.connectionpool.exceptions.OperationTimeoutException;
 %0A%0Apu
@@ -2268,120 +2268,8 @@
  e);
-%0A%0A            if (isRecoverable(e)) %7B%0A                throw new RecoverableException(errorMsg, e);%0A            %7D
 %0A   
@@ -2636,32 +2636,32 @@
 s.propagate(e);%0A
+
         %7D%0A    %7D%0A
@@ -2663,452 +2663,7 @@
    %7D
-%0A%0A    private boolean isRecoverable(Exception e) %7B%0A        // OperationTimeoutException could be caused by a C* tombstone issue that is%0A        // currently non-recoverable%0A        return Throwables.getCausalChain(e).stream()%0A                .noneMatch(%0A                        cause -%3E MissingResourceException.class.isInstance(cause)%0A                                %7C%7C OperationTimeoutException.class.isInstance(cause)%0A                );%0A    %7D
 %0A%7D%0A

Disable retrying on any exception for content bootstrap