/*
 * Copyright 2012-2015 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package it.ozimov.springboot.mail.service.defaultimpl;

import it.ozimov.springboot.mail.configuration.EmailSchedulerProperties;
import it.ozimov.springboot.mail.logging.EmailLogRenderer;
import it.ozimov.springboot.mail.model.Email;
import it.ozimov.springboot.mail.model.EmailSchedulingData;
import it.ozimov.springboot.mail.model.InlinePicture;
import it.ozimov.springboot.mail.model.defaultimpl.DefaultEmailSchedulingData;
import it.ozimov.springboot.mail.model.defaultimpl.TemplateEmailSchedulingData;
import it.ozimov.springboot.mail.service.EmailService;
import it.ozimov.springboot.mail.service.PersistenceService;
import it.ozimov.springboot.mail.service.EmailSchedulerService;
import it.ozimov.springboot.mail.service.ServiceStatus;
import it.ozimov.springboot.mail.service.exception.CannotSendEmailException;
import it.ozimov.springboot.mail.utils.TimeUtils;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnExpression;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.PreDestroy;
import java.time.Duration;
import java.time.OffsetDateTime;
import java.time.temporal.ChronoUnit;
import java.util.Collection;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import static com.google.common.base.Preconditions.checkArgument;
import static it.ozimov.springboot.mail.service.defaultimpl.ConditionalExpression.SCHEDULER_IS_ENABLED;
import static java.lang.Math.max;
import static java.lang.Math.min;
import static java.util.Comparator.comparing;
import static java.util.Objects.nonNull;

/**
 * The class provides a {@linkplain EmailSchedulerService} implementation with priority queues and persistence.
 * <p>
 * Main logic for the thread wait-notify mechanism comes from {@see http://stackoverflow.com/a/8980307/1339429 }
 */
@Service("priorityQueueEmailSchedulerService")
@ConditionalOnExpression(SCHEDULER_IS_ENABLED)
@Slf4j
public class PriorityQueueEmailSchedulerService implements EmailSchedulerService {

    /**
     * millisecs elapsed form the call of the send method and the actual sending by SMTP server
     */
    protected static final Duration CONSUMER_CYCLE_LENGTH = Duration.of(1, ChronoUnit.SECONDS);

    /**
     * millisecs elapsed form the call of the send method and the actual sending by SMTP server
     */
    protected static final Duration RESUMER_CYCLE_LENGTH = Duration.of(5, ChronoUnit.SECONDS);

    private final int batchSize;
    private final int minInMemory;
    private final int maxInMemory;

    private volatile ServiceStatus serviceStatus = ServiceStatus.RUNNING;

    private AtomicLong timeOfNextScheduledMessage;

    private final PriorityQueueManager priorityQueueManager;

    private final EmailService emailService;

    private final Consumer consumer;

    private final Resumer resumer;

    private final ExecutorService executor = Executors.newFixedThreadPool(5);

    private Optional<PersistenceService> persistenceServiceOptional;

    private EmailLogRenderer emailLogRenderer;

    private final Lock schedulerLock = new ReentrantLock();

    @Autowired
    public PriorityQueueEmailSchedulerService(
            final EmailService emailService,
            final EmailSchedulerProperties emailSchedulerProperties,
            final Optional<PersistenceService> persistenceServiceOptional,
            final EmailLogRenderer emailLogRenderer) throws InterruptedException {

        this.emailService = emailService;
        this.persistenceServiceOptional = persistenceServiceOptional;
        this.emailLogRenderer = emailLogRenderer.registerLogger(log);

        timeOfNextScheduledMessage = new AtomicLong();

        batchSize = nonNull(emailSchedulerProperties.getPersistence()) ?
                emailSchedulerProperties.getPersistence().getDesiredBatchSize() : 0;
        minInMemory = nonNull(emailSchedulerProperties.getPersistence()) ?
                emailSchedulerProperties.getPersistence().getMinKeptInMemory() : 1;
        maxInMemory = nonNull(emailSchedulerProperties.getPersistence()) ?
                emailSchedulerProperties.getPersistence().getMaxKeptInMemory() : Integer.MAX_VALUE;

        final int numberOfPriorityLevels = emailSchedulerProperties.getPriorityLevels();
        priorityQueueManager = new PriorityQueueManager(numberOfPriorityLevels, persistenceServiceOptional.isPresent(),
                maxInMemory, CONSUMER_CYCLE_LENGTH);

        //CREATING EMAIL CONSUMER
        consumer = new Consumer();
        startConsumer();

        //CREATING EMAIL RESUMER
        if (this.persistenceServiceOptional.isPresent()) {
            resumer = new Resumer();
            startResumer();
        } else {
            resumer = null;
        }
    }

    @Override
    @Async
    public void schedule(@NonNull final Email mimeEmail, final int desiredPriorityLevel) {
        scheduleEmail(mimeEmail, TimeUtils.offsetDateTimeNow(), desiredPriorityLevel);
    }

    @Override
    @Async
    public void schedule(@NonNull final Email mimeEmail, @NonNull final OffsetDateTime scheduledDateTime, final int desiredPriorityLevel) {
        scheduleEmail(mimeEmail, scheduledDateTime, desiredPriorityLevel);
    }

    @Override
    @Async
    public void schedule(@NonNull final Email mimeEmail, final int desiredPriorityLevel, @NonNull final String template,
                         @NonNull final Map<String, Object> modelObject, final InlinePicture... inlinePictures) throws CannotSendEmailException {
        scheduleTemplateEmail(mimeEmail, TimeUtils.offsetDateTimeNow(), desiredPriorityLevel, template, modelObject, inlinePictures);
    }

    @Override
    @Async
    public void schedule(@NonNull final Email mimeEmail, @NonNull final OffsetDateTime scheduledDateTime, final int desiredPriorityLevel,
                         @NonNull final String template, @NonNull final Map<String, Object> modelObject, final InlinePicture... inlinePictures) throws CannotSendEmailException {
        scheduleTemplateEmail(mimeEmail, scheduledDateTime, desiredPriorityLevel, template, modelObject, inlinePictures);
    }

    private void scheduleEmail(final Email mimeEmail, final OffsetDateTime scheduledDateTime, final int desiredPriorityLevel) {
        checkPriorityLevel(desiredPriorityLevel);

        final int assignedPriorityLevel = normalizePriority(desiredPriorityLevel);
        final EmailSchedulingData emailSchedulingData = buildEmailSchedulingData(mimeEmail, scheduledDateTime, desiredPriorityLevel, assignedPriorityLevel);
        schedule(emailSchedulingData);

        emailLogRenderer.info("Scheduled email {} at UTC time {} with priority {}", mimeEmail, scheduledDateTime, desiredPriorityLevel);
        notifyConsumerIfCouldFire(scheduledDateTime);
    }

    private void scheduleTemplateEmail(final Email mimeEmail, final OffsetDateTime scheduledDateTime, final int desiredPriorityLevel,
                         final String template, final Map<String, Object> modelObject, final InlinePicture... inlinePictures) throws CannotSendEmailException {
        checkPriorityLevel(desiredPriorityLevel);

        final int assignedPriorityLevel = normalizePriority(desiredPriorityLevel);
        final EmailSchedulingData emailTemplateSchedulingData = buildEmailSchedulingData(mimeEmail, scheduledDateTime, desiredPriorityLevel, template, modelObject, assignedPriorityLevel, inlinePictures);
        schedule(emailTemplateSchedulingData);

        emailLogRenderer.info("Scheduled email {} at UTC time {} with priority {} with template", mimeEmail, scheduledDateTime, desiredPriorityLevel);
        notifyConsumerIfCouldFire(scheduledDateTime);
    }

    protected EmailSchedulingData buildEmailSchedulingData(@NonNull Email mimeEmail, @NonNull OffsetDateTime scheduledDateTime, int desiredPriorityLevel, int assignedPriorityLevel) {
        return DefaultEmailSchedulingData.defaultEmailSchedulingDataBuilder()
                .email(mimeEmail)
                .scheduledDateTime(scheduledDateTime)
                .assignedPriority(assignedPriorityLevel)
                .desiredPriority(desiredPriorityLevel)
                .build();
    }

    protected EmailSchedulingData buildEmailSchedulingData(@NonNull Email mimeEmail, @NonNull OffsetDateTime scheduledDateTime, int desiredPriorityLevel, @NonNull String template, @NonNull Map<String, Object> modelObject, int assignedPriorityLevel, InlinePicture[] inlinePictures) {
        return TemplateEmailSchedulingData.templateEmailSchedulingDataBuilder()
                .email(mimeEmail)
                .scheduledDateTime(scheduledDateTime)
                .assignedPriority(assignedPriorityLevel)
                .desiredPriority(desiredPriorityLevel)
                .template(template)
                .modelObject(modelObject)
                .inlinePictures(inlinePictures)
                .build();
    }

    protected synchronized void schedule(final EmailSchedulingData emailSchedulingData) {
        enqueueFromScheduler(emailSchedulingData);
        addToPersistenceLayer(emailSchedulingData);
        completeEnqueue();
    }

    protected synchronized void startResumer() throws InterruptedException {
        startAndWaitForWaitingState(resumer);
    }

    protected synchronized void startConsumer() throws InterruptedException {
        startAndWaitForWaitingState(consumer);
    }

    private void startAndWaitForWaitingState(final Thread thread) throws InterruptedException {
        thread.start();
        while (thread.getState() == Thread.State.RUNNABLE) {
            TimeUnit.MILLISECONDS.sleep(50);
        }
    }

    private void notifyConsumerIfCouldFire(@NonNull OffsetDateTime scheduledDateTime) {
        final boolean canFire = isTimeOfNextSchedulerMessageNotSet() || scheduledDateTime.toInstant().toEpochMilli() < timeOfNextScheduledMessage.get();
        if (canFire && consumer.enabled()) {
            executor.submit(() -> {
                synchronized (consumer) {
                    consumer.notify(); //the consumer, if waiting, is notified and can try to send next scheduled message
                }
            });
        }
    }

    // Returns true if the email is enqueued
    private boolean enqueueFromScheduler(final EmailSchedulingData emailSchedulingData) {
        return enqueue(emailSchedulingData, false);
    }


    private boolean enqueueFromPersistenceLayer(final EmailSchedulingData emailSchedulingData) {
        return enqueue(emailSchedulingData, true);
    }

    // Returns true if the email is enqueued
    private boolean enqueue(final EmailSchedulingData emailSchedulingData, final boolean isFromPersistenceLayer) {
        if (serviceStatus == ServiceStatus.RUNNING) {
            return priorityQueueManager.enqueue(emailSchedulingData, isFromPersistenceLayer);
        }
        return false;
    }

    private void completeEnqueue() {
        if (serviceStatus == ServiceStatus.RUNNING) {
            priorityQueueManager.completeEnqueue();
        }
    }

    protected void addToPersistenceLayer(final EmailSchedulingData emailSchedulingData) {
        if (serviceStatus == ServiceStatus.RUNNING) {
            log.debug("Adding to persistence layer");
            persistenceServiceOptional.ifPresent(
                    persistenceService ->
                            persistenceService.add(emailSchedulingData)
            );
        }
    }

    protected void deleteFromPersistenceLayer(final EmailSchedulingData emailSchedulingData) {
        //This part is disabled for now, I'm not sure that this could not provide overhead to the persistence layer
        if (serviceStatus == ServiceStatus.RUNNING) {
            persistenceServiceOptional.ifPresent(
                    persistenceService -> {
                        persistenceService.remove(emailSchedulingData.getId());
                        priorityQueueManager.completeDequeue();
                    }
            );
        }
    }

    protected void loadNextBatch() {
        if (serviceStatus == ServiceStatus.RUNNING) {
            persistenceServiceOptional.ifPresent(
                    persistenceService -> {
                        final int currentlyInMemory = currentlyInMemory();
                        if (currentlyInMemory < minInMemory) {
                            //Currently REDIS loads all from the beginning so we need to count also those currentlyInMemory. This must be improved.
                            final int expectedFromPersistenceLayer = Math.min(currentlyInMemory + batchSize, maxInMemory);
                            final Collection<EmailSchedulingData> emailSchedulingDataList =
                                    persistenceService.getNextBatch(expectedFromPersistenceLayer);
                            if (!emailSchedulingDataList.isEmpty()) {
                                enqueueBatch(emailSchedulingDataList);
                            }
                        }
                    }
            );
        }
    }

    protected void enqueueBatch(final Collection<EmailSchedulingData> emailSchedulingDataCollection) {
        if (!emailSchedulingDataCollection.isEmpty()) {
            EmailSchedulingData lastEmailSchedulingData = emailSchedulingDataCollection.stream()
                    .filter(Objects::nonNull)
                    .max((comparing(EmailSchedulingData::getScheduledDateTime))).get();
            if (serviceStatus == ServiceStatus.RUNNING) {
                int countAdded = 0;
                for (final EmailSchedulingData emailSchedulingData : emailSchedulingDataCollection) {
                    synchronized (this) {
                        if (enqueueFromPersistenceLayer(emailSchedulingData)) {
                            countAdded++;
                        }
                        completeEnqueue();
                    }
                }
                log.debug("Enqueued batch of {} emails of {} loaded from persistence layer.",
                        countAdded, emailSchedulingDataCollection.size());
            }

            if (nonNull(lastEmailSchedulingData)) {
                notifyConsumerIfCouldFire(lastEmailSchedulingData.getScheduledDateTime());
            }
        }
    }

    protected int normalizePriority(int priorityLevel) {
        //the priority level must be between 0 and numberOfPriorityLevels
        final int maxLevel = priorityQueueManager.numberOfLevels();
        if (priorityLevel > maxLevel) {
            log.warn("Scheduled email with priority level {}, while priority level {} was requested. Reason: max level exceeded",
                    maxLevel, priorityLevel);
        }
        return max(1, min(priorityLevel, maxLevel));
    }

    private Optional<EmailSchedulingData> dequeue() throws InterruptedException {
        Optional<EmailSchedulingData> emailSchedulingDataOptional = Optional.empty();
        timeOfNextScheduledMessage.set(0);
        boolean consumerEnabled = consumer.enabled();
        while (consumerEnabled && !emailSchedulingDataOptional.isPresent()) {
            if (consumer.enabled()) {
                //try to find a message in queue
                if (priorityQueueManager.hasElements()) {
                    emailSchedulingDataOptional = priorityQueueManager.dequeueNext(CONSUMER_CYCLE_LENGTH);
                }
                if (!emailSchedulingDataOptional.isPresent()) {
                    //no message was found, let's sleep, some message may arrive in the meanwhile
                    timeOfNextScheduledMessage.set(priorityQueueManager.millisToNextEmail());
                    if (consumer.enabled()) {
                        if (isTimeOfNextSchedulerMessageNotSet()) { //all the queues are empty
                            consumer.waitForNotify(); //the consumer starts waiting for a new email to be scheduled or
                        } else {
                            final long waitTime = timeOfNextScheduledMessage.get() - TimeUtils.now() - CONSUMER_CYCLE_LENGTH.toMillis();
                            if (waitTime > 0) {
                                consumer.waitForMillis(waitTime); //wait before sending the most imminent scheduled email
                            }
                        }
                    }
                }
            } else {
                consumerEnabled = false;
            }
        }
        //here emailSchedulingData is the message to send
        return emailSchedulingDataOptional;
    }

    private boolean isTimeOfNextSchedulerMessageNotSet() {
        return timeOfNextScheduledMessage.get() == 0L;
    }

    private void checkPriorityLevel(int priorityLevel) {
        checkArgument(priorityLevel > 0, "The priority level index cannot be negative");
    }

    private int queueIndex(final EmailSchedulingData emailSchedulingData) {
        return emailSchedulingData.getAssignedPriority() - 1;
    }

    private boolean canAddOneInMemory() {
        return !persistenceServiceOptional.isPresent() || currentlyInMemory() < maxInMemory;
    }

    private int currentlyInMemory() {
        return priorityQueueManager.currentlyInQueue();
    }

    @PreDestroy
    protected void cleanUp() throws Exception {
        log.info("Closing EmailScheduler");
        try {
            executor.shutdownNow();

            schedulerLock.lock();
            try {
                this.serviceStatus = ServiceStatus.CLOSING;
            } finally {
                schedulerLock.unlock();
            }

            log.debug("EMAIL SCHEDULER -- Closing PriorityQueueManager");
            priorityQueueManager.close();
            if (nonNull(resumer)) {
                log.debug("EMAIL SCHEDULER -- Closing Resumer");
                resumer.close();
            }
            log.debug("EMAIL SCHEDULER -- Closing Consumer");
            consumer.close();

        } catch (Exception e) {
            log.warn("An issue occurred while stopping EmailScheduler, it should be due to a thread interruption.", e);
        } finally {
            this.serviceStatus = ServiceStatus.CLOSED;
        }
        log.info("Closed EmailScheduler");
    }


    private class Consumer extends Thread {

        public Consumer() {
            super(PriorityQueueEmailSchedulerService.class.getSimpleName() + " -- " + Consumer.class.getSimpleName());
        }

        public void run() {
            log.info("Email scheduler consumer started");
            while (enabled()) {
                try {
                    final Optional<EmailSchedulingData> emailSchedulingDataOptional = dequeue();
                    if (enabled() && emailSchedulingDataOptional.isPresent()) {
                        final EmailSchedulingData emailSchedulingData = emailSchedulingDataOptional.get();
                        if (emailSchedulingData instanceof TemplateEmailSchedulingData) {
                            final TemplateEmailSchedulingData emailTemplateSchedulingData = (TemplateEmailSchedulingData) emailSchedulingData;
                            try {
                                emailService.send(emailTemplateSchedulingData.getEmail(),
                                        emailTemplateSchedulingData.getTemplate(),
                                        emailTemplateSchedulingData.getModelObject(),
                                        emailTemplateSchedulingData.getInlinePictures());
                            } catch (final CannotSendEmailException e) {
                                log.error("An error occurred while sending the email", e);
                            }
                        } else {
                            emailService.send(emailSchedulingData.getEmail());
                        }
                        if (enabled() && !persistenceServiceOptional.isPresent()) {
                            priorityQueueManager.completeDequeue();
                        }

                        if (enabled()) deleteFromPersistenceLayer(emailSchedulingData);
                    }
                } catch (final InterruptedException e) {
                    log.error("Email scheduler consumer interrupted", e);
                }
            }
            log.info("Email scheduler consumer stopped");

        }

        public boolean enabled() {
            return serviceStatus == ServiceStatus.RUNNING && !isInterrupted();
        }

        public synchronized void waitForNotify() throws InterruptedException {
            if (enabled()) {
                log.debug("Email scheduler consumer starts waiting");
                wait();
            }
        }

        public synchronized void waitForMillis(final long timeoutInMillis) throws InterruptedException {
            if (enabled()) {
                log.debug("Email scheduler consumer starts waiting for {} millis", timeoutInMillis);
                wait(timeoutInMillis);
            }
        }

        public void close() throws InterruptedException {
            try {
                if (!isInterrupted()) {
                    log.info("Interrupting email scheduler consumer");
                    interrupt();
                    synchronized (this) {
                        notify();
                    }
                    join();
                } else {
                    log.info("Email scheduler consumer already interrupted");
                }
            } catch (InterruptedException e) {
            }
        }

    }

    private class Resumer extends Thread {

        public Resumer() {
            super(PriorityQueueEmailSchedulerService.class.getSimpleName() + " -- " + Resumer.class.getSimpleName());
        }

        public void run() {
            if (persistenceServiceOptional.isPresent()) {
                log.info("Email scheduler resumer started");
                while (enabled()) {
                    try {
                        if (canAddOneInMemory()) {
                            if (enabled()) {
                                loadNextBatch();
                            }
                            if (enabled()) {
                                synchronized (this) {
                                    wait(RESUMER_CYCLE_LENGTH.toMillis());
                                }
                            }
                        }
                    } catch (final InterruptedException e) {
                        log.error("Email scheduler consumer interrupted", e);
                    }
                }
                log.info("Email scheduler resumer stopped");
            } else {
                log.warn("Email scheduler resumer won't start because there is no email PersistenceService.");
            }
        }

        public boolean enabled() {
            return serviceStatus == ServiceStatus.RUNNING && !isInterrupted();
        }

        public synchronized void close() throws InterruptedException {
            try {
                if (!isInterrupted()) {
                    log.info("Interrupting email scheduler resumer");
                    interrupt();
                    synchronized (this) {
                        notify();
                    }
                    join();
                } else {
                    log.info("Email scheduler resumer already interrupted");
                }
            } catch (InterruptedException e) {
            }
        }

    }

}