/**
 * 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.pulsar.schema.compatibility;

import com.google.common.collect.Sets;
import lombok.extern.slf4j.Slf4j;
import org.apache.pulsar.broker.auth.MockedPulsarServiceBaseTest;
import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.ConsumerBuilder;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.ProducerBuilder;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SchemaSerializationException;
import org.apache.pulsar.client.api.schema.SchemaDefinition;
import org.apache.pulsar.common.naming.NamespaceName;
import org.apache.pulsar.common.naming.TopicDomain;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterData;
import org.apache.pulsar.common.policies.data.SchemaCompatibilityStrategy;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.testng.Assert;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

import java.util.Collections;
import java.util.concurrent.ThreadLocalRandom;

import static org.apache.pulsar.common.naming.TopicName.PUBLIC_TENANT;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;


@Slf4j
public class SchemaCompatibilityCheckTest extends MockedPulsarServiceBaseTest {
    private final static String CLUSTER_NAME = "test";

    @BeforeMethod
    @Override
    public void setup() throws Exception {
        super.internalSetup();

        // Setup namespaces
        admin.clusters().createCluster(CLUSTER_NAME, new ClusterData(pulsar.getBrokerServiceUrl()));
        TenantInfo tenantInfo = new TenantInfo();
        tenantInfo.setAllowedClusters(Collections.singleton(CLUSTER_NAME));
        admin.tenants().createTenant(PUBLIC_TENANT, tenantInfo);
    }

    @AfterMethod
    @Override
    public void cleanup() throws Exception {
        super.internalCleanup();
    }

    @DataProvider(name = "CanReadLastSchemaCompatibilityStrategy")
    public Object[] canReadLastSchemaCompatibilityStrategy() {
        return new Object[] {
                SchemaCompatibilityStrategy.BACKWARD,
                SchemaCompatibilityStrategy.FORWARD_TRANSITIVE,
                SchemaCompatibilityStrategy.FORWARD,
                SchemaCompatibilityStrategy.FULL
        };
    }

    @DataProvider(name = "ReadAllCheckSchemaCompatibilityStrategy")
    public Object[] readAllCheckSchemaCompatibilityStrategy() {
        return new Object[] {
                SchemaCompatibilityStrategy.BACKWARD_TRANSITIVE,
                SchemaCompatibilityStrategy.FULL_TRANSITIVE
        };
    }

    @DataProvider(name = "AllCheckSchemaCompatibilityStrategy")
    public Object[] allCheckSchemaCompatibilityStrategy() {
        return new Object[] {
                SchemaCompatibilityStrategy.BACKWARD,
                SchemaCompatibilityStrategy.FORWARD_TRANSITIVE,
                SchemaCompatibilityStrategy.FORWARD,
                SchemaCompatibilityStrategy.FULL,
                SchemaCompatibilityStrategy.BACKWARD_TRANSITIVE,
                SchemaCompatibilityStrategy.FULL_TRANSITIVE
        };
    }

    @Test(dataProvider =  "CanReadLastSchemaCompatibilityStrategy")
    public void testConsumerCompatibilityCheckCanReadLastTest(SchemaCompatibilityStrategy schemaCompatibilityStrategy) throws Exception {
        final String tenant = PUBLIC_TENANT;
        final String topic = "test-consumer-compatibility";

            String namespace = "test-namespace-" + randomName(16);
            String fqtn = TopicName.get(
                    TopicDomain.persistent.value(),
                    tenant,
                    namespace,
                    topic
            ).toString();

            NamespaceName namespaceName = NamespaceName.get(tenant, namespace);

            admin.namespaces().createNamespace(
                    tenant + "/" + namespace,
                    Sets.newHashSet(CLUSTER_NAME)
            );

            admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy);
            admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo());
            admin.schemas().createSchema(fqtn, Schema.AVRO(SchemaDefinition.builder()
                    .withAlwaysAllowNull(false).withPojo(Schemas.PersonTwo.class).build()).getSchemaInfo());

            Consumer<Schemas.PersonThree> consumerThree = pulsarClient.newConsumer(Schema.AVRO(
                    SchemaDefinition.<Schemas.PersonThree>builder().withAlwaysAllowNull
                            (false).withSupportSchemaVersioning(true).
                            withPojo(Schemas.PersonThree.class).build()))
                    .subscriptionName("test")
                    .topic(fqtn)
                    .subscribe();

            Producer<Schemas.PersonOne> producerOne = pulsarClient
                    .newProducer(Schema.AVRO(Schemas.PersonOne.class))
                    .topic(fqtn)
                    .create();


            Schemas.PersonOne personOne = new Schemas.PersonOne();
            personOne.id = 1;

            producerOne.send(personOne);
            Message<Schemas.PersonThree> message = null;

            try {
                message = consumerThree.receive();
                message.getValue();
            } catch (Exception e) {
                Assert.assertTrue(e instanceof SchemaSerializationException);
                consumerThree.acknowledge(message);
            }

            Producer<Schemas.PersonTwo> producerTwo = pulsarClient
                    .newProducer(Schema.AVRO(SchemaDefinition.<Schemas.PersonTwo>builder().withAlwaysAllowNull
                            (false).withSupportSchemaVersioning(true).
                            withPojo(Schemas.PersonTwo.class).build()))
                    .topic(fqtn)
                    .create();

            Schemas.PersonTwo personTwo = new Schemas.PersonTwo();
            personTwo.id = 1;
            personTwo.name = "Jerry";
            producerTwo.send(personTwo);

            message = consumerThree.receive();
            Schemas.PersonThree personThree = message.getValue();
            consumerThree.acknowledge(message);

            assertEquals(personThree.id, 1);
            assertEquals(personThree.name, "Jerry");

            consumerThree.close();
            producerOne.close();
            producerTwo.close();
    }

    @Test(dataProvider = "ReadAllCheckSchemaCompatibilityStrategy")
    public void testConsumerCompatibilityReadAllCheckTest(SchemaCompatibilityStrategy schemaCompatibilityStrategy) throws Exception {
        final String tenant = PUBLIC_TENANT;
        final String topic = "test-consumer-compatibility";
        String namespace = "test-namespace-" + randomName(16);
        String fqtn = TopicName.get(
                TopicDomain.persistent.value(),
                tenant,
                namespace,
                topic
        ).toString();

        NamespaceName namespaceName = NamespaceName.get(tenant, namespace);

        admin.namespaces().createNamespace(
                tenant + "/" + namespace,
                Sets.newHashSet(CLUSTER_NAME)
        );

        admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy);
        admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo());
        admin.schemas().createSchema(fqtn, Schema.AVRO(SchemaDefinition.builder()
                .withAlwaysAllowNull(false).withPojo(Schemas.PersonTwo.class).build()).getSchemaInfo());
        try {
            pulsarClient.newConsumer(Schema.AVRO(
                    SchemaDefinition.<Schemas.PersonThree>builder().withAlwaysAllowNull
                            (false).withSupportSchemaVersioning(true).
                            withPojo(Schemas.PersonThree.class).build()))
                    .subscriptionName("test")
                    .topic(fqtn)
                    .subscribe();

        } catch (Exception e) {
            Assert.assertTrue(e.getMessage().contains("Unable to read schema"));
        }
    }

    @Test(dataProvider =  "AllCheckSchemaCompatibilityStrategy")
    public void testIsAutoUpdateSchema(SchemaCompatibilityStrategy schemaCompatibilityStrategy) throws Exception {
        final String tenant = PUBLIC_TENANT;
        final String topic = "test-consumer-compatibility";

        String namespace = "test-namespace-" + randomName(16);
        String fqtn = TopicName.get(
                TopicDomain.persistent.value(),
                tenant,
                namespace,
                topic
        ).toString();

        NamespaceName namespaceName = NamespaceName.get(tenant, namespace);

        admin.namespaces().createNamespace(
                tenant + "/" + namespace,
                Sets.newHashSet(CLUSTER_NAME)
        );

        assertEquals(admin.namespaces().getSchemaCompatibilityStrategy(namespaceName.toString()),
                SchemaCompatibilityStrategy.FULL);
        
        admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy);
        admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo());

        admin.namespaces().setIsAllowAutoUpdateSchema(namespaceName.toString(), false);
        ProducerBuilder<Schemas.PersonTwo> producerThreeBuilder = pulsarClient
                .newProducer(Schema.AVRO(SchemaDefinition.<Schemas.PersonTwo>builder().withAlwaysAllowNull
                        (false).withSupportSchemaVersioning(true).
                        withPojo(Schemas.PersonTwo.class).build()))
                .topic(fqtn);
        try {
            producerThreeBuilder.create();
        } catch (Exception e) {
            Assert.assertTrue(e.getMessage().contains("Schema not found and schema auto updating is disabled."));
        }

        admin.namespaces().setIsAllowAutoUpdateSchema(namespaceName.toString(), true);
        ConsumerBuilder<Schemas.PersonTwo> comsumerBuilder = pulsarClient.newConsumer(Schema.AVRO(
                SchemaDefinition.<Schemas.PersonTwo>builder().withAlwaysAllowNull
                        (false).withSupportSchemaVersioning(true).
                        withPojo(Schemas.PersonTwo.class).build()))
                .subscriptionName("test")
                .topic(fqtn);

        Producer<Schemas.PersonTwo> producer = producerThreeBuilder.create();
        Consumer<Schemas.PersonTwo> consumerTwo = comsumerBuilder.subscribe();

        producer.send(new Schemas.PersonTwo(2, "Lucy"));
        Message<Schemas.PersonTwo> message = consumerTwo.receive();

        Schemas.PersonTwo personTwo = message.getValue();
        consumerTwo.acknowledge(message);

        assertEquals(personTwo.id, 2);
        assertEquals(personTwo.name, "Lucy");

        producer.close();
        consumerTwo.close();

        admin.namespaces().setIsAllowAutoUpdateSchema(namespaceName.toString(), false);

        producer = producerThreeBuilder.create();
        consumerTwo = comsumerBuilder.subscribe();

        producer.send(new Schemas.PersonTwo(2, "Lucy"));
        message = consumerTwo.receive();

        personTwo = message.getValue();
        consumerTwo.acknowledge(message);

        assertEquals(personTwo.id, 2);
        assertEquals(personTwo.name, "Lucy");

        consumerTwo.close();
        producer.close();
    }

    @Test(dataProvider = "AllCheckSchemaCompatibilityStrategy")
    public void testProducerSendWithOldSchemaAndConsumerCanRead(SchemaCompatibilityStrategy schemaCompatibilityStrategy) throws Exception {
        final String tenant = PUBLIC_TENANT;
        final String topic = "test-consumer-compatibility";
        String namespace = "test-namespace-" + randomName(16);
        String fqtn = TopicName.get(
                TopicDomain.persistent.value(),
                tenant,
                namespace,
                topic
        ).toString();

        NamespaceName namespaceName = NamespaceName.get(tenant, namespace);

        admin.namespaces().createNamespace(
                tenant + "/" + namespace,
                Sets.newHashSet(CLUSTER_NAME)
        );

        admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy);
        admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo());
        admin.schemas().createSchema(fqtn, Schema.AVRO(SchemaDefinition.builder()
                .withAlwaysAllowNull(false).withPojo(Schemas.PersonTwo.class).build()).getSchemaInfo());

        Producer<Schemas.PersonOne> producerOne = pulsarClient
                .newProducer(Schema.AVRO(Schemas.PersonOne.class))
                .topic(fqtn)
                .create();


        Schemas.PersonOne personOne = new Schemas.PersonOne(10);

        Consumer<Schemas.PersonOne> consumerOne = pulsarClient.newConsumer(Schema.AVRO(
                SchemaDefinition.<Schemas.PersonOne>builder().withAlwaysAllowNull
                        (false).withSupportSchemaVersioning(true).
                        withPojo(Schemas.PersonOne.class).build()))
                .subscriptionName("test")
                .topic(fqtn)
                .subscribe();

        producerOne.send(personOne);
        Message<Schemas.PersonOne> message = consumerOne.receive();
        personOne = message.getValue();

        assertEquals(10, personOne.id);

        consumerOne.close();
        producerOne.close();

    }

    @Test(dataProvider =  "CanReadLastSchemaCompatibilityStrategy")
    public void testConsumerWithNotCompatibilitySchema(SchemaCompatibilityStrategy schemaCompatibilityStrategy) throws Exception {
        final String tenant = PUBLIC_TENANT;
        final String topic = "test-consumer-compatibility";

        String namespace = "test-namespace-" + randomName(16);
        String fqtn = TopicName.get(
                TopicDomain.persistent.value(),
                tenant,
                namespace,
                topic
        ).toString();

        NamespaceName namespaceName = NamespaceName.get(tenant, namespace);

        admin.namespaces().createNamespace(
                tenant + "/" + namespace,
                Sets.newHashSet(CLUSTER_NAME)
        );

        admin.namespaces().setSchemaCompatibilityStrategy(namespaceName.toString(), schemaCompatibilityStrategy);
        admin.schemas().createSchema(fqtn, Schema.AVRO(Schemas.PersonOne.class).getSchemaInfo());
        admin.schemas().createSchema(fqtn, Schema.AVRO(SchemaDefinition.builder()
                .withAlwaysAllowNull(false).withPojo(Schemas.PersonTwo.class).build()).getSchemaInfo());

        try {
            pulsarClient.newConsumer(Schema.AVRO(
                    SchemaDefinition.<Schemas.PersonFour>builder().withAlwaysAllowNull
                            (false).withSupportSchemaVersioning(true).
                            withPojo(Schemas.PersonFour.class).build()))
                    .subscriptionName("test")
                    .topic(fqtn)
                    .subscribe();
        } catch (Exception e) {
            Assert.assertTrue(e.getMessage().contains("Unable to read schema"));
        }

    }
    public static String randomName(int numChars) {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < numChars; i++) {
            sb.append((char) (ThreadLocalRandom.current().nextInt(26) + 'a'));
        }
        return sb.toString();
    }
}
