/*
 * 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.cassandra.service.accord.txn;

import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

import javax.annotation.Nullable;

import com.google.common.annotations.VisibleForTesting;

import org.apache.cassandra.cql3.FieldIdentifier;
import org.apache.cassandra.cql3.Operation;
import org.apache.cassandra.cql3.RowUpdateBuilder;
import org.apache.cassandra.cql3.terms.Constants;
import org.apache.cassandra.cql3.terms.Lists;
import org.apache.cassandra.cql3.terms.Maps;
import org.apache.cassandra.cql3.terms.MultiElements;
import org.apache.cassandra.cql3.terms.Sets;
import org.apache.cassandra.cql3.terms.Term;
import org.apache.cassandra.cql3.terms.UserTypes;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.CollectionType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.ListType;
import org.apache.cassandra.db.marshal.MapType;
import org.apache.cassandra.db.marshal.SetType;
import org.apache.cassandra.db.marshal.TupleType;
import org.apache.cassandra.db.marshal.UserType;
import org.apache.cassandra.db.rows.CellPath;
import org.apache.cassandra.io.ParameterisedUnversionedSerializer;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.schema.ColumnMetadata;
import org.apache.cassandra.schema.TableMetadata;
import org.apache.cassandra.service.accord.serializers.AccordSerializers;
import org.apache.cassandra.service.accord.serializers.TableMetadatas;
import org.apache.cassandra.utils.ByteBufferUtil;

import static org.apache.cassandra.db.marshal.CollectionType.Kind.MAP;
import static org.apache.cassandra.service.accord.serializers.AccordSerializers.columnMetadataSerializer;

public class TxnReferenceOperation
{
    private static final Map<Class<? extends Operation>, Kind> operationKindMap = initOperationKindMap();

    @VisibleForTesting
    static Map<Class<? extends Operation>, Kind> initOperationKindMap()
    {
        Map<Class<? extends Operation>, Kind> temp = new HashMap<>();
        temp.put(Constants.Adder.class, Kind.ConstantAdder);
        temp.put(Constants.Setter.class, Kind.ConstantSetter);
        temp.put(Constants.Substracter.class, Kind.ConstantSubtracter);
        temp.put(Lists.Appender.class, Kind.ListAppender);
        temp.put(Lists.Discarder.class, Kind.ListDiscarder);
        temp.put(Lists.DiscarderByIndex.class, Kind.ListDiscarderByIndex);
        temp.put(Lists.Prepender.class, Kind.ListPrepender);
        temp.put(Lists.Setter.class, Kind.ListSetter);
        temp.put(Lists.SetterByIndex.class, Kind.ListSetterByIndex);
        temp.put(Maps.DiscarderByKey.class, Kind.MapDiscarderByKey);
        temp.put(Maps.Putter.class, Kind.MapPutter);
        temp.put(Maps.Setter.class, Kind.MapSetter);
        temp.put(Maps.SetterByKey.class, Kind.MapSetterByKey);
        temp.put(Sets.Adder.class, Kind.SetAdder);
        temp.put(Sets.Discarder.class, Kind.SetDiscarder);
        temp.put(Sets.ElementDiscarder.class, Kind.SetElementDiscarder);
        temp.put(Sets.Setter.class, Kind.SetSetter);
        temp.put(UserTypes.Setter.class, Kind.UserTypeSetter);
        temp.put(UserTypes.SetterByField.class, Kind.UserTypeSetterByField);
        return temp;
    }

    private interface ToOperation
    {
        Operation apply(ColumnMetadata column, Term keyOrIndex, FieldIdentifier field, Term value);
    }

    public enum Kind
    {
        SetAdder((byte) 1, (column, keyOrIndex, field, value) -> new Sets.Adder(column, value)),
        ConstantAdder((byte) 2, (column, keyOrIndex, field, value) -> new Constants.Adder(column, value)),
        ListAppender((byte) 3, (column, keyOrIndex, field, value) -> new Lists.Appender(column, value)),
        SetDiscarder((byte) 4, (column, keyOrIndex, field, value) -> new Sets.Discarder(column, value)),
        ListDiscarder((byte) 5, (column, keyOrIndex, field, value) -> new Lists.Discarder(column, value)),
        ListPrepender((byte) 6, (column, keyOrIndex, field, value) -> new Lists.Prepender(column, value)),
        MapPutter((byte) 7, (column, keyOrIndex, field, value) -> new Maps.Putter(column, value)),
        ListSetter((byte) 8, (column, keyOrIndex, field, value) -> new Lists.Setter(column, value)),
        SetSetter((byte) 9, (column, keyOrIndex, field, value) -> new Sets.Setter(column, value)),
        MapSetter((byte) 10, (column, keyOrIndex, field, value) -> new Maps.Setter(column, value)),
        UserTypeSetter((byte) 11, (column, keyOrIndex, field, value) -> new UserTypes.Setter(column, value)),
        ConstantSetter((byte) 12, (column, keyOrIndex, field, value) -> new Constants.Setter(column, value)),
        ConstantSubtracter((byte) 13, (column, keyOrIndex, field, value) -> new Constants.Substracter(column, value)),
        MapSetterByKey((byte) 14, (column, keyOrIndex, field, value) -> new Maps.SetterByKey(column, keyOrIndex, value)),
        ListSetterByIndex((byte) 15, (column, keyOrIndex, field, value) -> new Lists.SetterByIndex(column, keyOrIndex, value)),
        UserTypeSetterByField((byte) 16, (column, keyOrIndex, field, value) -> new UserTypes.SetterByField(column, field, value)),
        ListDiscarderByIndex((byte) 17, (column, keyOrIndex, field, value) -> new Lists.DiscarderByIndex(column, value)),
        MapDiscarderByKey((byte) 18, (column, keyOrIndex, field, value) -> new Maps.DiscarderByKey(column, value)),
        SetElementDiscarder((byte) 19, (column, keyOrIndex, field, value) -> new Sets.ElementDiscarder(column, value));

        private final byte id;
        private final ToOperation toOperation;

        Kind(byte id, ToOperation toOperation)
        {
            this.id = id;
            this.toOperation = toOperation;
        }

        public static Kind from(byte b)
        {
            for (Kind k : values())
                if (k.id == b)
                    return k;

            throw new IllegalArgumentException("There is no kind with id: " + b);
        }

        public static Kind from(Operation operation)
        {
            Class<? extends Operation> clazz = operation.getClass();
            Kind kind = operationKindMap.get(clazz);
            if (kind == null)
                throw new IllegalArgumentException("There is no Kind associated with operation: " + clazz);
            return kind;
        }
        
        public static Kind setterFor(ColumnMetadata column)
        {
            if (column.type instanceof ListType)
                return ListSetter;
            else if (column.type instanceof SetType)
                return SetSetter;
            else if (column.type instanceof MapType)
                return MapSetter;
            else if (column.type instanceof UserType)
                return UserTypeSetter;

            return ConstantSetter;
        }
        
        public Operation toOperation(ColumnMetadata column, Term keyOrIndex, FieldIdentifier field, Term value)
        {
            return toOperation.apply(column, keyOrIndex, field, value);
        }
    }

    private final Kind kind;
    private final ColumnMetadata receiver;
    public final TableMetadata table;
    private final @Nullable ByteBuffer keyOrIndex;
    private final @Nullable ByteBuffer field;
    private final TxnReferenceValue value;
    private final @Nullable AbstractType<?> keyOrIndexType;
    private final AbstractType<?> valueType;

    public TxnReferenceOperation(Kind kind, ColumnMetadata receiver, TableMetadata table,
                                 @Nullable ByteBuffer keyOrIndex, @Nullable ByteBuffer field, TxnReferenceValue value)
    {
        this.kind = kind;
        this.receiver = receiver;
        this.table = table;
        this.keyOrIndex = keyOrIndex;
        this.field = field;

        // We don't expect operators on clustering keys, but unwrap just in case.
        AbstractType<?> receiverType = receiver.type.unwrap();

        if (kind == TxnReferenceOperation.Kind.SetDiscarder && receiverType.isCollection() && (((CollectionType<?>) receiverType).kind == MAP))
        {
            // The value for a map subtraction is actually a set (see Operation.Substraction)
            this.valueType = SetType.getInstance(((MapType<?, ?>) receiverType).getKeysType(), true);
            this.keyOrIndexType = null;
        }
        else if (kind == Kind.MapDiscarderByKey || kind == Kind.SetElementDiscarder)
        {
            CollectionType<?> ct = (CollectionType<?>) receiverType;
            this.keyOrIndexType = null;
            this.valueType = ct.nameComparator();
        }
        else if (kind == Kind.MapSetterByKey || kind == Kind.ListSetterByIndex)
        {
            CollectionType<?> ct = (CollectionType<?>) receiverType;
            this.keyOrIndexType = ct.nameComparator();
            this.valueType = ct.valueComparator();
        }
        else if (kind == Kind.ListDiscarderByIndex)
        {
            this.valueType = Int32Type.instance;
            this.keyOrIndexType = null;
        }
        else if (kind == Kind.UserTypeSetterByField)
        {
            UserType userType = (UserType) receiverType;
            CellPath fieldPath = userType.cellPathForField(new FieldIdentifier(field));
            this.valueType = userType.fieldType(fieldPath);
            this.keyOrIndexType = null;
        }
        else
        {
            this.valueType = receiverType;
            this.keyOrIndexType = null;
        }

        this.value = value;
    }

    @Override
    public boolean equals(Object o)
    {
        if (this == o) return true;
        if (o == null || getClass() != o.getClass()) return false;
        TxnReferenceOperation that = (TxnReferenceOperation) o;
        return Objects.equals(receiver, that.receiver) 
               && kind == that.kind
               && Objects.equals(keyOrIndex, that.keyOrIndex)
               && Objects.equals(field, that.field)
               && Objects.equals(value, that.value);
    }

    @Override
    public int hashCode()
    {
        return Objects.hash(receiver, kind, keyOrIndex, field, value);
    }


    @Override
    public String toString()
    {
        return "TxnReferenceOperation{" +
               "kind=" + kind +
               ", receiver=" + receiver +
               ", table=" + table +
               ", key=" + keyOrIndex +
               ", field=" + field +
               ", value=" + value +
               '}';
    }

    public void collect(TableMetadatas.Collector collector)
    {
        collector.add(table);
        value.collect(collector);
    }

    public ColumnMetadata receiver()
    {
        return receiver;
    }

    public void apply(TxnData data, DecoratedKey key, RowUpdateBuilder up)
    {
        Operation operation = toOperation(data);
        operation.execute(key, up);
    }

    @VisibleForTesting
    Operation toOperation(TxnData data)
    {
        FieldIdentifier fieldIdentifier = field == null ? null : new FieldIdentifier(field);
        Term valueTerm = toTerm(data, valueType);
        Term keyorIndexTerm = keyOrIndex == null ? null : toTerm(keyOrIndex, keyOrIndexType);
        return kind.toOperation(receiver, keyorIndexTerm, fieldIdentifier, valueTerm);
    }

    private Term toTerm(TxnData data, AbstractType<?> receivingType)
    {
        ByteBuffer bytes = value.compute(data, receivingType);
        if (bytes == null)
            return Constants.NULL_VALUE;
        return toTerm(bytes, receivingType);
    }

    private Term toTerm(ByteBuffer bytes, AbstractType<?> receivingType)
    {
        if (receivingType.isCollection())
            return AccordSerializers.deserializeCqlCollectionAsTerm(bytes, receivingType);
        else if (receivingType.isUDT())
            return MultiElements.Value.fromSerialized(bytes, (UserType) receivingType);
        else if (receivingType.isTuple())
            return MultiElements.Value.fromSerialized(bytes, (TupleType) receivingType);

        return new Constants.Value(bytes);
    }

    static final ParameterisedUnversionedSerializer<TxnReferenceOperation, TableMetadatas> serializer = new ParameterisedUnversionedSerializer<>()
    {
        @Override
        public void serialize(TxnReferenceOperation operation, TableMetadatas tables, DataOutputPlus out) throws IOException
        {
            out.writeByte(operation.kind.id);
            tables.serialize(operation.table, out);
            columnMetadataSerializer.serialize(operation.receiver, operation.table, out);
            TxnReferenceValue.serializer.serialize(operation.value, tables, out);

            out.writeBoolean(operation.keyOrIndex != null);
            if (operation.keyOrIndex != null)
                ByteBufferUtil.writeWithVIntLength(operation.keyOrIndex, out);

            out.writeBoolean(operation.field != null);
            if (operation.field != null)
                ByteBufferUtil.writeWithVIntLength(operation.field, out);
        }

        @Override
        public TxnReferenceOperation deserialize(TableMetadatas tables, DataInputPlus in) throws IOException
        {
            Kind kind = Kind.from(in.readByte());
            TableMetadata table = tables.deserialize(in);
            ColumnMetadata receiver = columnMetadataSerializer.deserialize(table, in);
            TxnReferenceValue value = TxnReferenceValue.serializer.deserialize(tables, in);
            ByteBuffer key = in.readBoolean() ? ByteBufferUtil.readWithVIntLength(in) : null;
            ByteBuffer field = in.readBoolean() ? ByteBufferUtil.readWithVIntLength(in) : null;
            return new TxnReferenceOperation(kind, receiver, table, key, field, value);
        }

        @Override
        public long serializedSize(TxnReferenceOperation operation, TableMetadatas tables)
        {
            long size = Byte.BYTES;
            size +=  tables.serializedSize(operation.table);
            size += columnMetadataSerializer.serializedSize(operation.receiver, operation.table);
            size += TxnReferenceValue.serializer.serializedSize(operation.value, tables);

            size += TypeSizes.sizeof(operation.keyOrIndex != null);
            if (operation.keyOrIndex != null)
                size += ByteBufferUtil.serializedSizeWithVIntLength(operation.keyOrIndex);

            size += TypeSizes.sizeof(operation.field != null);
            if (operation.field != null)
                size += ByteBufferUtil.serializedSizeWithVIntLength(operation.field);

            return size;
        }
    };
}
