/*
 * 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.google.common.collect.ImmutableMap;
import com.google.common.collect.Iterators;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * A User Defined Type (UDT).
 *
 * <p>A UDT is a essentially a named collection of fields (with a name and a type).
 */
public class UserType extends DataType implements Iterable<UserType.Field> {

  static final String TYPE_NAME = "type_name";
  private static final String COLS_NAMES = "field_names";
  static final String COLS_TYPES = "field_types";

  private final String keyspace;
  private final String typeName;
  private final boolean frozen;
  private final ProtocolVersion protocolVersion;

  // can be null, if this object is being constructed from a response message
  // see Responses.Result.Rows.Metadata.decode()
  private volatile CodecRegistry codecRegistry;

  // Note that we don't expose the order of fields, from an API perspective this is a map
  // of String->Field, but internally we care about the order because the serialization format
  // of UDT expects a particular order.
  final Field[] byIdx;
  // For a given name, we can only have one field with that name, so we don't need a int[] in
  // practice. However, storing one element arrays save allocations in UDTValue.getAllIndexesOf
  // implementation.
  final Map<String, int[]> byName;

  private UserType(
      Name name,
      String keyspace,
      String typeName,
      boolean frozen,
      ProtocolVersion protocolVersion,
      CodecRegistry codecRegistry,
      Field[] byIdx,
      Map<String, int[]> byName) {
    super(name);
    this.keyspace = keyspace;
    this.typeName = typeName;
    this.frozen = frozen;
    this.protocolVersion = protocolVersion;
    this.codecRegistry = codecRegistry;
    this.byIdx = byIdx;
    this.byName = byName;
  }

  UserType(
      String keyspace,
      String typeName,
      boolean frozen,
      Collection<Field> fields,
      ProtocolVersion protocolVersion,
      CodecRegistry codecRegistry) {
    this(
        DataType.Name.UDT,
        keyspace,
        typeName,
        frozen,
        protocolVersion,
        codecRegistry,
        fields.toArray(new Field[fields.size()]),
        mapByName(fields));
  }

  private static ImmutableMap<String, int[]> mapByName(Collection<Field> fields) {
    ImmutableMap.Builder<String, int[]> builder = new ImmutableMap.Builder<String, int[]>();
    int i = 0;
    for (Field field : fields) {
      builder.put(field.getName(), new int[] {i});
      i += 1;
    }
    return builder.build();
  }

  static UserType build(
      KeyspaceMetadata ksm,
      Row row,
      VersionNumber version,
      Cluster cluster,
      Map<String, UserType> userTypes) {
    ProtocolVersion protocolVersion =
        cluster.getConfiguration().getProtocolOptions().getProtocolVersion();
    CodecRegistry codecRegistry = cluster.getConfiguration().getCodecRegistry();

    String keyspace = row.getString(KeyspaceMetadata.KS_NAME);
    String name = row.getString(TYPE_NAME);

    List<String> fieldNames = row.getList(COLS_NAMES, String.class);
    List<String> fieldTypes = row.getList(COLS_TYPES, String.class);

    List<Field> fields = new ArrayList<Field>(fieldNames.size());
    for (int i = 0; i < fieldNames.size(); i++) {
      DataType fieldType;
      if (version.getMajor() >= 3.0) {
        fieldType =
            DataTypeCqlNameParser.parse(
                fieldTypes.get(i), cluster, ksm.getName(), userTypes, ksm.userTypes, false, false);
      } else {
        fieldType =
            DataTypeClassNameParser.parseOne(fieldTypes.get(i), protocolVersion, codecRegistry);
      }
      fields.add(new Field(fieldNames.get(i), fieldType));
    }
    return new UserType(keyspace, name, false, fields, protocolVersion, codecRegistry);
  }

  /**
   * Returns a new empty value for this user type definition.
   *
   * @return an empty value for this user type definition.
   */
  public UDTValue newValue() {
    return new UDTValue(this);
  }

  /**
   * The name of the keyspace this UDT is part of.
   *
   * @return the name of the keyspace this UDT is part of.
   */
  public String getKeyspace() {
    return keyspace;
  }

  /**
   * The name of this user type.
   *
   * @return the name of this user type.
   */
  public String getTypeName() {
    return typeName;
  }

  /**
   * Returns the number of fields in this UDT.
   *
   * @return the number of fields in this UDT.
   */
  public int size() {
    return byIdx.length;
  }

  /**
   * Returns whether this UDT contains a given field.
   *
   * @param name the name to check. Note that {@code name} obey the usual CQL identifier rules: it
   *     should be quoted if it denotes a case sensitive identifier (you can use {@link
   *     Metadata#quote} for the quoting).
   * @return {@code true} if this UDT contains a field named {@code name}, {@code false} otherwise.
   */
  public boolean contains(String name) {
    return byName.containsKey(Metadata.handleId(name));
  }

  /**
   * Returns an iterator over the fields of this UDT.
   *
   * @return an iterator over the fields of this UDT.
   */
  @Override
  public Iterator<Field> iterator() {
    return Iterators.forArray(byIdx);
  }

  /**
   * Returns the names of the fields of this UDT.
   *
   * @return the names of the fields of this UDT as a collection.
   */
  public Collection<String> getFieldNames() {
    return byName.keySet();
  }

  /**
   * Returns the type of a given field.
   *
   * @param name the name of the field. Note that {@code name} obey the usual CQL identifier rules:
   *     it should be quoted if it denotes a case sensitive identifier (you can use {@link
   *     Metadata#quote} for the quoting).
   * @return the type of field {@code name} if this UDT has a field of this name, {@code null}
   *     otherwise.
   * @throws IllegalArgumentException if {@code name} is not a field of this UDT definition.
   */
  public DataType getFieldType(String name) {
    int[] idx = byName.get(Metadata.handleId(name));
    if (idx == null)
      throw new IllegalArgumentException(name + " is not a field defined in this definition");

    return byIdx[idx[0]].getType();
  }

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

  public UserType copy(boolean newFrozen) {
    if (newFrozen == frozen) {
      return this;
    } else {
      return new UserType(
          name, keyspace, typeName, newFrozen, protocolVersion, codecRegistry, byIdx, byName);
    }
  }

  @Override
  public int hashCode() {
    int result = name.hashCode();
    result = 31 * result + keyspace.hashCode();
    result = 31 * result + typeName.hashCode();
    result = 31 * result + Arrays.hashCode(byIdx);
    return result;
  }

  @Override
  public boolean equals(Object o) {
    if (!(o instanceof UserType)) return false;

    UserType other = (UserType) o;

    // Note: we don't test byName because it's redundant with byIdx in practice,
    // but also because the map holds 'int[]' which don't have proper equal.
    return name.equals(other.name)
        && keyspace.equals(other.keyspace)
        && typeName.equals(other.typeName)
        && Arrays.equals(byIdx, other.byIdx);
  }

  /**
   * Returns a CQL query representing this user type in human readable form.
   *
   * <p>This method is equivalent to {@link #asCQLQuery} but the ouptut is formatted to be human
   * readable (for some definition of human readable).
   *
   * @return the CQL query representing this user type.
   */
  public String exportAsString() {
    return asCQLQuery(true);
  }

  /**
   * Returns a CQL query representing this user type.
   *
   * <p>This method returns a single 'CREATE TYPE' query corresponding to this UDT definition.
   *
   * <p>Note that the returned string is a single line; the returned query is not formatted in any
   * way.
   *
   * @return the 'CREATE TYPE' query corresponding to this user type.
   * @see #exportAsString
   */
  public String asCQLQuery() {
    return asCQLQuery(false);
  }

  /**
   * Return the protocol version that has been used to deserialize this UDT, or that will be used to
   * serialize it. In most cases this should be the version currently in use by the cluster instance
   * that this UDT belongs to, as reported by {@link ProtocolOptions#getProtocolVersion()}.
   *
   * @return the protocol version that has been used to deserialize this UDT, or that will be used
   *     to serialize it.
   */
  ProtocolVersion getProtocolVersion() {
    return protocolVersion;
  }

  CodecRegistry getCodecRegistry() {
    return codecRegistry;
  }

  void setCodecRegistry(CodecRegistry codecRegistry) {
    this.codecRegistry = codecRegistry;
  }

  private String asCQLQuery(boolean formatted) {
    StringBuilder sb = new StringBuilder();

    sb.append("CREATE TYPE ")
        .append(Metadata.quoteIfNecessary(keyspace))
        .append('.')
        .append(Metadata.quoteIfNecessary(typeName))
        .append(" (");
    if (formatted) {
      TableMetadata.spaceOrNewLine(sb, true);
    }
    for (int i = 0; i < byIdx.length; i++) {
      sb.append(byIdx[i]);
      if (i < byIdx.length - 1) {
        sb.append(',');
        TableMetadata.spaceOrNewLine(sb, formatted);
      } else {
        TableMetadata.newLine(sb, formatted);
      }
    }

    return sb.append(");").toString();
  }

  @Override
  public String toString() {
    String str =
        Metadata.quoteIfNecessary(getKeyspace()) + "." + Metadata.quoteIfNecessary(getTypeName());
    return isFrozen() ? "frozen<" + str + ">" : str;
  }

  @Override
  public String asFunctionParameterString() {
    return Metadata.quoteIfNecessary(getTypeName());
  }

  /** A UDT field. */
  public static class Field {
    private final String name;
    private final DataType type;

    Field(String name, DataType type) {
      this.name = name;
      this.type = type;
    }

    /**
     * Returns the name of the field.
     *
     * @return the name of the field.
     */
    public String getName() {
      return name;
    }

    /**
     * Returns the type of the field.
     *
     * @return the type of the field.
     */
    public DataType getType() {
      return type;
    }

    @Override
    public final int hashCode() {
      return Arrays.hashCode(new Object[] {name, type});
    }

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

      Field other = (Field) o;
      return name.equals(other.name) && type.equals(other.type);
    }

    @Override
    public String toString() {
      return Metadata.quoteIfNecessary(name) + ' ' + type;
    }
  }

  /**
   * A "shallow" definition of a UDT that only contains the keyspace and type name, without any
   * information about the type's structure.
   *
   * <p>This is used for internal dependency analysis only, and never returned to the client.
   *
   * @since 3.0.0
   */
  static class Shallow extends DataType {

    final String keyspaceName;
    final String typeName;
    final boolean frozen;

    Shallow(String keyspaceName, String typeName, boolean frozen) {
      super(Name.UDT);
      this.keyspaceName = keyspaceName;
      this.typeName = typeName;
      this.frozen = frozen;
    }

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