/*
 * 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 com.datastax.driver.core;

import com.datastax.driver.core.exceptions.DriverInternalError;
import com.datastax.driver.core.utils.MoreObjects;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import io.netty.buffer.ByteBuf;
import java.util.ArrayList;
import java.util.Collections;
import java.util.EnumMap;
import java.util.List;
import java.util.Map;
import java.util.Set;

/** Data types supported by cassandra. */
public abstract class DataType {

  /** The CQL type name. */
  public enum Name {
    CUSTOM(0),
    ASCII(1),
    BIGINT(2),
    BLOB(3),
    BOOLEAN(4),
    COUNTER(5),
    DECIMAL(6),
    DOUBLE(7),
    FLOAT(8),
    INT(9),
    TEXT(10) {
      @Override
      public boolean isCompatibleWith(Name that) {
        return this == that || that == VARCHAR;
      }
    },
    TIMESTAMP(11),
    UUID(12),
    VARCHAR(13) {
      @Override
      public boolean isCompatibleWith(Name that) {
        return this == that || that == TEXT;
      }
    },
    VARINT(14),
    TIMEUUID(15),
    INET(16),
    DATE(17, ProtocolVersion.V4),
    TIME(18, ProtocolVersion.V4),
    SMALLINT(19, ProtocolVersion.V4),
    TINYINT(20, ProtocolVersion.V4),
    DURATION(21, ProtocolVersion.V5),
    LIST(32),
    MAP(33),
    SET(34),
    UDT(48, ProtocolVersion.V3),
    TUPLE(49, ProtocolVersion.V3);

    final int protocolId;

    final ProtocolVersion minProtocolVersion;

    private static final Name[] nameToIds;

    static {
      int maxCode = -1;
      for (Name name : Name.values()) maxCode = Math.max(maxCode, name.protocolId);
      nameToIds = new Name[maxCode + 1];
      for (Name name : Name.values()) {
        if (nameToIds[name.protocolId] != null) throw new IllegalStateException("Duplicate Id");
        nameToIds[name.protocolId] = name;
      }
    }

    private Name(int protocolId) {
      this(protocolId, ProtocolVersion.V1);
    }

    private Name(int protocolId, ProtocolVersion minProtocolVersion) {
      this.protocolId = protocolId;
      this.minProtocolVersion = minProtocolVersion;
    }

    static Name fromProtocolId(int id) {
      Name name = nameToIds[id];
      if (name == null) throw new DriverInternalError("Unknown data type protocol id: " + id);
      return name;
    }

    /**
     * Return {@code true} if the provided Name is equal to this one, or if they are aliases for
     * each other, and {@code false} otherwise.
     *
     * @param that the Name to compare with the current one.
     * @return {@code true} if the provided Name is equal to this one, or if they are aliases for
     *     each other, and {@code false} otherwise.
     */
    public boolean isCompatibleWith(Name that) {
      return this == that;
    }

    @Override
    public String toString() {
      return super.toString().toLowerCase();
    }
  }

  private static final Map<Name, DataType> primitiveTypeMap =
      new EnumMap<Name, DataType>(Name.class);

  static {
    primitiveTypeMap.put(Name.ASCII, new DataType.NativeType(Name.ASCII));
    primitiveTypeMap.put(Name.BIGINT, new DataType.NativeType(Name.BIGINT));
    primitiveTypeMap.put(Name.BLOB, new DataType.NativeType(Name.BLOB));
    primitiveTypeMap.put(Name.BOOLEAN, new DataType.NativeType(Name.BOOLEAN));
    primitiveTypeMap.put(Name.COUNTER, new DataType.NativeType(Name.COUNTER));
    primitiveTypeMap.put(Name.DECIMAL, new DataType.NativeType(Name.DECIMAL));
    primitiveTypeMap.put(Name.DOUBLE, new DataType.NativeType(Name.DOUBLE));
    primitiveTypeMap.put(Name.FLOAT, new DataType.NativeType(Name.FLOAT));
    primitiveTypeMap.put(Name.INET, new DataType.NativeType(Name.INET));
    primitiveTypeMap.put(Name.INT, new DataType.NativeType(Name.INT));
    primitiveTypeMap.put(Name.TEXT, new DataType.NativeType(Name.TEXT));
    primitiveTypeMap.put(Name.TIMESTAMP, new DataType.NativeType(Name.TIMESTAMP));
    primitiveTypeMap.put(Name.UUID, new DataType.NativeType(Name.UUID));
    primitiveTypeMap.put(Name.VARCHAR, new DataType.NativeType(Name.VARCHAR));
    primitiveTypeMap.put(Name.VARINT, new DataType.NativeType(Name.VARINT));
    primitiveTypeMap.put(Name.TIMEUUID, new DataType.NativeType(Name.TIMEUUID));
    primitiveTypeMap.put(Name.SMALLINT, new DataType.NativeType(Name.SMALLINT));
    primitiveTypeMap.put(Name.TINYINT, new DataType.NativeType(Name.TINYINT));
    primitiveTypeMap.put(Name.DATE, new DataType.NativeType(Name.DATE));
    primitiveTypeMap.put(Name.TIME, new DataType.NativeType(Name.TIME));
    primitiveTypeMap.put(Name.DURATION, new DataType.NativeType(Name.DURATION));
  }

  private static final Set<DataType> primitiveTypeSet =
      ImmutableSet.copyOf(primitiveTypeMap.values());

  protected final DataType.Name name;

  protected DataType(DataType.Name name) {
    this.name = name;
  }

  static DataType decode(
      ByteBuf buffer, ProtocolVersion protocolVersion, CodecRegistry codecRegistry) {
    Name name = Name.fromProtocolId(buffer.readUnsignedShort());
    switch (name) {
      case CUSTOM:
        String className = CBUtil.readString(buffer);
        if (DataTypeClassNameParser.isDuration(className)) {
          return DataType.duration();
        } else if (DataTypeClassNameParser.isUserType(className)
            || DataTypeClassNameParser.isTupleType(className)) {
          return DataTypeClassNameParser.parseOne(className, protocolVersion, codecRegistry);
        } else {
          return custom(className);
        }
      case LIST:
        return list(decode(buffer, protocolVersion, codecRegistry));
      case SET:
        return set(decode(buffer, protocolVersion, codecRegistry));
      case MAP:
        DataType keys = decode(buffer, protocolVersion, codecRegistry);
        DataType values = decode(buffer, protocolVersion, codecRegistry);
        return map(keys, values);
      case UDT:
        String keyspace = CBUtil.readString(buffer);
        String type = CBUtil.readString(buffer);
        int nFields = buffer.readShort() & 0xffff;
        List<UserType.Field> fields = new ArrayList<UserType.Field>(nFields);
        for (int i = 0; i < nFields; i++) {
          String fieldName = CBUtil.readString(buffer);
          DataType fieldType = decode(buffer, protocolVersion, codecRegistry);
          fields.add(new UserType.Field(fieldName, fieldType));
        }
        return new UserType(keyspace, type, false, fields, protocolVersion, codecRegistry);
      case TUPLE:
        nFields = buffer.readShort() & 0xffff;
        List<DataType> types = new ArrayList<DataType>(nFields);
        for (int i = 0; i < nFields; i++) {
          types.add(decode(buffer, protocolVersion, codecRegistry));
        }
        return new TupleType(types, protocolVersion, codecRegistry);
      default:
        return primitiveTypeMap.get(name);
    }
  }

  /**
   * Returns the ASCII type.
   *
   * @return The ASCII type.
   */
  public static DataType ascii() {
    return primitiveTypeMap.get(Name.ASCII);
  }

  /**
   * Returns the BIGINT type.
   *
   * @return The BIGINT type.
   */
  public static DataType bigint() {
    return primitiveTypeMap.get(Name.BIGINT);
  }

  /**
   * Returns the BLOB type.
   *
   * @return The BLOB type.
   */
  public static DataType blob() {
    return primitiveTypeMap.get(Name.BLOB);
  }

  /**
   * Returns the BOOLEAN type.
   *
   * @return The BOOLEAN type.
   */
  public static DataType cboolean() {
    return primitiveTypeMap.get(Name.BOOLEAN);
  }

  /**
   * Returns the COUNTER type.
   *
   * @return The COUNTER type.
   */
  public static DataType counter() {
    return primitiveTypeMap.get(Name.COUNTER);
  }

  /**
   * Returns the DECIMAL type.
   *
   * @return The DECIMAL type.
   */
  public static DataType decimal() {
    return primitiveTypeMap.get(Name.DECIMAL);
  }

  /**
   * Returns the DOUBLE type.
   *
   * @return The DOUBLE type.
   */
  public static DataType cdouble() {
    return primitiveTypeMap.get(Name.DOUBLE);
  }

  /**
   * Returns the FLOAT type.
   *
   * @return The FLOAT type.
   */
  public static DataType cfloat() {
    return primitiveTypeMap.get(Name.FLOAT);
  }

  /**
   * Returns the INET type.
   *
   * @return The INET type.
   */
  public static DataType inet() {
    return primitiveTypeMap.get(Name.INET);
  }

  /**
   * Returns the TINYINT type.
   *
   * @return The TINYINT type.
   */
  public static DataType tinyint() {
    return primitiveTypeMap.get(Name.TINYINT);
  }

  /**
   * Returns the SMALLINT type.
   *
   * @return The SMALLINT type.
   */
  public static DataType smallint() {
    return primitiveTypeMap.get(Name.SMALLINT);
  }

  /**
   * Returns the INT type.
   *
   * @return The INT type.
   */
  public static DataType cint() {
    return primitiveTypeMap.get(Name.INT);
  }

  /**
   * Returns the TEXT type.
   *
   * @return The TEXT type.
   */
  public static DataType text() {
    return primitiveTypeMap.get(Name.TEXT);
  }

  /**
   * Returns the TIMESTAMP type.
   *
   * @return The TIMESTAMP type.
   */
  public static DataType timestamp() {
    return primitiveTypeMap.get(Name.TIMESTAMP);
  }

  /**
   * Returns the DATE type.
   *
   * @return The DATE type.
   */
  public static DataType date() {
    return primitiveTypeMap.get(Name.DATE);
  }

  /**
   * Returns the TIME type.
   *
   * @return The TIME type.
   */
  public static DataType time() {
    return primitiveTypeMap.get(Name.TIME);
  }

  /**
   * Returns the UUID type.
   *
   * @return The UUID type.
   */
  public static DataType uuid() {
    return primitiveTypeMap.get(Name.UUID);
  }

  /**
   * Returns the VARCHAR type.
   *
   * @return The VARCHAR type.
   */
  public static DataType varchar() {
    return primitiveTypeMap.get(Name.VARCHAR);
  }

  /**
   * Returns the VARINT type.
   *
   * @return The VARINT type.
   */
  public static DataType varint() {
    return primitiveTypeMap.get(Name.VARINT);
  }

  /**
   * Returns the TIMEUUID type.
   *
   * @return The TIMEUUID type.
   */
  public static DataType timeuuid() {
    return primitiveTypeMap.get(Name.TIMEUUID);
  }

  /**
   * Returns the type of lists of {@code elementType} elements.
   *
   * @param elementType the type of the list elements.
   * @param frozen whether the list is frozen.
   * @return the type of lists of {@code elementType} elements.
   */
  public static CollectionType list(DataType elementType, boolean frozen) {
    return new DataType.CollectionType(Name.LIST, ImmutableList.of(elementType), frozen);
  }

  /**
   * Returns the type of "not frozen" lists of {@code elementType} elements.
   *
   * <p>This is a shorthand for {@code list(elementType, false);}.
   *
   * @param elementType the type of the list elements.
   * @return the type of "not frozen" lists of {@code elementType} elements.
   */
  public static CollectionType list(DataType elementType) {
    return list(elementType, false);
  }

  /**
   * Returns the type of frozen lists of {@code elementType} elements.
   *
   * <p>This is a shorthand for {@code list(elementType, true);}.
   *
   * @param elementType the type of the list elements.
   * @return the type of frozen lists of {@code elementType} elements.
   */
  public static CollectionType frozenList(DataType elementType) {
    return list(elementType, true);
  }

  /**
   * Returns the type of sets of {@code elementType} elements.
   *
   * @param elementType the type of the set elements.
   * @param frozen whether the set is frozen.
   * @return the type of sets of {@code elementType} elements.
   */
  public static CollectionType set(DataType elementType, boolean frozen) {
    return new DataType.CollectionType(Name.SET, ImmutableList.of(elementType), frozen);
  }

  /**
   * Returns the type of "not frozen" sets of {@code elementType} elements.
   *
   * <p>This is a shorthand for {@code set(elementType, false);}.
   *
   * @param elementType the type of the set elements.
   * @return the type of "not frozen" sets of {@code elementType} elements.
   */
  public static CollectionType set(DataType elementType) {
    return set(elementType, false);
  }

  /**
   * Returns the type of frozen sets of {@code elementType} elements.
   *
   * <p>This is a shorthand for {@code set(elementType, true);}.
   *
   * @param elementType the type of the set elements.
   * @return the type of frozen sets of {@code elementType} elements.
   */
  public static CollectionType frozenSet(DataType elementType) {
    return set(elementType, true);
  }

  /**
   * Returns the type of maps of {@code keyType} to {@code valueType} elements.
   *
   * @param keyType the type of the map keys.
   * @param valueType the type of the map values.
   * @param frozen whether the map is frozen.
   * @return the type of maps of {@code keyType} to {@code valueType} elements.
   */
  public static CollectionType map(DataType keyType, DataType valueType, boolean frozen) {
    return new DataType.CollectionType(Name.MAP, ImmutableList.of(keyType, valueType), frozen);
  }

  /**
   * Returns the type of "not frozen" maps of {@code keyType} to {@code valueType} elements.
   *
   * <p>This is a shorthand for {@code map(keyType, valueType, false);}.
   *
   * @param keyType the type of the map keys.
   * @param valueType the type of the map values.
   * @return the type of "not frozen" maps of {@code keyType} to {@code valueType} elements.
   */
  public static CollectionType map(DataType keyType, DataType valueType) {
    return map(keyType, valueType, false);
  }

  /**
   * Returns the type of frozen maps of {@code keyType} to {@code valueType} elements.
   *
   * <p>This is a shorthand for {@code map(keyType, valueType, true);}.
   *
   * @param keyType the type of the map keys.
   * @param valueType the type of the map values.
   * @return the type of frozen maps of {@code keyType} to {@code valueType} elements.
   */
  public static CollectionType frozenMap(DataType keyType, DataType valueType) {
    return map(keyType, valueType, true);
  }

  /**
   * Returns a Custom type.
   *
   * <p>A custom type is defined by the name of the class used on the Cassandra side to implement
   * it. Note that the support for custom types by the driver is limited.
   *
   * <p>The use of custom types is rarely useful and is thus not encouraged.
   *
   * @param typeClassName the server-side fully qualified class name for the type.
   * @return the custom type for {@code typeClassName}.
   */
  public static DataType.CustomType custom(String typeClassName) {
    if (typeClassName == null) throw new NullPointerException();
    return new DataType.CustomType(Name.CUSTOM, typeClassName);
  }

  /**
   * Returns the Duration type, introduced in Cassandra 3.10.
   *
   * <p>Note that a Duration type does not have a native representation in CQL, and technically, is
   * merely a special {@link DataType#custom(String) custom type} from the driver's point of view.
   *
   * @return the Duration type. The returned instance is a singleton.
   */
  public static DataType duration() {
    return primitiveTypeMap.get(Name.DURATION);
  }

  /**
   * Returns the name of that type.
   *
   * @return the name of that type.
   */
  public Name getName() {
    return name;
  }

  /**
   * Returns whether this data type is frozen.
   *
   * <p>This applies to User Defined Types, tuples and nested collections. Frozen types are
   * serialized as a single value in Cassandra's storage engine, whereas non-frozen types are stored
   * in a form that allows updates to individual subfields.
   *
   * @return whether this data type is frozen.
   */
  public abstract boolean isFrozen();

  /**
   * Returns whether this data type represent a CQL {@link
   * com.datastax.driver.core.DataType.CollectionType collection type}, that is, a list, set or map.
   *
   * @return whether this data type name represent the name of a collection type.
   */
  public boolean isCollection() {
    return this instanceof CollectionType;
  }

  /**
   * Returns the type arguments of this type.
   *
   * <p>Note that only the collection types (LIST, MAP, SET) have type arguments. For the other
   * types, this will return an empty list.
   *
   * <p>For the collection types:
   *
   * <ul>
   *   <li>For lists and sets, this method returns one argument, the type of the elements.
   *   <li>For maps, this method returns two arguments, the first one is the type of the map keys,
   *       the second one is the type of the map values.
   * </ul>
   *
   * @return an immutable list containing the type arguments of this type.
   */
  public List<DataType> getTypeArguments() {
    return Collections.emptyList();
  }

  /**
   * Returns a set of all the primitive types, where primitive types are defined as the types that
   * don't have type arguments (that is excluding lists, sets, maps, tuples and udts).
   *
   * @return returns a set of all the primitive types.
   */
  public static Set<DataType> allPrimitiveTypes() {
    return primitiveTypeSet;
  }

  /**
   * Returns a String representation of this data type suitable for inclusion as a parameter type in
   * a function or aggregate signature.
   *
   * <p>In such places, the String representation might vary from the canonical one as returned by
   * {@link #toString()}; e.g. the {@code frozen} keyword is not accepted.
   *
   * @return a String representation of this data type suitable for inclusion as a parameter type in
   *     a function or aggregate signature.
   */
  public String asFunctionParameterString() {
    return toString();
  }

  /** Instances of this class represent CQL native types, also known as CQL primitive types. */
  public static class NativeType extends DataType {

    private NativeType(DataType.Name name) {
      super(name);
    }

    @Override
    public boolean isFrozen() {
      return false;
    }

    @Override
    public final int hashCode() {
      return (name == Name.TEXT) ? Name.VARCHAR.hashCode() : name.hashCode();
    }

    @Override
    public final boolean equals(Object o) {
      if (!(o instanceof DataType.NativeType)) return false;

      NativeType that = (DataType.NativeType) o;
      return this.name.isCompatibleWith(that.name);
    }

    @Override
    public String toString() {
      return name.toString();
    }
  }

  /** Instances of this class represent collection types, that is, lists, sets or maps. */
  public static class CollectionType extends DataType {

    private final List<DataType> typeArguments;
    private boolean frozen;

    private CollectionType(DataType.Name name, List<DataType> typeArguments, boolean frozen) {
      super(name);
      this.typeArguments = typeArguments;
      this.frozen = frozen;
    }

    @Override
    public boolean isFrozen() {
      return frozen;
    }

    @Override
    public List<DataType> getTypeArguments() {
      return typeArguments;
    }

    @Override
    public final int hashCode() {
      return MoreObjects.hashCode(name, typeArguments);
    }

    @Override
    public final boolean equals(Object o) {
      if (!(o instanceof DataType.CollectionType)) return false;

      DataType.CollectionType d = (DataType.CollectionType) o;
      return name == d.name && typeArguments.equals(d.typeArguments);
    }

    @Override
    public String toString() {
      if (name == Name.MAP) {
        String template = frozen ? "frozen<%s<%s, %s>>" : "%s<%s, %s>";
        return String.format(template, name, typeArguments.get(0), typeArguments.get(1));
      } else {
        String template = frozen ? "frozen<%s<%s>>" : "%s<%s>";
        return String.format(template, name, typeArguments.get(0));
      }
    }

    @Override
    public String asFunctionParameterString() {
      if (name == Name.MAP) {
        String template = "%s<%s, %s>";
        return String.format(
            template,
            name,
            typeArguments.get(0).asFunctionParameterString(),
            typeArguments.get(1).asFunctionParameterString());
      } else {
        String template = "%s<%s>";
        return String.format(template, name, typeArguments.get(0).asFunctionParameterString());
      }
    }
  }

  /**
   * A "custom" type is a type that cannot be expressed as a CQL type.
   *
   * <p>Each custom type is merely identified by the fully qualified {@link
   * #getCustomTypeClassName() class name} that represents this type server-side.
   *
   * <p>The driver provides a minimal support for such types through instances of this class.
   *
   * <p>A codec for custom types can be obtained via {@link TypeCodec#custom(DataType.CustomType)}.
   */
  public static class CustomType extends DataType {

    private final String customClassName;

    private CustomType(DataType.Name name, String className) {
      super(name);
      this.customClassName = className;
    }

    @Override
    public boolean isFrozen() {
      return false;
    }

    /**
     * Returns the fully qualified name of the subtype of {@code
     * org.apache.cassandra.db.marshal.AbstractType} that represents this type server-side.
     *
     * @return the fully qualified name of the subtype of {@code
     *     org.apache.cassandra.db.marshal.AbstractType} that represents this type server-side.
     */
    public String getCustomTypeClassName() {
      return customClassName;
    }

    @Override
    public final int hashCode() {
      return MoreObjects.hashCode(name, customClassName);
    }

    @Override
    public final boolean equals(Object o) {
      if (!(o instanceof DataType.CustomType)) return false;

      DataType.CustomType d = (DataType.CustomType) o;
      return name == d.name && MoreObjects.equal(customClassName, d.customClassName);
    }

    @Override
    public String toString() {
      return String.format("'%s'", customClassName);
    }
  }
}
