chore: add sync type and validation for orchestrator TaskSync (NangoHQ#2256)

TBonnin · web-flow · commit 124f67cd0c4f · 2024-06-04T19:27:15.000Z
Just the types and validation. No sync are yet scheduled or processed

## Checklist before requesting a review (skip if just adding/editing
APIs &amp; templates)
- [ ] I added tests, otherwise the reason is: 
- [ ] I added observability, otherwise the reason is:
- [ ] I added analytics, otherwise the reason is:
diff --git a/packages/jobs/lib/processor/handler.ts b/packages/jobs/lib/processor/handler.ts
@@ -1,4 +1,4 @@
-import type { OrchestratorTask, TaskWebhook, TaskAction, TaskPostConnection } from '@nangohq/nango-orchestrator';
+import type { OrchestratorTask, TaskWebhook, TaskAction, TaskPostConnection, TaskSync } from '@nangohq/nango-orchestrator';
 import { jsonSchema } from '@nangohq/nango-orchestrator';
 import type { JsonValue } from 'type-fest';
 import { Err, Ok } from '@nangohq/utils';
@@ -13,6 +13,9 @@ export async function handler(task: OrchestratorTask): Promise<Result<JsonValue>
     task.abortController.signal.onabort = () => {
         abort(task);
     };
+    if (task.isSync()) {
+        return sync(task);
+    }
     if (task.isAction()) {
         return action(task);
     }
@@ -30,6 +33,10 @@ async function abort(_task: OrchestratorTask): Promise<Result<void>> {
     return Ok(undefined);
 }
 
+async function sync(task: TaskSync): Promise<Result<JsonValue>> {
+    return Err(`Not implemented: ${JSON.stringify({ taskId: task.id })}`);
+}
+
 async function action(task: TaskAction): Promise<Result<JsonValue>> {
     const providerConfig = await configService.getProviderConfig(task.connection.provider_config_key, task.connection.environment_id);
     if (providerConfig === null) {
diff --git a/packages/orchestrator/lib/clients/client.ts b/packages/orchestrator/lib/clients/client.ts
@@ -15,9 +15,6 @@ import type {
     ExecuteReturn,
     ExecuteWebhookProps,
     ExecutePostConnectionProps,
-    TaskAction,
-    TaskWebhook,
-    TaskPostConnection,
     OrchestratorTask
 } from './types.js';
 import { validateTask } from './validate.js';
@@ -155,15 +152,7 @@ export class OrchestratorClient {
         return this.execute(schedulingProps);
     }
 
-    public async search({
-        ids,
-        groupKey,
-        limit
-    }: {
-        ids?: string[];
-        groupKey?: string;
-        limit?: number;
-    }): Promise<Result<(TaskWebhook | TaskAction | TaskPostConnection)[], ClientError>> {
+    public async search({ ids, groupKey, limit }: { ids?: string[]; groupKey?: string; limit?: number }): Promise<Result<OrchestratorTask[], ClientError>> {
         const body = {
             ...(ids ? { ids } : {}),
             ...(groupKey ? { groupKey } : {}),
@@ -239,13 +228,7 @@ export class OrchestratorClient {
         }
     }
 
-    public async succeed({
-        taskId,
-        output
-    }: {
-        taskId: string;
-        output: JsonValue;
-    }): Promise<Result<TaskAction | TaskWebhook | TaskPostConnection, ClientError>> {
+    public async succeed({ taskId, output }: { taskId: string; output: JsonValue }): Promise<Result<OrchestratorTask, ClientError>> {
         const res = await this.routeFetch(putTaskRoute)({
             params: { taskId },
             body: { output, state: 'SUCCEEDED' }
@@ -265,7 +248,7 @@ export class OrchestratorClient {
         }
     }
 
-    public async failed({ taskId, error }: { taskId: string; error: Error }): Promise<Result<TaskAction | TaskWebhook | TaskPostConnection, ClientError>> {
+    public async failed({ taskId, error }: { taskId: string; error: Error }): Promise<Result<OrchestratorTask, ClientError>> {
         const output = { name: error.name, message: error.message };
         const res = await this.routeFetch(putTaskRoute)({
             params: { taskId },
@@ -286,7 +269,7 @@ export class OrchestratorClient {
         }
     }
 
-    public async cancel({ taskId, reason }: { taskId: string; reason: string }): Promise<Result<TaskAction | TaskWebhook | TaskPostConnection, ClientError>> {
+    public async cancel({ taskId, reason }: { taskId: string; reason: string }): Promise<Result<OrchestratorTask, ClientError>> {
         const res = await this.routeFetch(putTaskRoute)({
             params: { taskId },
             body: { output: reason, state: 'CANCELLED' }
diff --git a/packages/orchestrator/lib/clients/processor.integration.test.ts b/packages/orchestrator/lib/clients/processor.integration.test.ts
@@ -8,7 +8,7 @@ import { EventsHandler } from '../events.js';
 import { Ok, Err } from '@nangohq/utils';
 import type { Result } from '@nangohq/utils';
 import type { JsonValue } from 'type-fest';
-import type { OrchestratorTask, TaskAction, TaskWebhook, TaskPostConnection } from './types.js';
+import type { OrchestratorTask } from './types.js';
 import { tracer } from 'dd-trace';
 
 const dbClient = getTestDbClient();
@@ -99,7 +99,7 @@ describe('OrchestratorProcessor', async () => {
     });
 });
 
-async function processN(handler: (task: TaskAction | TaskWebhook | TaskPostConnection) => Promise<Result<JsonValue>>, groupKey: string, n: number) {
+async function processN(handler: (task: OrchestratorTask) => Promise<Result<JsonValue>>, groupKey: string, n: number) {
     const processor = new OrchestratorProcessor({
         handler,
         opts: { orchestratorClient, groupKey, maxConcurrency: n, checkForTerminatedInterval: 100 }
diff --git a/packages/orchestrator/lib/clients/types.ts b/packages/orchestrator/lib/clients/types.ts
@@ -5,6 +5,18 @@ import type { TaskState } from '@nangohq/scheduler';
 
 export type SchedulingProps = Omit<PostSchedule['Body'], 'scheduling'>;
 
+interface SyncArgs {
+    syncId: string;
+    syncName: string;
+    debug: boolean;
+    connection: {
+        id: number;
+        provider_config_key: string;
+        environment_id: number;
+        connection_id: string;
+    };
+    syncJobId: number;
+}
 interface ActionArgs {
     actionName: string;
     connection: {
@@ -46,20 +58,40 @@ export type ExecuteActionProps = Omit<ExecuteProps, 'args'> & { args: ActionArgs
 export type ExecuteWebhookProps = Omit<ExecuteProps, 'args'> & { args: WebhookArgs };
 export type ExecutePostConnectionProps = Omit<ExecuteProps, 'args'> & { args: PostConnectionArgs };
 
-export type OrchestratorTask = TaskAction | TaskWebhook | TaskPostConnection;
+export type OrchestratorTask = TaskSync | TaskAction | TaskWebhook | TaskPostConnection;
 
 interface TaskCommonFields {
     id: string;
     name: string;
     state: TaskState;
 }
 interface TaskCommon extends TaskCommonFields {
+    isSync(this: OrchestratorTask): this is TaskSync;
     isWebhook(this: OrchestratorTask): this is TaskWebhook;
     isAction(this: OrchestratorTask): this is TaskAction;
     isPostConnection(this: OrchestratorTask): this is TaskPostConnection;
     abortController: AbortController;
 }
 
+export interface TaskSync extends TaskCommon, SyncArgs {}
+export function TaskSync(props: TaskCommonFields & SyncArgs): TaskSync {
+    return {
+        id: props.id,
+        name: props.name,
+        syncId: props.syncId,
+        syncName: props.syncName,
+        syncJobId: props.syncJobId,
+        debug: props.debug,
+        connection: props.connection,
+        state: props.state,
+        abortController: new AbortController(),
+        isSync: () => true,
+        isWebhook: () => false,
+        isAction: () => false,
+        isPostConnection: () => false
+    };
+}
+
 export interface TaskAction extends TaskCommon, ActionArgs {}
 export function TaskAction(props: TaskCommonFields & ActionArgs): TaskAction {
     return {
@@ -71,6 +103,7 @@ export function TaskAction(props: TaskCommonFields & ActionArgs): TaskAction {
         activityLogId: props.activityLogId,
         input: props.input,
         abortController: new AbortController(),
+        isSync: () => false,
         isWebhook: () => false,
         isAction: () => true,
         isPostConnection: () => false
@@ -89,6 +122,7 @@ export function TaskWebhook(props: TaskCommonFields & WebhookArgs): TaskWebhook
         activityLogId: props.activityLogId,
         input: props.input,
         abortController: new AbortController(),
+        isSync: () => false,
         isWebhook: () => true,
         isAction: () => false,
         isPostConnection: () => false
@@ -106,6 +140,7 @@ export function TaskPostConnection(props: TaskCommonFields & PostConnectionArgs)
         fileLocation: props.fileLocation,
         activityLogId: props.activityLogId,
         abortController: new AbortController(),
+        isSync: () => false,
         isWebhook: () => false,
         isAction: () => false,
         isPostConnection: () => true
diff --git a/packages/orchestrator/lib/clients/validate.ts b/packages/orchestrator/lib/clients/validate.ts
@@ -1,17 +1,63 @@
 import { taskStates } from '@nangohq/scheduler';
 import type { Task } from '@nangohq/scheduler';
-import { TaskAction, TaskWebhook, TaskPostConnection } from './types.js';
+import type { OrchestratorTask } from './types.js';
+import { TaskAction, TaskWebhook, TaskPostConnection, TaskSync } from './types.js';
 import { z } from 'zod';
-import { actionArgsSchema, webhookArgsSchema, postConnectionArgsSchema } from '../routes/v1/postSchedule.js';
 import { Err, Ok } from '@nangohq/utils';
 import type { Result } from '@nangohq/utils';
+import { jsonSchema } from '../utils/validation.js';
+
+export const commonSchemaArgsFields = {
+    connection: z.object({
+        id: z.number().positive(),
+        connection_id: z.string().min(1),
+        provider_config_key: z.string().min(1),
+        environment_id: z.number().positive()
+    })
+};
+
+export const syncArgsSchema = z.object({
+    type: z.literal('sync'),
+    syncId: z.string().min(1),
+    syncName: z.string().min(1),
+    syncJobId: z.number().int().positive(),
+    debug: z.boolean(),
+    ...commonSchemaArgsFields
+});
+
+export const actionArgsSchema = z.object({
+    type: z.literal('action'),
+    actionName: z.string().min(1),
+    activityLogId: z.number().positive(),
+    input: jsonSchema,
+    ...commonSchemaArgsFields
+});
+export const webhookArgsSchema = z.object({
+    type: z.literal('webhook'),
+    webhookName: z.string().min(1),
+    parentSyncName: z.string().min(1),
+    activityLogId: z.number().positive(),
+    input: jsonSchema,
+    ...commonSchemaArgsFields
+});
+export const postConnectionArgsSchema = z.object({
+    type: z.literal('post-connection-script'),
+    postConnectionName: z.string().min(1),
+    fileLocation: z.string().min(1),
+    activityLogId: z.number().positive(),
+    ...commonSchemaArgsFields
+});
 
 const commonSchemaFields = {
     id: z.string().uuid(),
     name: z.string().min(1),
     groupKey: z.string().min(1),
     state: z.enum(taskStates)
 };
+const syncSchema = z.object({
+    ...commonSchemaFields,
+    payload: syncArgsSchema
+});
 const actionSchema = z.object({
     ...commonSchemaFields,
     payload: actionArgsSchema
@@ -25,7 +71,22 @@ const postConnectionSchema = z.object({
     payload: postConnectionArgsSchema
 });
 
-export function validateTask(task: Task): Result<TaskAction | TaskWebhook | TaskPostConnection> {
+export function validateTask(task: Task): Result<OrchestratorTask> {
+    const sync = syncSchema.safeParse(task);
+    if (sync.success) {
+        return Ok(
+            TaskSync({
+                id: sync.data.id,
+                state: sync.data.state,
+                name: sync.data.name,
+                syncId: sync.data.payload.syncId,
+                syncName: sync.data.payload.syncName,
+                connection: sync.data.payload.connection,
+                syncJobId: sync.data.payload.syncJobId,
+                debug: sync.data.payload.debug
+            })
+        );
+    }
     const action = actionSchema.safeParse(task);
     if (action.success) {
         return Ok(
diff --git a/packages/orchestrator/lib/routes/v1/postSchedule.ts b/packages/orchestrator/lib/routes/v1/postSchedule.ts
@@ -4,8 +4,8 @@ import type { Scheduler } from '@nangohq/scheduler';
 import type { ApiError, Endpoint } from '@nangohq/types';
 import type { EndpointRequest, EndpointResponse, RouteHandler, Route } from '@nangohq/utils';
 import { validateRequest } from '@nangohq/utils';
-import { jsonSchema } from '../../utils/validation.js';
 import type { TaskType } from '../../types.js';
+import { syncArgsSchema, actionArgsSchema, postConnectionArgsSchema, webhookArgsSchema } from '../../clients/validate.js';
 
 const path = '/v1/schedule';
 const method = 'POST';
@@ -32,40 +32,13 @@ export type PostSchedule = Endpoint<{
     Success: { taskId: string };
 }>;
 
-const commonSchemaFields = {
-    connection: z.object({
-        id: z.number().positive(),
-        connection_id: z.string().min(1),
-        provider_config_key: z.string().min(1),
-        environment_id: z.number().positive()
-    }),
-    activityLogId: z.number().positive(),
-    input: jsonSchema
-};
-
-export const actionArgsSchema = z.object({
-    type: z.literal('action'),
-    actionName: z.string().min(1),
-    ...commonSchemaFields
-});
-export const webhookArgsSchema = z.object({
-    type: z.literal('webhook'),
-    webhookName: z.string().min(1),
-    parentSyncName: z.string().min(1),
-    ...commonSchemaFields
-});
-export const postConnectionArgsSchema = z.object({
-    type: z.literal('post-connection-script'),
-    postConnectionName: z.string().min(1),
-    fileLocation: z.string().min(1),
-    ...commonSchemaFields
-});
-
 const validate = validateRequest<PostSchedule>({
     parseBody: (data: any) => {
         function argsSchema(data: any) {
             if ('args' in data && 'type' in data.args) {
                 switch (data.args.type) {
+                    case 'sync':
+                        return syncArgsSchema;
                     case 'action':
                         return actionArgsSchema;
                     case 'webhook':
