#RI-5661 - add rdi auth

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/modules/rdi/client/api.rdi.client.ts

@@ -1,81 +1,110 @@
-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 { RDI_TIMEOUT, RdiUrl, TOKEN_TRESHOLD } from 'src/modules/rdi/constants';
 import { RdiDryRunJobDto, RdiDryRunJobResponseDto, RdiTestConnectionResult } from 'src/modules/rdi/dto';
-import { RdiPipelineDeployFailedException } from 'src/modules/rdi/exceptions';
+import { RdiPipelineDeployFailedException, wrapRdiPipelineError } from 'src/modules/rdi/exceptions';
 import {
-  RdiJob,
   RdiPipeline,
   RdiStatisticsResult,
-  RdiType,
-  RdiDryRunJobResult,
-  RdiDyRunJobStatus,
   RdiStatisticsStatus,
-  RdiStatisticsData,
+  RdiStatisticsData, RdiClientMetadata, Rdi,
 } from 'src/modules/rdi/models';
 import { convertKeysToCamelCase } from 'src/utils/base.helper';
 
 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);
+    try {
+      const response = await this.client.post(RdiUrl.TestConnections, config);
 
-    return response.data;
+      return response.data;
+    } 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 +120,32 @@ 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();
+    }
   }
 }

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

@@ -1,25 +1,27 @@
 import {
-  RdiClientMetadata, RdiJob, RdiPipeline, RdiType, RdiDryRunJobResult, RdiStatisticsResult,
+  Rdi,
+  RdiClientMetadata, RdiJob, RdiPipeline, RdiStatisticsResult,
 } from 'src/modules/rdi/models';
 import {
   RdiDryRunJobDto, RdiDryRunJobResponseDto, RdiTestConnectionResult,
 } from 'src/modules/rdi/dto';
+import { IDLE_TRESHOLD } from 'src/modules/rdi/constants';
 
 export abstract class RdiClient {
-  abstract type: RdiType;
-
   public readonly id: string;
 
   public lastUsed: number = Date.now();
 
-  constructor(
-    protected readonly metadata: RdiClientMetadata,
-    protected readonly client: unknown,
+  protected constructor(
+    public readonly metadata: RdiClientMetadata,
+    protected readonly rdi: Rdi,
   ) {
     this.id = RdiClient.generateId(this.metadata);
   }
 
-  abstract isConnected(): Promise<boolean>;
+  public isIdle(): boolean {
+    return Date.now() - this.lastUsed > IDLE_TRESHOLD;
+  }
 
   abstract getSchema(): Promise<object>;
 
@@ -32,8 +34,6 @@ export abstract class RdiClient {
 
   abstract deploy(pipeline: RdiPipeline): Promise<void>;
 
-  abstract deployJob(job: RdiJob): Promise<RdiJob>;
-
   abstract dryRunJob(data: RdiDryRunJobDto): Promise<RdiDryRunJobResponseDto>;
 
   abstract testConnections(config: string): Promise<RdiTestConnectionResult>;
@@ -44,7 +44,9 @@ export abstract class RdiClient {
 
   abstract getJobFunctions(): Promise<object>;
 
-  abstract disconnect(): Promise<void>;
+  abstract ensureAuth(): Promise<void>;
+
+  abstract connect(): Promise<void>;
 
   public setLastUsed(): void {
     this.lastUsed = Date.now();

redisinsight/api/src/modules/rdi/constants/index.ts

@@ -9,4 +9,10 @@ export enum RdiUrl {
   TestConnections = 'api/v1/pipelines/targets/dry-run',
   GetStatistics = 'api/v1/monitoring/statistics',
   GetPipelineStatus = 'api/v1/status',
+  Login = 'api/v1/login',
 }
+
+export const IDLE_TRESHOLD = 10 * 60 * 1000; // 10 min
+export const RDI_TIMEOUT = 30_000; // 30 sec
+export const TOKEN_TRESHOLD = 2 * 60 * 1000; // 2 min
+export const RDI_SYNC_INTERVAL = 5 * 60 * 1_000; // 5 min

redisinsight/api/src/modules/rdi/dto/create.rdi.dto.ts

@@ -1,16 +1,4 @@
 import { OmitType } from '@nestjs/swagger';
-import { Rdi, RdiType } from 'src/modules/rdi/models';
-import { ValidateIf } from 'class-validator';
+import { Rdi } from 'src/modules/rdi/models';
 
-export class CreateRdiDto extends OmitType(Rdi, [
-  'id', 'lastConnection',
-] as const) {
-  @ValidateIf(({ type }) => type === RdiType.API)
-  url?: string;
-
-  @ValidateIf(({ type }) => type === RdiType.GEARS)
-  host?: string;
-
-  @ValidateIf(({ type }) => type === RdiType.GEARS)
-  port?: number;
-}
+export class CreateRdiDto extends OmitType(Rdi, ['id', 'lastConnection', 'version'] as const) {}