Julien Gautier
428d19f60f
CI / commits (pull_request) Successful in 3m34s
CI / scan (pull_request) Successful in 3m47s
CI / check (pull_request) Successful in 6m15s
CI / a11y (pull_request) Successful in 2m26s
Docs site / build (pull_request) Successful in 1m7s
CI / perf (pull_request) Successful in 4m33s
Lands the BFF-side skeleton for the apf-ai-service relay per ADR-0024, step 2 of the chantier (skeleton + tests against an in-process fake gRPC server; no live HTTP endpoint yet — that ships in the SSE-bridge PR). What ships: - contract/proto vendoring at apps/portal-bff/src/grpc/proto/apf-ai/ (common, chat, rag, ingestion, models). pnpm run grpc:sync refreshes from the sibling apf-ai-service tree; both .proto and the ts-proto TypeScript stubs under grpc/gen/ are committed for hermetic builds and reviewable diffs. - Codegen via pnpm run grpc:codegen using grpc-tools' bundled protoc and ts-proto (outputServices=grpc-js, esModuleInterop, forceLong, useOptionals=messages). Generated tree is excluded from Prettier and ESLint so hand-rules do not chase codegen output. - assertAiServiceConfig() pre-flight validator alongside the other config validators (AI_SERVICE_GRPC_ENDPOINT, AI_SERVICE_CLIENT_ID, AI_SERVICE_GRPC_TLS). - AiClientModule provides ChatClient / RagClient / IngestionClient / ModelsClient wrappers, a GrpcMetadataBuilder that stamps every call with x-client-id + x-correlation-id (W3C trace-id from OTel when a span is active, explicit override or UUID fallback otherwise), and a PrincipalMapper that hashes the Entra oid via HashUserIdService so the proto Principal.subject matches audit.events.actor_id_hash exactly (ADR-0013 cross-reference contract). - Specs cover: env validator, principal mapper, metadata builder (OTel span / override / fallback paths), chat client streaming + cancellation against an in-process fake gRPC ChatService, rag client unary happy path + error propagation, and module bootstrap. AiClientModule is NOT imported in AppModule yet — the SSE-bridge controller and its live route ship in the next PR.
70 lines
2.7 KiB
TypeScript
70 lines
2.7 KiB
TypeScript
import { Inject, Injectable } from '@nestjs/common';
|
|
import type { ClientReadableStream } from '@grpc/grpc-js';
|
|
import type { ChatEvent, ChatRequest, ChatServiceClient } from '../gen/apf-ai/chat';
|
|
import { GrpcMetadataBuilder } from './grpc-metadata.builder';
|
|
import { AI_CHAT_GRPC_CLIENT } from './tokens';
|
|
|
|
/**
|
|
* Wrapper around the generated `ChatServiceClient` for the
|
|
* server-streaming `Chat` RPC, per ADR-0024 §"Sub-decision 1".
|
|
*
|
|
* The wrapper does three things on top of the raw gRPC stub:
|
|
*
|
|
* 1. Injects the `x-client-id` + `x-correlation-id` metadata via
|
|
* `GrpcMetadataBuilder` so every call carries the contract
|
|
* headers from `apf-ai-service/docs/contract.md`.
|
|
* 2. Wires an optional `AbortSignal` to `call.cancel()` so the SSE
|
|
* bridge (next PR in the chantier) can propagate browser
|
|
* disconnects up to the AI service in one line.
|
|
* 3. Returns the raw `ClientReadableStream<ChatEvent>` — Node
|
|
* `Readable` streams are async-iterable by default, so the SSE
|
|
* bridge consumes the stream with `for await ... of` and no
|
|
* intermediate adapter.
|
|
*
|
|
* Cancellation flows in both directions: a `.cancel()` from inside
|
|
* the BFF (timeout, abort) reaches the AI service as a
|
|
* `grpc.status.CANCELLED`, which translates upstream to
|
|
* `ServerCallContext.CancellationToken` stopping LLM generation
|
|
* (per `apf-ai-service/docs/streaming.md`).
|
|
*/
|
|
@Injectable()
|
|
export class ChatClient {
|
|
constructor(
|
|
@Inject(AI_CHAT_GRPC_CLIENT) private readonly grpc: ChatServiceClient,
|
|
private readonly metadata: GrpcMetadataBuilder,
|
|
) {}
|
|
|
|
/**
|
|
* Start a streaming Chat call.
|
|
*
|
|
* The returned `ClientReadableStream<ChatEvent>` emits one event
|
|
* per AI service `ChatEvent` and ends after the terminal
|
|
* `ChatEvent.done` (or on cancellation / error).
|
|
*
|
|
* @param request The full proto request, already populated with
|
|
* the conversation history and a `Principal` (build the
|
|
* principal via `PrincipalMapper.fromInputs`).
|
|
* @param options.signal Optional AbortSignal. When it aborts, the
|
|
* underlying gRPC call is cancelled and the stream closes.
|
|
* @param options.correlationId Override the metadata
|
|
* correlation-id (defaults to the active OTel trace-id).
|
|
*/
|
|
chat(
|
|
request: ChatRequest,
|
|
options: { signal?: AbortSignal; correlationId?: string } = {},
|
|
): ClientReadableStream<ChatEvent> {
|
|
const metadata = this.metadata.build({ correlationId: options.correlationId });
|
|
const call = this.grpc.chat(request, metadata);
|
|
|
|
if (options.signal) {
|
|
if (options.signal.aborted) {
|
|
call.cancel();
|
|
} else {
|
|
options.signal.addEventListener('abort', () => call.cancel(), { once: true });
|
|
}
|
|
}
|
|
|
|
return call;
|
|
}
|
|
}
|