Rate limit and attachment support (#24)

## Summary

> Note: Should be reviewed alongside
devrev/airdrop-trello-snap-in#11.

These were the changes needed while I was developing functionality for:

- Attachment extraction
- Rate limiting

These changes include, but not limited to:

- Updates to `mock_devrev_server.py`
- Proxy server implementation
- Improving `attachment-extraction.md`
- Improving document with rules for emitting events
- Added a guide for metadata extraction (copied from the fern API docs)
- Updates in `test_data/`
- Updates in the Plain template

## Related issues
https://app.devrev.ai/devrev/works/ISS-180117

Author: tjazerzen

base_folder/package.json

@@ -52,7 +52,7 @@
     "yargs": "^17.6.2"
   },
   "dependencies": {
-    "@devrev/ts-adaas": "1.5.1",
+    "@devrev/ts-adaas": "1.9.0",
     "@devrev/typescript-sdk": "1.1.63",
     "axios": "^1.9.0",
     "dotenv": "^16.0.3",

base_folder/src/core/types.ts

@@ -0,0 +1,52 @@
+/**
+ * Type definitions for DevRev function inputs and related types
+ */
+
+export type Context = {
+    // ID of the dev org for which the function is being invoked.
+    dev_oid: string;
+    // ID of the automation/command/snap-kit Action/Event Source for which the function is being invoked.
+    source_id: string;
+    // ID of the snap-in as part of which the function is being invoked.
+    snap_in_id: string;
+    // ID of the snap-in Version as part of which the function is being invoked.
+    snap_in_version_id: string;
+    // ID of the service account.
+    service_account_id: string;
+    // This secrets map would contain some secrets which platform would provide to the snap-in.
+    // `service_account_token`: This is the token of the service account which belongs to this snap-in. This can be used to make API calls to DevRev. 
+    // `actor_session_token`: For commands, and snap-kits, where the user is performing some action, this is the token of the user who is performing the action.
+    secrets: Record<string, string>;
+  };
+  
+  export type ExecutionMetadata = {
+    // A unique id for the function invocation. Can be used to filter logs for a particular invocation.
+    request_id: string;
+    // Function name as defined in the manifest being invoked.
+    function_name: string;
+    // Type of event that triggered the function invocation as defined in manifest.
+    event_type: string;
+    // DevRev endpoint to which the function can make API calls.
+    // Example : "https://api.devrev.ai/"
+    devrev_endpoint: string;
+  };
+  
+  export type InputData = {
+    // Map of organization inputs and their corresponding values stored in snap-in.
+    // The values are passed as string and typing need to be handled by the function
+    global_values: Record<string, string>;
+    // Map of event sources and their corresponding ids stored in snap-in.
+    // These could be used to schedule events on a schedule based event source.
+    event_sources: Record<string, string>;
+  };
+  
+  // Event sent to our app.
+  export type FunctionInput = {
+    // Actual payload of the event.
+    payload: Record<string, any>;
+    // Context of the function invocation.
+    context: Context;
+    // Metadata of the function invocation.
+    execution_metadata: ExecutionMetadata;
+    input_data: InputData;
+  };

base_folder/src/core/utils.ts

@@ -0,0 +1,30 @@
+import { AirdropEvent, AirdropMessage } from '@devrev/ts-adaas';
+import { FunctionInput } from './types';
+
+export function convertToAirdropEvent(fi: FunctionInput): AirdropEvent {
+    // Create a properly structured AirdropMessage
+    const airdropMessage: AirdropMessage = {
+      connection_data: fi.payload.connection_data,
+      event_context: {
+        ...fi.payload.event_context,
+        ...fi.context,
+        request_id: fi.execution_metadata?.request_id,
+      },
+      event_type: fi.payload.event_type,
+      event_data: fi.payload.event_data || {},
+    };
+    
+    return {
+      context: {
+        ...fi.context,
+        secrets: {
+          service_account_token: fi.context.secrets?.service_account_token || '',
+          ...fi.context.secrets,
+        },
+      },
+      payload: airdropMessage,
+      execution_metadata: fi.execution_metadata,
+      input_data: fi.input_data,
+    };
+  }
+  

devrev-snapin-template.plain

@@ -11,8 +11,10 @@
   - Please note that EventContextIn and EventContextOut interfaces are deprecated.
 
 - The External Domain Metadata is a JSON object that describes the logical domain model of an external system. It defines the types of external record types, the relationships between those record types, and the metadata associated with each field, including the field’s data type, human-readable name, and other descriptive attributes.
+  - Note: Resource "attachments" should not be included in The External Domain Metadata JSON object even if the external system supports attachments. This is handled automatically by The Airdrop SDK Library.
 
 - The Initial Domain Mapping is a JSON object that defines how the logical domain model described by The External Domain Metadata maps to the target domain model in DevRev. It specifies the correspondence between each external record type and DevRev leaf type, as well as the mapping of individual fields, including transformation logic, value conversions, and any fixed or default values.
+  - Note: Resource "attachments" should not be included in The Initial Domain Mapping JSON object even if the external system supports attachments. This is handled automatically by The Airdrop SDK Library.
 
 - The External Sync Unit Item Count is a numeric field representing the total number of items (such as cards) contained within a given external sync unit. It is accessible as "item_count" field of the external sync unit.
 
@@ -27,7 +29,7 @@
 
 - The Snap-In will run on Node.js as the runtime environment.
 
-- The Snap-In should use The Airdrop SDK Library version "1.8.0" for handling data extraction and loading, pushing data, event-driven actions, state management, and artifact handling.
+- The Snap-In should use The Airdrop SDK Library version "1.9.0" for handling data extraction and loading, pushing data, event-driven actions, state management, and artifact handling.
 
 - Use snake_case for JSON keys.
 
@@ -57,41 +59,59 @@
 - The Conformance Tests should use the The Test Snap-In Server - do not use mock server.
 
 - 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 DevRev Server runs at http://localhost:8003, so it must not be mocked.
 
 - The Worker Data Server is part of The DevRev Server and is accessible at http://localhost:8003/external-worker, so it must not be mocked.
 
+- The API server is running at http://localhost:8004, so it must not be mocked.
+
 - The `jest.setup.js` file is managed externally and cannot be modified within The Conformance Tests.
 
 - All test files from The Conformance Tests must use the `.test.ts` suffix to be discoverable by Jest's default test pattern matching.
 
 - The Conformance Tests should send a single event to the Snap-In Server in the request body without wrapping it in an array.
+  - It is safe to assume that event payload will never be undefined, null, or empty.
+
+- If generating The Acceptance Test that references a JSON file (The Attached JSON Of The Acceptance Test) in the resource, you should:
+  - Store The Attached JSON Of The Acceptance Test in a separate JSON file.
+  - Make sure to replace placeholders of the credentials (in `event["payload"]["connection_data"]["key"]` and `event["payload"]["connection_data"]["org_id"]`) in The Attached JSON Of The Acceptance Test with the actual values (credentials from The API).
 
+- The Conformance Tests should encourage code reuse. Most notably, you should avoid duplication of the following:
+  - Setting up The Callback Server,
+  - Reading environment variables,
+  - Defining the test event payload sent to The Snap-In Server.
 
 # The TypeScript Node.js application boilerplate
 
 ***Non-Functional Requirements:***
 
 - Logic of pushing data to The DevRev Servers must be implemented in the worker thread and invoked with the "spawn" method from The Airdrop SDK Library.
 
+- The Worker Thread must be implemented in TypeScript and *The Worker Thread MUST BE* referenced with a .ts (TypeScript) extension, not .js (JavaScript).
+
 - The Function being implemented *MUST not* be called "extraction" or "loading".
 
 ***Functional Requirements:***
 
 - Implement The Function that only checks if The Function can be invoked.
 
-- Implement The Function called "test_external_sync_units" that provides a test of the 'external sync units' part of the extraction workflow as described in the resource [external-sync-units-extraction.mdx](docs/external-sync-units-extraction.mdx). Only check if the extraction can be triggered and that it successfully completes. Disregard all other details.
+- Implement The Function called "test_external_sync_units" (The External Sync Units Test Function) that provides a test of the 'external sync units' part of the extraction workflow as described in the resource [external-sync-units-extraction.mdx](docs/external-sync-units-extraction.mdx).
+  - If "event_type" equals "EXTRACTION_EXTERNAL_SYNC_UNITS_START", The External Sync Units Test Function should:
+    - Only emit event "EXTRACTION_EXTERNAL_SYNC_UNITS_DONE".
 
   ***Acceptance Tests:***
 
   - Test The Function using the resource [external_sync_unit_check.json](test_data/external_sync_unit_check.json). Test is successful if The Callback Server receives from DevRev an event with "event_type" that equals "EXTRACTION_EXTERNAL_SYNC_UNITS_DONE". The test must not send the event directly to The Callback Server.
 
 - Implement The Function called "data_extraction_check" (The Data Extraction Check Function) that provides a test of the 'data extraction' part of the extraction workflow as described in the resource [data-extraction.md](docs/data-extraction.md). The Data Extraction Check Function should:
-  - Initialize a single repo "users"
-  - normalize the users data using The Normalization Function
-  - push the normalized users data to The DevRev Servers
-  - Note: The Data Extraction Check Function should be simplified and should disregard all other details.
+  - if "event_type" equals "EXTRACTION_DATA_START" or "EXTRACTION_ATTACHMENTS_CONTINUE", The Data Extraction Check Function should:
+    - Initialize a single repo "users"
+    - normalize the users data using The Normalization Function
+    - push the normalized users data to The DevRev Servers
+    - Emit event "EXTRACTION_DATA_DONE".
+    - Note: The Data Extraction Check Function should be simplified and should disregard all other details.
 
   ***Acceptance Tests:***
 

docs/attachment_normalization.md

@@ -0,0 +1,52 @@
+# Attachment normalization
+
+Attachments have a special way of normalization compared to other resources. Function for normalizing "attachments" should return an object "NormalizedAttachment" instead of "NormalizedItem".
+
+## NormalizedAttachment
+
+NormalizedAttachment represents the standardized structure of an attachment after normalization in the Airdrop platform. This interface defines the essential properties needed to identify and link attachments to their parent items.
+
+### Properties
+
+- _url_
+
+  Required. A **string** representing the URL where the attachment can be accessed.
+
+- _id_
+
+  Required. A **string** that uniquely identifies the normalized attachment.
+
+- _file_name_
+
+  Required. A **string** representing the name of the attachment file.
+
+- _parent_id_
+
+  Required. A **string** identifying the parent item this attachment belongs to.
+
+- _author_id_
+
+  Optional. A **string** identifying the author or creator of the attachment.
+
+- _grand_parent_id_
+
+  Optional. A **number** identifying a higher-level parent entity, if applicable.
+
+### Example
+
+```typescript
+export function normalizeAttachment(item: any): NormalizedAttachment {
+  return {
+    id: item.gid,
+    url: item.download_url,
+    file_name: item.name,
+    parent_id: item.parent_id,
+  };
+}
+```
+
+### Further remarks
+
+Note:
+
+- In the example above, parent_id should be the ID of the resource that the attachment belongs to. For example, if we're normalizing an attachment for a task, parent_id should be the ID of the task.

docs/attachments-extraction.md

@@ -22,19 +22,41 @@ import {
 The `getAttachmentStream` function is responsible for fetching and streaming attachments from their source URLs.
 
 ```typescript
+import {
+    axiosClient,
+    ExternalSystemAttachmentStreamingParams,
+    ExternalSystemAttachmentStreamingResponse,
+    axios,
+    serializeAxiosError,
+    ... // Other imports from @devrev/ts-adaas
+  } from '@devrev/ts-adaas';
+
 const getAttachmentStream = async ({
   item,
 }: ExternalSystemAttachmentStreamingParams): Promise<ExternalSystemAttachmentStreamingResponse> => {
+  // IMPORTANT: "url" is not necessarily deployed on the base URL of The API. It could also be an external URL (e.g. https://example.com/attachment.pdf, https://devrev.ai, ...)
   const { id, url } = item;
 
+  // NOTE: Import axiosClient directly from @devrev/ts-adaas
   try {
+    // IMPORTANT: If the URL is protected by authentication from The API, you should also use the appropriate credentials.
     const fileStreamResponse = await axiosClient.get(url, {
       responseType: 'stream',
       headers: {
         'Accept-Encoding': 'identity',
+        'Authorization': ... // TODO: Authorization if needed
       },
     });
 
+    // Check if we were rate limited
+    if (fileStreamResponse.status === 429) {
+      const delay = ... // TODO: Calculate the delay from The API
+      return {
+        delay: delay
+      };
+    }
+
+    // IMPORTANT: "httpStream" should be directly Axios response stream (including headers, data, and everything else)
     return { httpStream: fileStreamResponse };
   } catch (error) {
     // Error handling logic
@@ -102,13 +124,6 @@ processTask({
 - Failed attachment metadata is logged for debugging
 - Error responses include the attachment ID for traceability
 
-### Event Emission
-The worker emits different events based on the processing outcome:
-- `ExtractionAttachmentsDelay`: When processing needs to be delayed
-- `ExtractionAttachmentsError`: When an error occurs
-- `ExtractionAttachmentsDone`: When processing completes successfully
-- `ExtractionAttachmentsProgress`: During timeout handling
-
 ### Timeout Handling
 - On timeout, the current state is posted to the platform
 - A progress event is emitted with 50% completion

docs/data-extraction-rules-for-emitting-events.md

@@ -0,0 +1,39 @@
+# Data Extraction Rules for Emitting Events
+
+## Input event: `EXTRACTION_DATA_START` or `EXTRACTION_DATA_CONTINUE`
+
+Here are the rules when `EXTRACTION_DATA_START` or `EXTRACTION_DATA_CONTINUE` event is received by the snap-in:
+
+- If there's a timeout ("onTimeout" method in "spawn" method), the snap-in should emit an event with type `EXTRACTION_DATA_PROGRESS` (no "data" payload).
+- If the extraction is successful and all the resources have been extracted, the snap-in should emit an event with type `EXTRACTION_DATA_DONE` (no "data" payload).
+- In case of unexpected error, the snap-in should emit an event with type `EXTRACTION_DATA_ERROR` ("data" payload should contain attribute "error" - error message as a string).
+- In case of rate limiting (status code 429) for "data extraction", the snap-in *must* emit an event with type `EXTRACTION_DATA_DELAY`. If emitting event of type `EXTRACTION_DATA_DELAY`, you *must* also provide a "delay" attribute, specifying the delay in seconds as an integer.
+  - Example: `await adapter.emit(ExtractorEventType.ExtractionDataDelay, { delay: "30" });`
+
+## Input event: `EXTRACTION_EXTERNAL_SYNC_UNITS_START`
+
+Here are the rules when `EXTRACTION_EXTERNAL_SYNC_UNITS_START` event is received by the snap-in:
+
+- If "external sync unit extraction" is successful and the snap-in has extracted all the external sync units, the snap-in should emit an event with type `EXTRACTION_EXTERNAL_SYNC_UNITS_DONE` (no "data" payload).
+- In case of unexpected error, the snap-in should emit an event with type `EXTRACTION_EXTERNAL_SYNC_UNITS_ERROR` ("data" payload should contain attribute "error" - error message as a string).
+- In case of rate limiting (status code 429) for "external sync unit extraction", the snap-in should also emit an event with type `EXTRACTION_EXTERNAL_SYNC_UNITS_ERROR`.
+
+## Input event: `EXTRACTION_METADATA_START`
+
+Here are the rules when `EXTRACTION_METADATA_START` event is received by the snap-in:
+- If "metadata extraction" is successful and the snap-in has extracted all the metadata, the snap-in should emit an event with type `EXTRACTION_METADATA_DONE` (no "data" payload).
+- In case of unexpected error, the snap-in should emit an event with type `EXTRACTION_METADATA_ERROR` ("data" payload should contain attribute "error" - error message as a string).
+
+## Input event: `EXTRACTION_ATTACHMENTS_START` or `EXTRACTION_ATTACHMENTS_CONTINUE`
+
+Here are the rules when `EXTRACTION_ATTACHMENTS_START` or `EXTRACTION_ATTACHMENTS_CONTINUE` event is received by the snap-in:
+- If "attachments extraction" is successful and the snap-in has extracted all the attachments, the snap-in should emit an event with type "EXTRACTION_ATTACHMENTS_DONE"
+- If case of unexpected error, the snap-in should emit an event with type "EXTRACTION_ATTACHMENTS_ERROR" ("data" payload should contain attribute "error" - error message as a string).
+- In case of rate limiting (status code 429) for "attachments extraction", the snap-in should also emit an event with type "EXTRACTION_ATTACHMENTS_DELAY". If emitting event of type "EXTRACTION_ATTACHMENTS_DELAY", you *must* also provide a "delay" attribute, specifying the delay in seconds as an integer.
+  - Example: `await adapter.emit(ExtractorEventType.ExtractionAttachmentsDelay, { delay: "30" });`
+- If there's a timeout ("onTimeout" method in "spawn" method), the snap-in should emit an event with type "EXTRACTION_ATTACHMENTS_PROGRESS".
+
+
+## IMPORTANT FOR ALL INPUT EVENTS
+
+- In all cases, only a single event should be emitted.

docs/data-extraction.md

@@ -100,6 +100,8 @@ processTask<ExtractorState>({
 });
 ```
 
+NOTE: Do not call `initializeRepos` multiple times. You should call it only once with all the repositories.
+
 After initialization of repositories using `initializeRepos`,
 items should be then retrieved from the external system and stored in the correct repository by calling the `push` function.
 

docs/function_invocation.mdx

@@ -3,7 +3,7 @@ A function can be invoked synchronously or asynchronously.
 You need to implement the run method in your function. The run method is called when the function is invoked. The run method signature is defined below:
 
 ```typescript
-async function run(events: any[]): any;
+async function run(events: FunctionInput[]): any;
 ```
 
 The value returned from the `run` method is passed back in synchronous execution modes, such as commands, snap kit actions, and event source synchronous execution. In asynchronous execution modes, such as automation execution, the return value is ignored.