apps/backend/src/modules/observability/application/telemetry-runtime.service.tsTypeScript
import { Injectable } from '@nestjs/common';
import {
TELEMETRY_REMOTE_CATEGORY_KEYS,
TelemetrySettingsService,
type TelemetryRemoteCategory,
} from './telemetry-settings.service';
import {
TelemetryDispatchPolicyService,
type RemoteTelemetryQueueItem,
} from './telemetry-dispatch-policy.service';
export type TelemetryDataOrigin = 'production' | 'synthetic';
export interface RemoteTelemetryEvent extends RemoteTelemetryQueueItem {
organizationId: string;
dataOrigin: TelemetryDataOrigin;
occurredAt: string;
payload: Record<string, unknown>;
}
/**
* A transport is deliberately supplied by the caller. There is no default
* remote transport and no fallback transport: the category policy and the
* data-origin boundary must be applied before a provider can receive data.
*/
export interface RemoteTelemetryTransport {
readonly dataOrigin: TelemetryDataOrigin;
send(event: RemoteTelemetryEvent): Promise<void>;
}
export interface TelemetryDispatchResult {
id: string;
sent: boolean;
action: 'send' | 'retain_local' | 'discard_existing' | 'manual_only';
reason:
| 'category_enabled'
| 'category_disabled'
| 'pilot_mode_required'
| 'manual_upload_required';
}
export interface TelemetryQueueDispatchResult {
sent: string[];
retainedLocal: string[];
discarded: string[];
}
@Injectable()
export class TelemetryRuntimeService {
constructor(
private readonly settings: TelemetrySettingsService,
private readonly policy: TelemetryDispatchPolicyService,
) {}
async dispatch(
event: RemoteTelemetryEvent,
transport: RemoteTelemetryTransport,
): Promise<TelemetryDispatchResult> {
this.assertEvent(event);
this.assertTransport(event, transport);
const settings = await this.settings.get(event.organizationId);
const decision = this.policy.decide(event.category, settings);
if (!decision.allowed) {
return {
id: event.id,
sent: false,
action: decision.action,
reason: decision.reason,
};
}
await transport.send(event);
return {
id: event.id,
sent: true,
action: decision.action,
reason: decision.reason,
};
}
async dispatchQueue(
events: RemoteTelemetryEvent[],
transport: RemoteTelemetryTransport,
): Promise<TelemetryQueueDispatchResult> {
if (!Array.isArray(events) || events.some((event) => !this.isEvent(event))) {
throw new Error('telemetry.runtime_queue_invalid');
}
if (events.length === 0) {
return {sent: [], retainedLocal: [], discarded: []};
}
this.assertTransport(events[0], transport);
if (events.some((event) => event.organizationId !== events[0].organizationId)) {
throw new Error('telemetry.runtime_queue_scope_invalid');
}
const settings = await this.settings.get(events[0].organizationId);
const queue = events.map((event) => ({id: event.id, category: event.category}));
const classified = this.policy.applyQueuePolicy(queue, settings);
const sendable = new Set(classified.sendable.map((item) => item.id));
const result: TelemetryQueueDispatchResult = {
sent: [],
retainedLocal: classified.retainedLocal.map((item) => item.id),
discarded: classified.discarded.map((item) => item.id),
};
// Sending stops on the first provider failure. The unsent events are not
// re-routed: the caller retains them locally for a later policy decision.
for (const event of events) {
if (!sendable.has(event.id)) continue;
await transport.send(event);
result.sent.push(event.id);
}
return result;
}
private assertTransport(
event: RemoteTelemetryEvent,
transport: RemoteTelemetryTransport,
): void {
if (
!transport ||
(transport.dataOrigin !== 'production' && transport.dataOrigin !== 'synthetic') ||
transport.dataOrigin !== event.dataOrigin ||
typeof transport.send !== 'function'
) {
throw new Error('telemetry.runtime_transport_origin_invalid');
}
}
private assertEvent(event: RemoteTelemetryEvent): void {
if (!this.isEvent(event)) throw new Error('telemetry.runtime_event_invalid');
}
private isEvent(value: unknown): value is RemoteTelemetryEvent {
if (!value || typeof value !== 'object' || Array.isArray(value)) return false;
const event = value as Record<string, unknown>;
return (
typeof event.id === 'string' &&
/^[a-zA-Z0-9._:-]{1,200}$/.test(event.id) &&
typeof event.organizationId === 'string' &&
event.organizationId.trim().length > 0 &&
event.organizationId.length <= 200 &&
this.isCategory(event.category) &&
(event.dataOrigin === 'production' || event.dataOrigin === 'synthetic') &&
typeof event.occurredAt === 'string' &&
!Number.isNaN(Date.parse(event.occurredAt)) &&
Boolean(event.payload) &&
typeof event.payload === 'object' &&
!Array.isArray(event.payload) &&
Object.keys(event.payload as object).length <= 64
);
}
private isCategory(value: unknown): value is TelemetryRemoteCategory {
return (
typeof value === 'string' &&
(TELEMETRY_REMOTE_CATEGORY_KEYS as readonly string[]).includes(value)
);
}
}