feat: enhance federation-sdk with metrics tracking and new utility functions

packages/federation-sdk/package.json

@@ -17,6 +17,7 @@
 	},
 	"dependencies": {
 		"@opentelemetry/api": "^1.9.0",
+		"prom-client": "^15.1.3",
 		"@rocket.chat/emitter": "^0.31.25",
 		"@rocket.chat/federation-core": "workspace:*",
 		"@rocket.chat/federation-crypto": "workspace:*",

packages/federation-sdk/src/index.ts

@@ -78,6 +78,19 @@ export {
 	type ITracedClassOptions,
 } from './utils/tracing';
 
+export { federationMetrics, initMetrics } from './metrics';
+export {
+	bucketizeEduCount,
+	bucketizePduCount,
+	determineMessageType,
+	extractOriginFromMatrixRoomId,
+	extractOriginFromMatrixUserId,
+	getEventTypeLabel,
+} from './metrics/helpers';
+
+// Event emitter types
+export type { EventHandlerExceptionHandler } from './services/event-emitter.service';
+
 export type HomeserverEventSignatures = {
 	'homeserver.ping': {
 		message: string;

packages/federation-sdk/src/metrics/helpers.ts

@@ -0,0 +1,87 @@
+/**
+ * Extracts the origin server domain from a Matrix room ID.
+ * @example extractOriginFromMatrixRoomId('!room:matrix.org') // 'matrix.org'
+ */
+export function extractOriginFromMatrixRoomId(roomId: string): string {
+	return roomId.split(':').pop() || 'unknown';
+}
+
+/**
+ * Extracts the origin server domain from a Matrix user ID.
+ * @example extractOriginFromMatrixUserId('@user:matrix.org') // 'matrix.org'
+ */
+export function extractOriginFromMatrixUserId(userId: string): string {
+	return userId.split(':').pop() || 'unknown';
+}
+
+// File types for message type detection
+const fileTypes = ['m.image', 'm.video', 'm.audio', 'm.file'];
+
+/**
+ * Determines the message type from a Matrix event for metrics labeling.
+ * @returns 'text' | 'file' | 'encrypted'
+ */
+export function determineMessageType(event: {
+	type?: string;
+	content?: { msgtype?: string };
+}): 'text' | 'file' | 'encrypted' {
+	if (event.type === 'm.room.encrypted') {
+		return 'encrypted';
+	}
+
+	const msgtype = event.content?.msgtype;
+	if (msgtype && fileTypes.includes(msgtype)) {
+		return 'file';
+	}
+
+	return 'text';
+}
+
+/**
+ * Bucketizes PDU count for metrics labeling to avoid high cardinality.
+ * Groups counts into buckets: 0, 1, 2-5, 6-10, 11-50, 51+
+ */
+export function bucketizePduCount(count: number): string {
+	if (count === 0) return '0';
+	if (count === 1) return '1';
+	if (count <= 5) return '2-5';
+	if (count <= 10) return '6-10';
+	if (count <= 50) return '11-50';
+	return '51+';
+}
+
+/**
+ * Bucketizes EDU count for metrics labeling to avoid high cardinality.
+ * Groups counts into buckets: 0, 1, 2-5, 6-10, 11+
+ */
+export function bucketizeEduCount(count: number): string {
+	if (count === 0) return '0';
+	if (count === 1) return '1';
+	if (count <= 5) return '2-5';
+	if (count <= 10) return '6-10';
+	return '11+';
+}
+
+/**
+ * Maps event emitter event types to simplified event_type labels for metrics.
+ */
+export function getEventTypeLabel(event: string): string {
+	// Map homeserver.matrix.* events to simplified labels
+	const mapping: Record<string, string> = {
+		'homeserver.matrix.message': 'message',
+		'homeserver.matrix.encrypted': 'message',
+		'homeserver.matrix.membership': 'membership',
+		'homeserver.matrix.redaction': 'redaction',
+		'homeserver.matrix.reaction': 'reaction',
+		'homeserver.matrix.typing': 'typing',
+		'homeserver.matrix.presence': 'presence',
+		'homeserver.matrix.room.create': 'room_create',
+		'homeserver.matrix.room.name': 'room_update',
+		'homeserver.matrix.room.topic': 'room_update',
+		'homeserver.matrix.room.power_levels': 'room_update',
+		'homeserver.matrix.room.server_acl': 'room_update',
+		'homeserver.matrix.room.role': 'role_change',
+		'homeserver.matrix.encryption': 'encryption',
+	};
+	return mapping[event] || event.replace('homeserver.matrix.', '');
+}

packages/federation-sdk/src/metrics/index.ts

@@ -0,0 +1,131 @@
+import client, { type Registry } from 'prom-client';
+
+let registry: Registry = client.register;
+
+export function initMetrics(opts: { registry: Registry }) {
+	registry = opts.registry;
+}
+
+const percentiles = [0.01, 0.1, 0.5, 0.9, 0.95, 0.99, 1];
+
+/**
+ * Gets an existing metric from the registry or creates it if it doesn't exist.
+ * This ensures we don't get duplicate registration errors when the SDK
+ * is used alongside other apps that also register metrics.
+ */
+function getOrCreateMetric<T extends client.Metric>(
+	name: string,
+	createFn: () => T,
+): T {
+	const existing = registry.getSingleMetric(name);
+	if (existing) {
+		return existing as T;
+	}
+	return createFn();
+}
+
+/**
+ * Federation metrics for incoming operations.
+ */
+export const federationMetrics = {
+	/** Counter for federation events processed */
+	get federationEventsProcessed() {
+		return getOrCreateMetric(
+			'rocketchat_federation_events_processed',
+			() =>
+				new client.Counter({
+					name: 'rocketchat_federation_events_processed',
+					labelNames: ['event_type', 'direction'],
+					help: 'Total federation events processed',
+					registers: [registry],
+				}),
+		);
+	},
+
+	/** Counter for failed federation events */
+	get federationEventsFailed() {
+		return getOrCreateMetric(
+			'rocketchat_federation_events_failed',
+			() =>
+				new client.Counter({
+					name: 'rocketchat_federation_events_failed',
+					labelNames: ['event_type', 'direction', 'error_type'],
+					help: 'Total federation events that failed to process',
+					registers: [registry],
+				}),
+		);
+	},
+
+	/** Counter for messages received from other federated servers */
+	get federatedMessagesReceived() {
+		return getOrCreateMetric(
+			'rocketchat_federation_messages_received',
+			() =>
+				new client.Counter({
+					name: 'rocketchat_federation_messages_received',
+					labelNames: ['message_type', 'origin'],
+					help: 'Total federated messages received',
+					registers: [registry],
+				}),
+		);
+	},
+
+	/** Counter for rooms joined */
+	get federatedRoomsJoined() {
+		return getOrCreateMetric(
+			'rocketchat_federation_rooms_joined',
+			() =>
+				new client.Counter({
+					name: 'rocketchat_federation_rooms_joined',
+					labelNames: ['origin'],
+					help: 'Total federated rooms joined',
+					registers: [registry],
+				}),
+		);
+	},
+
+	/** Duration to process incoming federation transaction */
+	get federationTransactionProcessDuration() {
+		return getOrCreateMetric(
+			'rocketchat_federation_transaction_process_duration_seconds',
+			() =>
+				new client.Summary({
+					name: 'rocketchat_federation_transaction_process_duration_seconds',
+					labelNames: ['pdu_count', 'edu_count', 'origin'],
+					help: 'Time to process incoming federation transaction',
+					percentiles,
+					registers: [registry],
+				}),
+		);
+	},
+
+	/** Duration to process incoming federated message */
+	get federationIncomingMessageProcessDuration() {
+		return getOrCreateMetric(
+			'rocketchat_federation_incoming_message_process_duration_seconds',
+			() =>
+				new client.Summary({
+					name: 'rocketchat_federation_incoming_message_process_duration_seconds',
+					labelNames: ['message_type'],
+					help: 'Time to process incoming federated message',
+					percentiles,
+					registers: [registry],
+				}),
+		);
+	},
+
+	/** Duration to join a federated room */
+	get federationRoomJoinDuration() {
+		return getOrCreateMetric(
+			'rocketchat_federation_room_join_duration_seconds',
+			() =>
+				new client.Summary({
+					name: 'rocketchat_federation_room_join_duration_seconds',
+					labelNames: ['origin'],
+					help: 'Time to join a federated room (invite acceptance)',
+					percentiles,
+					registers: [registry],
+				}),
+		);
+	},
+};