Skip to content

Commit

Permalink
feat: add JsonCanonicalizer and use in serialization
Browse files Browse the repository at this point in the history
  • Loading branch information
@scasplte2
scasplte2 committed Feb 13, 2025
1 parent bde0f68 commit 03f50d5
Show file tree
Hide file tree
Showing 8 changed files with 161 additions and 108 deletions.
49 changes: 19 additions & 30 deletions src/main/scala/io/constellationnetwork/metagraph_sdk/MetagraphCommonService.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,63 +7,52 @@ import org.tessellation.currency.dataApplication._
import org.tessellation.currency.dataApplication.dataApplication.DataApplicationBlock
import org.tessellation.security.signature.Signed

import io.constellationnetwork.metagraph_sdk.std.JsonBinaryCodec
import io.constellationnetwork.metagraph_sdk.std.JsonBinaryCodec.{simpleJsonDeserialization, simpleJsonSerialization}
import io.constellationnetwork.metagraph_sdk.std.JsonBinaryCodec._

import io.circe.{Decoder, Encoder}
import org.http4s.EntityDecoder
import org.http4s.circe.CirceEntityCodec.circeEntityDecoder

abstract class MetagraphCommonService[F[_], TX <: DataUpdate, PUB <: DataOnChainState, PRV <: DataCalculatedState](
implicit
txEncoder: Encoder[TX],
txDecoder: Decoder[TX],
prvEncoder: Encoder[PRV],
prvDecoder: Decoder[PRV],
val txBinCodec: JsonBinaryCodec[F, TX],
val pubBinCodec: JsonBinaryCodec[F, PUB],
val prvBinCodec: JsonBinaryCodec[F, PRV],
async: Async[F]
txEncoder: Encoder[TX],
pubEncoder: Encoder[PUB],
prvEncoder: Encoder[PRV],
txDecoder: Decoder[TX],
pubDecoder: Decoder[PUB],
prvDecoder: Decoder[PRV],
async: Async[F]
) {

val signedDataEntityDecoder: EntityDecoder[F, Signed[TX]] = circeEntityDecoder

implicit val dataBlockCodec: JsonBinaryCodec[F, Signed[DataApplicationBlock]] =
new JsonBinaryCodec[F, Signed[DataApplicationBlock]] {
implicit def dataEncoder: Encoder[DataUpdate] = txEncoder.contramap(_.asInstanceOf[TX])
implicit def dataUpdateEncoder: Encoder[DataUpdate] = txEncoder.contramap(_.asInstanceOf[TX])

implicit def dataDecoder: Decoder[DataUpdate] = txDecoder.widen
implicit def dataUpdateDecoder: Decoder[DataUpdate] = txDecoder.widen

override def serialize(obj: Signed[DataApplicationBlock]): F[Array[Byte]] =
simpleJsonSerialization(obj)

override def deserialize(bytes: Array[Byte]): F[Either[Throwable, Signed[DataApplicationBlock]]] =
simpleJsonDeserialization(bytes)
}
val signedDataEntityDecoder: EntityDecoder[F, Signed[TX]] = circeEntityDecoder

def serializeState(state: PUB): F[Array[Byte]] =
pubBinCodec.serialize(state)
state.toBinary

def deserializeState(bytes: Array[Byte]): F[Either[Throwable, PUB]] =
pubBinCodec.deserialize(bytes)
bytes.fromBinary[PUB]

def serializeCalculatedState(calculatedState: PRV): F[Array[Byte]] =
prvBinCodec.serialize(calculatedState)
calculatedState.toBinary

def deserializeCalculatedState(bytes: Array[Byte]): F[Either[Throwable, PRV]] =
prvBinCodec.deserialize(bytes)
bytes.fromBinary[PRV]

def serializeUpdate(update: TX): F[Array[Byte]] =
txBinCodec.serialize(update)
update.toBinary

def deserializeUpdate(bytes: Array[Byte]): F[Either[Throwable, TX]] =
txBinCodec.deserialize(bytes)
bytes.fromBinary[TX]

def serializeBlock(block: Signed[DataApplicationBlock]): F[Array[Byte]] =
dataBlockCodec.serialize(block)
block.toBinary

def deserializeBlock(bytes: Array[Byte]): F[Either[Throwable, Signed[DataApplicationBlock]]] =
dataBlockCodec.deserialize(bytes)
bytes.fromBinary[Signed[DataApplicationBlock]]

def dataEncoder: Encoder[TX] = txEncoder

Expand Down
83 changes: 44 additions & 39 deletions src/main/scala/io/constellationnetwork/metagraph_sdk/std/JsonBinaryCodec.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,62 +6,67 @@ import cats.effect.Sync
import cats.implicits.{toFlatMapOps, toFunctorOps}

import org.tessellation.currency.dataApplication.DataUpdate
import org.tessellation.currency.schema.currency.CurrencyIncrementalSnapshot
import org.tessellation.schema.ID.Id

import io.constellationnetwork.metagraph_sdk.std.JsonCanonicalizer.JsonPrinterEncodeOps

import io.circe.jawn.JawnParser
import io.circe.syntax.EncoderOps
import io.circe.{Decoder, Encoder, Printer}
import io.circe.{Decoder, Encoder}
import org.bouncycastle.util.encoders.Base64

trait JsonBinaryCodec[F[_], A] {
def serialize(content: A): F[Array[Byte]]
def deserialize(content: Array[Byte]): F[Either[Throwable, A]]
def serialize(content: A): F[Array[Byte]]
def deserialize(bytes: Array[Byte]): F[Either[Throwable, A]]
}

object JsonBinaryCodec {

private val printer = Printer(dropNullValues = true, indent = "", sortKeys = true)

def apply[F[_], A](implicit ev: JsonBinaryCodec[F, A]): JsonBinaryCodec[F, A] = ev

def simpleJsonSerialization[F[_]: Sync, A: Encoder](content: A): F[Array[Byte]] =
Sync[F].delay(content.asJson.printWith(printer).getBytes("UTF-8"))

def simpleJsonDeserialization[F[_]: Sync, A: Decoder](content: Array[Byte]): F[Either[Throwable, A]] =
Sync[F].delay(JawnParser(false).decodeByteArray[A](content))

def serializeDataUpdate[F[_]: Sync, U <: DataUpdate: Encoder](content: U): F[Array[Byte]] = for {
jsonBytes <- simpleJsonSerialization(content)
base64String <- Sync[F].delay(Base64.toBase64String(jsonBytes))
prefixedString = s"\u0019Constellation Signed Data:\n${base64String.length}\n$base64String"
} yield prefixedString.getBytes("UTF-8")
def fromBinary[F[_], A](bytes: Array[Byte])(implicit codec: JsonBinaryCodec[F, A]): F[Either[Throwable, A]] =
codec.deserialize(bytes)

def deserializeDataUpdate[F[_]: Sync, U <: DataUpdate: Decoder](
bytes: Array[Byte]
): F[Either[Throwable, DataUpdate]] = for {
base64String <- Sync[F].pure(new String(bytes, StandardCharsets.UTF_8).split("\n").drop(2).mkString)
jsonBytes <- Sync[F].delay(Base64.decode(base64String))
result <- simpleJsonDeserialization(jsonBytes)
} yield result
implicit def derive[F[_]: Sync, A: Encoder: Decoder]: JsonBinaryCodec[F, A] =
new JsonBinaryCodec[F, A] {

implicit def currencyIncrementalSnapshotCodec[F[_]: Sync]: JsonBinaryCodec[F, CurrencyIncrementalSnapshot] =
new JsonBinaryCodec[F, CurrencyIncrementalSnapshot] {
def serialize(content: A): F[Array[Byte]] = for {
json <- content.asCanonicalJson.map(_.noSpaces)
bytes <- Sync[F].delay(json.getBytes("UTF-8"))
} yield bytes

override def serialize(obj: CurrencyIncrementalSnapshot): F[Array[Byte]] =
simpleJsonSerialization(obj)
def deserialize(bytes: Array[Byte]): F[Either[Throwable, A]] = for {
str <- Sync[F].delay(new String(bytes, "UTF-8"))
result <- Sync[F].delay(JawnParser(false).decode[A](str))
} yield result
}

override def deserialize(bytes: Array[Byte]): F[Either[Throwable, CurrencyIncrementalSnapshot]] =
simpleJsonDeserialization(bytes)
implicit def deriveDataUpdate[F[_]: Sync, U <: DataUpdate: Encoder: Decoder]: JsonBinaryCodec[F, U] =
new JsonBinaryCodec[F, U] {

def serialize(content: U): F[Array[Byte]] = for {
json <- content.asCanonicalJson.map(_.noSpaces)
jsonBytes <- Sync[F].delay(json.getBytes("UTF-8"))
base64String <- Sync[F].delay(Base64.toBase64String(jsonBytes))
prefixedString = s"\u0019Constellation Signed Data:\n${base64String.length}\n$base64String"
result <- Sync[F].delay(prefixedString.getBytes("UTF-8"))
} yield result

def deserialize(bytes: Array[Byte]): F[Either[Throwable, U]] = for {
base64String <- Sync[F].pure(new String(bytes, StandardCharsets.UTF_8).split("\n").drop(2).mkString)
jsonBytes <- Sync[F].delay(Base64.decode(base64String))
str <- Sync[F].delay(new String(jsonBytes, "UTF-8"))
result <- Sync[F].delay(JawnParser(false).decode[U](str))
} yield result
}

implicit def idCodec[F[_]: Sync]: JsonBinaryCodec[F, Id] =
new JsonBinaryCodec[F, Id] {
implicit class JsonBinaryEncodeOps[F[_], A](val _v: A) extends AnyVal {

override def serialize(obj: Id): F[Array[Byte]] =
simpleJsonSerialization(obj)
def toBinary(implicit codec: JsonBinaryCodec[F, A]): F[Array[Byte]] =
codec.serialize(_v)
}

override def deserialize(bytes: Array[Byte]): F[Either[Throwable, Id]] =
simpleJsonDeserialization(bytes)
}
implicit class JsonBinaryDecodeOps[F[_]](val _v: Array[Byte]) extends AnyVal {

def fromBinary[A](implicit codec: JsonBinaryCodec[F, A]): F[Either[Throwable, A]] =
codec.deserialize(_v)
}
}
66 changes: 66 additions & 0 deletions src/main/scala/io/constellationnetwork/metagraph_sdk/std/JsonCanonicalizer.scala
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,66 @@
package io.constellationnetwork.metagraph_sdk.std

import cats.effect.Sync
import cats.syntax.functor._
import cats.syntax.traverse._
import cats.{Applicative, ApplicativeThrow}

import scala.math.Ordering.Implicits.seqOrdering

import io.circe.syntax.EncoderOps
import io.circe.{Encoder, Json, JsonNumber, JsonObject}

trait JsonCanonicalizer[F[_], A] {
def toJson(content: A): F[Json]
}

object JsonCanonicalizer {

def apply[F[_], A](implicit ev: JsonCanonicalizer[F, A]): JsonCanonicalizer[F, A] = ev

private def formatNumber[F[_]: ApplicativeThrow](n: JsonNumber): F[String] =
n.toBigDecimal match {
case Some(bd) =>
Applicative[F].pure {
val str = bd.underlying.toPlainString
if (bd.compare(BigDecimal(0)) == 0 && str.startsWith("-")) "-0"
else if (str.contains(".")) str
else str + ".0"
}
case None =>
ApplicativeThrow[F].raiseError(
new IllegalArgumentException(s"Invalid number: $n")
)
}

// RFC 8785 compliant formatting
private def canonicalizeJson[F[_]: ApplicativeThrow](json: Json): F[Json] =
json.fold(
jsonNull = Applicative[F].pure(Json.Null),
jsonBoolean = b => Applicative[F].pure(Json.fromBoolean(b)),
jsonNumber = n => formatNumber[F](n).map(Json.fromString),
jsonString = s => Applicative[F].pure(Json.fromString(s)),
jsonArray = arr => arr.traverse(canonicalizeJson[F]).map(Json.fromValues),
jsonObject = obj =>
obj.toList
.traverse { case (k, v) =>
canonicalizeJson[F](v).map(k -> _)
}
.map(pairs =>
Json.fromJsonObject(
JsonObject.fromIterable(
pairs.sortBy(_._1)(Ordering.by[String, Seq[Int]](_.codePoints.toArray.toSeq))
)
)
)
)

implicit def derive[F[_]: Sync, A: Encoder]: JsonCanonicalizer[F, A] =
(content: A) => canonicalizeJson[F](content.asJson)

implicit class JsonPrinterEncodeOps[F[_], A](val _v: A) extends AnyVal {

def asCanonicalJson(implicit ae: ApplicativeThrow[F], enc: Encoder[A]): F[Json] =
canonicalizeJson[F](_v.asJson)
}
}
34 changes: 13 additions & 21 deletions src/main/scala/io/constellationnetwork/metagraph_sdk/storage/impl/LevelDbCollection.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@ import cats.data.EitherT
import cats.effect.{Resource, Sync}
import cats.implicits._

import io.constellationnetwork.metagraph_sdk.std.JsonBinaryCodec
import io.constellationnetwork.metagraph_sdk.std.JsonBinaryCodec._
import io.constellationnetwork.metagraph_sdk.storage.Collection

import io.circe.{Decoder, Encoder}
Expand All @@ -15,14 +15,6 @@ import org.iq80.leveldb.impl.Iq80DBFactory

object LevelDbCollection {

implicit private class localEncoderOps[F[_]: Sync, T: Encoder](t: T) {
def toJsonBytes: F[Array[Byte]] = JsonBinaryCodec.simpleJsonSerialization(t)
}

implicit private class localDecoderOps[F[_]: Sync](bytes: Array[Byte]) {
def fromJsonBytes[T: Decoder]: F[Either[Throwable, T]] = JsonBinaryCodec.simpleJsonDeserialization(bytes)
}

def make[F[_]: Sync, Key: Encoder: Decoder, Value: Encoder: Decoder](
path: Path
): Resource[F, Collection[F, Key, Value]] = for {
Expand All @@ -31,31 +23,31 @@ object LevelDbCollection {
} yield new Collection[F, Key, Value] {

def put(id: Key, t: Value): F[Unit] = for {
idB <- id.toJsonBytes
tB <- t.toJsonBytes
idB <- id.toBinary
tB <- t.toBinary
_ <- Sync[F].blocking(db.put(idB, tB))
} yield ()

def remove(id: Key): F[Unit] =
id.toJsonBytes.flatMap { idB =>
id.toBinary.flatMap { idB =>
Sync[F].blocking(db.delete(idB))
}

def get(id: Key): F[Option[Value]] =
id.toJsonBytes
id.toBinary
.flatMap(idB => Sync[F].blocking(Option(db.get(idB))))
.flatMap(_.flatTraverse(_.fromJsonBytes[Value].map(_.toOption)))
.flatMap(_.flatTraverse(_.fromBinary[Value].map(_.toOption)))

def contains(id: Key): F[Boolean] =
id.toJsonBytes.flatMap { idB =>
id.toBinary.flatMap { idB =>
Sync[F].blocking(db.get(idB)).map(_ != null)
}

def putBatch(updates: List[(Key, Value)]): F[Unit] =
createWriteResource.use { case (batch, wo) =>
for {
_ <- updates.traverse_ { case (id, t) =>
(id.toJsonBytes, t.toJsonBytes).tupled.map { case (idB, tB) => batch.put(idB, tB) }
(id.toBinary, t.toBinary).tupled.map { case (idB, tB) => batch.put(idB, tB) }
}
_ <- Sync[F].blocking(db.write(batch, wo.sync(true)))
} yield ()
Expand All @@ -65,7 +57,7 @@ object LevelDbCollection {
createWriteResource.use { case (batch, wo) =>
for {
_ <- deletions.traverse_ { id =>
id.toJsonBytes.map(idB => batch.delete(idB))
id.toBinary.map(idB => batch.delete(idB))
}
_ <- Sync[F].blocking(db.write(batch, wo.sync(true)))
} yield ()
Expand All @@ -74,9 +66,9 @@ object LevelDbCollection {
def getBatch(keys: List[Key]): F[List[(Key, Option[Value])]] =
createReadResource.use { readOptions =>
keys.traverse { id =>
id.toJsonBytes
id.toBinary
.flatMap(idB => Sync[F].blocking(Option(db.get(idB, readOptions))))
.flatMap(_.flatTraverse(_.fromJsonBytes[Value].map(_.toOption)))
.flatMap(_.flatTraverse(_.fromBinary[Value].map(_.toOption)))
.map((id, _))
}
}
Expand All @@ -99,8 +91,8 @@ object LevelDbCollection {
else {
(for {
entry <- EitherT(Sync[F].delay(_iter.next()).attempt)
key <- EitherT(entry.getKey.fromJsonBytes[Key])
value <- EitherT(entry.getValue.fromJsonBytes[Value])
key <- EitherT[F, Throwable, Key](entry.getKey.fromBinary[Key])
value <- EitherT[F, Throwable, Value](entry.getValue.fromBinary[Value])
} yield (key, value)).value.flatMap {
case Left(_) => Sync[F].pure(Right(_buf))
case Right((key, value)) =>
Expand Down
9 changes: 6 additions & 3 deletions ...cala/io/constellationnetwork/metagraph_sdk/syntax/CurrencyIncrementalSnapshotSyntax.scala
View file
Original file line number Diff line number Diff line change
Expand Up @@ -12,20 +12,23 @@ import org.tessellation.currency.dataApplication.dataApplication.DataApplication
import org.tessellation.currency.schema.currency.{CurrencyIncrementalSnapshot, DataApplicationPart}
import org.tessellation.security.signature.Signed

import io.constellationnetwork.metagraph_sdk.std.JsonBinaryCodec
import io.constellationnetwork.metagraph_sdk.std.JsonBinaryCodec.JsonBinaryDecodeOps

import io.circe.{Decoder, Encoder}

trait CurrencyIncrementalSnapshotSyntax {

implicit class CurrencyIncrementalSnapshotOps[F[_]: Sync](cis: CurrencyIncrementalSnapshot)(implicit
json2bin: JsonBinaryCodec[F, Signed[DataApplicationBlock]]
ue: Encoder[DataUpdate],
ud: Decoder[DataUpdate]
) {

def countUpdates: F[Long] = getBlocks.map(_.map(_.updates.size.toLong).sum)

def getBlocks: F[List[Signed[DataApplicationBlock]]] =
getPart.flatMap {
_.blocks.traverse { bytes =>
json2bin.deserialize(bytes).flatMap(Sync[F].fromEither)
bytes.fromBinary[Signed[DataApplicationBlock]].flatMap(Sync[F].fromEither)
}
}

Expand Down
Loading

0 comments on commit 03f50d5

Please sign in to comment.