import logger from '@mds-core/mds-logger' ; import { isUUID , now , ValidationError , normalizeToArray , NotFoundError , ServerError } from '@mds-core/mds-utils' ; import { isValidStop , isValidDevice , validateEvent , isValidTelemetry } from '@mds-core/mds-schema-validators' ; import db from '@mds-core/mds-db' ; import cache from '@mds-core/mds-agency-cache' ; import stream from '@mds-core/mds-stream' ; import { providerName } from '@mds-core/mds-providers' ; import { Device , VehicleEvent , Telemetry , ErrorObject , DeviceID , VEHICLE_STATUSES , EVENT_STATUS_MAP , VEHICLE_EVENT , VEHICLE_REASON , UUID } from '@mds-core/mds-types' ; import urls from 'url' ; import { parseRequest } from '@mds-core/mds-api-helpers' ; import { AgencyApiRequest , AgencyRegisterVehicleResponse , AgencyGetVehicleByIdResponse , AgencyGetVehiclesByProviderResponse , AgencyUpdateVehicleResponse , AgencySubmitVehicleEventResponse , AgencySubmitVehicleTelemetryResponse , AgencyRegisterStopResponse , AgencyReadStopsResponse , AgencyReadStopResponse } from './types' ; import { badDevice , getVehicles , lower , writeTelemetry , badEvent , badTelemetry , writeRegisterEvent , readPayload , computeCompositeVehicleData } from './utils' ; stream . initialize ( ) ; const agencyServerError = { error : 'server_error' , error_description : 'Unknown server error' } ; export const registerVehicle = async ( req , res ) => { const { body } = req ; const recorded = now ( ) ; const { provider_id } = res . locals ; const { device_id , vehicle_id , type , propulsion , year , mfgr , model } = body ; const status = VEHICLE_STATUSES . removed ; const device = { provider_id , device_id , vehicle_id , type , propulsion , year , mfgr , model , recorded , status } ; try { isValidDevice ( device ) ; } catch ( err ) { logger . info ( `Device ValidationError for ${ providerName ( provider_id ) }. Error: ${ err }` ) ; } const failure = badDevice ( device ) ; if ( failure ) { return res . status ( 400 ) . send ( failure ) ; } try { await db . writeDevice ( device ) ; try { await Promise . all ( [ cache . writeDevice ( device ) , stream . writeDevice ( device ) ] ) ; } catch ( err ) { logger . error ( 'failed to write device stream/cache' , err ) ; } logger . info ( 'new' , providerName ( res . locals . provider_id ) , 'vehicle added' , device ) ; try { await writeRegisterEvent ( device , recorded ) ; } catch ( err ) { logger . error ( 'writeRegisterEvent failure' , err ) ; } res . status ( 201 ) . send ( { } ) ; } catch ( err ) { if ( String ( err ) . includes ( 'duplicate' ) ) { res . status ( 409 ) . send ( { error : 'already_registered' , error_description : 'A vehicle with this device_id is already registered' } ) ; } else if ( String ( err ) . includes ( 'db' ) ) { logger . error ( providerName ( res . locals . provider_id ) , 'register vehicle failed:' , err ) ; res . status ( 500 ) . send ( agencyServerError ) ; } else { logger . error ( providerName ( res . locals . provider_id ) , 'register vehicle failed:' , err ) ; res . status ( 500 ) . send ( agencyServerError ) ; } } } ; export const getVehicleById = async ( req , res ) => { const { device_id } = req . params ; const { provider_id } = res . locals . scopes . includes ( 'vehicles:read' ) ? parseRequest ( req ) . query ( 'provider_id' ) : res . locals ; const payload = await readPayload ( device_id ) ; if ( ! payload . device || ( provider_id && payload . device . provider_id !== provider_id ) ) { res . status ( 404 ) . send ( { } ) ; return ; } const compositeData = computeCompositeVehicleData ( payload ) ; res . status ( 200 ) . send ( { ... compositeData } ) ; } ; export const getVehiclesByProvider = async ( req , res ) => { const PAGE_SIZE = 1000 ; const { skip = 0 , take = PAGE_SIZE } = parseRequest ( req , { parser : Number } ) . query ( 'skip' , 'take' ) ; const url = urls . format ( { protocol : req . get ( 'x-forwarded-proto' ) || req . protocol , host : req . get ( 'host' ) , pathname : req . path } ) ; const { provider_id } = res . locals . scopes . includes ( 'vehicles:read' ) ? parseRequest ( req ) . query ( 'provider_id' ) : res . locals ; try { const response = await getVehicles ( skip , take , url , req . query , provider_id ) ; return res . status ( 200 ) . send ( { ... response } ) ; } catch ( err ) { logger . error ( 'getVehicles fail' , err ) ; return res . status ( 500 ) . send ( agencyServerError ) ; } } ; export async function updateVehicleFail ( req , res , provider_id , device_id , err ) { if ( String ( err ) . includes ( 'not found' ) ) { res . status ( 404 ) . send ( { } ) ; } else if ( String ( err ) . includes ( 'invalid' ) ) { res . status ( 400 ) . send ( { error : 'bad_param' , error_description : 'Invalid parameters for vehicle were sent' } ) ; } else if ( ! provider_id ) { res . status ( 404 ) . send ( { } ) ; } else { logger . error ( providerName ( provider_id ) , `fail PUT /vehicles/${ device_id }` , req . body , err ) ; res . status ( 500 ) . send ( agencyServerError ) ; } } export const updateVehicle = async ( req , res ) => { const { device_id } = req . params ; const { vehicle_id } = req . body ; const update = { vehicle_id } ; const { provider_id } = res . locals ; try { const tempDevice = await db . readDevice ( device_id , provider_id ) ; if ( tempDevice . provider_id !== provider_id ) { await updateVehicleFail ( req , res , provider_id , device_id , 'not found' ) ; } else { const device = await db . updateDevice ( device_id , provider_id , update ) ; try { await Promise . all ( [ cache . writeDevice ( device ) , stream . writeDevice ( device ) ] ) ; } catch ( error ) { logger . warn ( `Error writing to cache/stream ${ error }` ) ; } return res . status ( 201 ) . send ( { } ) ; } } catch ( err ) { await updateVehicleFail ( req , res , provider_id , device_id , 'not found' ) ; } } ; export const submitVehicleEvent = async ( req , res ) => { const { device_id } = req . params ; const { provider_id } = res . locals ; const name = providerName ( provider_id || 'unknown' ) ; const recorded = now ( ) ; const event = { device_id : req . params . device_id , provider_id : res . locals . provider_id , event_type : lower ( req . body . event_type ) as VEHICLE_EVENT , event_type_reason : lower ( req . body . event_type_reason ) as VEHICLE_REASON , telemetry : req . body . telemetry ? { ... req . body . telemetry , provider_id : res . locals . provider_id } : null , timestamp : req . body . timestamp , trip_id : req . body . trip_id , recorded , telemetry_timestamp : undefined , service_area_id : null } ; try { validateEvent ( event ) ; } catch ( err ) { logger . info ( `Event ValidationError for ${ providerName ( provider_id ) }. Error: ${ err }` ) ; } if ( event . telemetry ) { event . telemetry_timestamp = event . telemetry . timestamp ; } async function success ( ) { function fin ( ) { res . status ( 201 ) . send ( { device_id , status : EVENT_STATUS_MAP [ event . event_type ] } ) ; } const delta = now ( ) - recorded ; if ( delta > 100 ) { logger . info ( name , 'post event took' , delta , 'ms' ) ; fin ( ) ; } else { fin ( ) ; } } async function fail ( err ) { const message = err . message || String ( err ) ; if ( message . includes ( 'duplicate' ) ) { logger . info ( name , 'duplicate event' , event . event_type ) ; res . status ( 400 ) . send ( { error : 'bad_param' , error_description : 'An event with this device_id and timestamp has already been received' } ) ; } else if ( message . includes ( 'not found' ) || message . includes ( 'unregistered' ) ) { logger . info ( name , 'event for unregistered' , event . device_id , event . event_type ) ; res . status ( 400 ) . send ( { error : 'unregistered' , error_description : 'The specified device_id has not been registered' } ) ; } else { logger . error ( 'post event fail:' , event , message ) ; res . status ( 500 ) . send ( agencyServerError ) ; } } try { const device = await db . readDevice ( event . device_id , provider_id ) ; try { await cache . readDevice ( event . device_id ) ; } catch ( err ) { try { await Promise . all ( [ cache . writeDevice ( device ) , stream . writeDevice ( device ) ] ) ; logger . info ( 'Re-adding previously deregistered device to cache' , err ) ; } catch ( error ) { logger . warn ( `Error writing to cache/stream ${ error }` ) ; } } if ( event . telemetry ) { event . telemetry . device_id = event . device_id ; } const failure = ( await badEvent ( event ) ) || ( event . telemetry ? badTelemetry ( event . telemetry ) : null ) ; if ( failure ) { logger . info ( name , 'event failure' , failure , event ) ; return res . status ( 400 ) . send ( failure ) ; } const { telemetry } = event ; if ( telemetry ) { await db . writeTelemetry ( normalizeToArray ( telemetry ) ) ; } const recorded_event = await db . writeEvent ( event ) ; try { await Promise . all ( [ cache . writeEvent ( recorded_event ) , stream . writeEvent ( recorded_event ) ] ) ; if ( telemetry ) { telemetry . recorded = recorded ; await Promise . all ( [ cache . writeTelemetry ( [ telemetry ] ) , stream . writeTelemetry ( [ telemetry ] ) ] ) ; } await success ( ) ; } catch ( err ) { logger . warn ( '/event exception cache/stream' , err ) ; await success ( ) ; } } catch ( err ) { await fail ( err ) ; } } ; export const submitVehicleTelemetry = async ( req , res ) => { const start = Date . now ( ) ; const { data } = req . body ; const { provider_id } = res . locals ; if ( ! provider_id ) { res . status ( 400 ) . send ( { error : 'bad_param' , error_description : 'Bad or missing provider_id' } ) ; return ; } const name = providerName ( provider_id ) ; const failures = [ ] ; const valid = [ ] ; const recorded = now ( ) ; const p = data . length === 1 && isUUID ( data [ 0 ] . device_id ) ? db . readDevice ( data [ 0 ] . device_id , provider_id ) : db . readDeviceIds ( provider_id ) ; try { const deviceOrDeviceIds = await p ; const deviceIds = Array . isArray ( deviceOrDeviceIds ) ? deviceOrDeviceIds : [ deviceOrDeviceIds ] ; for ( const item of data ) { const { gps } = item ; const telemetry = { device_id : item . device_id , provider_id , timestamp : item . timestamp , charge : item . charge , gps : { lat : gps . lat , lng : gps . lng , altitude : gps . altitude , heading : gps . heading , speed : gps . speed , accuracy : gps . hdop , satellites : gps . satellites } , recorded } ; try { isValidTelemetry ( telemetry ) ; } catch ( err ) { logger . info ( `Telemetry ValidationError for ${ providerName ( provider_id ) }. Error: ${ err }` ) ; } const bad_telemetry = badTelemetry ( telemetry ) ; if ( bad_telemetry ) { const msg = `bad telemetry for device_id ${ telemetry . device_id }: ${ bad_telemetry . error_description }` ; failures . push ( msg ) ; } else if ( ! deviceIds . some ( item2 => item2 . device_id === telemetry . device_id ) ) { const msg = `device_id ${ telemetry . device_id }: not found` ; failures . push ( msg ) ; } else { valid . push ( telemetry ) ; } } if ( valid . length ) { const recorded_telemetry = await writeTelemetry ( valid ) ; const delta = Date . now ( ) - start ; if ( delta > 300 ) { logger . info ( name , 'writeTelemetry' , valid . length , `(${ recorded_telemetry . length } unique)` , 'took' , delta , `ms (${ Math . round ( ( 1000 * valid . length ) / delta ) }/s)` ) ; } if ( recorded_telemetry . length ) { res . status ( 201 ) . send ( { result : `telemetry success for ${ valid . length } of ${ data . length }` , recorded : now ( ) , unique : recorded_telemetry . length , failures } ) ; } else { logger . info ( name , 'no unique telemetry in' , data . length , 'items' ) ; res . status ( 400 ) . send ( { error : 'invalid_data' , error_description : 'None of the provided data was valid' , error_details : failures } ) ; } } else { const body = `${ JSON . stringify ( req . body ) . substring ( 0 , 128 ) } ...` ; const fails = `${ JSON . stringify ( failures ) . substring ( 0 , 128 ) } ...` ; logger . info ( name , 'no valid telemetry in' , data . length , 'items:' , body , 'failures:' , fails ) ; res . status ( 400 ) . send ( { error : 'invalid_data' , error_description : 'None of the provided data was valid' , error_details : failures } ) ; } } catch ( err ) { res . status ( 500 ) . send ( { error : 'server_error' , error_description : 'None of the provided data was valid' , error_details : [ ` device_id ${ data [ 0 ] . device_id }: not found` ] } ) ; } } ; export const registerStop = async ( req , res ) => { const stop = req . body ; try { isValidStop ( stop ) ; const recorded_stop = await db . writeStop ( stop ) ; return res . status ( 201 ) . send ( { ... recorded_stop } ) ; } catch ( error ) { if ( error instanceof ValidationError ) { return res . status ( 400 ) . send ( { error } ) ; } return res . status ( 500 ) . send ( { error : new ServerError ( ) } ) ; } } ; export const readStop = async ( req , res ) => { const { stop_id } = req . params ; try { const recorded_stop = await db . readStop ( stop_id ) ; if ( ! recorded_stop ) { return res . status ( 404 ) . send ( { error : new NotFoundError ( 'Stop not found' ) } ) ; } res . status ( 200 ) . send ( { ... recorded_stop } ) ; } catch ( err ) { res . status ( 500 ) . send ( { error : new ServerError ( ) } ) ; } } ; export const readStops = async ( req , res ) => { try { const stops = await db . readStops ( ) ; if ( ! stops ) { return res . status ( 404 ) . send ( { error : new NotFoundError ( 'No stops were found' ) } ) ; } res . status ( 200 ) . send ( { stops } ) ; } catch ( err ) { return res . status ( 500 ) . send ( { error : new ServerError ( ) } ) ; } } ;