Merge pull request #3414 from RedisInsight/feature/RI-5661_rdi_auth

Feature/ri 5661 rdi auth

Author: AmirAllayarovSofteq

redisinsight/api/migration/1716370509836-rdi.ts

@@ -0,0 +1,14 @@
+import { MigrationInterface, QueryRunner } from "typeorm";
+
+export class Rdi1716370509836 implements MigrationInterface {
+    name = 'Rdi1716370509836'
+
+    public async up(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`CREATE TABLE "rdi" ("id" varchar PRIMARY KEY NOT NULL, "url" varchar, "name" varchar NOT NULL, "username" varchar NOT NULL, "password" varchar NOT NULL, "lastConnection" datetime, "version" varchar NOT NULL, "encryption" varchar)`);
+    }
+
+    public async down(queryRunner: QueryRunner): Promise<void> {
+        await queryRunner.query(`DROP TABLE "rdi"`);
+    }
+
+}

redisinsight/api/migration/index.ts

@@ -40,6 +40,7 @@ import { CloudCapiKeys1691061058385 } from './1691061058385-cloud-capi-keys';
 import { FeatureSso1691476419592 } from './1691476419592-feature-sso';
 import { AiHistory1713515657364 } from './1713515657364-ai-history';
 import { AiHistorySteps1714501203616 } from './1714501203616-ai-history-steps';
+import { Rdi1716370509836 } from './1716370509836-rdi';
 
 export default [
   initialMigration1614164490968,
@@ -84,4 +85,5 @@ export default [
   FeatureSso1691476419592,
   AiHistory1713515657364,
   AiHistorySteps1714501203616,
+  Rdi1716370509836,
 ];

redisinsight/api/package.json

@@ -34,7 +34,8 @@
     "test:api:cov": "nyc --reporter=html --reporter=text --reporter=text-summary yarn run test:api",
     "test:api:ci:cov": "cross-env nyc -r text -r text-summary -r html yarn run test:api --reporter mocha-multi-reporters --reporter-options configFile=test/api/reporters.json && nyc merge .nyc_output ./coverage/test-run-coverage.json",
     "typeorm:migrate": "cross-env NODE_ENV=staging yarn typeorm migration:generate ./migration/migration",
-    "typeorm:run": "yarn typeorm migration:run"
+    "typeorm:run": "yarn typeorm migration:run",
+    "typeorm:run:stage": "cross-env NODE_ENV=staging yarn typeorm migration:run"
   },
   "resolutions": {
     "nanoid": "^3.1.31",

redisinsight/api/src/__mocks__/index.ts

@@ -39,3 +39,4 @@ export * from './session';
 export * from './cloud-session';
 export * from './database-info';
 export * from './cloud-job';
+export * from './rdi';

redisinsight/api/src/__mocks__/rdi.ts

@@ -0,0 +1,64 @@
+import {
+  Rdi,
+  RdiClientMetadata,
+} from 'src/modules/rdi/models';
+import { ApiRdiClient } from 'src/modules/rdi/client/api.rdi.client';
+
+export const mockRdiId = 'rdiId';
+
+export class MockRdiClient extends ApiRdiClient {
+  constructor(metadata: RdiClientMetadata, client: any = jest.fn()) {
+    super(metadata, client);
+  }
+
+  public getSchema = jest.fn();
+
+  public getPipeline = jest.fn();
+
+  public getTemplate = jest.fn();
+
+  public getStrategies = jest.fn();
+
+  public deploy = jest.fn();
+
+  public deployJob = jest.fn();
+
+  public dryRunJob = jest.fn();
+
+  public testConnections = jest.fn();
+
+  public getStatistics = jest.fn();
+
+  public getPipelineStatus = jest.fn();
+
+  public getJobFunctions = jest.fn();
+
+  public connect = jest.fn();
+
+  public ensureAuth = jest.fn();
+}
+
+export const generateMockRdiClient = (
+  metadata: RdiClientMetadata,
+  client = jest.fn(),
+): MockRdiClient => new MockRdiClient(metadata as RdiClientMetadata, client);
+
+export const mockRdiClientMetadata: RdiClientMetadata = {
+  sessionMetadata: undefined,
+  id: mockRdiId,
+};
+
+export const mockRdi = Object.assign(new Rdi(), {
+  name: 'name',
+  version: '1.2',
+  url: 'http://localhost:4000',
+  password: 'pass',
+  username: 'user',
+});
+
+export const mockRdiUnauthorizedError = {
+  message: 'Request failed with status code 401',
+  response: {
+    status: 401,
+  },
+};

redisinsight/api/src/constants/custom-error-codes.ts

@@ -52,4 +52,6 @@ export enum CustomErrorCodes {
 
   // RDI errors [11400, 11599]
   RdiDeployPipelineFailure = 11_401,
+  RdiUnauthorized = 11_402,
+  RdiInternalServerError = 11_403,
 }

redisinsight/api/src/constants/error-messages.ts

@@ -105,4 +105,5 @@ export default {
   COMMON_DEFAULT_IMPORT_ERROR: 'Unable to import default data',
 
   RDI_DEPLOY_PIPELINE_FAILURE: 'Failed to deploy pipeline',
+  RDI_TIMEOUT_ERROR: 'Encountered a timeout error while attempting to retrieve data',
 };

redisinsight/api/src/modules/rdi/client/api.rdi.client.ts

@@ -1,81 +1,127 @@
-import { AxiosInstance } from 'axios';
+import axios, { AxiosInstance } from 'axios';
 import { plainToClass } from 'class-transformer';
+import { decode } from 'jsonwebtoken';
 
 import { RdiClient } from 'src/modules/rdi/client/rdi.client';
-import { RdiUrl } from 'src/modules/rdi/constants';
-import { RdiDryRunJobDto, RdiDryRunJobResponseDto, RdiTestConnectionResult } from 'src/modules/rdi/dto';
-import { RdiPipelineDeployFailedException } from 'src/modules/rdi/exceptions';
 import {
-  RdiJob,
+  RdiUrl,
+  RDI_TIMEOUT,
+  TOKEN_TRESHOLD,
+  POLLING_INTERVAL,
+  MAX_POLLING_TIME,
+} from 'src/modules/rdi/constants';
+import {
+  RdiDryRunJobDto,
+  RdiDryRunJobResponseDto,
+  RdiTestConnectionsResponseDto,
+} from 'src/modules/rdi/dto';
+import {
+  RdiPipelineDeployFailedException,
+  RdiPipelineInternalServerErrorException,
+  wrapRdiPipelineError,
+} from 'src/modules/rdi/exceptions';
+import {
   RdiPipeline,
   RdiStatisticsResult,
-  RdiType,
-  RdiDryRunJobResult,
-  RdiDyRunJobStatus,
   RdiStatisticsStatus,
-  RdiStatisticsData,
+  RdiStatisticsData, RdiClientMetadata, Rdi,
 } from 'src/modules/rdi/models';
 import { convertKeysToCamelCase } from 'src/utils/base.helper';
+import { RdiPipelineTimeoutException } from 'src/modules/rdi/exceptions/rdi-pipeline.timeout-error.exception';
 
 const RDI_DEPLOY_FAILED_STATUS = 'failed';
 
 export class ApiRdiClient extends RdiClient {
-  public type = RdiType.API;
-
   protected readonly client: AxiosInstance;
 
-  async isConnected(): Promise<boolean> {
-    // todo: check if needed and possible
-    return true;
+  private auth: { jwt: string, exp: number };
+
+  constructor(clientMetadata: RdiClientMetadata, rdi: Rdi) {
+    super(clientMetadata, rdi);
+    this.client = axios.create({
+      baseURL: rdi.url,
+      timeout: RDI_TIMEOUT,
+    });
   }
 
   async getSchema(): Promise<object> {
-    const response = await this.client.get(RdiUrl.GetSchema);
-    return response.data;
+    try {
+      const response = await this.client.get(RdiUrl.GetSchema);
+      return response.data;
+    } catch (e) {
+      throw wrapRdiPipelineError(e);
+    }
   }
 
   async getPipeline(): Promise<RdiPipeline> {
-    const response = await this.client.get(RdiUrl.GetPipeline);
-    return response.data;
+    try {
+      const response = await this.client.get(RdiUrl.GetPipeline);
+      return response.data;
+    } catch (e) {
+      throw wrapRdiPipelineError(e);
+    }
   }
 
   async getStrategies(): Promise<object> {
-    const response = await this.client.get(RdiUrl.GetStrategies);
-    return response.data;
+    try {
+      const response = await this.client.get(RdiUrl.GetStrategies);
+      return response.data;
+    } catch (e) {
+      throw wrapRdiPipelineError(e);
+    }
   }
 
   async getTemplate(options: object): Promise<object> {
-    const response = await this.client.get(RdiUrl.GetTemplate, { params: options });
-    return response.data;
+    try {
+      const response = await this.client.get(RdiUrl.GetTemplate, { params: options });
+      return response.data;
+    } catch (error) {
+      throw wrapRdiPipelineError(error);
+    }
   }
 
   async deploy(pipeline: RdiPipeline): Promise<void> {
-    const response = await this.client.post(RdiUrl.Deploy, { ...pipeline });
+    let response;
+    try {
+      response = await this.client.post(RdiUrl.Deploy, { ...pipeline });
+    } catch (error) {
+      throw wrapRdiPipelineError(error, error.response.data.message);
+    }
 
     if (response.data?.status === RDI_DEPLOY_FAILED_STATUS) {
       throw new RdiPipelineDeployFailedException(undefined, { error: response.data?.error });
     }
   }
 
-  async deployJob(job: RdiJob): Promise<RdiJob> {
-    return null;
-  }
-
   async dryRunJob(data: RdiDryRunJobDto): Promise<RdiDryRunJobResponseDto> {
-    const response = await this.client.post(RdiUrl.DryRunJob, data);
-    return response.data;
+    try {
+      const response = await this.client.post(RdiUrl.DryRunJob, data);
+      return response.data;
+    } catch (e) {
+      throw wrapRdiPipelineError(e);
+    }
   }
 
-  async testConnections(config: string): Promise<RdiTestConnectionResult> {
-    const response = await this.client.post(RdiUrl.TestConnections, config);
+  async testConnections(config: string): Promise<RdiTestConnectionsResponseDto> {
+    try {
+      const response = await this.client.post(RdiUrl.TestConnections, config);
+
+      const actionId = response.data.action_id;
 
-    return response.data;
+      return this.pollActionStatus(actionId);
+    } catch (e) {
+      throw wrapRdiPipelineError(e);
+    }
   }
 
   async getPipelineStatus(): Promise<any> {
-    const response = await this.client.get(RdiUrl.GetPipelineStatus);
+    try {
+      const response = await this.client.get(RdiUrl.GetPipelineStatus);
 
-    return response.data;
+      return response.data;
+    } catch (e) {
+      throw wrapRdiPipelineError(e);
+    }
   }
 
   async getStatistics(sections?: string): Promise<RdiStatisticsResult> {
@@ -91,11 +137,61 @@ export class ApiRdiClient extends RdiClient {
   }
 
   async getJobFunctions(): Promise<object> {
-    const response = await this.client.post(RdiUrl.JobFunctions);
-    return response.data;
+    try {
+      const response = await this.client.post(RdiUrl.JobFunctions);
+      return response.data;
+    } catch (e) {
+      throw wrapRdiPipelineError(e);
+    }
   }
 
-  async disconnect(): Promise<void> {
-    return undefined;
+  async connect(): Promise<void> {
+    try {
+      const response = await this.client.post(
+        RdiUrl.Login,
+        { username: this.rdi.username, password: this.rdi.password },
+      );
+      const accessToken = response.data.access_token;
+      const decodedJwt = decode(accessToken);
+
+      this.auth = { jwt: accessToken, exp: decodedJwt.exp };
+      this.client.defaults.headers.common['Authorization'] = `Bearer ${accessToken}`;
+    } catch (e) {
+      throw wrapRdiPipelineError(e);
+    }
+  }
+
+  async ensureAuth(): Promise<void> {
+    const expiresIn = this.auth.exp * 1_000 - Date.now();
+
+    if (expiresIn < TOKEN_TRESHOLD) {
+      await this.connect();
+    }
+  }
+
+  private async pollActionStatus(actionId: string): Promise<any> {
+    const startTime = Date.now();
+    while (true) {
+      if (Date.now() - startTime > MAX_POLLING_TIME) {
+        throw new RdiPipelineTimeoutException();
+      }
+
+      try {
+        const response = await this.client.get(`${RdiUrl.Action}/${actionId}`);
+        const { status, data, error } = response.data;
+
+        if (status === 'failed') {
+          throw new RdiPipelineInternalServerErrorException(error);
+        }
+
+        if (status === 'completed') {
+          return data;
+        }
+      } catch (e) {
+        throw wrapRdiPipelineError(e);
+      }
+
+      await new Promise((resolve) => setTimeout(resolve, POLLING_INTERVAL));
+    }
   }
 }