diff --git a/packages/analytics-browser/src/config.ts b/packages/analytics-browser/src/config.ts index cc8c9a309c..6bcf3a4575 100644 --- a/packages/analytics-browser/src/config.ts +++ b/packages/analytics-browser/src/config.ts @@ -87,6 +87,7 @@ export class BrowserConfig extends Config implements IBrowserConfig { public partnerId?: string, public plan?: Plan, public serverUrl: string = '', + public delayedEventsServerUrl?: string, public serverZone: ServerZoneType = DEFAULT_SERVER_ZONE, sessionId?: number, deferredSessionId?: number, @@ -398,6 +399,7 @@ export const useBrowserConfig = async ( options.partnerId, options.plan, options.serverUrl, + options.delayedEventsServerUrl, // Use earlyConfig.serverZone to ensure consistent serverZone earlyConfig?.serverZone ?? options.serverZone, sessionId, diff --git a/packages/analytics-browser/src/video-capture/video-capture.ts b/packages/analytics-browser/src/video-capture/video-capture.ts index 84452da63a..2f377b34dc 100644 --- a/packages/analytics-browser/src/video-capture/video-capture.ts +++ b/packages/analytics-browser/src/video-capture/video-capture.ts @@ -5,18 +5,27 @@ import { EmbeddedVideoPlayer, VideoVendor, UUID, + BaseEvent, + getHeartbeatInstance, } from '@amplitude/analytics-core'; +/** Playback states where a view session is still in progress (e.g. buffering). */ +const ACTIVE_PLAYBACK_STATES = new Set(['playing', 'waiting']); + export class VideoCapture { private videoEl: HTMLVideoElement | null = null; + private heartbeat: ReturnType; private embeddedVideoPlayer: EmbeddedVideoPlayer | null = null; private vendor?: VideoVendor; private extraEventProperties: Record = {}; - + private stopEvent: BaseEvent | null = null; private listeners: ((previousState: VideoState, nextState: VideoState) => void)[] = []; private onRemoveListeners: (() => void)[] = []; + private playId: string | null = null; - constructor(private readonly amplitude: BrowserClient) {} + constructor(private readonly amplitude: BrowserClient) { + this.heartbeat = getHeartbeatInstance(this.amplitude); + } /** * Specify a video element to capture events from @@ -67,12 +76,35 @@ export class VideoCapture { */ captureVideoStarted(): VideoCapture { this.listeners.push((previousState, nextState) => { - if (previousState.playbackState !== 'playing' && nextState.playbackState === 'playing') { - // TODO: placeholder for Heartbeat Start Event - this.amplitude.track('Video Content Started', { - ...nextState.lastEvent, - ...this.extraEventProperties, - }); + if (!ACTIVE_PLAYBACK_STATES.has(previousState.playbackState) && nextState.playbackState === 'playing') { + this.playId = UUID(); + const now = new Date().getTime(); + const startEvent: BaseEvent = { + insert_id: UUID(), + event_type: 'Video Content Started', + time: now, + event_properties: { + ...nextState.lastEvent, + ...this.parseStartEventProperties(nextState), + ...this.extraEventProperties, + play_id: this.playId, + }, + }; + this.stopEvent = { + ...startEvent, + insert_id: UUID(), + event_type: 'Video Content Stopped', + time: now + 1, + event_properties: { + ...nextState.lastEvent, + ...this.parseStopEventProperties(nextState), + ...this.extraEventProperties, + stop_reason: 'timeout', + play_id: this.playId, + }, + }; + this.heartbeat.trackNoDelay(startEvent).catch(this.stop.bind(this)); + this.heartbeat.track(this.stopEvent).catch(this.stop.bind(this)); } }); return this; @@ -84,18 +116,49 @@ export class VideoCapture { */ captureVideoStopped(): VideoCapture { this.listeners.push((previousState, nextState) => { - if (previousState.playbackState === 'playing' && nextState.playbackState !== 'playing') { - // placeholder for Heartbeat Stop Event - this.amplitude.track('Video Content Stopped', { - ...nextState.lastEvent, - watch_duration: nextState.watchTime, + // update the delayed event properties to have + // the most up-to-date values + if (this.stopEvent) { + this.stopEvent.event_properties = { + ...this.stopEvent.event_properties, + ...this.parseStopEventProperties(nextState), ...this.extraEventProperties, - }); + }; + this.stopEvent.time = new Date().getTime(); + void this.heartbeat.update(this.stopEvent); + } + if ( + ACTIVE_PLAYBACK_STATES.has(previousState.playbackState) && + !ACTIVE_PLAYBACK_STATES.has(nextState.playbackState) + ) { + this.flushStopEvent(nextState.playbackState); } }); return this; } + /** + * End the current play session by ingesting its queued delayed stop event immediately. + * + * The heartbeat is shared by every capture on the same Amplitude client, so the event is + * flushed rather than the heartbeat stopped, which would discard other captures' events. + * Flushing also drops the event from the heartbeat queue once ingested, so the interval + * winds down on its own. No-op when no play session is in progress. + */ + private flushStopEvent(stopReason: string) { + const stopEvent = this.stopEvent; + if (!stopEvent) { + return; + } + // the next play queues a fresh delayed stop event + this.stopEvent = null; + stopEvent.event_properties = { + ...stopEvent.event_properties, + stop_reason: stopReason, + }; + this.heartbeat.trackNoDelay(stopEvent).catch(this.stop.bind(this)); + } + // Placeholder: may need a generic state change listener to capture unusual events or to have // more control over the event tracking. // withStateChangeListener(listener: (previousState: VideoState, nextState: VideoState) => void): VideoCapture { @@ -132,9 +195,33 @@ export class VideoCapture { return this; } + /** + * Stop capturing analytics events for the video element. + * + * Observers are detached first so no playback state change can race with the final + * event, then any in-progress play session is closed out. + */ stop() { this.onRemoveListeners.forEach((listener) => listener()); this.onRemoveListeners = []; + this.flushStopEvent('untracked'); + } + + parseStartEventProperties(nextState: VideoState): Record { + return { + duration: nextState.lastEvent?.duration ?? 0, + start_time: nextState.lastEvent?.start_time ?? 0, + position: nextState.position ?? 0, + }; + } + + parseStopEventProperties(nextState: VideoState): Record { + const percentCompleted = ((nextState.position ?? 0) / (nextState.lastEvent?.duration ?? 0)) * 100; + return { + ...this.parseStartEventProperties(nextState), + watch_duration: nextState.watchTime ?? 0, + percent_completed: percentCompleted || 0, + }; } } diff --git a/packages/analytics-browser/test/config.test.ts b/packages/analytics-browser/test/config.test.ts index 87129e6a03..9754f7fb58 100644 --- a/packages/analytics-browser/test/config.test.ts +++ b/packages/analytics-browser/test/config.test.ts @@ -157,10 +157,18 @@ describe('config', () => { }, topLevelDomain: '.amplitude.com', enableRequestBodyCompression: false, + delayedEventsServerUrl: undefined, }); expect(getTopLevelDomain).toHaveBeenCalledTimes(1); }); + test('should pass delayedEventsServerUrl through', async () => { + jest.spyOn(Config, 'getTopLevelDomain').mockResolvedValueOnce('.amplitude.com'); + const delayedEventsServerUrl = 'https://example.com/2/httpapi/delayed'; + const config = await Config.useBrowserConfig(apiKey, { delayedEventsServerUrl }, new AmplitudeBrowser()); + expect(config.delayedEventsServerUrl).toBe(delayedEventsServerUrl); + }); + test('should fall back to memoryStorage when storageProvider is not enabled', async () => { const localStorageIsEnabledSpy = jest .spyOn(LocalStorageModule.LocalStorage.prototype, 'isEnabled') @@ -285,6 +293,7 @@ describe('config', () => { }, topLevelDomain: 'amplitude.com', enableRequestBodyCompression: false, + delayedEventsServerUrl: undefined, }); }); }); diff --git a/packages/analytics-browser/test/video-capture/video-capture.test.ts b/packages/analytics-browser/test/video-capture/video-capture.test.ts index 797b1846b9..b8b6ad4d58 100644 --- a/packages/analytics-browser/test/video-capture/video-capture.test.ts +++ b/packages/analytics-browser/test/video-capture/video-capture.test.ts @@ -1,30 +1,47 @@ -/* eslint-disable @typescript-eslint/no-non-null-assertion, @typescript-eslint/unbound-method -- jest expectations */ +/* eslint-disable @typescript-eslint/no-non-null-assertion, @typescript-eslint/unbound-method, @typescript-eslint/no-unsafe-return -- jest expectations */ import { AmplitudeBrowser } from '@amplitude/analytics-browser'; import { EmbeddedVideoPlayer, VideoState } from '@amplitude/analytics-core'; import { VideoCapture, trackVideo } from '../../src/video-capture/video-capture'; import { currentVideoObserver, resetMockVideoObserver } from './mock-video-observer'; +const mockGetHeartbeatInstance = jest.fn(); + jest.mock('@amplitude/analytics-core', () => { const actual = jest.requireActual('@amplitude/analytics-core'); const { MockVideoObserver } = jest.requireActual('./mock-video-observer'); return { ...actual, VideoObserver: MockVideoObserver, + getHeartbeatInstance: (client: Parameters[0]) => + mockGetHeartbeatInstance(client), }; }); describe('VideoCapture', () => { let mockAmplitude: AmplitudeBrowser; + /** Flush resetHeartbeat's setTimeout(0) macrotask before asserting track calls. */ + async function flushHeartbeat() { + await jest.advanceTimersByTimeAsync(0); + } + beforeEach(() => { + jest.useFakeTimers(); resetMockVideoObserver(); + mockGetHeartbeatInstance.mockImplementation( + jest.requireActual('@amplitude/analytics-core').getHeartbeatInstance, + ); mockAmplitude = { - track: jest.fn(), + track: jest.fn().mockReturnValue({ promise: Promise.resolve({ event: {}, code: 200, message: 'success' }) }), } as unknown as AmplitudeBrowser; }); + afterEach(() => { + jest.useRealTimers(); + }); + describe('kitchen sink', () => { - it('should track start and stop events', () => { + it('should track start and stop events', async () => { const capture = new VideoCapture(mockAmplitude) .withVideoElement(document.createElement('video')) .captureVideoStarted() @@ -39,23 +56,71 @@ describe('VideoCapture', () => { let previousState: VideoState = { playbackState: 'paused', lastEvent: undefined }; let nextState: VideoState = { playbackState: 'playing', lastEvent: { duration: 10, last_position: undefined } }; currentVideoObserver!.emitStateChange(previousState, nextState); - expect(mockAmplitude.track).toHaveBeenCalledWith('Video Content Started', { - duration: 10, - hello: 'world', - number: 123, - }); + await flushHeartbeat(); + expect(mockAmplitude.track).toHaveBeenNthCalledWith( + 1, + 'Video Content Started', + { + duration: 10, + hello: 'world', + number: 123, + play_id: expect.any(String), + position: 0, + start_time: 0, + }, + { + delay: { id: expect.any(String) }, + insert_id: expect.any(String), + time: expect.any(Number), + }, + ); + expect(mockAmplitude.track).toHaveBeenNthCalledWith( + 2, + 'Video Content Stopped', + { + duration: 10, + hello: 'world', + number: 123, + play_id: expect.any(String), + position: 0, + start_time: 0, + watch_duration: 0, + percent_completed: 0, + stop_reason: 'timeout', + }, + { + delay: { id: expect.any(String), timeout: 3_600_000 }, + insert_id: expect.any(String), + time: expect.any(Number), + }, + ); // mock a pause event previousState = nextState; - nextState = { playbackState: 'paused', lastEvent: { duration: 10, last_position: 5 } }; + nextState = { playbackState: 'paused', lastEvent: { duration: 10, last_position: 5 }, position: 5 }; currentVideoObserver!.emitStateChange(previousState, nextState); - expect(mockAmplitude.track).toHaveBeenCalledWith('Video Content Stopped', { - duration: 10, - last_position: 5, - hello: 'world', - number: 123, - }); - expect(mockAmplitude.track).toHaveBeenCalledTimes(2); + await flushHeartbeat(); + expect(mockAmplitude.track).toHaveBeenNthCalledWith( + 3, + 'Video Content Stopped', + { + duration: 10, + hello: 'world', + number: 123, + play_id: expect.any(String), + position: 5, + start_time: 0, + watch_duration: 0, + percent_completed: 50, + stop_reason: 'paused', + }, + { + delay: { id: expect.any(String) }, + insert_id: expect.any(String), + time: expect.any(Number), + }, + ); + expect(mockAmplitude.track).toHaveBeenCalledTimes(3); // stop the capture capture.stop(); @@ -66,7 +131,7 @@ describe('VideoCapture', () => { currentVideoObserver!.emitStateChange(previousState, nextState); // assert that the track method was not called again - expect(mockAmplitude.track).toHaveBeenCalledTimes(2); + expect(mockAmplitude.track).toHaveBeenCalledTimes(3); }); }); @@ -107,7 +172,7 @@ describe('VideoCapture', () => { beforeEach(() => { resetMockVideoObserver(); }); - it('should capture start and stop events', () => { + it('should capture start and stop events', async () => { const stopVideoCapture = trackVideo(mockAmplitude, document.createElement('video'), { vendor: 'mux', extraEventProperties: { hello: 'world', number: 123 }, @@ -117,32 +182,60 @@ describe('VideoCapture', () => { { playbackState: 'paused', lastEvent: undefined }, { playbackState: 'playing', lastEvent: { duration: 10, last_position: undefined } }, ); - expect(mockAmplitude.track).toHaveBeenCalledWith('Video Content Started', { - duration: 10, - hello: 'world', - number: 123, - view_session_id: expect.any(String), - }); + await flushHeartbeat(); + expect(mockAmplitude.track).toHaveBeenNthCalledWith( + 1, + 'Video Content Started', + { + duration: 10, + hello: 'world', + number: 123, + play_id: expect.any(String), + position: 0, + start_time: 0, + view_session_id: expect.any(String), + }, + { + delay: { id: expect.any(String) }, + insert_id: expect.any(String), + time: expect.any(Number), + }, + ); currentVideoObserver!.emitStateChange( { playbackState: 'playing', lastEvent: { duration: 10, last_position: undefined } }, - { playbackState: 'paused', lastEvent: { duration: 10, last_position: 5 } }, + { playbackState: 'paused', lastEvent: { duration: 10, last_position: 5 }, position: 5 }, + ); + await flushHeartbeat(); + expect(mockAmplitude.track).toHaveBeenNthCalledWith( + 3, + 'Video Content Stopped', + { + duration: 10, + hello: 'world', + number: 123, + play_id: expect.any(String), + position: 5, + start_time: 0, + watch_duration: 0, + percent_completed: 50, + stop_reason: 'paused', + view_session_id: expect.any(String), + }, + { + delay: { id: expect.any(String) }, + insert_id: expect.any(String), + time: expect.any(Number), + }, ); - expect(mockAmplitude.track).toHaveBeenCalledWith('Video Content Stopped', { - duration: 10, - last_position: 5, - hello: 'world', - number: 123, - view_session_id: expect.any(String), - }); typeof stopVideoCapture === 'function' && stopVideoCapture(); currentVideoObserver!.emitStateChange( { playbackState: 'paused', lastEvent: { duration: 10, last_position: 5 } }, { playbackState: 'playing', lastEvent: { duration: 10, last_position: undefined } }, ); - expect(mockAmplitude.track).toHaveBeenCalledTimes(2); + expect(mockAmplitude.track).toHaveBeenCalledTimes(3); }); - it('should capture start and stop events with embedded video player', () => { + it('should capture start and stop events with embedded video player', async () => { const stopVideoCapture = trackVideo(mockAmplitude, { onPlay: jest.fn(), onPause: jest.fn(), @@ -153,19 +246,47 @@ describe('VideoCapture', () => { { playbackState: 'paused', lastEvent: undefined }, { playbackState: 'playing', lastEvent: { duration: 10, last_position: undefined } }, ); - expect(mockAmplitude.track).toHaveBeenCalledWith('Video Content Started', { - duration: 10, - view_session_id: expect.any(String), - }); + await flushHeartbeat(); + expect(mockAmplitude.track).toHaveBeenNthCalledWith( + 1, + 'Video Content Started', + { + duration: 10, + play_id: expect.any(String), + position: 0, + start_time: 0, + view_session_id: expect.any(String), + }, + { + delay: { id: expect.any(String) }, + insert_id: expect.any(String), + time: expect.any(Number), + }, + ); currentVideoObserver!.emitStateChange( { playbackState: 'playing', lastEvent: { duration: 10, last_position: undefined } }, - { playbackState: 'paused', lastEvent: { duration: 10, last_position: 5 } }, + { playbackState: 'paused', lastEvent: { duration: 10, last_position: 5 }, position: 5 }, + ); + await flushHeartbeat(); + expect(mockAmplitude.track).toHaveBeenNthCalledWith( + 3, + 'Video Content Stopped', + { + duration: 10, + play_id: expect.any(String), + position: 5, + start_time: 0, + watch_duration: 0, + percent_completed: 50, + stop_reason: 'paused', + view_session_id: expect.any(String), + }, + { + delay: { id: expect.any(String) }, + insert_id: expect.any(String), + time: expect.any(Number), + }, ); - expect(mockAmplitude.track).toHaveBeenCalledWith('Video Content Stopped', { - duration: 10, - last_position: 5, - view_session_id: expect.any(String), - }); typeof stopVideoCapture === 'function' && stopVideoCapture(); }); @@ -174,4 +295,338 @@ describe('VideoCapture', () => { expect(stopVideoCapture).toBeInstanceOf(Error); }); }); + + describe('buffering (waiting state)', () => { + const playingState: VideoState = { + playbackState: 'playing', + lastEvent: { duration: 10, last_position: 0 }, + position: 0, + watchTime: 5, + }; + const waitingState: VideoState = { + playbackState: 'waiting', + lastEvent: { duration: 10, last_position: 5 }, + position: 5, + watchTime: 5, + }; + const pausedState: VideoState = { + playbackState: 'paused', + lastEvent: { duration: 10, last_position: 5 }, + position: 5, + watchTime: 5, + }; + + it('should not split the play session across buffering', async () => { + new VideoCapture(mockAmplitude) + .withVideoElement(document.createElement('video')) + .captureVideoStarted() + .captureVideoStopped() + .start(); + + currentVideoObserver!.emitStateChange({ playbackState: 'paused', lastEvent: undefined }, playingState); + await flushHeartbeat(); + + const playId = (mockAmplitude.track as jest.Mock).mock.calls[0][1].play_id; + + currentVideoObserver!.emitStateChange(playingState, waitingState); + await flushHeartbeat(); + expect(mockAmplitude.track).toHaveBeenCalledTimes(2); + + currentVideoObserver!.emitStateChange(waitingState, { + ...playingState, + position: 6, + watchTime: 6, + }); + await flushHeartbeat(); + expect(mockAmplitude.track).toHaveBeenCalledTimes(2); + expect((mockAmplitude.track as jest.Mock).mock.calls.every((call) => call[1].play_id === playId)).toBe(true); + + currentVideoObserver!.emitStateChange({ ...playingState, position: 6, watchTime: 6 }, pausedState); + await flushHeartbeat(); + expect(mockAmplitude.track).toHaveBeenCalledTimes(3); + expect(mockAmplitude.track).toHaveBeenNthCalledWith( + 3, + 'Video Content Stopped', + expect.objectContaining({ + play_id: playId, + stop_reason: 'paused', + }), + expect.any(Object), + ); + }); + + it('should heartbeat the delayed stop event with the latest playback progress', async () => { + new VideoCapture(mockAmplitude) + .withVideoElement(document.createElement('video')) + .captureVideoStarted() + .captureVideoStopped() + .start(); + + currentVideoObserver!.emitStateChange({ playbackState: 'paused', lastEvent: undefined }, playingState); + await flushHeartbeat(); + + currentVideoObserver!.emitStateChange(playingState, { + ...playingState, + position: 8, + watchTime: 8, + }); + jest.clearAllMocks(); + + // the delayed stop event is re-sent on the next heartbeat + await jest.advanceTimersByTimeAsync(60_000); + expect(mockAmplitude.track).toHaveBeenCalledTimes(1); + expect(mockAmplitude.track).toHaveBeenCalledWith( + 'Video Content Stopped', + expect.objectContaining({ + position: 8, + watch_duration: 8, + percent_completed: 80, + stop_reason: 'timeout', + }), + expect.objectContaining({ delay: { id: expect.any(String), timeout: 3_600_000 } }), + ); + }); + }); + + describe('stops capturing when track fails', () => { + const playingState: VideoState = { + playbackState: 'playing', + lastEvent: { duration: 10, last_position: undefined }, + }; + const pausedState: VideoState = { playbackState: 'paused', lastEvent: undefined }; + + /** The "Video Content Stopped" event flushed by stop(). */ + const untrackedStopEvent = expect.objectContaining({ + event_type: 'Video Content Stopped', + event_properties: expect.objectContaining({ stop_reason: 'untracked' }), + }); + + let track: jest.Mock; + let trackNoDelay: jest.Mock; + let capture: VideoCapture; + + beforeEach(() => { + track = jest.fn().mockResolvedValue({ code: 200, event: {} }); + trackNoDelay = jest.fn().mockResolvedValue({ code: 200, event: {} }); + mockGetHeartbeatInstance.mockReturnValue({ + track, + trackNoDelay, + stop: jest.fn(), + update: jest.fn(), + }); + capture = new VideoCapture(mockAmplitude) + .withVideoElement(document.createElement('video')) + .captureVideoStarted() + .start(); + }); + + afterEach(() => { + capture.stop(); + }); + + it('should stop when trackNoDelay rejects on video start', async () => { + trackNoDelay.mockRejectedValue(new Error('trackNoDelay failed')); + + currentVideoObserver!.emitStateChange(pausedState, playingState); + await jest.advanceTimersByTimeAsync(0); + + expect(trackNoDelay).toHaveBeenCalledWith(untrackedStopEvent); + + // the observer is detached, so no further events are captured + trackNoDelay.mockClear(); + track.mockClear(); + currentVideoObserver!.emitStateChange(pausedState, playingState); + expect(trackNoDelay).not.toHaveBeenCalled(); + expect(track).not.toHaveBeenCalled(); + }); + + it('should stop when track rejects on video start', async () => { + track.mockRejectedValue(new Error('track failed')); + + currentVideoObserver!.emitStateChange(pausedState, playingState); + await jest.advanceTimersByTimeAsync(0); + + expect(trackNoDelay).toHaveBeenCalledWith(untrackedStopEvent); + }); + + it('should stop when trackNoDelay rejects on video stop', async () => { + capture.stop(); + trackNoDelay + .mockResolvedValueOnce({ code: 200, event: {} }) + .mockRejectedValueOnce(new Error('trackNoDelay failed')); + capture = new VideoCapture(mockAmplitude) + .withVideoElement(document.createElement('video')) + .captureVideoStarted() + .captureVideoStopped() + .start(); + + currentVideoObserver!.emitStateChange(pausedState, playingState); + await jest.advanceTimersByTimeAsync(0); + currentVideoObserver!.emitStateChange(playingState, { + playbackState: 'paused', + lastEvent: { duration: 10, last_position: 5 }, + position: 5, + }); + await jest.advanceTimersByTimeAsync(0); + + // the stop event was already flushed with stop_reason "paused", so tearing down + // the capture must not send it a second time + expect(trackNoDelay).toHaveBeenCalledTimes(2); + expect(trackNoDelay).not.toHaveBeenCalledWith(untrackedStopEvent); + }); + }); + + describe('stop()', () => { + const idleState: VideoState = { playbackState: 'paused', lastEvent: undefined }; + const playingState: VideoState = { + playbackState: 'playing', + lastEvent: { duration: 10, last_position: 0 }, + position: 4, + watchTime: 4, + }; + + function startCapture(extraEventProperties: Record = {}) { + const capture = new VideoCapture(mockAmplitude) + .withVideoElement(document.createElement('video')) + .withExtraEventProperties(extraEventProperties) + .captureVideoStarted() + .captureVideoStopped() + .start(); + return { capture, observer: currentVideoObserver! }; + } + + it('should flush the delayed stop event when stopped mid-play', async () => { + const { capture, observer } = startCapture(); + observer.emitStateChange(idleState, playingState); + await flushHeartbeat(); + jest.clearAllMocks(); + + capture.stop(); + await flushHeartbeat(); + + expect(mockAmplitude.track).toHaveBeenCalledTimes(1); + expect(mockAmplitude.track).toHaveBeenCalledWith( + 'Video Content Stopped', + expect.objectContaining({ stop_reason: 'untracked', position: 4, watch_duration: 4 }), + expect.objectContaining({ delay: { id: expect.any(String) } }), + ); + + // the flushed event is ingested, so it is no longer heartbeated + jest.clearAllMocks(); + await jest.advanceTimersByTimeAsync(60_000); + expect(mockAmplitude.track).not.toHaveBeenCalled(); + }); + + it('should not send a stop event when playback already stopped', async () => { + const { capture, observer } = startCapture(); + observer.emitStateChange(idleState, playingState); + await flushHeartbeat(); + observer.emitStateChange(playingState, { ...playingState, playbackState: 'paused' }); + await flushHeartbeat(); + jest.clearAllMocks(); + + capture.stop(); + await flushHeartbeat(); + + expect(mockAmplitude.track).not.toHaveBeenCalled(); + }); + + it('should be safe to call multiple times', async () => { + const { capture, observer } = startCapture(); + observer.emitStateChange(idleState, playingState); + await flushHeartbeat(); + jest.clearAllMocks(); + + capture.stop(); + capture.stop(); + await flushHeartbeat(); + + expect(mockAmplitude.track).toHaveBeenCalledTimes(1); + }); + + it('should keep delayed events queued by other captures on the same client', async () => { + const first = startCapture({ video: 'first' }); + const second = startCapture({ video: 'second' }); + first.observer.emitStateChange(idleState, playingState); + second.observer.emitStateChange(idleState, playingState); + await flushHeartbeat(); + + first.capture.stop(); + await flushHeartbeat(); + jest.clearAllMocks(); + + // the second capture's delayed stop event is still heartbeated + await jest.advanceTimersByTimeAsync(60_000); + expect(mockAmplitude.track).toHaveBeenCalledTimes(1); + expect(mockAmplitude.track).toHaveBeenCalledWith( + 'Video Content Stopped', + expect.objectContaining({ video: 'second', stop_reason: 'timeout' }), + expect.objectContaining({ delay: { id: expect.any(String), timeout: 3_600_000 } }), + ); + }); + }); + + describe('parseStartEventProperties()', () => { + it('should parse start event properties', () => { + const capture = new VideoCapture(mockAmplitude); + expect( + capture.parseStartEventProperties({ + playbackState: 'playing', + lastEvent: { duration: 10, start_time: 2, last_position: 5 }, + position: 5, + }), + ).toEqual({ + duration: 10, + start_time: 2, + position: 5, + }); + }); + + it('should parse start event properties with empty lastEvent', () => { + const capture = new VideoCapture(mockAmplitude); + expect( + capture.parseStartEventProperties({ + playbackState: 'playing', + }), + ).toEqual({ + duration: 0, + start_time: 0, + position: 0, + }); + }); + }); + + describe('parseStopEventProperties()', () => { + it('should parse stop event properties', () => { + const capture = new VideoCapture(mockAmplitude); + expect( + capture.parseStopEventProperties({ + playbackState: 'paused', + lastEvent: { duration: 10, start_time: 2, last_position: 5 }, + position: 5, + watchTime: 30, + }), + ).toEqual({ + duration: 10, + start_time: 2, + position: 5, + watch_duration: 30, + percent_completed: 50, + }); + }); + + it('should parse stop event properties with empty lastEvent', () => { + const capture = new VideoCapture(mockAmplitude); + const properties = capture.parseStopEventProperties({ + playbackState: 'paused', + }); + expect(properties).toEqual({ + duration: 0, + start_time: 0, + position: 0, + watch_duration: 0, + percent_completed: 0, + }); + }); + }); }); diff --git a/packages/analytics-core/src/config.ts b/packages/analytics-core/src/config.ts index 7b747f10ac..c246deb0ae 100644 --- a/packages/analytics-core/src/config.ts +++ b/packages/analytics-core/src/config.ts @@ -43,6 +43,7 @@ export class Config implements IConfig { plan?: Plan; ingestionMetadata?: IngestionMetadata; serverUrl: string | undefined; + delayedEventsServerUrl?: string; serverZone?: ServerZoneType; transportProvider: Transport; storageProvider?: Storage; @@ -72,6 +73,7 @@ export class Config implements IConfig { this.offline = options.offline !== undefined ? options.offline : defaultConfig.offline; this.optOut = options.optOut ?? defaultConfig.optOut; this.serverUrl = options.serverUrl; + this.delayedEventsServerUrl = options.delayedEventsServerUrl; this.serverZone = options.serverZone || defaultConfig.serverZone; this.storageProvider = options.storageProvider; this.transportProvider = options.transportProvider; diff --git a/packages/analytics-core/src/observers/video.ts b/packages/analytics-core/src/observers/video.ts index 24d36f90e4..e5f3b7a701 100644 --- a/packages/analytics-core/src/observers/video.ts +++ b/packages/analytics-core/src/observers/video.ts @@ -10,7 +10,7 @@ import type { export type { Vendor }; -type PlaybackState = 'playing' | 'paused' | 'ended' | 'error' | 'seeking'; +type PlaybackState = 'playing' | 'paused' | 'ended' | 'error' | 'seeking' | 'waiting'; export type State = { playbackState: PlaybackState; @@ -33,6 +33,7 @@ export class VideoObserver { playbackState: 'paused', }; + private waitingInterval: ReturnType | null = null; private untrack: () => void; private onStateChange: (previousState: State, nextState: State) => void; private handler: VideoHandler = { @@ -86,6 +87,7 @@ export class VideoObserver { } private updateStateWithError(error: string) { + this.clearWaitingInterval(); const previousState = this.state; const nextState: State = { ...previousState, @@ -103,11 +105,37 @@ export class VideoObserver { position: event.last_position, }; this.updateState(nextState); + + this.clearWaitingInterval(); + if (playbackState !== 'playing') { + return; + } + // if it's in a playing state, but the playhead is not moving, + // then transition playback from 'playing' to 'waiting' + // (the "waiting" listener isn't reliable, so this is a fallback) + let prevPosition: number | null | undefined = this.state.position; + this.waitingInterval = setInterval(() => { + const position = this.state.position; + if (this.state.playbackState === 'playing' && typeof prevPosition === 'number' && prevPosition === position) { + // no new player event backs this transition, so carry the current position and metadata over + this.updateState({ ...this.state, playbackState: 'waiting' }); + } + prevPosition = position; + }, 1_000); + } + + private clearWaitingInterval() { + if (this.waitingInterval) { + clearInterval(this.waitingInterval); + this.waitingInterval = null; + } } private updateTime(event: TimeUpdateEvent) { const lastVideoEvent = this.state.lastEvent; - if (!lastVideoEvent || this.state.playbackState !== 'playing') { + const isWaiting = this.state.playbackState === 'waiting'; + const isPlaying = this.state.playbackState === 'playing'; + if (!lastVideoEvent || (!isPlaying && !isWaiting)) { return; } const isSeeking = event.isSeeking || this.state.isSeeking; @@ -123,6 +151,7 @@ export class VideoObserver { const timeDelta = nextPosition - lastPosition; const nextState: State = { ...this.state, + playbackState: 'playing', position: nextPosition, watchTime: (this.state.watchTime ?? 0) + timeDelta, }; @@ -136,6 +165,7 @@ export class VideoObserver { } destroy() { + this.clearWaitingInterval(); this.untrack(); } } diff --git a/packages/analytics-core/src/plugins/destination.ts b/packages/analytics-core/src/plugins/destination.ts index bcf3306e66..204979099b 100644 --- a/packages/analytics-core/src/plugins/destination.ts +++ b/packages/analytics-core/src/plugins/destination.ts @@ -1,5 +1,6 @@ import { DestinationPlugin } from '../types/plugin'; import { Event } from '../types/event/event'; +import { Delay } from '../types/event/base-event'; import { Result } from '../types/result'; import { Status } from '../types/status'; import { @@ -32,14 +33,18 @@ import { EventCallback } from '../types/event-callback'; import { IDiagnosticsClient } from '../diagnostics/diagnostics-client'; import { isSuccessStatusCode } from '../utils/status-code'; import { getStacktrace } from '../utils/debug'; +import { DelayedPayload, Payload } from '../types/payload'; export interface Context { event: Event; attempts: number; callback: EventCallback; timeout: number; + delay?: Delay; } +type DelayedEventsById = Record; + const DEFAULT_AMPLITUDE_SERVER_URLS = new Set([ AMPLITUDE_SERVER_URL, EU_AMPLITUDE_SERVER_URL, @@ -95,6 +100,7 @@ export class Destination implements DestinationPlugin { flushId: ReturnType | null = null; queue: Context[] = []; diagnosticsClient: IDiagnosticsClient | undefined; + inFlightDelayedEvents: Record = {}; constructor(context?: { diagnosticsClient: IDiagnosticsClient }) { this.diagnosticsClient = context?.diagnosticsClient; @@ -125,12 +131,50 @@ export class Destination implements DestinationPlugin { callback: (result: Result) => resolve(result), timeout: 0, }; + this.removeStaleDelayedEvents(event); this.queue.push(context); this.schedule(this.config.flushIntervalMillis); this.saveEvents(); }); } + /** + * If a stale delayed event is sitting in the queue, and it is not in flight, + * remove it and resolve as "stale" with status code 0. + * + * If this delayed event is already in flight, mark it as fresh, so that + * when it completes, it won't clean-up the updated event. + * @param incomingEvent { Event } the new event to check old events against + * @returns void + */ + private removeStaleDelayedEvents(incomingEvent: Event) { + try { + if (!incomingEvent.delay?.id) { + return; + } + /* istanbul ignore next */ + this.queue = this.queue.filter((context) => { + if ( + incomingEvent.delay && + context.event.delay && + context.event.delay.id === incomingEvent.delay.id && + context.event.insert_id === incomingEvent.insert_id + ) { + if (this.inFlightDelayedEvents[incomingEvent.delay.id]) { + incomingEvent.delay.isFresh = true; + return true; + } + context.callback(buildResult(context.event, 0, 'Stale event overwritten')); + return false; + } + return true; + }); + /* istanbul ignore next */ + } catch (e) { + // swallow error + } + } + removeEventsExceedFlushMaxRetries(list: Context[]) { return list.filter((context) => { context.attempts += 1; @@ -198,17 +242,35 @@ export class Destination implements DestinationPlugin { this.resetSchedule(); const list: Context[] = []; + const delayed: DelayedEventsById = {}; const later: Context[] = []; - this.queue.forEach((context) => (context.timeout === 0 ? list.push(context) : later.push(context))); + this.queue.forEach((context) => { + if (context.timeout !== 0) { + later.push(context); + } else if (context.event.delay?.id) { + const delay = context.event.delay; + delayed[delay.id] = delayed[delay.id] || []; + delayed[delay.id].push(context); + } else { + list.push(context); + } + }); const batches = chunk(list, this.config.flushQueueSize); // Promise.all() doesn't guarantee resolve order. // Sequentially resolve to make sure backend receives events in order - await batches.reduce(async (promise, batch) => { + const regularEventBatch = batches.reduce(async (promise, batch) => { await promise; return await this.send(batch, useRetry); }, Promise.resolve()); + const eventPromises = [regularEventBatch]; + + if (Object.keys(delayed).length > 0) { + eventPromises.push(this.sendDelayedEvents(delayed, useRetry)); + } + + await Promise.all(eventPromises); // Mark current flush is done this.flushId = null; @@ -216,32 +278,84 @@ export class Destination implements DestinationPlugin { this.scheduleEvents(this.queue); } - async send(list: Context[], useRetry = true) { + sendDelayedEvents(delayed: DelayedEventsById, useRetry: boolean) { + const eventPromises = []; + try { + for (const [delayId, contexts] of Object.entries(delayed)) { + this.inFlightDelayedEvents[delayId] = true; + const delayedEventsSend = this.send(contexts, useRetry, true); + eventPromises.push(delayedEventsSend); + delayedEventsSend.finally(() => { + delete this.inFlightDelayedEvents[delayId]; + }); + } + } catch (e) { + // swallow error + } + return eventPromises.reduce(async (promise, batch) => { + await promise; + return await batch; + }, Promise.resolve()); + } + + translatePayloadToDelayedPayload(payload: Payload & Partial, list: Context[]): void { + const delayedEvents: Event[] = []; + const instantEvents: Event[] = []; + const delayedContexts: Context[] = []; + const instantContexts: Context[] = []; + + list.forEach((context, index) => { + if (context.event.delay!.timeout) { + delayedEvents.push(payload.events[index]); + delayedContexts.push(context); + } else { + instantEvents.push(payload.events[index]); + instantContexts.push(context); + } + }); + + list.splice(0, list.length, ...delayedContexts, ...instantContexts); + + /* istanbul ignore next */ + payload.timeout = delayedContexts[0]?.event.delay!.timeout ?? 0; + payload.id = list[0].event.delay!.id; + payload.events = delayedEvents; + payload.instant_events = instantEvents; + } + + async send(list: Context[], useRetry = true, delay?: boolean) { if (!this.config.apiKey) { return this.fulfillRequest(list, 400, MISSING_API_KEY_MESSAGE); } - const payload = { + const payload: Payload = { api_key: this.config.apiKey, events: list.map((context) => { // eslint-disable-next-line @typescript-eslint/no-unused-vars - const { extra, ...eventWithoutExtra } = context.event; + const { extra, delay, ...eventWithoutExtra } = context.event; return eventWithoutExtra; }), options: { min_id_length: this.config.minIdLength, }, client_upload_time: new Date().toISOString(), - request_metadata: this.config.requestMetadata, + request_metadata: delay ? undefined : this.config.requestMetadata, }; - this.config.requestMetadata = new RequestMetadata(); + if (!delay) { + this.config.requestMetadata = new RequestMetadata(); + } try { - const { serverUrl } = createServerConfig(this.config.serverUrl, this.config.serverZone, this.config.useBatch); - const shouldCompressUploadBody = shouldCompressUploadBodyForRequest( + let { serverUrl } = createServerConfig(this.config.serverUrl, this.config.serverZone, this.config.useBatch); + let shouldCompressUploadBody = shouldCompressUploadBodyForRequest( serverUrl, this.config.enableRequestBodyCompression, ); + if (delay) { + serverUrl = this.config.delayedEventsServerUrl || `${serverUrl}/delayed`; + this.translatePayloadToDelayedPayload(payload, list); + shouldCompressUploadBody = false; // delayed events doesn't support compression + } const res = await this.config.transportProvider.send(serverUrl, payload, shouldCompressUploadBody); if (res === null) { this.fulfillRequest(list, 0, UNEXPECTED_ERROR_MESSAGE); @@ -437,10 +551,21 @@ export class Destination implements DestinationPlugin { * This is called on response comes back for a request */ removeEvents(eventsToRemove: Context[]) { + const insertIdsBeingRemoved = new Set(eventsToRemove.map((context) => context.event.insert_id)); + this.queue = this.queue.filter( - (queuedContext) => !eventsToRemove.some((context) => context.event.insert_id === queuedContext.event.insert_id), + (queuedContext) => + !eventsToRemove.some( + (context) => context.event.insert_id === queuedContext.event.insert_id && !queuedContext.event.delay?.isFresh, + ), ); + this.queue.forEach((context) => { + if (context.event.delay?.isFresh && insertIdsBeingRemoved.has(context.event.insert_id)) { + delete context.event.delay.isFresh; + } + }); + this.saveEvents(); } } diff --git a/packages/analytics-core/src/types/config/core-config.ts b/packages/analytics-core/src/types/config/core-config.ts index 1bf6be0233..cfcc6ff27e 100644 --- a/packages/analytics-core/src/types/config/core-config.ts +++ b/packages/analytics-core/src/types/config/core-config.ts @@ -76,6 +76,11 @@ export interface IConfig { * The URL where events are upload to. */ serverUrl?: string; + /** + * The URL where delayed events are uploaded to. + * @experimental this is experimental and subject to change. + */ + delayedEventsServerUrl?: string; /** * The Amplitude server zone. * Set this to EU for Amplitude projects created in EU data center. diff --git a/packages/analytics-core/src/types/event/base-event.ts b/packages/analytics-core/src/types/event/base-event.ts index b060db5bed..2b79043c06 100644 --- a/packages/analytics-core/src/types/event/base-event.ts +++ b/packages/analytics-core/src/types/event/base-event.ts @@ -4,6 +4,9 @@ import { IngestionMetadataEventProperty } from './ingestion-metadata'; export interface Delay { id: string; timeout?: number; + // indicates that this event was updated while in flight, + // so do not clean-up after it was sent + isFresh?: true; } export interface BaseEvent extends EventOptions { diff --git a/packages/analytics-core/src/types/payload.ts b/packages/analytics-core/src/types/payload.ts index a1ba8c6f50..a5fdaba143 100644 --- a/packages/analytics-core/src/types/payload.ts +++ b/packages/analytics-core/src/types/payload.ts @@ -12,3 +12,9 @@ export interface Payload { client_upload_time?: string; request_metadata?: RequestMetadata; } + +export interface DelayedPayload extends Payload { + id: string; + timeout: number; + instant_events?: readonly Event[]; +} diff --git a/packages/analytics-core/test/config.test.ts b/packages/analytics-core/test/config.test.ts index 1426f25556..52c74e21b3 100644 --- a/packages/analytics-core/test/config.test.ts +++ b/packages/analytics-core/test/config.test.ts @@ -33,6 +33,7 @@ describe('config', () => { plan: undefined, ingestionMetadata: undefined, serverUrl: 'https://api2.amplitude.com/2/httpapi', + delayedEventsServerUrl: undefined, serverZone: 'US', storageProvider: defaultConfig.storageProvider, transportProvider: defaultConfig.transportProvider, @@ -42,6 +43,18 @@ describe('config', () => { expect(config.optOut).toBe(false); }); + test('should set delayedEventsServerUrl', () => { + const defaultConfig = useDefaultConfig(); + const delayedEventsServerUrl = 'https://example.com/2/httpapi/delayed'; + const config = new Config({ + apiKey: API_KEY, + delayedEventsServerUrl, + storageProvider: defaultConfig.storageProvider, + transportProvider: defaultConfig.transportProvider, + }); + expect(config.delayedEventsServerUrl).toBe(delayedEventsServerUrl); + }); + test('should overwrite default config', () => { const defaultConfig = useDefaultConfig(); const config = new Config({ @@ -77,6 +90,7 @@ describe('config', () => { sourceVersion: '2.0.0', }, serverUrl: 'https://api2.amplitude.com/batch', + delayedEventsServerUrl: undefined, serverZone: 'US', storageProvider: defaultConfig.storageProvider, transportProvider: defaultConfig.transportProvider, diff --git a/packages/analytics-core/test/observers/video.test.ts b/packages/analytics-core/test/observers/video.test.ts index 9a7ad2bc2c..7ed65649b2 100644 --- a/packages/analytics-core/test/observers/video.test.ts +++ b/packages/analytics-core/test/observers/video.test.ts @@ -224,5 +224,111 @@ describe('VideoObserver', () => { ); }); }); + + describe('playing to waiting state', () => { + beforeEach(() => { + jest.useFakeTimers(); + onStateChange.mockClear(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + it('should transition to waiting state if the playhead is not moving for 1s', () => { + internalHandler.onPlay({ duration: 10, last_position: 5 }); + jest.advanceTimersByTime(500); + expect(onStateChange).not.toHaveBeenCalledWith( + expect.objectContaining({ playbackState: 'playing', position: 5 }), + expect.objectContaining({ playbackState: 'waiting', position: 5 }), + ); + jest.advanceTimersByTime(500); + expect(onStateChange).toHaveBeenCalledWith( + expect.objectContaining({ playbackState: 'playing', position: 5 }), + expect.objectContaining({ playbackState: 'waiting', position: 5 }), + ); + }); + + it('should transition from waiting to playing if the playhead starts moving', () => { + internalHandler.onPlay({ duration: 10, last_position: 5 }); + jest.advanceTimersByTime(1000); + internalHandler.onTimeUpdate({ position: 6, isSeeking: false }); + expect(onStateChange).toHaveBeenCalledWith( + expect.objectContaining({ playbackState: 'waiting', position: 5 }), + expect.objectContaining({ playbackState: 'playing', position: 6 }), + ); + }); + + it('should keep the current position and watch time when stalling mid-playback', () => { + internalHandler.onPlay({ duration: 10, last_position: 0 }); + internalHandler.onTimeUpdate({ position: 3, isSeeking: false }); + onStateChange.mockClear(); + + // the playhead is sampled every second, so a stall is detected on the second matching sample + jest.advanceTimersByTime(2000); + + expect(onStateChange).toHaveBeenLastCalledWith( + expect.objectContaining({ playbackState: 'playing', position: 3, watchTime: 3 }), + expect.objectContaining({ + playbackState: 'waiting', + position: 3, + watchTime: 3, + lastEvent: { duration: 10, last_position: 0 }, + }), + ); + }); + + it('should detect a second stall after the playhead resumes', () => { + internalHandler.onPlay({ duration: 10, last_position: 0 }); + jest.advanceTimersByTime(1000); + internalHandler.onTimeUpdate({ position: 3, isSeeking: false }); + jest.advanceTimersByTime(1000); + internalHandler.onTimeUpdate({ position: 6, isSeeking: false }); + onStateChange.mockClear(); + + jest.advanceTimersByTime(2000); + + expect(onStateChange).toHaveBeenCalledTimes(1); + expect(onStateChange).toHaveBeenLastCalledWith( + expect.objectContaining({ playbackState: 'playing', position: 6 }), + expect.objectContaining({ playbackState: 'waiting', position: 6 }), + ); + }); + + it('should stop watching the playhead after a playback error', () => { + internalHandler.onPlay({ duration: 10, last_position: 5 }); + internalHandler.onError('test error'); + onStateChange.mockClear(); + + jest.advanceTimersByTime(5000); + + expect(onStateChange).not.toHaveBeenCalled(); + expect(jest.getTimerCount()).toBe(0); + }); + + it('should resume watching the playhead when playback restarts after an error', () => { + internalHandler.onPlay({ duration: 10, last_position: 5 }); + internalHandler.onError('test error'); + internalHandler.onPlay({ duration: 10, last_position: 5 }); + onStateChange.mockClear(); + + jest.advanceTimersByTime(2000); + + expect(onStateChange).toHaveBeenLastCalledWith( + expect.objectContaining({ playbackState: 'playing', position: 5 }), + expect.objectContaining({ playbackState: 'waiting', position: 5 }), + ); + }); + + it('should stop watching the playhead once destroyed', () => { + internalHandler.onPlay({ duration: 10, last_position: 5 }); + videoObserver.destroy(); + onStateChange.mockClear(); + + jest.advanceTimersByTime(5000); + + expect(onStateChange).not.toHaveBeenCalled(); + }); + }); }); }); diff --git a/packages/analytics-core/test/plugins/destination.test.ts b/packages/analytics-core/test/plugins/destination.test.ts index 9d3ff5748f..32ad9cd6fb 100644 --- a/packages/analytics-core/test/plugins/destination.test.ts +++ b/packages/analytics-core/test/plugins/destination.test.ts @@ -111,6 +111,183 @@ describe('destination', () => { expect(schedule).toHaveBeenCalledTimes(1); expect(saveEvents).toHaveBeenCalledTimes(1); }); + + describe('stale delayed events', () => { + let destination: Destination; + const delayId = 'delay-123'; + const event1 = { + event_type: 'before', + insert_id: '123', + delay: { id: delayId }, + }; + const event2 = { + event_type: 'after', + insert_id: '123', + delay: { id: delayId }, + }; + + beforeEach(() => { + jest.useFakeTimers(); + destination = new Destination(); + destination.config = useDefaultConfig(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + test('should be removed from queue and resolved as "stale" if event is not in flight', async () => { + const staleResult = destination.execute(event1); + void destination.execute(event2); + + expect(destination.queue.length).toBe(1); + expect(destination.queue[0].event).toEqual(event2); + + await expect(staleResult).resolves.toEqual({ + event: event1, + code: 0, + message: 'Stale event overwritten', + }); + }); + + test('should not return "stale" result if the event is in flight', async () => { + let resolveSend!: (value: Response) => void; + const sendPromise = new Promise((resolve) => { + resolveSend = resolve; + }); + destination.config = { + ...useDefaultConfig(), + transportProvider: { + send: jest.fn().mockReturnValue(sendPromise), + }, + }; + + const staleResult = destination.execute(event1); + void destination.flush(true); + void destination.execute(event2); + // doesn't remove the event from the queue because it is in flight + expect(destination.queue).toHaveLength(2); + + resolveSend({ + status: Status.Success, + statusCode: 200, + body: { + eventsIngested: 1, + payloadSizeBytes: 1, + serverUploadTime: 1, + }, + }); + + await expect(staleResult).resolves.toEqual({ + event: event1, + code: 200, + message: SUCCESS_MESSAGE, + }); + }); + + test('should retain replacement event after in-flight predecessor completes', async () => { + let resolveSend!: (value: Response) => void; + const sendPromise = new Promise((resolve) => { + resolveSend = resolve; + }); + const successResponse = { + status: Status.Success, + statusCode: 200, + body: { + eventsIngested: 1, + payloadSizeBytes: 1, + serverUploadTime: 1, + }, + } as Response; + const send = jest.fn().mockReturnValueOnce(sendPromise).mockResolvedValueOnce(successResponse); + destination.config = { + ...useDefaultConfig(), + transportProvider: { send }, + }; + + const staleResult = destination.execute(event1); + const flushPromise = destination.flush(true); + const replacementResult = destination.execute(event2); + + expect(destination.queue).toHaveLength(2); + + resolveSend(successResponse); + + await staleResult; + await flushPromise; + + // removeEvents must not drop the replacement when fulfilling the in-flight send by insert_id + expect(destination.queue).toHaveLength(1); + expect(destination.queue[0].event).toEqual(event2); + + await destination.flush(true); + + await expect(replacementResult).resolves.toEqual({ + event: event2, + code: 200, + message: SUCCESS_MESSAGE, + }); + expect(send).toHaveBeenCalledTimes(2); + }); + + test('should retain isFresh when a parallel regular flush completes first', async () => { + let resolveDelayedSend!: (value: Response) => void; + const delayedSendPromise = new Promise((resolve) => { + resolveDelayedSend = resolve; + }); + const successResponse = { + status: Status.Success, + statusCode: 200, + body: { + eventsIngested: 1, + payloadSizeBytes: 1, + serverUploadTime: 1, + }, + } as Response; + const regularEvent = { event_type: 'regular', insert_id: 'regular-1' }; + const send = jest + .fn() + // first flush: delayed predecessor + regular event in parallel + .mockReturnValueOnce(delayedSendPromise) + .mockResolvedValueOnce(successResponse) + // second flush: replacement delayed event + .mockResolvedValueOnce(successResponse); + destination.config = { + ...useDefaultConfig(), + transportProvider: { send }, + }; + + const predecessorResult = destination.execute(event1); + const regularResult = destination.execute(regularEvent); + const flushPromise = destination.flush(true); + const replacementResult = destination.execute(event2); + + expect(destination.queue).toHaveLength(3); + expect(destination.queue[2].event.delay?.isFresh).toBe(true); + + // Regular batch completes while delayed predecessor is still in flight. + // removeEvents must not clear isFresh on the unrelated replacement. + await regularResult; + expect(destination.queue).toHaveLength(2); + expect(destination.queue.find((c) => c.event === event2)?.event.delay?.isFresh).toBe(true); + + resolveDelayedSend(successResponse); + await predecessorResult; + await flushPromise; + + expect(destination.queue).toHaveLength(1); + expect(destination.queue[0].event).toEqual(event2); + + await destination.flush(true); + + await expect(replacementResult).resolves.toEqual({ + event: event2, + code: 200, + message: SUCCESS_MESSAGE, + }); + expect(send).toHaveBeenCalledTimes(3); + }); + }); }); describe('removeEventsExceedFlushMaxRetries', () => { @@ -1700,4 +1877,264 @@ describe('destination', () => { expect(result).toBe(''); }); }); + + describe('delayed events', () => { + const successResponse = { + status: Status.Success, + statusCode: 200, + body: { + eventsIngested: 1, + payloadSizeBytes: 1, + serverUploadTime: 1, + }, + }; + const delayedUrl = `${AMPLITUDE_SERVER_URL}/delayed`; + + let destination: Destination; + let transportProvider: { send: jest.Mock }; + + const createContext = ( + event: { event_type: string; delay?: { id: string; timeout?: number } }, + callback = jest.fn(), + timeout = 0, + ): Context => ({ + attempts: 0, + callback, + event, + timeout, + }); + + const flushQueue = async (contexts: Context[]) => { + destination.queue = contexts; + await destination.flush(true); + }; + + const expectSuccess = (callback: jest.Mock, event: Context['event']) => { + expect(callback).toHaveBeenCalledWith({ + event, + code: 200, + message: SUCCESS_MESSAGE, + }); + }; + + beforeEach(async () => { + destination = new Destination(); + transportProvider = { + send: jest.fn().mockResolvedValue(successResponse), + }; + await destination.setup({ + ...useDefaultConfig(), + transportProvider, + apiKey: API_KEY, + serverUrl: AMPLITUDE_SERVER_URL, + }); + }); + + test('should send delayed events to /delayed endpoint', async () => { + const delayId = 'delay-123'; + const delayTimeout = 5000; + const event = { event_type: 'delayed_event', delay: { id: delayId, timeout: delayTimeout } }; + const callback = jest.fn(); + await flushQueue([createContext(event, callback)]); + + expect(transportProvider.send).toHaveBeenCalledTimes(1); + expect(transportProvider.send).toHaveBeenCalledWith( + delayedUrl, + expect.objectContaining({ + id: delayId, + timeout: delayTimeout, + events: [expect.objectContaining({ event_type: 'delayed_event' })], + instant_events: [], + }), + false, // delayed endpoint does not support compression + ); + expectSuccess(callback, event); + }); + + test('should not include delay in the sent events', async () => { + const delayId = 'delay-123'; + const delayTimeout = 5000; + const event = { event_type: 'delayed_event', delay: { id: delayId, timeout: delayTimeout } }; + const callback = jest.fn(); + await flushQueue([createContext(event, callback)]); + + expect(transportProvider.send).toHaveBeenCalledTimes(1); + const sentPayload = transportProvider.send.mock.calls[0][1] as Payload; + const sentEvent = sentPayload.events[0]; + expect(sentEvent).not.toHaveProperty('delay'); + expectSuccess(callback, event); + }); + + test('should send instant_events to /delayed endpoint when delay_timeout is not set', async () => { + const delayId = 'delay-123'; + const event = { event_type: 'instant_event', delay: { id: delayId } }; + const callback = jest.fn(); + await flushQueue([createContext(event, callback)]); + + expect(transportProvider.send).toHaveBeenCalledTimes(1); + expect(transportProvider.send).toHaveBeenCalledWith( + delayedUrl, + expect.objectContaining({ + id: delayId, + timeout: 0, + events: [], + instant_events: [expect.objectContaining({ event_type: 'instant_event' })], + }), + false, // delayed endpoint does not support compression + ); + expectSuccess(callback, event); + }); + + test('should send /delayed and regular events on same flush', async () => { + const delayId = 'delay-123'; + const delayTimeout = 5000; + const regularCallback = jest.fn(); + const delayedCallback = jest.fn(); + const regularContext = createContext({ event_type: 'regular_event' }, regularCallback); + const delayedContext = createContext( + { event_type: 'delayed_event', delay: { id: delayId, timeout: delayTimeout } }, + delayedCallback, + ); + + await flushQueue([regularContext, delayedContext]); + + expect(transportProvider.send).toHaveBeenCalledTimes(2); + expect(transportProvider.send).toHaveBeenCalledWith( + AMPLITUDE_SERVER_URL, + expect.objectContaining({ + events: [expect.objectContaining({ event_type: 'regular_event' })], + }), + true, + ); + expect(transportProvider.send).toHaveBeenCalledWith( + delayedUrl, + expect.objectContaining({ + id: delayId, + timeout: delayTimeout, + events: [expect.objectContaining({ event_type: 'delayed_event' })], + instant_events: [], + }), + false, // delayed endpoint does not support compression + ); + expectSuccess(regularCallback, regularContext.event); + expectSuccess(delayedCallback, delayedContext.event); + }); + + test('should send delayed events with different delay_ids on same flush', async () => { + const delayTimeout = 5000; + const callbackA = jest.fn(); + const callbackB = jest.fn(); + const delayedContextA = createContext( + { event_type: 'delayed_event_a', delay: { id: 'delay-a', timeout: delayTimeout } }, + callbackA, + ); + const delayedContextB = createContext( + { event_type: 'delayed_event_b', delay: { id: 'delay-b', timeout: delayTimeout } }, + callbackB, + ); + + await flushQueue([delayedContextA, delayedContextB]); + + expect(transportProvider.send).toHaveBeenCalledTimes(2); + expect(transportProvider.send).toHaveBeenNthCalledWith( + 1, + delayedUrl, + expect.objectContaining({ + id: 'delay-a', + timeout: delayTimeout, + events: [ + expect.objectContaining({ + event_type: 'delayed_event_a', + }), + ], + instant_events: [], + }), + false, // delayed endpoint does not support compression + ); + expect(transportProvider.send).toHaveBeenNthCalledWith( + 2, + delayedUrl, + expect.objectContaining({ + id: 'delay-b', + timeout: delayTimeout, + events: [ + expect.objectContaining({ + event_type: 'delayed_event_b', + }), + ], + instant_events: [], + }), + false, // delayed endpoint does not support compression + ); + expectSuccess(callbackA, delayedContextA.event); + expectSuccess(callbackB, delayedContextB.event); + }); + + test('should not send delayed events while backoff timeout is active', async () => { + const delayId = 'delay-123'; + const event = { event_type: 'delayed_event', delay: { id: delayId } }; + const callback = jest.fn(); + const context = createContext(event, callback, 1000); + + destination.queue = [context]; + await destination.flush(true); + + expect(transportProvider.send).not.toHaveBeenCalled(); + expect(callback).not.toHaveBeenCalled(); + expect(destination.queue).toEqual([context]); + }); + + test('should map invalid response indices to wire order for mixed delayed batches', async () => { + const delayId = 'delay-123'; + const delayTimeout = 5000; + // Instant event is queued first; after translation wire order is [delayed, instant]. + // Index 0 in the response must drop the delayed event, not the instant one. + transportProvider.send + .mockResolvedValueOnce({ + status: Status.Invalid, + statusCode: 400, + body: { + error: 'error', + missingField: '', + eventsWithInvalidFields: { a: [0] }, + eventsWithMissingFields: {}, + eventsWithInvalidIdLengths: {}, + silencedEvents: [], + }, + }) + .mockResolvedValueOnce(successResponse); + + destination.retryTimeout = 1; + destination.config.flushIntervalMillis = 1; + + const results = await Promise.all([ + destination.execute({ + event_type: 'instant_event', + insert_id: 'instant-0', + delay: { id: delayId }, + }), + destination.execute({ + event_type: 'delayed_event', + insert_id: 'delayed-0', + delay: { id: delayId, timeout: delayTimeout }, + }), + ]); + + expect(transportProvider.send).toHaveBeenCalledTimes(2); + expect(transportProvider.send).toHaveBeenNthCalledWith( + 1, + delayedUrl, + expect.objectContaining({ + events: [expect.objectContaining({ event_type: 'delayed_event' })], + instant_events: [expect.objectContaining({ event_type: 'instant_event' })], + }), + false, // delayed endpoint does not support compression + ); + expect(results[0].code).toBe(200); + expect(results[0].event.insert_id).toBe('instant-0'); + expect(results[1].code).toBe(400); + expect(results[1].event.insert_id).toBe('delayed-0'); + expect(destination.queue.length).toBe(0); + }); + }); }); diff --git a/packages/analytics-node/test/config.test.ts b/packages/analytics-node/test/config.test.ts index 998043ae68..e1c7dc31b0 100644 --- a/packages/analytics-node/test/config.test.ts +++ b/packages/analytics-node/test/config.test.ts @@ -26,6 +26,7 @@ describe('config', () => { plan: undefined, ingestionMetadata: undefined, serverUrl: 'https://api2.amplitude.com/2/httpapi', + delayedEventsServerUrl: undefined, serverZone: 'US', storageProvider: undefined, transportProvider: new Http(), @@ -60,6 +61,7 @@ describe('config', () => { plan: undefined, ingestionMetadata: undefined, serverUrl: 'https://api2.amplitude.com/2/httpapi', + delayedEventsServerUrl: undefined, serverZone: 'US', storageProvider: undefined, transportProvider: new Http(), diff --git a/packages/analytics-react-native/test/config.test.ts b/packages/analytics-react-native/test/config.test.ts index dd10f8fe7f..0cf7262988 100644 --- a/packages/analytics-react-native/test/config.test.ts +++ b/packages/analytics-react-native/test/config.test.ts @@ -43,6 +43,7 @@ describe('config', () => { plan: undefined, ingestionMetadata: undefined, serverUrl: 'https://api2.amplitude.com/2/httpapi', + delayedEventsServerUrl: undefined, serverZone: 'US', sessionTimeout: 300000, trackingOptions: { @@ -98,6 +99,7 @@ describe('config', () => { plan: undefined, ingestionMetadata: undefined, serverUrl: 'https://api2.amplitude.com/2/httpapi', + delayedEventsServerUrl: undefined, serverZone: 'US', sessionTimeout: 300000, storageProvider: new core.MemoryStorage(), @@ -179,6 +181,7 @@ describe('config', () => { sourceVersion: '2.0.0', }, serverUrl: 'https://api2.amplitude.com/2/httpapi', + delayedEventsServerUrl: undefined, serverZone: 'US', _sessionId: -1, sessionTimeout: 1, diff --git a/test-server/mock-api.js b/test-server/mock-api.js index 28dc7d8ae8..003db1a803 100644 --- a/test-server/mock-api.js +++ b/test-server/mock-api.js @@ -18,6 +18,14 @@ export function createMockApi() { // Function to configure mock API middleware that can be used by both dev and preview servers export function configureMockApiMiddleware(middlewares) { + // Mock delayed-events endpoint. + middlewares.use('/2/httpapi/delayed', (_req, res) => { + res.statusCode = 200; + res.setHeader('Content-Type', 'application/json'); + res.setHeader('Access-Control-Allow-Origin', '*'); + res.end(JSON.stringify({ code: 200 })); + }); + // Status code endpoint - responds with the status code specified in the URL middlewares.use((req, res, next) => { const statusMatch = req.url.match(/^\/api\/status\/(\d+)/); diff --git a/test-server/video-analytics/track-html-video.html b/test-server/video-analytics/track-html-video.html index 2ded25c720..7c3ad24884 100644 --- a/test-server/video-analytics/track-html-video.html +++ b/test-server/video-analytics/track-html-video.html @@ -28,7 +28,7 @@

Track HTML Video Test

// MUX VIDEO MONITORING // monitor the video for Mux analytics - mux.monitor(video, { + mux?.monitor(video, { data: { view_session_id: muxViewSessionId, viewer_user_id: userId, @@ -40,6 +40,7 @@

Track HTML Video Test

// initialize Amplitude amplitude.setUserId(userId); amplitude.init(import.meta.env.VITE_AMPLITUDE_API_KEY, { + delayedEventsServerUrl: `${location.origin}/2/httpapi/delayed`, fetchRemoteConfig: false, autocapture: false, }).promise.then(() => {