From 8fb4cd47943a6549b1e0c0ed70bf8ce4b243d99f Mon Sep 17 00:00:00 2001 From: radovanjorgic Date: Wed, 29 Jul 2026 09:31:21 +0200 Subject: [PATCH] chore: migrate connector to AirSync SDK v2 --- MIGRATION_REVIEW.md | 46 +++++++++++++++++++ code/package-lock.json | 14 +++--- code/package.json | 2 +- .../external-system/data-denormalization.ts | 2 +- .../external-system/data-normalization.ts | 2 +- .../functions/external-system/http-client.ts | 4 +- code/src/functions/extraction/index.ts | 4 +- .../workers/attachments-extraction.ts | 38 ++++----------- .../extraction/workers/data-extraction.ts | 13 ++---- .../workers/external-sync-units-extraction.ts | 22 ++++----- .../extraction/workers/metadata-extraction.ts | 15 +++--- code/src/functions/loading/index.ts | 4 +- .../loading/workers/load-attachments.ts | 20 ++------ .../functions/loading/workers/load-data.ts | 22 ++------- code/src/test-runner/test-runner.ts | 4 +- 15 files changed, 104 insertions(+), 108 deletions(-) create mode 100644 MIGRATION_REVIEW.md diff --git a/MIGRATION_REVIEW.md b/MIGRATION_REVIEW.md new file mode 100644 index 0000000..4178fd5 --- /dev/null +++ b/MIGRATION_REVIEW.md @@ -0,0 +1,46 @@ +# v2 Migration Review + +Migrated `@devrev/ts-adaas` 1.20.0 → `@devrev/airsync-sdk` 2.0.0-beta.6. +Baseline was already the latest stable v1 (1.20.0, migration-free), so no v1 +baseline upgrade was needed. Only sections needing a second look are noted; the +rest applied cleanly. + +## Sections + +- §1 package rename — high +- §2 `AirdropEvent` → `AirSyncEvent` — high +- §3 worker split (`processExtractionTask` / `processLoadingTask`) — high +- §4 emit → return — high +- §4 onTimeout rewrites — medium: the progress-only `onTimeout` handlers in + `data-extraction` (resumable) and both loading workers were **deleted** — the + SDK's phase-aware default emits the same continuation. The `metadata` and + `external-sync-units` handlers were **kept** and converted to + `return { status: 'error', error: {...} }` because each carried a custom + timeout message that the SDK default would otherwise replace with a generated + one. Confirm these two non-resumable phases should still surface those exact + messages. +- §6 loading pass-through (`load-data`, `load-attachments`) — high +- §6 attachment streaming (`attachments-extraction`) — medium: replaced the v1 + `streamAttachments` result branching (delay/error/done emits) with a straight + `return adapter.streamAttachments(...)`. This also dropped the surrounding + `try/catch` whose only action was `console.error` + swallow (it emitted + nothing). The returned `TaskResult` already encodes delay/error/success, so + behavior is preserved or improved; confirm no bespoke catch handling is + wanted here. +- §7 external sync units via repo — high (connector already pushed ESUs to the + `EXTERNAL_SYNC_UNITS` repo; only the trailing emit changed to `return`) +- §10 `Mappers` `.data` unwrap — high (only a commented-out template example was + updated; no live mapper read exists in this connector) + +Verified no-ops (no code changed): §5 (no `WorkerAdapter` annotations/ +constructions), §8 (no SDK fields in `ExtractorState`/`LoaderState`, no +`lastSync*`, no `AdapterState`), §9 (connector imports `axios` directly, never +the SDK axios surface; no `statusCode` on the streaming error), §11–§15 (no +legacy modules, deleted enum members, deep imports, or `workerPath`), §16 (the +connector has no jest test suite). + +## Verification + +tsc ✓ · lint ✓ · build ✓ · test — n/a (no jest suite; `npm test` reports "No +tests found" on both the pre-migration and post-migration tree — a pre-existing +condition, not caused by this migration). diff --git a/code/package-lock.json b/code/package-lock.json index 6cd8ed8..d991474 100644 --- a/code/package-lock.json +++ b/code/package-lock.json @@ -9,7 +9,7 @@ "version": "1.1.6", "license": "ISC", "dependencies": { - "@devrev/ts-adaas": "1.20.0", + "@devrev/airsync-sdk": "2.0.0-beta.6", "@devrev/typescript-sdk": "1.1.78", "axios": "^1.16.0", "dotenv": "^16.0.3", @@ -1877,19 +1877,19 @@ "@jridgewell/sourcemap-codec": "^1.4.10" } }, - "node_modules/@devrev/ts-adaas": { - "version": "1.20.0", - "resolved": "https://registry.npmjs.org/@devrev/ts-adaas/-/ts-adaas-1.20.0.tgz", - "integrity": "sha512-u6UmWnmxyagPMZ1M51bzZ6FLwKWWTyc4sG4DivQiNkuhdOjMiN3jjowHjQwI5bMJ5ySVNz3isb5FJyRcsLJTdA==", + "node_modules/@devrev/airsync-sdk": { + "version": "2.0.0-beta.6", + "resolved": "https://registry.npmjs.org/@devrev/airsync-sdk/-/airsync-sdk-2.0.0-beta.6.tgz", + "integrity": "sha512-Xw0a0P3ad6rmKi+iU/ZdM/7s5+PgPHYkiPy4cSkj3Vlph5ZuvAxrZ8wVmmmnYQUFLPj0t0xLQ9T72Pu7DoDv5Q==", "license": "ISC", "dependencies": { - "@devrev/typescript-sdk": "^1.1.78", + "@devrev/typescript-sdk": "^1.1.76", "axios": "^1.17.0", "axios-retry": "^4.5.0", "form-data": "^4.0.4", "js-jsonl": "^1.1.1", "ts-node": "^10.9.2", - "yargs": "^17.7.3" + "yargs": "^17.7.2" } }, "node_modules/@devrev/typescript-sdk": { diff --git a/code/package.json b/code/package.json index 18dfe4b..2699036 100644 --- a/code/package.json +++ b/code/package.json @@ -58,7 +58,7 @@ "yargs": "^17.6.2" }, "dependencies": { - "@devrev/ts-adaas": "1.20.0", + "@devrev/airsync-sdk": "2.0.0-beta.6", "@devrev/typescript-sdk": "1.1.78", "axios": "^1.16.0", "dotenv": "^16.0.3", diff --git a/code/src/functions/external-system/data-denormalization.ts b/code/src/functions/external-system/data-denormalization.ts index 3ea0cfb..7949f8d 100644 --- a/code/src/functions/external-system/data-denormalization.ts +++ b/code/src/functions/external-system/data-denormalization.ts @@ -1,4 +1,4 @@ -import { ExternalSystemAttachment, ExternalSystemItem } from '@devrev/ts-adaas'; +import { ExternalSystemAttachment, ExternalSystemItem } from '@devrev/airsync-sdk'; import { ExternalAttachment, ExternalTodo } from './types'; export function denormalizeTodo(item: ExternalSystemItem): ExternalTodo { diff --git a/code/src/functions/external-system/data-normalization.ts b/code/src/functions/external-system/data-normalization.ts index 6ea7d54..06a8281 100644 --- a/code/src/functions/external-system/data-normalization.ts +++ b/code/src/functions/external-system/data-normalization.ts @@ -1,4 +1,4 @@ -import { ExternalSyncUnit, NormalizedAttachment, NormalizedItem } from '@devrev/ts-adaas'; +import { ExternalSyncUnit, NormalizedAttachment, NormalizedItem } from '@devrev/airsync-sdk'; import { ExternalAttachment, ExternalTodo, ExternalTodoList, ExternalUser } from './types'; // TODO: Replace with your actual normalization functions that will be used to diff --git a/code/src/functions/external-system/http-client.ts b/code/src/functions/external-system/http-client.ts index c729d68..93b9276 100644 --- a/code/src/functions/external-system/http-client.ts +++ b/code/src/functions/external-system/http-client.ts @@ -1,4 +1,4 @@ -import { AirdropEvent, ExternalSystemItemLoadingResponse } from '@devrev/ts-adaas'; +import { AirSyncEvent, ExternalSystemItemLoadingResponse } from '@devrev/airsync-sdk'; import { ExternalAttachment, ExternalTodo, ExternalTodoList, ExternalUser } from './types'; // --------------------------------------------------------------------------- @@ -85,7 +85,7 @@ export class HttpClient { private apiEndpoint: string; private apiToken: string; - constructor(event: AirdropEvent) { + constructor(event: AirSyncEvent) { // TODO: Replace with the API endpoint of the external system. This is // passed through the event payload (e.g. event.payload.connection_data.org_id // or a keyring subdomain field). diff --git a/code/src/functions/extraction/index.ts b/code/src/functions/extraction/index.ts index 4bfa21d..6694f6f 100644 --- a/code/src/functions/extraction/index.ts +++ b/code/src/functions/extraction/index.ts @@ -1,4 +1,4 @@ -import { AirdropEvent, spawn } from '@devrev/ts-adaas'; +import { AirSyncEvent, spawn } from '@devrev/airsync-sdk'; import initialDomainMapping from '../external-system/initial_domain_mapping.json'; @@ -19,7 +19,7 @@ export const initialExtractorState: ExtractorState = { attachments: { completed: false }, }; -const run = async (events: AirdropEvent[]) => { +const run = async (events: AirSyncEvent[]) => { for (const event of events) { await spawn({ event, diff --git a/code/src/functions/extraction/workers/attachments-extraction.ts b/code/src/functions/extraction/workers/attachments-extraction.ts index c7a3448..c9aa5ad 100644 --- a/code/src/functions/extraction/workers/attachments-extraction.ts +++ b/code/src/functions/extraction/workers/attachments-extraction.ts @@ -3,9 +3,8 @@ import axios from 'axios'; import { ExternalSystemAttachmentStreamingParams, ExternalSystemAttachmentStreamingResponse, - ExtractorEventType, - processTask, -} from '@devrev/ts-adaas'; + processExtractionTask, +} from '@devrev/airsync-sdk'; // TODO: Replace with function for fetching attachment streams from the // external system. This function should return either a stream of the @@ -47,33 +46,14 @@ async function getFileStream({ } } -processTask({ +processExtractionTask({ task: async ({ adapter }) => { - try { - const response = await adapter.streamAttachments({ - stream: getFileStream, + return adapter.streamAttachments({ + stream: getFileStream, - // TODO: If needed you can specify how many attachments to stream at - // once. Minimum is 1 and maximum is 50. - // batchSize: 10, - }); - - if (response?.delay) { - await adapter.emit(ExtractorEventType.AttachmentExtractionDelayed, { - delay: response.delay, - }); - } else if (response?.error) { - await adapter.emit(ExtractorEventType.AttachmentExtractionError, { - error: response.error, - }); - } else { - await adapter.emit(ExtractorEventType.AttachmentExtractionDone); - } - } catch (error) { - console.error('An error occured while processing a task.', error); - } - }, - onTimeout: async ({ adapter }) => { - await adapter.emit(ExtractorEventType.AttachmentExtractionProgress); + // TODO: If needed you can specify how many attachments to stream at + // once. Minimum is 1 and maximum is 50. + // batchSize: 10, + }); }, }); diff --git a/code/src/functions/extraction/workers/data-extraction.ts b/code/src/functions/extraction/workers/data-extraction.ts index 1f0d055..ee00512 100644 --- a/code/src/functions/extraction/workers/data-extraction.ts +++ b/code/src/functions/extraction/workers/data-extraction.ts @@ -1,4 +1,4 @@ -import { ExtractorEventType, processTask } from '@devrev/ts-adaas'; +import { processExtractionTask } from '@devrev/airsync-sdk'; import { normalizeAttachment, normalizeTodo, normalizeUser } from '../../external-system/data-normalization'; import { HttpClient } from '../../external-system/http-client'; @@ -48,7 +48,7 @@ const itemTypesToExtract: ItemTypeToExtract[] = [ }, ]; -processTask({ +processExtractionTask({ task: async ({ adapter }) => { adapter.initializeRepos(repos); @@ -60,9 +60,9 @@ processTask({ // system. This is just an example how you can iterate over the item types, // extract them, push them to the repo, and save the state. for (const itemTypeToExtract of itemTypesToExtract) { - // If the worker is about to time out, exit early so that `onTimeout` can run and emit progress. + // If the worker is about to time out, hand off for continuation. if (adapter.isTimeout) { - return; + return { status: 'progress' }; } if (!adapter.shouldExtract(itemTypeToExtract.name)) { @@ -75,9 +75,6 @@ processTask({ adapter.state[itemTypeToExtract.name].completed = true; } - await adapter.emit(ExtractorEventType.DataExtractionDone); - }, - onTimeout: async ({ adapter }) => { - await adapter.emit(ExtractorEventType.DataExtractionProgress); + return { status: 'success' }; }, }); diff --git a/code/src/functions/extraction/workers/external-sync-units-extraction.ts b/code/src/functions/extraction/workers/external-sync-units-extraction.ts index a3c8280..5a33f6a 100644 --- a/code/src/functions/extraction/workers/external-sync-units-extraction.ts +++ b/code/src/functions/extraction/workers/external-sync-units-extraction.ts @@ -1,14 +1,13 @@ import { AirSyncDefaultItemTypes, ExternalSyncUnit, - ExtractorEventType, - processTask, -} from '@devrev/ts-adaas'; + processExtractionTask, +} from '@devrev/airsync-sdk'; import { normalizeTodoList } from '../../external-system/data-normalization'; import { HttpClient } from '../../external-system/http-client'; -processTask({ +processExtractionTask({ task: async ({ adapter }) => { adapter.initializeRepos([ { @@ -36,13 +35,12 @@ processTask({ .getRepo(AirSyncDefaultItemTypes.EXTERNAL_SYNC_UNITS) ?.push(externalSyncUnits); - await adapter.emit(ExtractorEventType.ExternalSyncUnitExtractionDone); - }, - onTimeout: async ({ adapter }) => { - await adapter.emit(ExtractorEventType.ExternalSyncUnitExtractionError, { - error: { - message: 'Failed to extract external sync units. Lambda timeout.', - }, - }); + return { status: 'success' }; }, + onTimeout: async () => ({ + status: 'error', + error: { + message: 'Failed to extract external sync units. Lambda timeout.', + }, + }), }); diff --git a/code/src/functions/extraction/workers/metadata-extraction.ts b/code/src/functions/extraction/workers/metadata-extraction.ts index 2d3b0b4..e864f0a 100644 --- a/code/src/functions/extraction/workers/metadata-extraction.ts +++ b/code/src/functions/extraction/workers/metadata-extraction.ts @@ -1,4 +1,4 @@ -import { ExtractorEventType, processTask } from '@devrev/ts-adaas'; +import { processExtractionTask } from '@devrev/airsync-sdk'; import staticExternalDomainMetadata from '../../external-system/external_domain_metadata.json'; @@ -8,7 +8,7 @@ const repos = [ }, ]; -processTask({ +processExtractionTask({ task: async ({ adapter }) => { adapter.initializeRepos(repos); @@ -21,11 +21,10 @@ processTask({ }; await adapter.getRepo('external_domain_metadata')?.push([externalDomainMetadata]); - await adapter.emit(ExtractorEventType.MetadataExtractionDone); - }, - onTimeout: async ({ adapter }) => { - await adapter.emit(ExtractorEventType.MetadataExtractionError, { - error: { message: 'Failed to extract metadata. Lambda timeout.' }, - }); + return { status: 'success' }; }, + onTimeout: async () => ({ + status: 'error', + error: { message: 'Failed to extract metadata. Lambda timeout.' }, + }), }); diff --git a/code/src/functions/loading/index.ts b/code/src/functions/loading/index.ts index 040c6dd..4a65edf 100644 --- a/code/src/functions/loading/index.ts +++ b/code/src/functions/loading/index.ts @@ -1,4 +1,4 @@ -import { AirdropEvent, spawn } from '@devrev/ts-adaas'; +import { AirSyncEvent, spawn } from '@devrev/airsync-sdk'; import initialDomainMapping from '../external-system/initial_domain_mapping.json'; @@ -11,7 +11,7 @@ export interface LoaderState {} // This state will be used as a starting point for the loading process. export const initialLoaderState: LoaderState = {}; -const run = async (events: AirdropEvent[]) => { +const run = async (events: AirSyncEvent[]) => { for (const event of events) { await spawn({ event, diff --git a/code/src/functions/loading/workers/load-attachments.ts b/code/src/functions/loading/workers/load-attachments.ts index b8e988e..2a7ed98 100644 --- a/code/src/functions/loading/workers/load-attachments.ts +++ b/code/src/functions/loading/workers/load-attachments.ts @@ -1,9 +1,8 @@ import { ExternalSystemAttachment, ExternalSystemItemLoadingParams, - LoaderEventType, - processTask, -} from '@devrev/ts-adaas'; + processLoadingTask, +} from '@devrev/airsync-sdk'; import { denormalizeAttachment } from '../../external-system/data-denormalization'; import { HttpClient } from '../../external-system/http-client'; @@ -23,21 +22,10 @@ async function createAttachment({ item, mappers, event }: ExternalSystemItemLoad return createAttachmentResponse; } -processTask({ +processLoadingTask({ task: async ({ adapter }) => { - const { reports, processed_files } = await adapter.loadAttachments({ + return adapter.loadAttachments({ create: createAttachment, }); - - await adapter.emit(LoaderEventType.AttachmentLoadingDone, { - reports, - processed_files, - }); - }, - onTimeout: async ({ adapter }) => { - await adapter.emit(LoaderEventType.AttachmentLoadingProgress, { - reports: adapter.reports, - processed_files: adapter.processedFiles, - }); }, }); diff --git a/code/src/functions/loading/workers/load-data.ts b/code/src/functions/loading/workers/load-data.ts index a7849fe..84af13d 100644 --- a/code/src/functions/loading/workers/load-data.ts +++ b/code/src/functions/loading/workers/load-data.ts @@ -2,9 +2,8 @@ import { ExternalSystemItem, ExternalSystemItemLoadingParams, ExternalSystemItemLoadingResponse, - LoaderEventType, - processTask, -} from '@devrev/ts-adaas'; + processLoadingTask, +} from '@devrev/airsync-sdk'; import { denormalizeTodo } from '../../external-system/data-denormalization'; import { HttpClient } from '../../external-system/http-client'; @@ -50,7 +49,7 @@ async function updateTodo({ // sync_unit: event.payload.event_context.sync_unit, // target: item.id.devrev, // }); - // const todoExternalId = syncMapperRecordResponse.data.sync_mapper_record.external_ids[0]; + // const todoExternalId = syncMapperRecordResponse.sync_mapper_record.external_ids[0]; const todo = denormalizeTodo(item); @@ -58,9 +57,9 @@ async function updateTodo({ return updateTodoResponse; } -processTask({ +processLoadingTask({ task: async ({ adapter }) => { - const { reports, processed_files } = await adapter.loadItemTypes({ + return adapter.loadItemTypes({ itemTypesToLoad: [ { itemType: 'todos', @@ -69,16 +68,5 @@ processTask({ }, ], }); - - await adapter.emit(LoaderEventType.DataLoadingDone, { - reports, - processed_files, - }); - }, - onTimeout: async ({ adapter }) => { - await adapter.emit(LoaderEventType.DataLoadingProgress, { - reports: adapter.reports, - processed_files: adapter.processedFiles, - }); }, }); diff --git a/code/src/test-runner/test-runner.ts b/code/src/test-runner/test-runner.ts index 5182dd3..560d650 100644 --- a/code/src/test-runner/test-runner.ts +++ b/code/src/test-runner/test-runner.ts @@ -2,7 +2,7 @@ import * as dotenv from 'dotenv'; import * as fs from 'fs'; import * as path from 'path'; -import { AirdropEvent, createMockEvent, DeepPartial, EventType, MockServer } from '@devrev/ts-adaas'; +import { AirSyncEvent, createMockEvent, DeepPartial, EventType, MockServer } from '@devrev/airsync-sdk'; import { functionFactory, FunctionFactoryType } from '../function-factory'; @@ -72,7 +72,7 @@ async function runWithFixtureDir(fixturesDir: string, functionName?: FunctionFac const statePath = path.join(fixturesDir, 'state.json'); const extractionScopePath = path.join(fixturesDir, 'extraction_scope.json'); - const fixtureEvent = readFixtureFile>(eventPath); + const fixtureEvent = readFixtureFile>(eventPath); const fixtureState = readFixtureFile>(statePath); const fixtureExtractionScope = readFixtureFile>(extractionScopePath);