diff --git a/docs/README.md b/docs/README.md index befef09..d6475a0 100644 --- a/docs/README.md +++ b/docs/README.md @@ -28,7 +28,7 @@ verified in their own repositories. ## Current compatibility -- package version: `0.5.0`; +- package version: `0.5.3`; - session JSON schema: `9`; - fingerprint schema: `6`; - minimum Dart SDK: `3.9.2`; diff --git a/docs/integration/collector.md b/docs/integration/collector.md index 8429ffa..459d789 100644 --- a/docs/integration/collector.md +++ b/docs/integration/collector.md @@ -159,7 +159,7 @@ The SDK calls: | Request | Purpose | | --- | --- | -| `POST /v1/sessions` | `session_start` and `session_end` lifecycle payloads | +| `POST /v1/sessions` | Session lifecycle and identity: `session_start`, `session_identify`, `session_end`, `traits_updated`, `user_changed` | | `POST /v1/events/batch` | JSON event batches | | `POST /v1/frames` | multipart PNG frame upload | @@ -173,10 +173,30 @@ an accepted start response and reads its `sessionId`; events, frames, and the end lifecycle request then use that collector-issued ID. Events and frames are not uploaded before this handshake completes. +Session payloads may include: + +- `traits` — full traits snapshot when the host has set a bag (`session_start`, + `session_identify`, `traits_updated`, `user_changed`); the collector stores + the bag as-is (no server-side partial merge); +- `traitsId` — pass-through of a prior collector-issued id when no new bag is + sent (for example `session_end`, or `session_start` after only an id is + cached). Ignored by the collector when `traits` is present. + +Accepted session responses (`202`) may return `traitsId`. The SDK caches that +value in process memory and stamps it onto subsequent event batches. Host apps +register traits with `TugboatReplay.setTraits` and change the runtime user with +`TugboatReplay.setUserId`. While `session_start` is still pending, both APIs +update in-memory identity only (folded into start at send time). After start, +changes within 3s coalesce into one lifecycle POST: `session_identify` when +both user and traits change, otherwise `user_changed` or `traits_updated`. +Debounced updates are flushed before `session_end`. The SDK does **not** call +`/v1/identify` or `/v1/events/identify`. + Event payloads contain: - event ID, type, `atMs`, and absolute UTC `triggeredAt`; - optional user/session/run/action IDs; +- optional `traitsId` (pass-through only; does not upsert the traits dictionary); - optional before/after frame references, related-event ID, and result; - serialized state and target anchors; - event-specific data under `payload`; diff --git a/packages/tugboat/CHANGELOG.md b/packages/tugboat/CHANGELOG.md index 57cd583..32ba4aa 100644 --- a/packages/tugboat/CHANGELOG.md +++ b/packages/tugboat/CHANGELOG.md @@ -1,3 +1,46 @@ +## 0.5.3 + +### Added + +- **Debounced identity coalesce** — after `session_start`, `setUserId` and + `setTraits` within 3s consolidate into one `session_identify` POST when + both change; otherwise `user_changed` or `traits_updated`. Pending updates are + flushed before `session_end`. + +### Changed + +- Pre-start identity still folds into a single `session_start` when values are + staged before or while start is pending. + +## 0.5.2 + +### Changed + +- **`setUserId` folds into pending `session_start`** — when a `session_start` + is still pending, `CollectorHttpSink.setUserId` updates the runtime id only + and skips `user_changed` (same coalesce already used by `setTraits`). Boot + identity can land on a single `session_start` POST. + +## 0.5.1 + +### Added + +- **User traits via collector sessions** — `TugboatReplay.setTraits` posts + `eventType: traits_updated` on `POST /v1/sessions` with a full traits bag, + caches the response `traitsId`, and stamps it on event batches. + `TugboatReplay.setUserId` posts `user_changed` and updates the runtime user + id. Pre-set traits are included on the next `session_start`. No + `/v1/identify` route. + +### Changed + +- **`setUserId` skips unchanged ids** — calling `TugboatReplay.setUserId` / + `CollectorHttpSink.setUserId` with the same value as the current runtime + user id does not post `user_changed`. +- **Pre-initialize identify** — `setTraits` / `setUserId` called after the + controller mounts but before the HTTP sink is created retain identity for + the next `session_start` instead of dropping it. + ## 0.5.0 ### Breaking changes diff --git a/packages/tugboat/README.md b/packages/tugboat/README.md index 4522b6f..31b02d1 100644 --- a/packages/tugboat/README.md +++ b/packages/tugboat/README.md @@ -5,7 +5,7 @@ checkpoints around meaningful interactions, compact structural anchors, route transitions, scrolling evidence, and optional viewport semantic maps. Capture can be sent to the local exploration WebSocket, the HTTP collector, or both. -The current package version is `0.5.0`. Session JSON writers emit schema +The current package version is `0.5.3`. Session JSON writers emit schema version `9`; compatibility readers should accept versions `6` through `9`. Structural fingerprints use fingerprint schema version `6`. @@ -153,11 +153,37 @@ Identity contract: - `captureSessionId` (`session.id`) — SDK-generated emitted evidence session - `collectorSessionId` — stamped after HTTP `session_start` acceptance - `explorationRunId` — exploration control-plane ID from config +- `traitsId` — collector-issued traits dictionary id after `setTraits` / session responses `TugboatReplay.activeSessionId` remains as a deprecated alias for `activationRequestId`. Inspect `TugboatReplay.health` for sink/outbox/screenshot budget pressure without reading protected content. +### User traits and user id + +When an HTTP collector is configured, register a full traits snapshot (not a +partial merge) via `POST /v1/sessions`: + +```dart +await TugboatReplay.setTraits({ + 'plan': 'pro', + 'seatCount': 3, +}); + +await TugboatReplay.setUserId(currentUserId); +``` + +- `setTraits` debounces `traits_updated` (3s) after start acceptance, or + `session_identify` when combined with a pending user change. Caches + `traitsId` and stamps it on event batches. While `session_start` is pending, + updates memory only (folded into start at send time). +- `setUserId` debounces `user_changed` (3s) after start acceptance, or + `session_identify` when combined with a pending traits change. Unchanged ids + are ignored. While `session_start` is pending, updates memory only. +- Pre-activate calls are retained in memory and included on the next + `session_start` when present. Pending debounced updates flush on `session_end`. + There is no `/v1/identify` route. + Optional durable HTTP delivery (Collector only, default off): ```dart @@ -263,7 +289,7 @@ Emitted event types currently include: immutable `origin`, `result`, `attribution`, and `evidenceEventIds`; - legacy gesture peers (`stream: legacy_projection` when canonical is on): `tap`, `tap_settled`, `swipe`, `tap_outside_tree`, `tap_gesture_resolved`; -- lifecycle: `session_start`, `session_end`; +- lifecycle: `session_start`, `session_identify`, `session_end`; - input: `pointer_cancel` (`stream: evidence`); - state/navigation evidence (`stream: evidence`): `state_change`, `route_change` (claimed routes also carry `causedByInteractionId`); @@ -380,11 +406,12 @@ so it is suitable for health polling and cannot grow with session duration. ## Public surface -The supported import exports `TugboatReplay`, `TugboatNavigatorObserver`, -`TugboatReplayConfig`, capture/semantic/masking enums and policies, collector -configuration and host helpers, markers (`TugboatSensitive`, `TugboatTag`, -`TugboatSubView`, `TugboatInternal`), anchor and session models, the controller, -and `TugboatExplorationTransport`. +The supported import exports `TugboatReplay` (including `setTraits` / +`setUserId`), `TugboatNavigatorObserver`, `TugboatReplayConfig`, +capture/semantic/masking enums and policies, collector configuration and host +helpers, markers (`TugboatSensitive`, `TugboatTag`, `TugboatSubView`, +`TugboatInternal`), anchor and session models, the controller, and +`TugboatExplorationTransport`. `TugboatCaptureSink` and the built-in sink implementations are internal today; config supports only the WebSocket and HTTP destinations above. A stable custom diff --git a/packages/tugboat/example/pubspec.yaml b/packages/tugboat/example/pubspec.yaml index 621d096..70c8656 100644 --- a/packages/tugboat/example/pubspec.yaml +++ b/packages/tugboat/example/pubspec.yaml @@ -32,7 +32,7 @@ resolution: workspace dependencies: flutter: sdk: flutter - tugboat: ^0.5.0 + tugboat: ^0.5.3 # The following adds the Cupertino Icons font to your application. # Use with the CupertinoIcons class for iOS style icons. diff --git a/packages/tugboat/lib/src/collector_http_sink.dart b/packages/tugboat/lib/src/collector_http_sink.dart index 5490633..e03f2f3 100644 --- a/packages/tugboat/lib/src/collector_http_sink.dart +++ b/packages/tugboat/lib/src/collector_http_sink.dart @@ -16,7 +16,18 @@ class CollectorHttpSink implements TugboatCaptureSink { CollectorHttpSink({ required TugboatCollectorConfig config, http.Client? client, + Map? initialTraits, + String? initialTraitsId, + String? initialUserId, + @visibleForTesting Duration? identityDebounceDuration, }) : _config = config, + _userId = initialUserId ?? config.userId, + _traits = initialTraits == null + ? null + : Map.from(initialTraits), + _traitsId = initialTraitsId, + _identityDebounceDuration = + identityDebounceDuration ?? _defaultIdentityDebounceDuration, _client = _CollectorHttpClient( inner: client ?? http.Client(), apiKey: config.apiKey, @@ -41,29 +52,71 @@ class CollectorHttpSink implements TugboatCaptureSink { final List> _pendingEvents = []; final List<_PendingFrameUpload> _pendingFrames = []; final List>> _retryBatches = []; - _PendingSessionLifecycle? _pendingLifecycle; - _PendingSessionLifecycle? _pendingLifecycleTail; + final List<_PendingSessionLifecycle> _pendingLifecycle = []; + + /// Runtime user id (may change via [setUserId]). + String? _userId; + + /// Full traits bag last provided by the host (process-local). + Map? _traits; + + /// Collector-issued traits dictionary id. + String? _traitsId; + + /// How long to wait after the last identity change before posting. + /// + /// Boot flows often resolve userId and traits in separate calls (auth restore, + /// billing, feature flags). Debouncing merges those into one lifecycle POST + /// (`session_identify` when both change) instead of back-to-back + /// `user_changed` / `traits_updated`. + static const _defaultIdentityDebounceDuration = Duration(seconds: 3); + + final Duration _identityDebounceDuration; + + Timer? _identityDebounceTimer; + bool _userDirty = false; + bool _traitsDirty = false; + DateTime? _userTriggeredAt; + DateTime? _traitsTriggeredAt; Uri get _baseUri => Uri.parse(_config.baseUrl.replaceAll(RegExp(r'/+$'), '')); bool get _hasCollectorSessionId => _collectorSessionId != null && _collectorSessionId!.isNotEmpty; + /// Last collector-issued traits id, if any. + String? get traitsId => _traitsId; + + /// Last host-provided traits bag, if any. + Map? get traits => + _traits == null ? null : Map.unmodifiable(_traits!); + + /// Current runtime user id stamped on sessions and events. + String? get userId => _userId; + @override void startSession(TugboatSession session) { if (_disposed) return; _sessionEpoch += 1; _session = session; // Clear any prior collector-issued id so a new session cannot route to the old one. + // Traits / traitsId persist across sessions for the process lifetime. _collectorSessionId = null; _pendingEvents.clear(); _pendingFrames.clear(); _retryBatches.clear(); - _pendingLifecycle = null; - _pendingLifecycleTail = null; + _pendingLifecycle.clear(); _framesNeedRetry = false; + _cancelIdentityDebounce(); + _userDirty = false; + _traitsDirty = false; + _userTriggeredAt = null; + _traitsTriggeredAt = null; _scheduleFlushTimer(); - _enqueueSessionLifecycle('session_start', session.startedAt); + _enqueueSessionLifecycle( + TugboatCollectorSessionEventType.sessionStart.wireValue, + session.startedAt, + ); unawaited(_kickFlush()); } @@ -76,13 +129,13 @@ class CollectorHttpSink implements TugboatCaptureSink { } final session = _session!; - // sessionId is stamped at send time once the collector id is known. + // sessionId / traitsId are stamped at send time once known. _pendingEvents.add( mapTugboatEventToCollectorEvent( event: event, sessionStartedAt: session.startedAt, collectorConfig: _config, - userId: _config.userId, + userId: _userId, ), ); _trimPendingEvents(); @@ -149,11 +202,108 @@ class CollectorHttpSink implements TugboatCaptureSink { await flush(); } + /// Registers a full traits snapshot with the collector. + /// + /// No-ops when [traits] equals the cached bag via [mapEquals] (shallow: + /// nested maps/lists compared with `==`). While `session_start` is still + /// pending, updates memory only (folded into start at send time). Otherwise + /// debounces `traits_updated` or `session_identify` when combined with a + /// pending user change within [_identityDebounceDuration]. + Future setTraits(Map traits) async { + if (_disposed) return; + if (mapEquals(_traits, traits)) return; + _traits = Map.from(traits); + if (_session == null) return; + if (_isSessionStartPending()) return; + _traitsDirty = true; + _traitsTriggeredAt = DateTime.now(); + _scheduleIdentityDebounce(); + } + + /// Updates the runtime user id and notifies the collector. + /// + /// No-ops when [userId] equals the current runtime id. While `session_start` + /// is still pending, updates memory only (folded into start at send time). + /// Otherwise debounces `user_changed` or `session_identify` when combined + /// with a pending traits change within [_identityDebounceDuration]. + Future setUserId(String? userId) async { + if (_disposed) return; + if (userId == _userId) return; + _userId = userId; + if (_session == null) return; + if (_isSessionStartPending()) return; + _userDirty = true; + _userTriggeredAt = DateTime.now(); + _scheduleIdentityDebounce(); + } + + bool _isSessionStartPending() => _pendingLifecycle.any( + (p) => + p.eventType == TugboatCollectorSessionEventType.sessionStart.wireValue, + ); + + void _scheduleIdentityDebounce() { + _identityDebounceTimer?.cancel(); + _identityDebounceTimer = Timer(_identityDebounceDuration, () { + _identityDebounceTimer = null; + if (_disposed) return; + _enqueueCoalescedIdentityUpdate(); + unawaited(_drainLifecyclePosts()); + }); + } + + void _cancelIdentityDebounce() { + _identityDebounceTimer?.cancel(); + _identityDebounceTimer = null; + } + + /// Flushes any debounced identity update immediately (no-op when clean). + Future _flushIdentityDebounce() async { + _cancelIdentityDebounce(); + if (!_userDirty && !_traitsDirty) return; + _enqueueCoalescedIdentityUpdate(); + await _drainLifecyclePosts(); + } + + DateTime _coalescedIdentityTriggeredAt() { + if (_userDirty && _traitsDirty) { + final userAt = _userTriggeredAt; + final traitsAt = _traitsTriggeredAt; + if (userAt != null && traitsAt != null) { + return userAt.isAfter(traitsAt) ? userAt : traitsAt; + } + return userAt ?? traitsAt ?? DateTime.now(); + } + if (_userDirty) return _userTriggeredAt ?? DateTime.now(); + return _traitsTriggeredAt ?? DateTime.now(); + } + + void _enqueueCoalescedIdentityUpdate() { + if (_disposed || _session == null) return; + if (!_userDirty && !_traitsDirty) return; + + final eventType = _userDirty && _traitsDirty + ? TugboatCollectorSessionEventType.sessionIdentify.wireValue + : _userDirty + ? TugboatCollectorSessionEventType.userChanged.wireValue + : TugboatCollectorSessionEventType.traitsUpdated.wireValue; + + _userDirty = false; + _traitsDirty = false; + _enqueueSessionLifecycle(eventType, _coalescedIdentityTriggeredAt()); + _userTriggeredAt = null; + _traitsTriggeredAt = null; + } + @override Future endSession() async { if (_disposed || _session == null) return; + await _flushIdentityDebounce(); await flush(); - _enqueueSessionLifecycle('session_end', DateTime.now()); + _enqueueSessionLifecycle( + TugboatCollectorSessionEventType.sessionEnd.wireValue, + DateTime.now(), + ); await _drainLifecyclePosts(); _cancelFlushTimer(); } @@ -162,12 +312,16 @@ class CollectorHttpSink implements TugboatCaptureSink { void dispose() { _disposed = true; _cancelFlushTimer(); + _cancelIdentityDebounce(); + _userDirty = false; + _traitsDirty = false; + _userTriggeredAt = null; + _traitsTriggeredAt = null; _client.close(); _pendingEvents.clear(); _pendingFrames.clear(); _retryBatches.clear(); - _pendingLifecycle = null; - _pendingLifecycleTail = null; + _pendingLifecycle.clear(); _flushInFlight = null; _session = null; _collectorSessionId = null; @@ -189,31 +343,27 @@ class CollectorHttpSink implements TugboatCaptureSink { } void _enqueueSessionLifecycle(String eventType, DateTime triggeredAt) { - final post = _PendingSessionLifecycle( - eventType: eventType, - triggeredAt: triggeredAt, + _pendingLifecycle.add( + _PendingSessionLifecycle(eventType: eventType, triggeredAt: triggeredAt), ); - if (_pendingLifecycle == null) { - _pendingLifecycle = post; - return; - } - _pendingLifecycleTail = post; } Future _drainLifecyclePosts() async { - while (!_disposed && - (_pendingLifecycle != null || _pendingLifecycleTail != null)) { - final head = _pendingLifecycle; + while (!_disposed && _pendingLifecycle.isNotEmpty) { + final head = _pendingLifecycle.first; await flush(); // Stop only when the head was not accepted (still retrying). Advancing // from session_start → session_end must continue draining. - if (identical(head, _pendingLifecycle)) return; + if (_pendingLifecycle.isNotEmpty && + identical(head, _pendingLifecycle.first)) { + return; + } } } Future _flushLifecyclePosts() async { - final pending = _pendingLifecycle; - if (pending == null || _disposed) return; + if (_pendingLifecycle.isEmpty || _disposed) return; + final pending = _pendingLifecycle.first; final epoch = _sessionEpoch; final result = await _sendSessionLifecycle( @@ -223,8 +373,7 @@ class CollectorHttpSink implements TugboatCaptureSink { ); if (!_isCurrentEpoch(epoch)) return; if (result == _SendResult.accepted) { - _pendingLifecycle = _pendingLifecycleTail; - _pendingLifecycleTail = null; + _pendingLifecycle.removeAt(0); } } @@ -238,16 +387,29 @@ class CollectorHttpSink implements TugboatCaptureSink { if (session == null) return _SendResult.accepted; // session_start uses the local id; later lifecycle uses the collector id when known. - final sessionId = eventType == 'session_start' + final sessionId = + eventType == TugboatCollectorSessionEventType.sessionStart.wireValue ? session.id : (_collectorSessionId ?? session.id); + final includeFullTraits = + _traits != null && + (eventType == TugboatCollectorSessionEventType.sessionStart.wireValue || + eventType == + TugboatCollectorSessionEventType.sessionIdentify.wireValue || + eventType == + TugboatCollectorSessionEventType.traitsUpdated.wireValue || + eventType == + TugboatCollectorSessionEventType.userChanged.wireValue); + final body = mapTugboatSessionLifecycleToCollectorSession( eventType: eventType, sessionId: sessionId, triggeredAt: triggeredAt, config: _config, - userId: _config.userId, + userId: _userId, + traits: includeFullTraits ? _traits : null, + traitsId: includeFullTraits ? null : _traitsId, ); try { @@ -258,13 +420,35 @@ class CollectorHttpSink implements TugboatCaptureSink { ); final result = _classifyResponse(response.statusCode); - if (result == _SendResult.accepted && eventType == 'session_start') { - final decoded = jsonDecode(response.body) as Map; - final serverId = decoded['sessionId'] as String?; - if (_isCurrentEpoch(epoch)) { - _collectorSessionId = (serverId != null && serverId.isNotEmpty) - ? serverId - : session.id; + // Status alone decides acceptance (same as events/frames). Body is + // optional enrichment; empty or non-JSON must not turn 202 into retry. + if (result == _SendResult.accepted && _isCurrentEpoch(epoch)) { + final raw = response.body.trim(); + if (raw.isEmpty) { + if (eventType == + TugboatCollectorSessionEventType.sessionStart.wireValue) { + _collectorSessionId ??= session.id; + } + } else { + try { + final decoded = jsonDecode(raw) as Map; + if (eventType == + TugboatCollectorSessionEventType.sessionStart.wireValue) { + final serverId = decoded['sessionId'] as String?; + _collectorSessionId = (serverId != null && serverId.isNotEmpty) + ? serverId + : session.id; + } + final responseTraitsId = decoded['traitsId'] as String?; + if (responseTraitsId != null && responseTraitsId.isNotEmpty) { + _traitsId = responseTraitsId; + } + } on Object { + if (eventType == + TugboatCollectorSessionEventType.sessionStart.wireValue) { + _collectorSessionId ??= session.id; + } + } } } return result; @@ -316,8 +500,12 @@ class CollectorHttpSink implements TugboatCaptureSink { final sessionId = _collectorSessionId; if (sessionId == null || sessionId.isEmpty) return _SendResult.retry; + final traitsId = _traitsId; for (final event in events) { event['sessionId'] = sessionId; + if (traitsId != null && traitsId.isNotEmpty) { + event['traitsId'] = traitsId; + } } try { diff --git a/packages/tugboat/lib/src/collector_mapper.dart b/packages/tugboat/lib/src/collector_mapper.dart index 53515c3..09e1b65 100644 --- a/packages/tugboat/lib/src/collector_mapper.dart +++ b/packages/tugboat/lib/src/collector_mapper.dart @@ -2,12 +2,26 @@ import 'anchors.dart'; import 'collector_config.dart'; import 'models.dart'; +/// Wire values for `POST /v1/sessions` `eventType`. +enum TugboatCollectorSessionEventType { + sessionStart('session_start'), + sessionIdentify('session_identify'), + sessionEnd('session_end'), + traitsUpdated('traits_updated'), + userChanged('user_changed'); + + const TugboatCollectorSessionEventType(this.wireValue); + + final String wireValue; +} + Map mapTugboatEventToCollectorEvent({ required TugboatEvent event, required DateTime sessionStartedAt, required TugboatCollectorConfig collectorConfig, String? sessionId, String? userId, + String? traitsId, }) { final triggeredAt = sessionStartedAt.add(Duration(milliseconds: event.atMs)); @@ -34,6 +48,7 @@ Map mapTugboatEventToCollectorEvent({ if (event.actionId != null) 'actionId': event.actionId, if (event.beforeFrame != null) 'beforeFrame': event.beforeFrame, if (event.afterFrame != null) 'afterFrame': event.afterFrame, + if (traitsId != null) 'traitsId': traitsId, 'stateAnchor': event.stateAnchor?.toJson() ?? {}, 'targetAnchor': event.targetAnchor?.toJson() ?? {}, if (event.result != null) 'result': event.result!.name, @@ -61,6 +76,8 @@ Map mapTugboatSessionLifecycleToCollectorSession({ required DateTime triggeredAt, required TugboatCollectorConfig config, String? userId, + Map? traits, + String? traitsId, }) { return { 'sessionId': sessionId, @@ -73,6 +90,9 @@ Map mapTugboatSessionLifecycleToCollectorSession({ 'device': config.deviceInfo.toJson(), 'ipInfo': config.ipInfo.toJson(), 'locale': config.locale.toJson(), + // Full traits bag wins over traitsId pass-through. + if (traits != null) 'traits': traits, + if (traits == null && traitsId != null) 'traitsId': traitsId, }; } diff --git a/packages/tugboat/lib/src/controller.dart b/packages/tugboat/lib/src/controller.dart index a3f244c..dd425b2 100644 --- a/packages/tugboat/lib/src/controller.dart +++ b/packages/tugboat/lib/src/controller.dart @@ -736,7 +736,17 @@ class TugboatReplayController extends ChangeNotifier { required GlobalKey boundaryKey, this.activationRequestId, this.sessionEpoch = 0, - }) : _boundaryKey = boundaryKey; + Map? initialTraits, + String? initialTraitsId, + String? initialUserId, + bool initialUserIdOverride = false, + }) : _boundaryKey = boundaryKey, + _initialTraits = initialTraits == null + ? null + : Map.from(initialTraits), + _initialTraitsId = initialTraitsId, + _initialUserId = initialUserId, + _initialUserIdOverride = initialUserIdOverride; final TugboatReplayConfig config; final GlobalKey _boundaryKey; @@ -747,6 +757,11 @@ class TugboatReplayController extends ChangeNotifier { /// Monotonic gate epoch fencing evidence to this capture mount. final int sessionEpoch; + Map? _initialTraits; + final String? _initialTraitsId; + String? _initialUserId; + bool _initialUserIdOverride; + final Stopwatch _clock = Stopwatch(); Future _queue = Future.value(); int _queuedTaskCount = 0; @@ -1188,8 +1203,16 @@ class TugboatReplayController extends ChangeNotifier { } final collectorConfig = config.collector; if (collectorConfig != null) { + // [_initialTraits] / [_initialUserId] may have been updated by + // setTraits/setUserId after construction but before initialize. + final userId = _initialUserIdOverride + ? _initialUserId + : collectorConfig.withUserId(config.userId).userId; _collectorHttpSink = CollectorHttpSink( - config: collectorConfig.withUserId(config.userId), + config: collectorConfig.withUserId(userId), + initialTraits: _initialTraits, + initialTraitsId: _initialTraitsId, + initialUserId: userId, ); TugboatCaptureSink httpSink = _collectorHttpSink!; if (config.outbox.enabled) { @@ -1219,6 +1242,34 @@ class TugboatReplayController extends ChangeNotifier { await _outboxStore?.clear(); } + /// See [TugboatReplay.setTraits]. + Future setTraits(Map traits) { + final sink = _collectorHttpSink; + if (sink != null) return sink.setTraits(traits); + // Capture can call identify after the controller mounts but before + // [initialize] builds the HTTP sink — retain for session_start. + _initialTraits = Map.from(traits); + return Future.value(); + } + + /// See [TugboatReplay.setUserId]. + Future setUserId(String? userId) { + final sink = _collectorHttpSink; + if (sink != null) return sink.setUserId(userId); + _initialUserId = userId; + _initialUserIdOverride = true; + return Future.value(); + } + + /// Collector traits id cached on the HTTP sink, if any. + String? get collectorTraitsId => _collectorHttpSink?.traitsId; + + /// Collector traits bag cached on the HTTP sink, if any. + Map? get collectorTraits => _collectorHttpSink?.traits; + + /// Runtime user id on the HTTP sink, if any. + String? get collectorUserId => _collectorHttpSink?.userId; + TugboatSdkHealth healthSnapshot() { final outbox = _outboxStore; return TugboatSdkHealth( diff --git a/packages/tugboat/lib/src/sdk_version.dart b/packages/tugboat/lib/src/sdk_version.dart index 3544f9c..f933218 100644 --- a/packages/tugboat/lib/src/sdk_version.dart +++ b/packages/tugboat/lib/src/sdk_version.dart @@ -1,3 +1,3 @@ // Keep this in sync with packages/tugboat/pubspec.yaml. The SDK version test // reads pubspec.yaml directly so release bumps fail fast if this drifts. -const tugboatSdkVersion = '0.5.0'; +const tugboatSdkVersion = '0.5.3'; diff --git a/packages/tugboat/lib/src/tugboat.dart b/packages/tugboat/lib/src/tugboat.dart index 88bb429..0849bdb 100644 --- a/packages/tugboat/lib/src/tugboat.dart +++ b/packages/tugboat/lib/src/tugboat.dart @@ -1,6 +1,7 @@ import 'dart:async'; import 'dart:io'; +import 'package:flutter/foundation.dart'; import 'package:flutter/material.dart'; import 'capture_boundary.dart'; @@ -35,6 +36,12 @@ class TugboatReplay { debugLabel: 'tugboat-capture-boundary', ); + /// Process-local identity retained across controller mount/unmount. + static Map? _pendingTraits; + static String? _pendingTraitsId; + static String? _pendingUserId; + static bool _pendingUserIdSet = false; + /// Convenience root [NavigatorObserver]. Prefer this for the app's primary /// Navigator. static final TugboatNavigatorObserver navigatorObserver = @@ -74,11 +81,76 @@ class TugboatReplay { static set disabled(bool value) { _lifecycle.setDisabled(value); if (value) { + _syncIdentityFromController(); _controller?.dispose(); _controller = null; } } + /// Registers a full user-traits snapshot with the collector. + /// + /// When a capture session is active and the bag changes, debounces lifecycle + /// posts (3s). Combined with a pending user change, posts + /// `session_identify`; otherwise `traits_updated`. While `session_start` is + /// still pending, updates memory only (folded into start at send time). + static Future setTraits(Map traits) async { + if (mapEquals(_pendingTraits, traits)) return; + _pendingTraits = Map.from(traits); + final controller = _controller; + if (controller == null) return; + await controller.setTraits(traits); + _pendingTraitsId = controller.collectorTraitsId ?? _pendingTraitsId; + } + + /// Updates the runtime user id used on collector sessions and events. + /// + /// Always records a remount override via [hasPendingUserIdOverride]. Collector + /// posting no-ops when [userId] equals the current runtime id. When a capture + /// session is active and the id changes, debounces lifecycle posts (3s). + /// Combined with a pending traits change, posts `session_identify`; otherwise + /// `user_changed`. While `session_start` is still pending, updates memory only. + static Future setUserId(String? userId) async { + _pendingUserId = userId; + _pendingUserIdSet = true; + final controller = _controller; + if (controller == null) return; + await controller.setUserId(userId); + _pendingTraitsId = controller.collectorTraitsId ?? _pendingTraitsId; + } + + /// Pending traits bag applied when the next [CollectorHttpSink] is created. + static Map? get pendingTraits => _pendingTraits == null + ? null + : Map.unmodifiable(_pendingTraits!); + + /// Pending collector traits id applied when the next sink is created. + static String? get pendingTraitsId => _pendingTraitsId; + + /// Whether [setUserId] has been called (including with `null`). + static bool get hasPendingUserIdOverride => _pendingUserIdSet; + + /// Pending user id from [setUserId], when [hasPendingUserIdOverride] is true. + static String? get pendingUserId => _pendingUserId; + + static void _syncIdentityFromController() { + final controller = _controller; + if (controller == null) return; + final traits = controller.collectorTraits; + if (traits != null) { + _pendingTraits = Map.from(traits); + } + final traitsId = controller.collectorTraitsId; + if (traitsId != null && traitsId.isNotEmpty) { + _pendingTraitsId = traitsId; + } + // Only refresh when the host explicitly called setUserId. Config-applied + // collectorUserId must not promote hasPendingUserIdOverride, or remounts + // ignore updated TugboatReplayConfig.userId. + if (_pendingUserIdSet) { + _pendingUserId = controller.collectorUserId; + } + } + /// Whether capture machinery is allowed to run ([disabled] is `false`). static bool get isEnabled => !_lifecycle.disabled; @@ -137,10 +209,15 @@ class TugboatReplay { /// Resets lifecycle state between tests. @visibleForTesting static void resetForTest() { + _syncIdentityFromController(); _controller?.dispose(); _controller = null; debugConfigureControllerForTest = null; _lifecycle.resetForTest(); + _pendingTraits = null; + _pendingTraitsId = null; + _pendingUserId = null; + _pendingUserIdSet = false; } } @@ -325,6 +402,10 @@ class _TugboatReplayRootState extends State<_TugboatReplayRoot> boundaryKey: TugboatReplay._boundaryKey, activationRequestId: widget.activationRequestId, sessionEpoch: widget.sessionEpoch, + initialTraits: TugboatReplay.pendingTraits, + initialTraitsId: TugboatReplay.pendingTraitsId, + initialUserId: TugboatReplay.pendingUserId, + initialUserIdOverride: TugboatReplay.hasPendingUserIdOverride, ); TugboatReplay._controller = controller; TugboatReplay.debugConfigureControllerForTest?.call(controller); @@ -410,6 +491,7 @@ class _TugboatReplayRootState extends State<_TugboatReplayRoot> WidgetsBinding.instance.removeObserver(this); inputCapture?.dispose(); if (identical(TugboatReplay._controller, controller)) { + TugboatReplay._syncIdentityFromController(); TugboatReplay._controller = null; } controller.dispose(); diff --git a/packages/tugboat/pubspec.yaml b/packages/tugboat/pubspec.yaml index 58b6b1c..69019a7 100644 --- a/packages/tugboat/pubspec.yaml +++ b/packages/tugboat/pubspec.yaml @@ -1,7 +1,7 @@ name: tugboat description: >- Screenshot-based session replay with compact interaction anchors for Tugboat. -version: 0.5.0 +version: 0.5.3 repository: https://github.com/blendto/tugboat-flutter issue_tracker: https://github.com/blendto/tugboat-flutter/issues homepage: https://github.com/blendto/tugboat-flutter diff --git a/packages/tugboat/test/collector_http_sink_test.dart b/packages/tugboat/test/collector_http_sink_test.dart index 1a9c457..9d7192b 100644 --- a/packages/tugboat/test/collector_http_sink_test.dart +++ b/packages/tugboat/test/collector_http_sink_test.dart @@ -21,6 +21,12 @@ void main() { var sessionFailuresRemaining = 0; var eventResponseDelay = Duration.zero; var frameResponseDelay = Duration.zero; + String? sessionResponseTraitsId; + + /// When non-null, written verbatim as the `/v1/sessions` response body + /// (including `''` for empty). When null, the default JSON acceptance map + /// is used. + String? sessionResponseBody; final collectorConfig = TugboatCollectorConfig( baseUrl: 'http://127.0.0.1:0', @@ -57,6 +63,8 @@ void main() { sessionFailuresRemaining = 0; eventResponseDelay = Duration.zero; frameResponseDelay = Duration.zero; + sessionResponseTraitsId = null; + sessionResponseBody = null; server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); baseUri = Uri.parse('http://127.0.0.1:${server.port}'); @@ -79,9 +87,22 @@ void main() { return 503; })() : sessionStatus; - request.response - ..statusCode = status - ..write(jsonEncode({'accepted': true, 'sessionId': 'sess_server'})); + request.response.statusCode = status; + final overrideBody = sessionResponseBody; + if (overrideBody != null) { + if (overrideBody.isNotEmpty) { + request.response.write(overrideBody); + } + } else { + request.response.write( + jsonEncode({ + 'accepted': true, + 'sessionId': 'sess_server', + if (sessionResponseTraitsId != null) + 'traitsId': sessionResponseTraitsId, + }), + ); + } } else if (path == '/v1/events/batch') { final body = jsonDecode(await utf8.decoder.bind(request).join()) as Map; final events = (body['events'] as List) @@ -184,6 +205,29 @@ void main() { expect(headers['X-Sdk-Version'], tugboatSdkVersion); } + /// Short debounce so identity tests avoid ~3s wall-clock sleeps. + const testIdentityDebounce = Duration(milliseconds: 40); + + CollectorHttpSink createIdentitySink({ + Map? initialTraits, + String? initialTraitsId, + String? initialUserId, + }) { + return CollectorHttpSink( + config: configForServer(), + initialTraits: initialTraits, + initialTraitsId: initialTraitsId, + initialUserId: initialUserId, + identityDebounceDuration: testIdentityDebounce, + ); + } + + Future awaitIdentityDebounce() async { + await Future.delayed( + testIdentityDebounce + const Duration(milliseconds: 40), + ); + } + test('collector defaults flush low-volume events every 3 seconds', () { expect( TugboatCollectorConfig( @@ -556,6 +600,56 @@ void main() { sink.dispose(); }); + test('accepts empty 202 lifecycle body without retrying', () async { + sessionResponseBody = ''; + final sink = CollectorHttpSink(config: configForServer()); + final session = createSession(); + sink.startSession(session); + + await Future.delayed(const Duration(milliseconds: 50)); + expect(sessionPosts.map((post) => post['eventType']), ['session_start']); + + // Empty session_start body falls back to local session id for uploads. + for (var i = 0; i < 10; i++) { + sink.recordEvent(createEvent(i)); + } + await Future.delayed(const Duration(milliseconds: 100)); + expect(batchPosts, hasLength(1)); + expect(batchPosts.first.first['sessionId'], session.id); + + await sink.endSession(); + expect(sessionPosts.map((post) => post['eventType']), [ + 'session_start', + 'session_end', + ]); + + final postsAfterEnd = sessionPosts.length; + await sink.flush(); + await Future.delayed(const Duration(milliseconds: 50)); + expect(sessionPosts, hasLength(postsAfterEnd)); + sink.dispose(); + }); + + test('accepts non-JSON 202 lifecycle body without retrying', () async { + sessionResponseBody = 'not-json'; + final sink = CollectorHttpSink(config: configForServer()); + sink.startSession(createSession()); + + await Future.delayed(const Duration(milliseconds: 50)); + expect(sessionPosts, hasLength(1)); + + await sink.endSession(); + expect(sessionPosts.map((post) => post['eventType']), [ + 'session_start', + 'session_end', + ]); + + final postsAfterEnd = sessionPosts.length; + await sink.flush(); + expect(sessionPosts, hasLength(postsAfterEnd)); + sink.dispose(); + }); + test('session lifecycle retries after transient failure', () async { sessionStatus = 503; final sink = CollectorHttpSink(config: configForServer()); @@ -716,4 +810,387 @@ void main() { expect(framePosts.single['frameNos'], ['2', '3', '4']); sink.dispose(); }); + + test( + 'setTraits posts traits_updated, caches traitsId, stamps next events', + () async { + sessionResponseTraitsId = 'trt_abc'; + final sink = createIdentitySink(); + final session = createSession(); + sink.startSession(session); + await Future.delayed(const Duration(milliseconds: 50)); + expect(sessionPosts, hasLength(1)); + + await sink.setTraits({'plan': 'pro', 'seatCount': 3}); + await awaitIdentityDebounce(); + expect(sessionPosts, hasLength(2)); + final traitsPost = sessionPosts.last; + expect(traitsPost['eventType'], 'traits_updated'); + expect(traitsPost['traits'], {'plan': 'pro', 'seatCount': 3}); + expect(traitsPost.containsKey('traitsId'), isFalse); + expect(traitsPost['sessionId'], 'sess_server'); + expect(sink.traitsId, 'trt_abc'); + + sink.recordEvent(createEvent(0)); + await sink.flush(); + await Future.delayed(const Duration(milliseconds: 50)); + + expect(batchPosts, isNotEmpty); + expect(batchPosts.last.first['traitsId'], 'trt_abc'); + sink.dispose(); + }, + ); + + test('session_start includes pre-set traits bag', () async { + sessionResponseTraitsId = 'trt_from_start'; + final sink = CollectorHttpSink( + config: configForServer(), + initialTraits: {'plan': 'free'}, + ); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + + expect(sessionPosts, hasLength(1)); + expect(sessionPosts.first['eventType'], 'session_start'); + expect(sessionPosts.first['traits'], {'plan': 'free'}); + expect(sessionPosts.first.containsKey('traitsId'), isFalse); + expect(sink.traitsId, 'trt_from_start'); + sink.dispose(); + }); + + test('setTraits before startSession lands on session_start', () async { + sessionResponseTraitsId = 'trt_pre'; + final sink = CollectorHttpSink(config: configForServer()); + await sink.setTraits({'plan': 'starter'}); + expect(sessionPosts, isEmpty); + + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + + expect(sessionPosts, hasLength(1)); + expect(sessionPosts.single['eventType'], 'session_start'); + expect(sessionPosts.single['traits'], {'plan': 'starter'}); + expect( + sessionPosts.where((post) => post['eventType'] == 'traits_updated'), + isEmpty, + ); + sink.dispose(); + }); + + test( + 'setTraits while session_start pending updates start payload only', + () async { + sessionStatus = 503; + final sink = CollectorHttpSink(config: configForServer()); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + // Failed attempts are still recorded by the test server. + expect( + sessionPosts.where((post) => post['eventType'] == 'session_start'), + isNotEmpty, + ); + + await sink.setTraits({'plan': 'pro'}); + expect( + sessionPosts.where((post) => post['eventType'] == 'traits_updated'), + isEmpty, + ); + + sessionStatus = 202; + sessionResponseTraitsId = 'trt_pending'; + await sink.flush(); + await Future.delayed(const Duration(milliseconds: 50)); + + final starts = sessionPosts + .where((post) => post['eventType'] == 'session_start') + .toList(); + expect(starts.last['traits'], {'plan': 'pro'}); + expect( + sessionPosts.where((post) => post['eventType'] == 'traits_updated'), + isEmpty, + ); + sink.dispose(); + }, + ); + + test( + 'new session_start includes traits bag after mid-session setTraits', + () async { + sessionResponseTraitsId = 'trt_mid'; + final sink = createIdentitySink(); + sink.startSession(createSession(id: 'session-a')); + await Future.delayed(const Duration(milliseconds: 50)); + expect(sessionPosts, hasLength(1)); + + await sink.setTraits({'plan': 'pro'}); + await awaitIdentityDebounce(); + expect(sessionPosts, hasLength(2)); + expect(sessionPosts.last['eventType'], 'traits_updated'); + expect(sessionPosts.last['traits'], {'plan': 'pro'}); + + await sink.endSession(); + sessionPosts.clear(); + + sink.startSession(createSession(id: 'session-b')); + await Future.delayed(const Duration(milliseconds: 50)); + + expect(sessionPosts.first['eventType'], 'session_start'); + expect(sessionPosts.first['traits'], {'plan': 'pro'}); + expect(sessionPosts.first.containsKey('traitsId'), isFalse); + sink.dispose(); + }, + ); + + test('session lifecycle without traits bag sends cached traitsId', () async { + final sink = CollectorHttpSink( + config: configForServer(), + initialTraitsId: 'trt_cached', + ); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + + expect(sessionPosts.first['eventType'], 'session_start'); + expect(sessionPosts.first['traitsId'], 'trt_cached'); + expect(sessionPosts.first.containsKey('traits'), isFalse); + + await sink.endSession(); + final endPost = sessionPosts.last; + expect(endPost['eventType'], 'session_end'); + expect(endPost['traitsId'], 'trt_cached'); + expect(endPost.containsKey('traits'), isFalse); + sink.dispose(); + }); + + test('setUserId posts user_changed with cached traits', () async { + sessionResponseTraitsId = 'trt_user'; + final sink = createIdentitySink( + initialTraits: {'plan': 'pro'}, + initialUserId: 'user_a', + ); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + + await sink.setUserId('user_b'); + await awaitIdentityDebounce(); + expect(sink.userId, 'user_b'); + final changed = sessionPosts.last; + expect(changed['eventType'], 'user_changed'); + expect(changed['userId'], 'user_b'); + expect(changed['traits'], {'plan': 'pro'}); + expect(changed.containsKey('traitsId'), isFalse); + sink.dispose(); + }); + + test('setUserId before startSession lands on session_start', () async { + final sink = CollectorHttpSink(config: configForServer()); + await sink.setUserId('user_pre'); + expect(sessionPosts, isEmpty); + expect(sink.userId, 'user_pre'); + + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + + expect(sessionPosts, hasLength(1)); + expect(sessionPosts.single['eventType'], 'session_start'); + expect(sessionPosts.single['userId'], 'user_pre'); + expect( + sessionPosts.where((post) => post['eventType'] == 'user_changed'), + isEmpty, + ); + sink.dispose(); + }); + + test( + 'setUserId while session_start pending updates start payload only', + () async { + sessionStatus = 503; + final sink = CollectorHttpSink(config: configForServer()); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + expect( + sessionPosts.where((post) => post['eventType'] == 'session_start'), + isNotEmpty, + ); + + await sink.setUserId('user_pending'); + expect(sink.userId, 'user_pending'); + expect( + sessionPosts.where((post) => post['eventType'] == 'user_changed'), + isEmpty, + ); + + sessionStatus = 202; + await sink.flush(); + await Future.delayed(const Duration(milliseconds: 50)); + + final starts = sessionPosts + .where((post) => post['eventType'] == 'session_start') + .toList(); + expect(starts.last['userId'], 'user_pending'); + expect( + sessionPosts.where((post) => post['eventType'] == 'user_changed'), + isEmpty, + ); + sink.dispose(); + }, + ); + + test('setUserId no-ops when user id is unchanged', () async { + final sink = createIdentitySink(initialUserId: 'user_a'); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + expect(sessionPosts, hasLength(1)); + expect(sessionPosts.single['eventType'], 'session_start'); + + await sink.setUserId('user_a'); + expect(sink.userId, 'user_a'); + expect(sessionPosts, hasLength(1)); + + await sink.setUserId(null); + await awaitIdentityDebounce(); + expect(sink.userId, isNull); + expect(sessionPosts, hasLength(2)); + expect(sessionPosts.last['eventType'], 'user_changed'); + expect(sessionPosts.last['userId'], isNull); + + await sink.setUserId(null); + expect(sessionPosts, hasLength(2)); + sink.dispose(); + }); + + test('setTraits no-ops when traits bag is unchanged', () async { + sessionResponseTraitsId = 'trt_skip'; + final sink = createIdentitySink(); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + expect(sessionPosts, hasLength(1)); + + await sink.setTraits({'plan': 'pro'}); + await awaitIdentityDebounce(); + expect(sessionPosts, hasLength(2)); + expect(sessionPosts.last['eventType'], 'traits_updated'); + expect(sessionPosts.last['traits'], {'plan': 'pro'}); + + await sink.setTraits({'plan': 'pro'}); + expect(sessionPosts, hasLength(2)); + + await sink.setTraits({'plan': 'enterprise'}); + await awaitIdentityDebounce(); + expect(sessionPosts, hasLength(3)); + expect(sessionPosts.last['eventType'], 'traits_updated'); + expect(sessionPosts.last['traits'], {'plan': 'enterprise'}); + sink.dispose(); + }); + + test( + 'setUserId then setTraits within debounce posts session_identify once', + () async { + sessionResponseTraitsId = 'trt_both'; + final sink = createIdentitySink(); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + expect(sessionPosts, hasLength(1)); + + await sink.setUserId('user_coalesce'); + await sink.setTraits({'plan': 'pro', 'seatCount': 2}); + await awaitIdentityDebounce(); + + expect(sessionPosts, hasLength(2)); + expect(sessionPosts.last['eventType'], 'session_identify'); + expect(sessionPosts.last['userId'], 'user_coalesce'); + expect(sessionPosts.last['traits'], {'plan': 'pro', 'seatCount': 2}); + expect( + sessionPosts.where((post) => post['eventType'] == 'user_changed'), + isEmpty, + ); + expect( + sessionPosts.where((post) => post['eventType'] == 'traits_updated'), + isEmpty, + ); + sink.dispose(); + }, + ); + + test('debounced identity uses first dirty time as triggeredAt', () async { + final sink = createIdentitySink(); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + + final before = DateTime.now(); + await sink.setTraits({'plan': 'pro'}); + final afterCall = DateTime.now(); + await awaitIdentityDebounce(); + final afterDebounce = DateTime.now(); + + expect(sessionPosts, hasLength(2)); + expect(sessionPosts.last['eventType'], 'traits_updated'); + final triggeredAt = DateTime.parse( + sessionPosts.last['triggeredAt'] as String, + ); + expect( + triggeredAt.isAfter(before.subtract(const Duration(milliseconds: 50))), + isTrue, + ); + expect( + triggeredAt.isBefore(afterCall.add(const Duration(milliseconds: 50))), + isTrue, + ); + expect( + triggeredAt.isBefore( + afterDebounce.subtract(const Duration(milliseconds: 20)), + ), + isTrue, + ); + sink.dispose(); + }); + + test( + 'coalesced identity triggeredAt reflects latest dirty mark when both change', + () async { + final sink = createIdentitySink(); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + + await sink.setUserId('user_at'); + await Future.delayed(const Duration(milliseconds: 15)); + final beforeTraits = DateTime.now(); + await sink.setTraits({'plan': 'pro'}); + final afterTraits = DateTime.now(); + await awaitIdentityDebounce(); + + expect(sessionPosts.last['eventType'], 'session_identify'); + final triggeredAt = DateTime.parse( + sessionPosts.last['triggeredAt'] as String, + ); + expect( + triggeredAt.isAfter( + beforeTraits.subtract(const Duration(milliseconds: 50)), + ), + isTrue, + ); + expect( + triggeredAt.isBefore(afterTraits.add(const Duration(milliseconds: 50))), + isTrue, + ); + sink.dispose(); + }, + ); + + test('endSession flushes debounced identity before session_end', () async { + sessionResponseTraitsId = 'trt_end'; + final sink = createIdentitySink(); + sink.startSession(createSession()); + await Future.delayed(const Duration(milliseconds: 50)); + + await sink.setTraits({'plan': 'pro'}); + expect(sessionPosts, hasLength(1)); + + await sink.endSession(); + expect(sessionPosts.map((post) => post['eventType']), [ + 'session_start', + 'traits_updated', + 'session_end', + ]); + sink.dispose(); + }); } diff --git a/packages/tugboat/test/collector_mapper_test.dart b/packages/tugboat/test/collector_mapper_test.dart index c5d4033..c5a2af2 100644 --- a/packages/tugboat/test/collector_mapper_test.dart +++ b/packages/tugboat/test/collector_mapper_test.dart @@ -142,7 +142,7 @@ void main() { test('maps session lifecycle payloads for collector sessions endpoint', () { final mapped = mapTugboatSessionLifecycleToCollectorSession( - eventType: 'session_start', + eventType: TugboatCollectorSessionEventType.sessionStart.wireValue, sessionId: 'sess_123', triggeredAt: DateTime.utc(2026, 6, 19), config: collectorConfig, @@ -159,6 +159,70 @@ void main() { expect((mapped['locale'] as Map)['language'], 'en'); expect(mapped['platform'], 'ios'); expect(mapped['fingerprintSchemaVersion'], tugboatFingerprintSchemaVersion); + expect(mapped.containsKey('traits'), isFalse); + expect(mapped.containsKey('traitsId'), isFalse); + }); + + test('session map prefers full traits bag over traitsId', () { + final mapped = mapTugboatSessionLifecycleToCollectorSession( + eventType: TugboatCollectorSessionEventType.traitsUpdated.wireValue, + sessionId: 'sess_123', + triggeredAt: DateTime.utc(2026, 6, 19), + config: collectorConfig, + traits: {'plan': 'pro'}, + traitsId: 'trt_ignored', + ); + + expect(mapped['eventType'], 'traits_updated'); + expect(mapped['traits'], {'plan': 'pro'}); + expect(mapped.containsKey('traitsId'), isFalse); + }); + + test('session map sends traitsId when no traits bag is provided', () { + final mapped = mapTugboatSessionLifecycleToCollectorSession( + eventType: TugboatCollectorSessionEventType.sessionEnd.wireValue, + sessionId: 'sess_123', + triggeredAt: DateTime.utc(2026, 6, 19), + config: collectorConfig, + traitsId: 'trt_cached', + ); + + expect(mapped['traitsId'], 'trt_cached'); + expect(mapped.containsKey('traits'), isFalse); + }); + + test('event map includes optional traitsId', () { + final mapped = mapTugboatEventToCollectorEvent( + event: TugboatEvent(id: 'event-1', atMs: 0, type: 'tap'), + sessionStartedAt: DateTime.utc(2026, 6, 19), + collectorConfig: collectorConfig, + traitsId: 'trt_evt', + ); + + expect(mapped['traitsId'], 'trt_evt'); + }); + + test('session event type wire values match collector contract', () { + expect( + TugboatCollectorSessionEventType.sessionIdentify.wireValue, + 'session_identify', + ); + expect( + TugboatCollectorSessionEventType.sessionStart.wireValue, + 'session_start', + ); + expect( + TugboatCollectorSessionEventType.sessionEnd.wireValue, + 'session_end', + ); + expect( + TugboatCollectorSessionEventType.traitsUpdated.wireValue, + 'traits_updated', + ); + expect( + TugboatCollectorSessionEventType.userChanged.wireValue, + 'user_changed', + ); }); test('keeps deprecated packageName legacy constructor compatibility', () { diff --git a/packages/tugboat/test/tugboat_replay_test.dart b/packages/tugboat/test/tugboat_replay_test.dart index d852e1d..3b8b085 100644 --- a/packages/tugboat/test/tugboat_replay_test.dart +++ b/packages/tugboat/test/tugboat_replay_test.dart @@ -1,5 +1,6 @@ import 'dart:async'; import 'dart:convert'; +import 'dart:io'; import 'dart:typed_data'; import 'package:flutter/material.dart'; @@ -1176,6 +1177,177 @@ void main() { expect(TugboatReplay.activationRequestId, 'req-b'); }); + testWidgets( + 'config userId applies on remount when setUserId was never called', + (tester) async { + addTearDown(TugboatReplay.resetForTest); + + late HttpServer server; + await tester.runAsync(() async { + server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + server.listen((request) async { + request.response + ..statusCode = 202 + ..write(jsonEncode({'accepted': true, 'sessionId': 'sess_server'})); + await request.response.close(); + }); + }); + addTearDown(() async { + await server.close(force: true); + }); + + TugboatCollectorConfig collectorConfig() => TugboatCollectorConfig( + baseUrl: 'http://127.0.0.1:${server.port}', + apiKey: 'pmk_test', + eventFlushInterval: const Duration(hours: 1), + appInfo: const TugboatCollectorAppInfo( + name: 'Example App', + version: '1.0.0', + buildNumber: '1', + installationId: 'inst_1', + appId: 'com.example.app', + ), + deviceInfo: const TugboatCollectorDeviceInfo( + id: 'device_client', + platform: 'ios', + screenSize: TugboatCollectorScreenSize(width: 390, height: 844), + screenDensity: 3, + screenDpi: 460, + screenPixelDensity: 3, + ), + ipInfo: const TugboatCollectorIpInfo(ip: '127.0.0.1'), + locale: const TugboatCollectorLocaleInfo(language: 'en'), + ); + + Future pumpWithUserId(String userId) async { + await tester.pumpWidget( + MaterialApp( + builder: (context, child) => TugboatReplay.wrapApp( + config: _testConfig.copyWith( + profile: TugboatCaptureProfile.dormant, + userId: userId, + collector: collectorConfig(), + ), + child: child!, + ), + home: const Scaffold(body: Text('Identity')), + ), + ); + await tester.pump(); + } + + await pumpWithUserId('user_a'); + TugboatReplay.activate( + activationRequestId: 'req-user-a', + profile: TugboatCaptureProfile.exploration, + ); + await _waitForCaptures(tester); + expect(TugboatReplay.controller!.collectorUserId, 'user_a'); + expect(TugboatReplay.hasPendingUserIdOverride, isFalse); + + TugboatReplay.deactivate(); + await tester.pump(); + await tester.pump(const Duration(milliseconds: 50)); + expect(TugboatReplay.controller, isNull); + expect(TugboatReplay.hasPendingUserIdOverride, isFalse); + + await pumpWithUserId('user_b'); + TugboatReplay.activate( + activationRequestId: 'req-user-b', + profile: TugboatCaptureProfile.exploration, + ); + await _waitForCaptures(tester); + expect(TugboatReplay.controller!.collectorUserId, 'user_b'); + expect(TugboatReplay.hasPendingUserIdOverride, isFalse); + }, + ); + + testWidgets('setUserId override survives remount over config userId', ( + tester, + ) async { + addTearDown(TugboatReplay.resetForTest); + + late HttpServer server; + await tester.runAsync(() async { + server = await HttpServer.bind(InternetAddress.loopbackIPv4, 0); + server.listen((request) async { + request.response + ..statusCode = 202 + ..write(jsonEncode({'accepted': true, 'sessionId': 'sess_server'})); + await request.response.close(); + }); + }); + addTearDown(() async { + await server.close(force: true); + }); + + TugboatCollectorConfig collectorConfig() => TugboatCollectorConfig( + baseUrl: 'http://127.0.0.1:${server.port}', + apiKey: 'pmk_test', + eventFlushInterval: const Duration(hours: 1), + appInfo: const TugboatCollectorAppInfo( + name: 'Example App', + version: '1.0.0', + buildNumber: '1', + installationId: 'inst_1', + appId: 'com.example.app', + ), + deviceInfo: const TugboatCollectorDeviceInfo( + id: 'device_client', + platform: 'ios', + screenSize: TugboatCollectorScreenSize(width: 390, height: 844), + screenDensity: 3, + screenDpi: 460, + screenPixelDensity: 3, + ), + ipInfo: const TugboatCollectorIpInfo(ip: '127.0.0.1'), + locale: const TugboatCollectorLocaleInfo(language: 'en'), + ); + + Future pumpWithUserId(String userId) async { + await tester.pumpWidget( + MaterialApp( + builder: (context, child) => TugboatReplay.wrapApp( + config: _testConfig.copyWith( + profile: TugboatCaptureProfile.dormant, + userId: userId, + collector: collectorConfig(), + ), + child: child!, + ), + home: const Scaffold(body: Text('Override')), + ), + ); + await tester.pump(); + } + + await pumpWithUserId('user_a'); + TugboatReplay.activate( + activationRequestId: 'req-override-a', + profile: TugboatCaptureProfile.exploration, + ); + await _waitForCaptures(tester); + await tester.runAsync(() => TugboatReplay.setUserId('runtime')); + expect(TugboatReplay.controller!.collectorUserId, 'runtime'); + expect(TugboatReplay.hasPendingUserIdOverride, isTrue); + + TugboatReplay.deactivate(); + await tester.pump(); + await tester.pump(const Duration(milliseconds: 50)); + expect(TugboatReplay.controller, isNull); + expect(TugboatReplay.hasPendingUserIdOverride, isTrue); + expect(TugboatReplay.pendingUserId, 'runtime'); + + await pumpWithUserId('user_b'); + TugboatReplay.activate( + activationRequestId: 'req-override-b', + profile: TugboatCaptureProfile.exploration, + ); + await _waitForCaptures(tester); + expect(TugboatReplay.controller!.collectorUserId, 'runtime'); + expect(TugboatReplay.hasPendingUserIdOverride, isTrue); + }); + testWidgets('identical activate request is idempotent', (tester) async { addTearDown(TugboatReplay.resetForTest);