%dw 2.0
output application/dw

import dw::modules::ComplexData

var airportDetails = 
readUrl("classpath://airportInfoTiny.csv","application/csv")
groupBy $.IATA

var keysToRemove = [
  "openFlightsAirportId",
  "type",
  "source"
]

fun listKeys(anObject : Object) = 
anObject pluck $$

fun flightsJoinedWithAirportInfo(payload) =
payload.flights map ( (flight) ->
   flight ++
   {
      airportInfo:
      airportDetails[flight.destination][0]
      //--["openFlightsAirportId", "type", "source"]
      -- keysToRemove
   } 
)

fun flightsJoinedWithAirportInfoV2(payload) =
payload.flights as Array map (flight, index) -> 
//"flight$(index)": 
flight mapObject (value,key) ->
    key match {
     	case "destination" -> joinAirportDetails(flight)
      else -> { (key): value }
    }

fun joinAirportDetails(flight) =
destination: (
   {airportCode: flight.destination} 
   ++ 
   (
      filterDetailsKeys(
         airportDetails[flight.destination][0], keysToRemove
      ) 
   )
)

fun filterDetailsKeys( detailsObject, keysArray ) =
detailsObject mapObject ( (destValue, destKey) -> 
    if ( keysArray contains (destKey as String) ) {}
    else { (destKey): destValue }
)

//fun buildJsonObject( objectArray, parentTag) =
fun buildJsonObject( objectArray: Array<Object>, parentTag: String ) =
{( 
   objectArray map (element, index) -> {
      ( parentTag ++ index ): element
   }
)}

var joinedFlightsXml = ComplexData::buildXmlObject(
	flightsJoinedWithAirportInfoV2( payload ), "flight"
)

fun formatDataStructure (anyInput) =
	anyInput match {
		case is Null -> ""
		case is String -> anyInput
		case is Array -> anyInput map formatDataStructure($)
		case is Object -> 
			anyInput mapObject (v,k) -> (dw::core::Strings::dasherize(k)):formatDataStructure(v)
		case is Number -> anyInput as enrichedNumber
		else -> anyInput
	}

type enrichedNumber = String {format: "+#,##0.00;-#"}

fun deltaFlightsKpi (flightsInput:Array) =
	flightsInput reduce (flight,acc={
		delta: {count: 0, price: 0, revenue: 0}
	}) ->
		flight.airlineName match {
			case delta if(lower(delta) contains "del") -> 
				delta: {
					count: acc.delta.count + 1, 
					price: acc.delta.price + flight.price,
					revenue: acc.delta.revenue +  flight.availableSeats * flight.price
				}
			else -> acc
		}
	
---
//formatDataStructure(joinedFlightsXml)

deltaFlightsKpi(flightsJoinedWithAirportInfoV2( payload ))