implicit val ec = scala.concurrent.ExecutionContext.global
val database: Database = FDB.selectAPIVersion(630).open(null, ec)

final case class Book(isbn: String, title: String, publishedOn: LocalDate)

val booksSubspace = new TypedSubspace[Book, String] {
  override val subspace: Subspace = new Subspace(Tuple.from("books"))
  override def toKey(entity: Book): String = entity.isbn
  override def toRawValue(entity: Book): Array[Byte] = {
    Tuple.from(entity.title, entity.publishedOn.toString).pack
  }
  override def toTupledKey(key: String): Tuple = Tuple.from(key)
  override def toKey(tupledKey: Tuple): String = tupledKey.getString(0)
  override def toEntity(key: String, value: Array[Byte]): Book = {
    val tupledValue = Tuple.fromBytes(value)
    val publishedOn = LocalDate.parse(tupledValue.getString(1))
    Book(isbn = key, title = tupledValue.getString(0), publishedOn = publishedOn)
  }
}

val dbio: DBIO[Option[Book]] = for {
  _ <- booksSubspace.set(Book("978-0451205766", "The Godfather", LocalDate.parse("2002-03-01")))
  maybeBook <- booksSubspace.get("978-0451205766").toDBIO
} yield maybeBook

val maybeBook: Future[Option[Book]] = dbio.transact(database)

val fdb4sVersion = "0.12.0"

libraryDependencies ++= Seq(
  "com.github.pwliwanow.foundationdb4s" %% "core" % fdb4sVersion,
  "com.github.pwliwanow.foundationdb4s" %% "schema" % fdb4sVersion,
  "com.github.pwliwanow.foundationdb4s" %% "akka-streams" % fdb4sVersion
)

import com.github.pwliwanow.foundationdb4s.schema._
import shapeless.{::, HNil}

implicit val ec = scala.concurrent.ExecutionContext.global
val database: Database = FDB.selectAPIVersion(630).open(null, ec)

object Language extends Enumeration {
  type Language = Value
  val En, De = Value
}
case class ISBN(value: String) extends AnyVal
import Language._

final case class Book(language: Language, isbn: ISBN, title: String, publishedOn: LocalDate)

object Codecs {
  implicit val localDateEnc = implicitly[TupleEncoder[Long]].contramap[LocalDate](_.toEpochDay)
  implicit val localDateDec = implicitly[TupleDecoder[Long]].map(LocalDate.ofEpochDay)
  implicit val languageEnc = implicitly[TupleEncoder[String]].contramap[Language](_.toString)
  implicit val languageDec = implicitly[TupleDecoder[String]].map(Language.withName)
}
import Codecs._

object BookSchema extends Schema {
  type Entity = Book
  type KeySchema = Language :: LocalDate :: ISBN :: HNil
  type ValueSchema = String :: HNil

  override def toKey(entity: Book): BookSchema.KeySchema =
    entity.language :: entity.publishedOn :: entity.isbn :: HNil
  override def toValue(entity: Book): BookSchema.ValueSchema =
    entity.title :: HNil
  override def toEntity(key: BookSchema.KeySchema, valueRepr: BookSchema.ValueSchema): Book = {
    val language :: publishedOn :: isbn :: HNil = key
    val title :: HNil = valueRepr
    Book(language, isbn, title, publishedOn)
  }
}
  
val booksNamespace = new BookSchema.Namespace(new Subspace(Tuple.from("books")))

val dbio: ReadDBIO[Seq[Book]] = booksNamespace.getRange((Language.En, LocalDate.of(2018, 7, 10)))
// or booksNamespace.getRange(Language.En :: LocalDate.of(2018, 7, 10) :: HNil)
val result: Future[Seq[Book]] = dbio.transact(database)

// `Parallel` type classes are defined in `DBIO` and `ReadDBIO` companion objects, 
// so there is no need to import them as they are in scope automatically 
import cats.implicits._

val dbios: List[ReadDBIO[Option[Book]]] = 
  List(booksSubspace.get("978-0451205766"), booksSubspace.get("978-1491962299"))
// instruct `dbios` to be run in parallel
val dbio: ReadDBIO[List[Option[Book]]] = dbios.parSequence
val result: Future[List[Option[Book]]] = dbio.transact(database)

implicit val ec = scala.concurrent.ExecutionContext.global
val database: Database = FDB.selectAPIVersion(630).open(null, ec)

case class EventKey(eventType: String, versionstamp: Versionstamp)
case class Event(key: EventKey, content: Array[Byte])

val eventsSubspace = new SubspaceWithVersionstampedKeys[Event, EventKey] {
  override val subspace: Subspace = new Subspace(Tuple.from("events"))
  override def toKey(entity: Event): EventKey = entity.key
  override def toRawValue(entity: Event): Array[Byte] = event.content
  override def toTupledKey(key: EventKey): Tuple = Tuple.from(key.eventType, key.versiostamp)
  override def toKey(tupledKey: Tuple): EventKey = {
    EventKey(tupledKey.getString(0), tupledKey.getVersiostamp(1))
  }
  override def toEntity(key: EventKey, value: Array[Byte]): Event = Event(key, value)
  override def extractVersionstamp(key: EventKey): Versionstamp = key.versionstamp
}

val event = Event(
  key = EventKey("UserAdded", Versionstamp.incomplete(0)), 
  content = Tuple.from("""{ "name": "John Smith" }""").pack)

// save new event
val setDbio: DBIO[Unit] = eventsSubspace.set(event)
val completedVersionstamp: Versionstamp = 
  Await.result(
    setDbio
      .transactVersionstamped(database, userVersion = 0)
      .map { case (_, Some(versionstamp)) => versionstamp },
    Duration.Inf)

// update previously persisted event
val updatedEvent = event.copy(key = event.key.copy(versionstamp = completedVersionstamp))
val updateDbio = eventsSubspace.set(updatedEvent)

updateDbio.transact(database)

// reusing first example
final case class Book(isbn: String, title: String, publishedOn: LocalDate)
val booksSubspace: TypedSubspace[Book, String] = ???

val key = "978-0451205766"
val dbio: DBIO[Promise[Unit]] = booksSubspace.watch(key)
val futureWatch: Future[Promise[Unit]] = dbio.transact(database)

val source: Source[Entity, _] = SubspaceSource.from(subspace, database)

val processEntityFlow: Flow[Entity, Entity, NotUsed] = ???
val commitOffsetSink: Sink[Entity, NotUsed] = ???
val lastSeenKey: Array[Byte] = fetchLastSeenKey()
val source = Source[Entity, _] = 
  InfinitePollingSubspaceSource.from(
    typedSubspace, 
    database, 
    pollingInterval = 100.millis, 
    begin = KeySelector.firstGreaterThan(lastSeenKey))
source.via(processEntityFlow).to(commitOffsetSink).run()