/*
 * Copyright (c) 2022 the Eclipse Milo Authors
 *
 * This program and the accompanying materials are made
 * available under the terms of the Eclipse Public License 2.0
 * which is available at https://www.eclipse.org/legal/epl-2.0/
 *
 * SPDX-License-Identifier: EPL-2.0
 */

package org.eclipse.milo.opcua.sdk.server.subscriptions;

import java.util.Date;
import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.TimeUnit;

import com.google.common.collect.Lists;
import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
import org.eclipse.milo.opcua.stack.core.types.structured.RequestHeader;
import org.eclipse.milo.opcua.stack.server.services.ServiceRequest;
import org.jetbrains.annotations.Nullable;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PublishQueue {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    private final LinkedList<ServiceRequest> serviceQueue = new LinkedList<>();

    private final LinkedHashMap<UInteger, WaitingSubscription> waitList = new LinkedHashMap<>();

    /**
     * Add a Publish {@link ServiceRequest} to the queue.
     * <p>
     * If there are wait-listed Subscriptions this request will be used immediately, otherwise it will be queued for
     * later use by a Subscription whose publish timer has expired and has notifications to send.
     *
     * @param service the Publish {@link ServiceRequest}.
     */
    public synchronized void addRequest(ServiceRequest service) {
        List<WaitingSubscription> waitingSubscriptions = Lists.newArrayList(waitList.values());

        if (waitingSubscriptions.isEmpty()) {
            serviceQueue.add(service);

            logger.debug(
                "Queued PublishRequest requestHandle={}, size={}",
                service.getRequest().getRequestHeader().getRequestHandle(),
                serviceQueue.size()
            );
        } else {
            logger.debug("{} subscriptions waiting", waitingSubscriptions.size());

            WaitingSubscription subscription = null;

            int maxPriority = 0;
            long minWaitingSince = Long.MAX_VALUE;

            for (WaitingSubscription waiting : waitingSubscriptions) {
                final int priority = waiting.getSubscription().getPriority();
                final long waitingSince = waiting.getWaitingSince().getTime();

                logger.debug(
                    "subscription id={} priority={} waitingSince={}",
                    waiting.getSubscription().getId(),
                    priority,
                    waitingSince
                );

                if (priority > maxPriority) {
                    maxPriority = priority;
                    minWaitingSince = Long.MAX_VALUE;
                }
                if (priority >= maxPriority && waitingSince < minWaitingSince) {
                    minWaitingSince = waitingSince;
                    subscription = waiting;

                    logger.debug(
                        "subscription id={} priority={} now next in line",
                        waiting.getSubscription().getId(),
                        priority
                    );
                }
            }

            if (subscription != null) {
                waitList.remove(subscription.subscription.getId());

                logger.debug(
                    "delivering PublishRequest to subscription id={} priority={}",
                    subscription.getSubscription().getId(),
                    subscription.getSubscription().getPriority()
                );

                final WaitingSubscription ws = subscription;

                service.getServer().getConfig().getExecutor().execute(
                    () -> ws.subscription.onPublish(service)
                );
            } else {
                serviceQueue.add(service);
            }
        }
    }

    /**
     * Add a subscription to the wait list.
     * <p>
     * A subscription should be added to the wait list when either:
     * <p>
     * a) The previous Publish response indicated that there were still more Notifications ready to be transferred and
     * there were no more Publish requests queued to transfer them.
     * <p>
     * b) The publishing timer of a Subscription expired and there were either Notifications to be sent or a keep-alive
     * Message to be sent.
     *
     * @param subscription the subscription to wait-list.
     */
    public synchronized void addSubscription(Subscription subscription) {
        if (waitList.isEmpty() && !serviceQueue.isEmpty()) {
            ServiceRequest request = poll();

            if (request != null) {
                request.getServer().getConfig().getExecutor().execute(
                    () ->
                        subscription.onPublish(request)
                );
            } else {
                waitList.putIfAbsent(subscription.getId(), new WaitingSubscription(subscription));
            }
        } else {
            waitList.putIfAbsent(subscription.getId(), new WaitingSubscription(subscription));
        }
    }

    public synchronized boolean isEmpty() {
        return serviceQueue.isEmpty();
    }

    public synchronized boolean isNotEmpty() {
        return !isEmpty();
    }

    public synchronized boolean isWaitListEmpty() {
        return waitList.isEmpty();
    }

    @Nullable
    public synchronized ServiceRequest poll() {
        long nowNanos = System.nanoTime();

        while (true) {
            ServiceRequest serviceRequest = serviceQueue.poll();

            if (serviceRequest == null) {
                return null;
            } else {
                RequestHeader requestHeader = serviceRequest
                    .getRequest()
                    .getRequestHeader();

                long millisSinceReceived = TimeUnit.MILLISECONDS.convert(
                    nowNanos - serviceRequest.getReceivedAtNanos(),
                    TimeUnit.NANOSECONDS
                );

                long timeoutHint = requestHeader.getTimeoutHint().longValue();

                if (timeoutHint == 0 || millisSinceReceived < timeoutHint) {
                    return serviceRequest;
                } else {
                    logger.debug(
                        "Discarding expired PublishRequest requestHandle={} timestamp={} timeoutHint={}",
                        serviceRequest.getRequest().getRequestHeader().getRequestHandle(),
                        requestHeader.getTimestamp().getJavaDate(),
                        timeoutHint
                    );

                    serviceRequest.setServiceFault(StatusCodes.Bad_Timeout);
                }
            }
        }
    }

    /**
     * Get the number of queued Publish ServiceRequests.
     *
     * @return the number of queued Publish ServiceRequests.
     */
    public synchronized int size() {
        return serviceQueue.size();
    }

    public static class WaitingSubscription {

        private final Date waitingSince = new Date();

        private final Subscription subscription;

        public WaitingSubscription(Subscription subscription) {
            this.subscription = subscription;
        }

        public Subscription getSubscription() {
            return subscription;
        }

        public Date getWaitingSince() {
            return waitingSince;
        }

    }

}
