<commit_before>package actors

import actors.SplitsConversion.splitMessageToApiSplits
import akka.actor._
import akka.persistence._
import controllers.GetTerminalCrunch
import drt.shared.FlightsApi._
import drt.shared._
import server.protobuf.messages.CrunchState._
import services.Crunch._
import services.SDate
import services.workloadcalculator.PaxLoadCalculator.MillisSinceEpoch

import scala.collection.immutable._
import scala.language.postfixOps

case object GetFlights

class CrunchStateActor(portQueues: Map[TerminalName, Seq[QueueName]]) extends PersistentActor with ActorLogging {
  override def persistenceId: String = "crunch-state"

  var state: Option[CrunchState] = None

  val snapshotInterval = 100

  val oneDayMinutes = 1440

  override def receiveRecover: Receive = {
    case SnapshotOffer(metadata, snapshot) =>
      log.info(s"Recovery received SnapshotOffer ${metadata.timestamp} with ${snapshot.getClass}")
      setStateFromSnapshot(snapshot)

    case cdm: CrunchDiffMessage =>
      log.info(s"Recovery received CrunchDiffMessage")
      val newState = stateFromDiff(cdm, state)
      newState match {
        case None => log.info(s"Recovery: state is None")
        case Some(s) => log.info(s"Recovery: state contains ${s.flights.size} flights and ${s.crunchMinutes.size} crunch minutes")
      }
      state = newState

    case RecoveryCompleted =>
      log.info("Finished restoring crunch state")

    case u =>
      log.info(s"Recovery: received unexpected ${u.getClass}")
  }

  def setStateFromSnapshot(snapshot: Any): Unit = {
    snapshot match {
      case sm@CrunchStateSnapshotMessage(_, _, _, _) =>
        log.info(s"Using snapshot to restore")
        state = Option(snapshotMessageToState(sm))
      case somethingElse =>
        log.info(s"Ignoring unexpected snapshot ${somethingElse.getClass}")
    }
  }

  def stateFromDiff(cdm: CrunchDiffMessage, existingState: Option[CrunchState]): Option[CrunchState] = {
    val diff = crunchDiffFromMessage(cdm)
    val newState = existingState match {
      case None =>
        log.info(s"Creating an empty CrunchState to apply CrunchDiff")
        Option(CrunchState(
          flights = applyFlightsWithSplitsDiff(diff, Set()),
          crunchMinutes = applyCrunchDiff(diff, Set()),
          crunchFirstMinuteMillis = cdm.crunchStart.getOrElse(0L),
          numberOfMinutes = oneDayMinutes
        ))
      case Some(cs) =>
        log.info(s"Applying CrunchDiff to CrunchState")
        Option(cs.copy(
          flights = applyFlightsWithSplitsDiff(diff, cs.flights),
          crunchMinutes = applyCrunchDiff(diff, cs.crunchMinutes)))
    }
    newState
  }

  def crunchDiffFromMessage(diffMessage: CrunchDiffMessage): CrunchDiff = {
    CrunchDiff(
      flightRemovals = diffMessage.flightIdsToRemove.map(RemoveFlight).toSet,
      flightUpdates = diffMessage.flightsToUpdate.map(flightWithSplitsFromMessage).toSet,
      crunchMinuteRemovals = diffMessage.crunchMinutesToRemove.map(m => RemoveCrunchMinute(m.terminalName.getOrElse(""), m.queueName.getOrElse(""), m.minute.getOrElse(0L))).toSet,
      crunchMinuteUpdates = diffMessage.crunchMinutesToUpdate.map(crunchMinuteFromMessage).toSet
    )
  }

  def updateStateFromCrunchState(newState: CrunchState): Unit = {
    val existingState = state match {
      case None =>
        log.info(s"updating from no existing state")
        CrunchState(newState.crunchFirstMinuteMillis, newState.numberOfMinutes, Set(), Set())
      case Some(s) => s
    }
    val (crunchesToRemove, crunchesToUpdate) = crunchMinutesDiff(existingState.crunchMinutes, newState.crunchMinutes)
    val (flightsToRemove, flightsToUpdate) = flightsDiff(existingState.flights, newState.flights)
    val diff = CrunchDiff(flightsToRemove, flightsToUpdate, crunchesToRemove, crunchesToUpdate)

    val cmsFromDiff = applyCrunchDiff(diff, existingState.crunchMinutes)
    if (cmsFromDiff != newState.crunchMinutes) {
      log.error(s"new CrunchMinutes do not match update from diff")
    }
    val flightsFromDiff = applyFlightsWithSplitsDiff(diff, existingState.flights)
    if (flightsFromDiff != newState.flights) {
      log.error(s"new Flights do not match update from diff")
    }

    val diffToPersist = CrunchDiffMessage(
      createdAt = Option(SDate.now().millisSinceEpoch),
      crunchStart = Option(newState.crunchFirstMinuteMillis),
      flightIdsToRemove = diff.flightRemovals.map(rf => rf.flightId).toList,
      flightsToUpdate = diff.flightUpdates.map(FlightMessageConversion.flightWithSplitsToMessage).toList,
      crunchMinutesToRemove = diff.crunchMinuteRemovals.map(rc => RemoveCrunchMinuteMessage(Option(rc.terminalName), Option(rc.queueName), Option(rc.minute))).toList,
      crunchMinutesToUpdate = diff.crunchMinuteUpdates.map(cm => CrunchMinuteMessage(Option(cm.terminalName), Option(cm.queueName), Option(cm.minute), Option(cm.paxLoad), Option(cm.workLoad), Option(cm.deskRec), Option(cm.waitTime))).toList
    )

    persist(diffToPersist) { (diff: CrunchDiffMessage) =>
      log.info(s"Persisting ${diff.getClass}")
      context.system.eventStream.publish(diff)
    }

    val updatedState = existingState.copy(
      crunchFirstMinuteMillis = newState.crunchFirstMinuteMillis,
      crunchMinutes = cmsFromDiff,
      flights = flightsFromDiff)

    state = Option(updatedState)
  }

  override def receiveCommand: Receive = {
    case cs@CrunchState(_, _, _, _) =>
      log.info(s"received CrunchState. storing")
      updateStateFromCrunchState(cs)
      saveSnapshotAtInterval(cs)

    case GetFlights =>
      state match {
        case Some(CrunchState(_, _, flights, _)) =>
          sender() ! FlightsWithSplits(flights.toList)
        case None => FlightsNotReady
      }

    case GetPortWorkload =>
      state match {
        case Some(CrunchState(_, _, _, crunchMinutes)) =>
          sender() ! portWorkload(crunchMinutes)
        case None =>
          sender() ! WorkloadsNotReady()
      }

    case GetTerminalCrunch(terminalName) =>
      state match {
        case Some(CrunchState(startMillis, _, _, crunchMinutes)) =>
          sender() ! TerminalCrunchResult(queueCrunchResults(terminalName, startMillis, crunchMinutes).toList)
        case _ =>
          sender() ! TerminalCrunchResult(List[(QueueName, Either[NoCrunchAvailable, CrunchResult])]())
      }

    case SaveSnapshotSuccess(md) =><commit_msg>DRT-4687 Update log message for consistency<commit_after>package actors

import actors.SplitsConversion.splitMessageToApiSplits
import akka.actor._
import akka.persistence._
import controllers.GetTerminalCrunch
import drt.shared.FlightsApi._
import drt.shared._
import server.protobuf.messages.CrunchState._
import services.Crunch._
import services.SDate
import services.workloadcalculator.PaxLoadCalculator.MillisSinceEpoch

import scala.collection.immutable._
import scala.language.postfixOps

case object GetFlights

class CrunchStateActor(portQueues: Map[TerminalName, Seq[QueueName]]) extends PersistentActor with ActorLogging {
  override def persistenceId: String = "crunch-state"

  var state: Option[CrunchState] = None

  val snapshotInterval = 100

  val oneDayMinutes = 1440

  override def receiveRecover: Receive = {
    case SnapshotOffer(metadata, snapshot) =>
      log.info(s"Recovery: received SnapshotOffer ${metadata.timestamp} with ${snapshot.getClass}")
      setStateFromSnapshot(snapshot)

    case cdm: CrunchDiffMessage =>
      log.info(s"Recovery: received CrunchDiffMessage")
      val newState = stateFromDiff(cdm, state)
      newState match {
        case None => log.info(s"Recovery: state is None")
        case Some(s) => log.info(s"Recovery: state contains ${s.flights.size} flights and ${s.crunchMinutes.size} crunch minutes")
      }
      state = newState

    case RecoveryCompleted =>
      log.info("Finished restoring crunch state")

    case u =>
      log.info(s"Recovery: received unexpected ${u.getClass}")
  }

  def setStateFromSnapshot(snapshot: Any): Unit = {
    snapshot match {
      case sm@CrunchStateSnapshotMessage(_, _, _, _) =>
        log.info(s"Using snapshot to restore")
        state = Option(snapshotMessageToState(sm))
      case somethingElse =>
        log.info(s"Ignoring unexpected snapshot ${somethingElse.getClass}")
    }
  }

  def stateFromDiff(cdm: CrunchDiffMessage, existingState: Option[CrunchState]): Option[CrunchState] = {
    val diff = crunchDiffFromMessage(cdm)
    val newState = existingState match {
      case None =>
        log.info(s"Creating an empty CrunchState to apply CrunchDiff")
        Option(CrunchState(
          flights = applyFlightsWithSplitsDiff(diff, Set()),
          crunchMinutes = applyCrunchDiff(diff, Set()),
          crunchFirstMinuteMillis = cdm.crunchStart.getOrElse(0L),
          numberOfMinutes = oneDayMinutes
        ))
      case Some(cs) =>
        log.info(s"Applying CrunchDiff to CrunchState")
        Option(cs.copy(
          flights = applyFlightsWithSplitsDiff(diff, cs.flights),
          crunchMinutes = applyCrunchDiff(diff, cs.crunchMinutes)))
    }
    newState
  }

  def crunchDiffFromMessage(diffMessage: CrunchDiffMessage): CrunchDiff = {
    CrunchDiff(
      flightRemovals = diffMessage.flightIdsToRemove.map(RemoveFlight).toSet,
      flightUpdates = diffMessage.flightsToUpdate.map(flightWithSplitsFromMessage).toSet,
      crunchMinuteRemovals = diffMessage.crunchMinutesToRemove.map(m => RemoveCrunchMinute(m.terminalName.getOrElse(""), m.queueName.getOrElse(""), m.minute.getOrElse(0L))).toSet,
      crunchMinuteUpdates = diffMessage.crunchMinutesToUpdate.map(crunchMinuteFromMessage).toSet
    )
  }

  def updateStateFromCrunchState(newState: CrunchState): Unit = {
    val existingState = state match {
      case None =>
        log.info(s"updating from no existing state")
        CrunchState(newState.crunchFirstMinuteMillis, newState.numberOfMinutes, Set(), Set())
      case Some(s) => s
    }
    val (crunchesToRemove, crunchesToUpdate) = crunchMinutesDiff(existingState.crunchMinutes, newState.crunchMinutes)
    val (flightsToRemove, flightsToUpdate) = flightsDiff(existingState.flights, newState.flights)
    val diff = CrunchDiff(flightsToRemove, flightsToUpdate, crunchesToRemove, crunchesToUpdate)

    val cmsFromDiff = applyCrunchDiff(diff, existingState.crunchMinutes)
    if (cmsFromDiff != newState.crunchMinutes) {
      log.error(s"new CrunchMinutes do not match update from diff")
    }
    val flightsFromDiff = applyFlightsWithSplitsDiff(diff, existingState.flights)
    if (flightsFromDiff != newState.flights) {
      log.error(s"new Flights do not match update from diff")
    }

    val diffToPersist = CrunchDiffMessage(
      createdAt = Option(SDate.now().millisSinceEpoch),
      crunchStart = Option(newState.crunchFirstMinuteMillis),
      flightIdsToRemove = diff.flightRemovals.map(rf => rf.flightId).toList,
      flightsToUpdate = diff.flightUpdates.map(FlightMessageConversion.flightWithSplitsToMessage).toList,
      crunchMinutesToRemove = diff.crunchMinuteRemovals.map(rc => RemoveCrunchMinuteMessage(Option(rc.terminalName), Option(rc.queueName), Option(rc.minute))).toList,
      crunchMinutesToUpdate = diff.crunchMinuteUpdates.map(cm => CrunchMinuteMessage(Option(cm.terminalName), Option(cm.queueName), Option(cm.minute), Option(cm.paxLoad), Option(cm.workLoad), Option(cm.deskRec), Option(cm.waitTime))).toList
    )

    persist(diffToPersist) { (diff: CrunchDiffMessage) =>
      log.info(s"Persisting ${diff.getClass}")
      context.system.eventStream.publish(diff)
    }

    val updatedState = existingState.copy(
      crunchFirstMinuteMillis = newState.crunchFirstMinuteMillis,
      crunchMinutes = cmsFromDiff,
      flights = flightsFromDiff)

    state = Option(updatedState)
  }

  override def receiveCommand: Receive = {
    case cs@CrunchState(_, _, _, _) =>
      log.info(s"Received CrunchState. storing")
      updateStateFromCrunchState(cs)
      saveSnapshotAtInterval(cs)

    case GetFlights =>
      state match {
        case Some(CrunchState(_, _, flights, _)) =>
          sender() ! FlightsWithSplits(flights.toList)
        case None => FlightsNotReady
      }

    case GetPortWorkload =>
      state match {
        case Some(CrunchState(_, _, _, crunchMinutes)) =>
          sender() ! portWorkload(crunchMinutes)
        case None =>
          sender() ! WorkloadsNotReady()
      }

    case GetTerminalCrunch(terminalName) =>
      state match {
        case Some(CrunchState(startMillis, _, _, crunchMinutes)) =>
          sender() ! TerminalCrunchResult(queueCrunchResults(terminalName, startMillis, crunchMinutes).toList)
        case _ =>
          sender() ! TerminalCrunchResult(List[(QueueName, Either[NoCrunchAvailable, CrunchResult])]())
      }

    case SaveSnapshotSuccess(md) =>
{'fraction_non_alphanumeric': 0.10004678727386151, 'fraction_numerical': 0.0017155333749220212, 'mean_word_length': 3.946008484381026, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4664243', 'n_tokens_mistral': 3965, 'n_tokens_neox': 3877, 'n_words': 719}