/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you 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 org.apache.plc4x.java.ads.protocol;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToMessageCodec;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.configuration2.SystemConfiguration;
import org.apache.plc4x.java.ads.api.commands.*;
import org.apache.plc4x.java.ads.api.commands.types.*;
import org.apache.plc4x.java.ads.api.generic.AmsHeader;
import org.apache.plc4x.java.ads.api.generic.AmsPacket;
import org.apache.plc4x.java.ads.api.generic.types.*;
import org.apache.plc4x.java.ads.protocol.exception.AdsProtocolOverflowException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

public class Ads2PayloadProtocol extends MessageToMessageCodec<ByteBuf, AmsPacket> {

    private static final Logger LOGGER = LoggerFactory.getLogger(Ads2PayloadProtocol.class);

    private static final Configuration CONF = new SystemConfiguration();
    private static final long MAX_NUM_STAMPS = CONF.getLong("plc4x.ads2payloadprotocol.max_num_stamps", 512L);
    private static final long MAX_NUM_SAMPLES = CONF.getLong("plc4x.ads2payloadprotocol.max_num_samples", 1024L);
    private static final long ADS_READ_COMMAND_MAX_BYTES = CONF.getLong("plc4x.ads2payloadprotocol.ads_read_command_max_bytes", 134217728L);
    private static final long ADS_WRITE_COMMAND_MAX_BYTES = CONF.getLong("plc4x.ads2payloadprotocol.ads_write_command_max_bytes", 134217728L);
    private static final long ADS_WRITE_CONTROL_COMMAND_MAX_BYTES = CONF.getLong("plc4x.ads2payloadprotocol.ads_write_control_command_max_bytes", 134217728L);
    private static final long ADS_NOTIFICATION_SAMPLE_MAX_BYTES = CONF.getLong("plc4x.ads2payloadprotocol.ads_notification_sample_max_bytes", 134217728L);
    private static final long ADS_READ_WRITE_COMMAND_REQUEST_MAX_BYTES = CONF.getLong("plc4x.ads2payloadprotocol.ads_read_write_command_request_max_bytes", 134217728L);
    private static final long ADS_READ_WRITE_COMMAND_RESPONSE_MAX_BYTES = CONF.getLong("plc4x.ads2payloadprotocol.ads_read_write_command_response_max_bytes", 134217728L);

    private final ConcurrentMap<Invoke, AmsPacket> requests;

    public Ads2PayloadProtocol() {
        this.requests = new ConcurrentHashMap<>();
    }

    /**
     * Resets this protocol and discard all send requests.
     */
    public void reset() {
        requests.clear();
    }

    @Override
    protected void encode(ChannelHandlerContext channelHandlerContext, AmsPacket amsPacket, List<Object> out) {
        LOGGER.trace("(<--OUT): {}, {}, {}", channelHandlerContext, amsPacket, out);
        Invoke invokeId = amsPacket.getAmsHeader().getInvokeId();
        if (invokeId != Invoke.NONE) {
            requests.put(invokeId, amsPacket);
        }
        out.add(amsPacket.getByteBuf());
    }

    @Override
    protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> out) {
        if (byteBuf == Unpooled.EMPTY_BUFFER) {
            // Cleanup...
            reset();
            return;
        }
        LOGGER.trace("(-->IN): {}, {}, {}", channelHandlerContext, byteBuf, out);
        AmsNetId targetAmsNetId = AmsNetId.of(byteBuf);
        AmsPort targetAmsPort = AmsPort.of(byteBuf);
        AmsNetId sourceAmsNetId = AmsNetId.of(byteBuf);
        AmsPort sourceAmsPort = AmsPort.of(byteBuf);
        Command commandId = Command.of(byteBuf);
        State stateId = State.of(byteBuf);
        DataLength dataLength = DataLength.of(byteBuf);
        AmsError errorCode = AmsError.of(byteBuf);
        Invoke invoke = Invoke.of(byteBuf);
        AmsPacket correlatedAmsPacket = requests.remove(invoke);
        if (correlatedAmsPacket != null) {
            LOGGER.debug("Correlated packet received {}", correlatedAmsPacket);
        }
        if (dataLength.getAsLong() > Integer.MAX_VALUE) {
            throw new AdsProtocolOverflowException(Integer.class, dataLength.getAsLong());
        }
        ByteBuf commandBuffer = byteBuf.readBytes((int) dataLength.getAsLong());
        try {
            AmsHeader amsHeader = AmsHeader.of(targetAmsNetId, targetAmsPort, sourceAmsNetId, sourceAmsPort, commandId,
                stateId, dataLength, errorCode, invoke);
            final AmsPacket amsPacket;
            switch (commandId) {
                case INVALID:
                    amsPacket = handleInvalidCommand(commandBuffer, amsHeader);
                    break;
                case ADS_READ_DEVICE_INFO:
                    amsPacket = handleADSReadDeviceInfoCommand(stateId, commandBuffer, amsHeader);
                    break;
                case ADS_READ:
                    amsPacket = handleADSReadCommand(stateId, commandBuffer, amsHeader);
                    break;
                case ADS_WRITE:
                    amsPacket = handleADSWriteCommand(stateId, commandBuffer, amsHeader);
                    break;
                case ADS_READ_STATE:
                    amsPacket = handleADSReadStateCommand(stateId, commandBuffer, amsHeader);
                    break;
                case ADS_WRITE_CONTROL:
                    amsPacket = handleADSWriteControlCommand(stateId, commandBuffer, amsHeader);
                    break;
                case ADS_ADD_DEVICE_NOTIFICATION:
                    amsPacket = handleADSAddDeviceNotificationCommand(stateId, commandBuffer, amsHeader);
                    break;
                case ADS_DELETE_DEVICE_NOTIFICATION:
                    amsPacket = handADSDeleteDeviceNotificationCommand(stateId, commandBuffer, amsHeader);
                    break;
                case ADS_DEVICE_NOTIFICATION:
                    amsPacket = handleADSDeviceNotificationCommand(stateId, commandBuffer, amsHeader);
                    break;
                case ADS_READ_WRITE:
                    amsPacket = handleADSReadWriteCommand(stateId, commandBuffer, amsHeader);
                    break;
                case UNKNOWN:
                default:
                    amsPacket = handleUnknownCommand(commandBuffer, amsHeader);
            }
            LOGGER.debug("Received amsPacket {}", amsPacket);
            out.add(amsPacket);
            if (commandBuffer.readableBytes() > 0) {
                throw new IllegalStateException("Unread bytes left: " + commandBuffer.readableBytes());
            }
        } finally {
            commandBuffer.release();
        }
    }


    private AmsPacket handleInvalidCommand(ByteBuf commandBuffer, AmsHeader amsHeader) {
        return UnknownCommand.of(amsHeader, commandBuffer);
    }

    private AmsPacket handleADSReadDeviceInfoCommand(State stateId, ByteBuf commandBuffer, AmsHeader amsHeader) {
        AmsPacket amsPacket;
        if (stateId.isRequest()) {
            amsPacket = AdsReadDeviceInfoRequest.of(amsHeader);
        } else {
            Result result = Result.of(commandBuffer);
            MajorVersion majorVersion = MajorVersion.of(commandBuffer);
            MinorVersion minorVersion = MinorVersion.of(commandBuffer);
            Version version = Version.of(commandBuffer);
            Device device = Device.of(commandBuffer);
            amsPacket = AdsReadDeviceInfoResponse.of(amsHeader, result, majorVersion, minorVersion, version, device);
        }
        return amsPacket;
    }

    private AmsPacket handleADSReadCommand(State stateId, ByteBuf commandBuffer, AmsHeader amsHeader) {
        AmsPacket amsPacket;
        if (stateId.isRequest()) {
            IndexGroup indexGroup = IndexGroup.of(commandBuffer);
            IndexOffset indexOffset = IndexOffset.of(commandBuffer);
            Length length = Length.of(commandBuffer);
            amsPacket = AdsReadRequest.of(amsHeader, indexGroup, indexOffset, length);
        } else {
            Result result = Result.of(commandBuffer);
            Length length = Length.of(commandBuffer);
            if (length.getAsLong() > Integer.MAX_VALUE) {
                throw new AdsProtocolOverflowException(Integer.class, length.getAsLong());
            }
            if (length.getAsLong() > ADS_READ_COMMAND_MAX_BYTES) {
                throw new AdsProtocolOverflowException("ADS_READ_COMMAND_MAX_BYTES", ADS_READ_COMMAND_MAX_BYTES, length.getAsLong());
            }
            byte[] dataToRead = new byte[(int) length.getAsLong()];
            commandBuffer.readBytes(dataToRead);
            Data data = Data.of(dataToRead);
            amsPacket = AdsReadResponse.of(amsHeader, result, length, data);
        }
        return amsPacket;
    }

    private AmsPacket handleADSWriteCommand(State stateId, ByteBuf commandBuffer, AmsHeader amsHeader) {
        AmsPacket amsPacket;
        if (stateId.isRequest()) {
            IndexGroup indexGroup = IndexGroup.of(commandBuffer);
            IndexOffset indexOffset = IndexOffset.of(commandBuffer);
            Length length = Length.of(commandBuffer);
            if (length.getAsLong() > Integer.MAX_VALUE) {
                throw new AdsProtocolOverflowException(Integer.class, length.getAsLong());
            }
            if (length.getAsLong() > ADS_WRITE_COMMAND_MAX_BYTES) {
                throw new AdsProtocolOverflowException("ADS_WRITE_COMMAND_MAX_BYTES", ADS_WRITE_COMMAND_MAX_BYTES, length.getAsLong());
            }
            byte[] dataToRead = new byte[(int) length.getAsLong()];
            commandBuffer.readBytes(dataToRead);
            Data data = Data.of(dataToRead);
            amsPacket = AdsWriteRequest.of(amsHeader, indexGroup, indexOffset, length, data);
        } else {
            Result result = Result.of(commandBuffer);
            amsPacket = AdsWriteResponse.of(amsHeader, result);
        }
        return amsPacket;
    }


    private AmsPacket handleADSReadStateCommand(State stateId, ByteBuf commandBuffer, AmsHeader amsHeader) {
        AmsPacket amsPacket;
        if (stateId.isRequest()) {
            amsPacket = AdsReadStateRequest.of(amsHeader);
        } else {
            Result result = Result.of(commandBuffer);
            AdsState adsState = AdsState.of(commandBuffer);
            DeviceState deviceState = DeviceState.of(commandBuffer);
            amsPacket = AdsReadStateResponse.of(amsHeader, result, adsState, deviceState);
        }
        return amsPacket;
    }

    private AmsPacket handleADSWriteControlCommand(State stateId, ByteBuf commandBuffer, AmsHeader amsHeader) {
        AmsPacket amsPacket;
        if (stateId.isRequest()) {
            AdsState adsState = AdsState.of(commandBuffer);
            DeviceState deviceState = DeviceState.of(commandBuffer);
            Length length = Length.of(commandBuffer);
            if (length.getAsLong() > Integer.MAX_VALUE) {
                throw new AdsProtocolOverflowException(Integer.class, length.getAsLong());
            }
            if (length.getAsLong() > ADS_WRITE_CONTROL_COMMAND_MAX_BYTES) {
                throw new AdsProtocolOverflowException("ADS_WRITE_CONTROL_COMMAND_MAX_BYTES", ADS_WRITE_CONTROL_COMMAND_MAX_BYTES, length.getAsLong());
            }
            byte[] dataToRead = new byte[(int) length.getAsLong()];
            commandBuffer.readBytes(dataToRead);
            Data data = Data.of(dataToRead);
            amsPacket = AdsWriteControlRequest.of(amsHeader, adsState, deviceState, length, data);
        } else {
            Result result = Result.of(commandBuffer);
            amsPacket = AdsWriteControlResponse.of(amsHeader, result);
        }
        return amsPacket;
    }

    private AmsPacket handleADSAddDeviceNotificationCommand(State stateId, ByteBuf commandBuffer, AmsHeader amsHeader) {
        AmsPacket amsPacket;
        if (stateId.isRequest()) {
            IndexGroup indexGroup = IndexGroup.of(commandBuffer);
            IndexOffset indexOffset = IndexOffset.of(commandBuffer);
            Length length = Length.of(commandBuffer);
            TransmissionMode transmissionMode = TransmissionMode.of(commandBuffer);
            MaxDelay maxDelay = MaxDelay.of(commandBuffer);
            CycleTime cycleTime = CycleTime.of(commandBuffer);
            commandBuffer.skipBytes(AdsAddDeviceNotificationRequest.Reserved.NUM_BYTES);
            amsPacket = AdsAddDeviceNotificationRequest.of(amsHeader, indexGroup, indexOffset, length, transmissionMode, maxDelay, cycleTime);
        } else {
            Result result = Result.of(commandBuffer);
            NotificationHandle notificationHandle = NotificationHandle.of(commandBuffer);
            amsPacket = AdsAddDeviceNotificationResponse.of(amsHeader, result, notificationHandle);
        }
        return amsPacket;
    }

    private AmsPacket handADSDeleteDeviceNotificationCommand(State stateId, ByteBuf commandBuffer, AmsHeader amsHeader) {
        AmsPacket amsPacket;
        if (stateId.isRequest()) {
            NotificationHandle notificationHandle = NotificationHandle.of(commandBuffer);
            amsPacket = AdsDeleteDeviceNotificationRequest.of(amsHeader, notificationHandle);
        } else {
            Result result = Result.of(commandBuffer);
            amsPacket = AdsDeleteDeviceNotificationResponse.of(amsHeader, result);
        }
        return amsPacket;
    }

    private AmsPacket handleADSDeviceNotificationCommand(State stateId, ByteBuf commandBuffer, AmsHeader amsHeader) {
        AmsPacket amsPacket;
        if (stateId.isRequest()) {
            Length length = Length.of(commandBuffer);
            if (length.getAsLong() > Integer.MAX_VALUE) {
                throw new AdsProtocolOverflowException(Integer.class, length.getAsLong());
            }
            Stamps stamps = Stamps.of(commandBuffer);
            if (stamps.getAsLong() > Integer.MAX_VALUE) {
                throw new AdsProtocolOverflowException(Integer.class, stamps.getAsLong());
            }
            // Note: the length includes the 4 Bytes of stamps which we read already so we substract.
            ByteBuf adsDeviceNotificationBuffer = commandBuffer.readBytes((int) length.getAsLong() - Stamps.NUM_BYTES);
            try {
                List<AdsStampHeader> adsStampHeaders = new ArrayList<>((int) stamps.getAsLong());
                if (stamps.getAsLong() > MAX_NUM_STAMPS) {
                    throw new AdsProtocolOverflowException("MAX_NUM_STAMPS", MAX_NUM_STAMPS, length.getAsLong());
                }
                for (int i = 1; i <= stamps.getAsLong(); i++) {
                    AdsStampHeader adsStampHeader = handleStampHeader(adsDeviceNotificationBuffer);
                    adsStampHeaders.add(adsStampHeader);
                }
                amsPacket = AdsDeviceNotificationRequest.of(amsHeader, length, stamps, adsStampHeaders);
            } finally {
                adsDeviceNotificationBuffer.release();
            }
        } else {
            amsPacket = UnknownCommand.of(amsHeader, commandBuffer);
        }
        return amsPacket;
    }

    private AdsStampHeader handleStampHeader(ByteBuf adsDeviceNotificationBuffer) {
        TimeStamp timeStamp = TimeStamp.of(adsDeviceNotificationBuffer);
        Samples samples = Samples.of(adsDeviceNotificationBuffer);

        List<AdsNotificationSample> adsNotificationSamples = new LinkedList<>();
        if (samples.getAsLong() > MAX_NUM_SAMPLES) {
            throw new AdsProtocolOverflowException("MAX_NUM_SAMPLES", MAX_NUM_SAMPLES, samples.getAsLong());
        }
        for (int i = 1; i <= samples.getAsLong(); i++) {
            AdsNotificationSample adsNotificationSample = handleAdsNotificartionSample(adsDeviceNotificationBuffer);
            adsNotificationSamples.add(adsNotificationSample);

        }
        return AdsStampHeader.of(timeStamp, samples, adsNotificationSamples);
    }

    private AdsNotificationSample handleAdsNotificartionSample(ByteBuf adsDeviceNotificationBuffer) {
        NotificationHandle notificationHandle = NotificationHandle.of(adsDeviceNotificationBuffer);
        SampleSize sampleSize = SampleSize.of(adsDeviceNotificationBuffer);
        if (sampleSize.getAsLong() > Integer.MAX_VALUE) {
            throw new AdsProtocolOverflowException(Integer.class, sampleSize.getAsLong());
        }
        if (sampleSize.getAsLong() > ADS_NOTIFICATION_SAMPLE_MAX_BYTES) {
            throw new AdsProtocolOverflowException("ADS_NOTIFICATION_SAMPLE_MAX_BYTES", ADS_NOTIFICATION_SAMPLE_MAX_BYTES, sampleSize.getAsLong());
        }
        // TODO: do we need a special marker class for: Notice: If your handle becomes invalid, one notification without data will be send once as advice.
        byte[] dataToRead = new byte[(int) sampleSize.getAsLong()];
        adsDeviceNotificationBuffer.readBytes(dataToRead);
        Data data = Data.of(dataToRead);
        return AdsNotificationSample.of(notificationHandle, sampleSize, data);
    }

    private AmsPacket handleADSReadWriteCommand(State stateId, ByteBuf commandBuffer, AmsHeader amsHeader) {
        AmsPacket amsPacket;
        if (stateId.isRequest()) {
            IndexGroup indexGroup = IndexGroup.of(commandBuffer);
            IndexOffset indexOffset = IndexOffset.of(commandBuffer);
            ReadLength readLength = ReadLength.of(commandBuffer);
            if (readLength.getAsLong() > Integer.MAX_VALUE) {
                throw new AdsProtocolOverflowException(Integer.class, readLength.getAsLong());
            }
            WriteLength writeLength = WriteLength.of(commandBuffer);
            if (writeLength.getAsLong() > Integer.MAX_VALUE) {
                throw new AdsProtocolOverflowException(Integer.class, writeLength.getAsLong());
            }
            if (readLength.getAsLong() + writeLength.getAsLong() > Integer.MAX_VALUE) {
                throw new AdsProtocolOverflowException(Integer.class, readLength.getAsLong() + writeLength.getAsLong());
            }
            if (readLength.getAsLong() > ADS_READ_WRITE_COMMAND_REQUEST_MAX_BYTES) {
                throw new AdsProtocolOverflowException("ADS_READ_WRITE_COMMAND_REQUEST_MAX_BYTES", ADS_READ_WRITE_COMMAND_REQUEST_MAX_BYTES, readLength.getAsLong());
            }
            byte[] dataToRead = new byte[(int) readLength.getAsLong()];
            commandBuffer.readBytes(dataToRead);
            Data data = Data.of(dataToRead);
            amsPacket = AdsReadWriteRequest.of(amsHeader, indexGroup, indexOffset, readLength, writeLength, data);
        } else {
            Result result = Result.of(commandBuffer);
            Length length = Length.of(commandBuffer);
            if (length.getAsLong() > Integer.MAX_VALUE) {
                throw new AdsProtocolOverflowException(Integer.class, length.getAsLong());
            }
            if (length.getAsLong() > ADS_READ_WRITE_COMMAND_RESPONSE_MAX_BYTES) {
                throw new AdsProtocolOverflowException("ADS_READ_WRITE_COMMAND_RESPONSE_MAX_BYTES", ADS_READ_WRITE_COMMAND_RESPONSE_MAX_BYTES, length.getAsLong());
            }
            byte[] dataToRead = new byte[(int) length.getAsLong()];
            commandBuffer.readBytes(dataToRead);
            Data data = Data.of(dataToRead);
            amsPacket = AdsReadWriteResponse.of(amsHeader, result, length, data);
        }
        return amsPacket;
    }

    private AmsPacket handleUnknownCommand(ByteBuf commandBuffer, AmsHeader amsHeader) {
        return UnknownCommand.of(amsHeader, commandBuffer);
    }
}
