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

import java.io.BufferedWriter;
import java.util.Iterator;

/**
 * Generates Kafka MessageData classes.
 */
public final class JsonConverterGenerator implements MessageClassGenerator {
    private final static String SUFFIX = "JsonConverter";
    private final String packageName;
    private final StructRegistry structRegistry;
    private final HeaderGenerator headerGenerator;
    private final CodeBuffer buffer;

    JsonConverterGenerator(String packageName) {
        this.packageName = packageName;
        this.structRegistry = new StructRegistry();
        this.headerGenerator = new HeaderGenerator(packageName);
        this.buffer = new CodeBuffer();
    }

    @Override
    public String outputName(MessageSpec spec) {
        return spec.dataClassName() + SUFFIX;
    }

    @Override
    public void generateAndWrite(MessageSpec message, BufferedWriter writer)
            throws Exception {
        structRegistry.register(message);
        headerGenerator.addStaticImport(String.format("%s.%s.*",
            packageName, message.dataClassName()));
        buffer.printf("public class %s {%n",
            MessageGenerator.capitalizeFirst(outputName(message)));
        buffer.incrementIndent();
        generateConverters(message.dataClassName(), message.struct(),
            message.validVersions());
        for (Iterator<StructRegistry.StructInfo> iter = structRegistry.structs();
                iter.hasNext(); ) {
            StructRegistry.StructInfo info = iter.next();
            buffer.printf("%n");
            buffer.printf("public static class %s {%n",
                MessageGenerator.capitalizeFirst(info.spec().name() + SUFFIX));
            buffer.incrementIndent();
            generateConverters(MessageGenerator.capitalizeFirst(info.spec().name()),
                info.spec(), info.parentVersions());
            buffer.decrementIndent();
            buffer.printf("}%n");
        }
        buffer.decrementIndent();
        buffer.printf("}%n");
        headerGenerator.generate();
        headerGenerator.buffer().write(writer);
        buffer.write(writer);
    }

    private void generateConverters(String name,
                                    StructSpec spec,
                                    Versions parentVersions) {
        generateRead(name, spec, parentVersions);
        generateWrite(name, spec, parentVersions);
        generateOverloadWrite(name);
    }

    private void generateRead(String className,
                              StructSpec struct,
                              Versions parentVersions) {
        headerGenerator.addImport(MessageGenerator.JSON_NODE_CLASS);
        buffer.printf("public static %s read(JsonNode _node, short _version) {%n",
            className);
        buffer.incrementIndent();
        buffer.printf("%s _object = new %s();%n", className, className);
        VersionConditional.forVersions(struct.versions(), parentVersions).
            allowMembershipCheckAlwaysFalse(false).
            ifNotMember(__ -> {
                headerGenerator.addImport(MessageGenerator.UNSUPPORTED_VERSION_EXCEPTION_CLASS);
                buffer.printf("throw new UnsupportedVersionException(\"Can't read " +
                    "version \" + _version + \" of %s\");%n", className);
            }).
            generate(buffer);
        Versions curVersions = parentVersions.intersect(struct.versions());
        for (FieldSpec field : struct.fields()) {
            String sourceVariable = String.format("_%sNode", field.camelCaseName());
            buffer.printf("JsonNode %s = _node.get(\"%s\");%n",
                sourceVariable,
                field.camelCaseName());
            buffer.printf("if (%s == null) {%n", sourceVariable);
            buffer.incrementIndent();
            Versions mandatoryVersions = field.versions().subtract(field.taggedVersions());
            VersionConditional.forVersions(mandatoryVersions, curVersions).
                ifMember(__ -> {
                    buffer.printf("throw new RuntimeException(\"%s: unable to locate " +
                            "field \'%s\', which is mandatory in version \" + _version);%n",
                        className, field.camelCaseName());
                }).
                ifNotMember(__ -> {
                    buffer.printf("_object.%s = %s;%n", field.camelCaseName(),
                        field.fieldDefault(headerGenerator, structRegistry));
                }).
                generate(buffer);
            buffer.decrementIndent();
            buffer.printf("} else {%n");
            buffer.incrementIndent();
            VersionConditional.forVersions(struct.versions(), curVersions).
                ifMember(presentVersions -> {
                    generateTargetFromJson(new Target(field,
                            sourceVariable,
                            className,
                        input -> String.format("_object.%s = %s", field.camelCaseName(), input)),
                        curVersions);
                }).ifNotMember(__ -> {
                    buffer.printf("throw new RuntimeException(\"%s: field \'%s\' is not " +
                        "supported in version \" + _version);%n",
                        className, field.camelCaseName());
                }).generate(buffer);
            buffer.decrementIndent();
            buffer.printf("}%n");
        }
        buffer.printf("return _object;%n");
        buffer.decrementIndent();
        buffer.printf("}%n");
    }

    private void generateTargetFromJson(Target target, Versions curVersions) {
        if (target.field().type() instanceof FieldType.BoolFieldType) {
            buffer.printf("if (!%s.isBoolean()) {%n", target.sourceVariable());
            buffer.incrementIndent();
            buffer.printf("throw new RuntimeException(\"%s expected Boolean type, " +
                "but got \" + _node.getNodeType());%n", target.humanReadableName());
            buffer.decrementIndent();
            buffer.printf("}%n");
            buffer.printf("%s;%n", target.assignmentStatement(
                target.sourceVariable() + ".asBoolean()"));
        } else if (target.field().type() instanceof FieldType.Int8FieldType) {
            headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("MessageUtil.jsonNodeToByte(%s, \"%s\")",
                    target.sourceVariable(), target.humanReadableName())));
        } else if (target.field().type() instanceof FieldType.Int16FieldType) {
            headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("MessageUtil.jsonNodeToShort(%s, \"%s\")",
                    target.sourceVariable(), target.humanReadableName())));
        } else if (target.field().type() instanceof FieldType.Uint16FieldType) {
            headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("MessageUtil.jsonNodeToUnsignedShort(%s, \"%s\")",
                    target.sourceVariable(), target.humanReadableName())));
        } else if (target.field().type() instanceof FieldType.Uint32FieldType) {
            headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                 String.format("MessageUtil.jsonNodeToUnsignedInt(%s, \"%s\")",
                     target.sourceVariable(), target.humanReadableName())));
        } else if (target.field().type() instanceof FieldType.Int32FieldType) {
            headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("MessageUtil.jsonNodeToInt(%s, \"%s\")",
                    target.sourceVariable(), target.humanReadableName())));
        } else if (target.field().type() instanceof FieldType.Int64FieldType) {
            headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("MessageUtil.jsonNodeToLong(%s, \"%s\")",
                    target.sourceVariable(), target.humanReadableName())));
        } else if (target.field().type() instanceof FieldType.UUIDFieldType) {
            buffer.printf("if (!%s.isTextual()) {%n", target.sourceVariable());
            buffer.incrementIndent();
            buffer.printf("throw new RuntimeException(\"%s expected a JSON string " +
                "type, but got \" + _node.getNodeType());%n", target.humanReadableName());
            buffer.decrementIndent();
            buffer.printf("}%n");
            headerGenerator.addImport(MessageGenerator.UUID_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(String.format(
                "Uuid.fromString(%s.asText())", target.sourceVariable())));
        } else if (target.field().type() instanceof FieldType.Float64FieldType) {
            headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("MessageUtil.jsonNodeToDouble(%s, \"%s\")",
                    target.sourceVariable(), target.humanReadableName())));
        } else {
            // Handle the variable length types.  All of them are potentially
            // nullable, so handle that here.
            IsNullConditional.forName(target.sourceVariable()).
                nullableVersions(target.field().nullableVersions()).
                possibleVersions(curVersions).
                conditionalGenerator((name, negated) ->
                    String.format("%s%s.isNull()", negated ? "!" : "", name)).
                ifNull(() -> {
                    buffer.printf("%s;%n", target.assignmentStatement("null"));
                }).
                ifShouldNotBeNull(() -> {
                    generateVariableLengthTargetFromJson(target, curVersions);
                }).
                generate(buffer);
        }
    }

    private void generateVariableLengthTargetFromJson(Target target, Versions curVersions) {
        if (target.field().type().isString()) {
            buffer.printf("if (!%s.isTextual()) {%n", target.sourceVariable());
            buffer.incrementIndent();
            buffer.printf("throw new RuntimeException(\"%s expected a string " +
                "type, but got \" + _node.getNodeType());%n", target.humanReadableName());
            buffer.decrementIndent();
            buffer.printf("}%n");
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("%s.asText()", target.sourceVariable())));
        } else if (target.field().type().isBytes()) {
            headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
            if (target.field().zeroCopy()) {
                headerGenerator.addImport(MessageGenerator.BYTE_BUFFER_CLASS);
                buffer.printf("%s;%n", target.assignmentStatement(
                    String.format("ByteBuffer.wrap(MessageUtil.jsonNodeToBinary(%s, \"%s\"))",
                        target.sourceVariable(), target.humanReadableName())));
            } else {
                buffer.printf("%s;%n", target.assignmentStatement(
                    String.format("MessageUtil.jsonNodeToBinary(%s, \"%s\")",
                        target.sourceVariable(), target.humanReadableName())));
            }
        } else if (target.field().type().isRecords()) {
            headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
            headerGenerator.addImport(MessageGenerator.BYTE_BUFFER_CLASS);
            headerGenerator.addImport(MessageGenerator.MEMORY_RECORDS_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("MemoryRecords.readableRecords(ByteBuffer.wrap(MessageUtil.jsonNodeToBinary(%s, \"%s\")))",
                    target.sourceVariable(), target.humanReadableName())));
        } else if (target.field().type().isArray()) {
            buffer.printf("if (!%s.isArray()) {%n", target.sourceVariable());
            buffer.incrementIndent();
            buffer.printf("throw new RuntimeException(\"%s expected a JSON " +
                "array, but got \" + _node.getNodeType());%n", target.humanReadableName());
            buffer.decrementIndent();
            buffer.printf("}%n");
            String type = target.field().concreteJavaType(headerGenerator, structRegistry);
            buffer.printf("%s _collection = new %s(%s.size());%n", type, type, target.sourceVariable());
            buffer.printf("%s;%n", target.assignmentStatement("_collection"));
            headerGenerator.addImport(MessageGenerator.JSON_NODE_CLASS);
            buffer.printf("for (JsonNode _element : %s) {%n", target.sourceVariable());
            buffer.incrementIndent();
            generateTargetFromJson(target.arrayElementTarget(
                input -> String.format("_collection.add(%s)", input)),
                curVersions);
            buffer.decrementIndent();
            buffer.printf("}%n");
        } else if (target.field().type().isStruct()) {
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("%s%s.read(%s, _version)",
                target.field().type().toString(), SUFFIX, target.sourceVariable())));
        } else {
            throw new RuntimeException("Unexpected type " + target.field().type());
        }
    }

    private void generateOverloadWrite(String className) {
        buffer.printf("public static JsonNode write(%s _object, short _version) {%n",
                className);
        buffer.incrementIndent();
        buffer.printf("return write(_object, _version, true);%n");
        buffer.decrementIndent();
        buffer.printf("}%n");
    }

    private void generateWrite(String className,
                               StructSpec struct,
                               Versions parentVersions) {
        headerGenerator.addImport(MessageGenerator.JSON_NODE_CLASS);
        buffer.printf("public static JsonNode write(%s _object, short _version, boolean _serializeRecords) {%n",
            className);
        buffer.incrementIndent();
        VersionConditional.forVersions(struct.versions(), parentVersions).
            allowMembershipCheckAlwaysFalse(false).
            ifNotMember(__ -> {
                headerGenerator.addImport(MessageGenerator.UNSUPPORTED_VERSION_EXCEPTION_CLASS);
                buffer.printf("throw new UnsupportedVersionException(\"Can't write " +
                    "version \" + _version + \" of %s\");%n", className);
            }).
            generate(buffer);
        Versions curVersions = parentVersions.intersect(struct.versions());
        headerGenerator.addImport(MessageGenerator.OBJECT_NODE_CLASS);
        headerGenerator.addImport(MessageGenerator.JSON_NODE_FACTORY_CLASS);
        buffer.printf("ObjectNode _node = new ObjectNode(JsonNodeFactory.instance);%n");
        for (FieldSpec field : struct.fields()) {
            Target target = new Target(field,
                String.format("_object.%s", field.camelCaseName()),
                field.camelCaseName(),
                input -> String.format("_node.set(\"%s\", %s)", field.camelCaseName(), input));
            VersionConditional cond = VersionConditional.forVersions(field.versions(), curVersions).
                ifMember(presentVersions -> {
                    VersionConditional.forVersions(field.taggedVersions(), presentVersions).
                        ifMember(presentAndTaggedVersions -> {
                            field.generateNonDefaultValueCheck(headerGenerator,
                                structRegistry, buffer, "_object.", field.nullableVersions());
                            buffer.incrementIndent();
                            if (field.defaultString().equals("null")) {
                                // If the default was null, and we already checked that this field was not
                                // the default, we can omit further null checks.
                                generateTargetToJson(target.nonNullableCopy(), presentAndTaggedVersions);
                            } else {
                                generateTargetToJson(target, presentAndTaggedVersions);
                            }
                            buffer.decrementIndent();
                            buffer.printf("}%n");
                        }).
                        ifNotMember(presentAndNotTaggedVersions -> {
                            generateTargetToJson(target, presentAndNotTaggedVersions);
                        }).
                        generate(buffer);
                });
            if (!field.ignorable()) {
                cond.ifNotMember(__ -> {
                    field.generateNonIgnorableFieldCheck(headerGenerator,
                        structRegistry, "_object.", buffer);
                });
            }
            cond.generate(buffer);
        }
        buffer.printf("return _node;%n");
        buffer.decrementIndent();
        buffer.printf("}%n");
    }

    private void generateTargetToJson(Target target, Versions versions) {
        if (target.field().type() instanceof FieldType.BoolFieldType) {
            headerGenerator.addImport(MessageGenerator.BOOLEAN_NODE_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("BooleanNode.valueOf(%s)", target.sourceVariable())));
        } else if ((target.field().type() instanceof FieldType.Int8FieldType) ||
                (target.field().type() instanceof FieldType.Int16FieldType)) {
            headerGenerator.addImport(MessageGenerator.SHORT_NODE_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("new ShortNode(%s)", target.sourceVariable())));
        } else if ((target.field().type() instanceof FieldType.Int32FieldType) ||
                (target.field().type() instanceof FieldType.Uint16FieldType)) {
            headerGenerator.addImport(MessageGenerator.INT_NODE_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("new IntNode(%s)", target.sourceVariable())));
        } else if (target.field().type() instanceof FieldType.Int64FieldType ||
                (target.field().type() instanceof FieldType.Uint32FieldType)) {
            headerGenerator.addImport(MessageGenerator.LONG_NODE_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("new LongNode(%s)", target.sourceVariable())));
        } else if (target.field().type() instanceof FieldType.UUIDFieldType) {
            headerGenerator.addImport(MessageGenerator.TEXT_NODE_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("new TextNode(%s.toString())", target.sourceVariable())));
        } else if (target.field().type() instanceof FieldType.Float64FieldType) {
            headerGenerator.addImport(MessageGenerator.DOUBLE_NODE_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("new DoubleNode(%s)", target.sourceVariable())));
        } else {
            // Handle the variable length types.  All of them are potentially
            // nullable, so handle that here.
            IsNullConditional.forName(target.sourceVariable()).
                nullableVersions(target.field().nullableVersions()).
                possibleVersions(versions).
                conditionalGenerator((name, negated) ->
                    String.format("%s %s= null", name, negated ? "!" : "=")).
                ifNull(() -> {
                    headerGenerator.addImport(MessageGenerator.NULL_NODE_CLASS);
                    buffer.printf("%s;%n", target.assignmentStatement("NullNode.instance"));
                }).
                ifShouldNotBeNull(() -> {
                    generateVariableLengthTargetToJson(target, versions);
                }).
                generate(buffer);
        }
    }

    private void generateVariableLengthTargetToJson(Target target, Versions versions) {
        if (target.field().type().isString()) {
            headerGenerator.addImport(MessageGenerator.TEXT_NODE_CLASS);
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("new TextNode(%s)", target.sourceVariable())));
        } else if (target.field().type().isBytes()) {
            headerGenerator.addImport(MessageGenerator.BINARY_NODE_CLASS);
            if (target.field().zeroCopy()) {
                headerGenerator.addImport(MessageGenerator.MESSAGE_UTIL_CLASS);
                buffer.printf("%s;%n", target.assignmentStatement(
                    String.format("new BinaryNode(MessageUtil.byteBufferToArray(%s))",
                        target.sourceVariable())));
            } else {
                headerGenerator.addImport(MessageGenerator.ARRAYS_CLASS);
                buffer.printf("%s;%n", target.assignmentStatement(
                    String.format("new BinaryNode(Arrays.copyOf(%s, %s.length))",
                            target.sourceVariable(), target.sourceVariable())));
            }
        } else if (target.field().type().isRecords()) {
            headerGenerator.addImport(MessageGenerator.BINARY_NODE_CLASS);
            headerGenerator.addImport(MessageGenerator.INT_NODE_CLASS);
            // KIP-673: When logging requests/responses, we do not serialize the record, instead we
            // output its sizeInBytes, because outputting the bytes is not very useful and can be
            // quite expensive. Otherwise, we will serialize the record.
            buffer.printf("if (_serializeRecords) {%n");
            buffer.incrementIndent();
            buffer.printf("%s;%n", target.assignmentStatement("new BinaryNode(new byte[]{})"));
            buffer.decrementIndent();
            buffer.printf("} else {%n");
            buffer.incrementIndent();
            buffer.printf("_node.set(\"%sSizeInBytes\", new IntNode(%s.sizeInBytes()));%n",
                    target.field().camelCaseName(),
                    target.sourceVariable());
            buffer.decrementIndent();
            buffer.printf("}%n");
        } else if (target.field().type().isArray()) {
            headerGenerator.addImport(MessageGenerator.ARRAY_NODE_CLASS);
            headerGenerator.addImport(MessageGenerator.JSON_NODE_FACTORY_CLASS);
            FieldType.ArrayType arrayType = (FieldType.ArrayType) target.field().type();
            FieldType elementType = arrayType.elementType();
            String arrayInstanceName = String.format("_%sArray",
                target.field().camelCaseName());
            buffer.printf("ArrayNode %s = new ArrayNode(JsonNodeFactory.instance);%n",
                arrayInstanceName);
            buffer.printf("for (%s _element : %s) {%n",
                elementType.getBoxedJavaType(headerGenerator), target.sourceVariable());
            buffer.incrementIndent();
            generateTargetToJson(target.arrayElementTarget(
                input -> String.format("%s.add(%s)", arrayInstanceName, input)),
                versions);
            buffer.decrementIndent();
            buffer.printf("}%n");
            buffer.printf("%s;%n", target.assignmentStatement(arrayInstanceName));
        } else if (target.field().type().isStruct()) {
            buffer.printf("%s;%n", target.assignmentStatement(
                String.format("%sJsonConverter.write(%s, _version, _serializeRecords)",
                    target.field().type().toString(), target.sourceVariable())));
        } else {
            throw new RuntimeException("unknown type " + target.field().type());
        }
    }

}