/*
 * Copyright (c) 2019 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.items;

import java.nio.ByteBuffer;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.atomic.AtomicBoolean;

import org.eclipse.milo.opcua.sdk.server.OpcUaServer;
import org.eclipse.milo.opcua.sdk.server.Session;
import org.eclipse.milo.opcua.sdk.server.api.EventItem;
import org.eclipse.milo.opcua.sdk.server.events.EventContentFilter;
import org.eclipse.milo.opcua.sdk.server.events.FilterContext;
import org.eclipse.milo.opcua.sdk.server.model.nodes.objects.BaseEventTypeNode;
import org.eclipse.milo.opcua.sdk.server.subscriptions.Subscription;
import org.eclipse.milo.opcua.stack.core.Identifiers;
import org.eclipse.milo.opcua.stack.core.StatusCodes;
import org.eclipse.milo.opcua.stack.core.UaException;
import org.eclipse.milo.opcua.stack.core.serialization.UaStructure;
import org.eclipse.milo.opcua.stack.core.types.builtin.ByteString;
import org.eclipse.milo.opcua.stack.core.types.builtin.DateTime;
import org.eclipse.milo.opcua.stack.core.types.builtin.ExtensionObject;
import org.eclipse.milo.opcua.stack.core.types.builtin.LocalizedText;
import org.eclipse.milo.opcua.stack.core.types.builtin.NodeId;
import org.eclipse.milo.opcua.stack.core.types.builtin.QualifiedName;
import org.eclipse.milo.opcua.stack.core.types.builtin.StatusCode;
import org.eclipse.milo.opcua.stack.core.types.builtin.Variant;
import org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.UInteger;
import org.eclipse.milo.opcua.stack.core.types.enumerated.MonitoringMode;
import org.eclipse.milo.opcua.stack.core.types.enumerated.TimestampsToReturn;
import org.eclipse.milo.opcua.stack.core.types.structured.ContentFilter;
import org.eclipse.milo.opcua.stack.core.types.structured.ContentFilterElementResult;
import org.eclipse.milo.opcua.stack.core.types.structured.EventFieldList;
import org.eclipse.milo.opcua.stack.core.types.structured.EventFilter;
import org.eclipse.milo.opcua.stack.core.types.structured.EventFilterResult;
import org.eclipse.milo.opcua.stack.core.types.structured.MonitoringFilter;
import org.eclipse.milo.opcua.stack.core.types.structured.ReadValueId;
import org.eclipse.milo.opcua.stack.core.types.structured.SimpleAttributeOperand;
import org.jetbrains.annotations.NotNull;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.uint;
import static org.eclipse.milo.opcua.stack.core.types.builtin.unsigned.Unsigned.ushort;
import static org.eclipse.milo.opcua.stack.core.util.ConversionUtil.l;

public class MonitoredEventItem extends BaseMonitoredItem<Variant[]> implements EventItem {

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

    private volatile EventFilter filter;
    private volatile EventFilterResult filterResult;
    private volatile boolean filterResultGood;

    private final AtomicBoolean eventOverflow = new AtomicBoolean(false);

    private final FilterContext filterContext;

    public MonitoredEventItem(
        OpcUaServer server,
        Session session,
        UInteger id,
        UInteger subscriptionId,
        ReadValueId readValueId,
        MonitoringMode monitoringMode,
        TimestampsToReturn timestamps,
        UInteger clientHandle,
        double samplingInterval,
        UInteger queueSize,
        boolean discardOldest
    ) {

        super(server, session, id, subscriptionId, readValueId, monitoringMode,
            timestamps, clientHandle, samplingInterval, queueSize, discardOldest);

        filterContext = new FilterContext() {
            @Override
            public OpcUaServer getServer() {
                return server;
            }

            @Override
            public Optional<Session> getSession() {
                return Optional.of(session);
            }
        };
    }

    @Override
    public void onEvent(BaseEventTypeNode eventNode) {
        try {
            if (filterResultGood) {
                ContentFilter whereClause = filter.getWhereClause();

                boolean matches = EventContentFilter.evaluate(
                    filterContext,
                    whereClause,
                    eventNode
                );

                if (matches) {
                    enqueue(selectEventFields(eventNode));
                }
            }
        } catch (UaException e) {
            logger.error("Filter evaluation failed: {}", e.getMessage(), e);
        }
    }


    @NotNull
    private Variant[] selectEventFields(BaseEventTypeNode eventNode) {
        SimpleAttributeOperand[] selectClauses = filter.getSelectClauses();

        if (selectClauses != null) {
            return EventContentFilter.select(
                filterContext,
                selectClauses,
                eventNode
            );
        } else {
            return new Variant[0];
        }
    }

    @Override
    protected synchronized void enqueue(Variant[] value) {
        if (queue.size() < queue.maxSize()) {
            queue.add(value);
        } else {
            if (getQueueSize() > 1) {
                eventOverflow.set(true);

                Subscription subscription = session.getSubscriptionManager().getSubscription(subscriptionId);

                if (subscription != null) {
                    subscription.getSubscriptionDiagnostics().getEventQueueOverflowCount().increment();
                }
            }

            if (discardOldest) {
                queue.add(value);
            } else {
                queue.set(queue.maxSize() - 1, value);
            }
        }
    }

    @Override
    public synchronized boolean getNotifications(List<UaStructure> notifications, int max) {
        if (eventOverflow.compareAndSet(true, false)) {
            Variant[] eventFields = generateOverflowEventFields();

            if (discardOldest) {
                // insert overflow event at beginning
                notifications.add(wrapQueueValue(eventFields));

                return super.getNotifications(notifications, max);
            } else {
                // insert overflow event at end
                boolean more = super.getNotifications(notifications, max);
                notifications.add(wrapQueueValue(eventFields));

                return more;
            }
        } else {
            return super.getNotifications(notifications, max);
        }
    }

    @NotNull
    private Variant[] generateOverflowEventFields() {
        BaseEventTypeNode overflowEvent = null;

        try {
            UUID eventId = UUID.randomUUID();

            overflowEvent = server.getEventFactory().createEvent(
                new NodeId(1, eventId),
                Identifiers.EventQueueOverflowEventType
            );

            overflowEvent.setBrowseName(new QualifiedName(1, "EventQueueOverflow"));
            overflowEvent.setDisplayName(LocalizedText.english("EventQueueOverflow"));

            ByteBuffer buffer = ByteBuffer.allocate(64);
            buffer.putLong(eventId.getMostSignificantBits());
            buffer.putLong(eventId.getLeastSignificantBits());

            overflowEvent.setEventId(ByteString.of(buffer.array()));
            overflowEvent.setEventType(Identifiers.EventQueueOverflowEventType);
            overflowEvent.setSourceNode(Identifiers.Server);
            overflowEvent.setSourceName("Server");
            overflowEvent.setTime(DateTime.now());
            overflowEvent.setReceiveTime(DateTime.NULL_VALUE);
            overflowEvent.setMessage(LocalizedText.english("Event Queue Overflow"));
            overflowEvent.setSeverity(ushort(0));

            return selectEventFields(overflowEvent);
        } catch (UaException e) {
            logger.error("Error creating overflow event: {}", e.getMessage(), e);

            return new Variant[0];
        } finally {
            if (overflowEvent != null) {
                overflowEvent.delete();
            }
        }
    }

    @Override
    public ExtensionObject getFilterResult() {
        return ExtensionObject.encode(server.getSerializationContext(), filterResult);
    }

    @Override
    public void installFilter(MonitoringFilter filter) throws UaException {
        if (filter instanceof EventFilter) {
            this.filter = (EventFilter) filter;

            filterResult = EventContentFilter.validate(filterContext, this.filter);

            boolean selectClauseGood = l(filterResult.getSelectClauseResults())
                .stream()
                .allMatch(StatusCode::isGood);

            boolean whereClauseGood = l(filterResult.getWhereClauseResult().getElementResults())
                .stream()
                .map(ContentFilterElementResult::getStatusCode)
                .allMatch(StatusCode::isGood);

            filterResultGood = selectClauseGood && whereClauseGood;
        } else {
            filterResultGood = false;

            throw new UaException(StatusCodes.Bad_MonitoredItemFilterUnsupported);
        }
    }
    
    @Override
    protected EventFieldList wrapQueueValue(Variant[] value) {
        return new EventFieldList(uint(getClientHandle()), value);
    }

    @Override
    public boolean isSamplingEnabled() {
        return getMonitoringMode() != MonitoringMode.Disabled;
    }

}
