diff --git a/packages/agent-bff/src/action/action-routes-middleware.ts b/packages/agent-bff/src/action/action-routes-middleware.ts index f47c1adddb..67ea9896c4 100644 --- a/packages/agent-bff/src/action/action-routes-middleware.ts +++ b/packages/agent-bff/src/action/action-routes-middleware.ts @@ -4,6 +4,7 @@ import type { AgentActionClient, AgentActionClientOptions, } from './agent-action-client'; +import type { ActivityLogWriter } from '../activity-log/activity-log-writer'; import type { AgentTransport } from '../agent/agent-transport'; import type { Logger } from '../ports/logger-port'; import type ReadModelStore from '../read-model/read-model-store'; @@ -28,7 +29,9 @@ import { requireAgentToken, resolveReadModel, } from '../http/agent-route-helpers'; +import { BffHttpError } from '../http/bff-http-error'; import { + ACTION_REQUIRES_APPROVAL_TYPE, actionError, actionRequiresApproval, invalidRequest, @@ -37,6 +40,18 @@ import { const ACTION_ROUTE = /^\/agent\/v1\/([^/]+)\/actions\/([^/]+)\/(form|execute)$/; +const EXECUTE_VERB = 'execute'; + +/** + * An approval request is a business outcome, not a failure: the action was routed for review. The + * BFF answers it with a 403, but recording the entry as `failed` would make the same event count + * differently here and in mcp-server, which records it as a success — and action-failure statistics + * would mix refusals with runs that never happened. + */ +function isApprovalRequest(error: unknown): boolean { + return error instanceof BffHttpError && error.type === ACTION_REQUIRES_APPROVAL_TYPE; +} + interface ActionRequestBody { recordIds?: unknown; values?: unknown; @@ -94,6 +109,7 @@ export interface ActionRoutesMiddlewareOptions { store: ReadModelStore; transport: AgentTransport; logger: Logger; + activityLogs: ActivityLogWriter; createClient?: (options: AgentActionClientOptions) => AgentActionClient; } @@ -190,6 +206,7 @@ export default function createActionRoutesMiddleware({ store, transport, logger, + activityLogs, createClient = defaultCreateAgentActionClient, }: ActionRoutesMiddlewareOptions): Middleware { return async function actionRoutesMiddleware(ctx, next) { @@ -209,12 +226,16 @@ export default function createActionRoutesMiddleware({ // The read-model's action map IS the allow-list, so an absent action cannot be told from a // known-but-disallowed one — every non-exposed action maps to 404 here; `action_not_allowed` - // (403) has no local trigger, mirroring `collection_not_allowed`/`relation_not_allowed`. The - // URL identity is resolved before the body, so a bad action 404s before its payload is read. + // (403) has no local trigger, mirroring `collection_not_allowed`/`relation_not_allowed`. // TODO(PRD-673): distinguish disallowed from unknown when a separate exposure source exists. - if (!readModel.isActionAllowed(collection, actionName)) { + const allowed = readModel.isActionAllowed(collection, actionName); + + const refuseUnknownAction = () => { throw unknownAction(`Unknown action: ${collection}.${actionName}`); - } + }; + + // The form is unaudited, so it keeps refusing before the body is read. + if (!allowed && verb !== EXECUTE_VERB) refuseUnknownAction(); const body = (ctx.request.body ?? {}) as ActionRequestBody; assertKnownBodyKeys(body as Record); @@ -227,23 +248,49 @@ export default function createActionRoutesMiddleware({ actionEndpoints: readModel.getActionEndpoints(), }); - const action = await callAgent( - () => - client.loadAction({ - collection, - actionName, - recordIds, - timezone: ctx.state.timezone as string, - }), - logger, - ); - - const handlerArgs = { ctx, action, values, logger }; - - if (verb === 'execute') { - await handleExecute(handlerArgs); - } else { - await handleForm(handlerArgs); + const loadAction = () => + callAgent( + () => + client.loadAction({ + collection, + actionName, + recordIds, + timezone: ctx.state.timezone as string, + }), + logger, + ); + + // The form is not audited, mirroring mcp-server, whose get-action-form tool writes no log + // either: the record-touching event the trail records is the execution. + if (verb !== EXECUTE_VERB) { + const action = await loadAction(); + + await handleForm({ ctx, action, values, logger }); + + return; } + + // The whole sequence is audited, loadAction and setFields included, so the intent is recorded + // even when the attempt never reaches the agent's execute. The allow-list refusal is inside + // too, mirroring mcp-server, whose execute-action tool resolves the action within its own + // wrapper: an attempt on an action the caller may not trigger is exactly what the trail is + // for, and the record ids it names are only known once the body is read. + await activityLogs.record({ + ctx, + action: 'action', + context: { + collectionName: collection, + recordIds, + label: `triggered the action "${actionName}"`, + }, + isCompletedDespite: isApprovalRequest, + operation: async () => { + if (!allowed) refuseUnknownAction(); + + const action = await loadAction(); + + await handleExecute({ ctx, action, values, logger }); + }, + }); }; } diff --git a/packages/agent-bff/src/activity-log/activity-log-drainer.ts b/packages/agent-bff/src/activity-log/activity-log-drainer.ts new file mode 100644 index 0000000000..6112297d49 --- /dev/null +++ b/packages/agent-bff/src/activity-log/activity-log-drainer.ts @@ -0,0 +1,73 @@ +interface InFlightOperation { + promise: Promise; + /** What the drain names when a deadline leaves this one unfinished. Carries no record payload. */ + description: string; +} + +/** + * Holds the audited requests and the status transitions they fire without `await`. Nothing else + * keeps the transitions alive: `server.close()` waits for connections, and one sent after the + * response is attached to none — without this, every deploy would leave entries stuck in `pending`. + * + * The requests are tracked too, and not only their transitions, for the embedded deployment: there + * the host owns the connections, so `stop()` returns while requests are still running and their + * transitions are not registered yet. + */ +export default class ActivityLogDrainer { + private readonly inFlight = new Set(); + + track(operation: () => Promise, description: string): Promise { + const promise = operation(); + const entry: InFlightOperation = { promise, description }; + this.inFlight.add(entry); + promise.finally(() => this.inFlight.delete(entry)).catch(() => {}); + + return promise; + } + + /** + * Loops rather than settling one snapshot: a transition is registered only once the request it + * audits has finished, so a single pass would return before the work that outlives it. + * + * `timeoutMs` is the shutdown deadline the caller shares: a stalled audit store would otherwise + * hold the process past the grace its orchestrator gives it, and be SIGKILLed mid-drain. Returns + * what the deadline left unfinished, empty when everything settled. + */ + async drain(timeoutMs?: number): Promise { + const deadline = timeoutMs === undefined ? undefined : Date.now() + timeoutMs; + + while (this.inFlight.size > 0) { + const remainingMs = deadline === undefined ? undefined : deadline - Date.now(); + + if (remainingMs !== undefined && remainingMs <= 0) break; + + // eslint-disable-next-line no-await-in-loop + await this.settle(remainingMs); + } + + return [...this.inFlight].map(entry => entry.description); + } + + private async settle(timeoutMs?: number): Promise { + const settled = Promise.allSettled([...this.inFlight].map(entry => entry.promise)); + + if (timeoutMs === undefined) { + await settled; + + return; + } + + let timer: NodeJS.Timeout | undefined; + + try { + await Promise.race([ + settled, + new Promise(resolve => { + timer = setTimeout(resolve, timeoutMs); + }), + ]); + } finally { + clearTimeout(timer); + } + } +} diff --git a/packages/agent-bff/src/activity-log/activity-log-writer.ts b/packages/agent-bff/src/activity-log/activity-log-writer.ts new file mode 100644 index 0000000000..02b3a781c3 --- /dev/null +++ b/packages/agent-bff/src/activity-log/activity-log-writer.ts @@ -0,0 +1,54 @@ +import type { ActivityLogContext, BffActivityLogAction } from './activity-logs-creator'; +import type { ActivityLogsWriter } from './activity-logs-service'; +import type { Logger } from '../ports/logger-port'; +import type { Context } from 'koa'; + +import ActivityLogDrainer from './activity-log-drainer'; +import withActivityLog from './with-activity-log'; + +export interface RecordActivityLogOptions { + ctx: Context; + action: BffActivityLogAction; + context?: ActivityLogContext; + operation: () => Promise; + isCompletedDespite?: (error: unknown) => boolean; +} + +export interface ActivityLogWriter { + record(options: RecordActivityLogOptions): Promise; + /** + * Waits for the audited requests still running and for the status transitions no connection + * holds. Called when the server stops, which shares its deadline through `timeoutMs`; returns + * the operations that deadline left unfinished. + */ + drain(timeoutMs?: number): Promise; +} + +export interface ActivityLogWriterOptions { + service: ActivityLogsWriter; + logger: Logger; +} + +function describeRequest(action: BffActivityLogAction, collectionName?: string): string { + return collectionName ? `'${action}' request on '${collectionName}'` : `'${action}' request`; +} + +export default function createActivityLogWriter({ + service, + logger, +}: ActivityLogWriterOptions): ActivityLogWriter { + const drainer = new ActivityLogDrainer(); + + return { + record(options: RecordActivityLogOptions): Promise { + return drainer.track( + () => withActivityLog({ ...options, service, drainer, logger }), + describeRequest(options.action, options.context?.collectionName), + ); + }, + + drain(timeoutMs?: number): Promise { + return drainer.drain(timeoutMs); + }, + }; +} diff --git a/packages/agent-bff/src/activity-log/activity-logs-creator.ts b/packages/agent-bff/src/activity-log/activity-logs-creator.ts new file mode 100644 index 0000000000..8a94e598d9 --- /dev/null +++ b/packages/agent-bff/src/activity-log/activity-logs-creator.ts @@ -0,0 +1,341 @@ +import type ActivityLogDrainer from './activity-log-drainer'; +import type { ActivityLogsWriter } from './activity-logs-service'; +import type { Logger } from '../ports/logger-port'; +import type { + ActivityLogAction, + ActivityLogResponse, + ActivityLogType, +} from '@forestadmin/forestadmin-client'; +import type { Context } from 'koa'; + +import { HttpError } from '@forestadmin/forestadmin-client'; + +import { invalidateApiKeyIdentity } from '../api-key/api-key-middleware'; +import { + resolveForestServerToken, + resolveRenderingId, +} from '../auth/forest-server-token-middleware'; +import { sessionExpired } from '../http/bff-http-error'; +import { + AUDIT_RETRY_AFTER_SECONDS, + auditNotAuthorized, + auditUnavailable, + isUnretryableAuditFailure, +} from '../http/bff-local-errors'; + +/** The actions the BFF writes: its data routes read, and its action route writes. */ +export type BffActivityLogAction = Extract< + ActivityLogAction, + 'index' | 'search' | 'filter' | 'listRelatedData' | 'action' +>; + +/** + * Fail policy for the audit trail, keyed by action type: a write whose activity log cannot be + * created is blocked (no unaudited side effect), while a read proceeds with a warning (an audit + * store outage must not take down the read surface). + * + * One case is arbitrated by the cause instead of the action type: an authorization refusal (403) + * propagates for reads too — the read itself is not authorized either. + */ +const ACTION_TO_TYPE: Record = { + index: 'read', + search: 'read', + filter: 'read', + listRelatedData: 'read', + action: 'write', +}; + +const MAX_STATUS_ATTEMPTS = 5; +const STATUS_RETRY_DELAY_MS = 500; + +const NO_RENDERING_MESSAGE = 'This request carries no usable rendering'; + +export interface ActivityLogContext { + collectionName?: string; + recordId?: string | number; + recordIds?: string[] | number[]; + label?: string; +} + +/** + * The token that created the log is kept for the status transition: the transition is fired after + * the response, when the session it came from may already be unreachable. + */ +export interface PendingActivityLog { + activityLog: ActivityLogResponse; + forestServerToken: string; +} + +export interface CreatePendingActivityLogOptions { + ctx: Context; + service: ActivityLogsWriter; + action: BffActivityLogAction; + context?: ActivityLogContext; + logger: Logger; +} + +interface AuditCredentials { + forestServerToken: string; + renderingId: string; +} + +function describeCause(error: unknown): string { + return error instanceof Error ? `${error.name}: ${error.message}` : String(error); +} + +const FORBIDDEN = 403; +const UNAUTHORIZED = 401; +const SERVER_ERROR = 500; +/** Not found: the document may not be visible yet. 0: unreachable. 408: timeout. 429: throttled. */ +const RETRYABLE_TRANSITION_STATUSES = new Set([0, 404, 408, 429]); +const AUDIT_ENDPOINT_ABSENT_STATUSES = new Set([404, 501]); + +const NO_AUDIT_ENDPOINT_MESSAGE = + 'The Forest server does not expose the endpoint the activity log is written through, so the ' + + 'operation was not performed'; + +/** + * A 403 only. A 401 is not the caller being refused: the bearer the BFF audits with is minted by + * the Forest server and cached, so a 401 says that token expired — answering the caller with + * `audit_not_authorized` would refuse a read the fail-open policy lets through. + */ +function isAuthorizationRefusal(error: unknown): boolean { + return error instanceof HttpError && error.status === FORBIDDEN; +} + +function isExpiredAuditCredential(error: unknown): boolean { + return error instanceof HttpError && error.status === UNAUTHORIZED; +} + +/** + * A Forest server that does not serve the activity-log endpoint at all, rather than one failing to + * answer it. No retry can make the route appear, so the caller must not be handed a `Retry-After` + * it would keep honouring on every write. + */ +function isAuditEndpointAbsent(error: unknown): boolean { + return error instanceof HttpError && AUDIT_ENDPOINT_ABSENT_STATUSES.has(error.status); +} + +/** + * What locates the failure for support, and nothing else: the credential, the record ids and the + * label are the payload this must never carry. + */ +function auditIdentifiers( + ctx: Context, + context?: ActivityLogContext, +): Record { + const renderingId = resolveRenderingId(ctx); + + return { + ...(renderingId === undefined ? {} : { renderingId }), + ...(context?.collectionName === undefined ? {} : { collectionName: context.collectionName }), + }; +} + +/** An empty string is an answer, not a value: it locates no document, so it fails the guard. */ +function isPresent(value: string | undefined | null): boolean { + return value !== null && value !== undefined && value !== ''; +} + +/** + * The status transition reads both the id and the index, so an answer carrying only an id strands + * the entry `pending`: the fail-closed policy has to engage here, not asynchronously afterwards. + */ +function isTransitionable(activityLog: ActivityLogResponse): boolean { + return isPresent(activityLog?.id) && isPresent(activityLog?.attributes?.index); +} + +interface UnresolvedCredentialsReport { + ctx: Context; + action: BffActivityLogAction; + context?: ActivityLogContext; + logger: Logger; + error: unknown; +} + +/** + * A credential this deployment never mints is a degradation the Forest server declares by sending + * none (`api-key/api-key-client.ts`), so it warns; only a resolution that actually failed is an + * error. The single report for either: `with-activity-log` states nothing of its own, which used to + * double every line of the read path. + */ +function reportUnresolvedCredentials({ + ctx, + action, + context, + logger, + error, +}: UnresolvedCredentialsReport): void { + const identifiers = { ...auditIdentifiers(ctx, context), cause: describeCause(error) }; + + if (isUnretryableAuditFailure(error)) { + logger( + 'Warn', + `Activity log for '${action}' was not created: this deployment has no credential to write ` + + 'it with', + identifiers, + ); + + return; + } + + logger( + 'Error', + `Activity log for '${action}' has no credentials to be created with`, + identifiers, + ); +} + +async function resolveCredentials(ctx: Context): Promise { + const renderingId = resolveRenderingId(ctx); + + if (renderingId === undefined) throw sessionExpired(NO_RENDERING_MESSAGE); + + return { + forestServerToken: await resolveForestServerToken(ctx), + renderingId: String(renderingId), + }; +} + +export default async function createPendingActivityLog({ + ctx, + service, + action, + context, + logger, +}: CreatePendingActivityLogOptions): Promise { + const type = ACTION_TO_TYPE[action]; + + let credentials: AuditCredentials; + + try { + credentials = await resolveCredentials(ctx); + } catch (error) { + reportUnresolvedCredentials({ ctx, action, context, logger, error }); + + if (type === 'write') throw error; + + return null; + } + + const { forestServerToken, renderingId } = credentials; + + let activityLog: ActivityLogResponse; + + try { + activityLog = await service.createMcpActivityLog({ + forestServerToken, + renderingId, + action, + type, + collectionName: context?.collectionName, + recordId: context?.recordId, + recordIds: context?.recordIds, + label: context?.label, + }); + } catch (error) { + logger('Error', `Activity log for '${action}' could not be created`, { + ...auditIdentifiers(ctx, context), + cause: describeCause(error), + }); + + if (isAuthorizationRefusal(error)) throw auditNotAuthorized(); + if (isExpiredAuditCredential(error)) invalidateApiKeyIdentity(ctx); + + if (type === 'write') { + throw isAuditEndpointAbsent(error) + ? auditUnavailable(undefined, NO_AUDIT_ENDPOINT_MESSAGE) + : auditUnavailable(AUDIT_RETRY_AFTER_SECONDS); + } + + return null; + } + + if (!isTransitionable(activityLog)) { + logger( + 'Error', + `Activity log for '${action}' could not be created: the server answered with no activity ` + + 'log id or index, so the audit store dropped the write', + auditIdentifiers(ctx, context), + ); + + if (type === 'write') throw auditUnavailable(AUDIT_RETRY_AFTER_SECONDS); + + return null; + } + + return { activityLog, forestServerToken }; +} + +export interface MarkActivityLogOptions { + service: ActivityLogsWriter; + drainer: ActivityLogDrainer; + pending: PendingActivityLog; + status: 'completed' | 'failed'; + logger: Logger; +} + +/** + * The transition is fired after the response, so nothing retries it downstream: an entry whose + * transition is dropped stays `pending` for good and skews the action-failure statistics. Retried + * are the failures a later attempt can land — the document not existing yet, a timeout, a throttle, + * an unreachable server (status 0) and any 5xx — plus a transport failure the client rethrows raw, + * which carries no status at all. A refusal or a malformed request is left to fail at once. + */ +function isRetryableTransitionFailure(error: unknown): boolean { + if (!(error instanceof HttpError)) return true; + + return error.status >= SERVER_ERROR || RETRYABLE_TRANSITION_STATUSES.has(error.status); +} + +async function updateStatus(options: MarkActivityLogOptions, attempt = 1): Promise { + const { service, pending, status, logger } = options; + + try { + await service.updateActivityLogStatus({ + forestServerToken: pending.forestServerToken, + activityLog: pending.activityLog, + status, + }); + } catch (error) { + if (isRetryableTransitionFailure(error) && attempt < MAX_STATUS_ATTEMPTS) { + logger('Debug', `Activity log status transition failed, retrying it`, { + attempt, + attempts: MAX_STATUS_ATTEMPTS, + cause: describeCause(error), + }); + + await new Promise(resolve => { + // Unreferenced: a pending retry must not outlive the shutdown grace the drainer enforces. + setTimeout(resolve, STATUS_RETRY_DELAY_MS).unref(); + }); + + await updateStatus(options, attempt + 1); + + return; + } + + throw error; + } +} + +/** + * Fire-and-forget on purpose: the caller's response must not wait for the audit store. The drainer + * holds the promise so a shutdown can wait for it instead. + */ +export function markActivityLog(options: MarkActivityLogOptions): void { + const { drainer, status, logger } = options; + + drainer + .track( + () => updateStatus(options), + `'${status}' transition of the activity log ${options.pending.activityLog.id}`, + ) + .catch(error => { + logger('Error', `Failed to mark the activity log as '${status}'`, { + activityLogId: options.pending.activityLog.id, + index: options.pending.activityLog.attributes?.index, + cause: describeCause(error), + }); + }); +} diff --git a/packages/agent-bff/src/activity-log/activity-logs-service.ts b/packages/agent-bff/src/activity-log/activity-logs-service.ts new file mode 100644 index 0000000000..a9feaabde1 --- /dev/null +++ b/packages/agent-bff/src/activity-log/activity-logs-service.ts @@ -0,0 +1,31 @@ +import type { + ActivityLogResponse, + CreateActivityLogParams, + UpdateActivityLogStatusParams, +} from '@forestadmin/forestadmin-client'; + +import { ActivityLogsService, ForestHttpApi } from '@forestadmin/forestadmin-client'; + +export const APPLICATION_SOURCE_HEADER = 'Forest-Application-Source'; +export const BFF_APPLICATION_SOURCE = 'BFF'; + +/** + * The slice of `ActivityLogsService` the BFF uses. Named so a fake can stand in for the two calls + * without carrying the rest of the Forest client. + */ +export interface ActivityLogsWriter { + createMcpActivityLog(params: CreateActivityLogParams): Promise; + updateActivityLogStatus(params: UpdateActivityLogStatusParams): Promise; +} + +/** + * Its own instance rather than the client `oauth/forest-server-client.ts` already holds: + * `ForestAdminClientOptions` carries no `headers`, so that one cannot tell the server which channel + * wrote the log. + */ +export default function createBffActivityLogsService(forestServerUrl: string): ActivityLogsWriter { + return new ActivityLogsService(new ForestHttpApi(), { + forestServerUrl, + headers: { [APPLICATION_SOURCE_HEADER]: BFF_APPLICATION_SOURCE }, + }); +} diff --git a/packages/agent-bff/src/activity-log/with-activity-log.ts b/packages/agent-bff/src/activity-log/with-activity-log.ts new file mode 100644 index 0000000000..e7b22e9050 --- /dev/null +++ b/packages/agent-bff/src/activity-log/with-activity-log.ts @@ -0,0 +1,52 @@ +import type ActivityLogDrainer from './activity-log-drainer'; +import type { ActivityLogContext, BffActivityLogAction } from './activity-logs-creator'; +import type { ActivityLogsWriter } from './activity-logs-service'; +import type { Logger } from '../ports/logger-port'; +import type { Context } from 'koa'; + +import createPendingActivityLog, { markActivityLog } from './activity-logs-creator'; + +const COMPLETED = 'completed'; +const FAILED = 'failed'; + +export interface WithActivityLogOptions { + ctx: Context; + service: ActivityLogsWriter; + drainer: ActivityLogDrainer; + action: BffActivityLogAction; + context?: ActivityLogContext; + logger: Logger; + operation: () => Promise; + /** + * Errors the log records as `completed` rather than `failed` — the operation reached a business + * outcome the BFF answers with an error status. + */ + isCompletedDespite?: (error: unknown) => boolean; +} + +/** + * Runs an operation under an activity log: the pending log is awaited before the operation starts, + * so nothing runs unaudited, and the status transition is fired without `await` afterwards. + */ +export default async function withActivityLog(options: WithActivityLogOptions): Promise { + const { ctx, service, drainer, action, context, logger, operation, isCompletedDespite } = options; + + // No log line of its own when nothing came back: the creator reports the case it hit, with the + // rendering and the collection this one could not name. A read proceeds unaudited from here. + const pending = await createPendingActivityLog({ ctx, service, action, context, logger }); + + try { + const result = await operation(); + + if (pending) markActivityLog({ service, drainer, pending, status: COMPLETED, logger }); + + return result; + } catch (error) { + if (pending) { + const status = isCompletedDespite?.(error) ? COMPLETED : FAILED; + markActivityLog({ service, drainer, pending, status, logger }); + } + + throw error; + } +} diff --git a/packages/agent-bff/src/api-key/api-key-authenticator.ts b/packages/agent-bff/src/api-key/api-key-authenticator.ts index a6c540de02..99d65bdd12 100644 --- a/packages/agent-bff/src/api-key/api-key-authenticator.ts +++ b/packages/agent-bff/src/api-key/api-key-authenticator.ts @@ -25,10 +25,21 @@ export interface ApiKeyAuthenticatorOptions { export interface AuthenticatedApiKey { agentToken: string; identity: ResolvedApiKeyIdentity; + /** The Forest server token the resolve response carried, cached with the identity. */ + forestServerToken?: string; } export interface ApiKeyAuthenticator { authenticate(rawKey: string): Promise; + /** + * Forgets what a key resolved to, so the next request resolves it against the Forest server + * again. The resolution carries a short-lived server token the BFF caches with the identity: + * once that token is refused, the whole entry has to go. + * + * `credential` is the refused token, which tells a repeat refusal of the same one from the + * refusal of the token the previous invalidation went and fetched. + */ + invalidate(rawKey: string, credential?: string): void; } function mapResolveError(error: ApiKeyResolveError): ApiKeyError { @@ -54,7 +65,11 @@ export default function createApiKeyAuthenticator({ authSecret, }: ApiKeyAuthenticatorOptions): ApiKeyAuthenticator { function mint(identity: ResolvedApiKeyIdentity): AuthenticatedApiKey { - return { agentToken: issueAgentToken({ identity, authSecret }), identity }; + return { + agentToken: issueAgentToken({ identity, authSecret }), + identity, + forestServerToken: identity.saasAccessToken, + }; } return { @@ -91,5 +106,13 @@ export default function createApiKeyAuthenticator({ return authenticated; }, + + invalidate(rawKey, credential) { + const parsed = parseApiKey(rawKey); + + if (!parsed) return; + + cache.invalidate(hashApiKey(parsed.keyId, parsed.secret), credential); + }, }; } diff --git a/packages/agent-bff/src/api-key/api-key-client.ts b/packages/agent-bff/src/api-key/api-key-client.ts index f6ef887c51..d58cc59d11 100644 --- a/packages/agent-bff/src/api-key/api-key-client.ts +++ b/packages/agent-bff/src/api-key/api-key-client.ts @@ -17,6 +17,12 @@ export interface ResolvedApiKeyIdentity { user: ApiKeyIdentityUser; renderingId: number; allowedOrigins: string[]; + /** + * Short-lived, user-scoped Forest server token, used to write the activity log. Optional so a + * Forest server that does not send one yet still resolves keys: the audit trail then degrades on + * its own terms (a read proceeds unaudited, a write is blocked) instead of taking auth down. + */ + saasAccessToken?: string; } export interface ApiKeyClientOptions { @@ -90,12 +96,18 @@ export default class ApiKeyClient { private static isResolvedIdentity(body: unknown): body is ResolvedApiKeyIdentity { if (typeof body !== 'object' || body === null) return false; - const candidate = body as { user?: unknown; renderingId?: unknown; allowedOrigins?: unknown }; + const candidate = body as { + user?: unknown; + renderingId?: unknown; + allowedOrigins?: unknown; + saasAccessToken?: unknown; + }; return ( typeof candidate.renderingId === 'number' && Array.isArray(candidate.allowedOrigins) && candidate.allowedOrigins.every(entry => typeof entry === 'string') && + (candidate.saasAccessToken === undefined || typeof candidate.saasAccessToken === 'string') && ApiKeyClient.isIdentityUser(candidate.user) ); } diff --git a/packages/agent-bff/src/api-key/api-key-middleware.ts b/packages/agent-bff/src/api-key/api-key-middleware.ts index fdc60da445..4419fed441 100644 --- a/packages/agent-bff/src/api-key/api-key-middleware.ts +++ b/packages/agent-bff/src/api-key/api-key-middleware.ts @@ -1,12 +1,14 @@ import type { ApiKeyAuthenticator, AuthenticatedApiKey } from './api-key-authenticator'; import type { Logger } from '../ports/logger-port'; -import type { Middleware } from 'koa'; +import type { Context, Middleware } from 'koa'; import { fingerprintApiKey } from './api-key'; import { ApiKeyError } from './api-key-error'; export const BFF_KEY_HEADER = 'X-Forest-Bff-Key'; +export type ApiKeyIdentityInvalidator = () => void; + export interface ApiKeyMiddlewareOptions { authenticator: ApiKeyAuthenticator; logger: Logger; @@ -48,8 +50,15 @@ export default function createApiKeyMiddleware({ throw error; } + // Bound to the token this request was authenticated with: the cache uses it to tell a repeat + // refusal of the same credential from the refusal of the one it just went and fetched. + const invalidateIdentity: ApiKeyIdentityInvalidator = () => + authenticator.invalidate(rawKey, authenticated.forestServerToken); + + ctx.state.invalidateApiKeyIdentity = invalidateIdentity; ctx.state.agentToken = authenticated.agentToken; ctx.state.apiKeyIdentity = authenticated.identity; + ctx.state.forestServerToken = authenticated.forestServerToken; ctx.set('Cache-Control', 'no-store'); logger('Info', 'Resolved BFF API key', { keyHash: fingerprintApiKey(rawKey), @@ -59,3 +68,16 @@ export default function createApiKeyMiddleware({ await next(); }; } + +/** + * Forgets the identity this request was authenticated with. Called when the Forest server refuses + * the token that came with it: the token is cached with the identity, so the next request must + * resolve the key again instead of replaying the refused one for the rest of the cache window. + * + * A no-op outside api-key mode — nothing else lands an invalidator. + */ +export function invalidateApiKeyIdentity(ctx: Context): void { + const invalidate = ctx.state.invalidateApiKeyIdentity as ApiKeyIdentityInvalidator | undefined; + + invalidate?.(); +} diff --git a/packages/agent-bff/src/api-key/resolve-cache.ts b/packages/agent-bff/src/api-key/resolve-cache.ts index 6c891ec6a2..f32f54e62b 100644 --- a/packages/agent-bff/src/api-key/resolve-cache.ts +++ b/packages/agent-bff/src/api-key/resolve-cache.ts @@ -6,6 +6,18 @@ export interface ResolveCache { getNegative(hash: string): ApiKeyError | undefined; setPositive(hash: string, identity: ResolvedApiKeyIdentity): void; setNegative(hash: string, error: ApiKeyError): void; + /** + * Forgets a key. Bounded because the caller is a refusal the Forest server may repeat on every + * request: invalidating each time would defeat the cache and cost two round trips per request + * instead of one extra per window. + * + * `credential` is the server token that was refused. A window opened by one token still lets a + * second, different one through: the re-resolution that followed the first refusal caches a + * fresh token, and suppressing its refusal too would replay a credential the server rejects for + * the rest of the window. Only the second is allowed, so the bound holds whatever the server + * hands back. + */ + invalidate(hash: string, credential?: string): void; size(): number; } @@ -30,6 +42,14 @@ interface NegativeEntry { type CacheEntry = PositiveEntry | NegativeEntry; +interface InvalidationWindow { + until: number; + /** The server token whose refusal opened or reset the window. */ + credential?: string; + /** Whether a second, different credential has already reset it. */ + retried: boolean; +} + const DEFAULT_POSITIVE_TTL_SECONDS = 60; const DEFAULT_NEGATIVE_TTL_SECONDS = 10; const DEFAULT_MAX_ENTRIES = 10_000; @@ -41,6 +61,11 @@ export default function createResolveCache({ maxEntries = DEFAULT_MAX_ENTRIES, }: ResolveCacheOptions): ResolveCache { const entries = new Map(); + /** + * Per key, the window opened by its last invalidation. Bounded by `maxEntries` like the entries + * it guards: invalidations of distinct keys would otherwise grow it without limit. + */ + const invalidatedUntil = new Map(); function purgeExpired(): void { const current = now(); @@ -48,6 +73,10 @@ export default function createResolveCache({ for (const [hash, entry] of entries) { if (current >= entry.expiresAt) entries.delete(hash); } + + for (const [hash, window] of invalidatedUntil) { + if (current >= window.until) invalidatedUntil.delete(hash); + } } function liveEntry(hash: string): CacheEntry | undefined { @@ -63,13 +92,16 @@ export default function createResolveCache({ return entry; } + function evictOldestIfFull(map: Map, hash: string): void { + if (map.has(hash) || map.size < maxEntries) return; + + const oldest = map.keys().next().value; + if (oldest !== undefined) map.delete(oldest); + } + function store(hash: string, entry: CacheEntry): void { purgeExpired(); - - if (!entries.has(hash) && entries.size >= maxEntries) { - const oldest = entries.keys().next().value; - if (oldest !== undefined) entries.delete(oldest); - } + evictOldestIfFull(entries, hash); entries.set(hash, entry); } @@ -95,6 +127,30 @@ export default function createResolveCache({ store(hash, { kind: 'negative', error, expiresAt: now() + negativeTtlSeconds * 1000 }); }, + invalidate(hash, credential) { + const open = invalidatedUntil.get(hash); + const live = open !== undefined && now() < open.until; + + if (live) { + if (open.retried || open.credential === credential) return; + + // The window is kept: a second refusal buys one more resolution, not a later deadline. + invalidatedUntil.set(hash, { ...open, credential, retried: true }); + entries.delete(hash); + + return; + } + + purgeExpired(); + evictOldestIfFull(invalidatedUntil, hash); + invalidatedUntil.set(hash, { + until: now() + positiveTtlSeconds * 1000, + credential, + retried: false, + }); + entries.delete(hash); + }, + size() { purgeExpired(); diff --git a/packages/agent-bff/src/auth/auth-mode.ts b/packages/agent-bff/src/auth/auth-mode.ts index 5752427310..a118763eb2 100644 --- a/packages/agent-bff/src/auth/auth-mode.ts +++ b/packages/agent-bff/src/auth/auth-mode.ts @@ -7,12 +7,20 @@ export type AuthMode = 'oauth' | 'api-key'; const BEARER_PATTERN = /^Bearer[ \t]+(.+)$/i; const POSITIVE_INTEGER = /^[1-9]\d*$/; +export function readRenderingId(principal: BffAccessTokenPayload): number | undefined { + if (!POSITIVE_INTEGER.test(String(principal.rendering_id))) return undefined; + + return Number(principal.rendering_id); +} + export function requireRenderingId(principal: BffAccessTokenPayload): number { - if (!POSITIVE_INTEGER.test(String(principal.rendering_id))) { + const renderingId = readRenderingId(principal); + + if (renderingId === undefined) { throw unauthorized('The session carries no usable rendering'); } - return Number(principal.rendering_id); + return renderingId; } export function extractBearerToken(authorization: string | undefined): string | undefined { diff --git a/packages/agent-bff/src/auth/forest-server-token-middleware.ts b/packages/agent-bff/src/auth/forest-server-token-middleware.ts new file mode 100644 index 0000000000..70ed75ef97 --- /dev/null +++ b/packages/agent-bff/src/auth/forest-server-token-middleware.ts @@ -0,0 +1,124 @@ +import type { ResolvedApiKeyIdentity } from '../api-key/api-key-client'; +import type { BffAccessTokenPayload } from '../oauth/bff-token'; +import type ForestServerClient from '../oauth/forest-server-client'; +import type { SessionStore } from '../oauth/session-store'; +import type { Logger } from '../ports/logger-port'; +import type { Context, Middleware } from 'koa'; + +import { readRenderingId } from './auth-mode'; +import { extractErrorMessage } from '../errors'; +import { sessionExpired } from '../http/bff-http-error'; +import { AUDIT_RETRY_AFTER_SECONDS, auditUnavailable } from '../http/bff-local-errors'; +import { OAuthRequestError } from '../oauth/oauth-error'; +import ensureFreshServerAccess from '../oauth/session-lifecycle'; + +export type ForestServerTokenResolver = () => Promise; + +export interface OAuthSessionAccess { + store: SessionStore; + serverClient: ForestServerClient; +} + +export interface ForestServerTokenMiddlewareOptions { + session?: OAuthSessionAccess; + logger: Logger; +} + +const NO_SESSION_MESSAGE = 'The session behind this request could not be resolved'; +const NO_RESOLVER_MESSAGE = 'This request carries no Forest server credentials'; +/** + * Carries no `Retry-After`: the key resolution came back without an audit credential at all — a + * Forest server that does not mint one yet — so no retry can succeed until that server ships it. + */ +const NO_AUDIT_CREDENTIAL_MESSAGE = + 'The Forest server does not provide the credential the activity log is written with, so the ' + + 'operation was not performed'; +const UNAUTHORIZED = 401; + +async function resolveToken( + ctx: Context, + logger: Logger, + session?: OAuthSessionAccess, +): Promise { + if (ctx.state.authMode === 'api-key') { + const token = ctx.state.forestServerToken as string | undefined; + + if (!token) throw auditUnavailable(undefined, NO_AUDIT_CREDENTIAL_MESSAGE); + + return token; + } + + const principal = ctx.state.principal as BffAccessTokenPayload | undefined; + + if (!principal || !session) throw sessionExpired(NO_SESSION_MESSAGE); + + try { + return await ensureFreshServerAccess({ + sid: principal.sid, + store: session.store, + serverClient: session.serverClient, + }); + } catch (error) { + // The errors below carry neither the cause nor a `cause` field, so this line is the only place + // the operator ever sees what actually failed — a broken session store reads as an audit + // outage otherwise. + logger('Error', 'Could not resolve the Forest server access of this session', { + renderingId: readRenderingId(principal), + cause: extractErrorMessage(error), + }); + + // Only a session the Forest server rejected, or one that vanished, makes re-authenticating the + // answer. Everything else — the server being unreachable, above all — is retryable, and a 401 + // would log every user out over a blip instead of failing the audit write alone. + if (error instanceof OAuthRequestError && error.status === UNAUTHORIZED) { + throw sessionExpired(NO_SESSION_MESSAGE); + } + + throw auditUnavailable(AUDIT_RETRY_AFTER_SECONDS); + } +} + +/** + * Lands a lazy resolver of the Forest server bearer on the context, for both auth modes. Lazy on + * purpose: the routes that audit nothing — /health, the permissions, context, OpenAPI and docs + * routes — must not pay a session lookup, and the permissions one is hit on every page load. + * + * Keeping both modes here is what lets the data and action routes read one function off the context + * instead of taking the session store and the Forest server client as dependencies. + */ +export default function createForestServerTokenMiddleware({ + session, + logger, +}: ForestServerTokenMiddlewareOptions): Middleware { + return async function forestServerTokenMiddleware(ctx, next) { + let pending: Promise | undefined; + + const resolver: ForestServerTokenResolver = () => { + pending ??= resolveToken(ctx, logger, session); + + return pending; + }; + + ctx.state.resolveForestServerToken = resolver; + + await next(); + }; +} + +export function resolveForestServerToken(ctx: Context): Promise { + const resolver = ctx.state.resolveForestServerToken as ForestServerTokenResolver | undefined; + + if (!resolver) throw sessionExpired(NO_RESOLVER_MESSAGE); + + return resolver(); +} + +export function resolveRenderingId(ctx: Context): number | undefined { + if (ctx.state.authMode === 'api-key') { + return (ctx.state.apiKeyIdentity as ResolvedApiKeyIdentity | undefined)?.renderingId; + } + + const principal = ctx.state.principal as BffAccessTokenPayload | undefined; + + return principal ? readRenderingId(principal) : undefined; +} diff --git a/packages/agent-bff/src/build-bff.ts b/packages/agent-bff/src/build-bff.ts index 81ebbe9282..05c87ebca3 100644 --- a/packages/agent-bff/src/build-bff.ts +++ b/packages/agent-bff/src/build-bff.ts @@ -1,3 +1,4 @@ +import type { ActivityLogWriter } from './activity-log/activity-log-writer'; import type { AgentTransport } from './agent/agent-transport'; import type { AgentDispatcher } from './agent/in-process-transport'; import type { BFFConfig } from './config/env-config'; @@ -14,6 +15,8 @@ import { bodyParser } from '@koa/bodyparser'; import Koa from 'koa'; import createActionRoutesMiddleware from './action/action-routes-middleware'; +import createActivityLogWriter from './activity-log/activity-log-writer'; +import createBffActivityLogsService from './activity-log/activity-logs-service'; import createConsoleLogger from './adapters/console-logger'; import createAgentStubMiddleware from './agent/agent-stub'; import { createHttpTransport } from './agent/agent-transport'; @@ -25,6 +28,7 @@ import ApiKeyClient from './api-key/api-key-client'; import createApiKeyMiddleware from './api-key/api-key-middleware'; import createResolveCache from './api-key/resolve-cache'; import createAuthModeMiddleware from './auth/auth-mode-middleware'; +import createForestServerTokenMiddleware from './auth/forest-server-token-middleware'; import normalizeBasePath from './base-path'; import warnMissingConfig from './config/missing-config-warning'; import createContextRoutesMiddleware from './context/context-routes-middleware'; @@ -87,6 +91,14 @@ export interface Bff { * restarting on a customization refresh — calls this instead of waiting out the 24h TTL. */ invalidate(): void; + /** + * Waits for the activity-log status transitions still in flight. They are fired without `await`, + * so nothing else holds them: a host that stops without calling this leaves entries `pending`. + * Absent when the deployment writes no activity log. + * + * `timeoutMs` is the host's shutdown deadline; the returned descriptions name what it cut short. + */ + drainActivityLogs?: (timeoutMs?: number) => Promise; } const SESSION_TTL_SECONDS = 24 * 60 * 60; @@ -346,20 +358,27 @@ export function resolveUnfoldSource(config: BFFConfig, logger: Logger): UnfoldSo ); } +// The routes that write an activity log come with the writer holding their pending transitions, so +// the host can drain it when it stops. +interface AgentRouteEdge { + middlewares: Middleware[]; + activityLogs?: ActivityLogWriter; +} + // The data middleware falls through to the action middleware on a non-data path. function buildAgentRouteMiddlewares( bundle: ReadModelBundle | undefined, transport: AgentTransport | undefined, logger: Logger, permissionsCache: PermissionsCache, -): Middleware[] { +): AgentRouteEdge { if (!bundle) { logger( 'Warn', 'Data, action and permissions endpoints disabled: FOREST_SERVER_URL, FOREST_ENV_SECRET or FOREST_AUTH_SECRET is missing', ); - return [createAgentStubMiddleware()]; + return { middlewares: [createAgentStubMiddleware()] }; } const { store, apiKeyConfig } = bundle; @@ -377,14 +396,22 @@ function buildAgentRouteMiddlewares( if (!transport) { logger('Warn', 'Data and action endpoints disabled: AGENT_URL is missing'); - return [permissionsMiddleware, createAgentStubMiddleware()]; + return { middlewares: [permissionsMiddleware, createAgentStubMiddleware()] }; } - return [ - permissionsMiddleware, - createDataRoutesMiddleware({ store, transport, logger }), - createActionRoutesMiddleware({ store, transport, logger }), - ]; + const activityLogs = createActivityLogWriter({ + service: createBffActivityLogsService(apiKeyConfig.forestServerUrl), + logger, + }); + + return { + middlewares: [ + permissionsMiddleware, + createDataRoutesMiddleware({ store, transport, logger, activityLogs }), + createActionRoutesMiddleware({ store, transport, logger, activityLogs }), + ], + activityLogs, + }; } function buildAiMiddlewares(config: BFFConfig, oauth: OAuthEdge, logger: Logger): Middleware[] { @@ -415,6 +442,7 @@ function buildAiMiddlewares(config: BFFConfig, oauth: OAuthEdge, logger: Logger) interface AgentEdge { middlewares: Middleware[]; invalidate(): void; + activityLogs?: ActivityLogWriter; } function buildAgentMiddlewares( @@ -440,10 +468,13 @@ function buildAgentMiddlewares( const bundle = resolveReadModelBundle(config, logger, metrics); const source = toUnfoldSource(bundle, transport, logger); const permissionsCache = new PermissionsCache(); + const routeEdge = buildAgentRouteMiddlewares(bundle, transport, logger, permissionsCache); const chain: Middleware[] = [ createAuthModeMiddleware({ authSecret: forestAuthSecret }), apiKeyStep, + // After both auth middlewares: the resolver it lands reads what they put on the context. + createForestServerTokenMiddleware({ session: oauth.session, logger }), createRateLimitMiddleware({ maxRequests: config.rateLimitMaxRequests, windowMs: config.rateLimitWindowMs, @@ -469,7 +500,7 @@ function buildAgentMiddlewares( : []), ...aiMiddlewares, createTimezoneMiddleware({ defaultTimezone }), - ...buildAgentRouteMiddlewares(bundle, transport, logger, permissionsCache), + ...routeEdge.middlewares, ]; return { @@ -481,6 +512,7 @@ function buildAgentMiddlewares( bundle?.store.invalidate(); permissionsCache.clear(); }, + activityLogs: routeEdge.activityLogs, }; } @@ -572,5 +604,11 @@ export default async function buildBff({ const app = new Koa(); for (const middleware of middlewares) app.use(middleware); - return { callback: app.callback(), invalidate: agentEdge.invalidate }; + const { activityLogs } = agentEdge; + + return { + callback: app.callback(), + invalidate: agentEdge.invalidate, + drainActivityLogs: activityLogs && (timeoutMs => activityLogs.drain(timeoutMs)), + }; } diff --git a/packages/agent-bff/src/cli-core.ts b/packages/agent-bff/src/cli-core.ts index c19f0bbf45..fcf149a352 100644 --- a/packages/agent-bff/src/cli-core.ts +++ b/packages/agent-bff/src/cli-core.ts @@ -6,16 +6,57 @@ import { parseConfig } from './config/env-config'; import { extractErrorMessage } from './errors'; import BFFHttpServer from './http/bff-http-server'; +const SHUTDOWN_SIGNALS: NodeJS.Signals[] = ['SIGTERM', 'SIGINT']; + +let installedShutdownHandlers: { signal: NodeJS.Signals; handler: () => void }[] = []; + +/** + * Routes a termination signal to `stop()`, which drains the activity-log transitions no connection + * holds. Registered here rather than in the server: an embedded deployment does not own the process + * signals, so the drain has to be reachable through `stop()` instead. + * + * A process runs one BFF, so a second call replaces the handlers instead of adding a pair: the + * signal must reach the server that is listening, and nothing else. + */ +export function installShutdownHandlers(server: BFFHttpServer, logger: Logger): void { + for (const { signal, handler } of installedShutdownHandlers) { + process.removeListener(signal, handler); + } + + installedShutdownHandlers = SHUTDOWN_SIGNALS.map(signal => { + const handler = () => { + logger('Info', 'Stopping the Forest BFF', { signal }); + + server.stop().catch(error => { + logger('Error', 'The Forest BFF did not stop cleanly', { + cause: extractErrorMessage(error), + }); + }); + }; + + process.on(signal, handler); + + return { signal, handler }; + }); +} + export default async function runCli( env: NodeJS.ProcessEnv, logger: Logger = createConsoleLogger(), ): Promise { const config = parseConfig(env); - const { callback } = await buildBff({ config, logger }); + const { callback, drainActivityLogs } = await buildBff({ config, logger }); - const server = new BFFHttpServer({ port: config.httpPort, config, logger, callback }); + const server = new BFFHttpServer({ + port: config.httpPort, + config, + logger, + callback, + drainActivityLogs, + }); await server.start(); + installShutdownHandlers(server, logger); return server; } diff --git a/packages/agent-bff/src/data/agent-query.ts b/packages/agent-bff/src/data/agent-query.ts index be008cdf5d..710afa2937 100644 --- a/packages/agent-bff/src/data/agent-query.ts +++ b/packages/agent-bff/src/data/agent-query.ts @@ -199,6 +199,24 @@ function applySearch( if (body.searchExtended !== undefined) query.searchExtended = body.searchExtended; } +/** + * Whether the request actually searches, by the same rule `applySearch` forwards on. The audit trail + * names the operation it audits, so it has to answer this question the way the outgoing query does: + * a whitespace-only search must not be recorded as a search the agent never performed. + */ +export function hasSearch(body: Pick): boolean { + return (body.search?.trim() ?? '') !== ''; +} + +/** + * Whether the request actually filters. An empty object is how an absent filter is spelled — see + * `assertFilterNode`, which accepts a node carrying no key — so the audit trail must not record a + * plain page load as a filtered read. + */ +export function hasFilter(body: Pick): boolean { + return isPlainObject(body.filter) && Object.keys(body.filter).length > 0; +} + export function buildListAgentQuery( collection: string, timezone: string, diff --git a/packages/agent-bff/src/data/data-routes-middleware.ts b/packages/agent-bff/src/data/data-routes-middleware.ts index 507355a872..10ef2e6fed 100644 --- a/packages/agent-bff/src/data/data-routes-middleware.ts +++ b/packages/agent-bff/src/data/data-routes-middleware.ts @@ -5,6 +5,8 @@ import type { RelationCountRequestBody, RelationListRequestBody, } from './agent-query'; +import type { ActivityLogWriter } from '../activity-log/activity-log-writer'; +import type { BffActivityLogAction } from '../activity-log/activity-logs-creator'; import type { AgentTransport } from '../agent/agent-transport'; import type { Logger } from '../ports/logger-port'; import type { CapabilitiesResult } from '../read-model/capabilities-cache'; @@ -19,6 +21,8 @@ import { buildListAgentQuery, collectCountFieldPaths, collectListFieldPaths, + hasFilter, + hasSearch, parseCountRequest, parseListRequest, parseRelationCountRequest, @@ -46,6 +50,7 @@ export interface DataRoutesMiddlewareOptions { store: ReadModelStore; transport: AgentTransport; logger: Logger; + activityLogs: ActivityLogWriter; createClient?: (options: AgentDataClientOptions) => AgentDataClient; } @@ -57,6 +62,7 @@ interface RequestHandlerDeps { token: string; timezone: string; logger: Logger; + activityLogs: ActivityLogWriter; } type ListHandlerDeps = RequestHandlerDeps & { primaryKeys: PrimaryKeyField[] }; @@ -139,7 +145,14 @@ async function resolveOwnCapabilities( return result; } -async function handleList(ctx: Context, body: ListRequestBody, deps: ListHandlerDeps) { +function selectListAction(body: ListRequestBody): BffActivityLogAction { + if (hasSearch(body)) return 'search'; + if (hasFilter(body)) return 'filter'; + + return 'index'; +} + +async function listRecords(ctx: Context, body: ListRequestBody, deps: ListHandlerDeps) { assertNoRelationFieldPaths(collectListFieldPaths(body)); const validationInput = toValidationInput(body); @@ -162,6 +175,15 @@ async function handleList(ctx: Context, body: ListRequestBody, deps: ListHandler ctx.body = mapListResponse(deps.collection, records, primaryKeys); } +async function handleList(ctx: Context, body: ListRequestBody, deps: ListHandlerDeps) { + await deps.activityLogs.record({ + ctx, + action: selectListAction(body), + context: { collectionName: deps.collection }, + operation: () => listRecords(ctx, body, deps), + }); +} + async function handleCount(ctx: Context, body: CountRequestBody, deps: RequestHandlerDeps) { assertNoRelationFieldPaths(collectCountFieldPaths(body)); @@ -217,7 +239,18 @@ async function resolveExposedRelationCapabilities( return result; } -async function handleRelationList( +function relationListLabel(relation: string, body: RelationListRequestBody): string { + const refinements: string[] = []; + + if (hasSearch(body)) refinements.push('search'); + if (hasFilter(body)) refinements.push('filter'); + + const suffix = refinements.length > 0 ? ` with ${refinements.join(' and ')}` : ''; + + return `list relation "${relation}"${suffix}`; +} + +async function listRelatedRecords( ctx: Context, body: RelationListRequestBody, deps: RelationListHandlerDeps, @@ -244,6 +277,23 @@ async function handleRelationList( ctx.body = mapListResponse(deps.foreignCollection, records, primaryKeys); } +async function handleRelationList( + ctx: Context, + body: RelationListRequestBody, + deps: RelationListHandlerDeps, +) { + await deps.activityLogs.record({ + ctx, + action: 'listRelatedData', + context: { + collectionName: deps.collection, + recordId: body.parentId, + label: relationListLabel(deps.relation, body), + }, + operation: () => listRelatedRecords(ctx, body, deps), + }); +} + async function handleRelationCount( ctx: Context, body: RelationCountRequestBody, @@ -308,6 +358,7 @@ export default function createDataRoutesMiddleware({ store, transport, logger, + activityLogs, createClient = defaultCreateAgentDataClient, }: DataRoutesMiddlewareOptions): Middleware { return async function dataRoutesMiddleware(ctx, next) { @@ -341,6 +392,7 @@ export default function createDataRoutesMiddleware({ token, timezone: ctx.state.timezone as string, logger, + activityLogs, }; const rawBody = ctx.request.body ?? {}; diff --git a/packages/agent-bff/src/http/bff-http-server.ts b/packages/agent-bff/src/http/bff-http-server.ts index 83643c9ceb..14ff73a1e0 100644 --- a/packages/agent-bff/src/http/bff-http-server.ts +++ b/packages/agent-bff/src/http/bff-http-server.ts @@ -12,10 +12,21 @@ import createVersionHeaderMiddleware from './version-header-middleware'; import createConsoleLogger from '../adapters/console-logger'; import warnMissingConfig from '../config/missing-config-warning'; +/** How long `stop()` waits for the open connections before it destroys them and drains anyway. */ +export const SHUTDOWN_TIMEOUT_MS = 10_000; + interface BFFHttpServerBaseOptions { port: number; config: BFFConfig; logger?: Logger; + /** Overrides `SHUTDOWN_TIMEOUT_MS`, for a host whose orchestrator grants a different grace. */ + shutdownTimeoutMs?: number; + /** + * Waits for the work no connection holds: the activity-log status transitions are fired without + * `await`, so `close()` does not cover them and a shutdown would leave entries `pending`. Takes + * what is left of the shutdown deadline and returns what that deadline cut short. + */ + drainActivityLogs?: (timeoutMs?: number) => Promise; } /** The server assembles its own Koa app around `/health` and the version header. */ @@ -115,15 +126,63 @@ export default class BFFHttpServer { }); } + /** + * The drain shares the connection deadline rather than getting one of its own: `stop()` as a + * whole has to fit the grace the orchestrator gives the process, and a status transition against + * a slow audit store retries long enough to outlast it on its own. + */ async stop(): Promise { + const { drainActivityLogs } = this.options; + const timeoutMs = this.shutdownTimeoutMs; + const deadline = Date.now() + timeoutMs; + + await this.closeConnections(timeoutMs); + + if (!drainActivityLogs) return; + + const unfinished = await drainActivityLogs(Math.max(deadline - Date.now(), 0)); + + if (unfinished.length === 0) return; + + this.logger('Warn', 'Stopped the Forest BFF with activity logs still in flight', { + timeoutMs, + unfinished, + }); + } + + private get shutdownTimeoutMs(): number { + return this.options.shutdownTimeoutMs ?? SHUTDOWN_TIMEOUT_MS; + } + + /** + * Bounded on purpose: `close()` resolves only once the last connection is gone, so a single busy + * one would hold the shutdown until the orchestrator sends SIGKILL and the drain would never + * run. Idle keep-alive connections go first, the rest get the deadline and are then destroyed. + */ + private async closeConnections(timeoutMs: number): Promise { + const { server } = this; + + if (!server) return; + return new Promise((resolve, reject) => { - if (!this.server) { + let settled = false; + + const timer = setTimeout(() => { + settled = true; + this.logger('Warn', 'Forcing the Forest BFF shutdown: connections were still open', { + timeoutMs, + }); + server.closeAllConnections(); + this.server = null; resolve(); + }, timeoutMs); + + server.close(err => { + if (settled) return; - return; - } + settled = true; + clearTimeout(timer); - this.server.close(err => { if (err) { reject(err); } else { @@ -131,6 +190,8 @@ export default class BFFHttpServer { resolve(); } }); + + server.closeIdleConnections(); }); } diff --git a/packages/agent-bff/src/http/bff-local-errors.ts b/packages/agent-bff/src/http/bff-local-errors.ts index c986f5cb17..27f468f81e 100644 --- a/packages/agent-bff/src/http/bff-local-errors.ts +++ b/packages/agent-bff/src/http/bff-local-errors.ts @@ -107,11 +107,47 @@ export function tooManyRequests( }); } +export const ACTION_REQUIRES_APPROVAL_TYPE = 'action_requires_approval'; + export function actionRequiresApproval( message = 'This action requires an approval before it can run', details?: unknown, ): BffHttpError { - return new BffHttpError(403, 'action_requires_approval', message, { details }); + return new BffHttpError(403, ACTION_REQUIRES_APPROVAL_TYPE, message, { details }); +} + +export const AUDIT_RETRY_AFTER_SECONDS = 5; + +export const AUDIT_UNAVAILABLE_TYPE = 'audit_unavailable'; + +/** + * `retryAfter` is optional: a retry only helps while the audit store is expected to answer soon. + * A deployment whose Forest server cannot write the log at all must not advertise one. + */ +export function auditUnavailable( + retryAfter?: number, + message = 'The activity log could not be written, so the operation was not performed', +): BffHttpError { + return new BffHttpError(503, AUDIT_UNAVAILABLE_TYPE, message, { retryAfter }); +} + +/** + * The audit trail cannot be written in this deployment at all — no credential minted, no endpoint + * exposed — as opposed to an outage that a retry outlives. The missing `retryAfter` is the marker: + * it is what the callers above use to say a retry can never succeed. + */ +export function isUnretryableAuditFailure(error: unknown): boolean { + return ( + error instanceof BffHttpError && + error.type === AUDIT_UNAVAILABLE_TYPE && + error.retryAfter === undefined + ); +} + +export function auditNotAuthorized( + message = 'Not authorized to write the activity log for this request', +): BffHttpError { + return new BffHttpError(403, 'audit_not_authorized', message); } export function environmentUnresolved(): BffHttpError { diff --git a/packages/agent-bff/src/openapi/openapi-document.ts b/packages/agent-bff/src/openapi/openapi-document.ts index dab32bcbcb..70dabe1ebe 100644 --- a/packages/agent-bff/src/openapi/openapi-document.ts +++ b/packages/agent-bff/src/openapi/openapi-document.ts @@ -39,7 +39,7 @@ const SECURITY = [{ [SESSION_SCHEME]: [] }, { [API_KEY_SCHEME]: [] }]; const ERROR_STATUSES: Record = { 400: 'Malformed body, a malformed URL-encoded path segment, an invalid filter operator, a filter nested too deep, ambiguous credentials, an unsupported page, a missing or invalid timezone, an unknown submitted action field, a required action field left empty or a malformed file value at execute, or a rejected action form (type action_error)', 401: 'Missing, invalid, or expired credentials', - 403: 'The action needs approval before it runs (the body carries the approving roles), the Forest identity behind the API key is not allowed, the origin is not allowed for this key, or the agent refused the collection, relation, or action', + 403: 'The action needs approval before it runs (the body carries the approving roles), the Forest identity behind the API key is not allowed, the origin is not allowed for this key, the Forest server refused to write the activity log the request needs (type audit_not_authorized), or the agent refused the collection, relation, or action', 404: 'Unknown collection, relation, or action', 413: `The request body exceeds the BFF limit of ${BODY_LIMIT}`, 415: 'The request Content-Type is neither application/json nor an application/*+json type, including form-urlencoded, and is rejected with 415 instead of being silently dropped; a request carrying a body with no Content-Type at all is rejected the same way; or the declared character set cannot be decoded', @@ -48,7 +48,7 @@ const ERROR_STATUSES: Record = { 500: 'The agent payload could not be mapped to the BFF contract, or the BFF hit an unexpected error', 501: 'The BFF is running without an agent configured, so the proxy is not implemented', 502: 'The agent refused the connection, its host could not be resolved, or the transport failed another way (a connection reset mid-flight, a socket hang up, a TLS failure) — it failed outright rather than running out of time', - 503: 'The agent schema is unavailable, the agent returned a 5xx, the API key could not be resolved, or the Forest permissions could not be fetched and no fresh cache was left (type permissions_unavailable)', + 503: 'The agent schema is unavailable, the agent returned a 5xx, the API key could not be resolved, the activity log an action execution must be recorded in could not be written, so the action was not run (type audit_unavailable), or the Forest permissions could not be fetched and no fresh cache was left (type permissions_unavailable)', 504: 'The agent did not answer before the BFF timeout (BFF_AGENT_TIMEOUT_MS, 10s by default). The deadline is armed when the request starts, so at the default it also covers a host that accepts nothing and never resets the connection — raise the timeout past the OS connect timeout and that case reverts to 502', }; diff --git a/packages/agent-bff/test/action/action-routes-activity-log.test.ts b/packages/agent-bff/test/action/action-routes-activity-log.test.ts new file mode 100644 index 0000000000..bbaa72f1c5 --- /dev/null +++ b/packages/agent-bff/test/action/action-routes-activity-log.test.ts @@ -0,0 +1,323 @@ +import type { AgentActionClient } from '../../src/action/agent-action-client'; +import type { ActivityLogWriter } from '../../src/activity-log/activity-log-writer'; +import type { Logger } from '../../src/ports/logger-port'; +import type { Middleware } from 'koa'; + +import { ActionRequiresApprovalError } from '@forestadmin/agent-client'; +import { HttpError } from '@forestadmin/forestadmin-client'; +import { bodyParser } from '@koa/bodyparser'; +import Koa from 'koa'; +import request from 'supertest'; + +import createActionRoutesMiddleware from '../../src/action/action-routes-middleware'; +import { createHttpTransport } from '../../src/agent/agent-transport'; +import createErrorMiddleware from '../../src/http/error-middleware'; +import { TIMEZONE, clientOf, makeAction, readModel, storeOf } from '../helpers/action-routes'; +import { + ACTIVITY_LOG_ID, + ACTIVITY_LOG_INDEX, + API_KEY_SERVER_TOKEN, + RENDERING_ID, + activityLogsOf, + apiKeyCredentials, + fakeActivityLogsService, + forestServerTokenStep, + oauthCredentials, + sessionAccessToken, +} from '../helpers/activity-log'; + +const noopLogger: Logger = () => undefined; + +function buildApp({ + service, + client, + credentials = apiKeyCredentials(), + saasAccessToken, +}: { + service: ReturnType; + client: AgentActionClient; + credentials?: Middleware; + saasAccessToken?: string; +}): { app: Koa; activityLogs: ActivityLogWriter } { + const activityLogs = activityLogsOf(service, noopLogger); + const app = new Koa(); + app.silent = true; + app.use(createErrorMiddleware({ logger: noopLogger })); + app.use(bodyParser()); + app.use(credentials); + app.use(forestServerTokenStep(saasAccessToken)); + app.use(async (ctx, next) => { + ctx.state.timezone = TIMEZONE; + ctx.state.agentToken = 'agent-jwt'; + await next(); + }); + app.use( + createActionRoutesMiddleware({ + store: storeOf(readModel), + transport: createHttpTransport({ agentUrl: 'https://agent.example.com' }), + logger: noopLogger, + activityLogs, + createClient: () => client, + }), + ); + + return { app, activityLogs }; +} + +function executingAction() { + return makeAction({ execute: jest.fn(async () => ({ success: 'Done' })) }); +} + +describe('action routes activity log', () => { + describe('when executing an action', () => { + it('should record the action, its records and its label', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: clientOf(executingAction()) }); + + const response = await request(app.callback()) + .post('/agent/v1/users/actions/approve/execute') + .send({ recordIds: ['42', '43'] }); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).toHaveBeenCalledWith({ + forestServerToken: API_KEY_SERVER_TOKEN, + renderingId: String(RENDERING_ID), + action: 'action', + type: 'write', + collectionName: 'users', + recordId: undefined, + recordIds: ['42', '43'], + label: 'triggered the action "approve"', + }); + }); + + it('should record an attempt on an action the read model does not expose', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: clientOf(executingAction()) }); + + const response = await request(app.callback()) + .post('/agent/v1/users/actions/ghost/execute') + .send({ recordIds: ['42'] }); + + expect(response.status).toBe(404); + expect(service.createMcpActivityLog).toHaveBeenCalledWith({ + forestServerToken: API_KEY_SERVER_TOKEN, + renderingId: String(RENDERING_ID), + action: 'action', + type: 'write', + collectionName: 'users', + recordId: undefined, + recordIds: ['42'], + label: 'triggered the action "ghost"', + }); + }); + + it('should mark that attempt failed', async () => { + const service = fakeActivityLogsService(); + const { app, activityLogs } = buildApp({ service, client: clientOf(executingAction()) }); + + await request(app.callback()) + .post('/agent/v1/users/actions/ghost/execute') + .send({ recordIds: ['42'] }); + await activityLogs.drain(); + + expect(service.updateActivityLogStatus).toHaveBeenCalledWith( + expect.objectContaining({ status: 'failed' }), + ); + }); + + it('should leave the form of an unexposed action unaudited, like every other form', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: clientOf(executingAction()) }); + + const response = await request(app.callback()) + .post('/agent/v1/users/actions/ghost/form') + .send({ recordIds: ['42'] }); + + expect(response.status).toBe(404); + expect(service.createMcpActivityLog).not.toHaveBeenCalled(); + }); + + it('should mark the log completed once the action ran', async () => { + const service = fakeActivityLogsService(); + const { app, activityLogs } = buildApp({ service, client: clientOf(executingAction()) }); + + await request(app.callback()) + .post('/agent/v1/users/actions/approve/execute') + .send({ recordIds: ['42'] }); + await activityLogs.drain(); + + expect(service.updateActivityLogStatus).toHaveBeenCalledWith({ + forestServerToken: API_KEY_SERVER_TOKEN, + activityLog: { id: ACTIVITY_LOG_ID, attributes: { index: ACTIVITY_LOG_INDEX } }, + status: 'completed', + }); + }); + + it('should mark the log failed when the action throws', async () => { + const service = fakeActivityLogsService(); + const form = makeAction({ + execute: jest.fn(async () => { + throw new Error('the agent is down'); + }), + }); + const { app, activityLogs } = buildApp({ service, client: clientOf(form) }); + + const response = await request(app.callback()) + .post('/agent/v1/users/actions/approve/execute') + .send({ recordIds: ['42'] }); + await activityLogs.drain(); + + expect(response.status).toBe(502); + expect(service.updateActivityLogStatus).toHaveBeenCalledWith( + expect.objectContaining({ status: 'failed' }), + ); + }); + + it('should mark the log failed when the action cannot even be loaded', async () => { + const service = fakeActivityLogsService(); + const loadAction = jest.fn(async () => { + throw new Error('the agent is down'); + }); + const { app, activityLogs } = buildApp({ + service, + client: clientOf(executingAction(), loadAction as jest.Mock), + }); + + await request(app.callback()) + .post('/agent/v1/users/actions/approve/execute') + .send({ recordIds: ['42'] }); + await activityLogs.drain(); + + expect(service.updateActivityLogStatus).toHaveBeenCalledWith( + expect.objectContaining({ status: 'failed' }), + ); + }); + + it('should mark the log completed when the action was routed for approval', async () => { + const service = fakeActivityLogsService(); + const form = makeAction({ + execute: jest.fn(async () => { + throw new ActionRequiresApprovalError('Needs approval', [7]); + }), + }); + const { app, activityLogs } = buildApp({ service, client: clientOf(form) }); + + const response = await request(app.callback()) + .post('/agent/v1/users/actions/approve/execute') + .send({ recordIds: ['42'] }); + await activityLogs.drain(); + + expect(response.status).toBe(403); + expect(response.body.error.type).toBe('action_requires_approval'); + expect(service.updateActivityLogStatus).toHaveBeenCalledWith( + expect.objectContaining({ status: 'completed' }), + ); + }); + + it('should refuse with audit_unavailable and never reach the agent when the log cannot be created', async () => { + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => { + throw new Error('the audit store is down'); + }), + }); + const loadAction = jest.fn(async () => executingAction()); + const { app } = buildApp({ + service, + client: clientOf(executingAction(), loadAction as jest.Mock), + }); + + const response = await request(app.callback()) + .post('/agent/v1/users/actions/approve/execute') + .send({ recordIds: ['42'] }); + + expect(response.status).toBe(503); + expect(response.body.error.type).toBe('audit_unavailable'); + expect(response.headers['retry-after']).toBe('5'); + expect(loadAction).not.toHaveBeenCalled(); + }); + + it('should refuse with audit_unavailable when the audit endpoint returns no log id', async () => { + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => ({ attributes: { index: ACTIVITY_LOG_INDEX } })), + }); + const { app } = buildApp({ service, client: clientOf(executingAction()) }); + + const response = await request(app.callback()) + .post('/agent/v1/users/actions/approve/execute') + .send({ recordIds: ['42'] }); + + expect(response.status).toBe(503); + expect(response.body.error.type).toBe('audit_unavailable'); + }); + + it('should refuse with audit_not_authorized when the audit endpoint rejects the identity', async () => { + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => { + throw new HttpError('Forbidden', 403); + }), + }); + const { app } = buildApp({ service, client: clientOf(executingAction()) }); + + const response = await request(app.callback()) + .post('/agent/v1/users/actions/approve/execute') + .send({ recordIds: ['42'] }); + + expect(response.status).toBe(403); + expect(response.body.error.type).toBe('audit_not_authorized'); + }); + + it('should refuse with session_expired when the oauth session cannot be resolved', async () => { + const service = fakeActivityLogsService(); + const loadAction = jest.fn(async () => executingAction()); + const { app } = buildApp({ + service, + client: clientOf(executingAction(), loadAction as jest.Mock), + credentials: oauthCredentials(), + saasAccessToken: undefined, + }); + + const response = await request(app.callback()) + .post('/agent/v1/users/actions/approve/execute') + .send({ recordIds: ['42'] }); + + expect(response.status).toBe(401); + expect(response.body.error.type).toBe('session_expired'); + expect(loadAction).not.toHaveBeenCalled(); + expect(service.createMcpActivityLog).not.toHaveBeenCalled(); + }); + + it('should use the session token when the caller carries an oauth session', async () => { + const service = fakeActivityLogsService(); + const saasAccessToken = sessionAccessToken(); + const { app } = buildApp({ + service, + client: clientOf(executingAction()), + credentials: oauthCredentials(), + saasAccessToken, + }); + + await request(app.callback()) + .post('/agent/v1/users/actions/approve/execute') + .send({ recordIds: ['42'] }); + + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ forestServerToken: saasAccessToken }), + ); + }); + }); + + describe('when loading an action form', () => { + it('should write no log', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: clientOf(makeAction()) }); + + const response = await request(app.callback()) + .post('/agent/v1/users/actions/approve/form') + .send({ recordIds: ['42'] }); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/packages/agent-bff/test/action/action-routes-middleware.test.ts b/packages/agent-bff/test/action/action-routes-middleware.test.ts index 5a290f72fe..34277f80d0 100644 --- a/packages/agent-bff/test/action/action-routes-middleware.test.ts +++ b/packages/agent-bff/test/action/action-routes-middleware.test.ts @@ -26,6 +26,7 @@ import { readModel, storeOf, } from '../helpers/action-routes'; +import { passthroughActivityLogs } from '../helpers/activity-log'; const TRANSPORT = createHttpTransport({ agentUrl: 'https://agent.example.com' }); @@ -48,6 +49,7 @@ describe('action routes middleware', () => { store: storeOf(readModel), transport: createHttpTransport({ agentUrl: 'https://agent.example.com', timeoutMs: 2500 }), logger: noopLogger, + activityLogs: passthroughActivityLogs(), createClient, }), ); @@ -79,6 +81,7 @@ describe('action routes middleware', () => { store: storeOf(readModel), transport: TRANSPORT, logger: noopLogger, + activityLogs: passthroughActivityLogs(), createClient, }), ); diff --git a/packages/agent-bff/test/activity-log/activity-log-drainer.test.ts b/packages/agent-bff/test/activity-log/activity-log-drainer.test.ts new file mode 100644 index 0000000000..0612ddf512 --- /dev/null +++ b/packages/agent-bff/test/activity-log/activity-log-drainer.test.ts @@ -0,0 +1,115 @@ +import ActivityLogDrainer from '../../src/activity-log/activity-log-drainer'; + +const stalled = () => new Promise(() => {}); + +const TRANSITION = "'completed' transition of the activity log log-1"; +const REQUEST = "'index' request on 'books'"; + +describe('activity log drainer', () => { + it('should wait for a tracked transition to settle', async () => { + const drainer = new ActivityLogDrainer(); + let settled = false; + + drainer.track( + () => + new Promise(resolve => { + setTimeout(() => { + settled = true; + resolve(); + }, 10); + }), + TRANSITION, + ); + + await drainer.drain(); + + expect(settled).toBe(true); + }); + + it('should wait for a rejected transition without rethrowing it', async () => { + const drainer = new ActivityLogDrainer(); + + const tracked = drainer.track(async () => { + throw new Error('the audit store is down'); + }, TRANSITION); + tracked.catch(() => undefined); + + await expect(drainer.drain()).resolves.toEqual([]); + }); + + it('should resolve immediately when nothing is in flight', async () => { + const drainer = new ActivityLogDrainer(); + + await expect(drainer.drain()).resolves.toEqual([]); + }); + + it('should return the tracked result to its caller', async () => { + const drainer = new ActivityLogDrainer(); + + await expect(drainer.track(async () => 'done', TRANSITION)).resolves.toBe('done'); + }); + + it('should wait for work registered by an operation that was already in flight', async () => { + const drainer = new ActivityLogDrainer(); + let transitionSettled = false; + + drainer.track( + () => + new Promise(resolveRequest => { + setTimeout(() => { + drainer.track( + () => + new Promise(resolveTransition => { + setTimeout(() => { + transitionSettled = true; + resolveTransition(); + }, 10); + }), + TRANSITION, + ); + resolveRequest(); + }, 10); + }), + REQUEST, + ); + + await drainer.drain(); + + expect(transitionSettled).toBe(true); + }); + + describe('when a deadline is shared with the drain', () => { + it('should return once it expires, naming what was still in flight', async () => { + const drainer = new ActivityLogDrainer(); + + drainer.track(stalled, TRANSITION); + drainer.track(stalled, REQUEST); + + await expect(drainer.drain(20)).resolves.toEqual([TRANSITION, REQUEST]); + }); + + it('should return as soon as the work settles, well inside the deadline', async () => { + const drainer = new ActivityLogDrainer(); + const startedAt = Date.now(); + + drainer.track( + () => + new Promise(resolve => { + setTimeout(resolve, 10); + }), + TRANSITION, + ); + + await expect(drainer.drain(10_000)).resolves.toEqual([]); + expect(Date.now() - startedAt).toBeLessThan(5_000); + }); + + it('should give a stalled operation no grace at all once the deadline is spent', async () => { + const drainer = new ActivityLogDrainer(); + + drainer.track(stalled, TRANSITION); + + await expect(drainer.drain(0)).resolves.toEqual([TRANSITION]); + }); + }); +}); diff --git a/packages/agent-bff/test/activity-log/activity-logs-creator.test.ts b/packages/agent-bff/test/activity-log/activity-logs-creator.test.ts new file mode 100644 index 0000000000..2aa5250df0 --- /dev/null +++ b/packages/agent-bff/test/activity-log/activity-logs-creator.test.ts @@ -0,0 +1,623 @@ +import type { Logger } from '../../src/ports/logger-port'; +import type { Context } from 'koa'; + +import { HttpError, NotFoundError } from '@forestadmin/forestadmin-client'; + +import ActivityLogDrainer from '../../src/activity-log/activity-log-drainer'; +import createPendingActivityLog, { + markActivityLog, +} from '../../src/activity-log/activity-logs-creator'; +import { AUDIT_RETRY_AFTER_SECONDS, auditUnavailable } from '../../src/http/bff-local-errors'; +import { + ACTIVITY_LOG_ID, + ACTIVITY_LOG_INDEX, + API_KEY_SERVER_TOKEN, + RENDERING_ID, + fakeActivityLogsService, +} from '../helpers/activity-log'; + +const RETRY_DELAY_MS = 500; +const MAX_ATTEMPTS = 5; + +function ctxOf(invalidateApiKeyIdentity: () => void = () => undefined): Context { + return { + state: { + authMode: 'api-key', + apiKeyIdentity: { renderingId: RENDERING_ID }, + resolveForestServerToken: async () => API_KEY_SERVER_TOKEN, + invalidateApiKeyIdentity, + }, + } as unknown as Context; +} + +function ctxRejectingCredentials(error: unknown): Context { + return { + state: { + authMode: 'api-key', + apiKeyIdentity: { renderingId: RENDERING_ID }, + resolveForestServerToken: async () => { + throw error; + }, + }, + } as unknown as Context; +} + +function rejectingService(error: unknown) { + return fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => { + throw error; + }), + }); +} + +function loggerSpy(): jest.MockedFunction { + return jest.fn() as unknown as jest.MockedFunction; +} + +function pendingLog() { + return { + activityLog: { id: ACTIVITY_LOG_ID, attributes: { index: ACTIVITY_LOG_INDEX } }, + forestServerToken: API_KEY_SERVER_TOKEN, + }; +} + +describe('activity logs creator', () => { + describe('when the credentials cannot be resolved', () => { + it('should name the rendering the request carries', async () => { + const logger = loggerSpy(); + + await createPendingActivityLog({ + ctx: ctxRejectingCredentials(new Error('no token on this request')), + service: fakeActivityLogsService(), + action: 'index', + context: { collectionName: 'books' }, + logger, + }); + + expect(logger).toHaveBeenCalledWith( + 'Error', + "Activity log for 'index' has no credentials to be created with", + { + renderingId: RENDERING_ID, + collectionName: 'books', + cause: 'Error: no token on this request', + }, + ); + }); + + it('should keep Error for a resolution that failed and may recover', async () => { + const logger = loggerSpy(); + + await createPendingActivityLog({ + ctx: ctxRejectingCredentials(auditUnavailable(AUDIT_RETRY_AFTER_SECONDS)), + service: fakeActivityLogsService(), + action: 'index', + logger, + }); + + expect(logger).toHaveBeenCalledWith( + 'Error', + "Activity log for 'index' has no credentials to be created with", + { + renderingId: RENDERING_ID, + cause: + 'BffHttpError: The activity log could not be written, so the operation was not ' + + 'performed', + }, + ); + }); + }); + + describe('when the deployment mints no credential to write the log with', () => { + const noCredential = () => auditUnavailable(undefined, 'this server mints no audit token'); + + it('should report a read once, as a warning, and serve it unaudited', async () => { + const logger = loggerSpy(); + + const pending = await createPendingActivityLog({ + ctx: ctxRejectingCredentials(noCredential()), + service: fakeActivityLogsService(), + action: 'index', + context: { collectionName: 'books' }, + logger, + }); + + expect(pending).toBeNull(); + expect(logger).toHaveBeenCalledTimes(1); + expect(logger).toHaveBeenCalledWith( + 'Warn', + "Activity log for 'index' was not created: this deployment has no credential to write it " + + 'with', + { + renderingId: RENDERING_ID, + collectionName: 'books', + cause: 'BffHttpError: this server mints no audit token', + }, + ); + }); + + it('should block an action, still reporting the supported degradation once', async () => { + const logger = loggerSpy(); + + await expect( + createPendingActivityLog({ + ctx: ctxRejectingCredentials(noCredential()), + service: fakeActivityLogsService(), + action: 'action', + logger, + }), + ).rejects.toMatchObject({ status: 503, type: 'audit_unavailable', retryAfter: undefined }); + + expect(logger).toHaveBeenCalledTimes(1); + expect(logger).toHaveBeenCalledWith( + 'Warn', + "Activity log for 'action' was not created: this deployment has no credential to write it " + + 'with', + { renderingId: RENDERING_ID, cause: 'BffHttpError: this server mints no audit token' }, + ); + }); + }); + + describe('when the server accepts the creation but returns no id', () => { + it('should serve a read unaudited and say the audit store dropped the write', async () => { + const logger = loggerSpy(); + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => ({ id: null })), + }); + + const pending = await createPendingActivityLog({ + ctx: ctxOf(), + service, + action: 'index', + context: { collectionName: 'books' }, + logger, + }); + + expect(pending).toBeNull(); + expect(logger).toHaveBeenCalledWith( + 'Error', + expect.stringContaining('the audit store dropped the write'), + { renderingId: RENDERING_ID, collectionName: 'books' }, + ); + }); + + it('should block an action, which must not run unaudited, and record why', async () => { + const logger = loggerSpy(); + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => ({ id: null })), + }); + + await expect( + createPendingActivityLog({ + ctx: ctxOf(), + service, + action: 'action', + context: { collectionName: 'books', label: 'triggered the action "Refund"' }, + logger, + }), + ).rejects.toMatchObject({ status: 503, type: 'audit_unavailable' }); + + expect(logger).toHaveBeenCalledWith( + 'Error', + expect.stringContaining('the audit store dropped the write'), + { renderingId: RENDERING_ID, collectionName: 'books' }, + ); + }); + }); + + describe('when the server accepts the creation but returns no index', () => { + it('should serve a read unaudited, since the status transition could never land', async () => { + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => ({ id: ACTIVITY_LOG_ID })), + }); + + const pending = await createPendingActivityLog({ + ctx: ctxOf(), + service, + action: 'index', + logger: loggerSpy(), + }); + + expect(pending).toBeNull(); + }); + + it('should block an action instead of stranding its entry pending', async () => { + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => ({ id: ACTIVITY_LOG_ID })), + }); + + await expect( + createPendingActivityLog({ + ctx: ctxOf(), + service, + action: 'action', + logger: loggerSpy(), + }), + ).rejects.toMatchObject({ status: 503, type: 'audit_unavailable' }); + }); + }); + + describe('when the server answers with an empty id or index', () => { + it('should block an action whose entry would strand pending on an empty id', async () => { + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => ({ + id: '', + attributes: { index: ACTIVITY_LOG_INDEX }, + })), + }); + + await expect( + createPendingActivityLog({ ctx: ctxOf(), service, action: 'action', logger: loggerSpy() }), + ).rejects.toMatchObject({ status: 503, type: 'audit_unavailable' }); + }); + + it('should block an action on an empty index too', async () => { + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => ({ + id: ACTIVITY_LOG_ID, + attributes: { index: '' }, + })), + }); + + await expect( + createPendingActivityLog({ ctx: ctxOf(), service, action: 'action', logger: loggerSpy() }), + ).rejects.toMatchObject({ status: 503, type: 'audit_unavailable' }); + }); + + it('should serve a read unaudited rather than track an entry it cannot transition', async () => { + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => ({ id: '', attributes: { index: '' } })), + }); + + await expect( + createPendingActivityLog({ ctx: ctxOf(), service, action: 'index', logger: loggerSpy() }), + ).resolves.toBeNull(); + }); + }); + + describe('when the server does not expose the activity log endpoint', () => { + const NO_ENDPOINT_MESSAGE = + 'The Forest server does not expose the endpoint the activity log is written through, so ' + + 'the operation was not performed'; + + it('should block an action without a retry hint on a 404', async () => { + const service = rejectingService(new NotFoundError()); + + await expect( + createPendingActivityLog({ ctx: ctxOf(), service, action: 'action', logger: loggerSpy() }), + ).rejects.toMatchObject({ + status: 503, + type: 'audit_unavailable', + retryAfter: undefined, + message: NO_ENDPOINT_MESSAGE, + }); + }); + + it('should block an action without a retry hint on a 501', async () => { + const service = rejectingService(new HttpError('not implemented', 501)); + + await expect( + createPendingActivityLog({ ctx: ctxOf(), service, action: 'action', logger: loggerSpy() }), + ).rejects.toMatchObject({ + status: 503, + type: 'audit_unavailable', + retryAfter: undefined, + message: NO_ENDPOINT_MESSAGE, + }); + }); + + it('should keep the retry hint when the endpoint answered with a failure', async () => { + const service = rejectingService(new HttpError('the audit store is down', 500)); + + await expect( + createPendingActivityLog({ ctx: ctxOf(), service, action: 'action', logger: loggerSpy() }), + ).rejects.toMatchObject({ + status: 503, + type: 'audit_unavailable', + retryAfter: AUDIT_RETRY_AFTER_SECONDS, + }); + }); + + it('should serve a read unaudited rather than refuse it', async () => { + const service = rejectingService(new NotFoundError()); + + await expect( + createPendingActivityLog({ ctx: ctxOf(), service, action: 'index', logger: loggerSpy() }), + ).resolves.toBeNull(); + }); + }); + + describe('when the server refuses the creation with a 403', () => { + it('should refuse a read too, which is not authorized either', async () => { + const service = rejectingService(new HttpError('forbidden', 403)); + + await expect( + createPendingActivityLog({ + ctx: ctxOf(), + service, + action: 'index', + logger: loggerSpy(), + }), + ).rejects.toMatchObject({ status: 403, type: 'audit_not_authorized' }); + }); + }); + + describe('when the server refuses the creation with a 401', () => { + it('should serve a read unaudited rather than refuse it', async () => { + const service = rejectingService(new HttpError('expired', 401)); + + const pending = await createPendingActivityLog({ + ctx: ctxOf(), + service, + action: 'index', + logger: loggerSpy(), + }); + + expect(pending).toBeNull(); + }); + + it('should block an action with audit_unavailable, not audit_not_authorized', async () => { + const service = rejectingService(new HttpError('expired', 401)); + const logger = loggerSpy(); + + await expect( + createPendingActivityLog({ ctx: ctxOf(), service, action: 'action', logger }), + ).rejects.toMatchObject({ status: 503, type: 'audit_unavailable' }); + + expect(logger).toHaveBeenCalledWith( + 'Error', + "Activity log for 'action' could not be created", + { renderingId: RENDERING_ID, cause: 'HttpError: expired' }, + ); + }); + + it('should drop the cached identity so the next request re-resolves the key', async () => { + const invalidateApiKeyIdentity = jest.fn(); + const service = rejectingService(new HttpError('expired', 401)); + + await createPendingActivityLog({ + ctx: ctxOf(invalidateApiKeyIdentity), + service, + action: 'index', + logger: loggerSpy(), + }); + + expect(invalidateApiKeyIdentity).toHaveBeenCalledTimes(1); + }); + + it('should keep the cached identity when the refusal is a 403', async () => { + const invalidateApiKeyIdentity = jest.fn(); + const service = rejectingService(new HttpError('forbidden', 403)); + + await createPendingActivityLog({ + ctx: ctxOf(invalidateApiKeyIdentity), + service, + action: 'index', + logger: loggerSpy(), + }).catch(() => undefined); + + expect(invalidateApiKeyIdentity).not.toHaveBeenCalled(); + }); + }); + + describe('when the status transition lands before the document exists', () => { + beforeEach(() => { + jest.useFakeTimers(); + }); + + afterEach(() => { + jest.restoreAllMocks(); + jest.useRealTimers(); + }); + + it('should retry and succeed once the document is there', async () => { + const updateActivityLogStatus = jest + .fn() + .mockRejectedValueOnce(new NotFoundError()) + .mockResolvedValueOnce(undefined); + const service = fakeActivityLogsService({ updateActivityLogStatus }); + const drainer = new ActivityLogDrainer(); + + markActivityLog({ + service, + drainer, + pending: pendingLog(), + status: 'completed', + logger: loggerSpy(), + }); + + await jest.advanceTimersByTimeAsync(RETRY_DELAY_MS); + await drainer.drain(); + + expect(updateActivityLogStatus).toHaveBeenCalledTimes(2); + expect(updateActivityLogStatus).toHaveBeenLastCalledWith({ + forestServerToken: API_KEY_SERVER_TOKEN, + activityLog: { id: ACTIVITY_LOG_ID, attributes: { index: ACTIVITY_LOG_INDEX } }, + status: 'completed', + }); + }); + + it('should not let the wait it schedules keep the event loop alive', async () => { + const scheduleTimer = global.setTimeout; + const handles: Array<{ unref: jest.Mock }> = []; + const setTimeoutSpy = jest.spyOn(global, 'setTimeout').mockImplementation((( + callback: () => void, + delay: number, + ) => { + scheduleTimer(callback, delay); + const handle = { unref: jest.fn() }; + handles.push(handle); + + return handle; + }) as unknown as typeof global.setTimeout); + const updateActivityLogStatus = jest + .fn() + .mockRejectedValueOnce(new NotFoundError()) + .mockResolvedValueOnce(undefined); + const service = fakeActivityLogsService({ updateActivityLogStatus }); + const drainer = new ActivityLogDrainer(); + + markActivityLog({ + service, + drainer, + pending: pendingLog(), + status: 'completed', + logger: loggerSpy(), + }); + + await jest.advanceTimersByTimeAsync(RETRY_DELAY_MS); + await drainer.drain(); + + expect(setTimeoutSpy).toHaveBeenCalledWith(expect.any(Function), RETRY_DELAY_MS); + expect(handles[0].unref).toHaveBeenCalledTimes(1); + }); + + it('should give up after the last attempt and report the entry it could not mark', async () => { + const updateActivityLogStatus = jest.fn().mockRejectedValue(new NotFoundError()); + const service = fakeActivityLogsService({ updateActivityLogStatus }); + const drainer = new ActivityLogDrainer(); + const logger = loggerSpy(); + + markActivityLog({ + service, + drainer, + pending: pendingLog(), + status: 'failed', + logger, + }); + + await jest.advanceTimersByTimeAsync(RETRY_DELAY_MS * MAX_ATTEMPTS); + await drainer.drain(); + + expect(updateActivityLogStatus).toHaveBeenCalledTimes(MAX_ATTEMPTS); + expect(logger).toHaveBeenCalledWith('Error', "Failed to mark the activity log as 'failed'", { + activityLogId: ACTIVITY_LOG_ID, + index: ACTIVITY_LOG_INDEX, + cause: 'NotFoundError: Not found', + }); + }); + }); + + describe('when the status transition fails for a transient reason', () => { + beforeEach(() => { + jest.useFakeTimers(); + }); + + afterEach(() => { + jest.restoreAllMocks(); + jest.useRealTimers(); + }); + + it('should retry a 500 and succeed once the audit store answers', async () => { + const updateActivityLogStatus = jest + .fn() + .mockRejectedValueOnce(new HttpError('the audit store is down', 500)) + .mockResolvedValueOnce(undefined); + const service = fakeActivityLogsService({ updateActivityLogStatus }); + const drainer = new ActivityLogDrainer(); + + markActivityLog({ + service, + drainer, + pending: pendingLog(), + status: 'completed', + logger: loggerSpy(), + }); + + await jest.advanceTimersByTimeAsync(RETRY_DELAY_MS); + await drainer.drain(); + + expect(updateActivityLogStatus).toHaveBeenCalledTimes(2); + expect(updateActivityLogStatus).toHaveBeenLastCalledWith({ + forestServerToken: API_KEY_SERVER_TOKEN, + activityLog: { id: ACTIVITY_LOG_ID, attributes: { index: ACTIVITY_LOG_INDEX } }, + status: 'completed', + }); + }); + + it('should retry a transport failure carrying no status', async () => { + const updateActivityLogStatus = jest + .fn() + .mockRejectedValueOnce(new Error('connect ECONNREFUSED')) + .mockResolvedValueOnce(undefined); + const service = fakeActivityLogsService({ updateActivityLogStatus }); + const drainer = new ActivityLogDrainer(); + + markActivityLog({ + service, + drainer, + pending: pendingLog(), + status: 'failed', + logger: loggerSpy(), + }); + + await jest.advanceTimersByTimeAsync(RETRY_DELAY_MS); + await drainer.drain(); + + expect(updateActivityLogStatus).toHaveBeenCalledTimes(2); + }); + + it('should give up after the last attempt and report the entry it could not mark', async () => { + const updateActivityLogStatus = jest + .fn() + .mockRejectedValue(new HttpError('the audit store is down', 503)); + const service = fakeActivityLogsService({ updateActivityLogStatus }); + const drainer = new ActivityLogDrainer(); + const logger = loggerSpy(); + + markActivityLog({ + service, + drainer, + pending: pendingLog(), + status: 'completed', + logger, + }); + + await jest.advanceTimersByTimeAsync(RETRY_DELAY_MS * MAX_ATTEMPTS); + await drainer.drain(); + + expect(updateActivityLogStatus).toHaveBeenCalledTimes(MAX_ATTEMPTS); + expect(logger).toHaveBeenCalledWith( + 'Error', + "Failed to mark the activity log as 'completed'", + { + activityLogId: ACTIVITY_LOG_ID, + index: ACTIVITY_LOG_INDEX, + cause: 'HttpError: the audit store is down', + }, + ); + }); + }); + + describe('when the status transition is refused', () => { + it('should report it without retrying, since a retry recovers nothing', async () => { + const updateActivityLogStatus = jest + .fn() + .mockRejectedValue(new HttpError('this token may not write that log', 403)); + const service = fakeActivityLogsService({ updateActivityLogStatus }); + const drainer = new ActivityLogDrainer(); + const logger = loggerSpy(); + + markActivityLog({ + service, + drainer, + pending: pendingLog(), + status: 'completed', + logger, + }); + + await drainer.drain(); + + expect(updateActivityLogStatus).toHaveBeenCalledTimes(1); + expect(logger).toHaveBeenCalledWith( + 'Error', + "Failed to mark the activity log as 'completed'", + { + activityLogId: ACTIVITY_LOG_ID, + index: ACTIVITY_LOG_INDEX, + cause: 'HttpError: this token may not write that log', + }, + ); + }); + }); +}); diff --git a/packages/agent-bff/test/activity-log/activity-logs-service.test.ts b/packages/agent-bff/test/activity-log/activity-logs-service.test.ts new file mode 100644 index 0000000000..c8b761ee53 --- /dev/null +++ b/packages/agent-bff/test/activity-log/activity-logs-service.test.ts @@ -0,0 +1,19 @@ +import { ActivityLogsService, ForestHttpApi } from '@forestadmin/forestadmin-client'; + +import createBffActivityLogsService from '../../src/activity-log/activity-logs-service'; + +jest.mock('@forestadmin/forestadmin-client', () => ({ + ...jest.requireActual('@forestadmin/forestadmin-client'), + ActivityLogsService: jest.fn(), +})); + +describe('BFF activity logs service', () => { + it('should build the service with the BFF application source header', () => { + createBffActivityLogsService('https://api.forestadmin.com'); + + expect(ActivityLogsService).toHaveBeenCalledWith(expect.any(ForestHttpApi), { + forestServerUrl: 'https://api.forestadmin.com', + headers: { 'Forest-Application-Source': 'BFF' }, + }); + }); +}); diff --git a/packages/agent-bff/test/api-key/api-key-authenticator.test.ts b/packages/agent-bff/test/api-key/api-key-authenticator.test.ts index 55311521a1..e71bef42ca 100644 --- a/packages/agent-bff/test/api-key/api-key-authenticator.test.ts +++ b/packages/agent-bff/test/api-key/api-key-authenticator.test.ts @@ -47,6 +47,43 @@ describe('api key authenticator', () => { mintMock.mockClear(); }); + describe('invalidation', () => { + it('should resolve the key again on the next request', async () => { + const resolve = jest.fn(async () => IDENTITY); + const authenticator = buildAuthenticator(resolve, nowRef); + await authenticator.authenticate(RAW); + + authenticator.invalidate(RAW); + await authenticator.authenticate(RAW); + + expect(resolve).toHaveBeenCalledTimes(2); + }); + + it('should keep serving another key from the cache', async () => { + const resolve = jest.fn(async () => IDENTITY); + const authenticator = buildAuthenticator(resolve, nowRef); + const otherKey = `fbff_${'c'.repeat(16)}_${'d'.repeat(64)}`; + await authenticator.authenticate(RAW); + await authenticator.authenticate(otherKey); + + authenticator.invalidate(RAW); + await authenticator.authenticate(otherKey); + + expect(resolve).toHaveBeenCalledTimes(2); + }); + + it('should ignore a key it could never have cached', async () => { + const resolve = jest.fn(async () => IDENTITY); + const authenticator = buildAuthenticator(resolve, nowRef); + await authenticator.authenticate(RAW); + + authenticator.invalidate('not-a-key'); + await authenticator.authenticate(RAW); + + expect(resolve).toHaveBeenCalledTimes(1); + }); + }); + describe('valid key', () => { it('should resolve the key and mint an agent token from the identity', async () => { const resolve = jest.fn(async () => IDENTITY); diff --git a/packages/agent-bff/test/api-key/api-key-middleware.test.ts b/packages/agent-bff/test/api-key/api-key-middleware.test.ts index 4bf59daeee..284fdfcdd9 100644 --- a/packages/agent-bff/test/api-key/api-key-middleware.test.ts +++ b/packages/agent-bff/test/api-key/api-key-middleware.test.ts @@ -5,7 +5,10 @@ import Koa from 'koa'; import request from 'supertest'; import { invalidApiKey, keyResolutionUnavailable } from '../../src/api-key/api-key-error'; -import createApiKeyMiddleware, { BFF_KEY_HEADER } from '../../src/api-key/api-key-middleware'; +import createApiKeyMiddleware, { + BFF_KEY_HEADER, + invalidateApiKeyIdentity, +} from '../../src/api-key/api-key-middleware'; import createErrorMiddleware from '../../src/http/error-middleware'; const KEY_ID = 'a'.repeat(16); @@ -33,6 +36,8 @@ interface LogLine { } function buildApp(authenticate: ApiKeyAuthenticator['authenticate']) { + const invalidate = () => undefined; + const logs: LogLine[] = []; const logger = (level: LoggerLevel, message: string, context?: Record) => { @@ -42,7 +47,7 @@ function buildApp(authenticate: ApiKeyAuthenticator['authenticate']) { const app = new Koa(); app.silent = true; app.use(createErrorMiddleware({ logger })); - app.use(createApiKeyMiddleware({ authenticator: { authenticate }, logger })); + app.use(createApiKeyMiddleware({ authenticator: { authenticate, invalidate }, logger })); app.use(async ctx => { ctx.status = 200; ctx.body = { @@ -166,7 +171,12 @@ describe('api key middleware', () => { const app = new Koa(); app.silent = true; app.use(createErrorMiddleware({ logger })); - app.use(createApiKeyMiddleware({ authenticator: { authenticate }, logger })); + app.use( + createApiKeyMiddleware({ + authenticator: { authenticate, invalidate: () => undefined }, + logger, + }), + ); app.use(async () => { throw new Error('downstream boom'); }); @@ -192,4 +202,74 @@ describe('api key middleware', () => { expect(response.body).toEqual({ agentToken: null, identity: null }); }); }); + + describe('when a downstream middleware refuses the resolved identity', () => { + it('should drop the cached resolution of the key it was authenticated with', async () => { + const authenticate = jest.fn(async () => ({ + agentToken: 'minted-token', + identity: IDENTITY, + })); + const invalidate = jest.fn(); + const logger = () => undefined; + + const app = new Koa(); + app.silent = true; + app.use(createErrorMiddleware({ logger })); + app.use(createApiKeyMiddleware({ authenticator: { authenticate, invalidate }, logger })); + app.use(async ctx => { + invalidateApiKeyIdentity(ctx); + ctx.status = 204; + }); + + await request(app.callback()).get('/').set(BFF_KEY_HEADER, RAW); + + expect(invalidate).toHaveBeenCalledWith(RAW, undefined); + }); + + it('should name the refused server token, so a second one is not suppressed as a repeat', async () => { + const authenticate = jest.fn(async () => ({ + agentToken: 'minted-token', + identity: IDENTITY, + forestServerToken: 'saas-token', + })); + const invalidate = jest.fn(); + const logger = () => undefined; + + const app = new Koa(); + app.silent = true; + app.use(createErrorMiddleware({ logger })); + app.use(createApiKeyMiddleware({ authenticator: { authenticate, invalidate }, logger })); + app.use(async ctx => { + invalidateApiKeyIdentity(ctx); + ctx.status = 204; + }); + + await request(app.callback()).get('/').set(BFF_KEY_HEADER, RAW); + + expect(invalidate).toHaveBeenCalledWith(RAW, 'saas-token'); + }); + + it('should do nothing when the request carried no api key', async () => { + const invalidate = jest.fn(); + const logger = () => undefined; + + const app = new Koa(); + app.silent = true; + app.use(createErrorMiddleware({ logger })); + app.use( + createApiKeyMiddleware({ + authenticator: { authenticate: jest.fn(), invalidate }, + logger, + }), + ); + app.use(async ctx => { + invalidateApiKeyIdentity(ctx); + ctx.status = 204; + }); + + await request(app.callback()).get('/'); + + expect(invalidate).not.toHaveBeenCalled(); + }); + }); }); diff --git a/packages/agent-bff/test/api-key/resolve-cache.test.ts b/packages/agent-bff/test/api-key/resolve-cache.test.ts index cc63670d11..0e786e6a3d 100644 --- a/packages/agent-bff/test/api-key/resolve-cache.test.ts +++ b/packages/agent-bff/test/api-key/resolve-cache.test.ts @@ -43,6 +43,113 @@ describe('resolve cache', () => { }); }); + describe('invalidation', () => { + it('should forget a positive entry before its TTL', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60 }); + cache.setPositive('hash', IDENTITY); + + cache.invalidate('hash'); + + expect(cache.getPositive('hash')).toBeUndefined(); + }); + + it('should leave the other entries alone', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60 }); + cache.setPositive('hash', IDENTITY); + cache.setPositive('other-hash', IDENTITY); + + cache.invalidate('hash'); + + expect(cache.getPositive('other-hash')).toEqual(IDENTITY); + }); + + it('should ignore a second invalidation of the same key within the TTL window', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60 }); + cache.setPositive('hash', IDENTITY); + cache.invalidate('hash'); + cache.setPositive('hash', IDENTITY); + nowMs += 59_000; + + cache.invalidate('hash'); + + expect(cache.getPositive('hash')).toEqual(IDENTITY); + }); + + it('should still forget another key while one is within its window', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60 }); + cache.setPositive('hash', IDENTITY); + cache.setPositive('other-hash', IDENTITY); + cache.invalidate('hash'); + cache.setPositive('hash', IDENTITY); + + cache.invalidate('hash'); + cache.invalidate('other-hash'); + + expect(cache.getPositive('hash')).toEqual(IDENTITY); + expect(cache.getPositive('other-hash')).toBeUndefined(); + }); + + it('should invalidate again once the window has passed', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60 }); + cache.setPositive('hash', IDENTITY); + cache.invalidate('hash'); + nowMs += 60_000; + cache.setPositive('hash', IDENTITY); + + cache.invalidate('hash'); + + expect(cache.getPositive('hash')).toBeUndefined(); + }); + + it('should forget the entry when a second, different credential is refused in the window', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60 }); + cache.setPositive('hash', IDENTITY); + cache.invalidate('hash', 'first-token'); + cache.setPositive('hash', IDENTITY); + + cache.invalidate('hash', 'second-token'); + + expect(cache.getPositive('hash')).toBeUndefined(); + }); + + it('should ignore a repeat refusal of the credential that opened the window', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60 }); + cache.setPositive('hash', IDENTITY); + cache.invalidate('hash', 'first-token'); + cache.setPositive('hash', IDENTITY); + + cache.invalidate('hash', 'first-token'); + + expect(cache.getPositive('hash')).toEqual(IDENTITY); + }); + + it('should allow only one such retry, so a server minting a new token each time cannot thrash it', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60 }); + cache.setPositive('hash', IDENTITY); + cache.invalidate('hash', 'first-token'); + cache.setPositive('hash', IDENTITY); + cache.invalidate('hash', 'second-token'); + cache.setPositive('hash', IDENTITY); + + cache.invalidate('hash', 'third-token'); + + expect(cache.getPositive('hash')).toEqual(IDENTITY); + }); + + it('should not push the deadline back when a second credential resets it', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60 }); + cache.invalidate('hash', 'first-token'); + nowMs += 59_000; + cache.invalidate('hash', 'second-token'); + nowMs += 1_000; + cache.setPositive('hash', IDENTITY); + + cache.invalidate('hash', 'third-token'); + + expect(cache.getPositive('hash')).toBeUndefined(); + }); + }); + describe('negative entries', () => { it('should return the cached error within the negative TTL', () => { const cache = createResolveCache({ now, negativeTtlSeconds: 10 }); @@ -84,6 +191,32 @@ describe('resolve cache', () => { expect(cache.getPositive('c')).toEqual(IDENTITY); }); + it('should evict the oldest invalidation window once maxEntries is reached', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60, maxEntries: 2 }); + cache.invalidate('a'); + cache.invalidate('b'); + cache.invalidate('c'); + cache.setPositive('a', IDENTITY); + + cache.invalidate('a'); + + expect(cache.getPositive('a')).toBeUndefined(); + }); + + it('should drop an expired invalidation window rather than evict a live one', () => { + const cache = createResolveCache({ now, positiveTtlSeconds: 60, maxEntries: 2 }); + cache.invalidate('a'); + cache.invalidate('b'); + nowMs += 61_000; + cache.invalidate('a'); + cache.invalidate('c'); + cache.setPositive('a', IDENTITY); + + cache.invalidate('a'); + + expect(cache.getPositive('a')).toEqual(IDENTITY); + }); + it('should still overwrite an existing key when full', () => { const cache = createResolveCache({ now, maxEntries: 1 }); cache.setPositive('a', IDENTITY); diff --git a/packages/agent-bff/test/auth/forest-server-token-middleware.test.ts b/packages/agent-bff/test/auth/forest-server-token-middleware.test.ts new file mode 100644 index 0000000000..3bc3136bca --- /dev/null +++ b/packages/agent-bff/test/auth/forest-server-token-middleware.test.ts @@ -0,0 +1,240 @@ +import type ForestServerClient from '../../src/oauth/forest-server-client'; +import type { SessionStore } from '../../src/oauth/session-store'; +import type { Logger } from '../../src/ports/logger-port'; +import type { Context } from 'koa'; + +import jsonwebtoken from 'jsonwebtoken'; + +import createForestServerTokenMiddleware, { + resolveForestServerToken, +} from '../../src/auth/forest-server-token-middleware'; +import OAuthExchangeError from '../../src/oauth/oauth-exchange-error'; +import { + API_KEY_SERVER_TOKEN, + RENDERING_ID, + SESSION_ID, + sessionAccessToken, + unusedServerClient, +} from '../helpers/activity-log'; + +function contextOf(state: Record): Context { + return { state } as unknown as Context; +} + +function expiredAccessToken(): string { + return jsonwebtoken.sign({ scope: 'forest' }, 'session-secret', { expiresIn: '-1s' }); +} + +function storeOf(saasAccessToken: string | undefined, get = jest.fn()) { + const store = { + get: get.mockImplementation((sid: string) => + sid === SESSION_ID && saasAccessToken !== undefined ? { saasAccessToken } : undefined, + ), + } as unknown as SessionStore; + + return { store, get }; +} + +async function landResolver( + ctx: Context, + store?: SessionStore, + logger: Logger = () => undefined, +): Promise<() => Promise> { + const middleware = createForestServerTokenMiddleware({ + session: store ? { store, serverClient: unusedServerClient } : undefined, + logger, + }); + + await middleware(ctx, async () => undefined); + + return () => resolveForestServerToken(ctx); +} + +describe('forest server token middleware', () => { + describe('in api-key mode', () => { + it('should resolve the token the key resolution carried', async () => { + const ctx = contextOf({ + authMode: 'api-key', + apiKeyIdentity: { renderingId: RENDERING_ID }, + forestServerToken: API_KEY_SERVER_TOKEN, + }); + + const resolve = await landResolver(ctx); + + await expect(resolve()).resolves.toBe(API_KEY_SERVER_TOKEN); + }); + + it('should refuse without advertising a retry when the resolution carried no token', async () => { + const ctx = contextOf({ authMode: 'api-key', apiKeyIdentity: { renderingId: RENDERING_ID } }); + + const resolve = await landResolver(ctx); + + await expect(resolve()).rejects.toMatchObject({ + status: 503, + type: 'audit_unavailable', + retryAfter: undefined, + message: + 'The Forest server does not provide the credential the activity log is written with, ' + + 'so the operation was not performed', + }); + }); + }); + + describe('in oauth mode', () => { + it('should resolve the token held by the session', async () => { + const saasAccessToken = sessionAccessToken(); + const { store } = storeOf(saasAccessToken); + const ctx = contextOf({ + authMode: 'oauth', + principal: { sid: SESSION_ID, rendering_id: String(RENDERING_ID) }, + }); + + const resolve = await landResolver(ctx, store); + + await expect(resolve()).resolves.toBe(saasAccessToken); + }); + + it('should refuse with session_expired when the session is gone', async () => { + const { store } = storeOf(undefined); + const ctx = contextOf({ + authMode: 'oauth', + principal: { sid: SESSION_ID, rendering_id: String(RENDERING_ID) }, + }); + + const resolve = await landResolver(ctx, store); + + await expect(resolve()).rejects.toMatchObject({ + status: 401, + type: 'session_expired', + }); + }); + + it('should refuse with audit_unavailable when the Forest server cannot be reached', async () => { + const store = { + get: () => ({ saasAccessToken: expiredAccessToken() }), + getSaasRefreshToken: () => 'refresh-token', + } as unknown as SessionStore; + const serverClient = { + refreshServerToken: async () => { + throw new Error('connect ECONNREFUSED'); + }, + } as unknown as ForestServerClient; + const ctx = contextOf({ + authMode: 'oauth', + principal: { sid: SESSION_ID, rendering_id: String(RENDERING_ID) }, + }); + + const middleware = createForestServerTokenMiddleware({ + session: { store, serverClient }, + logger: () => undefined, + }); + await middleware(ctx, async () => undefined); + + await expect(resolveForestServerToken(ctx)).rejects.toMatchObject({ + status: 503, + type: 'audit_unavailable', + }); + }); + + it('should report the original failure, which the mapped error drops', async () => { + const store = { + get: () => { + throw new TypeError('sessions.get is not a function'); + }, + } as unknown as SessionStore; + const logger = jest.fn(); + const ctx = contextOf({ + authMode: 'oauth', + principal: { sid: SESSION_ID, rendering_id: String(RENDERING_ID) }, + }); + + const resolve = await landResolver(ctx, store, logger); + + await expect(resolve()).rejects.toMatchObject({ status: 503, type: 'audit_unavailable' }); + expect(logger).toHaveBeenCalledWith( + 'Error', + 'Could not resolve the Forest server access of this session', + { renderingId: RENDERING_ID, cause: 'sessions.get is not a function' }, + ); + }); + + it('should refuse with session_expired when the Forest server rejects the refresh token', async () => { + const store = { + get: () => ({ saasAccessToken: expiredAccessToken() }), + getSaasRefreshToken: () => 'refresh-token', + } as unknown as SessionStore; + const serverClient = { + refreshServerToken: async () => { + throw new OAuthExchangeError('invalid_grant', 'the refresh token was revoked'); + }, + } as unknown as ForestServerClient; + const ctx = contextOf({ + authMode: 'oauth', + principal: { sid: SESSION_ID, rendering_id: String(RENDERING_ID) }, + }); + + const logger = jest.fn(); + const middleware = createForestServerTokenMiddleware({ + session: { store, serverClient }, + logger, + }); + await middleware(ctx, async () => undefined); + + await expect(resolveForestServerToken(ctx)).rejects.toMatchObject({ + status: 401, + type: 'session_expired', + }); + expect(logger).toHaveBeenCalledWith( + 'Error', + 'Could not resolve the Forest server access of this session', + { renderingId: RENDERING_ID, cause: 'The Forest server rejected the refresh token' }, + ); + }); + + it('should refuse with session_expired when the deployment carries no session store', async () => { + const ctx = contextOf({ + authMode: 'oauth', + principal: { sid: SESSION_ID, rendering_id: String(RENDERING_ID) }, + }); + + const resolve = await landResolver(ctx); + + await expect(resolve()).rejects.toMatchObject({ + status: 401, + type: 'session_expired', + }); + }); + + it('should look the session up only once for repeated resolutions', async () => { + const { store, get } = storeOf(sessionAccessToken()); + const ctx = contextOf({ + authMode: 'oauth', + principal: { sid: SESSION_ID, rendering_id: String(RENDERING_ID) }, + }); + + const resolve = await landResolver(ctx, store); + await resolve(); + await resolve(); + + expect(get).toHaveBeenCalledTimes(1); + }); + }); + + it('should not look the session up when nothing resolves the token', async () => { + const { store, get } = storeOf(sessionAccessToken()); + const ctx = contextOf({ + authMode: 'oauth', + principal: { sid: SESSION_ID, rendering_id: String(RENDERING_ID) }, + }); + + await landResolver(ctx, store); + + expect(get).not.toHaveBeenCalled(); + }); + + it('should refuse with session_expired when no resolver was landed on the context', () => { + expect(() => resolveForestServerToken(contextOf({ authMode: 'oauth' }))).toThrow( + expect.objectContaining({ status: 401, type: 'session_expired' }), + ); + }); +}); diff --git a/packages/agent-bff/test/cli-shutdown.test.ts b/packages/agent-bff/test/cli-shutdown.test.ts new file mode 100644 index 0000000000..7b70234863 --- /dev/null +++ b/packages/agent-bff/test/cli-shutdown.test.ts @@ -0,0 +1,73 @@ +import type BFFHttpServer from '../src/http/bff-http-server'; +import type { Logger } from '../src/ports/logger-port'; + +import { installShutdownHandlers } from '../src/cli-core'; + +const noopLogger: Logger = () => undefined; + +const SIGNALS: NodeJS.Signals[] = ['SIGTERM', 'SIGINT']; + +function serverStub(stop = jest.fn(async () => undefined)) { + return { server: { stop } as unknown as BFFHttpServer, stop }; +} + +function installedHandlers(): { signal: NodeJS.Signals; handler: () => void }[] { + return SIGNALS.map(signal => ({ + signal, + handler: process.listeners(signal).at(-1) as () => void, + })); +} + +describe('shutdown handlers', () => { + let installed: { signal: NodeJS.Signals; handler: () => void }[] = []; + + afterEach(() => { + for (const { signal, handler } of installed) process.removeListener(signal, handler); + installed = []; + }); + + it.each(SIGNALS)('should stop the server on %s', signal => { + const { server, stop } = serverStub(); + + installShutdownHandlers(server, noopLogger); + installed = installedHandlers(); + installed.find(entry => entry.signal === signal)?.handler(); + + expect(stop).toHaveBeenCalledTimes(1); + }); + + it('should report a shutdown that failed, with what it failed on', async () => { + const logger = jest.fn(); + const closeError = new Error('close failed'); + const stop = jest.fn(async () => { + throw closeError; + }); + + installShutdownHandlers({ stop } as unknown as BFFHttpServer, logger as unknown as Logger); + installed = installedHandlers(); + installed[0].handler(); + + await expect(stop.mock.results[0].value).rejects.toBe(closeError); + + expect(logger).toHaveBeenCalledWith('Error', 'The Forest BFF did not stop cleanly', { + cause: 'close failed', + }); + }); + + it('should replace the handlers of a previous server instead of adding a pair', () => { + const first = serverStub(); + const second = serverStub(); + + installShutdownHandlers(first.server, noopLogger); + const before = process.listenerCount('SIGTERM'); + installShutdownHandlers(second.server, noopLogger); + installed = installedHandlers(); + + expect(process.listenerCount('SIGTERM')).toBe(before); + + installed[0].handler(); + + expect(first.stop).not.toHaveBeenCalled(); + expect(second.stop).toHaveBeenCalledTimes(1); + }); +}); diff --git a/packages/agent-bff/test/context/context-routes-middleware.test.ts b/packages/agent-bff/test/context/context-routes-middleware.test.ts index 2ee3bda46c..330467e021 100644 --- a/packages/agent-bff/test/context/context-routes-middleware.test.ts +++ b/packages/agent-bff/test/context/context-routes-middleware.test.ts @@ -71,6 +71,7 @@ function makeFullAgentEdge(fetchSchema: jest.Mock, allowedOrigins: string[] = [] agentToken: 'agent-token', identity: apiKeyIdentity(allowedOrigins), }), + invalidate: () => undefined, }, logger, }), diff --git a/packages/agent-bff/test/data/data-routes-activity-log.test.ts b/packages/agent-bff/test/data/data-routes-activity-log.test.ts new file mode 100644 index 0000000000..584a24cf43 --- /dev/null +++ b/packages/agent-bff/test/data/data-routes-activity-log.test.ts @@ -0,0 +1,397 @@ +import type { ActivityLogWriter } from '../../src/activity-log/activity-log-writer'; +import type { AgentDataClient } from '../../src/data/agent-data-client'; +import type { Logger } from '../../src/ports/logger-port'; +import type ReadModelStore from '../../src/read-model/read-model-store'; +import type { Middleware } from 'koa'; + +import { HttpError } from '@forestadmin/forestadmin-client'; +import { bodyParser } from '@koa/bodyparser'; +import Koa from 'koa'; +import request from 'supertest'; + +import { createHttpTransport } from '../../src/agent/agent-transport'; +import createDataRoutesMiddleware from '../../src/data/data-routes-middleware'; +import createErrorMiddleware from '../../src/http/error-middleware'; +import ReadModel from '../../src/read-model/read-model'; +import { + ACTIVITY_LOG_ID, + ACTIVITY_LOG_INDEX, + API_KEY_SERVER_TOKEN, + RENDERING_ID, + activityLogsOf, + apiKeyCredentials, + fakeActivityLogsService, + forestServerTokenStep, + oauthCredentials, + sessionAccessToken, +} from '../helpers/activity-log'; +import { collection, column, relation } from '../read-model/fixtures'; + +const AGENT_URL = 'https://agent.example.com'; +const TIMEZONE = 'Europe/Paris'; +const OPERATORS = ['present', 'blank', 'equal', 'not_equal', 'in', 'like']; +const EMAIL_FILTER = { field: 'email', operator: 'Equal', value: 'joe@example.com' }; +const TITLE_FILTER = { field: 'title', operator: 'Equal', value: 'hello' }; + +const noopLogger: Logger = () => undefined; + +const readModel = new ReadModel([ + collection('users', [column('id'), column('email'), relation('posts', 'HasMany', 'posts.id')]), + collection('posts', [column('id'), column('title')]), +]); + +function storeOf(): ReadModelStore { + return { + getReadModel: async () => readModel, + getCapabilities: async () => ({ + capabilities: { + fields: ['id', 'email', 'title'].map(name => ({ + name, + type: 'String', + operators: OPERATORS, + })), + }, + readModel, + }), + } as unknown as ReadModelStore; +} + +function buildApp({ + service, + client, + credentials = apiKeyCredentials(), + saasAccessToken, + logger = noopLogger, +}: { + service: ReturnType; + client: Partial; + credentials?: Middleware; + saasAccessToken?: string; + logger?: Logger; +}): { app: Koa; activityLogs: ActivityLogWriter } { + const activityLogs = activityLogsOf(service, logger); + const app = new Koa(); + app.silent = true; + app.use(createErrorMiddleware({ logger: noopLogger })); + app.use(bodyParser()); + app.use(credentials); + app.use(forestServerTokenStep(saasAccessToken)); + app.use(async (ctx, next) => { + ctx.state.timezone = TIMEZONE; + ctx.state.agentToken = 'agent-jwt'; + await next(); + }); + app.use( + createDataRoutesMiddleware({ + store: storeOf(), + transport: createHttpTransport({ agentUrl: AGENT_URL }), + logger, + activityLogs, + createClient: () => client as AgentDataClient, + }), + ); + + return { app, activityLogs }; +} + +describe('data routes activity log', () => { + describe('when listing records', () => { + it('should record a search when the body carries a search and a filter', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { list: async () => [] } }); + + const response = await request(app.callback()) + .post('/agent/v1/users/list') + .send({ search: 'joe', filter: EMAIL_FILTER }); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).toHaveBeenCalledWith({ + forestServerToken: API_KEY_SERVER_TOKEN, + renderingId: String(RENDERING_ID), + action: 'search', + type: 'read', + collectionName: 'users', + recordId: undefined, + recordIds: undefined, + label: undefined, + }); + }); + + it('should record a filter when the body carries a filter and no search', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { list: async () => [] } }); + + const response = await request(app.callback()) + .post('/agent/v1/users/list') + .send({ filter: EMAIL_FILTER }); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ action: 'filter', type: 'read', collectionName: 'users' }), + ); + }); + + it('should record a filter when the search is blank, which the agent never receives', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { list: async () => [] } }); + + const response = await request(app.callback()) + .post('/agent/v1/users/list') + .send({ search: ' ', filter: EMAIL_FILTER }); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ action: 'filter' }), + ); + }); + + it('should record an index when the search is blank and nothing else refines the list', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { list: async () => [] } }); + + const response = await request(app.callback()) + .post('/agent/v1/users/list') + .send({ search: ' ' }); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ action: 'index' }), + ); + }); + + it('should record an index when the filter is empty, which refines nothing', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { list: async () => [] } }); + + const response = await request(app.callback()) + .post('/agent/v1/users/list') + .send({ filter: {} }); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ action: 'index', type: 'read' }), + ); + }); + + it('should record an index when the body carries neither a search nor a filter', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { list: async () => [] } }); + + const response = await request(app.callback()).post('/agent/v1/users/list').send({}); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ action: 'index', type: 'read' }), + ); + }); + + it('should mark the log completed once the records are served', async () => { + const service = fakeActivityLogsService(); + const { app, activityLogs } = buildApp({ service, client: { list: async () => [] } }); + + await request(app.callback()).post('/agent/v1/users/list').send({}); + await activityLogs.drain(); + + expect(service.updateActivityLogStatus).toHaveBeenCalledWith({ + forestServerToken: API_KEY_SERVER_TOKEN, + activityLog: { id: ACTIVITY_LOG_ID, attributes: { index: ACTIVITY_LOG_INDEX } }, + status: 'completed', + }); + }); + + it('should mark the log failed when the agent refuses the list', async () => { + const service = fakeActivityLogsService(); + const { app, activityLogs } = buildApp({ + service, + client: { + list: async () => { + throw new Error('agent is down'); + }, + }, + }); + + const response = await request(app.callback()).post('/agent/v1/users/list').send({}); + await activityLogs.drain(); + + expect(response.status).toBe(502); + expect(service.updateActivityLogStatus).toHaveBeenCalledWith( + expect.objectContaining({ status: 'failed' }), + ); + }); + + it('should serve the records and report once that the log could not be created', async () => { + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => { + throw new Error('the audit store is down'); + }), + }); + const logger = jest.fn(); + const list = jest.fn(async () => []); + const { app } = buildApp({ service, client: { list }, logger }); + + const response = await request(app.callback()).post('/agent/v1/users/list').send({}); + + expect(response.status).toBe(200); + expect(list).toHaveBeenCalledTimes(1); + expect(logger).toHaveBeenCalledWith( + 'Error', + "Activity log for 'index' could not be created", + expect.objectContaining({ cause: 'Error: the audit store is down' }), + ); + expect(logger).not.toHaveBeenCalledWith('Warn', expect.stringContaining('Activity log')); + }); + + it('should refuse the list when the audit endpoint rejects the identity', async () => { + const service = fakeActivityLogsService({ + createMcpActivityLog: jest.fn(async () => { + throw new HttpError('Forbidden', 403); + }), + }); + const list = jest.fn(async () => []); + const { app } = buildApp({ service, client: { list } }); + + const response = await request(app.callback()).post('/agent/v1/users/list').send({}); + + expect(response.status).toBe(403); + expect(response.body.error.type).toBe('audit_not_authorized'); + expect(list).not.toHaveBeenCalled(); + }); + + it('should serve the records unaudited when the oauth session cannot be resolved', async () => { + const service = fakeActivityLogsService(); + const list = jest.fn(async () => []); + const { app } = buildApp({ + service, + client: { list }, + credentials: oauthCredentials(), + saasAccessToken: undefined, + }); + + const response = await request(app.callback()).post('/agent/v1/users/list').send({}); + + expect(response.status).toBe(200); + expect(list).toHaveBeenCalledTimes(1); + expect(service.createMcpActivityLog).not.toHaveBeenCalled(); + }); + + it('should use the session token when the caller carries an oauth session', async () => { + const service = fakeActivityLogsService(); + const saasAccessToken = sessionAccessToken(); + const { app } = buildApp({ + service, + client: { list: async () => [] }, + credentials: oauthCredentials(), + saasAccessToken, + }); + + await request(app.callback()).post('/agent/v1/users/list').send({}); + + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ forestServerToken: saasAccessToken }), + ); + }); + }); + + describe('when listing a relation', () => { + it('should record the parent record and label the refinements it was given', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { listRelation: async () => [] } }); + + const response = await request(app.callback()) + .post('/agent/v1/users/relations/posts/list') + .send({ parentId: 'users-1', search: 'hello', filter: TITLE_FILTER }); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ + action: 'listRelatedData', + type: 'read', + collectionName: 'users', + recordId: 'users-1', + label: 'list relation "posts" with search and filter', + }), + ); + }); + + it('should label a relation list carrying only a search', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { listRelation: async () => [] } }); + + await request(app.callback()) + .post('/agent/v1/users/relations/posts/list') + .send({ parentId: 'users-1', search: 'hello' }); + + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ label: 'list relation "posts" with search' }), + ); + }); + + it('should leave a blank search out of the label, like the outgoing query does', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { listRelation: async () => [] } }); + + await request(app.callback()) + .post('/agent/v1/users/relations/posts/list') + .send({ parentId: 'users-1', search: ' ', filter: TITLE_FILTER }); + + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ label: 'list relation "posts" with filter' }), + ); + }); + + it('should leave an empty filter out of the label, like the outgoing query does', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { listRelation: async () => [] } }); + + await request(app.callback()) + .post('/agent/v1/users/relations/posts/list') + .send({ parentId: 'users-1', filter: {} }); + + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ label: 'list relation "posts"' }), + ); + }); + + it('should label a plain relation list without a refinement suffix', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { listRelation: async () => [] } }); + + await request(app.callback()) + .post('/agent/v1/users/relations/posts/list') + .send({ parentId: 'users-1' }); + + expect(service.createMcpActivityLog).toHaveBeenCalledWith( + expect.objectContaining({ label: 'list relation "posts"' }), + ); + }); + }); + + describe('when counting records', () => { + it('should write no log for a count', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ service, client: { countRaw: async () => ({ count: 3 }) } }); + + const response = await request(app.callback()) + .post('/agent/v1/users/count') + .send({ filter: EMAIL_FILTER }); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).not.toHaveBeenCalled(); + }); + + it('should write no log for a relation count', async () => { + const service = fakeActivityLogsService(); + const { app } = buildApp({ + service, + client: { countRelationRaw: async () => ({ count: 1 }) }, + }); + + const response = await request(app.callback()) + .post('/agent/v1/users/relations/posts/count') + .send({ parentId: 'users-1' }); + + expect(response.status).toBe(200); + expect(service.createMcpActivityLog).not.toHaveBeenCalled(); + }); + }); +}); diff --git a/packages/agent-bff/test/data/data-routes-middleware.test.ts b/packages/agent-bff/test/data/data-routes-middleware.test.ts index 0974f66b78..3224a673f0 100644 --- a/packages/agent-bff/test/data/data-routes-middleware.test.ts +++ b/packages/agent-bff/test/data/data-routes-middleware.test.ts @@ -13,6 +13,7 @@ import createDataRoutesMiddleware from '../../src/data/data-routes-middleware'; import createErrorMiddleware from '../../src/http/error-middleware'; import SchemaUnavailableError from '../../src/read-model/errors'; import ReadModel from '../../src/read-model/read-model'; +import { passthroughActivityLogs } from '../helpers/activity-log'; import { collection, column, polymorphic, relation } from '../read-model/fixtures'; const TRANSPORT = createHttpTransport({ agentUrl: 'https://agent.example.com' }); @@ -81,6 +82,7 @@ function buildApp( store, transport: TRANSPORT, logger, + activityLogs: passthroughActivityLogs(), createClient, }), ); @@ -159,6 +161,7 @@ describe('data routes middleware', () => { store: storeOf(usersReadModel), transport: TRANSPORT, logger: noopLogger, + activityLogs: passthroughActivityLogs(), createClient, }), ); diff --git a/packages/agent-bff/test/data/fixtures/legacy-agent-harness.ts b/packages/agent-bff/test/data/fixtures/legacy-agent-harness.ts index bc5d88d888..b0e3de5100 100644 --- a/packages/agent-bff/test/data/fixtures/legacy-agent-harness.ts +++ b/packages/agent-bff/test/data/fixtures/legacy-agent-harness.ts @@ -15,6 +15,7 @@ import createErrorMiddleware from '../../../src/http/error-middleware'; import CapabilitiesCache from '../../../src/read-model/capabilities-cache'; import ReadModelStore from '../../../src/read-model/read-model-store'; import SchemaCache from '../../../src/read-model/schema-cache'; +import { passthroughActivityLogs } from '../../helpers/activity-log'; export const AUTH_SECRET = 'b0bdf0a639c16bae8851dd24ee3d79ef0a352e957c5b86cb'; @@ -136,6 +137,7 @@ export function buildLegacyApp(agentUrl: string, { liana }: { liana?: string } = store, transport: createHttpTransport({ agentUrl }), logger: noopLogger, + activityLogs: passthroughActivityLogs(), }), ); diff --git a/packages/agent-bff/test/helpers/action-routes.ts b/packages/agent-bff/test/helpers/action-routes.ts index bdc07cf9db..9d3e5e7296 100644 --- a/packages/agent-bff/test/helpers/action-routes.ts +++ b/packages/agent-bff/test/helpers/action-routes.ts @@ -5,6 +5,7 @@ import type ReadModelStore from '../../src/read-model/read-model-store'; import { bodyParser } from '@koa/bodyparser'; import Koa from 'koa'; +import { passthroughActivityLogs } from './activity-log'; import createActionRoutesMiddleware from '../../src/action/action-routes-middleware'; import { createHttpTransport } from '../../src/agent/agent-transport'; import createErrorMiddleware from '../../src/http/error-middleware'; @@ -125,6 +126,7 @@ export function buildApp( store, transport: createHttpTransport({ agentUrl: 'https://agent.example.com' }), logger, + activityLogs: passthroughActivityLogs(), createClient: () => client, }), ); @@ -151,6 +153,7 @@ export function buildAppWithTerminal(client: AgentActionClient) { store: storeOf(readModel), transport: createHttpTransport({ agentUrl: 'https://agent.example.com' }), logger: noopLogger, + activityLogs: passthroughActivityLogs(), createClient: () => client, }), ); diff --git a/packages/agent-bff/test/helpers/activity-log.ts b/packages/agent-bff/test/helpers/activity-log.ts new file mode 100644 index 0000000000..b89547306a --- /dev/null +++ b/packages/agent-bff/test/helpers/activity-log.ts @@ -0,0 +1,96 @@ +import type { + ActivityLogWriter, + RecordActivityLogOptions, +} from '../../src/activity-log/activity-log-writer'; +import type { ActivityLogsWriter } from '../../src/activity-log/activity-logs-service'; +import type ForestServerClient from '../../src/oauth/forest-server-client'; +import type { SessionStore } from '../../src/oauth/session-store'; +import type { Logger } from '../../src/ports/logger-port'; +import type { Middleware } from 'koa'; + +import jsonwebtoken from 'jsonwebtoken'; + +import createActivityLogWriter from '../../src/activity-log/activity-log-writer'; +import createForestServerTokenMiddleware from '../../src/auth/forest-server-token-middleware'; + +export const ACTIVITY_LOG_ID = 'log-1'; +export const ACTIVITY_LOG_INDEX = 'activity-logs-2024'; +export const API_KEY_SERVER_TOKEN = 'api-key-server-token'; +export const RENDERING_ID = 42; +export const SESSION_ID = 'sid-1'; + +export interface FakeActivityLogsService extends ActivityLogsWriter { + createMcpActivityLog: jest.Mock; + updateActivityLogStatus: jest.Mock; +} + +export function fakeActivityLogsService( + overrides: Partial = {}, +): FakeActivityLogsService { + return { + createMcpActivityLog: jest.fn(async () => ({ + id: ACTIVITY_LOG_ID, + attributes: { index: ACTIVITY_LOG_INDEX }, + })), + updateActivityLogStatus: jest.fn(async () => undefined), + ...overrides, + } as FakeActivityLogsService; +} + +export function activityLogsOf(service: ActivityLogsWriter, logger: Logger): ActivityLogWriter { + return createActivityLogWriter({ service, logger }); +} + +export function passthroughActivityLogs(): ActivityLogWriter { + return { + record(options: RecordActivityLogOptions): Promise { + return options.operation(); + }, + + drain(): Promise { + return Promise.resolve([]); + }, + }; +} + +export function sessionAccessToken(): string { + return jsonwebtoken.sign({ scope: 'forest' }, 'session-secret', { expiresIn: '15m' }); +} + +export function sessionStoreOf(saasAccessToken: string | undefined): SessionStore { + return { + get: (sid: string) => + sid === SESSION_ID && saasAccessToken !== undefined ? { saasAccessToken } : undefined, + } as unknown as SessionStore; +} + +export const unusedServerClient = {} as ForestServerClient; + +export function apiKeyCredentials( + forestServerToken: string | undefined = API_KEY_SERVER_TOKEN, +): Middleware { + return async function stubApiKeyCredentials(ctx, next) { + ctx.state.authMode = 'api-key'; + ctx.state.apiKeyIdentity = { renderingId: RENDERING_ID }; + ctx.state.forestServerToken = forestServerToken; + await next(); + }; +} + +export function oauthCredentials(): Middleware { + return async function stubOAuthCredentials(ctx, next) { + ctx.state.authMode = 'oauth'; + ctx.state.principal = { sid: SESSION_ID, rendering_id: String(RENDERING_ID) }; + await next(); + }; +} + +export function forestServerTokenStep( + saasAccessToken?: string, + logger: Logger = () => undefined, +): Middleware { + return createForestServerTokenMiddleware({ + session: { store: sessionStoreOf(saasAccessToken), serverClient: unusedServerClient }, + logger, + }); +} diff --git a/packages/agent-bff/test/http/bff-http-server.test.ts b/packages/agent-bff/test/http/bff-http-server.test.ts index c5ad7f0314..fd8fda5f5f 100644 --- a/packages/agent-bff/test/http/bff-http-server.test.ts +++ b/packages/agent-bff/test/http/bff-http-server.test.ts @@ -21,15 +21,30 @@ const VALID_ENV = { const noopLogger = () => undefined; +const SHUTDOWN_DEADLINE_MS = 20; + const teapot: BffCallback = (req, res) => { res.statusCode = 418; res.end(); }; -function createServer(env: NodeJS.ProcessEnv, port = 0, logger: Logger = noopLogger) { +function createServer( + env: NodeJS.ProcessEnv, + port = 0, + logger: Logger = noopLogger, + drainActivityLogs?: (timeoutMs?: number) => Promise, + shutdownTimeoutMs?: number, +) { const config = parseConfig(env); - return new BFFHttpServer({ port, version: VERSION, config, logger }); + return new BFFHttpServer({ + port, + version: VERSION, + config, + logger, + drainActivityLogs, + shutdownTimeoutMs, + }); } function createPrebuiltServer(env: NodeJS.ProcessEnv, logger: Logger = noopLogger) { @@ -268,6 +283,122 @@ describe('BFFHttpServer', () => { }); }); + describe('when stopping a server that writes activity logs', () => { + it('should drain the pending status transitions after closing the connections', async () => { + const events: string[] = []; + const server = createServer({ ...VALID_ENV }, 0, noopLogger, async () => { + events.push('drain'); + + return []; + }); + await server.start(); + (server as unknown as { server: Server }).server.on('close', () => events.push('close')); + + await server.stop(); + + expect(events).toEqual(['close', 'drain']); + }); + + it('should destroy the connections outliving the deadline and still drain', async () => { + const drain = jest.fn(async () => [] as string[]); + const logger = jest.fn(); + const server = createServer({ ...VALID_ENV }, 0, logger, drain, SHUTDOWN_DEADLINE_MS); + await server.start(); + + const internal = (server as unknown as { server: Server }).server; + const closeIdleConnections = jest.spyOn(internal, 'closeIdleConnections'); + const closeAllConnections = jest.spyOn(internal, 'closeAllConnections'); + jest.spyOn(internal, 'close').mockImplementation((() => internal) as Server['close']); + + await expect(server.stop()).resolves.toBeUndefined(); + + expect(closeIdleConnections).toHaveBeenCalledTimes(1); + expect(closeAllConnections).toHaveBeenCalledTimes(1); + expect(drain).toHaveBeenCalledTimes(1); + expect(logger).toHaveBeenCalledWith( + 'Warn', + 'Forcing the Forest BFF shutdown: connections were still open', + { timeoutMs: SHUTDOWN_DEADLINE_MS }, + ); + + jest.restoreAllMocks(); + await closeServer(internal); + }); + + it('should hand the drain what the connections left of the shutdown deadline', async () => { + const drain = jest.fn(async () => [] as string[]); + const server = createServer({ ...VALID_ENV }, 0, noopLogger, drain, SHUTDOWN_DEADLINE_MS); + await server.start(); + + const internal = (server as unknown as { server: Server }).server; + jest.spyOn(internal, 'close').mockImplementation((() => internal) as Server['close']); + + await server.stop(); + + expect(drain).toHaveBeenCalledWith(0); + + jest.restoreAllMocks(); + await closeServer(internal); + }); + + it('should name the activity logs the deadline left in flight', async () => { + const unfinished = ["'completed' transition of the activity log log-1"]; + const drain = jest.fn(async () => unfinished); + const logger = jest.fn(); + const server = createServer({ ...VALID_ENV }, 0, logger, drain, SHUTDOWN_DEADLINE_MS); + await server.start(); + + await server.stop(); + + expect(logger).toHaveBeenCalledWith( + 'Warn', + 'Stopped the Forest BFF with activity logs still in flight', + { timeoutMs: SHUTDOWN_DEADLINE_MS, unfinished }, + ); + }); + + it('should say nothing once everything drained', async () => { + const logger = jest.fn(); + const server = createServer( + { ...VALID_ENV }, + 0, + logger, + async () => [], + SHUTDOWN_DEADLINE_MS, + ); + await server.start(); + + await server.stop(); + + expect(logger).not.toHaveBeenCalledWith( + 'Warn', + 'Stopped the Forest BFF with activity logs still in flight', + expect.anything(), + ); + }); + + it('should not drain when the connections could not be closed', async () => { + const drain = jest.fn(async () => [] as string[]); + const server = createServer({ ...VALID_ENV }, 0, noopLogger, drain); + await server.start(); + + const closeError = new Error('close failed'); + const internal = (server as unknown as { server: Server }).server; + jest.spyOn(internal, 'close').mockImplementation(((cb: (err?: Error) => void) => { + cb(closeError); + + return internal; + }) as Server['close']); + + await expect(server.stop()).rejects.toBe(closeError); + + expect(drain).not.toHaveBeenCalled(); + + jest.restoreAllMocks(); + await closeServer(internal); + }); + }); + describe('when the underlying server fails to close', () => { it('should reject with the close error', async () => { const server = createServer({ ...VALID_ENV }); diff --git a/packages/agent-bff/test/http/bff-local-errors.test.ts b/packages/agent-bff/test/http/bff-local-errors.test.ts index 39b7dfefb0..66cb7f4111 100644 --- a/packages/agent-bff/test/http/bff-local-errors.test.ts +++ b/packages/agent-bff/test/http/bff-local-errors.test.ts @@ -1,5 +1,7 @@ import { actionNotAllowed, + auditNotAuthorized, + auditUnavailable, collectionNotAllowed, invalidRequest, mappingError, @@ -27,10 +29,19 @@ describe('bff local errors', () => { [unsupportedActionResult, 'unsupported_action_result', 501], [openapiDisabled, 'openapi_disabled', 404], [streamingUnsupported, 'streaming_unsupported', 501], + [auditNotAuthorized, 'audit_not_authorized', 403], ])('%p builds a %s error with status %d', (factory, type, status) => { expect(factory()).toMatchObject({ type, status }); }); + it('carries the retry delay on auditUnavailable', () => { + expect(auditUnavailable(5)).toMatchObject({ + type: 'audit_unavailable', + status: 503, + retryAfter: 5, + }); + }); + it('carries details on invalidRequest', () => { expect(invalidRequest('bad', { field: 'x' })).toMatchObject({ type: 'invalid_request', diff --git a/packages/agent-bff/test/http/error-contract.test.ts b/packages/agent-bff/test/http/error-contract.test.ts index af65a93255..9c541d6461 100644 --- a/packages/agent-bff/test/http/error-contract.test.ts +++ b/packages/agent-bff/test/http/error-contract.test.ts @@ -47,7 +47,12 @@ function buildEdge(authenticate: ApiKeyAuthenticator['authenticate']) { app.use(bodyParser({ jsonLimit: '16kb' })); app.use(createErrorMiddleware({ logger })); app.use(createAuthModeMiddleware({ authSecret: AUTH_SECRET })); - app.use(createApiKeyMiddleware({ authenticator: { authenticate }, logger })); + app.use( + createApiKeyMiddleware({ + authenticator: { authenticate, invalidate: () => undefined }, + logger, + }), + ); app.use(createPerKeyOriginMiddleware({ logger, serverAllowedOrigins: [] })); app.use(createTimezoneMiddleware({ defaultTimezone: undefined })); app.use(createAgentStubMiddleware()); diff --git a/packages/agent-bff/test/openapi/openapi-generated-client.test.ts b/packages/agent-bff/test/openapi/openapi-generated-client.test.ts index 44c72ecff0..bc5a4f8f34 100644 --- a/packages/agent-bff/test/openapi/openapi-generated-client.test.ts +++ b/packages/agent-bff/test/openapi/openapi-generated-client.test.ts @@ -18,6 +18,7 @@ import createDataRoutesMiddleware from '../../src/data/data-routes-middleware'; import createErrorMiddleware from '../../src/http/error-middleware'; import ReadModel from '../../src/read-model/read-model'; import createTimezoneMiddleware, { TIMEZONE_HEADER } from '../../src/timezone/timezone-middleware'; +import { passthroughActivityLogs } from '../helpers/activity-log'; import { action, collection, column, relation } from '../read-model/fixtures'; const MARK_AS_PAID = 'Mark as paid'; @@ -189,6 +190,7 @@ function buildApp(): Koa { store, transport: createHttpTransport({ agentUrl: ENV.AGENT_URL }), logger: noopLogger, + activityLogs: passthroughActivityLogs(), createClient: () => dataClient, }), ); @@ -197,6 +199,7 @@ function buildApp(): Koa { store, transport: createHttpTransport({ agentUrl: ENV.AGENT_URL }), logger: noopLogger, + activityLogs: passthroughActivityLogs(), createClient: () => actionClient, }), ); diff --git a/packages/agent-bff/test/rate-limit/rate-limit-middleware.test.ts b/packages/agent-bff/test/rate-limit/rate-limit-middleware.test.ts index 30580a28d9..4a97a406e4 100644 --- a/packages/agent-bff/test/rate-limit/rate-limit-middleware.test.ts +++ b/packages/agent-bff/test/rate-limit/rate-limit-middleware.test.ts @@ -48,7 +48,12 @@ function buildEdge( app.silent = true; app.use(createErrorMiddleware({ logger: () => undefined })); app.use(createAuthModeMiddleware({ authSecret: AUTH_SECRET })); - app.use(createApiKeyMiddleware({ authenticator: { authenticate }, logger: () => undefined })); + app.use( + createApiKeyMiddleware({ + authenticator: { authenticate, invalidate: () => undefined }, + logger: () => undefined, + }), + ); app.use(limiter); app.use(async ctx => { ctx.status = 200; diff --git a/packages/agent/src/agent.ts b/packages/agent/src/agent.ts index 37558cab62..8af63a4218 100644 --- a/packages/agent/src/agent.ts +++ b/packages/agent/src/agent.ts @@ -177,7 +177,7 @@ export default class Agent extends FrameworkMounter override async stop(): Promise { // Stop answering before the stack it dispatches into goes away: the host application keeps // whatever middleware it registered, so a stopped agent would otherwise still serve BFF data. - this.embeddedBff?.stop(); + await this.embeddedBff?.stop(); // Drain the embedded executor next, while the agent it depends on is still serving. await this.embeddedExecutor?.stop(); // Close anything related to ForestAdmin client diff --git a/packages/agent/src/embedded-bff.ts b/packages/agent/src/embedded-bff.ts index a92e0883dc..b64ed66a5c 100644 --- a/packages/agent/src/embedded-bff.ts +++ b/packages/agent/src/embedded-bff.ts @@ -3,6 +3,9 @@ import type { AgentDispatcher, BFFConfig, Bff } from '@forestadmin/agent-bff'; import { BFF_PREFIX, stripBffPrefix } from './bff-routes'; +/** How long `stop()` waits for the activity-log writes no connection holds. */ +export const SHUTDOWN_TIMEOUT_MS = 10_000; + /** * Serialize the BFF's structured log context onto the message: the agent's logger only accepts an * Error as its third argument, so the context would be dropped otherwise. Errors are unfolded by @@ -157,10 +160,35 @@ export default class EmbeddedBff { /** * Stop answering. The host application keeps whatever middleware it registered, so without this * a stopped agent would go on serving BFF data through a dispatcher pointing at a dead stack. + * + * Drains before returning: the activity-log status transitions are fired without `await`, so + * nothing else holds them and a shutdown would leave the entries `pending`. + * + * The drain is bounded, like the standalone server's. Here the host owns the connections, so + * there is no connection deadline to share and the budget is the caller's alone: unbounded, a + * stalled audit store would hold the process until its orchestrator sends SIGKILL, and the + * drain would not finish anyway. What the deadline leaves unfinished is logged by name. */ - stop(): void { + async stop(): Promise { + const { bff } = this; + this.bff = null; this.stopped = true; + + if (!bff?.drainActivityLogs) return; + + const timeoutMs = this.embedOptions.shutdownTimeoutMs ?? SHUTDOWN_TIMEOUT_MS; + const unfinished = await bff.drainActivityLogs(timeoutMs); + + if (unfinished.length === 0) return; + + this.options.logger( + 'Warn', + formatLog('Stopped the embedded BFF with activity logs still in flight', { + timeoutMs, + unfinished, + }), + ); } /** diff --git a/packages/agent/src/types.ts b/packages/agent/src/types.ts index d8e096f0b1..9b8391c6b6 100644 --- a/packages/agent/src/types.ts +++ b/packages/agent/src/types.ts @@ -188,6 +188,15 @@ export type BffEmbedOptions = { * exposed collection, relation and field. */ openapiEnabled?: boolean; + /** + * How long `stop()` waits for the activity-log writes still in flight. Defaults to 10s. + * + * Embedded, the host owns the connections, so `stop()` is reached while audited requests are + * still running and their status transitions are not registered yet. Unbounded, one stalled + * audit store would hold the process until its orchestrator sends SIGKILL, which is worse than + * the entries the deadline leaves `pending` — those are logged by name when it expires. + */ + shutdownTimeoutMs?: number; }; // Runtime view of `auditTrail`: the validator has built the SQL store from the connection string. diff --git a/packages/agent/test/agent-bff-lifecycle.test.ts b/packages/agent/test/agent-bff-lifecycle.test.ts index d0136fa55a..afe3313889 100644 --- a/packages/agent/test/agent-bff-lifecycle.test.ts +++ b/packages/agent/test/agent-bff-lifecycle.test.ts @@ -191,6 +191,80 @@ describe('the embedded BFF lifecycle', () => { message: 'The embedded BFF was stopped with the agent.', }); }); + + it('should drain the activity log transitions no connection holds', async () => { + const drainActivityLogs = jest.fn(async () => []); + mockBuildBff.mockResolvedValue({ + callback: mockBffCallback, + invalidate: mockInvalidate, + drainActivityLogs, + }); + const agent = buildAgent().addBff(); + await agent.start(); + + await agent.stop(); + + expect(drainActivityLogs).toHaveBeenCalledTimes(1); + }); + + it('should stop cleanly when the deployment writes no activity log', async () => { + const agent = buildAgent().addBff(); + await agent.start(); + + await expect(agent.stop()).resolves.toBeUndefined(); + }); + + it('should bound the drain, so a stalled audit store cannot hold the process', async () => { + const drainActivityLogs = jest.fn(async () => []); + mockBuildBff.mockResolvedValue({ + callback: mockBffCallback, + invalidate: mockInvalidate, + drainActivityLogs, + }); + const agent = buildAgent().addBff(); + await agent.start(); + + await agent.stop(); + + expect(drainActivityLogs).toHaveBeenCalledWith(10_000); + }); + + it('should give the drain the deadline the host asked for', async () => { + const drainActivityLogs = jest.fn(async () => []); + mockBuildBff.mockResolvedValue({ + callback: mockBffCallback, + invalidate: mockInvalidate, + drainActivityLogs, + }); + const agent = buildAgent().addBff({ shutdownTimeoutMs: 2_000 }); + await agent.start(); + + await agent.stop(); + + expect(drainActivityLogs).toHaveBeenCalledWith(2_000); + }); + + it('should name what the deadline left in flight, rather than drop it silently', async () => { + const logger = jest.fn(); + const drainActivityLogs = jest.fn(async () => ["'action' request on 'books'"]); + mockBuildBff.mockResolvedValue({ + callback: mockBffCallback, + invalidate: mockInvalidate, + drainActivityLogs, + }); + const agent = new Agent( + factories.forestAdminHttpDriverOptions.build({ skipSchemaUpdate: true, logger }), + ).addBff(); + await agent.start(); + + await agent.stop(); + + expect(logger).toHaveBeenCalledWith( + 'Warn', + '[BFF] Stopped the embedded BFF with activity logs still in flight ' + + '{"timeoutMs":10000,"unfinished":["\'action\' request on \'books\'"]}', + ); + }); }); describe('when stop() lands while the BFF is still being built', () => {