Support for sync from devrev to external system (#12)

Co-authored-by: Samo Dekleva <[email protected]>

Author: radovan-jorgic

README.md

@@ -2,6 +2,10 @@
 
 ## Release Notes
 
+#### v1.1.0
+
+- Support for sync from DevRev to external system. Known limitations: no support for loading attachments.
+
 #### v1.0.4
 
 - Fix logging from worker threads.

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@devrev/ts-adaas",
-  "version": "1.0.4",
+  "version": "1.1.0",
   "description": "Typescript library containing the ADaaS(AirDrop as a Service) control protocol.",
   "type": "commonjs",
   "main": "./dist/index.js",

src/common/constants.ts

@@ -5,9 +5,11 @@ export const STATELESS_EVENT_TYPES = [
   EventType.ExtractionMetadataStart,
   EventType.ExtractionDataDelete,
   EventType.ExtractionAttachmentsDelete,
+  EventType.StartDeletingLoaderState,
+  EventType.StartDeletingLoaderAttachmentsState,
 ];
 
-export const ALLOWED_EVENT_TYPES = [
+export const ALLOWED_EXTRACTION_EVENT_TYPES = [
   EventType.ExtractionExternalSyncUnitsStart,
   EventType.ExtractionMetadataStart,
   EventType.ExtractionDataStart,
@@ -18,6 +20,18 @@ export const ALLOWED_EVENT_TYPES = [
   EventType.ExtractionAttachmentsDelete,
 ];
 
+export const ALLOWED_LOADING_EVENT_TYPES = [
+  EventType.StartLoadingData,
+  EventType.ContinueLoadingData,
+  EventType.StartDeletingLoaderState,
+  EventType.StartDeletingLoaderAttachmentsState,
+];
+
+export const ALLOWED_EVENT_TYPES = [
+  ...ALLOWED_EXTRACTION_EVENT_TYPES,
+  ...ALLOWED_LOADING_EVENT_TYPES,
+];
+
 export const ARTIFACT_BATCH_SIZE = 2000;
 export const MAX_DEVREV_ARTIFACT_SIZE = 536870912; // 512MB
 

src/common/control-protocol.ts

@@ -5,12 +5,14 @@ import {
   EventData,
   ExtractorEvent,
   ExtractorEventType,
+  LoaderEvent,
 } from '../types/extraction';
+import { LoaderEventType } from '../types/loading';
 import { formatAxiosError } from '../logger/logger';
 
 export interface EmitInterface {
   event: AirdropEvent;
-  eventType: ExtractorEventType;
+  eventType: ExtractorEventType | LoaderEventType;
   data?: EventData;
 }
 
@@ -19,7 +21,7 @@ export const emit = async ({
   eventType,
   data,
 }: EmitInterface): Promise<void | Error> => {
-  const newEvent: ExtractorEvent = {
+  const newEvent: ExtractorEvent | LoaderEvent = {
     event_type: eventType,
     event_context: {
       uuid: event.payload.event_context.uuid,

src/common/helpers.ts

@@ -1,7 +1,19 @@
-import { EventType, ExtractorEventType } from '../types/extraction';
+import {
+  AirdropEvent,
+  EventType,
+  ExtractorEventType,
+} from '../types/extraction';
+import {
+  ActionType,
+  FileToLoad,
+  ItemTypeToLoad,
+  LoaderEventType,
+  LoaderReport,
+  StatsFileObject,
+} from '../types/loading';
 
-export function getErrorExtractorEventType(eventType: EventType): {
-  eventType: ExtractorEventType;
+export function getTimeoutErrorEventType(eventType: EventType): {
+  eventType: ExtractorEventType | LoaderEventType;
 } | null {
   switch (eventType) {
     case EventType.ExtractionMetadataStart:
@@ -30,11 +42,104 @@ export function getErrorExtractorEventType(eventType: EventType): {
       return {
         eventType: ExtractorEventType.ExtractionExternalSyncUnitsError,
       };
+
+    case EventType.StartLoadingData:
+    case EventType.ContinueLoadingData:
+      return {
+        eventType: LoaderEventType.DataLoadingError,
+      };
+    case EventType.StartDeletingLoaderState:
+    case EventType.StartDeletingLoaderAttachmentsState:
+      return {
+        eventType: LoaderEventType.LoaderStateDeletionError,
+      };
     default:
       console.error(
-        'Event type not recognized in getTimeoutExtractorEventType function: ' +
+        'Event type not recognized in getTimeoutErrorEventType function: ' +
           eventType
       );
       return null;
   }
 }
+
+export function getSyncDirection({ event }: { event: AirdropEvent }) {
+  return event.payload.event_context.mode;
+}
+
+export function getFilesToLoad({
+  itemTypesToLoad,
+  statsFile,
+}: {
+  itemTypesToLoad: ItemTypeToLoad[];
+  statsFile: StatsFileObject[];
+}): FileToLoad[] {
+  const filesToLoad = [];
+
+  if (itemTypesToLoad.length === 0 || statsFile.length === 0) {
+    return [];
+  }
+
+  const supportedItemTypes = itemTypesToLoad.map(
+    (itemTypeToLoad) => itemTypeToLoad.itemType
+  );
+
+  const filteredStatsFile = statsFile.filter((file) =>
+    supportedItemTypes.includes(file.item_type)
+  );
+
+  const orderedFiles = filteredStatsFile.sort((a, b) => {
+    const aIndex = supportedItemTypes.indexOf(a.item_type);
+    const bIndex = supportedItemTypes.indexOf(b.item_type);
+
+    return aIndex - bIndex;
+  });
+
+  for (const file of orderedFiles) {
+    filesToLoad.push({
+      id: file.id,
+      file_name: file.file_name,
+      itemType: file.item_type,
+      count: parseInt(file.count),
+      completed: false,
+      lineToProcess: 0,
+    });
+  }
+
+  return filesToLoad;
+}
+
+export function addReportToLoaderReport({
+  loaderReports,
+  report,
+}: {
+  loaderReports: LoaderReport[];
+  report: LoaderReport;
+}): LoaderReport[] {
+  const existingReport = loaderReports.find(
+    (loaderReport) => loaderReport.item_type === report.item_type
+  );
+
+  if (existingReport) {
+    existingReport[ActionType.CREATED] = existingReport[ActionType.CREATED]
+      ? report[ActionType.CREATED]
+        ? existingReport[ActionType.CREATED] + report[ActionType.CREATED]
+        : existingReport[ActionType.CREATED]
+      : report[ActionType.CREATED];
+
+    existingReport[ActionType.UPDATED] = existingReport[ActionType.UPDATED]
+      ? report[ActionType.UPDATED]
+        ? existingReport[ActionType.UPDATED] + report[ActionType.UPDATED]
+        : existingReport[ActionType.UPDATED]
+      : report[ActionType.UPDATED];
+
+    existingReport[ActionType.FAILED] = existingReport[ActionType.FAILED]
+      ? report[ActionType.FAILED]
+        ? existingReport[ActionType.FAILED] + report[ActionType.FAILED]
+        : existingReport[ActionType.FAILED]
+      : report[ActionType.FAILED];
+  } else {
+    loaderReports.push(report);
+  }
+
+  return loaderReports;
+}

src/logger/logger.test.ts

@@ -50,5 +50,8 @@ it('formatAxiosError should return formatted error', () => {
     status: 500,
     data: 'Internal server error',
     method: 'GET',
+    baseURL: undefined,
+    url: undefined,
+    payload: undefined,
   });
 });

src/mappers/mappers.interface.ts

@@ -0,0 +1,96 @@
+import { AirdropEvent } from '../types';
+import { DonV2 } from '../types/loading';
+import { WorkerAdapterOptions } from '../types/workers';
+
+export interface MappersFactoryInterface {
+  event: AirdropEvent;
+  options?: WorkerAdapterOptions;
+}
+
+export interface UpdateSyncMapperRecordParams {
+  external_ids: {
+    add: string[];
+  };
+  secondary_ids?: Record<string, string>;
+  targets: {
+    add: DonV2[];
+  };
+  status: SyncMapperRecordStatus;
+  input_files?: {
+    add: string[];
+  };
+  external_versions?: {
+    add: SyncMapperRecordExternalVersion[];
+  };
+  extra_data?: string;
+}
+
+export interface SyncMapperRecord {
+  id: DonV2;
+  external_ids: string[];
+  secondary_ids?: Record<string, string>;
+  targets: DonV2[];
+  status: SyncMapperRecordStatus;
+  input_files?: string[];
+  external_versions?: SyncMapperRecordExternalVersion[];
+  extra_data?: string;
+}
+
+export interface MappersGetByTargetIdParams {
+  sync_unit: DonV2;
+  target: DonV2;
+}
+
+export interface MappersGetByTargetIdResponse {
+  sync_mapper_record: SyncMapperRecord;
+}
+
+export interface MappersCreateParams {
+  sync_unit: DonV2;
+  external_ids: string[];
+  secondary_ids?: Record<string, string>;
+  targets: DonV2[];
+  status: SyncMapperRecordStatus;
+  input_files?: string[];
+  external_versions?: SyncMapperRecordExternalVersion[];
+  extra_data?: string;
+}
+
+export interface MappersCreateResponse {
+  sync_mapper_record: SyncMapperRecord;
+}
+
+export interface MappersUpdateParams {
+  id: DonV2;
+  sync_unit: DonV2;
+  external_ids: {
+    add: string[];
+  };
+  secondary_ids?: Record<string, string>;
+  targets: {
+    add: DonV2[];
+  };
+  status: SyncMapperRecordStatus;
+  input_files?: {
+    add: string[];
+  };
+  external_versions?: {
+    add: SyncMapperRecordExternalVersion[];
+  };
+  extra_data?: string;
+}
+
+export interface MappersUpdateResponse {
+  sync_mapper_record: SyncMapperRecord;
+}
+
+export enum SyncMapperRecordStatus {
+  OPERATIONAL = 'operational',
+  FILTERED = 'filtered',
+  IGNORED = 'ignored',
+}
+
+export interface SyncMapperRecordExternalVersion {
+  recipe_version: number;
+  modified_date: string;
+}

src/mappers/mappers.ts

@@ -0,0 +1,64 @@
+import axios, { AxiosResponse } from 'axios';
+
+import {
+  MappersFactoryInterface,
+  MappersCreateParams,
+  MappersCreateResponse,
+  MappersGetByTargetIdParams,
+  MappersGetByTargetIdResponse,
+  MappersUpdateParams,
+  MappersUpdateResponse,
+} from './mappers.interface';
+
+export class Mappers {
+  private endpoint: string;
+  private token: string;
+
+  constructor({ event }: MappersFactoryInterface) {
+    this.endpoint = event.execution_metadata.devrev_endpoint;
+    this.token = event.context.secrets.service_account_token;
+  }
+
+  async getByTargetId(
+    params: MappersGetByTargetIdParams
+  ): Promise<AxiosResponse<MappersGetByTargetIdResponse>> {
+    const { sync_unit, target } = params;
+    return axios.get<MappersGetByTargetIdResponse>(
+      `${this.endpoint}/internal/airdrop.sync-mapper-record.get-by-target`,
+      {
+        headers: {
+          Authorization: this.token,
+        },
+        params: { sync_unit, target },
+      }
+    );
+  }
+
+  async create(
+    params: MappersCreateParams
+  ): Promise<AxiosResponse<MappersCreateResponse>> {
+    return axios.post<MappersCreateResponse>(
+      `${this.endpoint}/internal/airdrop.sync-mapper-record.create`,
+      params,
+      {
+        headers: {
+          Authorization: this.token,
+        },
+      }
+    );
+  }
+
+  async update(
+    params: MappersUpdateParams
+  ): Promise<AxiosResponse<MappersUpdateResponse>> {
+    return axios.post<MappersUpdateResponse>(
+      `${this.endpoint}/internal/airdrop.sync-mapper-record.update`,
+      params,
+      {
+        headers: {
+          Authorization: this.token,
+        },
+      }
+    );
+  }
+}