WIP: Video analytics - #1957
Open
daniel-graham-amplitude wants to merge 5 commits into
Open
Conversation
- @amplitude/analytics-browser@2.46.0-video-analytics.0 - @amplitude/analytics-client-common@2.4.60-video-analytics.0 - @amplitude/analytics-core@2.55.0-video-analytics.0 - @amplitude/analytics-node@1.6.0-video-analytics.0 - @amplitude/analytics-react-native@1.7.0-video-analytics.0 - @amplitude/plugin-autocapture-browser@1.28.11-video-analytics.0 - @amplitude/plugin-custom-enrichment-browser@0.1.21-video-analytics.0 - @amplitude/plugin-event-property-attribution-browser@0.2.13-video-analytics.0 - @amplitude/plugin-experiment-browser@1.0.0-video-analytics.0 - @amplitude/plugin-network-capture-browser@1.10.13-video-analytics.0 - @amplitude/plugin-page-url-enrichment-browser@0.7.23-video-analytics.0 - @amplitude/plugin-page-view-tracking-browser@2.11.13-video-analytics.0 - @amplitude/plugin-session-replay-browser@1.33.8-video-analytics.0 - @amplitude/plugin-web-attribution-browser@2.2.23-video-analytics.0 - @amplitude/plugin-web-vitals-browser@1.1.45-video-analytics.0 - @amplitude/segment-session-replay-plugin@0.0.40-video-analytics.0 - @amplitude/session-replay-browser@1.48.2-video-analytics.0 - @amplitude/targeting@0.3.10-video-analytics.0 - @amplitude/unified@1.1.29-video-analytics.0
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 3 potential issues.
Autofix Details
Bugbot Autofix prepared fixes for all 3 issues found in the latest run.
- ✅ Fixed: Percent completed mishandles zero duration
- parseStopEventProperties now requires finite inputs and duration > 0, then clamps percent_completed to [0, 100], matching calculatePercentCompleted.
- ✅ Fixed: Batch mode builds wrong delayed URL
- Delayed uploads now resolve the HTTP V2 base URL (or a custom non-Amplitude serverUrl) instead of appending /delayed to the Batch API path.
- ✅ Fixed: Shared delay object blocks stale removal
- removeStaleDelayedEvents clones delay before setting isFresh so heartbeat-shared delay objects no longer keep the in-flight predecessor in the queue.
Or push these changes by commenting:
@cursor push 11e0afb1ad
Preview (11e0afb1ad)
diff --git a/packages/analytics-browser/src/video-capture/video-capture.ts b/packages/analytics-browser/src/video-capture/video-capture.ts
--- a/packages/analytics-browser/src/video-capture/video-capture.ts
+++ b/packages/analytics-browser/src/video-capture/video-capture.ts
@@ -216,11 +216,16 @@
}
parseStopEventProperties(nextState: VideoState): Record<string, string | number | boolean> {
- const percentCompleted = ((nextState.position ?? 0) / (nextState.lastEvent?.duration ?? 0)) * 100;
+ const position = nextState.position ?? 0;
+ const duration = nextState.lastEvent?.duration ?? 0;
+ let percentCompleted = 0;
+ if (Number.isFinite(position) && Number.isFinite(duration) && duration > 0) {
+ percentCompleted = Math.min(100, Math.max(0, (position / duration) * 100));
+ }
return {
...this.parseStartEventProperties(nextState),
watch_duration: nextState.watchTime ?? 0,
- percent_completed: percentCompleted || 0,
+ percent_completed: percentCompleted,
};
}
}
diff --git a/packages/analytics-browser/test/video-capture/video-capture.test.ts b/packages/analytics-browser/test/video-capture/video-capture.test.ts
--- a/packages/analytics-browser/test/video-capture/video-capture.test.ts
+++ b/packages/analytics-browser/test/video-capture/video-capture.test.ts
@@ -628,5 +628,61 @@
percent_completed: 0,
});
});
+
+ it('should return 0 percent_completed when duration is 0', () => {
+ const capture = new VideoCapture(mockAmplitude);
+ expect(
+ capture.parseStopEventProperties({
+ playbackState: 'paused',
+ lastEvent: { duration: 0, start_time: 0, last_position: 5 },
+ position: 5,
+ watchTime: 5,
+ }),
+ ).toEqual({
+ duration: 0,
+ start_time: 0,
+ position: 5,
+ watch_duration: 5,
+ percent_completed: 0,
+ });
+ });
+
+ it('should clamp percent_completed when position exceeds duration', () => {
+ const capture = new VideoCapture(mockAmplitude);
+ expect(
+ capture.parseStopEventProperties({
+ playbackState: 'ended',
+ lastEvent: { duration: 10, start_time: 0, last_position: 12 },
+ position: 12,
+ watchTime: 12,
+ }),
+ ).toEqual({
+ duration: 10,
+ start_time: 0,
+ position: 12,
+ watch_duration: 12,
+ percent_completed: 100,
+ });
+ });
+
+ it('should return 0 percent_completed for non-finite duration or position', () => {
+ const capture = new VideoCapture(mockAmplitude);
+ expect(
+ capture.parseStopEventProperties({
+ playbackState: 'paused',
+ lastEvent: { duration: Infinity, start_time: 0, last_position: 5 },
+ position: 5,
+ watchTime: 5,
+ }).percent_completed,
+ ).toBe(0);
+ expect(
+ capture.parseStopEventProperties({
+ playbackState: 'paused',
+ lastEvent: { duration: 10, start_time: 0, last_position: 0 },
+ position: NaN,
+ watchTime: 0,
+ }).percent_completed,
+ ).toBe(0);
+ });
});
});
diff --git a/packages/analytics-core/src/plugins/destination.ts b/packages/analytics-core/src/plugins/destination.ts
--- a/packages/analytics-core/src/plugins/destination.ts
+++ b/packages/analytics-core/src/plugins/destination.ts
@@ -161,7 +161,8 @@
context.event.insert_id === incomingEvent.insert_id
) {
if (this.inFlightDelayedEvents[incomingEvent.delay.id]) {
- incomingEvent.delay.isFresh = true;
+ // Clone so isFresh is not set on a delay object shared with the in-flight event.
+ incomingEvent.delay = { ...incomingEvent.delay, isFresh: true };
return true;
}
context.callback(buildResult(context.event, 0, 'Stale event overwritten'));
@@ -352,7 +353,15 @@
this.config.enableRequestBodyCompression,
);
if (delay) {
- serverUrl = this.config.delayedEventsServerUrl || `${serverUrl}/delayed`;
+ // Delayed ingest is HTTP V2 only; the Batch API has no /delayed route.
+ const { serverUrl: delayedBaseUrl } = createServerConfig(
+ this.config.serverUrl && !DEFAULT_AMPLITUDE_SERVER_URLS.has(this.config.serverUrl)
+ ? this.config.serverUrl
+ : '',
+ this.config.serverZone,
+ false,
+ );
+ serverUrl = this.config.delayedEventsServerUrl || `${delayedBaseUrl}/delayed`;
this.translatePayloadToDelayedPayload(payload, list);
shouldCompressUploadBody = false; // delayed events doesn't support compression
}
diff --git a/packages/analytics-core/test/plugins/destination.test.ts b/packages/analytics-core/test/plugins/destination.test.ts
--- a/packages/analytics-core/test/plugins/destination.test.ts
+++ b/packages/analytics-core/test/plugins/destination.test.ts
@@ -14,7 +14,7 @@
import { uuidPattern } from '../helpers/util';
import { DiagnosticsClient, RequestMetadata } from '../../src';
import { TrackEvent } from '../../src/types/event/event';
-import { AMPLITUDE_SERVER_URL } from '../../src/types/constants';
+import { AMPLITUDE_BATCH_SERVER_URL, AMPLITUDE_SERVER_URL } from '../../src/types/constants';
const jsons = (obj: any) => JSON.stringify(obj, null, 2);
@@ -287,6 +287,61 @@
});
expect(send).toHaveBeenCalledTimes(3);
});
+
+ test('should drop in-flight predecessor when replacement shares the same delay object', async () => {
+ const sharedDelay = { id: delayId };
+ const predecessor = {
+ event_type: 'before',
+ insert_id: '123',
+ delay: sharedDelay,
+ };
+ const replacement = {
+ event_type: 'after',
+ insert_id: '123',
+ delay: sharedDelay,
+ };
+ let resolveSend!: (value: Response) => void;
+ const sendPromise = new Promise<Response>((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(predecessor);
+ const flushPromise = destination.flush(true);
+ const replacementResult = destination.execute(replacement);
+
+ expect(destination.queue).toHaveLength(2);
+
+ resolveSend(successResponse);
+
+ await staleResult;
+ await flushPromise;
+
+ expect(destination.queue).toHaveLength(1);
+ expect(destination.queue[0].event).toEqual(replacement);
+
+ await destination.flush(true);
+
+ await expect(replacementResult).resolves.toEqual({
+ event: replacement,
+ code: 200,
+ message: SUCCESS_MESSAGE,
+ });
+ expect(send).toHaveBeenCalledTimes(2);
+ });
});
});
@@ -1930,6 +1985,79 @@
});
});
+ test('should send delayed events to default HTTP V2 /delayed when serverUrl is empty', async () => {
+ const emptyUrlDestination = new Destination();
+ const emptyUrlTransport = {
+ send: jest.fn().mockResolvedValue(successResponse),
+ };
+ await emptyUrlDestination.setup({
+ ...useDefaultConfig(),
+ transportProvider: emptyUrlTransport,
+ apiKey: API_KEY,
+ serverUrl: '',
+ serverZone: 'US',
+ });
+ const delayId = 'delay-123';
+ const event = { event_type: 'delayed_event', delay: { id: delayId, timeout: 5000 } };
+ emptyUrlDestination.queue = [createContext(event)];
+ await emptyUrlDestination.flush(true);
+
+ expect(emptyUrlTransport.send).toHaveBeenCalledWith(delayedUrl, expect.objectContaining({ id: delayId }), false);
+ });
+
+ test('should append /delayed to a custom server URL', async () => {
+ const customDestination = new Destination();
+ const customTransport = {
+ send: jest.fn().mockResolvedValue(successResponse),
+ };
+ const customServerUrl = 'https://proxy.example.com/2/httpapi';
+ await customDestination.setup({
+ ...useDefaultConfig(),
+ transportProvider: customTransport,
+ apiKey: API_KEY,
+ serverUrl: customServerUrl,
+ });
+ const delayId = 'delay-123';
+ const event = { event_type: 'delayed_event', delay: { id: delayId, timeout: 5000 } };
+ customDestination.queue = [createContext(event)];
+ await customDestination.flush(true);
+
+ expect(customTransport.send).toHaveBeenCalledWith(
+ `${customServerUrl}/delayed`,
+ expect.objectContaining({ id: delayId }),
+ false,
+ );
+ });
+
+ test('should send delayed events to HTTP V2 /delayed when useBatch is true', async () => {
+ const batchDestination = new Destination();
+ const batchTransport = {
+ send: jest.fn().mockResolvedValue(successResponse),
+ };
+ await batchDestination.setup({
+ ...useDefaultConfig(),
+ transportProvider: batchTransport,
+ apiKey: API_KEY,
+ useBatch: true,
+ serverUrl: AMPLITUDE_BATCH_SERVER_URL,
+ });
+ const delayId = 'delay-123';
+ const delayTimeout = 5000;
+ const event = { event_type: 'delayed_event', delay: { id: delayId, timeout: delayTimeout } };
+ batchDestination.queue = [createContext(event)];
+ await batchDestination.flush(true);
+
+ expect(batchTransport.send).toHaveBeenCalledTimes(1);
+ expect(batchTransport.send).toHaveBeenCalledWith(
+ delayedUrl,
+ expect.objectContaining({
+ id: delayId,
+ timeout: delayTimeout,
+ }),
+ false,
+ );
+ });
+
test('should send delayed events to /delayed endpoint', async () => {
const delayId = 'delay-123';
const delayTimeout = 5000;You can send follow-ups to the cloud agent here.
| ...this.parseStartEventProperties(nextState), | ||
| watch_duration: nextState.watchTime ?? 0, | ||
| percent_completed: percentCompleted || 0, | ||
| }; |
There was a problem hiding this comment.
Percent completed mishandles zero duration
Medium Severity
parseStopEventProperties divides playhead position by duration with no finite/duration > 0 guard, so a 0 duration (unloaded media, some live streams) yields Infinity. percentCompleted || 0 does not catch that, and JSON.stringify later turns it into null. The same helper also skips clamping, so position past duration can exceed 100. calculatePercentCompleted in track-video.ts already handles these cases, but this path overwrites that value.
Reviewed by Cursor Bugbot for commit 9a860a8. Configure here.
| ); | ||
| if (delay) { | ||
| serverUrl = this.config.delayedEventsServerUrl || `${serverUrl}/delayed`; | ||
| this.translatePayloadToDelayedPayload(payload, list); |
There was a problem hiding this comment.
Batch mode builds wrong delayed URL
Medium Severity
Delayed uploads fall back to `${serverUrl}/delayed` after createServerConfig is called with useBatch. With batching enabled and no delayedEventsServerUrl, that becomes https://api2.amplitude.com/batch/delayed (or the EU batch equivalent) instead of the HTTP V2 delayed path (/2/httpapi/delayed). Video heartbeat events then post to an endpoint that does not exist.
Reviewed by Cursor Bugbot for commit 9a860a8. Configure here.
| ) { | ||
| if (this.inFlightDelayedEvents[incomingEvent.delay.id]) { | ||
| incomingEvent.delay.isFresh = true; | ||
| return true; |
There was a problem hiding this comment.
Shared delay object blocks stale removal
Medium Severity
removeStaleDelayedEvents sets incomingEvent.delay.isFresh = true on the delay object itself. Heartbeat re-tracks queued events by passing that same delay reference into client.track(), so the in-flight queue entry is marked fresh too. removeEvents then keeps both copies, and the next flush can send the same video stop twice.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 9a860a8. Configure here.
size-limit report 📦
|
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes and found 1 potential issue.
There are 4 total unresolved issues (including 3 from previous reviews).
Bugbot Autofix prepared a fix for the issue found in the latest run.
- ✅ Fixed: In-flight replacements can duplicate events
- removeStaleDelayedEvents now drops prior isFresh replacements while a delay id is in flight, so only the latest update remains after the send completes.
Or push these changes by commenting:
@cursor push 9721635f78
Preview (9721635f78)
diff --git a/packages/analytics-core/src/plugins/destination.ts b/packages/analytics-core/src/plugins/destination.ts
--- a/packages/analytics-core/src/plugins/destination.ts
+++ b/packages/analytics-core/src/plugins/destination.ts
@@ -161,6 +161,12 @@
context.event.insert_id === incomingEvent.insert_id
) {
if (this.inFlightDelayedEvents[incomingEvent.delay.id]) {
+ // Keep the in-flight event, but drop any prior replacement so
+ // only the latest update remains after the send completes.
+ if (context.event.delay.isFresh) {
+ context.callback(buildResult(context.event, 0, 'Stale event overwritten'));
+ return false;
+ }
incomingEvent.delay.isFresh = true;
return true;
}
diff --git a/packages/analytics-core/test/plugins/destination.test.ts b/packages/analytics-core/test/plugins/destination.test.ts
--- a/packages/analytics-core/test/plugins/destination.test.ts
+++ b/packages/analytics-core/test/plugins/destination.test.ts
@@ -230,6 +230,64 @@
expect(send).toHaveBeenCalledTimes(2);
});
+ test('should keep only the latest replacement while predecessor is in flight', async () => {
+ let resolveSend!: (value: Response) => void;
+ const sendPromise = new Promise<Response>((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 event3 = {
+ event_type: 'after-after',
+ insert_id: '123',
+ delay: { id: delayId },
+ };
+
+ const staleResult = destination.execute(event1);
+ const flushPromise = destination.flush(true);
+ const firstReplacementResult = destination.execute(event2);
+ const secondReplacementResult = destination.execute(event3);
+
+ expect(destination.queue).toHaveLength(2);
+ expect(destination.queue[0].event).toEqual(event1);
+ expect(destination.queue[1].event).toEqual(event3);
+
+ await expect(firstReplacementResult).resolves.toEqual({
+ event: event2,
+ code: 0,
+ message: 'Stale event overwritten',
+ });
+
+ resolveSend(successResponse);
+
+ await staleResult;
+ await flushPromise;
+
+ expect(destination.queue).toHaveLength(1);
+ expect(destination.queue[0].event).toEqual(event3);
+
+ await destination.flush(true);
+
+ await expect(secondReplacementResult).resolves.toEqual({
+ event: event3,
+ 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<Response>((resolve) => {You can send follow-ups to the cloud agent here.
Reviewed by Cursor Bugbot for commit d9c4179. Configure here.
| return false; | ||
| } | ||
| return true; | ||
| }); |
There was a problem hiding this comment.
In-flight replacements can duplicate events
Low Severity
removeStaleDelayedEvents keeps every queued event with the same insert_id while that delay.id is in flight, instead of only the request currently being sent. A second heartbeat update during a long or hung send leaves multiple replacements in the queue, which are all flushed after the original request finishes.
Reviewed by Cursor Bugbot for commit d9c4179. Configure here.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Checklist
Note
Medium Risk
Changes core event upload routing and queue semantics for all delayed events, plus new video session lifecycle behavior that affects analytics accuracy if heartbeat or stall detection misbehaves.
Overview
Adds experimental delayed-event upload so events tagged with a shared
delay.idcan be sent to a dedicated/delayedendpoint (ordelayedEventsServerUrl), including payload shaping (instant_eventsvs timed events), in-flight deduplication viaisFresh, and queue cleanup when a newer version replaces a stale delayed event.Video analytics now emits Video Content Started immediately and keeps a Video Content Stopped event on the shared client heartbeat: the stop event is updated as playback progresses, heartbeated with a long timeout fallback, and flushed early on pause/end/teardown with
stop_reasonandplay_id. Buffering is modeled aswaitinginVideoObserver(stall detection when the playhead stops moving) so a single play session is not split across rebuffering.Browser init gains
delayedEventsServerUrlwiring; the test server and sample HTML page mock/configure the delayed endpoint for local video tracking.Reviewed by Cursor Bugbot for commit d9c4179. Bugbot is set up for automated code reviews on this repo. Configure here.