/*
 * 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.ignite.internal.processors.platform.client.cache;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;
import org.apache.ignite.binary.BinaryRawReader;
import org.apache.ignite.cache.CacheAtomicityMode;
import org.apache.ignite.cache.CacheKeyConfiguration;
import org.apache.ignite.cache.CacheMode;
import org.apache.ignite.cache.CacheRebalanceMode;
import org.apache.ignite.cache.CacheWriteSynchronizationMode;
import org.apache.ignite.cache.PartitionLossPolicy;
import org.apache.ignite.cache.QueryEntity;
import org.apache.ignite.cache.QueryIndex;
import org.apache.ignite.client.ClientException;
import org.apache.ignite.configuration.CacheConfiguration;
import org.apache.ignite.internal.binary.BinaryRawWriterEx;
import org.apache.ignite.internal.processors.platform.client.ClientProtocolContext;
import org.apache.ignite.internal.processors.platform.client.ClientProtocolVersionFeature;
import org.apache.ignite.internal.processors.platform.utils.PlatformConfigurationUtils;
import static java.util.Optional.ofNullable;
import static org.apache.ignite.internal.processors.platform.client.ClientProtocolVersionFeature.QUERY_ENTITY_PRECISION_AND_SCALE;

/**
 * Cache configuration serializer.
 */
public class ClientCacheConfigurationSerializer {
    /** Name. */
    private static final short NAME = 0;

    /** Common properties. */
    private static final short CACHE_MODE = 1;

    /** */
    private static final short ATOMICITY_MODE = 2;

    /** */
    private static final short BACKUPS = 3;

    /** */
    private static final short WRITE_SYNCHRONIZATION_MODE = 4;

    /** */
    private static final short COPY_ON_READ = 5;

    /** */
    private static final short READ_FROM_BACKUP = 6;

    /** Memory settings. */
    private static final short DATA_REGION_NAME = 100;

    /** */
    private static final short ONHEAP_CACHE_ENABLED = 101;

    /** SQL. */
    private static final short QUERY_ENTITIES = 200;

    /** */
    private static final short QUERY_PARALLELISM = 201;

    /** */
    private static final short QUERY_DETAIL_METRICS_SIZE = 202;

    /** */
    private static final short SQL_SCHEMA = 203;

    /** */
    private static final short SQL_INDEX_MAX_INLINE_SIZE = 204;

    /** */
    private static final short SQL_ESCAPE_ALL = 205;

    /** */
    private static final short MAX_QUERY_ITERATORS_COUNT = 206;

    /** Rebalance. */
    private static final short REBALANCE_MODE = 300;

    /** */
    private static final short REBALANCE_DELAY = 301;

    /** */
    private static final short REBALANCE_TIMEOUT = 302;

    /** */
    private static final short REBALANCE_BATCH_SIZE = 303;

    /** */
    private static final short REBALANCE_BATCHES_PREFETCH_COUNT = 304;

    /** */
    private static final short REBALANCE_ORDER = 305;

    /** */
    private static final short REBALANCE_THROTTLE = 306;

    /** Advanced. */
    private static final short GROUP_NAME = 400;

    /** */
    private static final short KEY_CONFIGURATION = 401;

    /** */
    private static final short DEFAULT_LOCK_TIMEOUT = 402;

    /** */
    private static final short MAX_CONCURRENT_ASYNC_OPERATIONS = 403;

    /** */
    private static final short PARTITION_LOSS_POLICY = 404;

    /** */
    private static final short EAGER_TTL = 405;

    /** */
    private static final short STATISTICS_ENABLED = 406;

    /** */
    private static final short EXPIRY_POLICY = 407;

    /**
     * Writes the cache configuration.
     * @param writer Writer.
     * @param cfg Configuration.
     * @param protocolCtx Client protocol context.
     */
    static void write(BinaryRawWriterEx writer, CacheConfiguration cfg, ClientProtocolContext protocolCtx) {
        assert writer != null;
        assert cfg != null;

        // Reserve for length.
        int pos = writer.reserveInt();

        PlatformConfigurationUtils.writeEnumInt(writer, cfg.getAtomicityMode(),
                CacheConfiguration.DFLT_CACHE_ATOMICITY_MODE);

        writer.writeInt(cfg.getBackups());
        writer.writeInt(ofNullable(cfg.getCacheMode()).orElse(CacheConfiguration.DFLT_CACHE_MODE).code());
        writer.writeBoolean(cfg.isCopyOnRead());
        writer.writeString(cfg.getDataRegionName());
        writer.writeBoolean(cfg.isEagerTtl());
        writer.writeBoolean(cfg.isStatisticsEnabled());
        writer.writeString(cfg.getGroupName());
        writer.writeLong(cfg.getDefaultLockTimeout());
        writer.writeInt(cfg.getMaxConcurrentAsyncOperations());
        writer.writeInt(cfg.getMaxQueryIteratorsCount());
        writer.writeString(cfg.getName());
        writer.writeBoolean(cfg.isOnheapCacheEnabled());
        writer.writeInt(cfg.getPartitionLossPolicy().ordinal());
        writer.writeInt(cfg.getQueryDetailMetricsSize());
        writer.writeInt(cfg.getQueryParallelism());
        writer.writeBoolean(cfg.isReadFromBackup());
        writer.writeInt(cfg.getRebalanceBatchSize());
        writer.writeLong(cfg.getRebalanceBatchesPrefetchCount());
        writer.writeLong(cfg.getRebalanceDelay());
        PlatformConfigurationUtils.writeEnumInt(writer, cfg.getRebalanceMode(), CacheConfiguration.DFLT_REBALANCE_MODE);
        writer.writeInt(cfg.getRebalanceOrder());
        writer.writeLong(cfg.getRebalanceThrottle());
        writer.writeLong(cfg.getRebalanceTimeout());
        writer.writeBoolean(cfg.isSqlEscapeAll());
        writer.writeInt(cfg.getSqlIndexMaxInlineSize());
        writer.writeString(cfg.getSqlSchema());
        PlatformConfigurationUtils.writeEnumInt(writer, cfg.getWriteSynchronizationMode());

        CacheKeyConfiguration[] keys = cfg.getKeyConfiguration();

        if (keys != null) {
            writer.writeInt(keys.length);

            for (CacheKeyConfiguration key : keys) {
                writer.writeString(key.getTypeName());
                writer.writeString(key.getAffinityKeyFieldName());
            }
        }
        else {
            writer.writeInt(0);
        }

        //noinspection unchecked
        Collection<QueryEntity> qryEntities = cfg.getQueryEntities();

        if (qryEntities != null) {
            writer.writeInt(qryEntities.size());

            for (QueryEntity e : qryEntities)
                writeQueryEntity(writer, e, protocolCtx);
        }
        else
            writer.writeInt(0);

        if (protocolCtx.isFeatureSupported(ClientProtocolVersionFeature.EXPIRY_POLICY))
            PlatformConfigurationUtils.writeExpiryPolicyFactory(writer, cfg.getExpiryPolicyFactory());

        // Write length (so that part of the config can be skipped).
        writer.writeInt(pos, writer.out().position() - pos - 4);
    }

    /**
     * Write query entity. Version for thin client.
     *
     * @param writer Writer.
     * @param qryEntity Query entity.
     * @param protocolCtx Protocol context.
     */
    private static void writeQueryEntity(BinaryRawWriterEx writer, QueryEntity qryEntity, ClientProtocolContext protocolCtx) {
        assert qryEntity != null;

        writer.writeString(qryEntity.getKeyType());
        writer.writeString(qryEntity.getValueType());
        writer.writeString(qryEntity.getTableName());
        writer.writeString(qryEntity.getKeyFieldName());
        writer.writeString(qryEntity.getValueFieldName());

        // Fields
        LinkedHashMap<String, String> fields = qryEntity.getFields();

        if (fields != null) {
            Set<String> keyFields = qryEntity.getKeyFields();
            Set<String> notNullFields = qryEntity.getNotNullFields();
            Map<String, Object> defVals = qryEntity.getDefaultFieldValues();
            Map<String, Integer> fieldsPrecision = qryEntity.getFieldsPrecision();
            Map<String, Integer> fieldsScale = qryEntity.getFieldsScale();

            writer.writeInt(fields.size());

            for (Map.Entry<String, String> field : fields.entrySet()) {
                writer.writeString(field.getKey());
                writer.writeString(field.getValue());
                writer.writeBoolean(keyFields != null && keyFields.contains(field.getKey()));
                writer.writeBoolean(notNullFields != null && notNullFields.contains(field.getKey()));
                writer.writeObject(defVals != null ? defVals.get(field.getKey()) : null);

                if (protocolCtx.isFeatureSupported(QUERY_ENTITY_PRECISION_AND_SCALE)) {
                    writer.writeInt(fieldsPrecision == null ? -1 : fieldsPrecision.getOrDefault(field.getKey(), -1));
                    writer.writeInt(fieldsScale == null ? -1 : fieldsScale.getOrDefault(field.getKey(), -1));
                }
            }
        }
        else
            writer.writeInt(0);

        // Aliases
        Map<String, String> aliases = qryEntity.getAliases();

        if (aliases != null) {
            writer.writeInt(aliases.size());

            for (Map.Entry<String, String> alias : aliases.entrySet()) {
                writer.writeString(alias.getKey());
                writer.writeString(alias.getValue());
            }
        }
        else
            writer.writeInt(0);

        // Indexes
        Collection<QueryIndex> indexes = qryEntity.getIndexes();

        if (indexes != null) {
            writer.writeInt(indexes.size());

            for (QueryIndex index : indexes)
                PlatformConfigurationUtils.writeQueryIndex(writer, index);
        }
        else
            writer.writeInt(0);
    }

    /**
     * Reads the cache configuration.
     *
     * @param reader Reader.
     * @param protocolCtx Client protocol context.
     * @return Configuration.
     */
    static CacheConfiguration read(BinaryRawReader reader, ClientProtocolContext protocolCtx) {
        reader.readInt();  // Skip length.

        short propCnt = reader.readShort();

        CacheConfiguration<?, ?> cfg = new CacheConfiguration<>();

        for (int i = 0; i < propCnt; i++) {
            short code = reader.readShort();

            switch (code) {
                case ATOMICITY_MODE:
                    cfg.setAtomicityMode(CacheAtomicityMode.fromOrdinal(reader.readInt()));
                    break;

                case BACKUPS:
                    cfg.setBackups(reader.readInt());
                    break;

                case CACHE_MODE:
                    cfg.setCacheMode(CacheMode.fromCode(reader.readInt()));
                    break;

                case COPY_ON_READ:
                    cfg.setCopyOnRead(reader.readBoolean());
                    break;

                case DATA_REGION_NAME:
                    cfg.setDataRegionName(reader.readString());
                    break;

                case EAGER_TTL:
                    cfg.setEagerTtl(reader.readBoolean());
                    break;

                case EXPIRY_POLICY:
                    cfg.setExpiryPolicyFactory(PlatformConfigurationUtils.readExpiryPolicyFactory(reader));
                    break;

                case STATISTICS_ENABLED:
                    cfg.setStatisticsEnabled(reader.readBoolean());
                    break;

                case GROUP_NAME:
                    cfg.setGroupName(reader.readString());
                    break;

                case DEFAULT_LOCK_TIMEOUT:
                    cfg.setDefaultLockTimeout(reader.readLong());
                    break;

                case MAX_CONCURRENT_ASYNC_OPERATIONS:
                    cfg.setMaxConcurrentAsyncOperations(reader.readInt());
                    break;

                case MAX_QUERY_ITERATORS_COUNT:
                    cfg.setMaxQueryIteratorsCount(reader.readInt());
                    break;

                case NAME:
                    cfg.setName(reader.readString());
                    break;

                case ONHEAP_CACHE_ENABLED:
                    cfg.setOnheapCacheEnabled(reader.readBoolean());
                    break;

                case PARTITION_LOSS_POLICY:
                    cfg.setPartitionLossPolicy(PartitionLossPolicy.fromOrdinal((byte)reader.readInt()));
                    break;

                case QUERY_DETAIL_METRICS_SIZE:
                    cfg.setQueryDetailMetricsSize(reader.readInt());
                    break;

                case QUERY_PARALLELISM:
                    cfg.setQueryParallelism(reader.readInt());
                    break;

                case READ_FROM_BACKUP:
                    cfg.setReadFromBackup(reader.readBoolean());
                    break;

                case REBALANCE_BATCH_SIZE:
                    cfg.setRebalanceBatchSize(reader.readInt());
                    break;

                case REBALANCE_BATCHES_PREFETCH_COUNT:
                    cfg.setRebalanceBatchesPrefetchCount(reader.readLong());
                    break;

                case REBALANCE_DELAY:
                    cfg.setRebalanceDelay(reader.readLong());
                    break;

                case REBALANCE_MODE:
                    cfg.setRebalanceMode(CacheRebalanceMode.fromOrdinal(reader.readInt()));
                    break;

                case REBALANCE_ORDER:
                    cfg.setRebalanceOrder(reader.readInt());
                    break;

                case REBALANCE_THROTTLE:
                    cfg.setRebalanceThrottle(reader.readLong());
                    break;

                case REBALANCE_TIMEOUT:
                    cfg.setRebalanceTimeout(reader.readLong());
                    break;

                case SQL_ESCAPE_ALL:
                    cfg.setSqlEscapeAll(reader.readBoolean());
                    break;

                case SQL_INDEX_MAX_INLINE_SIZE:
                    cfg.setSqlIndexMaxInlineSize(reader.readInt());
                    break;

                case SQL_SCHEMA:
                    cfg.setSqlSchema(reader.readString());
                    break;

                case WRITE_SYNCHRONIZATION_MODE:
                    cfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.fromOrdinal(reader.readInt()));
                    break;

                case KEY_CONFIGURATION:
                    int keyCnt = reader.readInt();

                    if (keyCnt > 0) {
                        CacheKeyConfiguration[] keys = new CacheKeyConfiguration[keyCnt];

                        for (int j = 0; j < keyCnt; j++) {
                            keys[j] = new CacheKeyConfiguration(reader.readString(), reader.readString());
                        }

                        cfg.setKeyConfiguration(keys);
                    }
                    break;

                case QUERY_ENTITIES:
                    int qryEntCnt = reader.readInt();

                    if (qryEntCnt > 0) {
                        Collection<QueryEntity> entities = new ArrayList<>(qryEntCnt);

                        for (int j = 0; j < qryEntCnt; j++)
                            entities.add(readQueryEntity(reader, protocolCtx));

                        cfg.setQueryEntities(entities);
                    }
                    break;
            }
        }

        if (cfg.getCacheMode() == null)
            throw new ClientException("Unsupported cache mode");

        return cfg;
    }

    /**
     * Reads the query entity. Version of function to be used from thin client.
     *
     * @param in Stream.
     * @param protocolCtx Client protocol version.
     * @return QueryEntity.
     */
    public static QueryEntity readQueryEntity(BinaryRawReader in, ClientProtocolContext protocolCtx) {
        QueryEntity res = new QueryEntity();

        res.setKeyType(in.readString());
        res.setValueType(in.readString());
        res.setTableName(in.readString());
        res.setKeyFieldName(in.readString());
        res.setValueFieldName(in.readString());

        // Fields
        int cnt = in.readInt();
        Set<String> keyFields = new HashSet<>(cnt);
        Set<String> notNullFields = new HashSet<>(cnt);
        Map<String, Object> defVals = new HashMap<>(cnt);
        Map<String, Integer> fieldsPrecision = new HashMap<>(cnt);
        Map<String, Integer> fieldsScale = new HashMap<>(cnt);

        if (cnt > 0) {
            LinkedHashMap<String, String> fields = new LinkedHashMap<>(cnt);

            for (int i = 0; i < cnt; i++) {
                String fieldName = in.readString();
                String fieldType = in.readString();

                fields.put(fieldName, fieldType);

                if (in.readBoolean())
                    keyFields.add(fieldName);

                if (in.readBoolean())
                    notNullFields.add(fieldName);

                Object defVal = in.readObject();
                if (defVal != null)
                    defVals.put(fieldName, defVal);

                if (protocolCtx.isFeatureSupported(QUERY_ENTITY_PRECISION_AND_SCALE)) {
                    int precision = in.readInt();

                    if (precision != -1)
                        fieldsPrecision.put(fieldName, precision);

                    int scale = in.readInt();

                    if (scale != -1)
                        fieldsScale.put(fieldName, scale);
                }
            }

            res.setFields(fields);

            if (!keyFields.isEmpty())
                res.setKeyFields(keyFields);

            if (!notNullFields.isEmpty())
                res.setNotNullFields(notNullFields);

            if (!defVals.isEmpty())
                res.setDefaultFieldValues(defVals);

            if (!fieldsPrecision.isEmpty())
                res.setFieldsPrecision(fieldsPrecision);

            if (!fieldsScale.isEmpty())
                res.setFieldsScale(fieldsScale);
        }

        // Aliases
        cnt = in.readInt();

        if (cnt > 0) {
            Map<String, String> aliases = new HashMap<>(cnt);

            for (int i = 0; i < cnt; i++)
                aliases.put(in.readString(), in.readString());

            res.setAliases(aliases);
        }

        // Indexes
        cnt = in.readInt();

        if (cnt > 0) {
            Collection<QueryIndex> indexes = new ArrayList<>(cnt);

            for (int i = 0; i < cnt; i++)
                indexes.add(PlatformConfigurationUtils.readQueryIndex(in));

            res.setIndexes(indexes);
        }

        return res;
    }
}
