import { Event , Options , Status , Response } from '@amplitude/types' ; import { BASE_RETRY_TIMEOUT_DEPRECATED , BASE_RETRY_TIMEOUT_DEPRECATED_WARNING } from '../constants' ; import { asyncSleep , collectInvalidEventIndices , logger } from '@amplitude/utils' ; import { BaseRetryHandler } from './baseRetry' ; interface RetryMetadata { shouldRetry : boolean ; shouldReduceEventCount : boolean ; eventIndicesToRemove : number [ ] ; response : Response ; } /**
 * Converts deprecated maxRetries option to retryTimeouts
 */ function convertMaxRetries ( maxRetries ) { const retryTimeouts = [ ] ; let currentTimeout = BASE_RETRY_TIMEOUT_DEPRECATED ; for ( let i = 0 ; i < maxRetries ; i ++ ) { retryTimeouts . push ( currentTimeout ) ; currentTimeout *= 2 ; } return retryTimeouts ; } function isNodeError ( err ) { return err . code !== undefined && err . errno !== undefined && err . syscall !== undefined ; } export class RetryHandler extends BaseRetryHandler { private readonly _idToBuffer : Map < string , Map < string , Event [ ] > > = new Map < string , Map < string , Event [ ] > > ( ) ; private _eventsInRetry = 0 ; public constructor ( apiKey , options = { } ) { super ( apiKey , options ) ; if ( this . _options . maxRetries !== undefined ) { logger . warn ( BASE_RETRY_TIMEOUT_DEPRECATED_WARNING ) ; this . _options . retryTimeouts = convertMaxRetries ( this . _options . maxRetries ) ; delete this . _options . maxRetries ; } } inheritDoc public async sendEventsWithRetry ( events ) { let response = { status : Status . Unknown , statusCode : 0 } ; let eventsToSend = [ ] ; try { eventsToSend = this . _pruneEvents ( events ) ; response = await this . _transport . sendPayload ( this . _getPayload ( eventsToSend ) ) ; if ( response . status !== Status . Success ) { throw new Error ( response . status ) ; } } catch ( err ) { if ( isNodeError ( err ) ) { response = { status : Status . SystemError , statusCode : 0 , error : err , } ; } else { logger . warn ( 'Unknown error caught when sending events' ) ; logger . warn ( err ) ; } if ( this . _shouldRetryEvents ( ) ) { this . _onEventsError ( eventsToSend , response ) ; } } return response ; } private _shouldRetryEvents ( ) { if ( this . _options . retryTimeouts . length === 0 ) { return false ; } return this . _eventsInRetry < this . _options . maxCachedEvents ; } private _pruneEvents ( events ) { const prunedEvents = [ ] ; if ( Array . isArray ( events ) ) { for ( const event of events ) { const { user_id : userId = '' , device_id : deviceId = '' } = event ; if ( userId . length > 0 || deviceId . length > 0 ) { const retryBuffer = this . _getRetryBuffer ( userId , deviceId ) ; if ( retryBuffer !== null ) { retryBuffer . push ( event ) ; this . _eventsInRetry ++ ; } else { prunedEvents . push ( event ) ; } } } } return prunedEvents ; } private _getRetryBuffer ( userId , deviceId ) { const deviceToBufferMap = this . _idToBuffer . get ( userId ) ; if ( deviceToBufferMap === undefined ) { return null ; } return deviceToBufferMap . get ( deviceId ) ?? null ; } private _cleanUpBuffer ( userId , deviceId ) { const deviceToBufferMap = this . _idToBuffer . get ( userId ) ; if ( deviceToBufferMap === undefined ) { return ; } const eventsToRetry = deviceToBufferMap . get ( deviceId ) ; if ( eventsToRetry !== undefined && eventsToRetry . length === 0 ) { deviceToBufferMap . delete ( deviceId ) ; } if ( deviceToBufferMap . size === 0 ) { this . _idToBuffer . delete ( userId ) ; } } private _onEventsError ( events , response ) { let eventsToRetry = events ; if ( response . status === Status . RateLimit && response . body !== undefined ) { const { exceededDailyQuotaUsers , exceededDailyQuotaDevices } = response . body ; eventsToRetry = events . filter ( ( { user_id : userId , device_id : deviceId } ) => { return ( ! ( userId !== undefined && userId in exceededDailyQuotaUsers ) && ! ( deviceId !== undefined && deviceId in exceededDailyQuotaDevices ) ) ; } ) ; } else if ( response . status === Status . Invalid ) { if ( typeof response . body ?. missingField === 'string' || events . length === 1 ) { return ; } else if ( response . body !== undefined ) { const invalidEventIndices = new Set < number > ( collectInvalidEventIndices ( response ) ) ; eventsToRetry = events . filter ( ( _ , index ) => ! invalidEventIndices . has ( index ) ) ; } } else if ( response . status === Status . Success ) { return ; } eventsToRetry . forEach ( ( event ) => { const { user_id : userId = '' , device_id : deviceId = '' } = event ; if ( userId . length > 0 || deviceId . length > 0 ) { let deviceToBufferMap = this . _idToBuffer . get ( userId ) ; if ( deviceToBufferMap === undefined ) { deviceToBufferMap = new Map < string , Event [ ] > ( ) ; this . _idToBuffer . set ( userId , deviceToBufferMap ) ; } let retryBuffer = deviceToBufferMap . get ( deviceId ) ; if ( retryBuffer === undefined ) { retryBuffer = [ ] ; deviceToBufferMap . set ( deviceId , retryBuffer ) ; setTimeout ( ( ) => { this . _retryEventsOnLoop ( userId , deviceId ) ; } , 0 ) ; } this . _eventsInRetry ++ ; retryBuffer . push ( event ) ; } } ) ; } private async _retryEventsOnce ( userId , deviceId , eventsToRetry ) { const response = await this . _transport . sendPayload ( this . _getPayload ( eventsToRetry ) ) ; let shouldRetry = true ; let shouldReduceEventCount = false ; let eventIndicesToRemove = [ ] ; if ( response . status === Status . RateLimit ) { if ( response . body !== undefined ) { const { exceededDailyQuotaUsers , exceededDailyQuotaDevices } = response . body ; if ( deviceId in exceededDailyQuotaDevices || userId in exceededDailyQuotaUsers ) { shouldRetry = false ; } } shouldReduceEventCount = true ; } else if ( response . status === Status . PayloadTooLarge ) { shouldReduceEventCount = true ; } else if ( response . status === Status . Invalid ) { if ( eventsToRetry . length === 1 ) { shouldRetry = false ; } else { eventIndicesToRemove = collectInvalidEventIndices ( response ) ; } } else if ( response . status === Status . Success ) { shouldRetry = false ; } return { shouldRetry , shouldReduceEventCount , eventIndicesToRemove , response } ; } private async _retryEventsOnLoop ( userId , deviceId ) { const eventsBuffer = this . _getRetryBuffer ( userId , deviceId ) ; if ( eventsBuffer === null || eventsBuffer . length === 0 ) { this . _cleanUpBuffer ( userId , deviceId ) ; return ; } let eventCount = eventsBuffer . length ; for ( let numRetries = 0 ; numRetries < this . _options . retryTimeouts . length ; numRetries ++ ) { const sleepDuration = this . _options . retryTimeouts [ numRetries ] ; await asyncSleep ( sleepDuration ) ; const isLastTry = numRetries === this . _options . retryTimeouts . length ; const eventsToRetry = eventsBuffer . slice ( 0 , eventCount ) ; const { shouldRetry , shouldReduceEventCount , eventIndicesToRemove , response } = await this . _retryEventsOnce ( userId , deviceId , eventsToRetry ) ; if ( this . _options . onRetry !== null ) this . _options . onRetry ( response , numRetries , numRetries === this . _options . retryTimeouts . length - 1 ) ; if ( eventIndicesToRemove . length > 0 ) { let numEventsRemoved = 0 ; Array . from ( eventIndicesToRemove ) . reverse ( ) . forEach ( index => { if ( index < eventCount ) { eventsBuffer . splice ( index , 1 ) ; numEventsRemoved += 1 ; } } ) ; eventCount -= numEventsRemoved ; this . _eventsInRetry -= eventCount ; if ( eventCount < 1 ) { break ; } } if ( ! shouldRetry ) { break ; } if ( shouldReduceEventCount && ! isLastTry ) { eventCount = Math . max ( eventCount >> 1 , 1 ) ; } } eventsBuffer . splice ( 0 , eventCount ) ; this . _eventsInRetry -= eventCount ; setTimeout ( ( ) => { this . _retryEventsOnLoop ( userId , deviceId ) ; } , 0 ) ; } }