/*
 * Copyright 2021-2023 Aklivity Inc
 *
 * Licensed under the Aklivity Community License (the "License"); you may not use
 * this file except in compliance with the License.  You may obtain a copy of the
 * License at
 *
 *   https://www.aklivity.io/aklivity-community-license/
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OF ANY KIND, either express or implied.  See the License for the
 * specific language governing permissions and limitations under the License.
 */
package io.aklivity.zilla.runtime.binding.grpc.kafka.internal.config;


import jakarta.json.Json;
import jakarta.json.JsonObject;
import jakarta.json.JsonObjectBuilder;
import jakarta.json.bind.adapter.JsonbAdapter;

import io.aklivity.zilla.runtime.binding.grpc.kafka.config.GrpcKafkaCorrelationConfig;
import io.aklivity.zilla.runtime.binding.grpc.kafka.config.GrpcKafkaIdempotencyConfig;
import io.aklivity.zilla.runtime.binding.grpc.kafka.config.GrpcKafkaOptionsConfig;
import io.aklivity.zilla.runtime.binding.grpc.kafka.config.GrpcKafkaReliabilityConfig;
import io.aklivity.zilla.runtime.binding.grpc.kafka.internal.GrpcKafkaBinding;
import io.aklivity.zilla.runtime.binding.grpc.kafka.internal.types.String16FW;
import io.aklivity.zilla.runtime.binding.grpc.kafka.internal.types.String8FW;
import io.aklivity.zilla.runtime.engine.config.OptionsConfig;
import io.aklivity.zilla.runtime.engine.config.OptionsConfigAdapterSpi;

public final class GrpcKafkaOptionsConfigAdapter implements OptionsConfigAdapterSpi, JsonbAdapter<OptionsConfig, JsonObject>
{
    private static final String RELIABILITY_NAME = "reliability";
    private static final String RELIABILITY_FIELD_NAME = "field";
    private static final String RELIABILITY_METADATA_NAME = "metadata";
    private static final String IDEMPOTENCY_NAME = "idempotency";
    private static final String IDEMPOTENCY_METADATA_NAME = "metadata";
    private static final String CORRELATION_NAME = "correlation";
    private static final String CORRELATION_HEADERS_NAME = "headers";
    private static final String CORRELATION_HEADERS_CORRELATION_ID_NAME = "correlation-id";
    private static final String CORRELATION_HEADERS_SERVICE_NAME = "service";
    private static final String CORRELATION_HEADERS_METHOD_NAME = "method";
    private static final String CORRELATION_HEADERS_REPLY_TO_NAME = "reply-to";

    private static final String16FW CORRELATION_HEADERS_CORRELATION_ID_DEFAULT = new String16FW("zilla:correlation-id");
    private static final String16FW CORRELATION_HEADERS_SERVICE_DEFAULT = new String16FW("zilla:service");
    private static final String16FW CORRELATION_HEADERS_METHOD_DEFAULT = new String16FW("zilla:method");
    private static final String16FW CORRELATION_HEADERS_REPLY_TO_DEFAULT = new String16FW("zilla:reply-to");

    private static final int FIELD_DEFAULT = 32767;
    private static final String8FW RELIABILITY_METADATA_DEFAULT = new String8FW("last-message-id");
    private static final String8FW IDEMPOTENCY_METADATA_DEFAULT = new String8FW("idempotency-key");
    private static final GrpcKafkaCorrelationConfig CORRELATION_DEFAULT =
        new GrpcKafkaCorrelationConfig(CORRELATION_HEADERS_CORRELATION_ID_DEFAULT,
            CORRELATION_HEADERS_SERVICE_DEFAULT, CORRELATION_HEADERS_METHOD_DEFAULT, CORRELATION_HEADERS_REPLY_TO_DEFAULT);

    private static final GrpcKafkaReliabilityConfig RELIABILITY_DEFAULT =
        new GrpcKafkaReliabilityConfig(FIELD_DEFAULT, RELIABILITY_METADATA_DEFAULT);

    private static final GrpcKafkaIdempotencyConfig IDEMPOTENCY_DEFAULT =
        new GrpcKafkaIdempotencyConfig(IDEMPOTENCY_METADATA_DEFAULT);

    public static final GrpcKafkaOptionsConfig DEFAULT =
        new GrpcKafkaOptionsConfig(RELIABILITY_DEFAULT, IDEMPOTENCY_DEFAULT, CORRELATION_DEFAULT);

    @Override
    public Kind kind()
    {
        return Kind.BINDING;
    }

    @Override
    public String type()
    {
        return GrpcKafkaBinding.NAME;
    }

    @Override
    public JsonObject adaptToJson(
        OptionsConfig options)
    {
        GrpcKafkaOptionsConfig grpcKafkaOptions = (GrpcKafkaOptionsConfig) options;

        JsonObjectBuilder object = Json.createObjectBuilder();

        GrpcKafkaReliabilityConfig reliability = grpcKafkaOptions.reliability;
        if (reliability != null &&
            !RELIABILITY_DEFAULT.equals(reliability))
        {
            JsonObjectBuilder newReliability = Json.createObjectBuilder();

            if (FIELD_DEFAULT != reliability.field)
            {
                newReliability.add(RELIABILITY_FIELD_NAME, reliability.field);
            }

            if (!RELIABILITY_METADATA_DEFAULT.equals(reliability.metadata))
            {
                newReliability.add(RELIABILITY_METADATA_NAME, reliability.metadata.asString());
            }

            object.add(RELIABILITY_NAME, newReliability);
        }

        GrpcKafkaIdempotencyConfig idempotency = grpcKafkaOptions.idempotency;
        if (idempotency != null &&
            !IDEMPOTENCY_DEFAULT.equals(idempotency))
        {
            JsonObjectBuilder newIdempotency = Json.createObjectBuilder();

            if (!IDEMPOTENCY_METADATA_DEFAULT.equals(idempotency.metadata))
            {
                newIdempotency.add(IDEMPOTENCY_METADATA_NAME, idempotency.metadata.asString());
            }

            object.add(IDEMPOTENCY_NAME, newIdempotency);
        }

        GrpcKafkaCorrelationConfig correlation = grpcKafkaOptions.correlation;
        if (correlation != null &&
            !CORRELATION_DEFAULT.equals(correlation))
        {
            JsonObjectBuilder newHeaders = Json.createObjectBuilder();

            if (!CORRELATION_HEADERS_SERVICE_DEFAULT.equals(correlation.service))
            {
                newHeaders.add(CORRELATION_HEADERS_SERVICE_NAME, correlation.service.asString());
            }

            if (!CORRELATION_HEADERS_METHOD_DEFAULT.equals(correlation.method))
            {
                newHeaders.add(CORRELATION_HEADERS_METHOD_NAME, correlation.method.asString());
            }

            if (!CORRELATION_HEADERS_CORRELATION_ID_DEFAULT.equals(correlation.correlationId))
            {
                newHeaders.add(CORRELATION_HEADERS_CORRELATION_ID_NAME, correlation.correlationId.asString());
            }

            if (!CORRELATION_HEADERS_REPLY_TO_DEFAULT.equals(correlation.replyTo))
            {
                newHeaders.add(CORRELATION_HEADERS_REPLY_TO_NAME, correlation.replyTo.asString());
            }

            JsonObjectBuilder newCorrelation = Json.createObjectBuilder();
            newCorrelation.add(CORRELATION_HEADERS_NAME, newHeaders);

            object.add(CORRELATION_NAME, newCorrelation);
        }

        return object.build();
    }

    @Override
    public OptionsConfig adaptFromJson(
        JsonObject object)
    {
        GrpcKafkaReliabilityConfig newReliability = RELIABILITY_DEFAULT;
        if (object.containsKey(RELIABILITY_NAME))
        {
            JsonObject reliability = object.getJsonObject(RELIABILITY_NAME);

            int newField = FIELD_DEFAULT;
            if (reliability.containsKey(RELIABILITY_FIELD_NAME))
            {
                int field = reliability.getInt(RELIABILITY_FIELD_NAME);
                if (field < 19000 || field > 19999)
                {
                    newField = field;
                }
            }

            String8FW newMetadata = RELIABILITY_METADATA_DEFAULT;
            if (reliability.containsKey(RELIABILITY_METADATA_NAME))
            {
                newMetadata = new String8FW(reliability.getString(RELIABILITY_METADATA_NAME));
            }

            newReliability = new GrpcKafkaReliabilityConfig(newField, newMetadata);
        }

        GrpcKafkaIdempotencyConfig newIdempotency = IDEMPOTENCY_DEFAULT;
        if (object.containsKey(IDEMPOTENCY_NAME))
        {
            JsonObject idempotency = object.getJsonObject(IDEMPOTENCY_NAME);

            String8FW newMetadata = IDEMPOTENCY_METADATA_DEFAULT;
            if (idempotency.containsKey(IDEMPOTENCY_METADATA_NAME))
            {
                newMetadata = new String8FW(idempotency.getString(IDEMPOTENCY_METADATA_NAME));
            }

            newIdempotency = new GrpcKafkaIdempotencyConfig(newMetadata);
        }

        GrpcKafkaCorrelationConfig newCorrelation = CORRELATION_DEFAULT;
        if (object.containsKey(CORRELATION_NAME))
        {
            JsonObject correlation = object.getJsonObject(CORRELATION_NAME);
            if (correlation.containsKey(CORRELATION_HEADERS_NAME))
            {
                JsonObject headers = correlation.getJsonObject(CORRELATION_HEADERS_NAME);

                String16FW newService = CORRELATION_HEADERS_SERVICE_DEFAULT;
                if (headers.containsKey(CORRELATION_HEADERS_SERVICE_NAME))
                {
                    newService = new String16FW(headers.getString(CORRELATION_HEADERS_SERVICE_NAME));
                }

                String16FW newMethod = CORRELATION_HEADERS_METHOD_DEFAULT;
                if (headers.containsKey(CORRELATION_HEADERS_METHOD_NAME))
                {
                    newMethod = new String16FW(headers.getString(CORRELATION_HEADERS_METHOD_NAME));
                }

                String16FW newCorrelationId = CORRELATION_HEADERS_CORRELATION_ID_DEFAULT;
                if (headers.containsKey(CORRELATION_HEADERS_CORRELATION_ID_NAME))
                {
                    newCorrelationId = new String16FW(headers.getString(CORRELATION_HEADERS_CORRELATION_ID_NAME));
                }

                String16FW newReplyTo = CORRELATION_HEADERS_REPLY_TO_DEFAULT;
                if (headers.containsKey(CORRELATION_HEADERS_REPLY_TO_NAME))
                {
                    newReplyTo = new String16FW(headers.getString(CORRELATION_HEADERS_REPLY_TO_NAME));
                }

                newCorrelation = new GrpcKafkaCorrelationConfig(newCorrelationId, newService, newMethod, newReplyTo);
            }
        }

        return new GrpcKafkaOptionsConfig(newReliability, newIdempotency, newCorrelation);
    }
}
