Merge branch 'main' of github.com:jembi/climate-mediator into add-endpoint-for-download

bradsawadye · bradsawadye · commit 67d993a0f332 · 2025-02-26T09:01:08.000+02:00
diff --git a/src/routes/index.ts b/src/routes/index.ts
@@ -46,7 +46,7 @@ const createSuccessResponse = (code: string, message: string): UploadResponse =>
   message,
 });
 
-const saveCsvToTmp = async (fileBuffer: Buffer, fileName: string): Promise<string> => {
+const saveToTmp = async (fileBuffer: Buffer, fileName: string): Promise<string> => {
   const tmpDir = path.join(process.cwd(), 'tmp');
   await fs.mkdir(tmpDir, { recursive: true });
 
@@ -77,7 +77,7 @@ const handleCsvFile = async (
     return createErrorResponse('INVALID_CSV_FORMAT', 'Invalid CSV file format');
   }
 
-  const fileUrl = await saveCsvToTmp(file.buffer, file.originalname);
+  const fileUrl = await saveToTmp(file.buffer, file.originalname);
   try {
     const uploadResult = await uploadToMinio(
       fileUrl,
@@ -96,11 +96,33 @@ const handleCsvFile = async (
   }
 };
 
-const handleJsonFile = (file: Express.Multer.File): UploadResponse => {
+const handleJsonFile = async (
+  file: Express.Multer.File,
+  bucket: string,
+  region: string
+): Promise<UploadResponse> => {
   if (!validateJsonFile(file.buffer)) {
     return createErrorResponse('INVALID_JSON_FORMAT', 'Invalid JSON file format');
   }
-  return createSuccessResponse('JSON_VALID', 'JSON file is valid - Future implementation');
+
+  const jsonString = file.buffer.toString().replace(/\n/g, '').replace(/\r/g, '');
+  const fileUrl = await saveToTmp(Buffer.from(jsonString), file.originalname);
+  try {
+    const uploadResult = await uploadToMinio(
+      fileUrl,
+      file.originalname,
+      bucket,
+      file.mimetype
+    );
+    await fs.unlink(fileUrl);
+
+    return uploadResult.success
+      ? createSuccessResponse('UPLOAD_SUCCESS', uploadResult.message)
+      : createErrorResponse('UPLOAD_FAILED', uploadResult.message);
+  } catch (error) {
+    logger.error('Error uploading file to Minio:', error);
+    throw error;
+  }
 };
 
 // Main route handler
@@ -138,7 +160,7 @@ routes.post('/upload', upload.single('file'), async (req, res) => {
     const response =
       file.mimetype === 'text/csv'
         ? await handleCsvFile(file, bucket, region)
-        : handleJsonFile(file);
+        : await handleJsonFile(file, bucket, region);
 
     if (createBucketIfNotExists && getConfig().runningMode !== 'testing') {
       await registerBucket(bucket, region);
diff --git a/src/utils/clickhouse.ts b/src/utils/clickhouse.ts
@@ -41,31 +41,55 @@ export async function createTable(fields: string[], tableName: string) {
   return true;
 }
 
+export async function createTableFromJson(
+  s3Path: string,
+  s3Config: { accessKey: string; secretKey: string },
+  tableName: string,
+  key: string
+) {
+  const client = createClient({
+    url,
+    password,
+  });
+
+  const normalizedTableName = tableName.replace(/-/g, '_');
+
+  //check if the table exists
+  try {
+    const existsResult = await client.query({
+      query: `desc ${normalizedTableName}`,
+    });
+    logger.info(`Table ${normalizedTableName} already exists`);
+    await client.close();
+    return false;
+  } catch (error) {
+    logger.error(`Table ${normalizedTableName} does not exist`);
+  }
+
+  const query = generateDDLFromJson(s3Path, s3Config, normalizedTableName, key);
+  await client.query({ query });
+  await client.close();
+}
+
 export function generateDDL(fields: string[], tableName: string) {
   return `CREATE TABLE ${tableName} (table_id UUID DEFAULT generateUUIDv4(),${fields.map((field) => `${field} VARCHAR`).join(', ')}) ENGINE = MergeTree ORDER BY (table_id)`;
 }
 
-export function flattenJson(json: any, prefix = ''): string[] {
-  const fields: string[] = [];
-  Object.keys(json).forEach((key) => {
-    const value = json[key];
-    if (typeof value === 'object') {
-      if (key === 'main') {
-        fields.push(...flattenJson(value));
-      } else {
-        // This is to avoid having a prefix starting with numbers
-        if (Array.isArray(json)) {
-          fields.push(...flattenJson(value, prefix));
-        } else {
-          fields.push(...flattenJson(value, `${key}_`));
-        }
-      }
-    } else {
-      fields.push(`${prefix}${key}`);
-    }
-  });
-  const fieldsSet = new Set(fields);
-  return Array.from(fieldsSet);
+export function generateDDLFromJson(
+  s3Path: string,
+  s3Config: { accessKey: string; secretKey: string },
+  tableName: string,
+  key: string
+) {
+  const query = `
+  CREATE TABLE IF NOT EXISTS \`default\`.${tableName}
+  ENGINE = MergeTree
+  ORDER BY ${key} EMPTY
+  AS SELECT * 
+  FROM s3('${s3Path}', '${s3Config.accessKey}', '${s3Config.secretKey}', JSONEachRow)
+  SETTINGS schema_inference_make_columns_nullable = 0
+  `;
+  return query;
 }
 
 export async function insertFromS3(
@@ -107,3 +131,41 @@ export async function insertFromS3(
     await client.close();
   }
 }
+
+export async function insertFromS3Json(
+  tableName: string,
+  s3Path: string,
+  s3Config: {
+    accessKey: string;
+    secretKey: string;
+  }
+) {
+  const client = createClient({
+    url,
+    password,
+  });
+
+  const normalizedTableName = tableName.replace(/-/g, '_');
+
+  try {
+    logger.debug(`Inserting data into ${normalizedTableName}`);
+    const query = `
+      INSERT INTO \`default\`.${normalizedTableName} 
+      SELECT * FROM s3(
+        '${s3Path}',
+        '${s3Config.accessKey}',
+        '${s3Config.secretKey}',
+        'JSONEachRow'
+      )
+    `;
+    await client.query({ query });
+    logger.info(`Successfully inserted data into ${normalizedTableName}`);
+    return true;
+  } catch (error) {
+    logger.error('Error inserting data from JSON');
+    logger.error(error);
+    return false;
+  } finally {
+    await client.close();
+  }
+}
diff --git a/src/utils/minioClient.ts b/src/utils/minioClient.ts
@@ -1,17 +1,22 @@
 import * as Minio from 'minio';
-import axios, { AxiosError } from 'axios';
+import axios from 'axios';
 
 import { getConfig } from '../config/config';
 import logger from '../logger';
 import crypto from 'crypto';
 import { readFile, rm } from 'fs/promises';
-import { createTable, flattenJson, insertFromS3 } from './clickhouse';
+import {
+  createTable,
+  createTableFromJson,
+  insertFromS3,
+  insertFromS3Json,
+} from './clickhouse';
 import { validateJsonFile, getCsvHeaders } from './file-validators';
 import { getOpenhimConfig, triggerProcessing } from '../openhim/openhim';
 import fs from 'fs/promises';
 import { createWriteStream } from 'fs';
 import path from 'path';
-import { timeStamp } from 'console';
+
 export interface Bucket {
   bucket: string;
   region?: string;
@@ -46,6 +51,23 @@ interface FileExistsResponse extends MinioResponse {
   exists: boolean;
 }
 
+/**
+ * Get the first field of a json object
+ * @param {any} json - The json object
+ * @returns {string} - The first field
+ */
+function getFirstField(json: any) {
+  let obj: any;
+  if (Array.isArray(json) && json.length > 0) {
+    obj = json[0];
+  } else {
+    obj = json;
+  }
+
+  const fields = Object.keys(obj);
+  return fields[0];
+}
+
 /**
  * Ensures a bucket exists, creates it if it doesn't
  * @param {string} bucket - Bucket name
@@ -217,32 +239,41 @@ export async function minioListenerHandler (bucket: string, file: string, tableN
   const extension = file.split('.').pop();
   logger.info(`File Downloaded - Type: ${extension}`);
 
-  if (['json', 'csv'].includes(extension as string)) {
-    let fields: string[] = [];
+  if (extension === 'json' && validateJsonFile(fileBuffer)) {
+    logger.info('File is a valid json file');
 
-    if (extension === 'json' && validateJsonFile(fileBuffer)) {
-      // flatten the json and pass it to clickhouse
-      fields = flattenJson(JSON.parse(fileBuffer.toString()));
-    } else if (getCsvHeaders(fileBuffer)) {
-      fields = (await readFile(`tmp/${file}`, 'utf8')).split('\n')[0].split(',');
-    }
+    // Construct the S3-style URL for the file
+    const minioUrl = `http://${endPoint}:${port}/${bucket}/${file}`;
 
-    if (fields.length) {
-      await createTable(fields, tableName);
+    const key = getFirstField(JSON.parse(fileBuffer.toString()));
 
-      // Construct the S3-style URL for the file
-      const minioUrl = `http://${endPoint}:${port}/${bucket}/${file}`;
+    // Create table from json
+    await createTableFromJson(minioUrl, { accessKey, secretKey }, tableName, key);
 
-      // Insert data into clickhouse
-      await insertFromS3(tableName, minioUrl, {
-        accessKey,
-        secretKey,
-      });
-    }
+    // Insert data into clickhouse
+    await insertFromS3Json(tableName, minioUrl, {
+      accessKey,
+      secretKey,
+    });
+  } else if (extension === 'csv' && getCsvHeaders(fileBuffer)) {
+    //get the first line of the csv file
+    const fields = (await readFile(`tmp/${file}`, 'utf8')).split('\n')[0].split(',');
+
+    await createTable(fields, tableName);
+
+    // If running locally and using docker compose, the minio host is 'minio'. This is to allow clickhouse to connect to the minio server
+
+    // Construct the S3-style URL for the file
+    const minioUrl = `http://${endPoint}:${port}/${bucket}/${file}`;
+
+    // Insert data into clickhouse
+    await insertFromS3(tableName, minioUrl, {
+      accessKey,
+      secretKey,
+    });
   } else {
     logger.warn(`Unknown file type - ${extension}`);
   }
-
   await rm(`tmp/${file}`);
   logger.debug(`File ${file} deleted from tmp directory`);
 }
diff --git a/tests/services/table-creation.test.ts b/tests/services/table-creation.test.ts
@@ -1,6 +1,6 @@
 import { expect } from 'chai';
 import { getCsvHeaders } from '../../src/utils/file-validators';
-import { createTable, flattenJson, generateDDL } from '../../src/utils/clickhouse';
+import { createTable, generateDDL } from '../../src/utils/clickhouse';
 
 describe('Create Tables based on files', function () {
   this.timeout(60_000);
@@ -20,13 +20,6 @@ describe('Create Tables based on files', function () {
     expect(fields).to.deep.equal(['id', 'name', 'age']);
   });
 
-  it('should extract columns based on a json file', async () => {
-    const jsonFile = Buffer.from('[{"id":1,"name":"John","age":20}]');
-    const json = JSON.parse(jsonFile.toString());
-    const fields = flattenJson(json[0]);
-    expect(fields).to.deep.equal(['id', 'name', 'age']);
-  });
-
   it('should generate a table create ddl based on a csv file', async () => {
     const csvFile = Buffer.from('id,name,age\n1,John,20\n2,Jane,21');
     const fields = getCsvHeaders(csvFile);
