AvroSchema

/**
  * Semi-automatic derivation done similarly to how Kittens does it
  *
  * @see https://github.com/typelevel/kittens/blob/master/core/src/main/scala/cats/derived/package.scala
  * */
final object generic {
  final object semi {
    def deriveSchema[A](implicit ev: MkAvroSchema[A]): HasAvroSchema[A] = ev
  }

  trait MkAvroSchemaType[A] extends HasAvroSchemaType[A] {
    override def schemaType(name: String): Schema = {
      val assembler =
        SchemaBuilder.builder().record(name).namespace("").fields()
      fields.getOrElse(List.empty).foldLeft(assembler) {
        case (assembler, (fieldName, fieldType)) =>
          assembler.name(fieldName).`type`(fieldType).noDefault()
      }
      assembler.endRecord()
    }
    def fields: Option[List[(String, Schema)]]
  }
  object MkAvroSchemaType extends MkAvroSchemaTypeDerivation {
    def apply[A](implicit schemaType: MkAvroSchemaType[A]) = schemaType

    def withSchema[A](f: String => Schema): MkAvroSchemaType[A] =
      new MkAvroSchemaType[A] {
        override def schemaType(name: String) = f(name)
        override def fields: Option[List[(String, Schema)]] = None
      }

    def withSchemaType[A](schema: Schema.Type): MkAvroSchemaType[A] =
      withSchema[A](_ => Schema.create(schema))
    def withFields[A](value: List[(String, Schema)]) = new MkAvroSchemaType[A] {
      override def fields: Option[List[(String, Schema)]] = Some(value)
    }
  }

  trait MkAvroSchemaTypeDerivation {
    implicit def optionType[A](
        implicit ev: HasAvroSchemaType[A] OrElse MkAvroSchemaType[A]
    ) =
      MkAvroSchemaType.withSchema[Option[A]](name =>
        Schema.createUnion(
          ev.unify.schemaType(name),
          Schema.create(Schema.Type.NULL)
        )
      )

    implicit def seqType[A](
        implicit ev: HasAvroSchemaType[A] OrElse MkAvroSchemaType[A]
    ): MkAvroSchemaType[Seq[A]] =
      MkAvroSchemaType.withSchema[Seq[A]](name =>
        Schema.createArray(ev.unify.schemaType(name))
      )

    implicit def mapType[A](
        implicit ev: HasAvroSchemaType[A] OrElse MkAvroSchemaType[A]
    ): MkAvroSchemaType[Map[String, A]] =
      MkAvroSchemaType.withSchema[Map[String, A]](name =>
        Schema.createMap(ev.unify.schemaType(name))
      )

    implicit val stringType: MkAvroSchemaType[String] =
      MkAvroSchemaType.withSchemaType(Schema.Type.STRING)
    implicit val doubleType: MkAvroSchemaType[Double] =
      MkAvroSchemaType.withSchemaType(Schema.Type.DOUBLE)
    implicit val longType: MkAvroSchemaType[Long] =
      MkAvroSchemaType.withSchemaType(Schema.Type.LONG)
    implicit val intType: MkAvroSchemaType[Int] =
      MkAvroSchemaType.withSchemaType(Schema.Type.INT)

    implicit val emptyProductDerivedSchemaType: MkAvroSchemaType[HNil] =
      MkAvroSchemaType.withFields(List.empty)
    implicit def productDerivedSchemaType[K <: Symbol, V, T <: HList](
        implicit key: Witness.Aux[K],
        typeV: HasAvroSchemaType[V] OrElse MkAvroSchemaType[V],
        typeT: MkAvroSchemaType[T]
    ): MkAvroSchemaType[FieldType[K, V] :: T] = {
      val fieldName = key.value.name
      val fieldAvroSchema = typeV.unify

      val fieldSchema = typeV.fold(
        _.schemaType(fieldName),
        _.fields match {
          case Some(fields) =>
            MkAvroSchemaType.withFields(fields).schemaType(fieldName)
          case None => fieldAvroSchema.schemaType(fieldName)
        }
      )

      MkAvroSchemaType.withFields(
        (fieldName, fieldSchema) :: typeT.fields.getOrElse(List.empty)
      )
    }

    implicit def genericDerivedSchemaType[A <: Product, R <: HList](
        implicit
        repr: LabelledGeneric.Aux[A, R],
        s: Lazy[MkAvroSchemaType[R]],
        auto: IsAutomaticallyDerived[A]
    ): MkAvroSchemaType[A] = {
      MkAvroSchemaType.withSchema(name => s.value.schemaType(name))
    }
  }

  trait MkAvroSchema[A] extends HasAvroSchema[A] with MkAvroSchemaType[A]
  object MkAvroSchema extends MkAvroSchemaDerivation {

    def withFields[A](value: List[(String, Schema)]) = new MkAvroSchema[A] {
      override def fields: Option[List[(String, Schema)]] = Some(value)
    }

    def withSchema[A](f: String => Schema): MkAvroSchema[A] =
      new MkAvroSchema[A] {
        override def schemaType(name: String) = f(name)

        override def fields: Option[List[(String, Schema)]] = None
      }
  }

  trait MkAvroSchemaDerivation {
    implicit val emptyProductDerivedSchemaType: MkAvroSchema[HNil] =
      MkAvroSchema.withFields(List.empty)
    implicit def productDerivedSchemaType[K <: Symbol, V, T <: HList](
        implicit key: Witness.Aux[K],
        typeV: HasAvroSchemaType[V] OrElse MkAvroSchemaType[V],
        typeT: MkAvroSchema[T]
    ): MkAvroSchema[FieldType[K, V] :: T] = {
      val fieldName = key.value.name
      val fieldAvroSchema = typeV.unify

      val fieldSchema = typeV.fold(
        _.schemaType(fieldName),
        _.fields match {
          case Some(fields) =>
            MkAvroSchemaType.withFields(fields).schemaType(fieldName)
          case None => fieldAvroSchema.schemaType(fieldName)
        }
      )

      MkAvroSchema.withFields(
        (fieldName, fieldSchema) :: typeT.fields.getOrElse(List.empty)
      )
    }

    implicit def genericDerivedSchemaType[A, R <: HList](
        implicit
        repr: LabelledGeneric.Aux[A, R],
        s: Lazy[MkAvroSchema[R]]
    ): MkAvroSchema[A] = {
      MkAvroSchema.withSchema(name => s.value.schemaType(name))
    }
  }

  /**
    * The goal is to automatically derive MkAvroSchemaType instances for tuples, not case classes.
    *
    * The idea is to mimick what most libraries do (circe, etc) and avoid black magic when it comes to deriving typeclasses.
    *
    * If a specific case class is supposed to have an AvroSchema instance, then one should be defined for the given type.
    *
    * Only tuples of size <= 5 are supported to encourage extracting case classes
    * */
  trait IsAutomaticallyDerived[A]
  object IsAutomaticallyDerived {
    def instance[A] = new IsAutomaticallyDerived[A] {}
    implicit def tuple1[A1] = instance[Tuple1[A1]]
    implicit def tuple2[A1, A2] = instance[Tuple2[A1, A2]]
    implicit def tuple3[A1, A2, A3] = instance[Tuple3[A1, A2, A3]]
    implicit def tuple4[A1, A2, A3, A4] = instance[Tuple4[A1, A2, A3, A4]]
    implicit def tuple5[A1, A2, A3, A4, A5] =
      instance[Tuple5[A1, A2, A3, A4, A5]]

  }
}

import org.apache.avro.Schema

trait HasAvroSchemaType[A] {
  def schemaType(name: String): Schema
}

object HasAvroSchemaType {
  def apply[A](implicit ev: HasAvroSchemaType[A]): HasAvroSchemaType[A] = ev

  def withSchema[A](f: String => Schema): HasAvroSchemaType[A] =
    new HasAvroSchemaType[A] {
      override def schemaType(name: String) = f(name)
    }

  def withSchemaType[A](schema: Schema.Type): HasAvroSchemaType[A] =
    withSchema[A](_ => Schema.create(schema))
}

trait HasAvroSchema[A] extends HasAvroSchemaType[A]

object HasAvroSchema {
  def apply[A](implicit enc: HasAvroSchema[A]) = enc
}