@onlineapps/infrastructure-tools

Infrastructure orchestration utilities for OA Drive infrastructure services.

Purpose

This library provides utilities for infrastructure services to coordinate their initialization, health tracking, and queue management. It is NOT intended for business services.

Installation

npm install @onlineapps/infrastructure-tools

Usage

Health Publisher

Publish health checks to infrastructure.health.checks queue:

const { createBaseClientAdapter } = require('@onlineapps/infrastructure-tools');
const BaseClient = require('@onlineapps/mq-client-core');

const mqClient = new BaseClient({ ... });
await mqClient.connect();

const healthPublisher = createBaseClientAdapter(
  mqClient,
  'gateway',
  () => ({
    mq: mqClient.isConnected() ? 'healthy' : 'unhealthy',
    redis: 'healthy',
    http: 'healthy'
  }),
  config.infrastructureHealth,
  logger
);

await healthPublisher.start();

Wait for Infrastructure Ready

Wait for all infrastructure services to be ready before creating queues:

const { waitForInfrastructureReady } = require('@onlineapps/infrastructure-tools');

await waitForInfrastructureReady({
  redisUrl: 'redis://api_node_cache:6379',
  maxWait: 300000, // 5 minutes
  checkInterval: 5000, // 5 seconds
  logger: logger
});

Note: waitForInfrastructureReady is re-exported from @onlineapps/service-common to avoid duplication. Both infrastructure and business services can use it.

Initialize Infrastructure Queues

Initialize infrastructure queues with correct parameters:

const { initInfrastructureQueues } = require('@onlineapps/infrastructure-tools');

await initInfrastructureQueues(channel, {
  queues: ['workflow.init'], // Only create specific queues
  connection: connection,
  serviceName: 'api-gateway',
  logger: logger
});

Storage Operations

Access MinIO storage for infrastructure services:

const { StorageCore } = require('@onlineapps/infrastructure-tools');

const storage = new StorageCore({
  // FAIL-FAST: topology must be explicit (no defaults for hosts/credentials)
  endPoint: process.env.MINIO_ENDPOINT,
  port: parseInt(process.env.MINIO_PORT, 10),
  accessKey: process.env.MINIO_ACCESS_KEY,
  secretKey: process.env.MINIO_SECRET_KEY
});

await storage.initialize();
await storage.ensureBucket('workflow');
await storage.putObject('workflow', 'path/to/file', buffer);

JWT Validation

Shared JWT validation for infrastructure services accepting client requests (gateway, delivery endpoint, meta). Standard: docs/standards/JWT_AUTH.md.

const {
  createJwtValidator,
  verifyAccessToken,
  extractTenantContext
} = require('@onlineapps/infrastructure-tools');

// Express middleware: validates Bearer token, sets req.auth
app.use('/api', createJwtValidator({
  logger,
  excludePaths: ['/health']
}));

// Pure function: verify token outside Express context (e.g. WebSocket handshake)
const decoded = verifyAccessToken(rawToken);

// Pure function: extract tenant context from auth + headers
const ctx = extractTenantContext(req.auth, req.headers);
// => { tenant_id, tenant_uuid, workspace_id, person_id, person_uuid, role }

Env: JWT_SECRET (min 16 chars, HMAC-SHA256). Fail-fast if missing.

API

waitForInfrastructureReady(options)

Waits for all infrastructure services to be reported as healthy by Registry.

Options:

• redisUrl (string): Redis URL
• maxWait (number): Maximum wait time in ms (default: 300000 = 5 minutes)
• checkInterval (number): Check interval in ms (default: 5000 = 5 seconds)
• logger (Object|Function): Required logger instance (or function). No implicit console fallback.

Returns: Promise<boolean>

initInfrastructureQueues(channel, options)

Initializes infrastructure queues with correct parameters from queueConfig.

Options:

• queues (Array): Specific queues to create (default: all infrastructure queues)
• connection (Object): RabbitMQ connection (for channel recreation)
• logger (Object): Logger instance (required)
• queueConfig (Object): Queue config instance (default: from mq-client-core)
• serviceName (string): Used in queue mismatch alerts (default: unknown-service)
• alertOnMismatch (boolean): Disable automatic 406 alerts (default: true)

Returns: Promise<void>

createHealthPublisher(options)

Creates a health publisher instance with custom publish function.

Options:

• serviceName (string): Service name (required)
• version (string): Service version (default: '1.0.0')
• publishFunction (Function): Function to publish message (required)
• getHealthData (Function): Function to get health data (required)
• config (Object): Configuration with infrastructureHealth settings
• logger (Object): Required logger instance (no implicit console fallback)

Returns: Health publisher instance with start(), stop(), publishNow() methods

createBaseClientAdapter(baseClient, serviceName, getHealthData, config, logger)

Creates health publisher adapter for BaseClient (from mq-client-core).

createAmqplibAdapter(connection, channel, serviceName, getHealthData, config, logger)

Creates health publisher adapter for amqplib (direct connection + channel).

verifyAccessToken(token)

Verify and decode an OA Drive access token (pure function).

Parameters:

• token (string): Raw JWT string (without "Bearer " prefix)

Returns: Decoded payload { sub, person_id, email, tenants, type, iss, iat, exp }

Throws:

• TokenExpiredError (from jsonwebtoken) if token is expired
• JsonWebTokenError if signature or issuer invalid
• Error with code: 'INVALID_TOKEN_TYPE' if token type is not access
• Error if JWT_SECRET env var is missing or too short

createJwtValidator({ logger, excludePaths })

Create Express middleware that validates JWT Bearer tokens.

Options:

• logger (Object): Logger with .warn() method (required)
• excludePaths (string[]): Paths to skip validation (default: [])

Behavior: Sets req.auth = { person_uuid, person_id, email, tenants } on success. Returns 401 JSON on failure.

extractTenantContext(auth, headers)

Extract tenant context from JWT auth data and request headers (pure function).

Parameters:

• auth (Object): req.auth from createJwtValidator: { person_uuid, person_id, email, tenants }
• headers (Object): HTTP request headers (lowercase keys)

Returns: { tenant_id, tenant_uuid, workspace_id, person_id, person_uuid, role }

Throws:

• Error with statusCode: 400 if multiple tenants and x-tenant-uuid header missing
• Error with statusCode: 403 if specified tenant not in memberships
• Error with statusCode: 401 if auth data is missing

StorageCore

Re-exported from @onlineapps/storage-core for infrastructure services.

Provides core MinIO storage operations:

• putObject(bucket, path, data, metadata?) - Upload object
• getObject(bucket, path) - Download object
• objectExists(bucket, path) - Check if object exists
• ensureBucket(bucket, region?) - Ensure bucket exists
• calculateFingerprint(content) - Generate SHA256 fingerprint
• getContentType(filename) - Get MIME type from filename
• getPresignedUrl(bucket, path, expiry?) - Generate presigned URL

See @onlineapps/storage-core for full API documentation.

Dependencies

• @onlineapps/mq-client-core - For queue configuration
• @onlineapps/service-common - For waitForInfrastructureReady (shared utility)
• @onlineapps/storage-core - For core MinIO storage operations
• jsonwebtoken - For JWT verification (HMAC-SHA256)

Related Libraries

• @onlineapps/mq-client-core - Core MQ operations and queue configuration
• @onlineapps/monitoring-core - Business monitoring (traces, metrics, logs)
• @onlineapps/storage-core - Core MinIO storage operations (re-exported here for infrastructure services)

License

MIT