listening and fetching configs from OpenHIM

Author: brett-onions

package-lock.json

@@ -41,6 +41,7 @@
   },
   "dependencies": {
     "@clickhouse/client": "^1.8.0",
+    "axios": "^1.7.7",
     "dotenv": "^16.4.5",
     "express": "^4.18.2",
     "express-async-handler": "^1.2.0",

package.json

@@ -1,25 +1,29 @@
 import express from 'express';
 import path from 'path';
-import { readFile } from 'fs/promises';
 import { getConfig } from './config/config';
 import logger from './logger';
 import routes from './routes/index';
-import { setupMediator } from './openhim/openhim';
-import {
-  createMinioBucketListeners,
-} from './utils/minioClient';
+import { getRegisterBuckets, setupMediator } from './openhim/openhim';
+import { createMinioBucketListeners, ensureBucketExists } from './utils/minioClient';
 
 const app = express();
 
 app.use('/', routes);
 
-createMinioBucketListeners();
-
-app.listen(getConfig().port, async() => {
+app.listen(getConfig().port, async () => {
   logger.info(`Server is running on port - ${getConfig().port}`);
 
   if (getConfig().runningMode !== 'testing' && getConfig().registerMediator) {
-    setupMediator(path.resolve(__dirname, './openhim/mediatorConfig.json'));
+    setupMediator();
   }
-});
 
+  const buckets = await getRegisterBuckets();
+
+  buckets.length === 0 && logger.warn('No buckets specified in the configuration');
+
+  for await (const { bucket, region } of buckets) {
+    await ensureBucketExists(bucket, region);
+  }
+
+  createMinioBucketListeners(buckets.map((bucket) => bucket.bucket));
+});

src/index.ts

@@ -20,7 +20,8 @@
         "instant"
       ],
       "methods": [
-        "POST", "GET"
+        "POST",
+        "GET"
       ],
       "type": "http"
     }
@@ -35,15 +36,26 @@
       "type": "http"
     }
   ],
-  "configDefs": [{
-    "param": "minio_buckets_registry",
+  "configDefs": [
+    {
+      "param": "minio_buckets_registry",
       "displayName": "Minio Buckets Registry",
       "description": "The available Minio buckets and their configurations",
       "type": "struct",
       "array": true,
-      "template":[{
-        "param": "bucket",
-        "type": "string"
-      }]
-  }]
-}
+      "template": [
+        {
+          "param": "bucket",
+          "displayName": "Bucket",
+          "type": "string"
+        },
+        {
+          "param": "region",
+          "displayName": "Region (optional)",
+          "type": "string",
+          "optional": true
+        }
+      ]
+    }
+  ]
+}

src/openhim/mediatorConfig.json

@@ -2,14 +2,28 @@ import logger from '../logger';
 import { MediatorConfig } from '../types/mediatorConfig';
 import { RequestOptions } from '../types/request';
 import { getConfig } from '../config/config';
-import { activateHeartbeat, fetchConfig, registerMediator } from 'openhim-mediator-utils';
+import axios, { AxiosError } from 'axios';
+import https from 'https';
+import {
+  activateHeartbeat,
+  fetchConfig,
+  registerMediator,
+  authenticate,
+  genAuthHeaders,
+} from 'openhim-mediator-utils';
+import { Bucket, createMinioBucketListeners, ensureBucketExists } from '../utils/minioClient';
+import path from 'path';
 
-const { openhimUsername, openhimPassword, openhimMediatorUrl, trustSelfSigned } = getConfig();
+const { openhimUsername, openhimPassword, openhimMediatorUrl, trustSelfSigned, runningMode } =
+  getConfig();
 
-const resolveMediatorConfig = (mediatorConfigFilePath: string): MediatorConfig => {
+const mediatorConfigFilePath = path.resolve(__dirname, './mediatorConfig.json');
+
+const resolveMediatorConfig = (): MediatorConfig => {
   let mediatorConfigFile;
 
   try {
+    logger.info(`Loading mediator config from: ${mediatorConfigFilePath}`);
     mediatorConfigFile = require(mediatorConfigFilePath);
   } catch (error) {
     logger.error(`Failed to parse JSON: ${error}`);
@@ -29,9 +43,9 @@ const resolveOpenhimConfig = (urn: string): RequestOptions => {
   };
 };
 
-export const setupMediator = (mediatorConfigFilePath: string) => {
+export const setupMediator = () => {
   try {
-    const mediatorConfig = resolveMediatorConfig(mediatorConfigFilePath);
+    const mediatorConfig = resolveMediatorConfig();
     const openhimConfig = resolveOpenhimConfig(mediatorConfig.urn);
 
     registerMediator(openhimConfig, mediatorConfig, (error: Error) => {
@@ -54,12 +68,78 @@ export const setupMediator = (mediatorConfigFilePath: string) => {
           logger.error(`Heartbeat failed: ${JSON.stringify(err)}`);
         });
 
-        emitter.on('config', (config: any) => {
-          logger.info(`Config: ${JSON.stringify(config)}`);
+        emitter.on('config', async (config: any) => {
+          logger.info('Received config from OpenHIM');
+
+          const buckets = config.minio_buckets_registry as Bucket[];
+
+          for await (const { bucket, region } of buckets) {
+            await ensureBucketExists(bucket, region);
+          }
+
+          createMinioBucketListeners(buckets.map((bucket) => bucket.bucket));
         });
       });
     });
   } catch (err) {
     logger.error('Unable to register mediator', err);
   }
 };
+
+//TODO: Add Typing and error handling.
+async function getMediatorConfig() {
+  logger.info('Fetching mediator config from OpenHIM');
+  const mediatorConfig = resolveMediatorConfig();
+  const openhimConfig = resolveOpenhimConfig(mediatorConfig.urn);
+
+  const { apiURL, urn, username, password, trustSelfSigned } = openhimConfig;
+
+  try {
+    const request = await axios.get(`${apiURL}/mediators/urn:mediator:climate-mediator`, {
+      auth: {
+        username,
+        password,
+      },
+      httpsAgent: new https.Agent({
+        rejectUnauthorized: !trustSelfSigned,
+      }),
+    });
+
+    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, ');
+    return null;
+  }
+}
+
+export async function getRegisterBuckets(): Promise<Bucket[]> {
+  if (runningMode !== 'testing') {
+    logger.info('Fetching registered buckets from OpenHIM');
+    const mediatorConfig = await getMediatorConfig();
+
+    //TODO: Handle errors, and undefined response.
+    const buckets = mediatorConfig.config?.minio_buckets_registry as Bucket[];
+    if (!buckets) {
+      return [];
+    }
+    return buckets;
+  } else {
+    logger.info('Running in testing mode, reading buckets from ENV');
+    const buckets = getConfig().minio.buckets.split(',');
+    return buckets.map((bucket) => ({ bucket, region: '' }));
+  }
+}
+
+export async function registerBucket(bucket: string, region?: string) {
+  if (runningMode !== 'testing') {
+    return true;
+  }
+  const mediatorConfig = await getMediatorConfig();
+  const existingBuckets = mediatorConfig.config?.minio_buckets_registry;
+
+  if (!existingBuckets) {
+    return [];
+  }
+}