package io.apicurio.registry.maven;
   
   import io.apicurio.registry.content.ContentHandle;
   import io.apicurio.registry.content.TypedContent;
   import io.apicurio.registry.rest.client.RegistryClient;
   import io.apicurio.registry.rest.client.models.ArtifactReference;
   import io.apicurio.registry.types.ContentTypes;
   import org.apache.avro.AvroTypeException;
   import org.apache.avro.Schema;
  import org.apache.avro.SchemaParseException;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  import java.io.File;
  import java.io.FileNotFoundException;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.List;
  import java.util.Map;
  import java.util.Objects;
  import java.util.Set;
  import java.util.concurrent.ExecutionException;
  import java.util.stream.Collectors;
  
  public class AvroDirectoryParser extends AbstractDirectoryParser<Schema> {
  
      private static final String AVRO_SCHEMA_EXTENSION = ".avsc";
      private static final Logger log = LoggerFactory.getLogger(AvroDirectoryParser.class);
  
      public AvroDirectoryParser(RegistryClient client) {
          super(client);
      }
  
      @Override
      public ParsedDirectoryWrapper<Schema> parse(File rootSchemaFile) {
          return parseDirectory(rootSchemaFile.getParentFile(), rootSchemaFile);
      }
  
      @Override
      public List<ArtifactReference> handleSchemaReferences(RegisterArtifact rootArtifact, Schema rootSchema,
              Map<String, TypedContent> fileContents)
              throws FileNotFoundException, ExecutionException, InterruptedException {
  
          Set<ArtifactReference> references = new HashSet<>();
  
          // Iterate through all the fields of the schema
          for (Schema.Field field : rootSchema.getFields()) {
              List<ArtifactReference> nestedArtifactReferences = new ArrayList<>();
              if (field.schema().getType() == Schema.Type.RECORD) { // If the field is a sub-schema, recursively
                                                                    // check for nested sub-schemas and register
                                                                    // all of them
  
                  RegisterArtifact nestedSchema = buildFromRoot(rootArtifact, field.schema().getFullName());
  
                  if (field.schema().hasFields()) {
                      nestedArtifactReferences = handleSchemaReferences(nestedSchema, field.schema(),
                              fileContents);
                  }
  
                  references.add(registerNestedSchema(field.schema().getFullName(), nestedArtifactReferences,
                          nestedSchema, fileContents.get(field.schema().getFullName()).getContent().content()));
              } else if (field.schema().getType() == Schema.Type.ENUM) { // If the nested schema is an enum,
                                                                         // just register
  
                  RegisterArtifact nestedSchema = buildFromRoot(rootArtifact, field.schema().getFullName());
                  references.add(registerNestedSchema(field.schema().getFullName(), nestedArtifactReferences,
                          nestedSchema, fileContents.get(field.schema().getFullName()).getContent().content()));
              } else if (isArrayWithSubschemaElement(field)) { // If the nested schema is an array and the
                                                               // element is a sub-schema, handle it
  
                  Schema elementSchema = field.schema().getElementType();
  
                  RegisterArtifact nestedSchema = buildFromRoot(rootArtifact, elementSchema.getFullName());
  
                  if (elementSchema.hasFields()) {
                      nestedArtifactReferences = handleSchemaReferences(nestedSchema, elementSchema,
                              fileContents);
                  }
  
                  references.add(registerNestedSchema(elementSchema.getFullName(), nestedArtifactReferences,
                          nestedSchema, fileContents.get(elementSchema.getFullName()).getContent().content()));
              }
          }
          return new ArrayList<>(references);
      }
  
      private ParsedDirectoryWrapper<Schema> parseDirectory(File directory, File rootSchema) {
          Set<File> typesToAdd = Arrays
                  .stream(Objects.requireNonNull(
                          directory.listFiles((dir, name) -> name.endsWith(AVRO_SCHEMA_EXTENSION))))
                  .filter(file -> !file.getName().equals(rootSchema.getName())).collect(Collectors.toSet());
  
          Map<String, Schema> processed = new HashMap<>();
          Map<String, TypedContent> schemaContents = new HashMap<>();
  
          Schema.Parser rootSchemaParser = new Schema.Parser();
          Schema.Parser partialParser = new Schema.Parser();
 
         while (processed.size() != typesToAdd.size()) {
             boolean fileParsed = false;
             for (File typeToAdd : typesToAdd) {
                 if (typeToAdd.getName().equals(rootSchema.getName())) {
                     continue;
                 }
                 try {
                     final ContentHandle schemaContent = readSchemaContent(typeToAdd);
                     final String contentType = ContentTypes.APPLICATION_JSON;
                     final TypedContent typedSchemaContent = TypedContent.create(schemaContent, contentType);
                     final Schema schema = partialParser.parse(schemaContent.content());
                     processed.put(schema.getFullName(), schema);
                     schemaContents.put(schema.getFullName(), typedSchemaContent);
                     fileParsed = true;
                 } catch (SchemaParseException | AvroTypeException ex) {
                     log.warn(
                             "Error processing Avro schema with name {}. This usually means that the references are not ready yet to parse it",
                             typeToAdd.getName());
                 }
             }
             partialParser = new Schema.Parser();
             partialParser.addTypes(processed);
 
             // If no schema has been processed during this iteration, that means there is an error in the
             // configuration, throw exception.
             if (!fileParsed) {
                 throw new IllegalStateException(
                         "Error found in the directory structure. Check that all required files are present.");
             }
 
         }
 
         rootSchemaParser.addTypes(processed);
 
         return new AvroSchemaWrapper(rootSchemaParser.parse(readSchemaContent(rootSchema).content()),
                 schemaContents);
     }
 
     private boolean isArrayWithSubschemaElement(Schema.Field field) {
         return field.schema().getType() == Schema.Type.ARRAY
                 && field.schema().getElementType().getType() == Schema.Type.RECORD;
     }
 
     public static class AvroSchemaWrapper implements ParsedDirectoryWrapper<Schema> {
         final Schema schema;
         final Map<String, TypedContent> fileContents; // Original file contents from the file system.
 
         public AvroSchemaWrapper(Schema schema, Map<String, TypedContent> fileContents) {
             this.schema = schema;
             this.fileContents = fileContents;
         }
 
         @Override
         public Schema getSchema() {
             return schema;
         }
 
         @Override
         public Map<String, TypedContent> getSchemaContents() {
             return fileContents;
         }
     }
 }