<commit_before>package actors

import actors.FlightMessageConversion._
import akka.actor._
import akka.event.LoggingReceive
import akka.pattern.AskableActorRef
import akka.persistence._
import akka.util.Timeout
import controllers.FlightState
import drt.shared.FlightsApi.{Flights, FlightsWithSplits}
import drt.shared.PassengerSplits.{FlightNotFound, VoyagePaxSplits}
import drt.shared.{Arrival, _}
import org.joda.time.LocalDate
import org.joda.time.format.DateTimeFormat
import passengersplits.core.PassengerInfoRouterActor.ReportVoyagePaxSplit
import server.protobuf.messages.FlightsMessage.{FlightLastKnownPaxMessage, FlightStateSnapshotMessage, FlightsMessage}
import services.SplitsProvider.SplitProvider
import services.{CSVPassengerSplitsProvider, FastTrackPercentages, SDate}

import scala.collection.immutable.Seq
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.language.postfixOps

case object GetFlights

case object GetFlightsWithSplits

class FlightsActor(crunchActorRef: ActorRef,
                   dqApiSplitsActorRef: AskableActorRef,
                   csvSplitsProvider: SplitProvider,
                   val airportConfig: AirportConfig)
  extends PersistentActor
    with ActorLogging
    with FlightState
    with DomesticPortList {
  implicit val timeout = Timeout(5 seconds)


  import SplitRatiosNs.SplitSources._

  val snapshotInterval = 20

  override def persistenceId = "flights-store"

  override protected def onRecoveryFailure(cause: Throwable, event: Option[Any]): Unit = {
    Recovery
    super.onRecoveryFailure(cause, event)
    log.error(cause, "recovery failed in flightsActors")
  }

  val receiveRecover: Receive = {
    case FlightsMessage(recoveredFlights) =>
      log.info(s"Recovering ${recoveredFlights.length} new flights")
      setFlights(recoveredFlights.map(flightMessageToApiFlight).map(f => (f.FlightID, f)).toMap)
    case SnapshotOffer(_, snapshot) =>
      val flightsFromSnapshot = snapshot match {
        case flightStateSnapshot: FlightStateSnapshotMessage =>
          log.info(s"Restoring snapshot from protobuf message")
          flightStateSnapshot.flightMessages.map(flightMessageToApiFlight).map(f => (f.FlightID, f)).toMap
        case flights: Map[Int, Any] =>
          log.info(s"Restoring snapshot from legacy ApiFlight")
          flights.mapValues {
            case f: ApiFlight => ApiFlightToArrival(f)
          }
      }<commit_msg>DRT-4600 Add case for Arrival for completeness<commit_after>package actors

import actors.FlightMessageConversion._
import akka.actor._
import akka.event.LoggingReceive
import akka.pattern.AskableActorRef
import akka.persistence._
import akka.util.Timeout
import controllers.FlightState
import drt.shared.FlightsApi.{Flights, FlightsWithSplits}
import drt.shared.PassengerSplits.{FlightNotFound, VoyagePaxSplits}
import drt.shared.{Arrival, _}
import org.joda.time.LocalDate
import org.joda.time.format.DateTimeFormat
import passengersplits.core.PassengerInfoRouterActor.ReportVoyagePaxSplit
import server.protobuf.messages.FlightsMessage.{FlightLastKnownPaxMessage, FlightStateSnapshotMessage, FlightsMessage}
import services.SplitsProvider.SplitProvider
import services.{CSVPassengerSplitsProvider, FastTrackPercentages, SDate}

import scala.collection.immutable.Seq
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.language.postfixOps

case object GetFlights

case object GetFlightsWithSplits

class FlightsActor(crunchActorRef: ActorRef,
                   dqApiSplitsActorRef: AskableActorRef,
                   csvSplitsProvider: SplitProvider,
                   val airportConfig: AirportConfig)
  extends PersistentActor
    with ActorLogging
    with FlightState
    with DomesticPortList {
  implicit val timeout = Timeout(5 seconds)

  import SplitRatiosNs.SplitSources._

  val snapshotInterval = 20

  override def persistenceId = "flights-store"

  override protected def onRecoveryFailure(cause: Throwable, event: Option[Any]): Unit = {
    Recovery
    super.onRecoveryFailure(cause, event)
    log.error(cause, "recovery failed in flightsActors")
  }

  val receiveRecover: Receive = {
    case FlightsMessage(recoveredFlights) =>
      log.info(s"Recovering ${recoveredFlights.length} new flights")
      setFlights(recoveredFlights.map(flightMessageToApiFlight).map(f => (f.FlightID, f)).toMap)
    case SnapshotOffer(_, snapshot) =>
      val flightsFromSnapshot = snapshot match {
        case flightStateSnapshot: FlightStateSnapshotMessage =>
          log.info(s"Restoring snapshot from protobuf message")
          flightStateSnapshot.flightMessages.map(flightMessageToApiFlight).map(f => (f.FlightID, f)).toMap
        case flights: Map[Int, Any] =>
          log.info(s"Restoring snapshot from legacy ApiFlight")
          flights.mapValues {
            case a: Arrival => a
            case f: ApiFlight => ApiFlightToArrival(f)
          }
      }
{'fraction_non_alphanumeric': 0.07693825212073387, 'fraction_numerical': 0.0019727756954034327, 'mean_word_length': 4.3936170212765955, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 18, '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': '521445', 'n_tokens_mistral': 1521, 'n_tokens_neox': 1449, 'n_words': 306}