package io.paradoxical.cassieq.dataAccess;

import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.Timer;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.querybuilder.QueryBuilder;
import com.godaddy.logging.Logger;
import com.google.common.collect.ImmutableList;
import com.google.inject.Inject;
import com.google.inject.assistedinject.Assisted;
import io.paradoxical.cassieq.dataAccess.interfaces.MessageDeleterJobProcessor;
import io.paradoxical.cassieq.dataAccess.interfaces.MonotonicRepository;
import io.paradoxical.cassieq.dataAccess.interfaces.PointerRepository;
import io.paradoxical.cassieq.dataAccess.interfaces.QueueRepository;
import io.paradoxical.cassieq.factories.DataContextFactory;
import io.paradoxical.cassieq.factories.MonotonicRepoFactory;
import io.paradoxical.cassieq.factories.PointerRepoFactory;
import io.paradoxical.cassieq.model.BucketSize;
import io.paradoxical.cassieq.model.GenericMessagePointer;
import io.paradoxical.cassieq.model.InvisibilityMessagePointer;
import io.paradoxical.cassieq.model.MessagePointer;
import io.paradoxical.cassieq.model.MonotonicIndex;
import io.paradoxical.cassieq.model.QueueId;
import io.paradoxical.cassieq.model.QueueStatus;
import org.apache.commons.collections4.CollectionUtils;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.stream.IntStream;

import static com.codahale.metrics.MetricRegistry.name;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.in;
import static com.godaddy.logging.LoggerFactory.getLogger;

public class MessageDeletorJobProcessorImpl implements MessageDeleterJobProcessor {
    private final Session session;
    private final MetricRegistry metricRegistry;
    private final DeletionJob job;
    private final PointerRepository pointerRepository;
    private final MonotonicRepository monotonicRepository;

    private Logger logger = getLogger(MessageDeletorJobProcessorImpl.class);
    private final QueueRepository queueRepository;

    @Inject
    public MessageDeletorJobProcessorImpl(
            Session session,
            DataContextFactory dataContextFactory,
            MetricRegistry metricRegistry,
            @Assisted DeletionJob job) {
        this.session = session;
        this.metricRegistry = metricRegistry;
        this.job = job;

        QueueId queueId = job.getQueueIdentifier();

        queueRepository = dataContextFactory.forAccount(job.getAccountName());
        pointerRepository = dataContextFactory.getPointerRepository(queueId);
        monotonicRepository = dataContextFactory.getMonotonicRepository(queueId);

        logger = logger.with(job);
    }

    /**
     * Handles deleting messages and marking the job as complete
     */
    @Override
    public void start() {
        try (Timer.Context ignored = getMetricTimer()) {
            final MessagePointer startPointer = getMinStartPointer(pointerRepository, job.getBucketSize());

            final MessagePointer endPointer = monotonicRepository.getCurrent();

            logger.debug("Starting deletion job");

            delete(startPointer, endPointer);

            logger.success("Deleted all messages!");

            monotonicRepository.deleteAll();

            logger.success("Deleted all monotons!");

            pointerRepository.deleteAll();

            logger.success("Deleted all pointers!");

            queueRepository.deleteQueueStats(job.getQueueStatsId());

            logger.success("Deleted all stats!");

            complete();

            logger.success("Complete");
        }
    }

    private void complete() {
        // queue definition is now totally inactive
        queueRepository.tryAdvanceQueueStatus(job.getQueueName(), QueueStatus.Inactive);

        queueRepository.deleteCompletionJob(job);
    }

    private MessagePointer getMinStartPointer(PointerRepository pointerRepository, BucketSize bucketSize) {
        final MonotonicIndex repairPointer = pointerRepository.getRepairCurrentBucketPointer().startOf(bucketSize);

        final InvisibilityMessagePointer currentInvisPointer = pointerRepository.getCurrentInvisPointer();

        return GenericMessagePointer.valueOf(Math.min(repairPointer.get(), currentInvisPointer.get()));
    }

    private void delete(MessagePointer from, MessagePointer to) {

        final int lastBucketNumber = to.toBucketPointer(job.getBucketSize()).get().intValue();
        final int firstBucketNumber = from.toBucketPointer(job.getBucketSize()).get().intValue();

        List<Integer> batchBucketsToDelete = new ArrayList<>();

        final Iterator<Integer> bucketRangeIterator = IntStream.range(firstBucketNumber, lastBucketNumber + 1)
                                                               .boxed()
                                                               .iterator();

        while (bucketRangeIterator.hasNext()) {
            batchBucketsToDelete.add(bucketRangeIterator.next());

            if (batchBucketsToDelete.size() == getDeleteBatchSize()) {
                deleteAllMessagesInBuckets(ImmutableList.copyOf(batchBucketsToDelete));

                batchBucketsToDelete.clear();
            }
        }

        if (!CollectionUtils.isEmpty(batchBucketsToDelete)) {
            deleteAllMessagesInBuckets(ImmutableList.copyOf(batchBucketsToDelete));
        }
    }

    private int getDeleteBatchSize() {
        return 100;
    }

    private void deleteAllMessagesInBuckets(final List<Integer> deletableBuckets) {
        final Statement delete = QueryBuilder.delete()
                                             .all()
                                             .from(Tables.Message.TABLE_NAME)
                                             .where(eq(Tables.Message.QUEUE_ID, job.getQueueIdentifier().get()))
                                             .and(in(Tables.Message.BUCKET_NUM, deletableBuckets));

        session.execute(delete);
    }

    private Timer.Context getMetricTimer() {

        try {
            return metricRegistry.timer(name("deletion", job.getQueueName().get())).time();
        }
        catch (Exception ex) {
            logger.warn(ex, "Error getting deletion metric");

            return new Timer().time();
        }
    }
}
