chore: refactor oibus engine service, controller and tests

backend/src/db/entity-migrations/v3.4.0.ts

@@ -0,0 +1,111 @@
+import { Knex } from 'knex';
+import { HISTORY_QUERIES_TABLE } from '../../repository/history-query.repository';
+import { NORTH_CONNECTORS_TABLE } from '../../repository/north-connector.repository';
+import path from 'node:path';
+import { filesExists } from '../../service/utils';
+import fs from 'node:fs/promises';
+import { HISTORY_ITEMS_TABLE } from '../../repository/history-query-item.repository';
+import { OIANALYTICS_MESSAGE_TABLE } from '../../repository/oianalytics-message.repository';
+import { COMMANDS_TABLE } from '../../repository/oianalytics-command.repository';
+import { SOUTH_CONNECTORS_TABLE } from '../../repository/south-connector.repository';
+
+const EXTERNAL_SOURCES_TABLE = 'external_sources';
+const EXTERNAL_SUBSCRIPTION_TABLE = 'external_subscription';
+
+export async function up(knex: Knex): Promise<void> {
+  await removeNorthOIBusConnectors(knex);
+  await removeNorthOIBusHistoryQueries(knex);
+  await removeExternalSubscriptions(knex);
+  await updateOIAMessageTable(knex);
+  await recreateCommandTable(knex);
+  await updateSouthConnectorsTable(knex);
+  await updateHistoryQueriesTable(knex);
+}
+
+async function removeNorthOIBusConnectors(knex: Knex): Promise<void> {
+  const northConnectors: Array<{ id: string; name: string }> = await knex(NORTH_CONNECTORS_TABLE)
+    .select('id', 'name')
+    .where('type', 'oibus');
+  await knex(NORTH_CONNECTORS_TABLE).delete().where('type', 'oibus');
+  for (const north of northConnectors) {
+    const baseFolder = path.resolve('./cache/data-stream', `north-${north.id}`);
+    if (await filesExists(baseFolder)) {
+      await fs.rm(baseFolder, { recursive: true });
+    }
+  }
+}
+
+async function removeNorthOIBusHistoryQueries(knex: Knex): Promise<void> {
+  const historyQueries: Array<{ id: string; name: string }> = await knex(HISTORY_QUERIES_TABLE)
+    .select('id', 'name')
+    .where('north_type', 'oibus');
+
+  for (const history of historyQueries) {
+    await knex(HISTORY_ITEMS_TABLE).delete().where('history_id', history.id);
+    const baseFolder = path.resolve('./cache/history-query', `history-${history.id}`);
+    if (await filesExists(baseFolder)) {
+      await fs.rm(baseFolder, { recursive: true });
+    }
+  }
+  await knex(HISTORY_QUERIES_TABLE).delete().where('north_type', 'oibus');
+}
+
+async function removeExternalSubscriptions(knex: Knex): Promise<void> {
+  await knex.schema.dropTableIfExists(EXTERNAL_SUBSCRIPTION_TABLE);
+  await knex.schema.dropTableIfExists(EXTERNAL_SOURCES_TABLE);
+}
+
+async function updateOIAMessageTable(knex: Knex): Promise<void> {
+  await knex.schema.raw(`delete
+                         from ${OIANALYTICS_MESSAGE_TABLE}`);
+  await knex.schema.alterTable(OIANALYTICS_MESSAGE_TABLE, table => {
+    table.dropColumn('content');
+  });
+}
+
+async function recreateCommandTable(knex: Knex): Promise<void> {
+  await knex.schema.raw(
+    `create table temporary_table
+     (
+       id               char(36) primary key,
+       created_at       datetime default CURRENT_TIMESTAMP not null,
+       updated_at       datetime default CURRENT_TIMESTAMP not null,
+       type             text                               not null,
+       status           text                               not null,
+       ack              boolean  default '0'               not null,
+       retrieved_date   varchar(255),
+       completed_date   varchar(255),
+       result           varchar(255),
+       upgrade_version  varchar(255),
+       upgrade_asset_id varchar(255),
+       command_content  text
+     );
+    `
+  );
+  await knex.schema.raw(`INSERT INTO temporary_table
+                         SELECT *, ''
+                         FROM ${COMMANDS_TABLE}`);
+  await knex.schema.raw(`DROP TABLE ${COMMANDS_TABLE}`);
+  await knex.schema.raw(`ALTER TABLE temporary_table
+    RENAME TO ${COMMANDS_TABLE}`);
+  await knex.schema.alterTable(COMMANDS_TABLE, table => {
+    table.string('south_connector_id', 255);
+    table.string('north_connector_id', 255);
+    table.string('scan_mode_id', 255);
+    table.string('target_version', 255);
+  });
+}
+
+async function updateSouthConnectorsTable(knex: Knex): Promise<void> {
+  await knex.schema.alterTable(SOUTH_CONNECTORS_TABLE, table => {
+    table.boolean('shared_connection').defaultTo(false);
+  });
+}
+
+async function updateHistoryQueriesTable(knex: Knex): Promise<void> {
+  await knex.schema.alterTable(HISTORY_QUERIES_TABLE, table => {
+    table.boolean('south_shared_connection').defaultTo(false);
+  });
+}
+
+export async function down(): Promise<void> {}

backend/src/engine/oibus-engine.spec.ts

@@ -17,8 +17,6 @@ import fs from 'node:fs/promises';
 import path from 'node:path';
 import { ScanModeDTO } from '../../../shared/model/scan-mode.model';
 import { filesExists } from '../service/utils';
-import HomeMetricsServiceMock from '../tests/__mocks__/service/home-metrics-service.mock';
-import HomeMetricsService from '../service/home-metrics.service';
 import { OIBusTimeValue } from '../../../shared/model/engine.model';
 
 jest.mock('../south/south-mqtt/south-mqtt');
@@ -34,7 +32,6 @@ const anotherLogger: pino.Logger = new PinoLogger();
 
 const southService: SouthService = new SouthServiceMock();
 const northService: NorthService = new NorthServiceMock();
-const homeMetrics: HomeMetricsService = new HomeMetricsServiceMock();
 const encryptionService: EncryptionService = new EncryptionServiceMock('', '');
 
 const nowDateString = '2020-02-02T02:02:02.222Z';
@@ -178,7 +175,7 @@ describe('OIBusEngine', () => {
     (southService.createSouth as jest.Mock).mockReturnValue(createdSouth);
     (northService.createNorth as jest.Mock).mockReturnValue(createdNorth);
 
-    engine = new OIBusEngine(encryptionService, northService, southService, homeMetrics, logger);
+    engine = new OIBusEngine(encryptionService, northService, southService, logger);
   });
 
   it('it should start', async () => {

backend/src/engine/oibus-engine.ts

@@ -14,7 +14,6 @@ import { NorthConnectorDTO } from '../../../shared/model/north-connector.model';
 import { Instant } from '../../../shared/model/types';
 import { PassThrough } from 'node:stream';
 import { ScanModeDTO } from '../../../shared/model/scan-mode.model';
-import HomeMetricsService from '../service/home-metrics.service';
 import { OIBusContent } from '../../../shared/model/engine.model';
 
 const CACHE_FOLDER = './cache/data-stream';
@@ -26,13 +25,7 @@ export default class OIBusEngine extends BaseEngine {
   private northConnectors: Map<string, NorthConnector> = new Map<string, NorthConnector>();
   private southConnectors: Map<string, SouthConnector> = new Map<string, SouthConnector>();
 
-  constructor(
-    encryptionService: EncryptionService,
-    northService: NorthService,
-    southService: SouthService,
-    private readonly homeMetricsService: HomeMetricsService,
-    logger: pino.Logger
-  ) {
+  constructor(encryptionService: EncryptionService, northService: NorthService, southService: SouthService, logger: pino.Logger) {
     super(encryptionService, northService, southService, logger, CACHE_FOLDER);
   }
 
@@ -131,7 +124,7 @@ export default class OIBusEngine extends BaseEngine {
       this.logger.child({ scopeType: 'south', scopeId: settings.id, scopeName: settings.name })
     );
     this.southConnectors.set(settings.id, south);
-    this.homeMetricsService.addSouth(south, south.settings.id);
+    // this.homeMetricsService.addSouth(south, south.settings.id);
   }
 
   async startSouth(southId: string): Promise<void> {
@@ -164,7 +157,7 @@ export default class OIBusEngine extends BaseEngine {
       this.logger.child({ scopeType: 'north', scopeId: settings.id, scopeName: settings.name })
     );
     this.northConnectors.set(settings.id, north);
-    this.homeMetricsService.addNorth(north, north.settings.id);
+    // this.homeMetricsService.addNorth(north, north.settings.id);
   }
 
   async startNorth(northId: string): Promise<void> {
@@ -196,7 +189,7 @@ export default class OIBusEngine extends BaseEngine {
    */
   async deleteSouth(southId: string, name: string): Promise<void> {
     await this.stopSouth(southId);
-    this.homeMetricsService.removeSouth(southId);
+    // this.homeMetricsService.removeSouth(southId);
     this.southConnectors.delete(southId);
     const baseFolder = path.resolve(this.cacheFolder, `south-${southId}`);
 
@@ -218,7 +211,7 @@ export default class OIBusEngine extends BaseEngine {
    */
   async deleteNorth(northId: string, name: string): Promise<void> {
     await this.stopNorth(northId);
-    this.homeMetricsService.removeNorth(northId);
+    // this.homeMetricsService.removeNorth(northId);
     this.northConnectors.delete(northId);
 
     const baseFolder = path.resolve(this.cacheFolder, `north-${northId}`);