@nebulae/tpiv2-interop-server

Toolkit de resiliencia del lado servidor para handlers de µAPIs en la plantilla de microservicios de la Nebula Engineering.

Proporciona 7 funciones independientes y componibles para proteger los handlers RxJS con idempotencia, circuit breaker, métricas, reintentos y construcción de respuestas CQRS — sin introducir RxJS como dependencia propia del paquete.

Instalación

npm install @nebulae/tpiv2-interop-server

Referencia de la API

checkIdempotency<T>(cache, key): Promise<void>

Verifica si ya existe una respuesta cacheada para la clave dada.

• Si la clave existe y no ha expirado → lanza IdempotencyError con error.previousResponse.
• Si no existe → resuelve sin valor (continuar ejecución normal).

import { checkIdempotency, IdempotencyCache } from '@nebulae/tpiv2-interop-server';

const cache = new IdempotencyCache({ ttlMs: 60_000 });
await checkIdempotency(cache, requestKey); // throws on duplicate

storeIdempotencyResult<T>(cache, key, result): void

Almacena el resultado de una operación exitosa en la caché. Llamar después de confirmar el éxito (típicamente en tap).

import { storeIdempotencyResult } from '@nebulae/tpiv2-interop-server';

tap(response => storeIdempotencyResult(cache, requestKey, response))

guardWithCircuitBreaker<T>(cb, fn, correlationId?): Promise<T>

Ejecuta fn bajo la protección de un CircuitBreaker.

• Si el circuito está OPEN → lanza CircuitOpenError sin llamar a fn.
• Si fn falla → registra la falla y relanza el error original.
• Si fn tiene éxito → registra el éxito y retorna el resultado.

import { guardWithCircuitBreaker, CircuitBreaker } from '@nebulae/tpiv2-interop-server';

const cb = new CircuitBreaker({ failureThreshold: 3, halfOpenTimeoutMs: 30_000 });

from(guardWithCircuitBreaker(cb, () =>
  AggregateTypeDA.getData$().toPromise()
))

emitApiMetric(collector, event): void

Emite un evento de métrica al colector configurado, inyectando el timestamp automáticamente.
Si collector es undefined, la función es un no-op — no requiere condicionales en el handler.

import { emitApiMetric } from '@nebulae/tpiv2-interop-server';

// En tap de éxito:
tap(() => emitApiMetric(collector, {
  type: 'request',
  client: 'server',
  endpoint: 'aggregatetype/listing',
  durationMs: Date.now() - start,
}))

// En catchError:
catchError(err => {
  emitApiMetric(collector, {
    type: 'error',
    client: 'server',
    endpoint: 'aggregatetype/listing',
    errorType: err.name,
    durationMs: Date.now() - start,
  });
  return of(handleError(err));
})

withRetry<T>(config, fn, correlationId?): Promise<T>

Ejecuta fn con reintentos automáticos usando backoff exponencial.
Lanza MaxRetriesExceededError cuando se agotan los intentos, con el error original en error.details.originalError.

import { withRetry } from '@nebulae/tpiv2-interop-server';

from(withRetry(
  { maxRetries: 3, initialDelay: 200 },
  () => ExternalServiceDA.fetchData$().toPromise(),
  correlationId
))

buildSuccessResponse<T>(data): CqrsResponse<T>

Equivalente síncrono de CqrsResponseHelper.buildSuccessResponse$. Devuelve el objeto de respuesta de éxito que el gateway espera, listo para envolver con map().

import { buildSuccessResponse } from '@nebulae/tpiv2-interop-server';

// en pipeline RxJS:
map(([listing, count]) => buildSuccessResponse({ listing, count }))

// fuera de RxJS:
const response = buildSuccessResponse(result);

El objeto retornado tiene la forma { result: { code: 200, message: 'OK' }, data: T }.

handleError(err, location?): CqrsResponse<null>

Equivalente síncrono de CqrsResponseHelper.handleError$. Mapea automáticamente los errores del SDK a códigos HTTP y devuelve el objeto de error que el gateway espera.

import { handleError } from '@nebulae/tpiv2-interop-server';

// en catchError de RxJS:
catchError(err => of(handleError(err)))

// con localización del endpoint:
catchError(err => of(handleError(err, 'aggregatetype/listing')))

El objeto retornado tiene la forma { result: { code: number, message: string, error: { name, code, location? } }, data: null }.

Mapeo de errores del SDK:

| Clase de error | Código HTTP | |---|---| | AuthenticationError | 401 | | AuthorizationError | 403 | | ValidationError | 400 | | TimeoutError | 504 | | CircuitOpenError | 503 | | MaxRetriesExceededError | 503 | | IdempotencyError | 409 | | RateLimitError | 429 | | ConnectionError | 502 | | ProtocolError | 502 | | ConfigurationError | 500 | | Cualquier otro error | 500 |

Patrón de uso completo en un handler

const { from, forkJoin, iif, throwError, of } = require('rxjs');
const { mergeMap, map, tap, catchError, toArray } = require('rxjs/operators');
const {
  checkIdempotency,
  storeIdempotencyResult,
  guardWithCircuitBreaker,
  emitApiMetric,
  buildSuccessResponse,
  handleError,
  IdempotencyCache,
  CircuitBreaker,
} = require('@nebulae/tpiv2-interop-server');

// Instancias singleton por handler (una vez al levantar el servicio)
const idempotencyCache = new IdempotencyCache({ ttlMs: 60_000 });
const circuitBreaker   = new CircuitBreaker({ failureThreshold: 3, halfOpenTimeoutMs: 30_000 });

// Handler
getAggregateTypeListing$(data, authToken, message) {
  const start        = Date.now();
  const correlationId = message?.id ?? authToken?.sub;
  const key           = `listing-${authToken.sub}-${JSON.stringify(data.query ?? {})}`;

  return from(checkIdempotency(idempotencyCache, key)).pipe(
    mergeMap(() =>
      from(guardWithCircuitBreaker(circuitBreaker, () =>
        forkJoin([
          AggregateTypeDA.getAggregateTypeList$(data.filterInput, data.paginationInput)
            .pipe(toArray()),
          AggregateTypeDA.getAggregateTypeSize$(data.filterInput),
        ]).toPromise(),
        correlationId
      ))
    ),
    map(([listing, count]) => buildSuccessResponse({ listing, count })),
    tap(response => {
      storeIdempotencyResult(idempotencyCache, key, response);
      emitApiMetric(collector, {
        type: 'request', client: 'server',
        endpoint: 'aggregatetype/listing',
        durationMs: Date.now() - start,
        correlationId,
      });
    }),
    catchError(err => {
      emitApiMetric(collector, {
        type: 'error', client: 'server',
        endpoint: 'aggregatetype/listing',
        errorType: err.name,
        durationMs: Date.now() - start,
        correlationId,
      });
      return iif(
        () => err.name === 'MongoDBTimeoutError',
        throwError(() => err),
        of(handleError(err, 'aggregatetype/listing'))
      );
    })
  );
}

Tipos re-exportados desde core

Para mayor comodidad, el paquete re-exporta los tipos y clases del core que necesitas:

| Símbolo | Descripción | |---|---| | IdempotencyCache | Caché LRU con TTL para idempotencia | | CircuitBreaker | Circuit breaker CLOSED→OPEN→HALF_OPEN | | RetryHandler | Motor de reintentos con backoff exponencial | | InteropError | Clase base de todos los errores del SDK | | AuthenticationError | Error de autenticación (HTTP 401) | | AuthorizationError | Error de autorización (HTTP 403) | | ValidationError | Error de validación de esquema (HTTP 400) | | TimeoutError | Error de timeout (HTTP 504) | | ConnectionError | Error de conexión (HTTP 502) | | ProtocolError | Error de protocolo SOAP/REST/X-Road (HTTP 502) | | ConfigurationError | Error de configuración (HTTP 500) | | IdempotencyError | Error lanzado en detección de duplicado (HTTP 409) | | CircuitOpenError | Error lanzado cuando el circuito está abierto (HTTP 503) | | MaxRetriesExceededError | Error lanzado al agotar los reintentos (HTTP 503) | | RateLimitError | Error de rate limiting (HTTP 429) | | CqrsResponse<T> | Tipo de respuesta CQRS que espera el gateway | | CqrsSuccessResult | Forma del campo result en respuestas exitosas | | CqrsErrorResult | Forma del campo result en respuestas de error | | CqrsErrorInfo | Forma del campo error dentro de CqrsErrorResult | | ServerMiddlewareConfig | Tipo auxiliar para configurar todos los mecanismos | | MetricsCollector | Interfaz para implementar un colector de métricas | | MetricEvent | Estructura del evento de métrica emitido |