IdempotentVersionedWritable.scala

import org.apache.commons.io.IOUtils
import uk.ac.wellcome.storage.streaming.Codec
import scala.util.{Success, Try}

trait IdempotentVersionedWritable[Id, V, T]
  extends Writable[Version[Id, V], T]
    with Readable[Version[Id, V], T] {

  implicit val codec: Codec[T]

  abstract override def put(id: Version[Id, V])(t: T): WriteEither = {
    super.put(id)(t) match {
      case Right(identifiedT) => Right(identifiedT)
      case Left(originalError: VersionAlreadyExistsError) => for {
          putStream <- codec.toStream(t)
          retrieved <- get(id) match {
            case Right(result) => Right(result)
            case Left(readError) => Left(IdempotentCheckWriteError(readError))
          }
          getStream <- codec.toStream(retrieved.identifiedT)
          tryEquals <- Try(IOUtils.contentEquals(putStream, getStream)) match {
            case Success(equality) => Right(equality)
            case Failure(err) => Left(IOExceptionWriteError(err))
          }
          result <- if(tryEquals) {
            Right(Identified(id, t))
          } else {
            Left(originalError)
          }
      } yield result
      case Left(err) => Left(err)
    }
  }
}