package uk.gov.indexer;

import com.google.common.collect.Lists;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import uk.gov.indexer.dao.DestinationDBUpdateDAO;
import uk.gov.indexer.dao.Entry;
import uk.gov.indexer.dao.FatEntry;
import uk.gov.indexer.dao.Item;
import uk.gov.indexer.dao.SourceDBQueryDAO;
import uk.gov.indexer.monitoring.CloudwatchRecordsProcessedUpdater;

import java.util.Collections;
import java.util.List;
import java.util.Optional;

public class IndexerTask implements Runnable {
    private final Logger LOGGER = LoggerFactory.getLogger(IndexerTask.class);

    private final String register;
    private final DestinationDBUpdateDAO destinationDBUpdateDAO;
    private final SourceDBQueryDAO sourceDBQueryDAO;
    private final Optional<CloudwatchRecordsProcessedUpdater> cloudwatchUpdater;

    public IndexerTask(Optional<CloudwatchRecordsProcessedUpdater> cloudwatchUpdater, String register, SourceDBQueryDAO sourceDBQueryDAO, DestinationDBUpdateDAO destinationDBUpdateDAO) {
        this.register = register;
        this.sourceDBQueryDAO = sourceDBQueryDAO;
        this.destinationDBUpdateDAO = destinationDBUpdateDAO;
        this.cloudwatchUpdater = cloudwatchUpdater;
    }

    @Override
    public void run() {
        try {
            LOGGER.info("Starting update for: " + register);
            update();
            LOGGER.info("Finished for register: " + register);
        } catch (Throwable e) {
            LOGGER.error(ExceptionFormatter.formatExceptionAsString(e));
            throw e;
        }
    }

    protected void update() {
        List<FatEntry> fatEntries = fetchNewFatEntries();

        if (fatEntries.isEmpty()) {
            updateCloudWatch(0);
        } else {
            do {
                int totalNewEntries = fatEntries.size();

                LOGGER.info(String.format("Register '%s': Found '%d' new entries in entries table.", register, totalNewEntries));

                destinationDBUpdateDAO.writeEntriesInBatch(register, fatEntries);

                LOGGER.info(String.format("Register '%s': Copied '%d' entries in ordered_entry_index from index '%d'.", register, totalNewEntries, fatEntries.get(0).serial_number));

                updateCloudWatch(totalNewEntries);

                LOGGER.info(String.format("Register '%s': Notified Cloudwatch about '%d' entries processed.", register, totalNewEntries));
            } while (!(fatEntries = fetchNewFatEntries()).isEmpty());
        }

        List<Entry> entries = fetchNewEntries();

        if (!entries.isEmpty()) {
            do {
                int totalNewEntries = entries.size();
                List<Item> items = fetchItemsByHex(Lists.transform(entries, Entry::getItemHash));

                LOGGER.info(String.format("Register '%s': Found '%d' new entries in entry table.", register, totalNewEntries));

                destinationDBUpdateDAO.writeEntriesAndItemsInBatch(register, entries, items);
            } while (!(entries = fetchNewEntries()).isEmpty());
        }
    }

    private void updateCloudWatch(final int totalEntriesWritten) {
        cloudwatchUpdater.ifPresent(cwu -> cwu.update(totalEntriesWritten));
    }

    private List<FatEntry> fetchNewFatEntries() {
        if (!sourceDBQueryDAO.entriesTableExists()) {
            return Collections.emptyList();
        }
        return sourceDBQueryDAO.read(destinationDBUpdateDAO.lastReadSerialNumber());
    }

    private List<Entry> fetchNewEntries() {
        return sourceDBQueryDAO.readEntries(destinationDBUpdateDAO.lastReadEntryNumber());
    }

    private List<Item> fetchItemsByHex(List<String> itemHexValues) {
        return sourceDBQueryDAO.readItems(itemHexValues);
    }
}

@@ -1580,24 +1580,122 @@
  update() %7B%0A
+        updateFatEntries();%0A        updateEntries();%0A    %7D%0A%0A    private void updateFatEntries() %7B%0A
         List
@@ -1769,32 +1769,78 @@
 es.isEmpty()) %7B%0A
+            if (!isMigrationComplete()) %7B%0A    
             upda
@@ -1856,16 +1856,30 @@
 tch(0);%0A
+            %7D%0A
         
@@ -2626,24 +2626,65 @@
 ;%0A        %7D%0A
+    %7D%0A%0A    private void updateEntries() %7B
 %0A        Lis
@@ -2734,17 +2734,16 @@
     if (
-!
 entries.
@@ -2747,32 +2747,141 @@
 es.isEmpty()) %7B%0A
+            if (isMigrationComplete()) %7B%0A                updateCloudWatch(0);%0A            %7D%0A        %7D else %7B%0A
             do %7B
@@ -3029,25 +3029,24 @@
 ItemHash));%0A
-%0A
             
@@ -3252,24 +3252,454 @@
 es, items);%0A
+                LOGGER.info(String.format(%22Register '%25s': Copied '%25d' entries in entry from index '%25d'.%22, register, totalNewEntries, entries.get(0).getEntryNumber()));%0A%0A                if (isMigrationComplete()) %7B%0A                    updateCloudWatch(totalNewEntries);%0A                    LOGGER.info(String.format(%22Register '%25s': Notified Cloudwatch about '%25d' entries processed.%22, register, totalNewEntries));%0A                %7D%0A
             
@@ -4442,11 +4442,171 @@
 );%0A    %7D
+%0A%0A    private boolean isMigrationComplete() %7B%0A        return destinationDBUpdateDAO.lastReadEntryNumber() %3E destinationDBUpdateDAO.lastReadSerialNumber();%0A    %7D
 %0A%7D%0A

Update Cloudwatch when migration is complete