import { createHash } from 'node:crypto'; import type { RuntimeHealth, RuntimeMessage, RuntimeScope, RuntimeSessionState, RuntimeStreamEvent, } from '@mosaicstack/types'; export type MatrixRuntimeTransportErrorCode = | 'forbidden' | 'invalid_request' | 'not_found' | 'unavailable'; export class MatrixRuntimeTransportError extends Error { constructor( readonly code: MatrixRuntimeTransportErrorCode, message: string, ) { super(message); this.name = MatrixRuntimeTransportError.name; } } /** Minimal injectable Matrix fetch surface; it avoids coupling this transport to an SDK. */ export interface MatrixFetchLike { ( url: string, init?: { method?: string; headers?: Record; body?: string }, ): Promise<{ ok: boolean; status: number; json(): Promise; text(): Promise; }>; } /** A server-configured runtime session binding. Callers never select a Matrix room. */ export interface MatrixNativeRuntimeSessionBinding { id: string; runtimeId: string; roomId: string; remoteUserId: string; parentSessionId?: string; state: RuntimeSessionState; createdAt: string; updatedAt: string; } export interface MatrixNativeRuntimeTransportOptions { homeserverUrl: string; accessToken: string; /** Matrix user authenticated by the service token. */ userId: string; sessions: readonly MatrixNativeRuntimeSessionBinding[]; fetchImpl?: MatrixFetchLike; now?: () => Date; } interface MatrixSyncEvent { type?: string; sender?: string; origin_server_ts?: number; content?: Record; } interface MatrixSyncResponse { next_batch?: string; rooms?: { join?: Record }; } /** * Concrete Matrix CS-API transport for the native provider. It authenticates * the configured sender before each operation and resolves rooms exclusively * from configured bindings, preventing client-selected room/identity routing. */ export class MatrixNativeRuntimeTransport { private readonly baseUrl: string; private readonly fetchImpl: MatrixFetchLike; private readonly now: () => Date; private readonly bindings: ReadonlyMap; constructor(private readonly options: MatrixNativeRuntimeTransportOptions) { this.baseUrl = validatedHomeserverUrl(options.homeserverUrl); if (!options.accessToken.trim()) { throw new MatrixRuntimeTransportError('invalid_request', 'Matrix access token is required'); } if (!options.userId.trim()) { throw new MatrixRuntimeTransportError('invalid_request', 'Matrix user identity is required'); } this.bindings = new Map( options.sessions.map((binding) => [binding.id, Object.freeze({ ...binding })]), ); if (this.bindings.size !== options.sessions.length) { throw new MatrixRuntimeTransportError( 'invalid_request', 'Matrix runtime session IDs must be unique', ); } this.fetchImpl = options.fetchImpl ?? (globalThis.fetch as unknown as MatrixFetchLike); this.now = options.now ?? (() => new Date()); } async health(_scope: RuntimeScope): Promise { try { const versions = await this.request('/_matrix/client/versions', { method: 'GET' }); if (!versions.ok) { return this.healthResult('down', `versions HTTP ${versions.status}`); } await this.assertIdentity(); return this.healthResult('healthy'); } catch (error: unknown) { return this.healthResult('down', message(error)); } } async listSessions(scope: RuntimeScope): Promise { await this.assertIdentity(); void scope; return [...this.bindings.values()].map((binding) => ({ ...binding })); } async verifySession( sessionId: string, scope: RuntimeScope, ): Promise { await this.assertIdentity(); void scope; const binding = this.bindings.get(sessionId); if (!binding) { throw new MatrixRuntimeTransportError( 'not_found', 'Matrix runtime session is not configured', ); } return { ...binding }; } async *stream( sessionId: string, cursor: string | undefined, scope: RuntimeScope, ): AsyncIterable { const binding = await this.verifySession(sessionId, scope); const query = new URLSearchParams({ timeout: '0' }); if (cursor?.trim()) query.set('since', cursor); const response = await this.request(`/_matrix/client/v3/sync?${query.toString()}`, { method: 'GET', headers: this.authHeaders(), }); if (!response.ok) { throw new MatrixRuntimeTransportError( 'unavailable', `Matrix sync failed: HTTP ${response.status}`, ); } const payload = (await response.json()) as MatrixSyncResponse; const nextCursor = payload.next_batch?.trim() || cursor?.trim() || 'initial'; const events = payload.rooms?.join?.[binding.roomId]?.timeline?.events ?? []; for (const event of events) { const normalized = normalizeEvent(event, binding, nextCursor); if (normalized) yield normalized; } } async send(sessionId: string, message: RuntimeMessage, scope: RuntimeScope): Promise { const binding = await this.verifySession(sessionId, scope); const txnId = transactionId('send', binding.id, message.idempotencyKey); const response = await this.request( `/_matrix/client/v3/rooms/${encodeURIComponent(binding.roomId)}/send/m.room.message/${encodeURIComponent(txnId)}`, { method: 'PUT', headers: this.authHeaders(), body: JSON.stringify({ msgtype: 'm.text', body: message.content, 'mosaic.runtime.v1': metadata(binding, scope, message.idempotencyKey), }), }, ); if (!response.ok) { throw new MatrixRuntimeTransportError( 'unavailable', `Matrix message delivery failed: HTTP ${response.status}`, ); } } async terminate(sessionId: string, approvalRef: string, scope: RuntimeScope): Promise { const binding = await this.verifySession(sessionId, scope); const txnId = transactionId('terminate', binding.id, `${approvalRef}:${scope.correlationId}`); const response = await this.request( `/_matrix/client/v3/rooms/${encodeURIComponent(binding.roomId)}/send/mosaic.runtime.terminate/${encodeURIComponent(txnId)}`, { method: 'PUT', headers: this.authHeaders(), body: JSON.stringify({ ...metadata(binding, scope), approval_ref: approvalRef, }), }, ); if (!response.ok) { throw new MatrixRuntimeTransportError( 'unavailable', `Matrix termination delivery failed: HTTP ${response.status}`, ); } } private async assertIdentity(): Promise { let response: Awaited>; try { response = await this.request('/_matrix/client/v3/account/whoami', { method: 'GET', headers: this.authHeaders(), }); } catch (error: unknown) { throw new MatrixRuntimeTransportError('unavailable', message(error)); } if (!response.ok) { throw new MatrixRuntimeTransportError( 'forbidden', `Matrix whoami failed: HTTP ${response.status}`, ); } const body = (await response.json()) as { user_id?: unknown }; if (body.user_id !== this.options.userId) { throw new MatrixRuntimeTransportError( 'forbidden', 'Matrix whoami identity does not match configuration', ); } } private request( path: string, init: { method?: string; headers?: Record; body?: string }, ) { return this.fetchImpl(`${this.baseUrl}${path}`, init); } private authHeaders(): Record { return { Authorization: `Bearer ${this.options.accessToken}`, 'Content-Type': 'application/json', }; } private healthResult(status: RuntimeHealth['status'], detail?: string): RuntimeHealth { return { status, checkedAt: this.now().toISOString(), ...(detail ? { detail } : {}) }; } } function metadata( binding: MatrixNativeRuntimeSessionBinding, scope: RuntimeScope, idempotencyKey?: string, ): Record { return { session_id: binding.id, runtime_id: binding.runtimeId, actor_id: scope.actorId, tenant_id: scope.tenantId, channel_id: scope.channelId, correlation_id: scope.correlationId, ...(idempotencyKey ? { idempotency_key: idempotencyKey } : {}), }; } function transactionId(kind: 'send' | 'terminate', sessionId: string, identity: string): string { return `mosaic-${kind}-${createHash('sha256').update(`${sessionId}\u0000${identity}`).digest('base64url')}`; } function normalizeEvent( event: MatrixSyncEvent, binding: MatrixNativeRuntimeSessionBinding, cursor: string, ): RuntimeStreamEvent | undefined { if (event.type !== 'mosaic.runtime.event' || event.sender !== binding.remoteUserId) return undefined; const payload = event.content?.['mosaic.runtime.v1']; if ( !isRecord(payload) || payload['session_id'] !== binding.id || typeof payload['type'] !== 'string' ) { return undefined; } const occurredAt = typeof payload['occurred_at'] === 'string' ? payload['occurred_at'] : new Date(event.origin_server_ts ?? 0).toISOString(); const eventCursor = typeof payload['cursor'] === 'string' ? payload['cursor'] : cursor; switch (payload['type']) { case 'session.state': return isState(payload['state']) ? { type: 'session.state', sessionId: binding.id, cursor: eventCursor, occurredAt, state: payload['state'], } : undefined; case 'message.delta': return typeof payload['content'] === 'string' ? { type: 'message.delta', sessionId: binding.id, cursor: eventCursor, occurredAt, content: payload['content'], } : undefined; case 'message.complete': return typeof payload['message_id'] === 'string' ? { type: 'message.complete', sessionId: binding.id, cursor: eventCursor, occurredAt, messageId: payload['message_id'], } : undefined; default: return undefined; } } function isRecord(value: unknown): value is Record { return typeof value === 'object' && value !== null; } function isState(value: unknown): value is RuntimeSessionState { return ( value === 'starting' || value === 'active' || value === 'idle' || value === 'stopped' || value === 'failed' ); } function validatedHomeserverUrl(value: string): string { let url: URL; try { url = new URL(value); } catch { throw new MatrixRuntimeTransportError( 'invalid_request', 'Matrix homeserver URL must be absolute', ); } if (url.protocol !== 'https:') { throw new MatrixRuntimeTransportError( 'invalid_request', 'Matrix homeserver URL must use HTTPS', ); } return url.toString().replace(/\/$/, ''); } function message(error: unknown): string { return error instanceof Error ? error.message : 'Matrix transport request failed'; }