diff --git a/projects/ngx-ui-watchtower/DESIGN.md b/projects/ngx-ui-watchtower/DESIGN.md index e3964e6..690f281 100644 --- a/projects/ngx-ui-watchtower/DESIGN.md +++ b/projects/ngx-ui-watchtower/DESIGN.md @@ -96,8 +96,7 @@ fails where it always did: the logger needs a log provider but no destination, scenarios need a destination and no log provider. The services are root singletons, so each concern exists once per injector, and all of them share the one destination and log provider. `settled$` is the core's -`onSettled` as an Observable that completes on destroy. It listens to the -core only while subscribed, so no record is copied for nobody; `traceScenario` is the +`onSettled` as an Observable that completes on destroy; `traceScenario` is the core's `./rxjs` operator, reporting through this package's dev-mode reporter. Framework-free applications compose the same three classes with `uwtCreateTelemetry`. @@ -211,10 +210,7 @@ providers: [ the same names. The dependency runs one way: the core imports neither Angular nor this package, which a lint rule enforces. The library build compiles against the core's built declarations, not its source, so it sees - the core exactly as an application does. An application upgrading from - 0.4 changes nothing. Another framework's adapter needs only the core's - public entry points, as this package does; - `docs/ADAPTERS.md` is a short guide. + the core exactly as an application does. - **What the core owns, this package re-exports.** The engine, the models and name registries, the configuration types and defaults (`uwtResolveTelemetryConfig` completes every `with*` feature), redaction, @@ -233,29 +229,10 @@ providers: [ broadcast plumbing stay internal so they can change freely. Redaction is the exception: a custom sink needs `uwtRedactMetadata`, `uwtNormalizeError`, `uwtScrubString` and `uwtRedactConfigFor`. -- **The RUM sink is the core's too.** Its logic — the lazy `aws-rum-web` - import, startup retries, credential refresh, the pre-init buffer, the - settings check — lives in `@absaoss-cps/ui-watchtower/rum`. This package's - `/rum` keeps `UWT_RUM_CREDENTIALS_PROVIDER`, `provideUwtTelemetryRumSink()` - and `UwtRumTelemetrySink`, an injectable subclass of the core's. - **The secondary entry keeps private copies of a few utilities.** ng-packagr fixes each entry point's `rootDir` to its own `src`, so `@absaoss-cps/ngx-ui-watchtower/rum` can't import the main entry's - internals by relative path and carries small copies of what it needs: its - Angular helpers and the reporter. -- **What is published is tested as published.** `npm run smoke` installs - both tarballs into consumers outside the workspace: a framework-free one - with neither Angular nor `aws-rum-web` installed, and an Angular - application, prerendered as a fragment without `aws-rum-web` — which then - holds no RUM code — and as a shell with `/rum`, whose RUM client is a chunk - loaded on demand. -- **Both packages share one version and release together.** Release Please - links their versions: a change to either releases both, and the same - release PR moves this package's dependency on the core to `^` the new - version. The release workflow publishes the core first and this package - only after it, so this package never depends on a core version npm doesn't - have. Publishing skips what npm already has, so running the workflow again - on the release tag completes a partial release. + internals by relative path and carries small copies of what it needs. - **Test doubles live in the specs that use them**, declared inline, never in a shared file and never exported. The one browser gap needing a stub is `BroadcastChannel`, which jsdom doesn't implement. @@ -1631,15 +1608,12 @@ convention, for an application that wants the pre-fallback behavior back. ## 11. Error handling and performance - **Telemetry never breaks the application.** Every public entry point is - wrapped and never rethrows. The core hands a suppressed error (`'error'`), - or data it had to drop (`'warn'`), to the `UwtReporter` it is given. This - package gives it one that writes to the console, prefixed + wrapped and never rethrows. The core prints nothing by itself: it hands a + suppressed error, or data it had to drop, to the `UwtReporter` it is + given. This package gives it one that writes to the console, prefixed `[ngx-ui-watchtower]`, in development mode only, so bugs surface in - development and production stays silent. What cannot work as configured - (`'config'`) — a second broadcast host, a host in a forwarding realm, RUM - settings of the wrong type — is printed in production too, as it always - was; without a reporter, the core prints it itself. Even a broken - `console.error` cannot turn a suppressed failure into a throw. + development and production stays silent. Even a broken `console.error` + cannot turn a suppressed failure into a throw. - **Fail-open.** A broker outage, expired credentials or an SDK throw all leave the application behaving exactly as with healthy telemetry. A failed broker request at startup is retried with backoff for five minutes. A @@ -2209,12 +2183,6 @@ providers: [ Code inside a fragment does not change: same services, same calls. One session, one budget, one bundle. -| Realm | Destination | RUM and `UWT_RUM_CREDENTIALS_PROVIDER` | `UWT_LOG_API_PROVIDER` | -| -------------- | --------------------------------------------------- | -------------------------------------- | -------------------------------------- | -| Standalone app | Its own: `provideUwtTelemetryRumSink()`, or another | The app's | The app's | -| Shell | Its own, plus `provideUwtTelemetryBroadcastHost()` | The shell's: the only ones on the page | The shell's, also for fragments' logs | -| Fragment | `provideUwtTelemetrySink('broadcast')` | None: the shell's records for it | Bound by `'broadcast'`, to the shell's | - Either side may be built without Angular. The protocol — the messages, the host, the forwarding sink and log provider — lives in `@absaoss-cps/ui-watchtower`, and this package's classes are injectable diff --git a/projects/ngx-ui-watchtower/README.md b/projects/ngx-ui-watchtower/README.md index dd7a479..dbe32a6 100644 --- a/projects/ngx-ui-watchtower/README.md +++ b/projects/ngx-ui-watchtower/README.md @@ -352,6 +352,9 @@ providers: [ ``` Fragment code doesn't change. Give each fragment its own `application` name. +A shell or a fragment needn't be Angular: one built on +[`@absaoss-cps/ui-watchtower`](https://www.npmjs.com/package/@absaoss-cps/ui-watchtower) +connects the same way. ## Seeing what is sent diff --git a/projects/ngx-ui-watchtower/public-api.snapshot.json b/projects/ngx-ui-watchtower/public-api.snapshot.json index 87fb6e9..274fa7f 100644 --- a/projects/ngx-ui-watchtower/public-api.snapshot.json +++ b/projects/ngx-ui-watchtower/public-api.snapshot.json @@ -96,6 +96,7 @@ "UwtBroadcastLogApiProvider": [ "class", "new (): UwtBroadcastLogApiProvider", + "dispose(): void", "flush(): void", "ngOnDestroy(): void", "query(filter: UwtLogQuery): Promise", @@ -104,6 +105,7 @@ "UwtBroadcastTelemetrySink": [ "class", "new (): UwtBroadcastTelemetrySink", + "dispose(): void", "flush(beacon?: boolean | undefined): void", "getSessionId(): string | undefined", "getUserId(): string | undefined", @@ -395,6 +397,7 @@ "UwtTelemetryBroadcastHost": [ "class", "new (): UwtTelemetryBroadcastHost", + "dispose(): void", "get received: number", "ngOnDestroy(): void" ], diff --git a/projects/ngx-ui-watchtower/src/lib/providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider.ts b/projects/ngx-ui-watchtower/src/lib/providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider.ts index db8cda5..00dbe08 100644 --- a/projects/ngx-ui-watchtower/src/lib/providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider.ts +++ b/projects/ngx-ui-watchtower/src/lib/providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider.ts @@ -1,29 +1,13 @@ -import { inject, Injectable, NgZone, OnDestroy } from '@angular/core'; +import { Injectable, OnDestroy } from '@angular/core'; +import { UwtBroadcastLogApiProvider as CoreBroadcastLogApiProvider } from '@absaoss-cps/ui-watchtower'; +import { uwtBroadcastConnectionOptions } from '../../sinks/uwt-broadcast/uwt-broadcast.messages'; import { - UwtLogApiProvider, - UwtLogQuery, - UwtLogRecord, - uwtUuid -} from '@absaoss-cps/ui-watchtower'; -import { uwtSafeVoid } from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; -import { - UwtBroadcastConnection, - uwtConnectBroadcastChannel, - uwtIsBroadcastLogRecord, - uwtIsBroadcastMessage -} from '../../sinks/uwt-broadcast/uwt-broadcast.messages'; - -/** - * How long a follower waits for the host to answer a log query before - * resolving to `[]`, in milliseconds. - */ -export const UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS = 10_000; + uwtDevModeReporter, + uwtRunInsideZone, + uwtRunOutsideZone +} from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; -/** A log query waiting for the host's answer. */ -interface UwtPendingLogQuery { - resolve: (records: UwtLogRecord[]) => void; - timer: ReturnType; -} +export { UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS } from '@absaoss-cps/ui-watchtower'; /** * Log API provider for a follower realm — the log counterpart of @@ -45,82 +29,20 @@ interface UwtPendingLogQuery { */ @Injectable() export class UwtBroadcastLogApiProvider - implements UwtLogApiProvider, OnDestroy + extends CoreBroadcastLogApiProvider + implements OnDestroy { - private readonly connection: UwtBroadcastConnection = - uwtConnectBroadcastChannel('broadcastLogApi'); - - private readonly pending = new Map(); - - /** Absent outside an application, e.g. a bare `Injector.create`. */ - private readonly zone = inject(NgZone, { optional: true }); - constructor() { - this.connection.onMessage((data) => this.onMessage(data)); - } - - /** @inheritdoc */ - send(record: UwtLogRecord): void { - this.connection.post({ kind: 'log', record }); - } - - /** @inheritdoc */ - query(filter: UwtLogQuery): Promise { - if (!this.connection.isOpen()) { - return Promise.resolve([]); - } - - return new Promise((resolve) => { - const id = uwtUuid(); - const timer = this.outsideZone(() => - setTimeout( - () => this.answer(id, []), - UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS - ) - ); - this.pending.set(id, { resolve, timer }); - this.connection.post({ kind: 'log-query', id, filter }); + super({ + ...uwtBroadcastConnectionOptions(), + runOutside: uwtRunOutsideZone(), + runInside: uwtRunInsideZone(), + report: uwtDevModeReporter }); } - /** @inheritdoc */ - flush(): void { - this.connection.post({ kind: 'log-flush' }); - } - /** Resolves every unanswered query to `[]`, then closes the channel. */ ngOnDestroy(): void { - for (const id of [...this.pending.keys()]) { - this.answer(id, []); - } - this.connection.close(); - } - - private onMessage(data: unknown): void { - uwtSafeVoid('broadcastLogApi.receive', () => { - if (!uwtIsBroadcastMessage(data) || data.kind !== 'log-query-result') { - return; - } - // Other followers' answers arrive here too; answer() ignores them. - this.answer(data.id, data.records.filter(uwtIsBroadcastLogRecord)); - }); - } - - private answer(id: string, records: UwtLogRecord[]): void { - const query = this.pending.get(id); - if (!query) { - return; - } - this.pending.delete(id); - clearTimeout(query.timer); - if (this.zone) { - this.zone.run(() => query.resolve(records)); - } else { - query.resolve(records); - } - } - - private outsideZone(fn: () => T): T { - return this.zone ? this.zone.runOutsideAngular(fn) : fn(); + this.dispose(); } } diff --git a/projects/ngx-ui-watchtower/src/lib/providers/uwt-telemetry-common.providers/uwt-telemetry-common.providers.ts b/projects/ngx-ui-watchtower/src/lib/providers/uwt-telemetry-common.providers/uwt-telemetry-common.providers.ts index 0d32998..25c7361 100644 --- a/projects/ngx-ui-watchtower/src/lib/providers/uwt-telemetry-common.providers/uwt-telemetry-common.providers.ts +++ b/projects/ngx-ui-watchtower/src/lib/providers/uwt-telemetry-common.providers/uwt-telemetry-common.providers.ts @@ -9,7 +9,10 @@ import { } from '@angular/core'; import { UwtTelemetryBroadcastHost } from '../../sinks/uwt-broadcast/uwt-broadcast-host.service'; import { UwtBroadcastTelemetrySink } from '../../sinks/uwt-broadcast/uwt-broadcast-telemetry.sink'; -import { UWT_BROADCAST_CHANNEL } from '../../sinks/uwt-broadcast/uwt-broadcast.messages'; +import { + UWT_BROADCAST_CHANNEL, + UwtBroadcastConnectionRef +} from '../../sinks/uwt-broadcast/uwt-broadcast.messages'; import { UwtNoopTelemetrySink } from '../../sinks/uwt-telemetry/uwt-noop-telemetry.sink/uwt-noop-telemetry.sink'; import { UWT_DEFAULT_TELEMETRY_CONFIG, @@ -376,6 +379,8 @@ export function provideUwtTelemetrySink( switch (mode) { case 'broadcast': return makeEnvironmentProviders([ + // One channel for the realm's sink and log provider. + UwtBroadcastConnectionRef, provideUwtTelemetryDestination(UwtBroadcastTelemetrySink), UwtBroadcastLogApiProvider, { diff --git a/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-host.service.ts b/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-host.service.ts index 7871ad0..c0c2813 100644 --- a/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-host.service.ts +++ b/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-host.service.ts @@ -1,39 +1,11 @@ import { inject, Injectable, OnDestroy } from '@angular/core'; import { - uwtClassifyTelemetryEvent, - UwtLogQuery, - UwtLogRecord, - uwtSessionElapsed, + UwtTelemetryBroadcastHost as CoreTelemetryBroadcastHost, UwtTelemetrySink } from '@absaoss-cps/ui-watchtower'; -import { - uwtSafe, - uwtSafeVoid, - uwtSafeVoidMaybeAsync -} from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; -import { - UwtBroadcastConnection, - uwtConnectBroadcastChannel, - uwtElectBroadcastHostLeader, - uwtIsBroadcastMessage -} from './uwt-broadcast.messages'; -import { uwtTelemetryEnabled } from '../../config/uwt-telemetry-common.config/uwt-telemetry-common.config'; -import { UwtBroadcastTelemetrySink } from './uwt-broadcast-telemetry.sink'; -import { UwtBroadcastLogApiProvider } from '../../providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider'; import { UWT_LOG_API_PROVIDER } from '../../providers/uwt-log-api.provider/uwt-log-api.provider'; - -/** The identity fields scenario, BI and log records carry. */ -interface UwtIdentified { - sessionId?: string; - userId?: string; - sessionElapsed?: number; -} - -/** - * Active host count per channel, scoped to this JS realm — distinguishes a - * true duplicate provider from another tab sharing the same channel. - */ -const hostsInThisRealm = new Map(); +import { uwtHostReporter } from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; +import { uwtBroadcastChannelOptions } from './uwt-broadcast.messages'; /** * Receives telemetry forwarded by follower realms and records it through this @@ -73,287 +45,21 @@ const hostsInThisRealm = new Map(); * @group Services */ @Injectable() -export class UwtTelemetryBroadcastHost implements OnDestroy { - private readonly sink = inject(UwtTelemetrySink); - private readonly logApi = inject(UWT_LOG_API_PROVIDER); - private readonly connection: UwtBroadcastConnection = - uwtConnectBroadcastChannel('broadcastHost'); - - /** Number of messages accepted, for tests and diagnostics. */ - private _received = 0; - - /** - * The session and user id last sent in an `identity` message. - * - * Compared against the sink's current values on every follower activity, so - * a late-resolving session id or user id gets announced once it exists. - */ - private lastAnnouncedSessionId?: string; - private lastAnnouncedUserId?: string; - - /** Whether this realm won the leader election; non-leaders stay passive. */ - private isLeader = false; - private releaseLeadership: () => void = () => undefined; - private readonly enabled = uwtTelemetryEnabled(); - +export class UwtTelemetryBroadcastHost + extends CoreTelemetryBroadcastHost + implements OnDestroy +{ constructor() { - if (!this.enabled) { - // Telemetry is off: no listening, no leader election. - return; - } - this.warnIfDuplicateInThisRealm(); - if (this.forwardsItsOwnRealm()) { - this.connection.close(); - return; - } - - this.connection.onMessage((data) => this.onMessage(data)); - this.releaseLeadership = uwtElectBroadcastHostLeader( - this.connection.channelName, - () => { - this.isLeader = true; - this.announceIdentity(); - } - ); - } - - /** How many follower messages have been accepted. */ - get received(): number { - return this._received; + super({ + sink: inject(UwtTelemetrySink), + logApi: inject(UWT_LOG_API_PROVIDER), + ...uwtBroadcastChannelOptions(), + report: uwtHostReporter + }); } /** @inheritdoc */ ngOnDestroy(): void { - if (this.enabled) { - this.forgetInThisRealm(); - } - this.releaseLeadership(); - this.connection.close(); - } - - private onMessage(data: unknown): void { - uwtSafeVoid('broadcastHost.receive', () => { - if (!this.isLeader || !uwtIsBroadcastMessage(data)) { - return; - } - - if ( - data.kind !== 'identity' && - data.kind !== 'identity-request' && - data.kind !== 'log-query-result' - ) { - this._received++; - } - - switch (data.kind) { - case 'event': - this.sink.record( - data.eventType, - this.stampEvent(data.eventType, data.payload), - data.metadata - ); - this.reannounceIfIdentityChanged(); - break; - case 'error': - this.sink.recordError(data.error, data.metadata); - this.reannounceIfIdentityChanged(); - break; - case 'user': - this.sink.setUserId(data.userId); - this.announceIdentity(); - break; - case 'flush': - this.sink.flush(data.beacon); - this.reannounceIfIdentityChanged(); - break; - case 'identity-request': - this.announceIdentity(); - break; - case 'identity': - // Another realm's host on this origin-wide channel — expected, - // not a problem. Same-document duplicates are caught by - // warnIfDuplicateInThisRealm() instead. - break; - case 'log': - this.deliverLog(this.withIdentity(data.record)); - this.reannounceIfIdentityChanged(); - break; - case 'log-flush': - uwtSafeVoidMaybeAsync('broadcastHost.logFlush', () => - this.logApi?.flush?.() - ); - break; - case 'log-query': - this.answerLogQuery(data.id, data.filter); - break; - case 'log-query-result': - // Hosts answer queries; they never ask them. - break; - } - }); - } - - /** - * A forwarded scenario or BI record with this realm's identity; any other - * payload as it is. - */ - private stampEvent(eventType: string, payload: object): object { - const { kind } = uwtClassifyTelemetryEvent(eventType, payload); - return kind === 'unknown' - ? payload - : this.withIdentity(payload as UwtIdentified, kind === 'scenario'); - } - - /** - * `record` with this realm's session and user id. A follower stamps the ids - * it was last told, which lag behind whenever they change here — the client - * starting after the follower asked, a session expiring, a sign-out — so its - * record could name a session other than the one it is sent in. With - * `measuresSession`, a changed session's `sessionElapsed` is measured again. - */ - private withIdentity( - record: T, - measuresSession = false - ): T { - const sessionId = this.currentSessionId(); - // '' is no user, as `setUserId('')` signs out. - const userId = this.currentUserId() || undefined; - if (record.sessionId === sessionId && record.userId === userId) { - return record; - } - // An id this realm doesn't have stays `undefined`, - // as in the library's own records. - const stamped: T = { ...record, sessionId, userId }; - if (measuresSession && record.sessionId !== sessionId) { - // The old session's value no longer applies; absent when unknown. - delete stamped.sessionElapsed; - const elapsed = uwtSessionElapsed(sessionId); - if (elapsed !== undefined) { - stamped.sessionElapsed = Math.round(elapsed); - } - } - return stamped; - } - - /** Hands a follower's record to this realm's log API provider. */ - private deliverLog(record: UwtLogRecord): void { - uwtSafeVoidMaybeAsync('broadcastHost.log', () => this.logApi.send(record)); - } - - /** - * Answers a follower's query from this realm's log API provider. Always - * answers — `[]` when there is no provider or it fails — so the follower - * isn't left waiting for its timeout. - */ - private answerLogQuery(id: string, filter: UwtLogQuery): void { - const reply = (records: unknown) => - this.connection.post({ - kind: 'log-query-result', - id, - records: Array.isArray(records) ? records : [] - }); - - const pending = uwtSafe( - 'broadcastHost.logQuery', - () => this.logApi.query(filter), - undefined - ); - - if (!pending) { - reply([]); - return; - } - // Promise.resolve: a provider breaking its contract with a plain value - // still gets an answer out. - Promise.resolve(pending).then(reply, () => reply([])); - } - - /** - * Whether this realm's own sink or log provider forwards over the - * channel. Hosting then would loop: the host receives its own realm's - * messages and hands them straight back to the channel. - */ - private forwardsItsOwnRealm(): boolean { - const forwards = - this.sink instanceof UwtBroadcastTelemetrySink || - this.logApi instanceof UwtBroadcastLogApiProvider; - - if (forwards) { - uwtSafeVoid('broadcastHost.forwardingRealmWarning', () => { - console.warn( - `[ngx-ui-watchtower] this realm forwards its own telemetry on channel "${this.connection.channelName}", so it cannot also host it; the host stays inactive. Provide the host only in the realm with the real sink and log provider` - ); - }); - } - return forwards; - } - - /** - * Warns when another host is already active on the same channel in this - * realm — a real misconfiguration. Not based on the `identity` broadcast, - * which can't distinguish this from a different, legitimate tab. - */ - private warnIfDuplicateInThisRealm(): void { - const channelName = this.connection.channelName; - const activeCount = hostsInThisRealm.get(channelName) ?? 0; - - if (activeCount > 0) { - uwtSafeVoid('broadcastHost.duplicateWarning', () => { - console.warn( - `[ngx-ui-watchtower] a second telemetry host is active on channel "${channelName}" in this document; only one realm should provide it` - ); - }); - } - - hostsInThisRealm.set(channelName, activeCount + 1); - } - - /** Releases this host's slot in {@link hostsInThisRealm}. */ - private forgetInThisRealm(): void { - const channelName = this.connection.channelName; - const activeCount = hostsInThisRealm.get(channelName) ?? 0; - - if (activeCount <= 1) { - hostsInThisRealm.delete(channelName); - } else { - hostsInThisRealm.set(channelName, activeCount - 1); - } - } - - private announceIdentity(): void { - const sessionId = this.currentSessionId(); - const userId = this.currentUserId(); - this.lastAnnouncedSessionId = sessionId; - this.lastAnnouncedUserId = userId; - this.connection.post({ kind: 'identity', sessionId, userId }); - } - - /** - * Re-announces identity when the sink's session id or user id has moved on - * from what followers were last told. - */ - private reannounceIfIdentityChanged(): void { - if ( - this.currentSessionId() !== this.lastAnnouncedSessionId || - this.currentUserId() !== this.lastAnnouncedUserId - ) { - this.announceIdentity(); - } - } - - private currentSessionId(): string | undefined { - return uwtSafe( - 'broadcastHost.getSessionId', - () => this.sink.getSessionId(), - undefined - ); - } - - private currentUserId(): string | undefined { - return uwtSafe( - 'broadcastHost.getUserId', - () => this.sink.getUserId(), - undefined - ); + this.dispose(); } } diff --git a/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-telemetry.sink.ts b/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-telemetry.sink.ts index ae9f424..9ba3775 100644 --- a/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-telemetry.sink.ts +++ b/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-telemetry.sink.ts @@ -1,16 +1,8 @@ import { inject, Injectable, OnDestroy } from '@angular/core'; -import { - UwtTelemetryError, - UwtTelemetryMetadata, - UwtTelemetrySink -} from '@absaoss-cps/ui-watchtower'; +import { UwtBroadcastTelemetrySink as CoreBroadcastTelemetrySink } from '@absaoss-cps/ui-watchtower'; import { UWT_TELEMETRY_IDENTITY } from '../../config/uwt-telemetry-common.config/uwt-telemetry-common.config'; -import { uwtSafeVoid } from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; -import { - UwtBroadcastConnection, - uwtConnectBroadcastChannel, - uwtIsBroadcastMessage -} from './uwt-broadcast.messages'; +import { uwtDevModeReporter } from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; +import { uwtBroadcastConnectionOptions } from './uwt-broadcast.messages'; /** * Telemetry sink for a follower realm — a micro-frontend fragment, a webview @@ -34,99 +26,19 @@ import { */ @Injectable() export class UwtBroadcastTelemetrySink - extends UwtTelemetrySink + extends CoreBroadcastTelemetrySink implements OnDestroy { - private readonly connection: UwtBroadcastConnection = - uwtConnectBroadcastChannel('broadcastSink'); - - /** The host's session id, once it has answered. */ - private sessionId?: string; - private userId?: string; - - private readonly config = inject(UWT_TELEMETRY_IDENTITY); - - /** - * This realm's own identity, stamped onto everything it forwards. - * - * The host records through its own RUM client, so without this, forwarded - * events would all be labelled as the host's realm. - */ - private readonly origin = { - application: this.config.application, - environment: this.config.environment, - appVersion: this.config.version - }; - constructor() { - super(); - - this.connection.onMessage((data) => this.onMessage(data)); - this.connection.post({ kind: 'identity-request' }); - } - - /** @inheritdoc */ - record( - eventType: string, - payload: object, - metadata?: UwtTelemetryMetadata - ): void { - this.connection.post({ - kind: 'event', - eventType, - payload, - metadata: { ...metadata, ...this.origin } - }); - } - - /** @inheritdoc */ - recordError(error: UwtTelemetryError, metadata?: UwtTelemetryMetadata): void { - this.connection.post({ - kind: 'error', - error, - metadata: { ...metadata, ...this.origin } + super({ + identity: inject(UWT_TELEMETRY_IDENTITY), + ...uwtBroadcastConnectionOptions(), + report: uwtDevModeReporter }); } - /** - * The host's session id. - * - * Returns `undefined` until the host answers. - * - * @returns the shared session id, when known - */ - getSessionId(): string | undefined { - return this.sessionId; - } - - /** @inheritdoc */ - setUserId(userId: string | undefined): void { - this.userId = userId; - this.connection.post({ kind: 'user', userId }); - } - - /** @inheritdoc */ - getUserId(): string | undefined { - return this.userId; - } - - /** @inheritdoc */ - flush(beacon = false): void { - this.connection.post({ kind: 'flush', beacon }); - } - /** @inheritdoc */ ngOnDestroy(): void { - this.connection.close(); - } - - private onMessage(data: unknown): void { - uwtSafeVoid('broadcastSink.receive', () => { - if (!uwtIsBroadcastMessage(data) || data.kind !== 'identity') { - return; - } - this.sessionId = data.sessionId; - this.userId = data.userId; - }); + this.dispose(); } } diff --git a/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.messages.ts b/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.messages.ts index 5249c4a..8e3f80e 100644 --- a/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.messages.ts +++ b/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.messages.ts @@ -1,23 +1,16 @@ -import { inject, InjectionToken } from '@angular/core'; +import { inject, Injectable, InjectionToken, OnDestroy } from '@angular/core'; import { - UWT_LOG_LEVEL_ORDER, - UwtLogQuery, - UwtLogRecord, - UwtTelemetryError, - UwtTelemetryMetadata + UWT_DEFAULT_BROADCAST_CHANNEL, + UwtBroadcastConnection, + uwtConnectBroadcastChannel as coreConnectBroadcastChannel } from '@absaoss-cps/ui-watchtower'; import { uwtTelemetryEnabled } from '../../config/uwt-telemetry-common.config/uwt-telemetry-common.config'; import { - uwtIsBrowser, - uwtSafeVoid + uwtDevModeReporter, + uwtIsBrowser } from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; -/** - * Default `BroadcastChannel` name used between telemetry realms. - * - * @group Utils - */ -export const UWT_DEFAULT_BROADCAST_CHANNEL = 'ngx-ui-watchtower'; +export { UWT_DEFAULT_BROADCAST_CHANNEL } from '@absaoss-cps/ui-watchtower'; /** * Overrides the channel name shared by the host and its followers. @@ -32,356 +25,65 @@ export const UWT_BROADCAST_CHANNEL = new InjectionToken( ); /** - * Everything that travels between telemetry realms. - * - * Structured-cloned by `BroadcastChannel`, so every payload must be plain - * data. + * Where this injector's realm connects: the channel name — the + * {@link UWT_BROADCAST_CHANNEL} override, or the default — and whether a + * channel may open at all: only in a browser, while telemetry is on. Must be + * called from an injection context. */ -export type UwtBroadcastMessage = - | { - kind: 'event'; - eventType: string; - payload: object; - metadata?: UwtTelemetryMetadata; - } - | { - kind: 'error'; - error: UwtTelemetryError; - metadata?: UwtTelemetryMetadata; - } - | { kind: 'user'; userId: string | undefined } - | { kind: 'flush'; beacon: boolean } - | { kind: 'identity-request' } - /** The host announcing shared identity. Always sends both fields, even when only one changed. */ - | { kind: 'identity'; sessionId?: string; userId?: string } - /** A follower's log record, for the host to hand to its log API provider. */ - | { kind: 'log'; record: UwtLogRecord } - /** A follower asking the host to flush its log API provider. */ - | { kind: 'log-flush' } - /** A follower reading records back through the host's log API provider. */ - | { kind: 'log-query'; id: string; filter: UwtLogQuery } - /** - * The host's answer to one `log-query`, matched by `id`. The records come - * from the host's backend unchecked; the asking follower keeps only those - * that pass {@link uwtIsBroadcastLogRecord}. - */ - | { kind: 'log-query-result'; id: string; records: unknown[] }; - -/** Every `kind` the union above accepts, for {@link uwtIsBroadcastMessage}. */ -const MESSAGE_KINDS: ReadonlySet = new Set([ - 'event', - 'error', - 'user', - 'flush', - 'identity-request', - 'identity', - 'log', - 'log-flush', - 'log-query', - 'log-query-result' -] as const); - -const LOG_LEVELS: ReadonlySet = new Set( - Object.keys(UWT_LOG_LEVEL_ORDER) -); - -/** Minimal `BroadcastChannel` surface this library relies on. */ -export interface UwtBroadcastChannelLike { - postMessage(message: unknown): void; - close(): void; - onmessage: ((event: { data: unknown }) => void) | null; -} - -/** - * Opens a broadcast channel, when the browser provides the API. - * - * Feature-detected: unavailable under jsdom (tests) and in server-side - * rendering. - */ -export function uwtOpenBroadcastChannel( - name: string -): UwtBroadcastChannelLike | undefined { - const Channel = ( - globalThis as { - BroadcastChannel?: new (name: string) => UwtBroadcastChannelLike; - } - ).BroadcastChannel; - - if (typeof Channel !== 'function') { - return undefined; - } - - try { - return new Channel(name); - } catch { - return undefined; - } -} - -/** Minimal Web Locks API surface this library relies on. */ -interface UwtLockManagerLike { - request(name: string, callback: () => Promise): Promise; -} - -/** - * Elects exactly one leader among same-origin realms on this channel, via - * the Web Locks API — prevents two shell tabs from both recording every - * forwarded message. Fails open (elects immediately) when Web Locks is - * unavailable. - * - * @returns releases the lock so the next queued realm can become leader. - */ -export function uwtElectBroadcastHostLeader( - channelName: string, - onElected: () => void -): () => void { - const locks = (globalThis as { navigator?: { locks?: UwtLockManagerLike } }) - .navigator?.locks; - - if (!locks) { - onElected(); - return () => undefined; - } - - let released = false; - let release: () => void = () => { - released = true; - }; - - const failOpen = () => { - if (!released) { - onElected(); - } +export function uwtBroadcastChannelOptions(): { + channelName: string; + enabled: boolean; +} { + return { + channelName: + inject(UWT_BROADCAST_CHANNEL, { optional: true }) ?? + UWT_DEFAULT_BROADCAST_CHANNEL, + enabled: uwtIsBrowser() && uwtTelemetryEnabled() }; - - try { - locks - .request( - `ngx-ui-watchtower-host:${channelName}`, - () => - new Promise((resolve) => { - if (released) { - resolve(); - return; - } - release = () => { - released = true; - resolve(); - }; - onElected(); - }) - ) - .catch(failOpen); - } catch { - failOpen(); - } - - return () => release(); -} - -/** A realm's live connection to the shared broadcast channel. */ -export interface UwtBroadcastConnection { - /** The channel name actually in use — the injected override, or the default. */ - readonly channelName: string; - - /** Whether a channel is open — false without `BroadcastChannel`, on the server, and once closed. */ - isOpen(): boolean; - - /** Sends a message to every other realm on this channel. No-op if unavailable. */ - post(message: UwtBroadcastMessage): void; - - /** Registers the handler for incoming messages. No-op if the channel could not be opened. */ - onMessage(handler: (data: unknown) => void): void; - - /** Closes the channel and stops any further delivery. Safe to call more than once. */ - close(): void; } /** - * Opens this realm's connection to the shared broadcast channel: resolves the - * channel name, feature-detects `BroadcastChannel`, and wraps every operation - * in the library's fail-open guard. - * - * Must be called from an injection context (a field initializer or a - * constructor). + * Opens this realm's connection to the shared broadcast channel, configured + * from this injector, reporting through the dev-mode reporter. Must be called + * from an injection context (a field initializer or a constructor). */ export function uwtConnectBroadcastChannel( operation: string ): UwtBroadcastConnection { - const channelName = - inject(UWT_BROADCAST_CHANNEL, { optional: true }) ?? - UWT_DEFAULT_BROADCAST_CHANNEL; - // No channel on the server or while telemetry is off: every call is a no-op. - let channel = - uwtIsBrowser() && uwtTelemetryEnabled() - ? uwtOpenBroadcastChannel(channelName) - : undefined; - - return { + const { channelName, enabled } = uwtBroadcastChannelOptions(); + return coreConnectBroadcastChannel({ channelName, - isOpen() { - return channel !== undefined; - }, - post(message) { - uwtSafeVoid(`${operation}.post`, () => channel?.postMessage(message)); - }, - onMessage(handler) { - if (channel) { - channel.onmessage = (event) => handler(event.data); - } - }, - close() { - uwtSafeVoid(`${operation}.close`, () => { - channel?.close(); - channel = undefined; - }); - } - }; -} - -/** - * Whether `value` is a valid `metadata` field: absent, or a non-array - * object whose values are all finite numbers, strings, booleans, or null — - * the same flat-primitives contract `uwtRedactMetadata` enforces on the way - * out. Without this, a same-origin sender (not necessarily this library's - * own sink) could post a nested object on the channel that reaches a - * receiving sink's own sanitizer — e.g. `UwtRumTelemetrySink.sanitize()` — - * unchecked, since that step trusts the shape rather than re-validating it. - */ -function isValidBroadcastMetadata(value: unknown): boolean { - if (value === undefined) { - return true; - } - if (typeof value !== 'object' || value === null || Array.isArray(value)) { - return false; - } - return Object.values(value).every( - (v) => - v === null || - typeof v === 'string' || - typeof v === 'boolean' || - (typeof v === 'number' && Number.isFinite(v)) - ); -} - -function isOptionalString(value: unknown): boolean { - return value === undefined || typeof value === 'string'; -} - -function isRecord(value: unknown): value is Record { - return typeof value === 'object' && value !== null && !Array.isArray(value); -} - -function isValidTelemetryError(value: unknown): boolean { - return ( - isRecord(value) && - typeof value.name === 'string' && - typeof value.message === 'string' && - isOptionalString(value.stack) - ); -} - -function isLogLevel(value: unknown): boolean { - return typeof value === 'string' && LOG_LEVELS.has(value); + open: enabled, + operation, + report: uwtDevModeReporter + }); } /** - * Whether `value` has the shape of a {@link UwtLogRecord}: the required - * strings, a known level, optional strings where the record has them, and - * metadata and error held to the same rules as on an `event` or `error` - * message. + * The one connection a realm's broadcast sink and log provider share: + * provided with them by `provideUwtTelemetrySink('broadcast')`, and closed + * with their injector. Internal. */ -export function uwtIsBroadcastLogRecord(value: unknown): value is UwtLogRecord { - return ( - isRecord(value) && - typeof value.timestamp === 'string' && - isLogLevel(value.level) && - typeof value.message === 'string' && - typeof value.application === 'string' && - typeof value.environment === 'string' && - typeof value.version === 'string' && - isOptionalString(value.logger) && - isOptionalString(value.context) && - isOptionalString(value.correlationId) && - isOptionalString(value.userId) && - isOptionalString(value.sessionId) && - isValidBroadcastMetadata(value.metadata) && - (value.error === undefined || isValidTelemetryError(value.error)) - ); -} +@Injectable() +export class UwtBroadcastConnectionRef implements OnDestroy { + readonly connection = uwtConnectBroadcastChannel('broadcast'); -function isValidLogQuery(value: unknown): boolean { - return ( - isRecord(value) && - isOptionalString(value.correlationId) && - isOptionalString(value.logger) && - (value.minLevel === undefined || isLogLevel(value.minLevel)) && - isOptionalString(value.from) && - isOptionalString(value.to) && - (value.limit === undefined || - (typeof value.limit === 'number' && Number.isFinite(value.limit))) - ); + /** @inheritdoc */ + ngOnDestroy(): void { + this.connection.close(); + } } /** - * Narrows an incoming `BroadcastChannel` payload to a telemetry message. - * - * Checks each kind's required fields, not just `kind`, so a same-named - * message from something else on the channel is rejected. + * How a broadcast sink or log provider connects: through its injector's + * shared connection when there is one, on its own otherwise. Must be called + * from an injection context. */ -export function uwtIsBroadcastMessage( - data: unknown -): data is UwtBroadcastMessage { - if (typeof data !== 'object' || data === null) { - return false; - } - - const message = data as Record; - const kind = message.kind as UwtBroadcastMessage['kind']; - if (!MESSAGE_KINDS.has(kind)) { - return false; - } - - switch (kind) { - case 'event': - return ( - typeof message.eventType === 'string' && - typeof message.payload === 'object' && - message.payload !== null && - isValidBroadcastMetadata(message.metadata) - ); - case 'error': - return ( - isValidTelemetryError(message.error) && - isValidBroadcastMetadata(message.metadata) - ); - case 'flush': - return typeof message.beacon === 'boolean'; - case 'user': - // Absent or `undefined` userId both mean "clear it". - return message.userId === undefined || typeof message.userId === 'string'; - case 'identity': - return ( - (message.sessionId === undefined || - typeof message.sessionId === 'string') && - (message.userId === undefined || typeof message.userId === 'string') - ); - case 'identity-request': - case 'log-flush': - return true; - case 'log': - return uwtIsBroadcastLogRecord(message.record); - case 'log-query': - return ( - typeof message.id === 'string' && - message.id !== '' && - isValidLogQuery(message.filter) - ); - case 'log-query-result': - return ( - typeof message.id === 'string' && - message.id !== '' && - Array.isArray(message.records) - ); - } +export function uwtBroadcastConnectionOptions(): + | { connection: UwtBroadcastConnection } + | { channelName: string; enabled: boolean } { + const shared = inject(UwtBroadcastConnectionRef, { optional: true }); + return shared + ? { connection: shared.connection } + : uwtBroadcastChannelOptions(); } diff --git a/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.spec.ts b/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.spec.ts index 6356206..ea28aba 100644 --- a/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.spec.ts +++ b/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.spec.ts @@ -31,9 +31,7 @@ import { import { UWT_BROADCAST_CHANNEL, UWT_DEFAULT_BROADCAST_CHANNEL, - uwtConnectBroadcastChannel, - uwtElectBroadcastHostLeader, - uwtIsBroadcastMessage + uwtConnectBroadcastChannel } from './uwt-broadcast.messages'; /** @@ -1087,327 +1085,6 @@ describe('broadcast telemetry across realms', () => { }); }); - describe('uwtElectBroadcastHostLeader', () => { - afterEach(() => LockManagerStub.uninstall()); - - it('should elect immediately when the Locks API is unavailable', () => { - const onElected = jest.fn(); - uwtElectBroadcastHostLeader('ngx-ui-watchtower', onElected); - - expect(onElected).toHaveBeenCalledTimes(1); - }); - - it('should elect once the lock is granted', () => { - LockManagerStub.install(); - const onElected = jest.fn(); - uwtElectBroadcastHostLeader('ngx-ui-watchtower', onElected); - - expect(onElected).toHaveBeenCalledTimes(1); - }); - - it('should fail open and still elect when request() itself rejects', async () => { - Object.defineProperty(globalThis.navigator, 'locks', { - value: { request: () => Promise.reject(new Error('not-fully-active')) }, - configurable: true - }); - const onElected = jest.fn(); - - uwtElectBroadcastHostLeader('ngx-ui-watchtower', onElected); - // Let the rejection's microtask settle. - await Promise.resolve(); - await Promise.resolve(); - - expect(onElected).toHaveBeenCalledTimes(1); - }); - - it('should fail open and still elect when request() itself throws synchronously', () => { - Object.defineProperty(globalThis.navigator, 'locks', { - value: { - request: () => { - throw new Error('locks unavailable in this context'); - } - }, - configurable: true - }); - const onElected = jest.fn(); - - expect(() => - uwtElectBroadcastHostLeader('ngx-ui-watchtower', onElected) - ).not.toThrow(); - expect(onElected).toHaveBeenCalledTimes(1); - }); - - it('should not elect a requester released while still queued, and should let the next requester through', async () => { - LockManagerStub.install(); - const onElectedHolder = jest.fn(); - const onElectedA = jest.fn(); - const onElectedB = jest.fn(); - - const releaseHolder = uwtElectBroadcastHostLeader( - 'ngx-ui-watchtower', - onElectedHolder - ); - const releaseA = uwtElectBroadcastHostLeader( - 'ngx-ui-watchtower', - onElectedA - ); - uwtElectBroadcastHostLeader('ngx-ui-watchtower', onElectedB); - - // A is released before ever being granted the lock. - releaseA(); - // The current holder releases, so the lock passes down the queue. - releaseHolder(); - for (let i = 0; i < 6; i++) { - await Promise.resolve(); - } - - expect(onElectedA).not.toHaveBeenCalled(); - expect(onElectedB).toHaveBeenCalledTimes(1); - }); - }); - - describe('uwtIsBroadcastMessage', () => { - it.each([null, undefined, 'string', 42, ['array']])( - 'should reject the non-object payload %p', - (data) => { - expect(uwtIsBroadcastMessage(data)).toBe(false); - } - ); - - it('should reject a payload with an unknown kind', () => { - expect(uwtIsBroadcastMessage({ kind: 'other library' })).toBe(false); - }); - - it('should accept a well-formed event message', () => { - expect( - uwtIsBroadcastMessage({ - kind: 'event', - eventType: 'com.uwt.bi', - payload: { eventName: 'x' } - }) - ).toBe(true); - }); - - it.each([ - { eventType: 'com.uwt.bi' }, - { payload: {} }, - { eventType: 1, payload: {} }, - { eventType: 'com.uwt.bi', payload: 'not-an-object' }, - { - eventType: 'com.uwt.bi', - payload: {}, - metadata: { nested: {} } - }, - { - eventType: 'com.uwt.bi', - payload: {}, - metadata: { list: [1, 2] } - }, - { - eventType: 'com.uwt.bi', - payload: {}, - metadata: 'not-an-object' - } - ])('should reject a malformed event message %p', (fields) => { - expect(uwtIsBroadcastMessage({ kind: 'event', ...fields })).toBe(false); - }); - - it('should accept an event message with flat-primitive metadata', () => { - expect( - uwtIsBroadcastMessage({ - kind: 'event', - eventType: 'com.uwt.bi', - payload: {}, - metadata: { count: 3, ok: true, note: 'x', blank: null } - }) - ).toBe(true); - }); - - it('should accept an event message with no metadata', () => { - expect( - uwtIsBroadcastMessage({ - kind: 'event', - eventType: 'com.uwt.bi', - payload: {} - }) - ).toBe(true); - }); - - it('should accept a well-formed error message', () => { - expect( - uwtIsBroadcastMessage({ - kind: 'error', - error: { name: 'Error', message: 'boom' } - }) - ).toBe(true); - }); - - it.each([ - {}, - { error: { name: 'Error' } }, - { error: { message: 'boom' } }, - { error: 'boom' }, - { error: { name: 'Error', message: 'boom', stack: 42 } }, - { - error: { name: 'Error', message: 'boom' }, - metadata: { nested: {} } - } - ])('should reject a malformed error message %p', (fields) => { - expect(uwtIsBroadcastMessage({ kind: 'error', ...fields })).toBe(false); - }); - - it('should accept an error message with flat-primitive metadata', () => { - expect( - uwtIsBroadcastMessage({ - kind: 'error', - error: { name: 'Error', message: 'boom' }, - metadata: { origin: 'checkout' } - }) - ).toBe(true); - }); - - it('should accept an error message with a string stack', () => { - expect( - uwtIsBroadcastMessage({ - kind: 'error', - error: { name: 'Error', message: 'boom', stack: 'at foo.ts:1' } - }) - ).toBe(true); - }); - - it('should accept a well-formed flush message', () => { - expect(uwtIsBroadcastMessage({ kind: 'flush', beacon: true })).toBe(true); - }); - - it('should reject a flush message with a non-boolean beacon', () => { - expect(uwtIsBroadcastMessage({ kind: 'flush', beacon: 'true' })).toBe( - false - ); - }); - - it.each([{ userId: 'u-1' }, { userId: undefined }, {}])( - 'should accept a well-formed user message %p', - (fields) => { - expect(uwtIsBroadcastMessage({ kind: 'user', ...fields })).toBe(true); - } - ); - - it('should reject a user message with a non-string, non-undefined userId', () => { - expect(uwtIsBroadcastMessage({ kind: 'user', userId: 42 })).toBe(false); - }); - - const logRecord = { - timestamp: '2026-01-01T00:00:00.000Z', - level: 'warn', - message: 'Careful', - application: 'cart', - environment: 'test', - version: '1.0.0' - }; - - it('should accept a well-formed log message', () => { - expect(uwtIsBroadcastMessage({ kind: 'log', record: logRecord })).toBe( - true - ); - expect( - uwtIsBroadcastMessage({ - kind: 'log', - record: { - ...logRecord, - logger: 'checkout', - context: 'Cart', - correlationId: 'c-1', - sessionId: 's', - userId: 'u', - metadata: { count: 1 }, - error: { name: 'Error', message: 'boom' } - } - }) - ).toBe(true); - }); - - it.each([ - { level: 'debug' }, - { level: 'toString' }, - { application: undefined }, - { message: 42 }, - { logger: 7 }, - { metadata: { nested: {} } }, - { error: { name: 'Error' } } - ])('should reject a log message whose record has %p', (fields) => { - expect( - uwtIsBroadcastMessage({ - kind: 'log', - record: { ...logRecord, ...fields } - }) - ).toBe(false); - }); - - it('should accept log-flush, log-query and log-query-result messages', () => { - expect(uwtIsBroadcastMessage({ kind: 'log-flush' })).toBe(true); - expect( - uwtIsBroadcastMessage({ - kind: 'log-query', - id: 'q-1', - filter: { correlationId: 'c-1', minLevel: 'warn', limit: 5 } - }) - ).toBe(true); - expect( - uwtIsBroadcastMessage({ - kind: 'log-query-result', - id: 'q-1', - records: [] - }) - ).toBe(true); - }); - - it.each([ - { id: '', filter: {} }, - { id: 'q-1', filter: null }, - { id: 'q-1', filter: { limit: 'ten' } }, - { id: 'q-1', filter: { minLevel: 'verbose' } } - ])('should reject a malformed log-query message %p', (fields) => { - expect(uwtIsBroadcastMessage({ kind: 'log-query', ...fields })).toBe( - false - ); - }); - - it('should reject a log-query-result without a records array', () => { - expect( - uwtIsBroadcastMessage({ - kind: 'log-query-result', - id: 'q-1', - records: 'none' - }) - ).toBe(false); - }); - - it('should accept an identity-request message with no other fields', () => { - expect(uwtIsBroadcastMessage({ kind: 'identity-request' })).toBe(true); - }); - - it.each([ - {}, - { sessionId: 's-1' }, - { userId: 'u-1' }, - { - sessionId: 's-1', - userId: 'u-1' - } - ])('should accept an identity message %p', (fields) => { - expect(uwtIsBroadcastMessage({ kind: 'identity', ...fields })).toBe(true); - }); - - it.each([{ sessionId: 42 }, { userId: 42 }, { sessionId: 42, userId: 42 }])( - 'should reject an identity message with a non-string, non-undefined field %p', - (fields) => { - expect(uwtIsBroadcastMessage({ kind: 'identity', ...fields })).toBe( - false - ); - } - ); - }); - describe('uwtConnectBroadcastChannel', () => { it('should default the channel name to UWT_DEFAULT_BROADCAST_CHANNEL', () => { const connection = TestBed.runInInjectionContext(() => diff --git a/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.wire.spec.ts b/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.wire.spec.ts index da002fe..2eb2305 100644 --- a/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.wire.spec.ts +++ b/projects/ngx-ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.wire.spec.ts @@ -1,12 +1,27 @@ import { DOCUMENT } from '@angular/common'; -import { Injectable, Injector, PLATFORM_ID } from '@angular/core'; +import { + createEnvironmentInjector, + EnvironmentInjector, + Injectable, + Injector, + PLATFORM_ID +} from '@angular/core'; +import { TestBed } from '@angular/core/testing'; import { existsSync, readFileSync, writeFileSync } from 'fs'; import { join } from 'path'; import { + UwtBroadcastLogApiProvider as CoreBroadcastLogApiProvider, + UwtBroadcastTelemetrySink as CoreBroadcastTelemetrySink, + uwtConnectBroadcastChannel, + uwtCreateTelemetry, UWT_DEFAULT_TELEMETRY_CONFIG, UwtLogApiProvider, + UwtLogger, UwtLogQuery, UwtLogRecord, + UwtScenario, + UwtScenarioOptions, + UwtTelemetryBroadcastHost as CoreTelemetryBroadcastHost, UwtTelemetryError, UwtTelemetryMetadata, UwtTelemetrySink @@ -20,6 +35,10 @@ import { UWT_LOG_CONFIG } from '../../config/uwt-log.config/uwt-log.config'; import { UWT_SCENARIO_TELEMETRY_CONFIG } from '../../config/uwt-scenario-telemetry.config/uwt-scenario-telemetry.config'; import { UWT_LOG_API_PROVIDER } from '../../providers/uwt-log-api.provider/uwt-log-api.provider'; import { UwtBroadcastLogApiProvider } from '../../providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider'; +import { + provideUwtTelemetry, + provideUwtTelemetrySink +} from '../../providers/uwt-telemetry-common.providers/uwt-telemetry-common.providers'; import { UwtBITelemetryService } from '../../services/uwt-bi-telemetry.service/uwt-bi-telemetry.service'; import { UwtLoggerService } from '../../services/uwt-logger.service/uwt-logger.service'; import { UwtScenarioTelemetryService } from '../../services/uwt-scenario-telemetry.service/uwt-scenario-telemetry.service'; @@ -28,14 +47,22 @@ import { UwtBroadcastTelemetrySink } from './uwt-broadcast-telemetry.sink'; /** * The broadcast protocol is a contract between realms deployed separately: - * a shell and its fragments may run different versions. This records every - * message a shell and a fragment exchange through one full session, and - * checks it against the fixture recorded from the implementation before the - * core was extracted. Record it again only deliberately, with - * `UPDATE_WIRE_FIXTURE=1`. + * a shell and its fragments may run different versions, and may be built + * with or without Angular. This records every message a shell and a fragment + * exchange through one full session, and checks it against the fixture + * recorded from the implementation before the core was extracted — for an + * Angular or a core-only realm on either side. Record it again only + * deliberately, with `UPDATE_WIRE_FIXTURE=1`. */ const FIXTURE = join(__dirname, 'uwt-broadcast.wire.fixture.json'); +const IDENTITY = { + application: 'realm', + environment: 'test', + version: '1.0.0', + eventNamespace: 'com.uwt' +}; + /** A message as posted: by which realm, and what it carried. */ interface Posted { from: string; @@ -68,6 +95,11 @@ class WireChannel { WireChannel.channels.clear(); } + /** The realms with a channel still open, one entry per channel. */ + static open(): string[] { + return [...WireChannel.channels.values()].flat().map((c) => c.realm); + } + postMessage(message: unknown): void { if (this.closed) { return; @@ -154,11 +186,12 @@ function stable(posted: Posted[]): Posted[] { /** The shell's real destination, with a session and user it controls. */ @Injectable() class ShellSink extends UwtTelemetrySink { + readonly events: string[] = []; sessionId: string | undefined = 'shell-session'; userId?: string; - record(_eventType: string, _payload: object, _meta?: UwtTelemetryMetadata) { - // Recorded on the wire, not here. + record(eventType: string, _payload: object, _meta?: UwtTelemetryMetadata) { + this.events.push(eventType); } recordError(_error: UwtTelemetryError, _meta?: UwtTelemetryMetadata) { @@ -204,21 +237,29 @@ class ShellLogApi implements UwtLogApiProvider { } } -/** A realm: its own injector, as the shell and each fragment have. */ +/** A shell: its real destination, and how it goes away. */ +interface Shell { + sink: ShellSink; + destroy(): void; +} + +/** A fragment, whatever it is built with: what its application code calls. */ +interface Fragment { + start(options: UwtScenarioOptions): UwtScenario; + track(eventName: string, metadata?: UwtTelemetryMetadata): void; + getLogger(name: string): UwtLogger; + sink: UwtTelemetrySink; + logApi: UwtLogApiProvider; + destroy(): void; +} + +/** An Angular realm: its own injector, as the shell and each fragment have. */ function createRealm(providers: unknown[]): Injector { return Injector.create({ providers: [ { provide: PLATFORM_ID, useValue: 'browser' }, { provide: DOCUMENT, useValue: document }, - { - provide: UWT_TELEMETRY_IDENTITY, - useValue: { - application: 'realm', - environment: 'test', - version: '1.0.0', - eventNamespace: 'com.uwt' - } - }, + { provide: UWT_TELEMETRY_IDENTITY, useValue: IDENTITY }, { provide: UWT_SCENARIO_TELEMETRY_CONFIG, useValue: UWT_DEFAULT_TELEMETRY_CONFIG.scenario @@ -240,6 +281,99 @@ function createRealm(providers: unknown[]): Injector { }); } +/** What application code calls in an Angular fragment. */ +function angularFragmentOf(injector: Injector, destroy: () => void): Fragment { + const scenarios = injector.get(UwtScenarioTelemetryService); + const bi = injector.get(UwtBITelemetryService); + const logs = injector.get(UwtLoggerService); + return { + start: (options) => scenarios.start(options), + track: (eventName, metadata) => bi.track(eventName, metadata), + getLogger: (name) => logs.getLogger(name), + sink: injector.get(UwtTelemetrySink), + logApi: injector.get(UWT_LOG_API_PROVIDER), + destroy + }; +} + +const shells: Record Shell> = { + 'an Angular shell': () => { + const shell = createRealm([ + ShellSink, + { provide: UwtTelemetrySink, useExisting: ShellSink }, + ShellLogApi, + { provide: UWT_LOG_API_PROVIDER, useExisting: ShellLogApi }, + UwtTelemetryBroadcastHost + ]); + const host = shell.get(UwtTelemetryBroadcastHost); + return { sink: shell.get(ShellSink), destroy: () => host.ngOnDestroy() }; + }, + 'a core shell': () => { + const sink = new ShellSink(); + const host = new CoreTelemetryBroadcastHost({ + sink, + logApi: new ShellLogApi() + }); + return { sink, destroy: () => host.dispose() }; + } +}; + +const fragments: Record Fragment> = { + // Each class opens its own channel, as before the extraction. + 'an Angular fragment': () => { + const realm = createRealm([ + UwtBroadcastTelemetrySink, + { provide: UwtTelemetrySink, useExisting: UwtBroadcastTelemetrySink }, + UwtBroadcastLogApiProvider, + { provide: UWT_LOG_API_PROVIDER, useExisting: UwtBroadcastLogApiProvider } + ]); + return angularFragmentOf(realm, () => { + realm.get(UwtBroadcastTelemetrySink).ngOnDestroy(); + realm.get(UwtBroadcastLogApiProvider).ngOnDestroy(); + }); + }, + // Wired by provideUwtTelemetrySink: one channel, shared. + 'an Angular fragment sharing one channel': () => { + const injector = createEnvironmentInjector( + [ + provideUwtTelemetry(IDENTITY), + provideUwtTelemetrySink('broadcast'), + UwtScenarioTelemetryService, + UwtBITelemetryService, + UwtLoggerService + ], + TestBed.inject(EnvironmentInjector) + ); + return angularFragmentOf(injector, () => injector.destroy()); + }, + 'a core-only fragment': () => { + const connection = uwtConnectBroadcastChannel(); + const sink = new CoreBroadcastTelemetrySink({ + identity: IDENTITY, + connection + }); + const logApi = new CoreBroadcastLogApiProvider({ connection }); + const telemetry = uwtCreateTelemetry({ + identity: IDENTITY, + destination: sink, + logApi + }); + return { + start: (options) => telemetry.scenarios.start(options), + track: (eventName, metadata) => telemetry.bi.track(eventName, metadata), + getLogger: (name) => telemetry.logging.getLogger(name), + sink, + logApi, + destroy: () => { + telemetry.dispose(); + sink.dispose(); + logApi.dispose(); + connection.close(); + } + }; + } +}; + describe('the broadcast wire protocol', () => { beforeEach(() => { localStorage.clear(); @@ -262,64 +396,99 @@ describe('the broadcast wire protocol', () => { } } - it('should carry what a shell and a fragment exchange, unchanged', async () => { + /** A shell and a fragment, each in its own realm. */ + async function connect(shellKind: string, fragmentKind: string) { WireChannel.realm = 'shell'; - const shell = createRealm([ - ShellSink, - { provide: UwtTelemetrySink, useExisting: ShellSink }, - ShellLogApi, - { provide: UWT_LOG_API_PROVIDER, useExisting: ShellLogApi }, - UwtTelemetryBroadcastHost - ]); - const shellSink = shell.get(ShellSink); - const host = shell.get(UwtTelemetryBroadcastHost); - + const shell = shells[shellKind](); WireChannel.realm = 'fragment'; - const fragment = createRealm([ - UwtBroadcastTelemetrySink, - { provide: UwtTelemetrySink, useExisting: UwtBroadcastTelemetrySink }, - UwtBroadcastLogApiProvider, - { provide: UWT_LOG_API_PROVIDER, useExisting: UwtBroadcastLogApiProvider } - ]); - const sink = fragment.get(UwtTelemetrySink); - const logApi = fragment.get(UWT_LOG_API_PROVIDER); - await settle(); - - // A scenario with a step, a business event, logs and a mirrored error. - const scenario = fragment - .get(UwtScenarioTelemetryService) - .start({ name: 'checkout', metadata: { cart: 'c-1' } }); - scenario.step('pay', { method: 'card' }); - scenario.endStep(); - scenario.complete({ metadata: { items: 2 } }); - fragment - .get(UwtBITelemetryService) - .track('export_clicked', { format: 'csv' }); - const logger = fragment.get(UwtLoggerService).getLogger('checkout'); - logger.warn('Slow payment', { metadata: { ms: 900 } }); - logger.error('Payment failed', { error: new Error('Declined') }); - await settle(); - - // A sign-in, flushes, and a query answered by the shell's backend. - sink.setUserId('user-1'); + const fragment = fragments[fragmentKind](); await settle(); - sink.flush(true); - logApi.flush?.(); - const found = logApi.query({ logger: 'checkout', limit: 1 }); - await settle(); - expect(await found).toHaveLength(1); + return { shell, fragment }; + } - // The shell's session moves on: the next record is stamped with it. - shellSink.sessionId = 'shell-session-2'; - fragment.get(UwtBITelemetryService).track('theme_changed'); - await settle(); + const pairs = [ + ['an Angular shell', 'an Angular fragment'], + ['an Angular shell', 'an Angular fragment sharing one channel'], + ['an Angular shell', 'a core-only fragment'], + ['a core shell', 'an Angular fragment'], + ['a core shell', 'an Angular fragment sharing one channel'], + ['a core shell', 'a core-only fragment'] + ]; + + it.each(pairs)( + 'should carry what %s and %s exchange, unchanged', + async (shellKind, fragmentKind) => { + const { shell, fragment } = await connect(shellKind, fragmentKind); + + // A scenario with a step, a business event, logs and a mirrored error. + const scenario = fragment.start({ + name: 'checkout', + metadata: { cart: 'c-1' } + }); + scenario.step('pay', { method: 'card' }); + scenario.endStep(); + scenario.complete({ metadata: { items: 2 } }); + fragment.track('export_clicked', { format: 'csv' }); + const logger = fragment.getLogger('checkout'); + logger.warn('Slow payment', { metadata: { ms: 900 } }); + logger.error('Payment failed', { error: new Error('Declined') }); + await settle(); + + // A sign-in, flushes, and a query answered by the shell's backend. + fragment.sink.setUserId('user-1'); + await settle(); + fragment.sink.flush(true); + fragment.logApi.flush?.(); + const found = fragment.logApi.query({ logger: 'checkout', limit: 1 }); + await settle(); + expect(await found).toHaveLength(1); + + // The shell's session moves on: the next record is stamped with it. + shell.sink.sessionId = 'shell-session-2'; + fragment.track('theme_changed'); + await settle(); + + shell.destroy(); + // The session, and the host going away, as the fixture recorded them. + const recorded = stable(WireChannel.posted); + fragment.destroy(); + if ( + process.env.UPDATE_WIRE_FIXTURE === '1' && + shellKind === pairs[0][0] && + fragmentKind === pairs[0][1] + ) { + writeFileSync(FIXTURE, JSON.stringify(recorded, null, 2) + '\n'); + } + expect(existsSync(FIXTURE)).toBe(true); + expect(recorded).toEqual(JSON.parse(readFileSync(FIXTURE, 'utf8'))); + expect(WireChannel.open()).toEqual([]); + } + ); + + it.each(Object.keys(fragments))( + 'should send nothing twice, and leave no channel open, when %s is remounted', + async (fragmentKind) => { + const { shell, fragment } = await connect( + 'an Angular shell', + fragmentKind + ); - host.ngOnDestroy(); - const recorded = stable(WireChannel.posted); - if (process.env.UPDATE_WIRE_FIXTURE === '1') { - writeFileSync(FIXTURE, JSON.stringify(recorded, null, 2) + '\n'); + fragment.track('export_clicked'); + const pending = fragment.logApi.query({}); + fragment.destroy(); + expect(await pending).toEqual([]); + expect(WireChannel.open()).toEqual(['shell']); + + WireChannel.realm = 'fragment'; + const remounted = fragments[fragmentKind](); + await settle(); + remounted.track('theme_changed'); + await settle(); + + expect(shell.sink.events).toEqual(['com.uwt.bi', 'com.uwt.bi']); + remounted.destroy(); + shell.destroy(); + expect(WireChannel.open()).toEqual([]); } - expect(existsSync(FIXTURE)).toBe(true); - expect(recorded).toEqual(JSON.parse(readFileSync(FIXTURE, 'utf8'))); - }); + ); }); diff --git a/projects/ngx-ui-watchtower/src/lib/utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util.ts b/projects/ngx-ui-watchtower/src/lib/utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util.ts index 103f03e..62ed090 100644 --- a/projects/ngx-ui-watchtower/src/lib/utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util.ts +++ b/projects/ngx-ui-watchtower/src/lib/utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util.ts @@ -43,6 +43,16 @@ export function uwtRunOutsideZone(): (callback: () => T) => T { return (callback) => zone?.runOutsideAngular(callback) ?? callback(); } +/** + * Runs a callback inside Angular's zone when there is one, so what it + * resolves is seen by change detection. Must be called from an injection + * context. + */ +export function uwtRunInsideZone(): (callback: () => T) => T { + const zone = inject(NgZone, { optional: true }); + return (callback) => (zone ? zone.run(callback) : callback()); +} + /** * How this package reports what telemetry handles silently: to the console, * prefixed, and only in development mode. Everything the core does on this @@ -54,6 +64,19 @@ export const uwtDevModeReporter: UwtReporter = (level, message, ...details) => { } }; +/** + * {@link uwtDevModeReporter}, except that warnings are printed in production + * too: the broadcast host's warnings about a realm that cannot host always + * were. + */ +export const uwtHostReporter: UwtReporter = (level, message, ...details) => { + if (level === 'warn') { + console.warn(`[ngx-ui-watchtower] ${message}`, ...details); + } else { + uwtDevModeReporter(level, message, ...details); + } +}; + /** * Runs a telemetry operation without letting it affect the application: * the core's fail-open wrapper, reporting through {@link uwtDevModeReporter}. diff --git a/projects/ngx-ui-watchtower/src/public-api.ts b/projects/ngx-ui-watchtower/src/public-api.ts index 9bac8c6..e6dec4d 100644 --- a/projects/ngx-ui-watchtower/src/public-api.ts +++ b/projects/ngx-ui-watchtower/src/public-api.ts @@ -99,10 +99,8 @@ export { UwtScenarioTelemetryService } from './lib/services/uwt-scenario-telemet // Sinks export { UwtTelemetryBroadcastHost } from './lib/sinks/uwt-broadcast/uwt-broadcast-host.service'; export { UwtBroadcastTelemetrySink } from './lib/sinks/uwt-broadcast/uwt-broadcast-telemetry.sink'; -export { - UWT_BROADCAST_CHANNEL, - UWT_DEFAULT_BROADCAST_CHANNEL -} from './lib/sinks/uwt-broadcast/uwt-broadcast.messages'; +export { UWT_BROADCAST_CHANNEL } from './lib/sinks/uwt-broadcast/uwt-broadcast.messages'; +export { UWT_DEFAULT_BROADCAST_CHANNEL } from '@absaoss-cps/ui-watchtower'; export { UwtTelemetrySink } from '@absaoss-cps/ui-watchtower'; export { UwtNoopTelemetrySink } from './lib/sinks/uwt-telemetry/uwt-noop-telemetry.sink/uwt-noop-telemetry.sink'; diff --git a/projects/ui-watchtower/README.md b/projects/ui-watchtower/README.md index 24a61f0..7f33136 100644 --- a/projects/ui-watchtower/README.md +++ b/projects/ui-watchtower/README.md @@ -8,8 +8,8 @@ JavaScript or TypeScript application. It imports no framework and touches no browser global when imported, so it loads in Node, on a server and in a worker as well as in a browser. -> **Work in progress.** The broadcast and AWS RUM destinations are still -> moving here from the Angular package. +> **Work in progress.** The AWS RUM destination is still moving here from the +> Angular package. ## Install @@ -71,6 +71,32 @@ abandoned and flushes; the engine does nothing after that, and leaves no listener or timer behind. The destination and log store you passed are yours to close. +## Shell and fragments + +In a page composed of fragments, each in its own JavaScript realm, the shell +sends everything and the fragments forward to it over a `BroadcastChannel`, +so a user is one session, not one per fragment: + +```ts +// shell: records what fragments forward, through its own destination and +// log store +const host = new UwtTelemetryBroadcastHost({ sink: destination, logApi }); + +// each fragment: forwards everything, over one channel +const identity = { application: 'cart', environment: 'prod', version: '1.4.0' }; +const connection = uwtConnectBroadcastChannel(); +const telemetry = uwtCreateTelemetry({ + identity, + destination: new UwtBroadcastTelemetrySink({ identity, connection }), + logApi: new UwtBroadcastLogApiProvider({ connection }) +}); +``` + +Either side can be an Angular application using +`@absaoss-cps/ngx-ui-watchtower`; they exchange the same messages. Dispose +the host, the sink, the log provider and the connection when their realm goes +away. + ## Declare your names Scenario, step, event and logger names are typed, so a typo is a compile @@ -125,18 +151,19 @@ customers$.pipe(traceScenario(scenario)).subscribe(); ## API -| Group | Exports | -| ------------- | ------------------------------------------------------------------------------------------------------------------------------------------------ | -| Engine | `uwtCreateTelemetry`, `UwtTelemetry`, `UwtTelemetryOptions`, `UwtTelemetryDocument` | -| Concerns | `UwtScenarioTelemetry`, `UwtBITelemetry`, `UwtLogTelemetry` and their options, for an adapter composing them itself | -| Configuration | `UwtTelemetryIdentity`, the per-concern config types, `UWT_DEFAULT_TELEMETRY_CONFIG`, `uwtResolveTelemetryConfig` | -| Models | scenario, step, BI event and log record types; `UWT_LOG_LEVEL_ORDER`, `UWT_TELEMETRY_EVENT_TYPE`, `uwtEventTypes` | -| Names | `UwtScenarioNames`, `UwtScenarioSteps`, `UwtBIEventNames`, `UwtLoggerNames` and their `Uwt*Name` types | -| Scenarios | `UwtScenario` | -| Contracts | `UwtTelemetrySink` and `UwtNoopTelemetrySink`; `UwtLogApiProvider`, `UwtLogQuery` and `UwtNoopLogApiProvider` | -| Redaction | `uwtRedactMetadata`, `uwtScrubString`, `uwtNormalizeError`, `uwtRedactConfigFor`, `UwtRedactConfig`, `UWT_DEFAULT_REDACT_CONFIG`, `UWT_REDACTED` | -| Utilities | `uwtClassifyTelemetryEvent`, `uwtIsDebugEnabled`, `uwtMarkSessionStart`, `uwtUuid`, `uwtSafe*`, `UwtReporter` | -| `/rxjs` | `traceScenario`, `UwtTraceScenarioOptions` | +| Group | Exports | +| ------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------- | +| Engine | `uwtCreateTelemetry`, `UwtTelemetry`, `UwtTelemetryOptions`, `UwtTelemetryDocument` | +| Concerns | `UwtScenarioTelemetry`, `UwtBITelemetry`, `UwtLogTelemetry` and their options, for an adapter composing them itself | +| Configuration | `UwtTelemetryIdentity`, the per-concern config types, `UWT_DEFAULT_TELEMETRY_CONFIG`, `uwtResolveTelemetryConfig` | +| Models | scenario, step, BI event and log record types; `UWT_LOG_LEVEL_ORDER`, `UWT_TELEMETRY_EVENT_TYPE`, `uwtEventTypes` | +| Names | `UwtScenarioNames`, `UwtScenarioSteps`, `UwtBIEventNames`, `UwtLoggerNames` and their `Uwt*Name` types | +| Scenarios | `UwtScenario` | +| Contracts | `UwtTelemetrySink` and `UwtNoopTelemetrySink`; `UwtLogApiProvider`, `UwtLogQuery` and `UwtNoopLogApiProvider` | +| Broadcast | `UwtTelemetryBroadcastHost`, `UwtBroadcastTelemetrySink`, `UwtBroadcastLogApiProvider`, `uwtConnectBroadcastChannel`, `UWT_DEFAULT_BROADCAST_CHANNEL` | +| Redaction | `uwtRedactMetadata`, `uwtScrubString`, `uwtNormalizeError`, `uwtRedactConfigFor`, `UwtRedactConfig`, `UWT_DEFAULT_REDACT_CONFIG`, `UWT_REDACTED` | +| Utilities | `uwtClassifyTelemetryEvent`, `uwtIsDebugEnabled`, `uwtMarkSessionStart`, `uwtUuid`, `uwtSafe*`, `UwtReporter` | +| `/rxjs` | `traceScenario`, `UwtTraceScenarioOptions` | ## License diff --git a/projects/ui-watchtower/public-api.snapshot.json b/projects/ui-watchtower/public-api.snapshot.json index b38bf28..1c0bbed 100644 --- a/projects/ui-watchtower/public-api.snapshot.json +++ b/projects/ui-watchtower/public-api.snapshot.json @@ -1,5 +1,11 @@ { "@absaoss-cps/ui-watchtower": { + "UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS": [ + "const: 10000" + ], + "UWT_DEFAULT_BROADCAST_CHANNEL": [ + "const: 'ngx-ui-watchtower'" + ], "UWT_DEFAULT_EVENT_NAMESPACE": [ "const: 'com.uwt'" ], @@ -61,9 +67,66 @@ "report?: UwtReporter | undefined", "sink: UwtTelemetrySink" ], + "UwtBroadcastConnection": [ + "interface", + "close(): void", + "isOpen(): boolean", + "onMessage(handler: (data: unknown) => void): () => void", + "post(message: UwtBroadcastMessage): void", + "readonly channelName: string" + ], + "UwtBroadcastConnectOptions": [ + "interface", + "channelName?: string | undefined", + "open?: boolean | undefined", + "operation?: string | undefined", + "report?: UwtReporter | undefined" + ], + "UwtBroadcastLogApiProvider": [ + "class", + "new (options?: UwtBroadcastLogApiProviderOptions | undefined): UwtBroadcastLogApiProvider", + "dispose(): void", + "flush(): void", + "query(filter: UwtLogQuery): Promise", + "send(record: UwtLogRecord): void" + ], + "UwtBroadcastLogApiProviderOptions": [ + "interface", + "channelName?: string | undefined", + "connection?: UwtBroadcastConnection | undefined", + "enabled?: boolean | undefined", + "report?: UwtReporter | undefined", + "runInside?: ((callback: () => T) => T) | undefined", + "runOutside?: ((callback: () => T) => T) | undefined" + ], + "UwtBroadcastMessage": [ + "type = { kind: 'event'; eventType: string; payload: object; metadata?: UwtTelemetryMetadata | undefined; } | { kind: 'error'; error: UwtTelemetryError; metadata?: UwtTelemetryMetadata | undefined; } | { kind: 'user'; userId: string | undefined; } | { kind: 'flush'; beacon: boolean; } | { kind: 'identity-request'; } | { kind: 'identity'; sessionId?: string | undefined; userId?: string | undefined; } | { kind: 'log'; record: UwtLogRecord; } | { kind: 'log-flush'; } | { kind: 'log-query'; id: string; filter: UwtLogQuery; } | { kind: 'log-query-result'; id: string; records: unknown[]; }" + ], + "UwtBroadcastTelemetrySink": [ + "class", + "new (options: UwtBroadcastTelemetrySinkOptions): UwtBroadcastTelemetrySink", + "dispose(): void", + "flush(beacon?: boolean | undefined): void", + "getSessionId(): string | undefined", + "getUserId(): string | undefined", + "record(eventType: string, payload: object, metadata?: UwtTelemetryMetadata | undefined): void", + "recordError(error: UwtTelemetryError, metadata?: UwtTelemetryMetadata | undefined): void", + "setUserId(userId: string | undefined): void" + ], + "UwtBroadcastTelemetrySinkOptions": [ + "interface", + "channelName?: string | undefined", + "connection?: UwtBroadcastConnection | undefined", + "enabled?: boolean | undefined", + "identity: UwtTelemetryIdentity", + "report?: UwtReporter | undefined" + ], "uwtClassifyTelemetryEvent": [ "function(eventType: string, payload: object): UwtTelemetrySinkEvent" ], + "uwtConnectBroadcastChannel": [ + "function(options?: UwtBroadcastConnectOptions | undefined): UwtBroadcastConnection" + ], "uwtCreateTelemetry": [ "function(options: UwtTelemetryOptions): UwtTelemetry" ], @@ -393,6 +456,20 @@ "sessionId?: string | undefined", "userId?: string | undefined" ], + "UwtTelemetryBroadcastHost": [ + "class", + "new (options: UwtTelemetryBroadcastHostOptions): UwtTelemetryBroadcastHost", + "dispose(): void", + "get received: number" + ], + "UwtTelemetryBroadcastHostOptions": [ + "interface", + "channelName?: string | undefined", + "enabled?: boolean | undefined", + "logApi: UwtLogApiProvider", + "report?: UwtReporter | undefined", + "sink: UwtTelemetrySink" + ], "UwtTelemetryConfig": [ "interface", "bi: UwtBITelemetryConfig", diff --git a/projects/ui-watchtower/src/lib/providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider.ts b/projects/ui-watchtower/src/lib/providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider.ts new file mode 100644 index 0000000..78abce5 --- /dev/null +++ b/projects/ui-watchtower/src/lib/providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider.ts @@ -0,0 +1,191 @@ +import { UwtLogRecord } from '../../models/uwt-log.models/uwt-log.models'; +import { + UwtBroadcastConnection, + uwtConnectBroadcastChannel, + uwtIsBroadcastLogRecord, + uwtIsBroadcastMessage +} from '../../sinks/uwt-broadcast/uwt-broadcast.messages'; +import { + UwtReporter, + uwtSafeVoid, + uwtUuid +} from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; +import { + UwtLogApiProvider, + UwtLogQuery +} from '../uwt-log-api.provider/uwt-log-api.provider'; + +/** + * How long a follower waits for the host to answer a log query before + * resolving to `[]`, in milliseconds. + * + * @group Utils + */ +export const UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS = 10_000; + +/** A log query waiting for the host's answer. */ +interface UwtPendingLogQuery { + resolve: (records: UwtLogRecord[]) => void; + timer: ReturnType; +} + +/** + * What {@link UwtBroadcastLogApiProvider} needs; all optional. + * + * @group Interfaces + */ +export interface UwtBroadcastLogApiProviderOptions { + /** + * The channel's name; must match the host's. Default + * {@link UWT_DEFAULT_BROADCAST_CHANNEL}. + */ + channelName?: string; + /** + * `false` forwards nothing, and every query resolves to `[]` at once — + * on a server, or while telemetry is off — even over a shared + * {@link connection}. Default `true`. + */ + enabled?: boolean; + /** + * A connection to share with the realm's {@link UwtBroadcastTelemetrySink}, + * instead of opening one of its own. Not used while {@link enabled} is + * `false`. Left open by {@link UwtBroadcastLogApiProvider.dispose}: whoever + * opened it closes it. + */ + connection?: UwtBroadcastConnection; + /** + * Schedules the query timeout outside the host framework's change + * detection: Angular passes `NgZone.runOutsideAngular`. + */ + runOutside?: (callback: () => T) => T; + /** + * Hands a query's answer back inside the host framework's change + * detection: Angular passes `NgZone.run`. + */ + runInside?: (callback: () => T) => T; + /** Receives failures; see {@link UwtReporter}. */ + report?: UwtReporter; +} + +/** + * Log API provider for a follower realm — the log counterpart of + * {@link UwtBroadcastTelemetrySink}. + * + * Ships nothing itself. Records are forwarded over the `BroadcastChannel` + * to the realm running a {@link UwtTelemetryBroadcastHost}, which hands + * them to its own log provider: one log backend, one set of credentials, + * for the whole composed page. + * + * `query()` is answered by the host's provider too. It resolves to `[]` + * when no host answers within {@link UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS}, + * and at once when no channel could be opened. + * + * @group Services + */ +export class UwtBroadcastLogApiProvider implements UwtLogApiProvider { + private readonly connection: UwtBroadcastConnection; + private readonly ownsConnection: boolean; + private readonly stopListening: () => void; + private readonly runOutside: (callback: () => T) => T; + private readonly runInside: (callback: () => T) => T; + private readonly report?: UwtReporter; + private readonly pending = new Map(); + /** Set by {@link dispose}: nothing is sent or asked after, even on a shared connection. */ + private disposed = false; + + constructor(options: UwtBroadcastLogApiProviderOptions = {}) { + this.report = options.report; + // Off wins over a shared connection: this provider then has a closed one + // of its own, and the shared one stays open for its other users. + const shared = options.enabled === false ? undefined : options.connection; + this.ownsConnection = !shared; + this.connection = + shared ?? + uwtConnectBroadcastChannel({ + channelName: options.channelName, + open: options.enabled !== false, + operation: 'broadcastLogApi', + report: options.report + }); + this.runOutside = options.runOutside ?? ((callback) => callback()); + this.runInside = options.runInside ?? ((callback) => callback()); + this.stopListening = this.connection.onMessage((data) => + this.onMessage(data) + ); + } + + /** @inheritdoc */ + send(record: UwtLogRecord): void { + if (!this.disposed) { + this.connection.post({ kind: 'log', record }); + } + } + + /** @inheritdoc */ + query(filter: UwtLogQuery): Promise { + if (this.disposed || !this.connection.isOpen()) { + return Promise.resolve([]); + } + + return new Promise((resolve) => { + const id = uwtUuid(); + const timer = this.runOutside(() => + setTimeout( + () => this.answer(id, []), + UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS + ) + ); + this.pending.set(id, { resolve, timer }); + this.connection.post({ kind: 'log-query', id, filter }); + }); + } + + /** @inheritdoc */ + flush(): void { + if (!this.disposed) { + this.connection.post({ kind: 'log-flush' }); + } + } + + /** + * Resolves every unanswered query to `[]`, stops listening, and closes + * the channel if this provider opened it. A shared connection stays open + * for its other users, but this provider sends nothing over it any more, + * and every later query resolves to `[]` at once. Calling it again does + * nothing. + */ + dispose(): void { + this.disposed = true; + for (const id of [...this.pending.keys()]) { + this.answer(id, []); + } + this.stopListening(); + if (this.ownsConnection) { + this.connection.close(); + } + } + + private onMessage(data: unknown): void { + uwtSafeVoid( + 'broadcastLogApi.receive', + () => { + if (!uwtIsBroadcastMessage(data) || data.kind !== 'log-query-result') { + return; + } + // Other followers' answers arrive here too; answer() ignores them. + this.answer(data.id, data.records.filter(uwtIsBroadcastLogRecord)); + }, + this.report + ); + } + + private answer(id: string, records: UwtLogRecord[]): void { + const query = this.pending.get(id); + if (!query) { + return; + } + this.pending.delete(id); + clearTimeout(query.timer); + this.runInside(() => query.resolve(records)); + } +} diff --git a/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-host.ts b/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-host.ts new file mode 100644 index 0000000..743baa5 --- /dev/null +++ b/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-host.ts @@ -0,0 +1,414 @@ +import { UwtLogRecord } from '../../models/uwt-log.models/uwt-log.models'; +import { UwtBroadcastLogApiProvider } from '../../providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider'; +import { + UwtLogApiProvider, + UwtLogQuery +} from '../../providers/uwt-log-api.provider/uwt-log-api.provider'; +import { uwtSessionElapsed } from '../../utils/uwt-session-start.util/uwt-session-start.util'; +import { uwtClassifyTelemetryEvent } from '../../utils/uwt-telemetry-event.util/uwt-telemetry-event.util'; +import { + uwtReport, + UwtReporter, + uwtSafe, + uwtSafeVoid, + uwtSafeVoidMaybeAsync +} from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; +import { UwtTelemetrySink } from '../uwt-telemetry/uwt-telemetry-abstract.sink/uwt-telemetry-abstract.sink'; +import { UwtBroadcastTelemetrySink } from './uwt-broadcast-telemetry.sink'; +import { + UwtBroadcastConnection, + uwtConnectBroadcastChannel, + uwtElectBroadcastHostLeader, + uwtIsBroadcastMessage +} from './uwt-broadcast.messages'; + +/** The identity fields scenario, BI and log records carry. */ +interface UwtIdentified { + sessionId?: string; + userId?: string; + sessionElapsed?: number; +} + +/** + * Active host count per channel, scoped to this JS realm — distinguishes a + * true duplicate host from another tab sharing the same channel. Realm-wide + * on purpose; only a host that is active in a browser counts, so server-side + * renders never see each other's. + */ +const hostsInThisRealm = new Map(); + +/** Where warnings about a host that cannot work go when no reporter is given. */ +const consoleReporter: UwtReporter = (level, message, ...details) => + console[level](`[ui-watchtower] ${message}`, ...details); + +/** + * What {@link UwtTelemetryBroadcastHost} needs. + * + * @group Interfaces + */ +export interface UwtTelemetryBroadcastHostOptions { + /** This realm's real destination, recording what followers forward. */ + sink: UwtTelemetrySink; + /** + * This realm's real log store: it receives followers' records and answers + * their queries. + */ + logApi: UwtLogApiProvider; + /** + * The channel's name; must match the followers'. Default + * {@link UWT_DEFAULT_BROADCAST_CHANNEL}. + */ + channelName?: string; + /** + * `false` hosts nothing — no channel, no election: on a server, or while + * telemetry is off. Default `true`. + */ + enabled?: boolean; + /** + * Receives failures, and the warnings about a host that cannot work as + * configured. Those warnings go to the console when no reporter is given: + * a misconfigured host is never silent. + */ + report?: UwtReporter; +} + +/** + * Receives telemetry forwarded by follower realms and records it through this + * realm's sink. + * + * Runs in the shell, the realm with the real sink. Fragments using + * {@link UwtBroadcastTelemetrySink} post their events here, so one AWS client, + * one session and one event budget serve the whole composed page. Their + * scenario, BI and log records are stamped with this realm's current session + * and user, which a fragment's own copy can lag behind. + * + * Fragments' log records arrive the same way, from + * {@link UwtBroadcastLogApiProvider}, and go to this realm's log provider — + * which also answers their `query()` calls. + * + * A realm that forwards its own telemetry cannot also host it: the host + * would receive its own realm's messages and forward them again, forever. + * Such a host warns and stays inactive. + * + * A Web Locks-based election keeps exactly one host active per channel; + * others stay passive. + * + * @group Services + */ +export class UwtTelemetryBroadcastHost { + private readonly sink: UwtTelemetrySink; + private readonly logApi: UwtLogApiProvider; + private readonly connection: UwtBroadcastConnection; + private readonly enabled: boolean; + private readonly report?: UwtReporter; + + /** Number of messages accepted, for tests and diagnostics. */ + private _received = 0; + + /** + * The session and user id last sent in an `identity` message. + * + * Compared against the sink's current values on every follower activity, so + * a late-resolving session id or user id gets announced once it exists. + */ + private lastAnnouncedSessionId?: string; + private lastAnnouncedUserId?: string; + + /** Whether this realm won the leader election; non-leaders stay passive. */ + private isLeader = false; + private releaseLeadership: () => void = () => undefined; + private disposed = false; + + constructor(options: UwtTelemetryBroadcastHostOptions) { + this.sink = options.sink; + this.logApi = options.logApi; + this.enabled = options.enabled !== false; + this.report = options.report; + this.connection = uwtConnectBroadcastChannel({ + channelName: options.channelName, + open: this.enabled, + operation: 'broadcastHost', + report: options.report + }); + + if (!this.enabled) { + // No listening, no leader election. + return; + } + this.warnIfDuplicateInThisRealm(); + if (this.forwardsItsOwnRealm()) { + this.connection.close(); + return; + } + + this.connection.onMessage((data) => this.onMessage(data)); + this.releaseLeadership = uwtElectBroadcastHostLeader( + this.connection.channelName, + () => { + this.isLeader = true; + this.announceIdentity(); + } + ); + } + + /** How many follower messages have been accepted. */ + get received(): number { + return this._received; + } + + /** + * Gives up leadership and closes the channel. Calling it again does + * nothing. + */ + dispose(): void { + if (this.disposed) { + return; + } + this.disposed = true; + if (this.enabled) { + this.forgetInThisRealm(); + } + this.releaseLeadership(); + this.connection.close(); + } + + private onMessage(data: unknown): void { + uwtSafeVoid( + 'broadcastHost.receive', + () => { + if (!this.isLeader || !uwtIsBroadcastMessage(data)) { + return; + } + + if ( + data.kind !== 'identity' && + data.kind !== 'identity-request' && + data.kind !== 'log-query-result' + ) { + this._received++; + } + + switch (data.kind) { + case 'event': + this.sink.record( + data.eventType, + this.stampEvent(data.eventType, data.payload), + data.metadata + ); + this.reannounceIfIdentityChanged(); + break; + case 'error': + this.sink.recordError(data.error, data.metadata); + this.reannounceIfIdentityChanged(); + break; + case 'user': + this.sink.setUserId(data.userId); + this.announceIdentity(); + break; + case 'flush': + this.sink.flush(data.beacon); + this.reannounceIfIdentityChanged(); + break; + case 'identity-request': + this.announceIdentity(); + break; + case 'identity': + // Another realm's host on this origin-wide channel — expected, + // not a problem. Same-document duplicates are caught by + // warnIfDuplicateInThisRealm() instead. + break; + case 'log': + this.deliverLog(this.withIdentity(data.record)); + this.reannounceIfIdentityChanged(); + break; + case 'log-flush': + uwtSafeVoidMaybeAsync( + 'broadcastHost.logFlush', + () => this.logApi?.flush?.(), + this.report + ); + break; + case 'log-query': + this.answerLogQuery(data.id, data.filter); + break; + case 'log-query-result': + // Hosts answer queries; they never ask them. + break; + } + }, + this.report + ); + } + + /** + * A forwarded scenario or BI record with this realm's identity; any other + * payload as it is. + */ + private stampEvent(eventType: string, payload: object): object { + const { kind } = uwtClassifyTelemetryEvent(eventType, payload); + return kind === 'unknown' + ? payload + : this.withIdentity(payload as UwtIdentified, kind === 'scenario'); + } + + /** + * `record` with this realm's session and user id. A follower stamps the ids + * it was last told, which lag behind whenever they change here — the client + * starting after the follower asked, a session expiring, a sign-out — so its + * record could name a session other than the one it is sent in. With + * `measuresSession`, a changed session's `sessionElapsed` is measured again. + */ + private withIdentity( + record: T, + measuresSession = false + ): T { + const sessionId = this.currentSessionId(); + // '' is no user, as `setUserId('')` signs out. + const userId = this.currentUserId() || undefined; + if (record.sessionId === sessionId && record.userId === userId) { + return record; + } + // An id this realm doesn't have stays `undefined`, + // as in the library's own records. + const stamped: T = { ...record, sessionId, userId }; + if (measuresSession && record.sessionId !== sessionId) { + // The old session's value no longer applies; absent when unknown. + delete stamped.sessionElapsed; + const elapsed = uwtSessionElapsed(sessionId); + if (elapsed !== undefined) { + stamped.sessionElapsed = Math.round(elapsed); + } + } + return stamped; + } + + /** Hands a follower's record to this realm's log API provider. */ + private deliverLog(record: UwtLogRecord): void { + uwtSafeVoidMaybeAsync( + 'broadcastHost.log', + () => this.logApi.send(record), + this.report + ); + } + + /** + * Answers a follower's query from this realm's log API provider. Always + * answers — `[]` when there is no provider or it fails — so the follower + * isn't left waiting for its timeout. + */ + private answerLogQuery(id: string, filter: UwtLogQuery): void { + const reply = (records: unknown) => + this.connection.post({ + kind: 'log-query-result', + id, + records: Array.isArray(records) ? records : [] + }); + + const pending = uwtSafe( + 'broadcastHost.logQuery', + () => this.logApi.query(filter), + undefined, + this.report + ); + + if (!pending) { + reply([]); + return; + } + // Promise.resolve: a provider breaking its contract with a plain value + // still gets an answer out. + Promise.resolve(pending).then(reply, () => reply([])); + } + + /** + * Whether this realm's own sink or log provider forwards over the + * channel. Hosting then would loop: the host receives its own realm's + * messages and hands them straight back to the channel. + */ + private forwardsItsOwnRealm(): boolean { + const forwards = + this.sink instanceof UwtBroadcastTelemetrySink || + this.logApi instanceof UwtBroadcastLogApiProvider; + + if (forwards) { + this.warn( + `this realm forwards its own telemetry on channel "${this.connection.channelName}", so it cannot also host it; the host stays inactive. Provide the host only in the realm with the real sink and log provider` + ); + } + return forwards; + } + + /** + * Warns when another host is already active on the same channel in this + * realm — a real misconfiguration. Not based on the `identity` broadcast, + * which can't distinguish this from a different, legitimate tab. + */ + private warnIfDuplicateInThisRealm(): void { + const channelName = this.connection.channelName; + const activeCount = hostsInThisRealm.get(channelName) ?? 0; + + if (activeCount > 0) { + this.warn( + `a second telemetry host is active on channel "${channelName}" in this document; only one realm should provide it` + ); + } + + hostsInThisRealm.set(channelName, activeCount + 1); + } + + /** Releases this host's slot in {@link hostsInThisRealm}. */ + private forgetInThisRealm(): void { + const channelName = this.connection.channelName; + const activeCount = hostsInThisRealm.get(channelName) ?? 0; + + if (activeCount <= 1) { + hostsInThisRealm.delete(channelName); + } else { + hostsInThisRealm.set(channelName, activeCount - 1); + } + } + + /** A host that cannot work as configured is never silent. */ + private warn(message: string): void { + uwtReport(this.report ?? consoleReporter, 'warn', message); + } + + private announceIdentity(): void { + const sessionId = this.currentSessionId(); + const userId = this.currentUserId(); + this.lastAnnouncedSessionId = sessionId; + this.lastAnnouncedUserId = userId; + this.connection.post({ kind: 'identity', sessionId, userId }); + } + + /** + * Re-announces identity when the sink's session id or user id has moved on + * from what followers were last told. + */ + private reannounceIfIdentityChanged(): void { + if ( + this.currentSessionId() !== this.lastAnnouncedSessionId || + this.currentUserId() !== this.lastAnnouncedUserId + ) { + this.announceIdentity(); + } + } + + private currentSessionId(): string | undefined { + return uwtSafe( + 'broadcastHost.getSessionId', + () => this.sink.getSessionId(), + undefined, + this.report + ); + } + + private currentUserId(): string | undefined { + return uwtSafe( + 'broadcastHost.getUserId', + () => this.sink.getUserId(), + undefined, + this.report + ); + } +} diff --git a/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-telemetry.sink.ts b/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-telemetry.sink.ts new file mode 100644 index 0000000..fe9f7b5 --- /dev/null +++ b/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast-telemetry.sink.ts @@ -0,0 +1,191 @@ +import { UwtTelemetryIdentity } from '../../config/uwt-telemetry-common.config/uwt-telemetry-common.config'; +import { + UwtTelemetryError, + UwtTelemetryMetadata +} from '../../models/uwt-telemetry-common.models/uwt-telemetry-common.models'; +import { + UwtReporter, + uwtSafeVoid +} from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; +import { UwtTelemetrySink } from '../uwt-telemetry/uwt-telemetry-abstract.sink/uwt-telemetry-abstract.sink'; +import { + UwtBroadcastConnection, + UwtBroadcastMessage, + uwtConnectBroadcastChannel, + uwtIsBroadcastMessage +} from './uwt-broadcast.messages'; + +/** + * What {@link UwtBroadcastTelemetrySink} needs. + * + * @group Interfaces + */ +export interface UwtBroadcastTelemetrySinkOptions { + /** This realm's identity, stamped onto everything it forwards. */ + identity: UwtTelemetryIdentity; + /** + * The channel's name; must match the host's. Default + * {@link UWT_DEFAULT_BROADCAST_CHANNEL}. + */ + channelName?: string; + /** + * `false` forwards nothing — on a server, or while telemetry is off — + * even over a shared {@link connection}. Default `true`. + */ + enabled?: boolean; + /** + * A connection to share with the realm's + * {@link UwtBroadcastLogApiProvider}, instead of opening one of its own. + * Not used while {@link enabled} is `false`. Left open by + * {@link UwtBroadcastTelemetrySink.dispose}: whoever opened it closes it. + */ + connection?: UwtBroadcastConnection; + /** Receives failures; see {@link UwtReporter}. */ + report?: UwtReporter; +} + +/** + * Telemetry sink for a follower realm — a micro-frontend fragment, a webview + * panel, any context with its own JavaScript realm but the same origin. + * + * Creates no AWS client of its own. Events are forwarded over a + * `BroadcastChannel` to the realm running a {@link UwtTelemetryBroadcastHost}, + * which records them through the one real sink. Application code sees no + * difference — it uses the same engine and calls the same methods. + * + * @group Services + */ +export class UwtBroadcastTelemetrySink extends UwtTelemetrySink { + private readonly connection: UwtBroadcastConnection; + private readonly ownsConnection: boolean; + private readonly stopListening: () => void; + private readonly report?: UwtReporter; + /** Set by {@link dispose}: nothing is forwarded after, even on a shared connection. */ + private disposed = false; + + /** The host's session id, once it has answered. */ + private sessionId?: string; + private userId?: string; + + /** + * This realm's own identity, stamped onto everything it forwards. + * + * The host records through its own RUM client, so without this, forwarded + * events would all be labelled as the host's realm. + */ + private readonly origin: { + application: string; + environment: string; + appVersion: string; + }; + + constructor(options: UwtBroadcastTelemetrySinkOptions) { + super(); + this.report = options.report; + // Off wins over a shared connection: this sink then has a closed one of + // its own, and the shared one stays open for its other users. + const shared = options.enabled === false ? undefined : options.connection; + this.ownsConnection = !shared; + this.connection = + shared ?? + uwtConnectBroadcastChannel({ + channelName: options.channelName, + open: options.enabled !== false, + operation: 'broadcastSink', + report: options.report + }); + this.origin = { + application: options.identity.application, + environment: options.identity.environment, + appVersion: options.identity.version + }; + + this.stopListening = this.connection.onMessage((data) => + this.onMessage(data) + ); + this.connection.post({ kind: 'identity-request' }); + } + + /** @inheritdoc */ + record( + eventType: string, + payload: object, + metadata?: UwtTelemetryMetadata + ): void { + this.post({ + kind: 'event', + eventType, + payload, + metadata: { ...metadata, ...this.origin } + }); + } + + /** @inheritdoc */ + recordError(error: UwtTelemetryError, metadata?: UwtTelemetryMetadata): void { + this.post({ + kind: 'error', + error, + metadata: { ...metadata, ...this.origin } + }); + } + + /** + * The host's session id. + * + * Returns `undefined` until the host answers. + * + * @returns the shared session id, when known + */ + getSessionId(): string | undefined { + return this.sessionId; + } + + /** @inheritdoc */ + setUserId(userId: string | undefined): void { + this.userId = userId; + this.post({ kind: 'user', userId }); + } + + /** @inheritdoc */ + getUserId(): string | undefined { + return this.userId; + } + + /** @inheritdoc */ + flush(beacon = false): void { + this.post({ kind: 'flush', beacon }); + } + + /** + * Stops listening and forwarding, and closes the channel if this sink + * opened it. A shared connection stays open for its other users, but this + * sink sends nothing over it any more. Calling it again does nothing. + */ + dispose(): void { + this.disposed = true; + this.stopListening(); + if (this.ownsConnection) { + this.connection.close(); + } + } + + private post(message: UwtBroadcastMessage): void { + if (!this.disposed) { + this.connection.post(message); + } + } + + private onMessage(data: unknown): void { + uwtSafeVoid( + 'broadcastSink.receive', + () => { + if (!uwtIsBroadcastMessage(data) || data.kind !== 'identity') { + return; + } + this.sessionId = data.sessionId; + this.userId = data.userId; + }, + this.report + ); + } +} diff --git a/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.messages.spec.ts b/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.messages.spec.ts new file mode 100644 index 0000000..f66da99 --- /dev/null +++ b/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.messages.spec.ts @@ -0,0 +1,369 @@ +/** + * @jest-environment jsdom + */ +import { + uwtElectBroadcastHostLeader, + uwtIsBroadcastMessage +} from './uwt-broadcast.messages'; + +/** Minimal Web Locks API stub: grants each named lock to one requester at a time. */ +class LockManagerStub { + private readonly held = new Set(); + private readonly queues = new Map void)[]>(); + + request(name: string, callback: () => Promise): Promise { + return new Promise((resolve) => { + const grant = () => { + this.held.add(name); + callback().then(() => { + this.held.delete(name); + resolve(); + this.queues.get(name)?.shift()?.(); + }); + }; + + if (this.held.has(name)) { + const queue = this.queues.get(name) ?? []; + queue.push(grant); + this.queues.set(name, queue); + } else { + grant(); + } + }); + } + + /** Installs the stub as `navigator.locks`. */ + static install(): void { + Object.defineProperty(globalThis.navigator, 'locks', { + value: new LockManagerStub(), + configurable: true + }); + } + + static uninstall(): void { + delete (globalThis.navigator as { locks?: unknown }).locks; + } +} + +describe('the broadcast protocol', () => { + describe('uwtElectBroadcastHostLeader', () => { + afterEach(() => LockManagerStub.uninstall()); + + it('should elect immediately when the Locks API is unavailable', () => { + const onElected = jest.fn(); + uwtElectBroadcastHostLeader('ngx-ui-watchtower', onElected); + + expect(onElected).toHaveBeenCalledTimes(1); + }); + + it('should elect once the lock is granted', () => { + LockManagerStub.install(); + const onElected = jest.fn(); + uwtElectBroadcastHostLeader('ngx-ui-watchtower', onElected); + + expect(onElected).toHaveBeenCalledTimes(1); + }); + + it('should fail open and still elect when request() itself rejects', async () => { + Object.defineProperty(globalThis.navigator, 'locks', { + value: { request: () => Promise.reject(new Error('not-fully-active')) }, + configurable: true + }); + const onElected = jest.fn(); + + uwtElectBroadcastHostLeader('ngx-ui-watchtower', onElected); + // Let the rejection's microtask settle. + await Promise.resolve(); + await Promise.resolve(); + + expect(onElected).toHaveBeenCalledTimes(1); + }); + + it('should fail open and still elect when request() itself throws synchronously', () => { + Object.defineProperty(globalThis.navigator, 'locks', { + value: { + request: () => { + throw new Error('locks unavailable in this context'); + } + }, + configurable: true + }); + const onElected = jest.fn(); + + expect(() => + uwtElectBroadcastHostLeader('ngx-ui-watchtower', onElected) + ).not.toThrow(); + expect(onElected).toHaveBeenCalledTimes(1); + }); + + it('should not elect a requester released while still queued, and should let the next requester through', async () => { + LockManagerStub.install(); + const onElectedHolder = jest.fn(); + const onElectedA = jest.fn(); + const onElectedB = jest.fn(); + + const releaseHolder = uwtElectBroadcastHostLeader( + 'ngx-ui-watchtower', + onElectedHolder + ); + const releaseA = uwtElectBroadcastHostLeader( + 'ngx-ui-watchtower', + onElectedA + ); + uwtElectBroadcastHostLeader('ngx-ui-watchtower', onElectedB); + + // A is released before ever being granted the lock. + releaseA(); + // The current holder releases, so the lock passes down the queue. + releaseHolder(); + for (let i = 0; i < 6; i++) { + await Promise.resolve(); + } + + expect(onElectedA).not.toHaveBeenCalled(); + expect(onElectedB).toHaveBeenCalledTimes(1); + }); + }); + + describe('uwtIsBroadcastMessage', () => { + it.each([null, undefined, 'string', 42, ['array']])( + 'should reject the non-object payload %p', + (data) => { + expect(uwtIsBroadcastMessage(data)).toBe(false); + } + ); + + it('should reject a payload with an unknown kind', () => { + expect(uwtIsBroadcastMessage({ kind: 'other library' })).toBe(false); + }); + + it('should accept a well-formed event message', () => { + expect( + uwtIsBroadcastMessage({ + kind: 'event', + eventType: 'com.uwt.bi', + payload: { eventName: 'x' } + }) + ).toBe(true); + }); + + it.each([ + { eventType: 'com.uwt.bi' }, + { payload: {} }, + { eventType: 1, payload: {} }, + { eventType: 'com.uwt.bi', payload: 'not-an-object' }, + { + eventType: 'com.uwt.bi', + payload: {}, + metadata: { nested: {} } + }, + { + eventType: 'com.uwt.bi', + payload: {}, + metadata: { list: [1, 2] } + }, + { + eventType: 'com.uwt.bi', + payload: {}, + metadata: 'not-an-object' + } + ])('should reject a malformed event message %p', (fields) => { + expect(uwtIsBroadcastMessage({ kind: 'event', ...fields })).toBe(false); + }); + + it('should accept an event message with flat-primitive metadata', () => { + expect( + uwtIsBroadcastMessage({ + kind: 'event', + eventType: 'com.uwt.bi', + payload: {}, + metadata: { count: 3, ok: true, note: 'x', blank: null } + }) + ).toBe(true); + }); + + it('should accept an event message with no metadata', () => { + expect( + uwtIsBroadcastMessage({ + kind: 'event', + eventType: 'com.uwt.bi', + payload: {} + }) + ).toBe(true); + }); + + it('should accept a well-formed error message', () => { + expect( + uwtIsBroadcastMessage({ + kind: 'error', + error: { name: 'Error', message: 'boom' } + }) + ).toBe(true); + }); + + it.each([ + {}, + { error: { name: 'Error' } }, + { error: { message: 'boom' } }, + { error: 'boom' }, + { error: { name: 'Error', message: 'boom', stack: 42 } }, + { + error: { name: 'Error', message: 'boom' }, + metadata: { nested: {} } + } + ])('should reject a malformed error message %p', (fields) => { + expect(uwtIsBroadcastMessage({ kind: 'error', ...fields })).toBe(false); + }); + + it('should accept an error message with flat-primitive metadata', () => { + expect( + uwtIsBroadcastMessage({ + kind: 'error', + error: { name: 'Error', message: 'boom' }, + metadata: { origin: 'checkout' } + }) + ).toBe(true); + }); + + it('should accept an error message with a string stack', () => { + expect( + uwtIsBroadcastMessage({ + kind: 'error', + error: { name: 'Error', message: 'boom', stack: 'at foo.ts:1' } + }) + ).toBe(true); + }); + + it('should accept a well-formed flush message', () => { + expect(uwtIsBroadcastMessage({ kind: 'flush', beacon: true })).toBe(true); + }); + + it('should reject a flush message with a non-boolean beacon', () => { + expect(uwtIsBroadcastMessage({ kind: 'flush', beacon: 'true' })).toBe( + false + ); + }); + + it.each([{ userId: 'u-1' }, { userId: undefined }, {}])( + 'should accept a well-formed user message %p', + (fields) => { + expect(uwtIsBroadcastMessage({ kind: 'user', ...fields })).toBe(true); + } + ); + + it('should reject a user message with a non-string, non-undefined userId', () => { + expect(uwtIsBroadcastMessage({ kind: 'user', userId: 42 })).toBe(false); + }); + + const logRecord = { + timestamp: '2026-01-01T00:00:00.000Z', + level: 'warn', + message: 'Careful', + application: 'cart', + environment: 'test', + version: '1.0.0' + }; + + it('should accept a well-formed log message', () => { + expect(uwtIsBroadcastMessage({ kind: 'log', record: logRecord })).toBe( + true + ); + expect( + uwtIsBroadcastMessage({ + kind: 'log', + record: { + ...logRecord, + logger: 'checkout', + context: 'Cart', + correlationId: 'c-1', + sessionId: 's', + userId: 'u', + metadata: { count: 1 }, + error: { name: 'Error', message: 'boom' } + } + }) + ).toBe(true); + }); + + it.each([ + { level: 'debug' }, + { level: 'toString' }, + { application: undefined }, + { message: 42 }, + { logger: 7 }, + { metadata: { nested: {} } }, + { error: { name: 'Error' } } + ])('should reject a log message whose record has %p', (fields) => { + expect( + uwtIsBroadcastMessage({ + kind: 'log', + record: { ...logRecord, ...fields } + }) + ).toBe(false); + }); + + it('should accept log-flush, log-query and log-query-result messages', () => { + expect(uwtIsBroadcastMessage({ kind: 'log-flush' })).toBe(true); + expect( + uwtIsBroadcastMessage({ + kind: 'log-query', + id: 'q-1', + filter: { correlationId: 'c-1', minLevel: 'warn', limit: 5 } + }) + ).toBe(true); + expect( + uwtIsBroadcastMessage({ + kind: 'log-query-result', + id: 'q-1', + records: [] + }) + ).toBe(true); + }); + + it.each([ + { id: '', filter: {} }, + { id: 'q-1', filter: null }, + { id: 'q-1', filter: { limit: 'ten' } }, + { id: 'q-1', filter: { minLevel: 'verbose' } } + ])('should reject a malformed log-query message %p', (fields) => { + expect(uwtIsBroadcastMessage({ kind: 'log-query', ...fields })).toBe( + false + ); + }); + + it('should reject a log-query-result without a records array', () => { + expect( + uwtIsBroadcastMessage({ + kind: 'log-query-result', + id: 'q-1', + records: 'none' + }) + ).toBe(false); + }); + + it('should accept an identity-request message with no other fields', () => { + expect(uwtIsBroadcastMessage({ kind: 'identity-request' })).toBe(true); + }); + + it.each([ + {}, + { sessionId: 's-1' }, + { userId: 'u-1' }, + { + sessionId: 's-1', + userId: 'u-1' + } + ])('should accept an identity message %p', (fields) => { + expect(uwtIsBroadcastMessage({ kind: 'identity', ...fields })).toBe(true); + }); + + it.each([{ sessionId: 42 }, { userId: 42 }, { sessionId: 42, userId: 42 }])( + 'should reject an identity message with a non-string, non-undefined field %p', + (fields) => { + expect(uwtIsBroadcastMessage({ kind: 'identity', ...fields })).toBe( + false + ); + } + ); + }); +}); diff --git a/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.messages.ts b/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.messages.ts new file mode 100644 index 0000000..76eb481 --- /dev/null +++ b/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.messages.ts @@ -0,0 +1,430 @@ +import { + UWT_LOG_LEVEL_ORDER, + UwtLogRecord +} from '../../models/uwt-log.models/uwt-log.models'; +import { + UwtTelemetryError, + UwtTelemetryMetadata +} from '../../models/uwt-telemetry-common.models/uwt-telemetry-common.models'; +import { UwtLogQuery } from '../../providers/uwt-log-api.provider/uwt-log-api.provider'; +import { + UwtReporter, + uwtSafeVoid +} from '../../utils/uwt-telemetry-safe.util/uwt-telemetry-safe.util'; + +/** + * Default `BroadcastChannel` name used between telemetry realms. + * + * Named for the package that introduced it. Kept: realms deployed + * separately, on different versions, must still meet on it. + * + * @group Utils + */ +export const UWT_DEFAULT_BROADCAST_CHANNEL = 'ngx-ui-watchtower'; + +/** + * Everything that travels between telemetry realms. + * + * Structured-cloned by `BroadcastChannel`, so every payload must be plain + * data. + */ +export type UwtBroadcastMessage = + | { + kind: 'event'; + eventType: string; + payload: object; + metadata?: UwtTelemetryMetadata; + } + | { + kind: 'error'; + error: UwtTelemetryError; + metadata?: UwtTelemetryMetadata; + } + | { kind: 'user'; userId: string | undefined } + | { kind: 'flush'; beacon: boolean } + | { kind: 'identity-request' } + /** The host announcing shared identity. Always sends both fields, even when only one changed. */ + | { kind: 'identity'; sessionId?: string; userId?: string } + /** A follower's log record, for the host to hand to its log API provider. */ + | { kind: 'log'; record: UwtLogRecord } + /** A follower asking the host to flush its log API provider. */ + | { kind: 'log-flush' } + /** A follower reading records back through the host's log API provider. */ + | { kind: 'log-query'; id: string; filter: UwtLogQuery } + /** + * The host's answer to one `log-query`, matched by `id`. The records come + * from the host's backend unchecked; the asking follower keeps only those + * that pass {@link uwtIsBroadcastLogRecord}. + */ + | { kind: 'log-query-result'; id: string; records: unknown[] }; + +/** Every `kind` the union above accepts, for {@link uwtIsBroadcastMessage}. */ +const MESSAGE_KINDS: ReadonlySet = new Set([ + 'event', + 'error', + 'user', + 'flush', + 'identity-request', + 'identity', + 'log', + 'log-flush', + 'log-query', + 'log-query-result' +] as const); + +const LOG_LEVELS: ReadonlySet = new Set( + Object.keys(UWT_LOG_LEVEL_ORDER) +); + +/** Minimal `BroadcastChannel` surface this library relies on. */ +export interface UwtBroadcastChannelLike { + postMessage(message: unknown): void; + close(): void; + onmessage: ((event: { data: unknown }) => void) | null; +} + +/** + * Opens a broadcast channel, when the browser provides the API. + * + * Feature-detected: unavailable under jsdom (tests) and in server-side + * rendering. + */ +export function uwtOpenBroadcastChannel( + name: string +): UwtBroadcastChannelLike | undefined { + const Channel = ( + globalThis as { + BroadcastChannel?: new (name: string) => UwtBroadcastChannelLike; + } + ).BroadcastChannel; + + if (typeof Channel !== 'function') { + return undefined; + } + + try { + return new Channel(name); + } catch { + return undefined; + } +} + +/** Minimal Web Locks API surface this library relies on. */ +interface UwtLockManagerLike { + request(name: string, callback: () => Promise): Promise; +} + +/** + * Elects exactly one leader among same-origin realms on this channel, via + * the Web Locks API — prevents two shell tabs from both recording every + * forwarded message. Fails open (elects immediately) when Web Locks is + * unavailable. + * + * @returns releases the lock so the next queued realm can become leader. + */ +export function uwtElectBroadcastHostLeader( + channelName: string, + onElected: () => void +): () => void { + const locks = (globalThis as { navigator?: { locks?: UwtLockManagerLike } }) + .navigator?.locks; + + if (!locks) { + onElected(); + return () => undefined; + } + + let released = false; + let release: () => void = () => { + released = true; + }; + + const failOpen = () => { + if (!released) { + onElected(); + } + }; + + try { + locks + .request( + `ngx-ui-watchtower-host:${channelName}`, + () => + new Promise((resolve) => { + if (released) { + resolve(); + return; + } + release = () => { + released = true; + resolve(); + }; + onElected(); + }) + ) + .catch(failOpen); + } catch { + failOpen(); + } + + return () => release(); +} + +/** + * A realm's live connection to the shared broadcast channel. + * + * @group Interfaces + */ +export interface UwtBroadcastConnection { + /** The channel name in use. */ + readonly channelName: string; + + /** Whether a channel is open — false without `BroadcastChannel`, when not opened, and once closed. */ + isOpen(): boolean; + + /** Sends a message to every other realm on this channel. No-op if unavailable. */ + post(message: UwtBroadcastMessage): void; + + /** + * Adds a handler for incoming messages. No-op if no channel is open. + * + * @returns removes the handler + */ + onMessage(handler: (data: unknown) => void): () => void; + + /** Closes the channel and stops any further delivery. Safe to call more than once. */ + close(): void; +} + +/** + * Where, and whether, to connect. + * + * @group Interfaces + */ +export interface UwtBroadcastConnectOptions { + /** + * The channel's name; every realm on it must use the same one. Default + * {@link UWT_DEFAULT_BROADCAST_CHANNEL}. + */ + channelName?: string; + /** + * `false` opens no channel, so every call is a no-op: on a server, or + * while telemetry is off. Default `true`. + */ + open?: boolean; + /** Names this connection in failure reports. Default `broadcast`. */ + operation?: string; + /** Receives failures; see {@link UwtReporter}. */ + report?: UwtReporter; +} + +/** + * Opens this realm's connection to the shared broadcast channel: + * feature-detects `BroadcastChannel`, and wraps every operation in the + * library's fail-open guard. One connection can serve several handlers — + * a realm's broadcast sink and log provider share one. + * + * @param options the channel name, and whether to open it at all + * @returns the connection; a no-op one when no channel could be opened + * + * @group Utils + */ +export function uwtConnectBroadcastChannel( + options: UwtBroadcastConnectOptions = {} +): UwtBroadcastConnection { + const channelName = options.channelName ?? UWT_DEFAULT_BROADCAST_CHANNEL; + const operation = options.operation ?? 'broadcast'; + const { report } = options; + const handlers = new Set<(data: unknown) => void>(); + let channel = + options.open === false ? undefined : uwtOpenBroadcastChannel(channelName); + + if (channel) { + channel.onmessage = (event) => { + for (const handler of [...handlers]) { + uwtSafeVoid(`${operation}.receive`, () => handler(event.data), report); + } + }; + } + + return { + channelName, + isOpen() { + return channel !== undefined; + }, + post(message) { + uwtSafeVoid( + `${operation}.post`, + () => channel?.postMessage(message), + report + ); + }, + onMessage(handler) { + if (!channel) { + return () => undefined; + } + handlers.add(handler); + return () => { + handlers.delete(handler); + }; + }, + close() { + uwtSafeVoid( + `${operation}.close`, + () => { + handlers.clear(); + channel?.close(); + channel = undefined; + }, + report + ); + } + }; +} + +/** + * Whether `value` is a valid `metadata` field: absent, or a non-array + * object whose values are all finite numbers, strings, booleans, or null — + * the same flat-primitives contract `uwtRedactMetadata` enforces on the way + * out. Without this, a same-origin sender (not necessarily this library's + * own sink) could post a nested object on the channel that reaches a + * receiving sink's own sanitizer — e.g. `UwtRumTelemetrySink.sanitize()` — + * unchecked, since that step trusts the shape rather than re-validating it. + */ +function isValidBroadcastMetadata(value: unknown): boolean { + if (value === undefined) { + return true; + } + if (typeof value !== 'object' || value === null || Array.isArray(value)) { + return false; + } + return Object.values(value).every( + (v) => + v === null || + typeof v === 'string' || + typeof v === 'boolean' || + (typeof v === 'number' && Number.isFinite(v)) + ); +} + +function isOptionalString(value: unknown): boolean { + return value === undefined || typeof value === 'string'; +} + +function isRecord(value: unknown): value is Record { + return typeof value === 'object' && value !== null && !Array.isArray(value); +} + +function isValidTelemetryError(value: unknown): boolean { + return ( + isRecord(value) && + typeof value.name === 'string' && + typeof value.message === 'string' && + isOptionalString(value.stack) + ); +} + +function isLogLevel(value: unknown): boolean { + return typeof value === 'string' && LOG_LEVELS.has(value); +} + +/** + * Whether `value` has the shape of a {@link UwtLogRecord}: the required + * strings, a known level, optional strings where the record has them, and + * metadata and error held to the same rules as on an `event` or `error` + * message. + */ +export function uwtIsBroadcastLogRecord(value: unknown): value is UwtLogRecord { + return ( + isRecord(value) && + typeof value.timestamp === 'string' && + isLogLevel(value.level) && + typeof value.message === 'string' && + typeof value.application === 'string' && + typeof value.environment === 'string' && + typeof value.version === 'string' && + isOptionalString(value.logger) && + isOptionalString(value.context) && + isOptionalString(value.correlationId) && + isOptionalString(value.userId) && + isOptionalString(value.sessionId) && + isValidBroadcastMetadata(value.metadata) && + (value.error === undefined || isValidTelemetryError(value.error)) + ); +} + +function isValidLogQuery(value: unknown): boolean { + return ( + isRecord(value) && + isOptionalString(value.correlationId) && + isOptionalString(value.logger) && + (value.minLevel === undefined || isLogLevel(value.minLevel)) && + isOptionalString(value.from) && + isOptionalString(value.to) && + (value.limit === undefined || + (typeof value.limit === 'number' && Number.isFinite(value.limit))) + ); +} + +/** + * Narrows an incoming `BroadcastChannel` payload to a telemetry message. + * + * Checks each kind's required fields, not just `kind`, so a same-named + * message from something else on the channel is rejected. + */ +export function uwtIsBroadcastMessage( + data: unknown +): data is UwtBroadcastMessage { + if (typeof data !== 'object' || data === null) { + return false; + } + + const message = data as Record; + const kind = message.kind as UwtBroadcastMessage['kind']; + if (!MESSAGE_KINDS.has(kind)) { + return false; + } + + switch (kind) { + case 'event': + return ( + typeof message.eventType === 'string' && + typeof message.payload === 'object' && + message.payload !== null && + isValidBroadcastMetadata(message.metadata) + ); + case 'error': + return ( + isValidTelemetryError(message.error) && + isValidBroadcastMetadata(message.metadata) + ); + case 'flush': + return typeof message.beacon === 'boolean'; + case 'user': + // Absent or `undefined` userId both mean "clear it". + return message.userId === undefined || typeof message.userId === 'string'; + case 'identity': + return ( + (message.sessionId === undefined || + typeof message.sessionId === 'string') && + (message.userId === undefined || typeof message.userId === 'string') + ); + case 'identity-request': + case 'log-flush': + return true; + case 'log': + return uwtIsBroadcastLogRecord(message.record); + case 'log-query': + return ( + typeof message.id === 'string' && + message.id !== '' && + isValidLogQuery(message.filter) + ); + case 'log-query-result': + return ( + typeof message.id === 'string' && + message.id !== '' && + Array.isArray(message.records) + ); + } +} diff --git a/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.spec.ts b/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.spec.ts new file mode 100644 index 0000000..578f68e --- /dev/null +++ b/projects/ui-watchtower/src/lib/sinks/uwt-broadcast/uwt-broadcast.spec.ts @@ -0,0 +1,245 @@ +/** + * @jest-environment jsdom + */ +import { UwtLogRecord } from '../../models/uwt-log.models/uwt-log.models'; +import { + UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS, + UwtBroadcastLogApiProvider +} from '../../providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider'; +import { UwtNoopLogApiProvider } from '../../providers/uwt-noop-log-api.provider/uwt-noop-log-api.provider'; +import { UwtNoopTelemetrySink } from '../uwt-telemetry/uwt-noop-telemetry.sink/uwt-noop-telemetry.sink'; +import { UwtTelemetryBroadcastHost } from './uwt-broadcast-host'; +import { UwtBroadcastTelemetrySink } from './uwt-broadcast-telemetry.sink'; +import { uwtConnectBroadcastChannel } from './uwt-broadcast.messages'; + +/** + * `BroadcastChannel` for the test: delivers asynchronously to every other + * open instance of the same name. + */ +class ChannelStub { + static readonly open = new Set(); + onmessage: ((event: { data: unknown }) => void) | null = null; + + constructor(readonly name: string) { + ChannelStub.open.add(this); + } + + postMessage(message: unknown): void { + for (const peer of ChannelStub.open) { + if (peer !== this && peer.name === this.name) { + const data = JSON.parse(JSON.stringify(message)); + setTimeout(() => peer.onmessage?.({ data }), 0); + } + } + } + + close(): void { + ChannelStub.open.delete(this); + } +} + +async function settle(): Promise { + for (let turn = 0; turn < 5; turn++) { + await new Promise((resolve) => setTimeout(resolve, 0)); + } +} + +const identity = { application: 'cart', environment: 'test', version: '1' }; + +describe('the core broadcast classes', () => { + let hosts: UwtTelemetryBroadcastHost[]; + + /** A host, disposed after the test whatever happens. */ + function hostOf( + options: ConstructorParameters[0] + ): UwtTelemetryBroadcastHost { + const host = new UwtTelemetryBroadcastHost(options); + hosts.push(host); + return host; + } + + beforeEach(() => { + hosts = []; + Object.defineProperty(globalThis, 'BroadcastChannel', { + value: ChannelStub, + configurable: true, + writable: true + }); + }); + + afterEach(() => { + hosts.forEach((host) => host.dispose()); + ChannelStub.open.clear(); + delete (globalThis as { BroadcastChannel?: unknown }).BroadcastChannel; + jest.useRealTimers(); + jest.restoreAllMocks(); + }); + + describe('a shared connection', () => { + it('should serve a sink and a log provider over one channel', async () => { + const connection = uwtConnectBroadcastChannel(); + const sink = new UwtBroadcastTelemetrySink({ identity, connection }); + const logApi = new UwtBroadcastLogApiProvider({ connection }); + const host = hostOf({ + sink: new UwtNoopTelemetrySink(), + logApi: { + send: () => undefined, + query: async () => [record('found')] + } + }); + await settle(); + + expect(ChannelStub.open.size).toBe(2); + const query = logApi.query({}); + await settle(); + expect((await query).map((r) => r.message)).toEqual(['found']); + + sink.flush(); + await settle(); + expect(host.received).toBe(2); + sink.dispose(); + logApi.dispose(); + connection.close(); + }); + + it('should be closed by whoever opened it, not by those it was lent to', () => { + const connection = uwtConnectBroadcastChannel(); + const sink = new UwtBroadcastTelemetrySink({ identity, connection }); + const logApi = new UwtBroadcastLogApiProvider({ connection }); + + sink.dispose(); + logApi.dispose(); + expect(connection.isOpen()).toBe(true); + + connection.close(); + expect(ChannelStub.open.size).toBe(0); + }); + + it('should be opened, and closed, by a sink given none', () => { + const sink = new UwtBroadcastTelemetrySink({ identity }); + expect(ChannelStub.open.size).toBe(1); + + sink.dispose(); + sink.dispose(); + expect(ChannelStub.open.size).toBe(0); + }); + + it('should open nothing while disabled', async () => { + const sink = new UwtBroadcastTelemetrySink({ identity, enabled: false }); + const logApi = new UwtBroadcastLogApiProvider({ enabled: false }); + + expect(ChannelStub.open.size).toBe(0); + expect(() => sink.record('com.uwt.bi', {})).not.toThrow(); + await expect(logApi.query({})).resolves.toEqual([]); + }); + }); + + describe('a log query', () => { + it('should resolve to [] when no host answers in time, its timer scheduled through runOutside', async () => { + jest.useFakeTimers(); + let outside = 0; + const logApi = new UwtBroadcastLogApiProvider({ + runOutside: (callback) => { + outside++; + return callback(); + } + }); + + const query = logApi.query({}); + jest.advanceTimersByTime(UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS); + + await expect(query).resolves.toEqual([]); + expect(outside).toBe(1); + expect(jest.getTimerCount()).toBe(0); + logApi.dispose(); + }); + + it('should resolve to [] when the provider is disposed, leaving no timer', async () => { + jest.useFakeTimers(); + const logApi = new UwtBroadcastLogApiProvider(); + + const query = logApi.query({}); + logApi.dispose(); + + await expect(query).resolves.toEqual([]); + expect(jest.getTimerCount()).toBe(0); + }); + }); + + describe('the host', () => { + const shell = () => ({ + sink: new UwtNoopTelemetrySink(), + logApi: new UwtNoopLogApiProvider() + }); + + it('should report a second host in this realm to its reporter', () => { + const report = jest.fn(); + const first = hostOf({ ...shell(), report }); + const second = hostOf({ ...shell(), report }); + + expect(report).toHaveBeenCalledWith( + 'warn', + expect.stringContaining('a second telemetry host is active') + ); + first.dispose(); + second.dispose(); + }); + + it('should warn on the console without a reporter: a misconfigured host is never silent', () => { + const warn = jest.spyOn(console, 'warn').mockImplementation(() => {}); + const forwarding = hostOf({ + sink: new UwtBroadcastTelemetrySink({ identity }), + logApi: new UwtNoopLogApiProvider() + }); + + expect(warn).toHaveBeenCalledWith( + expect.stringMatching(/^\[ui-watchtower\] .*cannot also host it/) + ); + forwarding.dispose(); + }); + + it('should neither count nor open anything while disabled', () => { + const report = jest.fn(); + const disabled = hostOf({ + ...shell(), + enabled: false, + report + }); + const active = hostOf({ ...shell(), report }); + + expect(report).not.toHaveBeenCalled(); + expect(ChannelStub.open.size).toBe(1); + disabled.dispose(); + active.dispose(); + }); + + it('should give up its place only once, however often it is disposed', () => { + const report = jest.fn(); + const first = hostOf({ ...shell(), report }); + const second = hostOf({ ...shell(), report }); + second.dispose(); + second.dispose(); + report.mockClear(); + + const third = hostOf({ ...shell(), report }); + + expect(report).toHaveBeenCalledWith( + 'warn', + expect.stringContaining('a second telemetry host is active') + ); + first.dispose(); + third.dispose(); + }); + }); +}); + +function record(message: string): UwtLogRecord { + return { + timestamp: '2026-01-01T00:00:00.000Z', + level: 'log', + message, + application: 'shell', + environment: 'test', + version: '1' + }; +} diff --git a/projects/ui-watchtower/src/public-api.ts b/projects/ui-watchtower/src/public-api.ts index 7cf1fbc..665096d 100644 --- a/projects/ui-watchtower/src/public-api.ts +++ b/projects/ui-watchtower/src/public-api.ts @@ -89,6 +89,27 @@ export type { } from './lib/providers/uwt-log-api.provider/uwt-log-api.provider'; export { UwtNoopLogApiProvider } from './lib/providers/uwt-noop-log-api.provider/uwt-noop-log-api.provider'; +// Shell and fragments: a fragment forwards its telemetry and logs over a +// BroadcastChannel to the shell's host, which records them +export { + UWT_DEFAULT_BROADCAST_CHANNEL, + uwtConnectBroadcastChannel +} from './lib/sinks/uwt-broadcast/uwt-broadcast.messages'; +export type { + UwtBroadcastConnection, + UwtBroadcastConnectOptions, + UwtBroadcastMessage +} from './lib/sinks/uwt-broadcast/uwt-broadcast.messages'; +export { UwtBroadcastTelemetrySink } from './lib/sinks/uwt-broadcast/uwt-broadcast-telemetry.sink'; +export type { UwtBroadcastTelemetrySinkOptions } from './lib/sinks/uwt-broadcast/uwt-broadcast-telemetry.sink'; +export { + UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS, + UwtBroadcastLogApiProvider +} from './lib/providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider'; +export type { UwtBroadcastLogApiProviderOptions } from './lib/providers/uwt-broadcast-log-api.provider/uwt-broadcast-log-api.provider'; +export { UwtTelemetryBroadcastHost } from './lib/sinks/uwt-broadcast/uwt-broadcast-host'; +export type { UwtTelemetryBroadcastHostOptions } from './lib/sinks/uwt-broadcast/uwt-broadcast-host'; + // Utilities an application or a destination needs export { uwtClassifyTelemetryEvent } from './lib/utils/uwt-telemetry-event.util/uwt-telemetry-event.util'; export type { UwtTelemetrySinkEvent } from './lib/utils/uwt-telemetry-event.util/uwt-telemetry-event.util'; diff --git a/projects/ui-watchtower/type-tests/compat.ts b/projects/ui-watchtower/type-tests/compat.ts index 7769c5d..47b06d2 100644 --- a/projects/ui-watchtower/type-tests/compat.ts +++ b/projects/ui-watchtower/type-tests/compat.ts @@ -11,6 +11,18 @@ import { UwtTraceScenarioOptions } from '@absaoss-cps/ui-watchtower/rxjs'; import { + UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS, + UWT_DEFAULT_BROADCAST_CHANNEL, + UwtBroadcastConnection, + UwtBroadcastConnectOptions, + UwtBroadcastLogApiProvider, + UwtBroadcastLogApiProviderOptions, + UwtBroadcastMessage, + UwtBroadcastTelemetrySink, + UwtBroadcastTelemetrySinkOptions, + uwtConnectBroadcastChannel, + UwtTelemetryBroadcastHost, + UwtTelemetryBroadcastHostOptions, UWT_DEFAULT_EVENT_NAMESPACE, UWT_DEFAULT_REDACT_CONFIG, UWT_DEFAULT_TELEMETRY_CONFIG, @@ -238,3 +250,47 @@ export function concerns( new UwtLogTelemetry(logOptions) ]; } + +/** A composed page without a framework: a fragment, and the shell hosting it. */ +export function composedPage( + shellSink: UwtTelemetrySink, + shellLogs: UwtLogApiProvider +): () => void { + const connectOptions: UwtBroadcastConnectOptions = { + channelName: UWT_DEFAULT_BROADCAST_CHANNEL, + open: true, + report + }; + const connection: UwtBroadcastConnection = + uwtConnectBroadcastChannel(connectOptions); + const sinkOptions: UwtBroadcastTelemetrySinkOptions = { + identity, + connection + }; + const logOptions: UwtBroadcastLogApiProviderOptions = { + connection, + runOutside: (callback) => callback(), + runInside: (callback) => callback() + }; + const fragment = uwtCreateTelemetry({ + identity, + destination: new UwtBroadcastTelemetrySink(sinkOptions), + logApi: new UwtBroadcastLogApiProvider(logOptions) + }); + + const hostOptions: UwtTelemetryBroadcastHostOptions = { + sink: shellSink, + logApi: shellLogs, + report + }; + const host = new UwtTelemetryBroadcastHost(hostOptions); + const flush: UwtBroadcastMessage = { kind: 'flush', beacon: true }; + connection.post(flush); + + return () => { + report('warn', `${host.received} of ${UWT_BROADCAST_LOG_QUERY_TIMEOUT_MS}`); + fragment.dispose(); + host.dispose(); + connection.close(); + }; +}