feat(agent-bff): fetch and cache agent schema, capabilities and allow-list

packages/agent-bff/package.json

@@ -31,6 +31,7 @@
     "test": "jest"
   },
   "dependencies": {
+    "@forestadmin/agent-client": "1.9.0",
     "@forestadmin/forestadmin-client": "1.40.3",
     "@koa/bodyparser": "^6.1.0",
     "jsonwebtoken": "^9.0.3",

packages/agent-bff/src/adapters/console-metrics.ts

@@ -0,0 +1,15 @@
+import type { Logger } from '../ports/logger-port';
+import type { MetricTags, Metrics } from '../ports/metrics-port';
+
+import createConsoleLogger from './console-logger';
+
+export default function createConsoleMetrics(logger: Logger = createConsoleLogger()): Metrics {
+  return {
+    increment(name: string, tags?: MetricTags): void {
+      logger('Info', 'metric.increment', { metric: name, ...(tags ?? {}) });
+    },
+    gauge(name: string, value: number, tags?: MetricTags): void {
+      logger('Info', 'metric.gauge', { metric: name, value, ...(tags ?? {}) });
+    },
+  };
+}

packages/agent-bff/src/ports/metrics-port.ts

@@ -0,0 +1,6 @@
+export type MetricTags = Record<string, string | number>;
+
+export interface Metrics {
+  increment(name: string, tags?: MetricTags): void;
+  gauge(name: string, value: number, tags?: MetricTags): void;
+}

packages/agent-bff/src/read-model/action-endpoint-resolver.ts

@@ -0,0 +1,55 @@
+import type ReadModel from './read-model';
+import type { Metrics } from '../ports/metrics-port';
+import type { ActionEndpointsByCollection } from '@forestadmin/agent-client';
+
+export const ACTION_ENDPOINT_MISS = 'action_endpoint_miss';
+export const ACTION_ENDPOINT_ERROR = 'action_endpoint_error';
+
+export type ActionEndpointInfo = ActionEndpointsByCollection[string][string];
+
+export type ReadModelProvider = () => Promise<ReadModel>;
+
+export interface ResolveActionContext {
+  rendering: string | number;
+}
+
+/**
+ * Resolves an action to its endpoint against the current read-model. Never throws: an absent
+ * mapping emits the miss counter, a failure to obtain the read-model emits the error counter.
+ * Both counters carry `rendering`, `collection`, and `action` tags. In normal flow the action
+ * allow-list already excludes endpoint-less actions, so the miss path is a defensive guard.
+ */
+export default class ActionEndpointResolver {
+  constructor(
+    private readonly getReadModel: ReadModelProvider,
+    private readonly metrics: Metrics,
+  ) {}
+
+  async resolve(
+    collection: string,
+    action: string,
+    { rendering }: ResolveActionContext,
+  ): Promise<ActionEndpointInfo | undefined> {
+    const tags = { rendering, collection, action };
+
+    let readModel: ReadModel;
+
+    try {
+      readModel = await this.getReadModel();
+    } catch {
+      this.metrics.increment(ACTION_ENDPOINT_ERROR, tags);
+
+      return undefined;
+    }
+
+    const info = readModel.getActionEndpoints()[collection]?.[action];
+
+    if (!info) {
+      this.metrics.increment(ACTION_ENDPOINT_MISS, tags);
+
+      return undefined;
+    }
+
+    return info;
+  }
+}

packages/agent-bff/src/read-model/agent-capabilities-fetcher.ts

@@ -0,0 +1,21 @@
+import type { CapabilitiesFetcher } from './capabilities-cache';
+
+import { createRemoteAgentClient } from '@forestadmin/agent-client';
+
+export interface AgentCapabilitiesFetcherOptions {
+  agentUrl: string;
+  token: string;
+}
+
+/**
+ * Builds a capabilities fetcher bound to a request's agent token. The cache calls it only on a
+ * miss, so the token of whichever request first populates a collection is the one used.
+ */
+export default function createAgentCapabilitiesFetcher({
+  agentUrl,
+  token,
+}: AgentCapabilitiesFetcherOptions): CapabilitiesFetcher {
+  const client = createRemoteAgentClient({ url: agentUrl, token });
+
+  return collection => client.collection(collection).capabilities();
+}

packages/agent-bff/src/read-model/capabilities-cache.ts

@@ -0,0 +1,68 @@
+import { ONE_DAY_MS } from './schema-cache';
+
+export interface CapabilitiesResult {
+  fields: { name: string; type: string; operators: string[] }[];
+}
+
+export type CapabilitiesFetcher = (collection: string) => Promise<CapabilitiesResult>;
+
+export interface CapabilitiesCacheOptions {
+  now?: () => number;
+  ttlMs?: number;
+}
+
+interface CacheEntry {
+  result: CapabilitiesResult;
+  fetchedAt: number;
+}
+
+/**
+ * Per-collection capabilities cache (24h). The fetcher is passed per call so it can be bound to
+ * the caller's request token. Invalidated together with the schema via `clear()`.
+ */
+export default class CapabilitiesCache {
+  private readonly now: () => number;
+  private readonly ttlMs: number;
+
+  private readonly entries = new Map<string, CacheEntry>();
+  private readonly inFlight = new Map<string, Promise<CapabilitiesResult>>();
+  private generation = 0;
+
+  constructor({ now = Date.now, ttlMs = ONE_DAY_MS }: CapabilitiesCacheOptions = {}) {
+    this.now = now;
+    this.ttlMs = ttlMs;
+  }
+
+  async get(collection: string, fetcher: CapabilitiesFetcher): Promise<CapabilitiesResult> {
+    const entry = this.entries.get(collection);
+    if (entry && this.now() - entry.fetchedAt < this.ttlMs) return entry.result;
+
+    const pending = this.inFlight.get(collection);
+    if (pending) return pending;
+
+    const { generation } = this;
+    const fetch = fetcher(collection)
+      .then(result => {
+        // Skip the write if a clear() (schema refresh) happened while this fetch was in flight —
+        // its result belongs to the old schema generation and must not repopulate the cache.
+        if (this.generation === generation) {
+          this.entries.set(collection, { result, fetchedAt: this.now() });
+        }
+
+        return result;
+      })
+      .finally(() => {
+        if (this.inFlight.get(collection) === fetch) this.inFlight.delete(collection);
+      });
+
+    this.inFlight.set(collection, fetch);
+
+    return fetch;
+  }
+
+  clear(): void {
+    this.generation += 1;
+    this.entries.clear();
+    this.inFlight.clear();
+  }
+}

packages/agent-bff/src/read-model/create-read-model.ts

@@ -0,0 +1,42 @@
+import type { Logger } from '../ports/logger-port';
+import type { Metrics } from '../ports/metrics-port';
+
+import ActionEndpointResolver from './action-endpoint-resolver';
+import CapabilitiesCache from './capabilities-cache';
+import ForestSchemaClient from './forest-schema-client';
+import ReadModelStore from './read-model-store';
+import SchemaCache from './schema-cache';
+import createConsoleMetrics from '../adapters/console-metrics';
+
+export interface CreateReadModelOptions {
+  forestServerUrl: string;
+  envSecret: string;
+  metrics?: Metrics;
+  logger?: Logger;
+  now?: () => number;
+}
+
+export interface ReadModelBundle {
+  store: ReadModelStore;
+  actionEndpointResolver: ActionEndpointResolver;
+}
+
+export default function createReadModel({
+  forestServerUrl,
+  envSecret,
+  metrics,
+  logger,
+  now,
+}: CreateReadModelOptions): ReadModelBundle {
+  const resolvedMetrics = metrics ?? createConsoleMetrics(logger);
+  const fetcher = new ForestSchemaClient({ forestServerUrl, envSecret });
+  const schemaCache = new SchemaCache({ fetcher, metrics: resolvedMetrics, now });
+  const capabilitiesCache = new CapabilitiesCache({ now });
+  const store = new ReadModelStore(schemaCache, capabilitiesCache);
+  const actionEndpointResolver = new ActionEndpointResolver(
+    () => store.getReadModel(),
+    resolvedMetrics,
+  );
+
+  return { store, actionEndpointResolver };
+}

packages/agent-bff/src/read-model/errors.ts

@@ -0,0 +1,9 @@
+export default class SchemaUnavailableError extends Error {
+  readonly cause?: unknown;
+
+  constructor(cause?: unknown) {
+    super('The agent schema is unavailable');
+    this.name = 'SchemaUnavailableError';
+    this.cause = cause;
+  }
+}

packages/agent-bff/src/read-model/forest-schema-client.ts

@@ -0,0 +1,24 @@
+import type { ForestSchemaCollection } from '@forestadmin/forestadmin-client';
+
+import { ForestHttpApi, SchemaService } from '@forestadmin/forestadmin-client';
+
+export interface ForestSchemaClientOptions {
+  forestServerUrl: string;
+  envSecret: string;
+}
+
+export interface SchemaFetcher {
+  fetchSchema(): Promise<ForestSchemaCollection[]>;
+}
+
+export default class ForestSchemaClient implements SchemaFetcher {
+  private readonly schemaService: SchemaService;
+
+  constructor({ forestServerUrl, envSecret }: ForestSchemaClientOptions) {
+    this.schemaService = new SchemaService(new ForestHttpApi(), { forestServerUrl, envSecret });
+  }
+
+  async fetchSchema(): Promise<ForestSchemaCollection[]> {
+    return this.schemaService.getSchema();
+  }
+}

packages/agent-bff/src/read-model/read-model-store.ts

@@ -0,0 +1,54 @@
+import type CapabilitiesCache from './capabilities-cache';
+import type { CapabilitiesFetcher, CapabilitiesResult } from './capabilities-cache';
+import type SchemaCache from './schema-cache';
+
+import ReadModel from './read-model';
+
+/**
+ * Single owner of the coupled schema + capabilities lifecycle. A successful schema refresh (a bump
+ * of the cache `revision`) rebuilds the read-model and clears capabilities atomically, so the
+ * allow-list and capabilities never split-brain across schema generations.
+ */
+export default class ReadModelStore {
+  private readonly schemaCache: SchemaCache;
+  private readonly capabilitiesCache: CapabilitiesCache;
+
+  private builtRevision = -1;
+  private readModel: ReadModel | null = null;
+
+  constructor(schemaCache: SchemaCache, capabilitiesCache: CapabilitiesCache) {
+    this.schemaCache = schemaCache;
+    this.capabilitiesCache = capabilitiesCache;
+  }
+
+  async getReadModel(): Promise<ReadModel> {
+    const collections = await this.schemaCache.get();
+
+    if (this.schemaCache.revision !== this.builtRevision || !this.readModel) {
+      this.readModel = new ReadModel(collections);
+      this.builtRevision = this.schemaCache.revision;
+      this.capabilitiesCache.clear();
+    }
+
+    return this.readModel;
+  }
+
+  async getCapabilities(
+    collection: string,
+    fetcher: CapabilitiesFetcher,
+  ): Promise<CapabilitiesResult> {
+    // Ensure any pending schema refresh (and its capabilities invalidation) runs first.
+    await this.getReadModel();
+
+    // TODO(wiring): possible TOCTOU once this is called from request handling. A concurrent schema
+    // refresh can clear capabilities while this fetch is in flight, so the caller could receive
+    // capabilities from the previous schema generation alongside the new allow-list. When wiring
+    // the data endpoints, re-check `schemaCache.revision` after the fetch resolves and retry on a
+    // mismatch so capabilities and schema stay atomically coupled.
+    return this.capabilitiesCache.get(collection, fetcher);
+  }
+
+  ageSeconds(): number | undefined {
+    return this.schemaCache.ageSeconds();
+  }
+}