jchapuis

def updateIndexFiles(newFile: IndexPage.File) =
  for {
    _     <- logger.info("Retrieving index page")
    files <- getIndexByNamePage.map(IndexPage.parseFiles)
    _     <- logger.info(s"Retrieved index page with ${files.size} files")
    _     <- uploadNewIndexFiles(files, newFile)
  } yield ()

def uploadNewIndexFiles(files: List[IndexPage.File], newFile: IndexPage.File) = {
  val newFiles              = files.toSet + newFile

jchapuis

def safeUpdateIndexFiles(newFile: IndexPage.File) =
  leaseRepository
    .acquire(LeaseID(ks"file-uploader"), HolderID.unique)
    .use(_.guard(updateIndexFiles(newFile)).map(_.embedError))

jchapuis

trait LeaseRepository[F[_]]: 
  def labels: List[Label]
  def acquire(id: LeaseID, holderID: HolderID, annotations: List[Annotation] = Nil)(implicit
      parameters: LeaseParameters = LeaseParameters.Default
  ): Resource[F, HeldLease[F]]
  def get(id: LeaseID): F[Option[Lease[F]]]
  def list: F[List[Lease[F]]]
  def watcher: fs2.Stream[F, LeaseEvent[F]]

jchapuis

trait Lease[F[_]]:
  def id: LeaseID
  def holder: F[HolderID]
  def labels: F[List[Label]]
  def annotations: F[List[Annotation]]
  def isExpired: F[Boolean]
  def expired: fs2.Stream[F, Unit]

jchapuis

def prepare(transferID: TransferID, transfer: Transfer): F[Branch.Vote[TransferFailure]] = {
  if (accountID === transfer.origin)
    Logger[F].debug(show"Preparing outgoing transfer $transferID: $transfer for account $accountID") >>
      account.prepareOutgoingTransfer(transferID, transfer)
        .onErrorRetryWithBackoff(Logger[F].warn(_)(show"Error preparing outgoing transfer $transferID, retrying in a bit"))
        .onLeftRetryWithBackoff { case Account.PendingOutgoingTransfer =>
          Logger[F].warn(show"Account $accountID has a pending outgoing transfer, retrying in a bit")
        }(retryParameters.onPendingTransfer)
        .flatMap {
          case Left(Account.Unknown) => Branch.Vote.Abort(TransferFailure.AccountNotFound(accountID)).pure[F]

jchapuis

def transfer(from: AccountID, to: AccountID, amount: PosAmount): F[TransferFailure \/ Unit] =
  coordinator
    .create(TransferID.random, Transfer(from, to, amount), from, to)
    .use(_.pollForFinalStatus())
    .flatMap {
      case Status.Committed => ().asRight[TransferFailure].pure
      case Status.Aborted(reason) =>
        reason match {
          case AbortReason.Timeout =>
            EitherT.leftT(TransferFailure.Timeout: TransferFailure).value

jchapuis

transactor.coordinator[TransferID, AccountID, Transfer, TransferFailure](
  "transfer",
  { accountID =>
    val account = sharding.entityFor(accountID)
    new TransferBranch(accountID, account)
  },
  Some(transferParameters.timeout)
)

jchapuis

trait Branch[F[_], TID, Q, R] 
  def prepare(id: TID, query: Q): F[Vote[R]]
  def commit(id: TID): F[Unit]
  def abort(id: TID): F[Unit]

jchapuis

jchapuis

import cats.{ Eq, Show }
import cats.syntax.eq._
import cats.syntax.show._
import cats.instances.int._
import FunctionHelpers._

object Differ {
  sealed trait Diff[T]
  object Diff {
    final case class Insert[T](revision: T)            extends Diff[T]