/*
 * 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.querybuilder;

import static com.google.common.base.Preconditions.checkNotNull;

import com.datastax.driver.core.CodecRegistry;
import com.datastax.driver.core.DataType;
import com.datastax.driver.core.Metadata;
import com.datastax.driver.core.ProtocolVersion;
import com.datastax.driver.core.Token;
import com.datastax.driver.core.TypeCodec;
import com.datastax.driver.core.exceptions.InvalidTypeException;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.regex.Pattern;

// Static utilities private to the query builder
abstract class Utils {

  private static final Pattern alphanumeric = Pattern.compile("\\w+"); // this includes _
  private static final Pattern cnamePattern = Pattern.compile("\\w+(?:\\[.+\\])?");

  /**
   * Deal with case sensitivity for a given element id (keyspace, table, column, etc.)
   *
   * <p>This method is used to convert identifiers provided by the client (through methods such as
   * getKeyspace(String)), to the format used internally by the driver.
   *
   * <p>We expect client-facing APIs to behave like cqlsh, that is: - identifiers that are
   * mixed-case or contain special characters should be quoted. - unquoted identifiers will be
   * lowercased: getKeyspace("Foo") will look for a keyspace named "foo"
   *
   * <p>Copied from {@link Metadata#handleId(String)} to avoid making it public.
   */
  static String handleId(String id) {
    // Shouldn't really happen for this method, but no reason to fail here
    if (id == null) return null;

    if (alphanumeric.matcher(id).matches()) return id.toLowerCase();

    // Check if it's enclosed in quotes. If it is, remove them and unescape internal double quotes
    if (!id.isEmpty() && id.charAt(0) == '"' && id.charAt(id.length() - 1) == '"')
      return id.substring(1, id.length() - 1).replaceAll("\"\"", "\"");

    // Otherwise, just return the id.
    // Note that this is a bit at odds with the rules explained above, because the client can pass
    // an
    // identifier that contains special characters, without the need to quote it.
    // Still it's better to be lenient here rather than throwing an exception.
    return id;
  }

  static StringBuilder joinAndAppend(
      StringBuilder sb,
      CodecRegistry codecRegistry,
      String separator,
      List<? extends Appendeable> values,
      List<Object> variables) {
    for (int i = 0; i < values.size(); i++) {
      if (i > 0) sb.append(separator);
      values.get(i).appendTo(sb, variables, codecRegistry);
    }
    return sb;
  }

  static StringBuilder joinAndAppendNames(
      StringBuilder sb, CodecRegistry codecRegistry, List<?> values) {
    for (int i = 0; i < values.size(); i++) {
      if (i > 0) sb.append(",");
      appendName(values.get(i), codecRegistry, sb);
    }
    return sb;
  }

  static StringBuilder joinAndAppendValues(
      StringBuilder sb, CodecRegistry codecRegistry, List<?> values, List<Object> variables) {
    for (int i = 0; i < values.size(); i++) {
      if (i > 0) sb.append(",");
      appendValue(values.get(i), codecRegistry, sb, variables);
    }
    return sb;
  }

  static StringBuilder appendValue(
      Object value, CodecRegistry codecRegistry, StringBuilder sb, List<Object> variables) {
    if (value == null) {
      sb.append("null");
    } else if (value instanceof BindMarker) {
      sb.append(value);
    } else if (value instanceof FCall) {
      FCall fcall = (FCall) value;
      sb.append(fcall.name).append('(');
      for (int i = 0; i < fcall.parameters.length; i++) {
        if (i > 0) sb.append(',');
        appendValue(fcall.parameters[i], codecRegistry, sb, variables);
      }
      sb.append(')');
    } else if (value instanceof Cast) {
      Cast cast = (Cast) value;
      sb.append("CAST(");
      appendName(cast.column, codecRegistry, sb);
      sb.append(" AS ").append(cast.targetType).append(")");
    } else if (value instanceof CName) {
      appendName(((CName) value).name, codecRegistry, sb);
    } else if (value instanceof RawString) {
      sb.append(value.toString());
    } else if (value instanceof List && !isSerializable(value)) {
      // bind variables are not supported inside collection literals
      appendList((List<?>) value, codecRegistry, sb);
    } else if (value instanceof Set && !isSerializable(value)) {
      // bind variables are not supported inside collection literals
      appendSet((Set<?>) value, codecRegistry, sb);
    } else if (value instanceof Map && !isSerializable(value)) {
      // bind variables are not supported inside collection literals
      appendMap((Map<?, ?>) value, codecRegistry, sb);
    } else if (variables == null || !isSerializable(value)) {
      // we are not collecting statement values (variables == null)
      // or the value is meant to be forcefully appended to the query string:
      // format it with the appropriate codec and append it now
      TypeCodec<Object> codec = codecRegistry.codecFor(value);
      sb.append(codec.format(value));
    } else {
      // Do not format the value nor append it to the query string:
      // use a bind marker instead,
      // but add the value the the statement's variables list
      sb.append('?');
      variables.add(value);
      return sb;
    }
    return sb;
  }

  private static StringBuilder appendList(
      List<?> l, CodecRegistry codecRegistry, StringBuilder sb) {
    sb.append('[');
    for (int i = 0; i < l.size(); i++) {
      if (i > 0) sb.append(',');
      appendValue(l.get(i), codecRegistry, sb, null);
    }
    sb.append(']');
    return sb;
  }

  private static StringBuilder appendSet(Set<?> s, CodecRegistry codecRegistry, StringBuilder sb) {
    sb.append('{');
    boolean first = true;
    for (Object elt : s) {
      if (first) first = false;
      else sb.append(',');
      appendValue(elt, codecRegistry, sb, null);
    }
    sb.append('}');
    return sb;
  }

  private static StringBuilder appendMap(
      Map<?, ?> m, CodecRegistry codecRegistry, StringBuilder sb) {
    sb.append('{');
    boolean first = true;
    for (Map.Entry<?, ?> entry : m.entrySet()) {
      if (first) first = false;
      else sb.append(',');
      appendValue(entry.getKey(), codecRegistry, sb, null);
      sb.append(':');
      appendValue(entry.getValue(), codecRegistry, sb, null);
    }
    sb.append('}');
    return sb;
  }

  static boolean containsBindMarker(Object value) {
    if (value instanceof BindMarker) return true;
    if (value instanceof FCall)
      for (Object param : ((FCall) value).parameters) if (containsBindMarker(param)) return true;
    if (value instanceof Collection)
      for (Object elt : (Collection<?>) value) if (containsBindMarker(elt)) return true;
    if (value instanceof Map)
      for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet())
        if (containsBindMarker(entry.getKey()) || containsBindMarker(entry.getValue())) return true;
    return false;
  }

  static boolean containsSpecialValue(Object value) {
    if (value instanceof BindMarker
        || value instanceof FCall
        || value instanceof CName
        || value instanceof RawString) return true;
    if (value instanceof Collection)
      for (Object elt : (Collection<?>) value) if (containsSpecialValue(elt)) return true;
    if (value instanceof Map)
      for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet())
        if (containsSpecialValue(entry.getKey()) || containsSpecialValue(entry.getValue()))
          return true;
    return false;
  }

  /**
   * Return true if the given value is likely to find a suitable codec to be serialized as a query
   * parameter. If the value is not serializable, it must be included in the query string. Non
   * serializable values include special values such as function calls, column names and bind
   * markers, and collections thereof. We also don't serialize fixed size number types. The reason
   * is that if we do it, we will force a particular size (4 bytes for ints, ...) and for the query
   * builder, we don't want users to have to bother with that.
   *
   * @param value the value to inspect.
   * @return true if the value is serializable, false otherwise.
   */
  static boolean isSerializable(Object value) {
    if (containsSpecialValue(value)) return false;
    if (value instanceof Number && !(value instanceof BigInteger || value instanceof BigDecimal))
      return false;
    if (value instanceof Collection)
      for (Object elt : (Collection<?>) value) if (!isSerializable(elt)) return false;
    if (value instanceof Map)
      for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet())
        if (!isSerializable(entry.getKey()) || !isSerializable(entry.getValue())) return false;
    return true;
  }

  static boolean isIdempotent(Object value) {
    if (value == null) {
      return true;
    } else if (value instanceof Assignment) {
      Assignment assignment = (Assignment) value;
      return assignment.isIdempotent();
    } else if (value instanceof FCall) {
      return false;
    } else if (value instanceof RawString) {
      return false;
    } else if (value instanceof Collection) {
      for (Object elt : ((Collection<?>) value)) {
        if (!isIdempotent(elt)) return false;
      }
      return true;
    } else if (value instanceof Map) {
      for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet()) {
        if (!isIdempotent(entry.getKey()) || !isIdempotent(entry.getValue())) return false;
      }
    } else if (value instanceof Clause) {
      Object clauseValue = ((Clause) value).firstValue();
      return isIdempotent(clauseValue);
    }
    return true;
  }

  static StringBuilder appendName(String name, StringBuilder sb) {
    name = name.trim();
    // FIXME: checking for token( specifically is uber ugly, we'll need some better solution.
    if (name.startsWith("\"") || name.startsWith("token(") || cnamePattern.matcher(name).matches())
      sb.append(name);
    else sb.append('"').append(name).append('"');
    return sb;
  }

  static StringBuilder appendName(Object name, CodecRegistry codecRegistry, StringBuilder sb) {
    if (name instanceof String) {
      appendName((String) name, sb);
    } else if (name instanceof CName) {
      appendName(((CName) name).name, sb);
    } else if (name instanceof Path) {
      String[] segments = ((Path) name).segments;
      for (int i = 0; i < segments.length; i++) {
        if (i > 0) sb.append('.');
        appendName(segments[i], sb);
      }
    } else if (name instanceof FCall) {
      FCall fcall = (FCall) name;
      sb.append(fcall.name).append('(');
      for (int i = 0; i < fcall.parameters.length; i++) {
        if (i > 0) sb.append(',');
        appendValue(fcall.parameters[i], codecRegistry, sb, null);
      }
      sb.append(')');
    } else if (name instanceof Alias) {
      Alias alias = (Alias) name;
      appendName(alias.column, codecRegistry, sb);
      sb.append(" AS ").append(alias.alias);
    } else if (name instanceof Cast) {
      Cast cast = (Cast) name;
      sb.append("CAST(");
      appendName(cast.column, codecRegistry, sb);
      sb.append(" AS ").append(cast.targetType).append(")");
    } else if (name instanceof RawString) {
      sb.append(((RawString) name).str);
    } else {
      throw new IllegalArgumentException(
          String.format("Invalid column %s of type unknown of the query builder", name));
    }
    return sb;
  }

  /**
   * Utility method to serialize user-provided values.
   *
   * <p>This method is a copy of the one declared in {@link
   * com.datastax.driver.core.SimpleStatement}, it was duplicated to avoid having to make it public.
   *
   * <p>It is useful in situations where there is no metadata available and the underlying CQL type
   * for the values is not known.
   *
   * <p>This situation happens when a {@link com.datastax.driver.core.SimpleStatement} or a {@link
   * com.datastax.driver.core.querybuilder.BuiltStatement} (Query Builder) contain values; in these
   * places, the driver has no way to determine the right CQL type to use.
   *
   * <p>This method performs a best-effort heuristic to guess which codec to use. Note that this is
   * not particularly efficient as the codec registry needs to iterate over the registered codecs
   * until it finds a suitable one.
   *
   * @param values The values to convert.
   * @param protocolVersion The protocol version to use.
   * @param codecRegistry The {@link CodecRegistry} to use.
   * @return The converted values.
   */
  static ByteBuffer[] convert(
      Object[] values, ProtocolVersion protocolVersion, CodecRegistry codecRegistry) {
    ByteBuffer[] serializedValues = new ByteBuffer[values.length];
    for (int i = 0; i < values.length; i++) {
      Object value = values[i];
      if (value == null) {
        // impossible to locate the right codec when object is null,
        // so forcing the result to null
        serializedValues[i] = null;
      } else {
        if (value instanceof Token) {
          // bypass CodecRegistry for Token instances
          serializedValues[i] = ((Token) value).serialize(protocolVersion);
        } else {
          try {
            TypeCodec<Object> codec = codecRegistry.codecFor(value);
            serializedValues[i] = codec.serialize(value, protocolVersion);
          } catch (Exception e) {
            // Catch and rethrow to provide a more helpful error message (one that include which
            // value is bad)
            throw new InvalidTypeException(
                String.format(
                    "Value %d of type %s does not correspond to any CQL3 type",
                    i, value.getClass()),
                e);
          }
        }
      }
    }
    return serializedValues;
  }

  /**
   * Utility method to assemble different routing key components into a single {@link ByteBuffer}.
   * Mainly intended for statements that need to generate a routing key out of their current values.
   *
   * <p>This method is a copy of the one declared in {@link
   * com.datastax.driver.core.SimpleStatement}, it was duplicated to avoid having to make it public.
   *
   * @param buffers the components of the routing key.
   * @return A ByteBuffer containing the serialized routing key
   */
  static ByteBuffer compose(ByteBuffer... buffers) {
    if (buffers.length == 1) return buffers[0];

    int totalLength = 0;
    for (ByteBuffer bb : buffers) totalLength += 2 + bb.remaining() + 1;

    ByteBuffer out = ByteBuffer.allocate(totalLength);
    for (ByteBuffer buffer : buffers) {
      ByteBuffer bb = buffer.duplicate();
      putShortLength(out, bb.remaining());
      out.put(bb);
      out.put((byte) 0);
    }
    out.flip();
    return out;
  }

  static void putShortLength(ByteBuffer bb, int length) {
    bb.put((byte) ((length >> 8) & 0xFF));
    bb.put((byte) (length & 0xFF));
  }

  abstract static class Appendeable {
    abstract void appendTo(StringBuilder sb, List<Object> values, CodecRegistry codecRegistry);

    abstract boolean containsBindMarker();
  }

  static class RawString {
    private final String str;

    RawString(String str) {
      this.str = str;
    }

    @Override
    public String toString() {
      return str;
    }
  }

  static class FCall {

    private final String name;
    private final Object[] parameters;

    FCall(String name, Object... parameters) {
      checkNotNull(name);
      this.name = name;
      this.parameters = parameters;
    }

    @Override
    public String toString() {
      StringBuilder sb = new StringBuilder();
      sb.append(name).append('(');
      for (int i = 0; i < parameters.length; i++) {
        if (i > 0) sb.append(',');
        sb.append(parameters[i]);
      }
      sb.append(')');
      return sb.toString();
    }
  }

  static class CName {
    private final String name;

    CName(String name) {
      this.name = name;
    }

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

  static class Alias {
    private final Object column;
    private final String alias;

    Alias(Object column, String alias) {
      this.column = column;
      this.alias = alias;
    }

    @Override
    public String toString() {
      return String.format("%s AS %s", column, alias);
    }
  }

  static class Cast {
    private final Object column;
    private final DataType targetType;

    Cast(Object column, DataType targetType) {
      this.column = column;
      this.targetType = targetType;
    }

    @Override
    public String toString() {
      return String.format("CAST(%s AS %s)", column, targetType);
    }
  }

  static class Path {

    private final String[] segments;

    Path(String... segments) {
      this.segments = segments;
    }
  }
}
