resolved issues with mediator registration

Author: brett-onions

src/index.ts

@@ -3,7 +3,7 @@ import path from 'path';
 import { getConfig } from './config/config';
 import logger from './logger';
 import routes from './routes/index';
-import { getRegisterBuckets, setupMediator } from './openhim/openhim';
+import { getRegisteredBuckets, setupMediator } from './openhim/openhim';
 import { createMinioBucketListeners, ensureBucketExists } from './utils/minioClient';
 
 const app = express();
@@ -14,10 +14,10 @@ app.listen(getConfig().port, async () => {
   logger.info(`Server is running on port - ${getConfig().port}`);
 
   if (getConfig().runningMode !== 'testing' && getConfig().registerMediator) {
-    setupMediator();
+    await setupMediator();
   }
 
-  const buckets = await getRegisterBuckets();
+  const buckets = await getRegisteredBuckets();
 
   buckets.length === 0 && logger.warn('No buckets specified in the configuration');
 

src/openhim/mediatorConfig.json

@@ -10,7 +10,7 @@
       "routes": [
         {
           "name": "Climate Mediator Endpoint",
-          "host": "172.17.0.1",
+          "host": "climate-mediator",
           "port": "3000",
           "primary": true,
           "type": "http"
@@ -29,7 +29,7 @@
   "endpoints": [
     {
       "name": "Climate Endpoint",
-      "host": "172.17.0.1",
+      "host": "climate-mediator",
       "path": "/test",
       "port": "3000",
       "primary": true,
@@ -57,5 +57,8 @@
         }
       ]
     }
-  ]
+  ],
+  "config": {
+    "minio_buckets_registry": []
+  }
 }

src/openhim/openhim.ts

@@ -1,16 +1,11 @@
 import logger from '../logger';
+import { MinioBucketsRegistry, Mediator as OpenHimAPIMediator } from '../types';
 import { MediatorConfig } from '../types/mediatorConfig';
 import { RequestOptions } from '../types/request';
 import { getConfig } from '../config/config';
 import axios, { AxiosError } from 'axios';
 import https from 'https';
-import {
-  activateHeartbeat,
-  fetchConfig,
-  registerMediator,
-  authenticate,
-  genAuthHeaders,
-} from 'openhim-mediator-utils';
+import { activateHeartbeat, fetchConfig, registerMediator } from 'openhim-mediator-utils';
 import { Bucket, createMinioBucketListeners, ensureBucketExists } from '../utils/minioClient';
 import path from 'path';
 
@@ -23,7 +18,7 @@ const resolveMediatorConfig = (): MediatorConfig => {
   let mediatorConfigFile;
 
   try {
-    logger.info(`Loading mediator config from: ${mediatorConfigFilePath}`);
+    logger.debug(`Loading mediator config from: ${mediatorConfigFilePath}`);
     mediatorConfigFile = require(mediatorConfigFilePath);
   } catch (error) {
     logger.error(`Failed to parse JSON: ${error}`);
@@ -43,12 +38,12 @@ const resolveOpenhimConfig = (urn: string): RequestOptions => {
   };
 };
 
-export const setupMediator = () => {
+export const setupMediator = async () => {
   try {
     const mediatorConfig = resolveMediatorConfig();
     const openhimConfig = resolveOpenhimConfig(mediatorConfig.urn);
 
-    registerMediator(openhimConfig, mediatorConfig, (error: Error) => {
+    await registerMediator(openhimConfig, mediatorConfig, (error: Error) => {
       if (error) {
         logger.error(`Failed to register mediator: ${JSON.stringify(error)}`);
         throw error;
@@ -87,8 +82,8 @@ export const setupMediator = () => {
 };
 
 //TODO: Add Typing and error handling.
-async function getMediatorConfig() {
-  logger.info('Fetching mediator config from OpenHIM');
+async function getMediatorConfig(): Promise<OpenHimAPIMediator | null> {
+  logger.debug('Fetching mediator config from OpenHIM');
   const mediatorConfig = resolveMediatorConfig();
   const openhimConfig = resolveOpenhimConfig(mediatorConfig.urn);
 
@@ -108,17 +103,65 @@ async function getMediatorConfig() {
     return request.data;
   } catch (e) {
     const error = e as AxiosError;
-    logger.error(`Failed to fetch mediator config: ${JSON.stringify(error)}`);
-    error.status === 404 && logger.warn('Mediator config not found in OpenHIM, ');
+
+    switch (error.status) {
+      case 401:
+        logger.error(`Failed to authenticate with OpenHIM, check your credentials`);
+        break;
+      case 404:
+        logger.debug(
+          'Mediator config not found in OpenHIM, This is expected on initial setup'
+        );
+        break;
+      default:
+        logger.error(`Failed to fetch mediator config: ${JSON.stringify(error)}`);
+        break;
+    }
     return null;
   }
 }
 
-export async function getRegisterBuckets(): Promise<Bucket[]> {
+async function putMediatorConfig(mediatorUrn: string, mediatorConfig: MinioBucketsRegistry[]) {
+  const openhimConfig = resolveOpenhimConfig(mediatorUrn);
+  const { apiURL, username, password, trustSelfSigned } = openhimConfig;
+  await axios.put(
+    `${apiURL}/mediators/urn:mediator:climate-mediator/config`,
+    {
+      minio_buckets_registry: mediatorConfig,
+    },
+    {
+      auth: { username, password },
+      httpsAgent: new https.Agent({
+        rejectUnauthorized: !trustSelfSigned,
+      }),
+    }
+  );
+
+  try {
+    logger.info('Successfully updated mediator config in OpenHIM');
+  } catch (error) {
+    const axiosError = error as AxiosError;
+    switch (axiosError.status) {
+      case 401:
+        logger.error(`Failed to authenticate with OpenHIM, check your credentials`);
+        break;
+      default:
+        logger.error(
+          `Failed to update mediator config in OpenHIM: ${JSON.stringify(axiosError)}`
+        );
+        break;
+    }
+  }
+}
+
+export async function getRegisteredBuckets(): Promise<Bucket[]> {
   if (runningMode !== 'testing') {
     logger.info('Fetching registered buckets from OpenHIM');
     const mediatorConfig = await getMediatorConfig();
 
+    if (!mediatorConfig) {
+      return [];
+    }
     //TODO: Handle errors, and undefined response.
     const buckets = mediatorConfig.config?.minio_buckets_registry as Bucket[];
     if (!buckets) {
@@ -133,13 +176,49 @@ export async function getRegisterBuckets(): Promise<Bucket[]> {
 }
 
 export async function registerBucket(bucket: string, region?: string) {
-  if (runningMode !== 'testing') {
+  // If we are in testing mode, we don't need to have the registered buckets persisted
+  if (runningMode === 'testing') {
+    logger.debug('Running in testing mode, skipping bucket registration');
     return true;
   }
+
+  //get the mediator config from OpenHIM
   const mediatorConfig = await getMediatorConfig();
-  const existingBuckets = mediatorConfig.config?.minio_buckets_registry;
 
-  if (!existingBuckets) {
-    return [];
+  //TODO: Change this to a debug log
+  logger.debug(`Mediator config: ${JSON.stringify(mediatorConfig)}`);
+
+  //if the mediator config is not found, log the issue and return false
+  if (mediatorConfig === null) {
+    logger.error('Mediator config not found in OpenHIM, unable to register bucket');
+    return false;
   }
+
+  const newBucket = {
+    bucket,
+    region: region || '',
+  };
+
+  //get the existing buckets from the mediator config
+  const existingConfig = mediatorConfig.config;
+
+  if (existingConfig === undefined) {
+    logger.info('Mediator config does not have a config section, creating new config');
+    mediatorConfig['config'] = {
+      minio_buckets_registry: [newBucket],
+    };
+  } else {
+    const existingBucket = existingConfig.minio_buckets_registry.find(
+      (bucket) => bucket.bucket === newBucket.bucket
+    );
+    if (existingBucket) {
+      logger.debug(`Bucket ${bucket} already exists in the config`);
+      return false;
+    }
+    logger.info(`Adding bucket ${bucket} to OpenHIM config`);
+    existingConfig.minio_buckets_registry.push(newBucket);
+    await putMediatorConfig(mediatorConfig.urn, existingConfig.minio_buckets_registry);
+  }
+
+  return true;
 }

src/routes/index.ts

@@ -5,8 +5,12 @@ import { getCsvHeaders } from '../utils/file-validators';
 import logger from '../logger';
 import fs from 'fs/promises';
 import path from 'path';
-import e from 'express';
-import { uploadToMinio } from '../utils/minioClient';
+import {
+  BucketDoesNotExistError,
+  ensureBucketExists,
+  uploadToMinio,
+} from '../utils/minioClient';
+import { registerBucket } from '../openhim/openhim';
 
 // Constants
 const VALID_MIME_TYPES = ['text/csv', 'application/json'] as const;
@@ -77,7 +81,6 @@ const handleCsvFile = async (
       fileUrl,
       file.originalname,
       bucket,
-      region,
       file.mimetype
     );
     await fs.unlink(fileUrl);
@@ -104,6 +107,7 @@ routes.post('/upload', upload.single('file'), async (req, res) => {
     const file = req.file;
     const bucket = req.query.bucket as string;
     const region = req.query.region as string;
+    const createBucketIfNotExists = req.query.createBucketIfNotExists === 'true';
 
     if (!file) {
       logger.error('No file uploaded');
@@ -115,21 +119,31 @@ routes.post('/upload', upload.single('file'), async (req, res) => {
       return res.status(400).json(createErrorResponse('BUCKET_MISSING', 'No bucket provided'));
     }
 
+    await ensureBucketExists(bucket, region, createBucketIfNotExists);
+
     const response =
       file.mimetype === 'text/csv'
         ? await handleCsvFile(file, bucket, region)
         : handleJsonFile(file);
 
+    createBucketIfNotExists && (await registerBucket(bucket, region));
+
     const statusCode = response.status === 'success' ? 201 : 400;
     return res.status(statusCode).json(response);
-  } catch (error) {
-    logger.error('Error processing upload:', error);
+  } catch (e) {
+    logger.error('Error processing upload:', e);
+
+    if (e instanceof BucketDoesNotExistError) {
+      const error = e as BucketDoesNotExistError;
+      return res.status(400).json(createErrorResponse('BUCKET_DOES_NOT_EXIST', error.message));
+    }
+
     return res
       .status(500)
       .json(
         createErrorResponse(
           'INTERNAL_SERVER_ERROR',
-          error instanceof Error ? error.message : 'Unknown error'
+          e instanceof Error ? e.message : 'Unknown error'
         )
       );
   }

src/types/index.ts

@@ -0,0 +1 @@
+export * from './openhim-api/mediator';

src/types/openhim-api/mediator.ts

@@ -0,0 +1,79 @@
+export interface Mediator {
+  _id: string;
+  urn: string;
+  version: string;
+  name: string;
+  description: string;
+  endpoints: Endpoint[];
+  defaultChannelConfig: DefaultChannelConfig[];
+  configDefs: ConfigDef[];
+  __v: number;
+  _lastHeartbeat: Date;
+  _uptime: number;
+  _configModifiedTS: Date;
+  config?: Config;
+}
+
+interface Config {
+  minio_buckets_registry: MinioBucketsRegistry[];
+}
+
+export interface MinioBucketsRegistry {
+  bucket: string;
+  region?: string;
+}
+
+interface ConfigDef {
+  param: string;
+  displayName: string;
+  description: string;
+  type: string;
+  values: any[];
+  template: Template[];
+  array: boolean;
+  _id: string;
+}
+
+interface Template {
+  param: string;
+  displayName: string;
+  type: string;
+  optional?: boolean;
+}
+
+interface DefaultChannelConfig {
+  name: string;
+  urlPattern: string;
+  isAsynchronousProcess: boolean;
+  methods: string[];
+  type: string;
+  allow: string[];
+  whitelist: any[];
+  authType: string;
+  routes: Endpoint[];
+  matchContentTypes: any[];
+  properties: any[];
+  txViewAcl: any[];
+  txViewFullAcl: any[];
+  txRerunAcl: any[];
+  status: string;
+  rewriteUrls: boolean;
+  addAutoRewriteRules: boolean;
+  autoRetryEnabled: boolean;
+  autoRetryPeriodMinutes: number;
+  _id: string;
+  alerts: any[];
+  rewriteUrlsConfig: any[];
+}
+
+interface Endpoint {
+  name: string;
+  type: string;
+  status: string;
+  host: string;
+  port: number;
+  primary: boolean;
+  forwardAuthHeader: boolean;
+  _id: string;
+  path?: string;
+}