Extraction function scaffolding in base folder

base_folder/src/function-factory.ts

@@ -1,5 +1,8 @@
+import extraction from './functions/extraction';
+
 export const functionFactory = {
   // Add your functions here
+  extraction,
 } as const;
 
 export type FunctionFactoryType = keyof typeof functionFactory;

base_folder/src/functions/extraction/index.ts

@@ -0,0 +1,42 @@
+import { convertToAirdropEvent } from '../../core/utils';
+import { FunctionInput } from '../../core/types';
+import { spawn, EventType } from '@devrev/ts-adaas';
+
+export interface ExtractorState {}
+
+export const initialState: ExtractorState = {};
+
+
+function getWorkerPerExtractionPhase(event: FunctionInput) {
+  let path;
+  switch (event.payload.event_type) {
+    case EventType.ExtractionExternalSyncUnitsStart:
+      path = __dirname + '/workers/external-sync-units-extraction';
+      break;
+    case EventType.ExtractionMetadataStart:
+      path = __dirname + '/workers/metadata-extraction';
+      break;
+    case EventType.ExtractionDataStart:
+    case EventType.ExtractionDataContinue:
+      path = __dirname + '/workers/data-extraction';
+      break;
+    case EventType.ExtractionAttachmentsStart:
+    case EventType.ExtractionAttachmentsContinue:
+      path = __dirname + '/workers/attachments-extraction';
+      break;
+  }
+  return path;
+}
+
+const run = async (events: FunctionInput[]) => {
+  for (const event of events) {
+    const file = getWorkerPerExtractionPhase(event);
+    await spawn({
+      event: convertToAirdropEvent(event),
+      workerPath: file,
+      initialState: initialState,
+    });
+  }
+};
+
+export default run;

base_folder/src/functions/extraction/workers/attachments-extraction.ts

@@ -0,0 +1,9 @@
+import { ExtractorEventType, processTask } from '@devrev/ts-adaas';
+import { ExtractorState } from "../index";
+
+processTask<ExtractorState>({
+  task: async ({ adapter }) => {},
+  onTimeout: async ({ adapter }) => {
+    await adapter.emit(ExtractorEventType.ExtractionAttachmentsProgress);
+  },
+});

base_folder/src/functions/extraction/workers/data-extraction.ts

@@ -0,0 +1,9 @@
+import { ExtractorEventType, processTask } from "@devrev/ts-adaas";
+import { ExtractorState } from "../index";
+
+processTask<ExtractorState>({
+  task: async ({ adapter }) => {},
+  onTimeout: async ({ adapter }) => {
+    await adapter.emit(ExtractorEventType.ExtractionDataProgress);
+  },
+});

base_folder/src/functions/extraction/workers/external-sync-units-extraction.ts

@@ -0,0 +1,13 @@
+import { ExtractorEventType, processTask } from "@devrev/ts-adaas";
+import { ExtractorState } from "../index";
+
+processTask<ExtractorState>({
+  task: async ({ adapter }) => {},
+  onTimeout: async ({ adapter }) => {
+    await adapter.emit(ExtractorEventType.ExtractionExternalSyncUnitsError, {
+      error: {
+        message: 'Failed to extract external sync units. Lambda timeout.',
+      },
+    });
+  },
+});

base_folder/src/functions/extraction/workers/metadata-extraction.ts

@@ -0,0 +1,11 @@
+import { processTask, ExtractorEventType } from '@devrev/ts-adaas';
+import { ExtractorState } from '../index';
+
+processTask<ExtractorState>({
+  task: async ({ adapter }) => {},
+  onTimeout: async ({ adapter }) => {
+    await adapter.emit(ExtractorEventType.ExtractionMetadataError, {
+      error: { message: 'Failed to extract metadata. Lambda timeout.' },
+    });
+  },
+});

devrev-snapin-template.plain

@@ -60,6 +60,7 @@
 
 - The Conformance Tests should spin a server at http://localhost:8002 for testing callbacks (The Callback Server).
   - The Conformance Tests *MUST NOT* send the event directly to The Callback Server.
+  - The Implementation Code *MUST NOT* send the event directly to The Callback Server.
 
 - The DevRev Server runs at http://localhost:8003, so it must not be mocked.
 

run_devrev_snapin_conformance_tests.sh

@@ -6,9 +6,9 @@ NPM_INSTALL_OUTPUT_FILTER="up to date in|added [0-9]* packages, removed [0-9]* p
 ANSI_ESCAPE_PATTERN="s/\x1b\[[0-9;]*[mK]//g"
 
 # Maximum number of characters to display from log files
-SNAP_IN_LOG_MAX_CHARS=30000
-DEVREV_SERVER_LOG_MAX_CHARS=15000
-PROXY_SERVER_LOG_MAX_CHARS=10000
+SNAP_IN_LOG_MAX_CHARS=80000
+DEVREV_SERVER_LOG_MAX_CHARS=40000
+PROXY_SERVER_LOG_MAX_CHARS=20000
 
 # Function to print a log file, truncating it if it's too large
 print_log_file() {

templates/attachment_extraction_acceptance_test.plain

@@ -1,10 +1,10 @@
-  - Test The Extraction Function with the following flow (The Attachment Test Flow):
-    - Step 1: Invoke The Extraction Function using the resource [{{ data_extraction_resource_name }}](test_data/{{ data_extraction_resource_name }}).
-      - Expect The Callback Server to receive from DevRev a **single** event with "event_type" that equals "EXTRACTION_DATA_DONE".
-    - Step 2: Invoke The Extraction Function using the resource [{{ attachment_extraction_resource_name }}](test_data/{{ attachment_extraction_resource_name }}).
-      - Expect The Callback Server to receive from DevRev a **single** event (`:= event_from_callback_server`) with "event_type" that equals "EXTRACTION_ATTACHMENTS_DONE".
-      - Expect `event_from_callback_server["event_data"]["artifacts"]` to be an array. Expect this array to not be empty. Expect this array to have length 1.
-      - Let `artifact_object = event_from_callback_server["event_data"]["artifacts"][0]`.
-        - Expect `artifact_object["item_type"] = "ssor_attachment"` and `artifact_object["item_count"] = {{ expected_attachment_count }}`.
-        - Make a GET request to "http://localhost:8003/is_uploaded/{artifact_object['id']}". Expect response code to be equal to 200. 
-    - Note: The Attachment Test Flow should be executed within a single test.
+Test The Extraction Function with the following flow (The Attachment Test Flow):
+  - Step 1: Invoke The Extraction Function using the resource [{{ data_extraction_resource_name }}](test_data/{{ data_extraction_resource_name }}).
+    - Expect The Callback Server to receive from DevRev a **single** event with "event_type" that equals "EXTRACTION_DATA_DONE".
+  - Step 2: Invoke The Extraction Function using the resource [{{ attachment_extraction_resource_name }}](test_data/{{ attachment_extraction_resource_name }}).
+    - Expect The Callback Server to receive from DevRev a **single** event (`:= event_from_callback_server`) with "event_type" that equals "EXTRACTION_ATTACHMENTS_DONE".
+    - Expect `event_from_callback_server["event_data"]["artifacts"]` to be an array. Expect this array to not be empty. Expect this array to have length 1.
+    - Let `artifact_object = event_from_callback_server["event_data"]["artifacts"][0]`.
+      - Expect `artifact_object["item_type"] = "ssor_attachment"` and `artifact_object["item_count"] = {{ expected_attachment_count }}`.
+      - Make a GET request to "http://localhost:8003/is_uploaded/{artifact_object['id']}". Expect response code to be equal to 200. 
+  - Note: The Attachment Test Flow should be executed within a single test.

templates/chef_cli_normalization_validation.plain

@@ -1,10 +1,10 @@
-    - Then, validate The Normalization Function with the chef-cli tool (The Chef CLI) using the command "chef-cli validate-data -m <The External Domain Metadata JSON object file> -r {{ record_type }}":
-      - To invoke the function that returns The External Domain Metadata JSON object, send the event [external_domain_metadata_event_payload.json](test_data/external_domain_metadata_event_payload.json) to The Snap-In server.
-      - The External Domain Metadata JSON object should then be stored in temporary file and provided as path to the validation of The Normalization Function.
-      - The Chef CLI reads the extracted file (The Extracted File) from stdin.
-      - The Extracted File is inside folder provided in the environment variable "EXTRACTED_FILES_FOLDER_PATH". To retrieve the relevant file name, command `ls EXTRACTED_FILES_FOLDER_PATH | grep extractor_{{ record_type }} | sort -r | head -n 1` should be executed. The test should always fail if The Extracted File is not present.
-      - Please note that the expected behaviour is that only after the invocation of The Extraction Function, the folder provided in the environment variable "EXTRACTED_FILES_FOLDER_PATH" should exist. The test should always fail if before the execution of The Chef CLI, the folder "EXTRACTED_FILES_FOLDER_PATH" does not exist or if The Extracted File is not found.
-      - The path to The Chef CLI executable is provided in the environment variable `CHEF_CLI_PATH`.
-      - For the validation of The Normalization Function to be successful The Chef CLI must return an empty output. The implementation of The Extraction Function directly infuences the output of The Chef CLI.
-      - The stdout and stderr of The Chef CLI *must* be printed to the console.
-      - The test should always fail if The Chef CLI is not available.
+validate The Normalization Function with the chef-cli tool (The Chef CLI) using the command "chef-cli validate-data -m <The External Domain Metadata JSON object file> -r {{ record_type }}":
+  - To invoke the function that returns The External Domain Metadata JSON object, send the event [external_domain_metadata_event_payload.json](test_data/external_domain_metadata_event_payload.json) to The Snap-In server.
+  - The External Domain Metadata JSON object should then be stored in temporary file and provided as path to the validation of The Normalization Function.
+  - The Chef CLI reads the extracted file (The Extracted File) from stdin.
+  - The Extracted File is inside folder provided in the environment variable "EXTRACTED_FILES_FOLDER_PATH". To retrieve the relevant file name, command `ls EXTRACTED_FILES_FOLDER_PATH | grep extractor_{{ record_type }} | sort -r | head -n 1` should be executed. The test should always fail if The Extracted File is not present.
+  - Please note that the expected behaviour is that only after the invocation of The Extraction Function, the folder provided in the environment variable "EXTRACTED_FILES_FOLDER_PATH" should exist. The test should always fail if before the execution of The Chef CLI, the folder "EXTRACTED_FILES_FOLDER_PATH" does not exist or if The Extracted File is not found.
+  - The path to The Chef CLI executable is provided in the environment variable `CHEF_CLI_PATH`.
+  - For the validation of The Normalization Function to be successful The Chef CLI must return an empty output. The implementation of The Extraction Function directly infuences the output of The Chef CLI.
+  - The stdout and stderr of The Chef CLI *must* be printed to the console.
+  - The test should always fail if The Chef CLI is not available.

templates/data_extraction_acceptance_test.plain

@@ -1,6 +1,6 @@
-  - Test The Extraction Function using the resource [{{ resource_name }}](test_data/{{ resource_name }}).
-    - Expect The Callback Server to receive from DevRev a **single** event with "event_type" that equals "EXTRACTION_DATA_DONE".
-    - Let `artifact_array = <event_from_callback_server>["event_data"]["artifacts"]` (array of objects).
-      - Expect `len(artifact_array) > 0`.
-      - Out of `artifact_array`, expect one of the elements to have "item_type" equal to "{{ artifact_name }}" (`:= {{ artifact_name }}_artifact`) and `{{ artifact_name }}_artifact["item_count"]={{ expected_item_count }}`.
-      - Note: If `{{ artifact_name }}_artifact["item_count"] < {{ expected_item_count }}`, this indicates that not all the {{ artifact_name }} data was extracted.
+Test The Extraction Function using the resource [{{ resource_name }}](test_data/{{ resource_name }}).
+  - Expect The Callback Server to receive from DevRev a **single** event with "event_type" that equals "EXTRACTION_DATA_DONE".
+  - Let `artifact_array = <event_from_callback_server>["event_data"]["artifacts"]` (array of objects).
+    - Expect `len(artifact_array) > 0`.
+    - Out of `artifact_array`, expect one of the elements to have "item_type" equal to "{{ artifact_name }}" (`:= {{ artifact_name }}_artifact`) and `{{ artifact_name }}_artifact["item_count"]={{ expected_item_count }}`.
+    - Note: If `{{ artifact_name }}_artifact["item_count"] < {{ expected_item_count }}`, this indicates that not all the {{ artifact_name }} data was extracted.

templates/test_rate_limiting_during_data_extraction.plain

@@ -1,7 +1,7 @@
-  - Test the rate limiting of The Function "{{ function_name }}" with the following flow:
-    - Make a POST request to "http://localhost:8004/start_rate_limiting" with body `{ "test_name": <identifyer of the test>}`.
-    - Invoke The Function "{{ function_name }}" with valid credentials and all required parameters.
-    - Expect: `status_code = 429`.
-    - Expect: `api_delay > 0` and `api_delay <= 3`. If api_delay is bigger than 3, assume the problem is that we did not calculate the api_delay correctly in The Implementation Code.
-    - Make a POST request to "http://localhost:8004/end_rate_limiting".
+Test the rate limiting of The Function "{{ function_name }}" with the following flow:
+  - Make a POST request to "http://localhost:8004/start_rate_limiting" with body `{ "test_name": <identifyer of the test>}`.
+  - Invoke The Function "{{ function_name }}" with valid credentials and all required parameters.
+  - Expect: `status_code = 429`.
+  - Expect: `api_delay > 0` and `api_delay <= 3`. If api_delay is bigger than 3, assume the problem is that we did not calculate the api_delay correctly in The Implementation Code.
+  - Make a POST request to "http://localhost:8004/end_rate_limiting".
 

templates/test_rate_limiting_during_emitting_events.plain

@@ -1,5 +1,5 @@
-  - When the input event is "{{ input_event_name }}", the Extraction Function must handle rate limiting using the following test flow:
-    - Step 1: Make a POST request to "http://localhost:8004/start_rate_limiting" with body `{ "test_name": <identifyer of the test>}`.
-    - Step 2: Invoke The Extraction Function using the resource [{{ resource_name }}](test_data/{{ resource_name }}).
-      - Expect The Callback Server to receive *a single* event with "event_type" "{{ expected_output_event_type }}".
-    - Step 3: Make a POST request to "http://localhost:8004/end_rate_limiting".
+When the input event is "{{ input_event_name }}", the Extraction Function must handle rate limiting using the following test flow:
+  - Step 1: Make a POST request to "http://localhost:8004/start_rate_limiting" with body `{ "test_name": <identifyer of the test>}`.
+  - Step 2: Invoke The Extraction Function using the resource [{{ resource_name }}](test_data/{{ resource_name }}).
+    - Expect The Callback Server to receive *a single* event with "event_type" "{{ expected_output_event_type }}".
+  - Step 3: Make a POST request to "http://localhost:8004/end_rate_limiting".