feat(orchestrator): implement task processor

.env.example

@@ -99,3 +99,6 @@ MAILGUN_API_KEY=
 
 # Redis (optional)
 NANGO_REDIS_URL=
+
+# Orchestrator
+ORCHESTRATOR_SERVICE_URL="http://localhost:3008"

packages/jobs/Dockerfile

@@ -16,6 +16,7 @@ COPY packages/data-ingestion/ packages/data-ingestion/
 COPY packages/jobs/ packages/jobs/
 COPY packages/logs/ packages/logs/
 COPY packages/runner/ packages/runner/
+COPY packages/scheduler/ packages/scheduler/
 COPY packages/orchestrator/ packages/orchestrator/
 COPY package*.json ./
 

packages/jobs/lib/app.ts

@@ -1,25 +1,41 @@
 import './tracer.js';
 import { Temporal } from './temporal.js';
+import { Processor } from './processor/processor.js';
 import { server } from './server.js';
 import { cronAutoIdleDemo } from './crons/autoIdleDemo.js';
 import { deleteOldActivityLogs } from './crons/deleteOldActivities.js';
 import { deleteSyncsData } from './crons/deleteSyncsData.js';
 import { reconcileTemporalSchedules } from './crons/reconcileTemporalSchedules.js';
 import { getLogger, stringifyError } from '@nangohq/utils';
-import { JOBS_PORT } from './constants.js';
-import { db } from '@nangohq/shared';
+import { db, featureFlags } from '@nangohq/shared';
+import { envs } from './env.js';
 
 const logger = getLogger('Jobs');
 
 try {
-    server.listen(JOBS_PORT);
-    logger.info(`🚀 service ready at http://localhost:${JOBS_PORT}`);
-    const temporalNs = process.env['TEMPORAL_NAMESPACE'] || 'default';
+    const port = envs.NANGO_JOBS_PORT;
+    const temporalNs = envs.TEMPORAL_NAMESPACE;
+    const orchestratorServiceUrl = envs.ORCHESTRATOR_SERVICE_URL;
+    server.listen(port);
+    logger.info(`🚀 service ready at http://localhost:${port}`);
     const temporal = new Temporal(temporalNs);
+    const processor = new Processor(orchestratorServiceUrl);
 
     // This promise never resolve
     void temporal.start();
 
+    // Start processor
+    const getFlag = () => featureFlags.isEnabled('orchestrator:dryrun:process', 'global', false, false);
+    const processorFlagTimer = setInterval(async () => {
+        const isProcessorEnabled = await getFlag();
+        if (isProcessorEnabled && processor.isStopped()) {
+            processor.start();
+        }
+        if (!isProcessorEnabled && !processor.isStopped()) {
+            processor.stop();
+        }
+    }, 1000);
+
     db.enableMetrics();
 
     // Register recurring tasks
@@ -31,6 +47,8 @@ try {
     // handle SIGTERM
     process.on('SIGTERM', () => {
         temporal.stop();
+        processor.stop();
+        clearInterval(processorFlagTimer);
         server.server.close(() => {
             process.exit(0);
         });

packages/jobs/lib/env.ts

@@ -0,0 +1,3 @@
+import { ENVS, parseEnvs } from '@nangohq/utils';
+
+export const envs = parseEnvs(ENVS.required({ ORCHESTRATOR_SERVICE_URL: true }));

packages/jobs/lib/processor/handler.ts

@@ -0,0 +1,34 @@
+import type { OrchestratorTask, TaskWebhook, TaskAction } from '@nangohq/nango-orchestrator';
+import type { JsonValue } from 'type-fest';
+import { Err, Ok } from '@nangohq/utils';
+import type { Result } from '@nangohq/utils';
+
+export async function handler(task: OrchestratorTask): Promise<Result<JsonValue>> {
+    task.abortController.signal.onabort = () => {
+        abort(task);
+    };
+    if (task.isAction()) {
+        return action(task);
+    }
+    if (task.isWebhook()) {
+        return webhook(task);
+    }
+    return Err(`Unreachable`);
+}
+
+async function abort(_task: OrchestratorTask): Promise<Result<void>> {
+    // TODO: Implement abort processing
+    return Ok(undefined);
+}
+
+async function action(task: TaskAction): Promise<Result<JsonValue>> {
+    // TODO: Implement action processing
+    // Returning a successful result for now
+    return Ok({ taskId: task.id, dryrun: true });
+}
+
+async function webhook(task: TaskWebhook): Promise<Result<JsonValue>> {
+    // TODO: Implement action processing
+    // Returning an error for now
+    return Err(`Not implemented: ${JSON.stringify({ taskId: task.id })}`);
+}

packages/jobs/lib/processor/processor.ts

@@ -0,0 +1,50 @@
+import { getLogger } from '@nangohq/utils';
+import { ProcessorWorker } from './processor.worker.js';
+
+const logger = getLogger('jobs.processor');
+
+export class Processor {
+    private orchestratorServiceUrl: string;
+    private workers: ProcessorWorker[];
+    private stopped: boolean;
+
+    constructor(orchestratorServiceUrl: string) {
+        this.orchestratorServiceUrl = orchestratorServiceUrl;
+        this.workers = [];
+        this.stopped = true;
+    }
+
+    isStopped() {
+        return this.stopped;
+    }
+
+    start() {
+        logger.info('Starting task processors');
+        try {
+            const actionWorker = new ProcessorWorker({
+                orchestratorUrl: this.orchestratorServiceUrl,
+                groupKey: 'action',
+                maxConcurrency: 100
+            });
+            actionWorker.start();
+
+            const webhookWorker = new ProcessorWorker({
+                orchestratorUrl: this.orchestratorServiceUrl,
+                groupKey: 'webhook',
+                maxConcurrency: 50
+            });
+            webhookWorker.start();
+            this.workers = [actionWorker, webhookWorker];
+            this.stopped = false;
+        } catch (e) {
+            logger.error(e);
+        }
+    }
+
+    stop() {
+        if (this.workers) {
+            this.workers.forEach((worker) => worker.stop());
+        }
+        this.stopped = true;
+    }
+}

packages/jobs/lib/processor/processor.worker.boot.ts

@@ -0,0 +1,11 @@
+import { isMainThread, parentPort, workerData } from 'node:worker_threads';
+import { getLogger } from '@nangohq/utils';
+import { ProcessorChild } from './processor.worker.js';
+
+const logger = getLogger('processor.worker.boot');
+
+if (!isMainThread && parentPort) {
+    new ProcessorChild(parentPort, workerData);
+} else {
+    logger.error('Processor should not be instantiated in the main thread');
+}

packages/jobs/lib/processor/processor.worker.ts

@@ -0,0 +1,94 @@
+import * as fs from 'fs';
+import type { MessagePort } from 'node:worker_threads';
+import { Worker, isMainThread } from 'node:worker_threads';
+import { getLogger, stringifyError } from '@nangohq/utils';
+import { OrchestratorClient, OrchestratorProcessor } from '@nangohq/nango-orchestrator';
+import { handler } from './handler.js';
+
+const logger = getLogger('jobs.processor.worker');
+
+export class ProcessorWorker {
+    private worker: Worker | null;
+    constructor({ orchestratorUrl, groupKey, maxConcurrency }: { orchestratorUrl: string; groupKey: string; maxConcurrency: number }) {
+        if (isMainThread) {
+            const url = new URL('../../dist/processor/processor.worker.boot.js', import.meta.url);
+            if (!fs.existsSync(url)) {
+                throw new Error(`Processor worker boot script not found at ${url}`);
+            }
+            this.worker = new Worker(url, { workerData: { orchestratorUrl, groupKey, maxConcurrency } });
+            this.worker.on('error', (err) => {
+                logger.error(`ProcessorWorker exited with error: ${stringifyError(err)}`);
+            });
+            this.worker.on('exit', (code) => {
+                if (code !== 0) {
+                    logger.error(`ProcessorWorker exited with exit code: ${code}`);
+                }
+            });
+        } else {
+            throw new Error('ProcessorWorker should be instantiated in the main thread');
+        }
+    }
+
+    start(): void {
+        this.worker?.postMessage('start');
+    }
+
+    stop(): void {
+        if (this.worker) {
+            this.worker.postMessage('stop');
+            this.worker = null;
+        }
+    }
+}
+
+export class ProcessorChild {
+    private parent: MessagePort;
+    private processor: OrchestratorProcessor;
+    private opts: {
+        orchestratorUrl: string;
+        groupKey: string;
+        maxConcurrency: number;
+    };
+
+    constructor(parent: MessagePort, workerData: { orchestratorUrl: string; groupKey: string; maxConcurrency: number }) {
+        if (isMainThread) {
+            throw new Error('Processor should not be instantiated in the main thread');
+        }
+        if (!workerData.orchestratorUrl || !workerData.groupKey || workerData.maxConcurrency <= 0) {
+            throw new Error(
+                `Missing required options for processor worker. Expecting orchestratorUrl, groupKey, maxConcurrency > 0, got: ${JSON.stringify(workerData)}`
+            );
+        }
+        this.opts = workerData;
+        this.parent = parent;
+        this.parent.on('message', async (msg: 'start' | 'stop') => {
+            switch (msg) {
+                case 'start':
+                    await this.start();
+                    break;
+                case 'stop':
+                    this.stop();
+                    break;
+            }
+        });
+        const client = new OrchestratorClient({ baseUrl: this.opts.orchestratorUrl });
+        this.processor = new OrchestratorProcessor({
+            handler,
+            opts: {
+                orchestratorClient: client,
+                groupKey: this.opts.groupKey,
+                maxConcurrency: this.opts.maxConcurrency
+            }
+        });
+    }
+
+    async start(): Promise<void> {
+        logger.info(`Starting Processor: ${JSON.stringify(this.opts)}`);
+        this.processor.start();
+    }
+
+    stop(): void {
+        logger.info(`Stopping Processor: ${JSON.stringify(this.opts)}`);
+        this.processor.stop();
+    }
+}

packages/jobs/lib/temporal.ts

@@ -5,10 +5,11 @@ import { createRequire } from 'module';
 import * as activities from './activities.js';
 import { SYNC_TASK_QUEUE, WEBHOOK_TASK_QUEUE } from '@nangohq/shared';
 import { isProd, isEnterprise, getLogger } from '@nangohq/utils';
+import { envs } from './env.js';
 
 const logger = getLogger('Jobs.Temporal');
 
-const TEMPORAL_WORKER_MAX_CONCURRENCY = parseInt(process.env['TEMPORAL_WORKER_MAX_CONCURRENCY'] || '0') || 500;
+const TEMPORAL_WORKER_MAX_CONCURRENCY = envs.TEMPORAL_WORKER_MAX_CONCURRENCY;
 
 export class Temporal {
     namespace: string;

packages/jobs/nodemon.json

@@ -1,5 +1,5 @@
 {
-    "watch": ["lib", "../shared/dist", "../utils/dist", "../records/dist", "../data-ingestion/dist", "../logs/dist", "../../.env"],
+    "watch": ["lib", "../shared/dist", "../utils/dist", "../records/dist", "../data-ingestion/dist", "../logs/dist", "../orchestrator/dist", "../../.env"],
     "ext": "js,ts,json",
     "ignore": ["lib/**/*.spec.ts"],
     "exec": "tsx -r dotenv/config lib/app.ts dotenv_config_path=./../../.env",

packages/jobs/package.json

@@ -47,6 +47,7 @@
         "@types/node": "^20.12.2",
         "nodemon": "^3.0.1",
         "typescript": "^5.3.3",
+        "type-fest": "4.14.0",
         "vitest": "0.33.0"
     }
 }