package com.thatdot.quine.graph

import java.nio.ByteBuffer
import java.nio.charset.StandardCharsets.UTF_8
import java.util.UUID
import java.util.concurrent.atomic.{AtomicLong, AtomicReference}

import scala.reflect.{ClassTag, classTag}
import scala.util.{Failure, Success, Try}

import memeid.{UUID => UUID4s}
import org.apache.commons.codec.digest.MurmurHash2

import com.thatdot.quine.model.{NamespacedIdProvider, QuineGraphLocation, QuineId, QuineIdProvider, QuineValue}

/** This provider is special: it is a no-op provider in the sense that none of the
  * conversions do any work. [[com.thatdot.quine.model.QuineId]] is the ID type.
  */
case object IdentityIdProvider extends QuineIdProvider {
  type CustomIdType = QuineId
  val customIdTag: ClassTag[QuineId] = classTag[QuineId]

  private val counter: AtomicReference[BigInt] = new AtomicReference(0)

  def newCustomId(): QuineId = QuineId(counter.getAndUpdate((b: BigInt) => b + 1).toByteArray)

  def hashedCustomId(bytes: Array[Byte]): QuineId = QuineId(bytes)

  def customIdToString(qid: QuineId): String = qid.toInternalString
  def customIdFromString(s: String): Try[QuineId] = Try(QuineId.fromInternalString(s))

  def customIdToBytes(qid: QuineId): Array[Byte] = qid.array
  def customIdFromBytes(qidBytes: Array[Byte]): Try[QuineId] = Success(QuineId(qidBytes))

  override def valueToQid(value: QuineValue): Option[QuineId] = value match {
    case QuineValue.Id(qid) => Some(qid)
    case QuineValue.Bytes(arr) => Some(QuineId(arr))
    case _ => None
  }
  override def qidToValue(qid: QuineId): QuineValue = QuineValue.Bytes(qid.array)
}

/** This provider uses UUID-like (see caveats below) values as IDs
  *
  * Caveats:
  * - [[QuineUUIDProvider.newCustomId]] provides valid UUIDv4 instances
  * - [[QuineUUIDProvider.customIdFromBytes]] allows using *any* 128 bits, not constrained to those that are valid UUIDs
  * - [[QuineUUIDProvider.hashedCustomId]] generates potentially invalid UUIDv3 instances (notably, the java API for
  *   UUIDv3 does not conform to the RFC specification for making hash-based UUIDs)
  * - [[QuineUUIDProvider.customIdFromString]] permits many non-UUID things as input that will succeed in generating a
  *   java.util.UUID. The java standard library's UUID implementation is very permissive in the UUID-like strings it
  *   will try to parse. For example: https://bugs.java.com/bugdatabase/view_bug.do?bug_id=8159339
  *
  * newCustomId UUIDs generated by this provider have 6 fixed bits and 122 pseudorandom bits. At 1,000,000 IDs per
  *   second, likely time to collision for newCustomId-generated IDs is 73069 years
  * customIdFromBytes generated by this provider have theoretically 128 entropic bits. At 1,000,000 IDs per second,
  *   second, likely time to collision for customIdFromBytes IDs is 584554 years
  * hashedCustomId generated by this provider have 6 fixed bits and 122 pseudorandom bits. At 1,000,000 IDs per
  *   second, likely time to collision for newCustomId-generated IDs is 73069 years
  */
case object QuineUUIDProvider extends QuineIdProvider {
  type CustomIdType = UUID
  val customIdTag: ClassTag[UUID] = classTag[UUID]

  def newCustomId(): UUID = UUID.randomUUID()

  def hashedCustomId(bytes: Array[Byte]): UUID =
    UUID.nameUUIDFromBytes(bytes)

  def customIdToString(typed: UUID): String = typed.toString
  def customIdFromString(s: String): Try[UUID] = Try(UUID.fromString(s))

  def customIdToBytes(uuid: UUID): Array[Byte] = ByteBuffer
    .allocate(16) // 128 bits
    .putLong(uuid.getMostSignificantBits)
    .putLong(uuid.getLeastSignificantBits)
    .array()
  def customIdFromBytes(qidBytes: Array[Byte]): Try[UUID] = Try {
    val bb = ByteBuffer.wrap(qidBytes)
    new UUID(bb.getLong(), bb.getLong())
  }
}

/** This provider uses sequential 64-bit integers as UUIDs
  *
  * @param initial the first integer to be assigned as a fresh ID
  *
  * @note the last-assigned ID is only stored at runtime, and not automatically restored at system startup. Use caution
  *       when rebooting a Quine instance with this provider as some IDs may be reused
  * @note there is no guarantee that a hashedCustomId-generated ID will not duplicate a newCustomId-generated ID, or
  *       visa versa.
  *
  * newCustomIds generated by this provider have 64 cyclically-generated bits. At 1,000,000 IDs per second, likely time
  *   to collision is 584554 years.
  */
final case class QuineIdLongProvider(initial: Long = 0L) extends QuineIdProvider {
  type CustomIdType = Long
  val customIdTag: ClassTag[Long] = classTag[Long]

  private val counter = new AtomicLong(initial)

  def newCustomId(): Long = counter.getAndIncrement()

  def hashedCustomId(bytes: Array[Byte]): Long =
    ByteBuffer.wrap(QuineIdProvider.hashToLength(bytes, 8)).getLong()

  def customIdToString(typed: Long): String = typed.toString
  def customIdFromString(s: String): Try[Long] = Try(s.toLong)

  def customIdToBytes(typed: Long): Array[Byte] = ByteBuffer.allocate(8).putLong(typed).array()
  def customIdFromBytes(bytes: Array[Byte]): Try[Long] = Try {
    val bb = ByteBuffer.wrap(bytes)
    require(bb.capacity() == 8)
    bb.getLong()
  }

  override def valueToQid(value: QuineValue): Option[QuineId] = value match {
    case QuineValue.Id(qid) => Some(qid)
    case QuineValue.Integer(lng) => Some(customIdToQid(lng))
    case _ => None
  }
  override def qidToValue(qid: QuineId): QuineValue =
    customIdFromQid(qid) match {
      case Success(lng) => QuineValue.Integer(lng)
      case Failure(_) => QuineValue.Id(qid)
    }
}

/** This provider uses random 53-bit integers as UUIDs
  *
  * @note this uses Longs to represent Javascript/JSON-safe integers, ie, Double integers. Not all 2^64 Longs
  *       are representable in Javascript without loss of precision. There are 2^53 integers representable as Doubles
  *       without loss of precision, and so that is the range of this function
  *
  * newCustomIds generated by this provider have 53 pseudorandom bits. At 1,000,000 IDs per second, likely time
  *   to collision is 95 seconds. (For reference, if all 64 bits of the Long were used, the likely time to collision
  *   would be 71.5 minutes)
  */
case object QuineIdRandomLongProvider extends QuineIdProvider {
  type CustomIdType = Long
  val customIdTag: ClassTag[Long] = classTag[Long]

  /* Map a [[Long]] into the range of integers Javascript can represent exactly,
   * aka `[-(2^53 - 1), 2^53 - 1]`. If the input [[Long]] is uniformly
   * distributed, so will the output (on the aforementioned range).
   *
   * Straying outside of this range silently causes confusing behaviour when
   * seeing IDs from the browser: the numbers will be rounded to the nearest
   * representable integer
   */
  final private def makeJsSafeLong(seed: Long): Long = seed % (1L << 53)

  def newCustomId(): Long = makeJsSafeLong(scala.util.Random.nextLong())

  def hashedCustomId(bytes: Array[Byte]): Long =
    makeJsSafeLong(ByteBuffer.wrap(QuineIdProvider.hashToLength(bytes, 8)).getLong())

  def customIdToString(typed: Long): String = typed.toString
  def customIdFromString(s: String): Try[Long] = Try(s.toLong)

  def customIdToBytes(typed: Long): Array[Byte] = ByteBuffer.allocate(8).putLong(typed).array()
  def customIdFromBytes(bytes: Array[Byte]): Try[Long] = Try {
    val bb = ByteBuffer.wrap(bytes)
    require(bb.capacity() == 8)
    bb.getLong()
  }

  override def valueToQid(value: QuineValue): Option[QuineId] = value match {
    case QuineValue.Id(qid) => Some(qid)
    case QuineValue.Integer(lng) => Some(customIdToQid(lng))
    case _ => None
  }
  override def qidToValue(qid: QuineId): QuineValue =
    customIdFromQid(qid) match {
      case Success(lng) => QuineValue.Integer(lng)
      case Failure(_) => QuineValue.Bytes(qid.array)
    }
}

object WithPartitioning {
  final case class Id[IdT](partition: Int, underlyingId: IdT)

  private val PartitionedUnderlying = "([0-9a-fA-F]{8})/(.*)".r
  def apply(underlying: QuineIdProvider): NamespacedIdProvider = underlying match {
    case alreadyNamespaced: NamespacedIdProvider => alreadyNamespaced
    case notNamespaced => new WithPartitioning(notNamespaced)
  }
}

/** A wrapper making a given IdProvider namespace-aware by mapping the provided namespaces into byte-space partitions.
  * Given a QuineId, the partition (0 through MAX_INT) may be recovered, but not the original namespace.
  *
  * newCustomIds generated by this provider have 32 bits more entropy than that of the underlying ID provider
  * hashedCustomIds generated by this provider have (approximately) the same amount of entropy as that of the underlying
  *   ID provider, up to a maximum of 256 bits
  */
final case class WithPartitioning private (underlying: QuineIdProvider) extends NamespacedIdProvider {
  type CustomIdType = WithPartitioning.Id[underlying.CustomIdType]

  val customIdTag: ClassTag[WithPartitioning.Id[underlying.CustomIdType]] =
    classTag[WithPartitioning.Id[underlying.CustomIdType]]

  def partitionFor(namespace: String): Int =
    MurmurHash2.hash32(namespace)

  def randomPartition: Int = scala.util.Random.nextInt()

  def newCustomIdInNamespace(namespace: String): Try[WithPartitioning.Id[underlying.CustomIdType]] =
    Success(WithPartitioning.Id(partitionFor(namespace), underlying.newCustomId()))

  def hashedCustomIdInNamespace(
    namespace: String,
    bytes: Array[Byte]
  ): Try[WithPartitioning.Id[underlying.CustomIdType]] =
    Success(WithPartitioning.Id(partitionFor(namespace), underlying.hashedCustomId(bytes)))

  def newCustomId(): WithPartitioning.Id[underlying.CustomIdType] =
    WithPartitioning.Id(randomPartition, underlying.newCustomId())

  def hashedCustomId(bytes: Array[Byte]): WithPartitioning.Id[underlying.CustomIdType] = {
    val hashed = QuineIdProvider.hashToLength(bytes, 32)
    // 32 bits for a partition leaves 224 bits of entropy. This is still a LOT.
    val bb = ByteBuffer.wrap(hashed)
    val partition = bb.getInt
    val bytesForUnderlying = bb.remainingBytes
    WithPartitioning.Id(partition, underlying.hashedCustomId(bytesForUnderlying))
  }

  def customIdToString(id: WithPartitioning.Id[underlying.CustomIdType]): String =
    "%08X".format(id.partition) + "/" + underlying.customIdToString(id.underlyingId)

  def customIdFromString(str: String): Try[WithPartitioning.Id[underlying.CustomIdType]] = {
    import WithPartitioning.PartitionedUnderlying
    str match {
      case PartitionedUnderlying(partitionHex, underlyingStr) =>
        for {
          partition <- Try(Integer.parseUnsignedInt(partitionHex, 16)).recoverWith { case err =>
            Failure(new IllegalArgumentException("Unable to decode partition marker portion of partitioned ID", err))
          }
          underlyingId <- underlying
            .customIdFromString(underlyingStr)
            .recoverWith { case err =>
              Failure(new IllegalArgumentException("Unable to decode underlying ID portion of partitioned ID", err))
            }
        } yield WithPartitioning.Id(partition, underlyingId)
      case doesntMatch =>
        Failure(
          new IllegalArgumentException(
            s"Provided ID string ($doesntMatch) was not in the required format (`partition`/`customIdString`) where " +
            "`partition` is two hex characters, and `customIdString` is a valid string representation for " +
            "the underlying id provider ($underlying)"
          )
        )
    }
  }

  def customIdToBytes(id: WithPartitioning.Id[underlying.CustomIdType]): Array[Byte] =
    ByteBuffer.allocate(4).putInt(id.partition).array ++ underlying.customIdToBytes(id.underlyingId)

  /** Extract an ID from its Quine-internal raw byte array format
    *
    * @note should be the inverse of [[customIdToBytes]]
    * @param bytes raw byte array representation of ID
    * @return node ID
    */
  def customIdFromBytes(bytes: Array[Byte]): Try[WithPartitioning.Id[underlying.CustomIdType]] = {
    val bb = ByteBuffer.wrap(bytes)
    for {
      partition <- Try(bb.getInt)
      tailBytes = bb.remainingBytes
      underlyingId <- underlying.customIdFromBytes(tailBytes)
    } yield WithPartitioning.Id(partition, underlyingId)
  }

  override def nodeLocation(qid: QuineId): QuineGraphLocation = customIdFromBytes(qid.array) match {
    case Failure(exception) =>
      logger.warn(
        s"""Couldn't parse out a partitioned Quine Id from provided id ${qid.pretty(this)}.
            |Falling back to the underlying node location algorithm""".stripMargin.replace('\n', ' '),
        exception
      )
      underlying.nodeLocation(qid)
    case Success(id) =>
      QuineGraphLocation(Some(id.partition), underlying.nodeLocation(qid).shardIdx)
  }
}

// Must use the same list of namespaces (in the same order) on all hosts!.
object NameSpacedUuidProvider {
  // match groups are greedy by default, so the left match group will match all but the last `--`. This allows
  // namespaces to contain "--".
  private val NamespaceAndId = "(.*)--(.*)".r
}

/** This provider uses String--UUID-like pairs as QuineIds
  *
  * This provider is subject to all of the same caveats around UUID usage as [[QuineUUIDProvider]].
  * newCustomIds and hashedCustomIds generated by this provider have the same entropy as those generated by
  * the [[QuineUUIDProvider]]
  *
  * @param localNamespaces Namespaces used in this cluster
  *                        INV: this list must contain the same elements in the same order on all clustered hosts, and
  *                        have the same number of elements as the number of hosts in the cluster
  * @param thisNamespaceIdx The index into [[localNamespaces]] corresponding to the current host's namespace
  *                         INV: this must be different on each clustered host
  */
final case class NameSpacedUuidProvider(
  localNamespaces: List[String],
  thisNamespaceIdx: Int
) extends NamespacedIdProvider {
  type CustomIdType = (String, UUID)
  val customIdTag: ClassTag[(String, UUID)] = classTag[(String, UUID)]

  logger.warn("NamespacedUuidProvider is deprecated - use a specific-version UUID provider with partitioning instead")

  require(thisNamespaceIdx <= localNamespaces.size - 1 && thisNamespaceIdx >= 0)
  private val localNamespace = localNamespaces(thisNamespaceIdx)
  private val namespaceCount = localNamespaces.size

  def newCustomId(): (String, UUID) = newCustomIdInNamespace(localNamespace).get

  def hashedCustomId(bytes: Array[Byte]): (String, UUID) = {
    val chosenNamespace = localNamespaces(
      Math.floorMod(ByteBuffer.wrap(QuineIdProvider.hashToLength(bytes, 4)).getInt, namespaceCount)
    )
    hashedCustomIdInNamespace(chosenNamespace, bytes).get
  }

  def customIdToString(typed: (String, UUID)): String = s"${typed._1}--${typed._2}"
  def customIdFromString(s: String): Try[(String, UUID)] =
    s match {
      case NameSpacedUuidProvider.NamespaceAndId(namespace, id) =>
        Try(UUID.fromString(id)).map(uuid => namespace -> uuid)
      case other => Failure(new IllegalArgumentException(s"Invalid namespace--UUID pair provided: ${other}"))
    }

  def customIdToBytes(typed: (String, UUID)): Array[Byte] = {
    val stringBytes = typed._1.getBytes(UTF_8)
    ByteBuffer
      .allocate(16 + stringBytes.length)
      .putLong(typed._2.getMostSignificantBits)
      .putLong(typed._2.getLeastSignificantBits)
      .put(stringBytes)
      .array()
  }
  def customIdFromBytes(bytes: Array[Byte]): Try[(String, UUID)] = Try {
    val bb = ByteBuffer.wrap(bytes)
    val uuid = new UUID(bb.getLong(), bb.getLong())
    new String(bb.remainingBytes, UTF_8) -> uuid
  }

  // Goal: Consistent choice of HostIdx directly based on the `localNamespace`. Random distribution among that host's shards.
  override def nodeLocation(qid: QuineId): QuineGraphLocation = {
    val custom = customIdFromQid(qid).get
    val hostIdx = Try(
      Math.abs(
        // This is the core definition of which host is responsible for a specific QID:
        custom._1
          .split("_")(1)
          .toInt // Distribute to the host that is literally specified (modded later if host count > this Int).
//    hashToLength(qid.array, 1).head.toInt           // Distribute evenly among shards
//    hashToLength(qid.array.drop(16), 1).head.toInt  // IDs in the same namespace go to the same shard
      )
    ).getOrElse { // In case that fails:
//      println(s"localNamespace is not of the expected type!  $localNamespace")
      Math.abs(ByteBuffer.wrap(QuineIdProvider.hashToLength(custom._1.getBytes(UTF_8), 4)).getInt())
    }

    val localShardIdx = Math.abs(
      ByteBuffer
        .wrap(
          QuineIdProvider.hashToLength(
            ByteBuffer
              .allocate(16)
              .putLong(custom._2.getMostSignificantBits)
              .putLong(custom._2.getLeastSignificantBits)
              .array(),
            4
          )
        )
        .getInt()
    )

    QuineGraphLocation(Some(hostIdx), localShardIdx)
  }

  def newCustomIdInNamespace(namespace: String): Try[(String, UUID)] = if (localNamespaces.contains(namespace)) {
    Success(namespace -> UUID.randomUUID())
  } else Failure(new IllegalArgumentException(s"Cannot create an ID in nonexistent namespace $namespace"))

  def hashedCustomIdInNamespace(namespace: String, bytes: Array[Byte]): Try[(String, UUID)] = if (
    localNamespaces.contains(namespace)
  ) {
    val bb = ByteBuffer.wrap(QuineIdProvider.hashToLength(bytes, 16))
    Try(namespace -> new UUID(bb.getLong(), bb.getLong()))
  } else Failure(new IllegalArgumentException(s"Cannot create an ID in nonexistent namespace $namespace"))

}

final case class WrongUuidVersion[V <: UUID4s: ClassTag](u: UUID4s)
    extends IllegalArgumentException(
      s"Got a UUID $u with V${u.version}, expected " + classTag[V].runtimeClass.getSimpleName
    )

/** Common supertype of UUID IDProviders that use only a single version of UUID (represented as a single subtype of
  * memeid's UUID type, aliased here as UUID4s
  */
sealed abstract class SingleVersionUuidProvider[UuidV <: UUID4s: ClassTag] extends QuineIdProvider {

  /** Checks whether the provided UUID is of the type managed by this IdProvider
    * @param u the UUID instance to check
    * @return true iff the instance is of the version represented by `UuidV`
    */
  def uuidVersionMatches(u: UUID4s): Boolean = {
    import memeid4s.UUID.RichUUID
    u.is[UuidV]
  }

  final val customIdTag: ClassTag[UUID] = classTag[UUID]
  final type CustomIdType = UUID

  def customIdToString(typed: UUID): String = typed.toString
  def customIdFromString(str: String): Try[UUID] =
    Try(UUID4s.fromString(str)).flatMap(asJavaUuid)

  def customIdToBytes(typed: UUID): Array[Byte] = uuidToBytes(UUID4s.fromUUID(typed))
  def customIdFromBytes(bytes: Array[Byte]): Try[UUID] =
    if (bytes.length != 128 / 8)
      Failure(new IllegalArgumentException(s"Byte array had ${bytes.length * 8} bits -- a UUID must have 128"))
    else
      Try {
        val bb = ByteBuffer.wrap(bytes)
        UUID4s.from(bb.getLong(), bb.getLong())
      }.flatMap(asJavaUuid)

  /** Turn a UUID losslessly into a freshly-allocated 16-byte (128-bit) array */
  final protected[this] def uuidToBytes(u: UUID4s): Array[Byte] =
    ByteBuffer.allocate(16).putLong(u.getMostSignificantBits).putLong(u.getLeastSignificantBits).array()

  /** Given a UUID of knowable version, convert it to a java UUID only if the version matches [[UuidV]]
    * @param u the uuid to convert
    * @return Success(uuid) where uuid is a valid java UUID matching the provided version. Failure otherwise
    */
  final protected[this] def asJavaUuid(u: UUID4s): Try[UUID] =
    Either.cond(uuidVersionMatches(u), u.asJava, WrongUuidVersion[UuidV](u)).toTry
}

/** This provider uses (strict) UUIDv5s
  *
  * This does NOT suffer from any of the caveats of [[QuineUUIDProvider]] or [[NameSpacedUuidProvider]] with respect to
  * correctness of IDs -- [[Uuid5Provider]] will always produce valid UUIDv5s. However, this provider will be slightly
  * slower than those when generating fresh IDs via newCustomId(), as this will do a SHA1 hash as part of that process.
  *
  * UUID5s use 128 bits, 6 of which are fixed, leaving 122 bits of entropy. At 1,000,000 IDs per second, likely time to
  * collision is 73069 years
  *
  * @param defaultNamespace the namespace in which this host will create new IDs
  * @see [[Uuid4Provider]] for a provider that uses v4 UUIDs, and hashes using a Quine-controlled protocol
  * @see [[Uuid3Provider]] for a provider that uses RFC-compliant v3 UUIDs (ie hashes via MD5)
  */
final case class Uuid5Provider(defaultNamespace: UUID = UUID4s.NIL.asJava())
    extends SingleVersionUuidProvider[UUID4s.V5] {

  private val defaultNamespace4s = UUID4s.fromUUID(defaultNamespace)

  /** Generate a fresh UUIDv5 with `defaultNamespace` as a namespace and a V4 (random) UUID's underlying bytes as a
    * value
    */
  def newCustomId(): UUID = UUID4s.V5.from(defaultNamespace4s, UUID4s.V4.random(), uuidToBytes).asJava()

  /** Generates a UUIDv5 with [[defaultNamespace]] as a namespace and [[bytes]] as a name
    */
  def hashedCustomId(bytes: Array[Byte]): UUID =
    UUID4s.V5.from[Array[Byte]](defaultNamespace4s, bytes, identity).asJava()
}

/** This provider uses (strict) UUIDv3s
  *
  * This does NOT suffer from any of the caveats of [[QuineUUIDProvider]] or [[NameSpacedUuidProvider]] with respect to
  * correctness of IDs -- [[Uuid3Provider]] will always produce valid UUIDv3s. However, this provider will be slightly
  * slower than those when generating fresh IDs via newCustomId(), as this will do an MD5 hash as part of that process.
  *
  * UUID3s use 128 bits, 6 of which are fixed, leaving 122 bits of entropy. At 1,000,000 IDs per second, likely time to
  * collision is 73069 years
  *
  * @param defaultNamespace the namespace in which this host will create new IDs
  * @see [[Uuid4Provider]] for a provider that uses v4 UUIDs, and hashes using a Quine-controlled protocol
  * @see [[Uuid5Provider]] for a provider that uses RFC-compliant v5 UUIDs (ie hashes via SHA)
  */
final case class Uuid3Provider(defaultNamespace: UUID = UUID4s.NIL.asJava())
    extends SingleVersionUuidProvider[UUID4s.V3] {

  private val defaultNamespace4s = UUID4s.fromUUID(defaultNamespace)

  /** Generate a fresh UUIDv3 with `defaultNamespace` as a namespace and a V4 (random) UUID's underlying bytes as a
    * value
    */
  def newCustomId(): UUID = UUID4s.V3.from(defaultNamespace4s, UUID4s.V4.random(), uuidToBytes).asJava()

  /** Generates a UUIDv3 with [[defaultNamespace]] as a namespace and [[bytes]] as a name
    */
  def hashedCustomId(bytes: Array[Byte]): UUID =
    UUID4s.V3.from[Array[Byte]](defaultNamespace4s, bytes, identity).asJava()
}

/** This provider uses standards-adherent UUIDv4s (given a loose definition of "pseudo-random", similar to an sqUUID)
  *
  * Specifically, UUIDs generated with this provider's hashedCustomId will be (more or less) evenly distributed through
  * the ID space, though they may be computationally distinguishable from randomly generated. This is accomplished by
  * "baking in" our hashing algorithm to the process of "pseudo-randomly" generating bytes for a hashedCustomId, rather
  * than using a cryptographic hashing algorithm.
  *
  * UUID4s use 128 bits, 6 of which are fixed, leaving 122 bits of entropy. At 1,000,000 IDs per second, likely time to
  * collision is 73069 years
  *
  * @note hashedCustomIds generated by this provider should not be considered opaque with respect to their input data
  * @note this provider should be the most performant of the specific-version UUID providers
  * @see [[Uuid5Provider]] for a provider that uses RFC-compliant v5 UUIDs (ie hashes via SHA)
  * @see [[Uuid3Provider]] for a provider that uses RFC-compliant v3 UUIDs (ie hashes via MD5)
  */
case object Uuid4Provider extends SingleVersionUuidProvider[UUID4s.V4] {

  def newCustomId(): UUID = UUID4s.V4.random().asJava()

  def hashedCustomId(bytes: Array[Byte]): UUID = {
    val underlyingBytes = ByteBuffer.wrap(QuineIdProvider.hashToLength(bytes, 16)) // 16 bytes = 128 bits
    UUID4s.V4.from(underlyingBytes.getLong, underlyingBytes.getLong).asJava()
  }
}
