Content monitoring made easy with Ior

IorMonitoring.scala

package com.itv.sif.db

import cats.{Order, Show}
import cats.data.{Ior, NonEmptyList, NonEmptySet, Validated}
import cats.kernel.Semigroup

import cats.syntax.apply._
import cats.syntax.option._
import cats.syntax.order._
import cats.syntax.show._
import cats.syntax.foldable._

import cats.instances.list._
import cats.instances.long._
import cats.instances.string._
import cats.instances.sortedSet._
import cats.instances.option._

object demo2 extends App {

  import GenericModels._
  import DomainModels._
  import SharedProcessing._
  import PrettyPrint._

  object GenericModels {
    type TimeStamp = Long
    type Identifier = Long
    type Reason = String

    case class InternalComponent(name: String, id: Identifier)
    object InternalComponent {
      implicit val order: Order[InternalComponent] = {
        val byName: Order[InternalComponent] = Order.by(_.name)
        val byId: Order[InternalComponent] = Order.by(_.id)

        Order.whenEqual(byName, byId)
      }
    }

    sealed trait Source
    object Source {
      case class ExternalSource(name: String) extends Source
      case class InternalSource(components: NonEmptySet[InternalComponent]) extends Source
      implicit val order: Order[Source] = (x: Source, y: Source) => (x, y) match {
        case (_:ExternalSource, _: InternalSource) =>
          -1
        case (_: InternalSource, _:ExternalSource) =>
          1
        case (InternalSource(cx), InternalSource(cy)) =>
          cx.toSortedSet.compare(cy.toSortedSet)
        case (ExternalSource(nameX), ExternalSource(nameY)) =>
          nameX.compare(nameY)
      }
    }

    case class EventIncompleteBecause(id: Identifier, timeStamp: TimeStamp, sources: NonEmptySet[Source], reasons: NonEmptyList[Reason])
    object EventIncompleteBecause {
      implicit val show: Show[EventIncompleteBecause] = entry => s"Error: {timeStamp: ${entry.timeStamp}, id: ${entry.id}, reasons: ${entry.reasons.toList.mkString(", ")}, sources: ${entry.sources.toSortedSet.mkString(", ")} }"
    }

    case class ErrorLog(sources: NonEmptySet[Source], reasons: NonEmptyList[Reason])
    object ErrorLog {
      def one(source: Source, reason: Reason): ErrorLog = ErrorLog(
        NonEmptySet.one(source),
        NonEmptyList.one(reason)
      )

      implicit val semigroup: Semigroup[ErrorLog] = (x: ErrorLog, y: ErrorLog) => ErrorLog(x.sources ++ y.sources, x.reasons ::: y.reasons)
    }

    case class Availability[E, A](value: Ior[E, A]) extends AnyVal
    object Availability {
      implicit def show[E: Show, A: Show]: Show[Availability[E,A]] = _.value.fold(
        e => s" - Error: ${e.show}".magenta,
        a => s" - Available: ${a.show}".cyan,
        (e, a) => s" - Unavailable because: ${e.show}".red ++  s" raw value: ${a.show}".yellow,
      )
    }

  }

  object PrettyPrint {
    implicit class ConsoleColorise(val str: String) extends AnyVal {
      import Console._
      def black     = s"$BLACK$str$RESET"
      def red       = s"$RED$str$RESET"
      def green     = s"$GREEN$str$RESET"
      def yellow    = s"$YELLOW$str$RESET"
      def blue      = s"$BLUE$str$RESET"
      def magenta   = s"$MAGENTA$str$RESET"
      def cyan      = s"$CYAN$str$RESET"
      def white     = s"$WHITE$str$RESET"

      def blackBg   = s"$BLACK_B$str$RESET"
      def redBg     = s"$RED_B$str$RESET"
      def greenBg   = s"$GREEN_B$str$RESET"
      def yellowBg  = s"$YELLOW_B$str$RESET"
      def blueBg    = s"$BLUE_B$str$RESET"
      def magentaBg = s"$MAGENTA_B$str$RESET"
      def cyanBg    = s"$CYAN_B$str$RESET"
      def whiteBg   = s"$WHITE_B$str$RESET"
    }
  }

  object DomainModels {

    trait Hobby
    object Hobby {
      case object BirdWatching extends Hobby
      case object Sports extends Hobby
    }

    case class ProfileEvent(id: Identifier, timeStamp: TimeStamp, maybeAge: Option[Int], maybeHobby: Option[Hobby])
    object ProfileEvent {
      val AliceId = 1L
      val BobId = 2L

      def ofBob(timeStamp: TimeStamp, age: Option[Int], hobby: Option[Hobby]): ProfileEvent = ProfileEvent(BobId, timeStamp, age, hobby)
      def ofAlice(timeStamp: TimeStamp, age: Option[Int], hobby: Option[Hobby]): ProfileEvent = ProfileEvent(AliceId, timeStamp, age, hobby)

      implicit val show: Show[ProfileEvent] = entry => s"{timeStamp: ${entry.timeStamp}, id: ${entry.id}, age: ${entry.maybeAge} , hobby: ${entry.maybeHobby}}"
    }
    case class Profile private (age: Int, hobby: Hobby)
    object Profile {
      def fromDbEntry(dbProfile: ProfileEvent): Validated[ErrorLog, Profile] = {
        (
          Validated.fromOption(dbProfile.maybeAge, ErrorLog.one(Source.ExternalSource("user"), "Age is not specified.")).andThen(
            i => Validated.cond(i> 0, i, ErrorLog.one(Source.ExternalSource("user"), "Age is negative."))
          ),
          Validated.fromOption(dbProfile.maybeHobby, ErrorLog.one(Source.ExternalSource("user"), "Hobby is not specified."))
          ).mapN{
          case (age, hobby) => Profile(age, hobby)
        }
      }
    }

    case class MatchEvent(id: Identifier, timeStamp: TimeStamp, p1: Option[ProfileEvent], p2: Option[ProfileEvent])
    object MatchEvent {
      implicit val show: Show[MatchEvent] = entry => s"{timeStamp: ${entry.timeStamp}, p1: ${entry.p1.show}, p2: ${entry.p2.show}}"
    }

    case class Match private (p1: Profile, p2: Profile)
    object Match {
      def fromDbEntry(dbMatch: MatchEvent): Validated[ErrorLog, Match] = {
        (
          Validated.fromOption(dbMatch.p1, ErrorLog.one(Source.ExternalSource("profile1"), "Profile 1 could not be found")),
          Validated.fromOption(dbMatch.p2, ErrorLog.one(Source.ExternalSource("profile2"), "Profile 2 could not be found"))
          ).mapN{
          case (p1, p2) =>
            (
              Profile.fromDbEntry(p1)
                .leftMap(_ => ErrorLog.one(Source.InternalSource(NonEmptySet.one(InternalComponent("profile1", p1.id))), "Profile was not complete")) // not passing on the full set of reasons, just giving enough info to find it later.
              ,
                Profile.fromDbEntry(p2)
                .leftMap(_ => ErrorLog.one(Source.InternalSource(NonEmptySet.one(InternalComponent("profile2", p2.id))), "Profile was not complete")) //same
              ).mapN{ case (profile1, profile2) =>
              Validated.cond(
                profile1.hobby == profile2.hobby,
                Match(profile1, profile2),
                ErrorLog.one(
                  Source.InternalSource(
                    NonEmptySet.of(
                      InternalComponent("profile1", p1.id),
                      InternalComponent("profile2", p2.id)
                    )
                  ),
                  "The profiles are both complete but do not have compatible hobbies."
                )
              )
            }.andThen(identity)

        }.andThen(identity)
      }
    }
  }

  object SharedProcessing {

    def process[E, A, B](value: A, validate: A => Validated[E, B]): Ior[E, A] = validate(value) match {
      case Validated.Valid(_) => Ior.right(value)
      case Validated.Invalid(e) => Ior.both(e, value)
    }

    var profileHistory: List[Ior[EventIncompleteBecause, ProfileEvent]] = Nil
    var matchHistory: List[Ior[EventIncompleteBecause, MatchEvent]] = Nil

    def profilesWithId(id: Identifier): List[Ior[EventIncompleteBecause, ProfileEvent]] = profileHistory.filter(
      _.bimap(_.id, _.id).toEither.merge == id
    )

    def processMatchEntry(record: MatchEvent): Ior[EventIncompleteBecause, MatchEvent] = {
      process(
        record,
        (entry: MatchEvent) => Match.fromDbEntry(entry)
      )
        .leftMap(l =>
          EventIncompleteBecause(record.id, record.timeStamp, l.sources, l.reasons)
        )
    }

    def aliceHistory: List[Ior[EventIncompleteBecause, ProfileEvent]] = profilesWithId(ProfileEvent.AliceId)
    def bobHistory: List[Ior[EventIncompleteBecause, ProfileEvent]] = profilesWithId(ProfileEvent.BobId)

  }

  trait HistoryRecorder[E, A] {
    def getHistory(): List[Ior[E, A]]
  }

  object RealTimeProcessing extends HistoryRecorder[EventIncompleteBecause, MatchEvent] {

    private def getLatestTimeStampTimeStamp(): TimeStamp = profileHistory.headOption.map(
      _.bimap(_.timeStamp, _.timeStamp).fold(identity, identity, math.max)
    ).getOrElse(0L)

    private def processProfile(record: ProfileEvent): Ior[EventIncompleteBecause, ProfileEvent] = {
      process(record, Profile.fromDbEntry)
        .leftMap(
        l => EventIncompleteBecause(record.id, record.timeStamp, l.sources, l.reasons)
      )
    }

    private def readLatestAliceAndBobMatchState(): MatchEvent = MatchEvent(42,
      getLatestTimeStampTimeStamp(),
      aliceHistory.headOption.flatMap(_.right),
      bobHistory.headOption.flatMap(_.right)
    )

    def processProfileEntry(record: ProfileEvent): Unit = {
      profileHistory = processProfile(record) :: profileHistory
      val matchEntry = readLatestAliceAndBobMatchState()


      val processed: Ior[EventIncompleteBecause, MatchEvent] = processMatchEntry(matchEntry)

      val processedWithBug: Ior[EventIncompleteBecause, MatchEvent] = {
        val bug = ErrorLog.one(Source.ExternalSource("universe"), "bug")
        val dbErrorLog = EventIncompleteBecause(42, record.timeStamp, bug.sources, bug.reasons)

        processed.fold(
          e => Ior.left(e), _ => Ior.left(dbErrorLog), (e, a) => Ior.both(e,a)
        )

      }

      matchHistory = processedWithBug :: matchHistory
    }

    override def getHistory(): List[Ior[EventIncompleteBecause, MatchEvent]] = matchHistory

  }

  object PostProcessing extends HistoryRecorder[EventIncompleteBecause, MatchEvent] {
    private def getProfile(identifier: Identifier, timeStamp: TimeStamp): Option[ProfileEvent] = {
      profilesWithId(identifier)
        .filter(
          _.bimap(_.timeStamp, _.timeStamp).toEither.merge <= timeStamp
        ).maximumOption(Order.by(_.bimap(_.timeStamp, _.timeStamp).toEither.merge))
        .flatMap(_.right)
    }

    private def getAllEventTimeStamps(): List[TimeStamp] = profileHistory.map(_.bimap(_.timeStamp, _.timeStamp).toEither.merge)

    private def reconstructedMatchEntries: List[MatchEvent] = getAllEventTimeStamps().map(t =>
      MatchEvent(
        42,
        t,
        getProfile(ProfileEvent.AliceId, t),
        getProfile(ProfileEvent.BobId, t)
      )
    )

    override def getHistory(): List[Ior[EventIncompleteBecause, MatchEvent]] = reconstructedMatchEntries.map(processMatchEntry)
  }

  RealTimeProcessing.processProfileEntry(ProfileEvent.ofAlice(1, age = none, hobby = none))
  RealTimeProcessing.processProfileEntry(ProfileEvent.ofAlice(2, none, Hobby.BirdWatching.some))
  RealTimeProcessing.processProfileEntry(ProfileEvent.ofAlice(3, 30.some, Hobby.BirdWatching.some))

  RealTimeProcessing.processProfileEntry(ProfileEvent.ofBob(4, none, none))
  RealTimeProcessing.processProfileEntry(ProfileEvent.ofBob(5, none, Hobby.Sports.some))
  RealTimeProcessing.processProfileEntry(ProfileEvent.ofBob(6, 29.some, Hobby.Sports.some))
  RealTimeProcessing.processProfileEntry(ProfileEvent.ofBob(7, 29.some, Hobby.BirdWatching.some))

  RealTimeProcessing.processProfileEntry(ProfileEvent.ofAlice(8, -1.some, Hobby.BirdWatching.some))

  //todo: color print these.
  println("\n***Stored Match values***\n".white)
  RealTimeProcessing.getHistory().reverse.map(Availability.apply).foreach(a => println(a.show))

  println("\n***Alice(Profile1)***\n".white)
  SharedProcessing.aliceHistory.reverse.map(Availability.apply).foreach(a => println(a.show))

  println("\n***Bob(Profile2)***\n".white)
  SharedProcessing.bobHistory.reverse.map(Availability.apply).foreach(a => println(a.show))

  println("\n***Reconstructed Match values***\n".white)
  PostProcessing.getHistory().reverse.map(Availability.apply).foreach(a => println(a.show))

}


//***Stored Match values***
//
//- Unavailable because: Error: {timeStamp: 1, id: 42, reasons: Profile 2 could not be found, sources: ExternalSource(profile2) } raw value: {timeStamp: 1, p1: Some({timeStamp: 1, id: 1, age: None , hobby: None}), p2: None}
//  - Unavailable because: Error: {timeStamp: 2, id: 42, reasons: Profile 2 could not be found, sources: ExternalSource(profile2) } raw value: {timeStamp: 2, p1: Some({timeStamp: 2, id: 1, age: None , hobby: Some(BirdWatching)}), p2: None}
//  - Unavailable because: Error: {timeStamp: 3, id: 42, reasons: Profile 2 could not be found, sources: ExternalSource(profile2) } raw value: {timeStamp: 3, p1: Some({timeStamp: 3, id: 1, age: Some(30) , hobby: Some(BirdWatching)}), p2: None}
//  - Unavailable because: Error: {timeStamp: 4, id: 42, reasons: Profile was not complete, sources: InternalSource(TreeSet(InternalComponent(profile2,2))) } raw value: {timeStamp: 4, p1: Some({timeStamp: 3, id: 1, age: Some(30) , hobby: Some(BirdWatching)}), p2: Some({timeStamp: 4, id: 2, age: None , hobby: None})}
//  - Unavailable because: Error: {timeStamp: 5, id: 42, reasons: Profile was not complete, sources: InternalSource(TreeSet(InternalComponent(profile2,2))) } raw value: {timeStamp: 5, p1: Some({timeStamp: 3, id: 1, age: Some(30) , hobby: Some(BirdWatching)}), p2: Some({timeStamp: 5, id: 2, age: None , hobby: Some(Sports)})}
//  - Unavailable because: Error: {timeStamp: 6, id: 42, reasons: The profiles are both complete but do not have compatible hobbies., sources: InternalSource(TreeSet(InternalComponent(profile1,1), InternalComponent(profile2,2))) } raw value: {timeStamp: 6, p1: Some({timeStamp: 3, id: 1, age: Some(30) , hobby: Some(BirdWatching)}), p2: Some({timeStamp: 6, id: 2, age: Some(29) , hobby: Some(Sports)})}
//  - Error: Error: {timeStamp: 7, id: 42, reasons: bug, sources: ExternalSource(universe) }
//  - Unavailable because: Error: {timeStamp: 8, id: 42, reasons: Profile was not complete, sources: InternalSource(TreeSet(InternalComponent(profile1,1))) } raw value: {timeStamp: 8, p1: Some({timeStamp: 8, id: 1, age: Some(-1) , hobby: Some(BirdWatching)}), p2: Some({timeStamp: 7, id: 2, age: Some(29) , hobby: Some(BirdWatching)})}
//
//  ***Alice(Profile1)***
//
//  - Unavailable because: Error: {timeStamp: 1, id: 1, reasons: Age is not specified., Hobby is not specified., sources: ExternalSource(user) } raw value: {timeStamp: 1, id: 1, age: None , hobby: None}
//  - Unavailable because: Error: {timeStamp: 2, id: 1, reasons: Age is not specified., sources: ExternalSource(user) } raw value: {timeStamp: 2, id: 1, age: None , hobby: Some(BirdWatching)}
//  - Available: {timeStamp: 3, id: 1, age: Some(30) , hobby: Some(BirdWatching)}
//  - Unavailable because: Error: {timeStamp: 8, id: 1, reasons: Age is negative., sources: ExternalSource(user) } raw value: {timeStamp: 8, id: 1, age: Some(-1) , hobby: Some(BirdWatching)}
//
//  ***Bob(Profile2)***
//
//  - Unavailable because: Error: {timeStamp: 4, id: 2, reasons: Age is not specified., Hobby is not specified., sources: ExternalSource(user) } raw value: {timeStamp: 4, id: 2, age: None , hobby: None}
//  - Unavailable because: Error: {timeStamp: 5, id: 2, reasons: Age is not specified., sources: ExternalSource(user) } raw value: {timeStamp: 5, id: 2, age: None , hobby: Some(Sports)}
//  - Available: {timeStamp: 6, id: 2, age: Some(29) , hobby: Some(Sports)}
//  - Available: {timeStamp: 7, id: 2, age: Some(29) , hobby: Some(BirdWatching)}
//
//  ***Reconstructed Match values***
//
//  - Unavailable because: Error: {timeStamp: 1, id: 42, reasons: Profile 2 could not be found, sources: ExternalSource(profile2) } raw value: {timeStamp: 1, p1: Some({timeStamp: 1, id: 1, age: None , hobby: None}), p2: None}
//  - Unavailable because: Error: {timeStamp: 2, id: 42, reasons: Profile 2 could not be found, sources: ExternalSource(profile2) } raw value: {timeStamp: 2, p1: Some({timeStamp: 2, id: 1, age: None , hobby: Some(BirdWatching)}), p2: None}
//  - Unavailable because: Error: {timeStamp: 3, id: 42, reasons: Profile 2 could not be found, sources: ExternalSource(profile2) } raw value: {timeStamp: 3, p1: Some({timeStamp: 3, id: 1, age: Some(30) , hobby: Some(BirdWatching)}), p2: None}
//  - Unavailable because: Error: {timeStamp: 4, id: 42, reasons: Profile was not complete, sources: InternalSource(TreeSet(InternalComponent(profile2,2))) } raw value: {timeStamp: 4, p1: Some({timeStamp: 3, id: 1, age: Some(30) , hobby: Some(BirdWatching)}), p2: Some({timeStamp: 4, id: 2, age: None , hobby: None})}
//  - Unavailable because: Error: {timeStamp: 5, id: 42, reasons: Profile was not complete, sources: InternalSource(TreeSet(InternalComponent(profile2,2))) } raw value: {timeStamp: 5, p1: Some({timeStamp: 3, id: 1, age: Some(30) , hobby: Some(BirdWatching)}), p2: Some({timeStamp: 5, id: 2, age: None , hobby: Some(Sports)})}
//  - Unavailable because: Error: {timeStamp: 6, id: 42, reasons: The profiles are both complete but do not have compatible hobbies., sources: InternalSource(TreeSet(InternalComponent(profile1,1), InternalComponent(profile2,2))) } raw value: {timeStamp: 6, p1: Some({timeStamp: 3, id: 1, age: Some(30) , hobby: Some(BirdWatching)}), p2: Some({timeStamp: 6, id: 2, age: Some(29) , hobby: Some(Sports)})}
//  - Available: {timeStamp: 7, p1: Some({timeStamp: 3, id: 1, age: Some(30) , hobby: Some(BirdWatching)}), p2: Some({timeStamp: 7, id: 2, age: Some(29) , hobby: Some(BirdWatching)})}
//  - Unavailable because: Error: {timeStamp: 8, id: 42, reasons: Profile was not complete, sources: InternalSource(TreeSet(InternalComponent(profile1,1))) } raw value: {timeStamp: 8, p1: Some({timeStamp: 8, id: 1, age: Some(-1) , hobby: Some(BirdWatching)}), p2: Some({timeStamp: 7, id: 2, age: Some(29) , hobby: Some(BirdWatching)})}
//