/*
 * 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 static com.datastax.driver.core.querybuilder.QueryBuilder.bindMarker;
import static com.datastax.driver.core.querybuilder.QueryBuilder.eq;
import static com.datastax.driver.core.querybuilder.QueryBuilder.insertInto;
import static com.datastax.driver.core.querybuilder.QueryBuilder.select;
import static org.assertj.core.api.Assertions.assertThat;

import com.datastax.driver.core.exceptions.InvalidTypeException;
import com.datastax.driver.core.querybuilder.BuiltStatement;
import com.datastax.driver.core.utils.CassandraVersion;
import com.datastax.driver.core.utils.MoreObjects;
import com.google.common.reflect.TypeParameter;
import com.google.common.reflect.TypeToken;
import java.math.BigDecimal;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.Test;

public class TypeCodecEncapsulationIntegrationTest extends CCMTestsSupport {

  // @formatter:off
  private static final TypeToken<NumberBox<Integer>> NUMBERBOX_OF_INTEGER_TOKEN =
      new TypeToken<NumberBox<Integer>>() {};
  private static final TypeToken<NumberBox<Long>> NUMBERBOX_OF_LONG_TOKEN =
      new TypeToken<NumberBox<Long>>() {};
  private static final TypeToken<NumberBox<Float>> NUMBERBOX_OF_FLOAT_TOKEN =
      new TypeToken<NumberBox<Float>>() {};
  private static final TypeToken<NumberBox<Double>> NUMBERBOX_OF_DOUBLE_TOKEN =
      new TypeToken<NumberBox<Double>>() {};
  private static final TypeToken<NumberBox<BigInteger>> NUMBERBOX_OF_BIGINTEGER_TOKEN =
      new TypeToken<NumberBox<BigInteger>>() {};
  private static final TypeToken<NumberBox<BigDecimal>> NUMBERBOX_OF_BIGDECIMAL_TOKEN =
      new TypeToken<NumberBox<BigDecimal>>() {};
  // @formatter:on

  private final String insertQuery =
      "INSERT INTO \"myTable\" (c_int, c_bigint, c_float, c_double, c_varint, c_decimal) VALUES (?, ?, ?, ?, ?, ?)";

  private final String selectQuery =
      "SELECT c_int, c_bigint, c_float, c_double, c_varint, c_decimal FROM \"myTable\" WHERE c_int = ? and c_bigint = ?";

  private BuiltStatement insertStmt;
  private BuiltStatement selectStmt;

  private int n_int = 42;
  private long n_bigint = 4242;
  private float n_float = 42.42f;
  private double n_double = 4242.42d;
  private BigInteger n_varint = new BigInteger("424242");
  private BigDecimal n_decimal = new BigDecimal("424242.42");

  @Override
  public void onTestContextInitialized() {
    execute(
        "CREATE TABLE \"myTable\" ("
            + "c_int int, "
            + "c_bigint bigint, "
            + "c_float float, "
            + "c_double double, "
            + "c_varint varint, "
            + "c_decimal decimal, "
            + "PRIMARY KEY (c_int, c_bigint)"
            + ")");
  }

  @Override
  public Cluster.Builder createClusterBuilder() {
    return Cluster.builder()
        .withCodecRegistry(
            new CodecRegistry()
                .register(
                    new NumberBoxCodec<Integer>(TypeCodec.cint()),
                    new NumberBoxCodec<Long>(TypeCodec.bigint()),
                    new NumberBoxCodec<Float>(TypeCodec.cfloat()),
                    new NumberBoxCodec<Double>(TypeCodec.cdouble()),
                    new NumberBoxCodec<BigInteger>(TypeCodec.varint()),
                    new NumberBoxCodec<BigDecimal>(TypeCodec.decimal())));
  }

  @BeforeMethod(groups = "short")
  public void createBuiltStatements() throws Exception {
    insertStmt =
        insertInto("\"myTable\"")
            .value("c_int", bindMarker())
            .value("c_bigint", bindMarker())
            .value("c_float", bindMarker())
            .value("c_double", bindMarker())
            .value("c_varint", bindMarker())
            .value("c_decimal", bindMarker());
    selectStmt =
        select("c_int", "c_bigint", "c_float", "c_double", "c_varint", "c_decimal")
            .from("\"myTable\"")
            .where(eq("c_int", bindMarker()))
            .and(eq("c_bigint", bindMarker()));
  }

  @Test(groups = "short")
  @CassandraVersion("2.0.0")
  public void should_use_custom_codecs_with_simple_statements() {
    session()
        .execute(
            insertQuery,
            n_int,
            new NumberBox<Long>(n_bigint),
            new NumberBox<Float>(n_float),
            new NumberBox<Double>(n_double),
            new NumberBox<BigInteger>(n_varint),
            new NumberBox<BigDecimal>(n_decimal));
    ResultSet rows = session().execute(selectQuery, n_int, new NumberBox<Long>(n_bigint));
    Row row = rows.one();
    assertRow(row);
  }

  @Test(groups = "short")
  public void should_use_custom_codecs_with_prepared_statements_1() {
    session()
        .execute(
            session()
                .prepare(insertQuery)
                .bind(
                    n_int,
                    new NumberBox<Long>(n_bigint),
                    new NumberBox<Float>(n_float),
                    new NumberBox<Double>(n_double),
                    new NumberBox<BigInteger>(n_varint),
                    new NumberBox<BigDecimal>(n_decimal)));
    PreparedStatement ps = session().prepare(selectQuery);
    ResultSet rows = session().execute(ps.bind(n_int, new NumberBox<Long>(n_bigint)));
    Row row = rows.one();
    assertRow(row);
  }

  @Test(groups = "short")
  public void should_use_custom_codecs_with_prepared_statements_2() {
    session()
        .execute(
            session()
                .prepare(insertQuery)
                .bind()
                .set(0, new NumberBox<Integer>(n_int), NUMBERBOX_OF_INTEGER_TOKEN)
                .set(1, new NumberBox<Long>(n_bigint), NUMBERBOX_OF_LONG_TOKEN)
                .set(2, new NumberBox<Float>(n_float), NUMBERBOX_OF_FLOAT_TOKEN)
                .set(3, new NumberBox<Double>(n_double), NUMBERBOX_OF_DOUBLE_TOKEN)
                .set(4, new NumberBox<BigInteger>(n_varint), NUMBERBOX_OF_BIGINTEGER_TOKEN)
                .set(5, new NumberBox<BigDecimal>(n_decimal), NUMBERBOX_OF_BIGDECIMAL_TOKEN));
    PreparedStatement ps = session().prepare(selectQuery);
    ResultSet rows =
        session()
            .execute(
                ps.bind()
                    .set(0, new NumberBox<Integer>(n_int), NUMBERBOX_OF_INTEGER_TOKEN)
                    .set(1, new NumberBox<Long>(n_bigint), NUMBERBOX_OF_LONG_TOKEN));
    Row row = rows.one();
    assertRow(row);
  }

  @Test(groups = "short")
  public void should_use_custom_codecs_with_built_statements_1() {
    session()
        .execute(
            session()
                .prepare(insertStmt)
                .bind(
                    n_int,
                    new NumberBox<Long>(n_bigint),
                    new NumberBox<Float>(n_float),
                    new NumberBox<Double>(n_double),
                    new NumberBox<BigInteger>(n_varint),
                    new NumberBox<BigDecimal>(n_decimal)));
    PreparedStatement ps = session().prepare(selectStmt);
    ResultSet rows = session().execute(ps.bind(n_int, new NumberBox<Long>(n_bigint)));
    Row row = rows.one();
    assertRow(row);
  }

  @Test(groups = "short")
  public void should_use_custom_codecs_with_built_statements_2() {
    session()
        .execute(
            session()
                .prepare(insertStmt)
                .bind()
                .set(0, new NumberBox<Integer>(n_int), NUMBERBOX_OF_INTEGER_TOKEN)
                .set(1, new NumberBox<Long>(n_bigint), NUMBERBOX_OF_LONG_TOKEN)
                .set(2, new NumberBox<Float>(n_float), NUMBERBOX_OF_FLOAT_TOKEN)
                .set(3, new NumberBox<Double>(n_double), NUMBERBOX_OF_DOUBLE_TOKEN)
                .set(4, new NumberBox<BigInteger>(n_varint), NUMBERBOX_OF_BIGINTEGER_TOKEN)
                .set(5, new NumberBox<BigDecimal>(n_decimal), NUMBERBOX_OF_BIGDECIMAL_TOKEN));
    PreparedStatement ps = session().prepare(selectStmt);
    ResultSet rows =
        session()
            .execute(
                ps.bind()
                    .set(0, new NumberBox<Integer>(n_int), NUMBERBOX_OF_INTEGER_TOKEN)
                    .set(1, new NumberBox<Long>(n_bigint), NUMBERBOX_OF_LONG_TOKEN));

    Row row = rows.one();
    assertRow(row);
  }

  private void assertRow(Row row) {
    // using getInt, etc: the default codecs are used
    // and values are deserialized the traditional way
    assertThat(row.getInt(0)).isEqualTo(n_int);
    assertThat(row.getLong(1)).isEqualTo(n_bigint);
    assertThat(row.getFloat(2)).isEqualTo(n_float);
    assertThat(row.getDouble(3)).isEqualTo(n_double);
    assertThat(row.getVarint(4)).isEqualTo(n_varint);
    assertThat(row.getDecimal(5)).isEqualTo(n_decimal);
    // with getObject, the first matching codec is the default one
    assertThat(row.getObject(0)).isEqualTo(n_int);
    assertThat(row.getObject(1)).isEqualTo(n_bigint);
    assertThat(row.getObject(2)).isEqualTo(n_float);
    assertThat(row.getObject(3)).isEqualTo(n_double);
    assertThat(row.getObject(4)).isEqualTo(n_varint);
    assertThat(row.getObject(5)).isEqualTo(n_decimal);
    // with get + type
    // we go back to the default codecs
    assertThat(row.get(0, Integer.class)).isEqualTo(n_int);
    assertThat(row.get(1, Long.class)).isEqualTo(n_bigint);
    assertThat(row.get(2, Float.class)).isEqualTo(n_float);
    assertThat(row.get(3, Double.class)).isEqualTo(n_double);
    assertThat(row.get(4, BigInteger.class)).isEqualTo(n_varint);
    assertThat(row.get(5, BigDecimal.class)).isEqualTo(n_decimal);
    // with get + type, but enforcing NumberBox types
    // we get the NumberBox codecs instead
    assertThat(row.get(0, NUMBERBOX_OF_INTEGER_TOKEN)).isEqualTo(new NumberBox<Integer>(n_int));
    assertThat(row.get(1, NUMBERBOX_OF_LONG_TOKEN)).isEqualTo(new NumberBox<Long>(n_bigint));
    assertThat(row.get(2, NUMBERBOX_OF_FLOAT_TOKEN)).isEqualTo(new NumberBox<Float>(n_float));
    assertThat(row.get(3, NUMBERBOX_OF_DOUBLE_TOKEN)).isEqualTo(new NumberBox<Double>(n_double));
    assertThat(row.get(4, NUMBERBOX_OF_BIGINTEGER_TOKEN))
        .isEqualTo(new NumberBox<BigInteger>(n_varint));
    assertThat(row.get(5, NUMBERBOX_OF_BIGDECIMAL_TOKEN))
        .isEqualTo(new NumberBox<BigDecimal>(n_decimal));
  }

  private class NumberBoxCodec<T extends Number> extends TypeCodec<NumberBox<T>> {

    private final TypeCodec<T> numberCodec;

    protected NumberBoxCodec(TypeCodec<T> numberCodec) {
      // @formatter:off
      super(
          numberCodec.getCqlType(),
          new TypeToken<NumberBox<T>>() {}.where(
              new TypeParameter<T>() {}, numberCodec.getJavaType()));
      // @formatter:on
      this.numberCodec = numberCodec;
    }

    public boolean accepts(Object value) {
      return value instanceof NumberBox && numberCodec.accepts(((NumberBox) value).getNumber());
    }

    @Override
    public ByteBuffer serialize(NumberBox<T> value, ProtocolVersion protocolVersion)
        throws InvalidTypeException {
      return numberCodec.serialize(value.getNumber(), protocolVersion);
    }

    @Override
    public NumberBox<T> deserialize(ByteBuffer bytes, ProtocolVersion protocolVersion)
        throws InvalidTypeException {
      return new NumberBox<T>(numberCodec.deserialize(bytes, protocolVersion));
    }

    @Override
    public NumberBox<T> parse(String value) throws InvalidTypeException {
      return new NumberBox<T>(numberCodec.parse(value));
    }

    @Override
    public String format(NumberBox<T> value) throws InvalidTypeException {
      return numberCodec.format(value.getNumber());
    }
  }

  private class NumberBox<T extends Number> {

    private final T number;

    private NumberBox(T number) {
      this.number = number;
    }

    public T getNumber() {
      return number;
    }

    @Override
    public boolean equals(Object o) {
      if (this == o) return true;
      if (o == null || getClass() != o.getClass()) return false;
      NumberBox numberBox = (NumberBox) o;
      return MoreObjects.equal(number, numberBox.number);
    }

    @Override
    public int hashCode() {
      return MoreObjects.hashCode(number);
    }
  }
}
