@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);

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).

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

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