express-worker-service.ts

/**
 * Express Worker Service — HTTP server + Conductor workers in one process
 *
 * Demonstrates running an Express.js HTTP server alongside Conductor
 * task workers. The HTTP server exposes endpoints to trigger workflows
 * and check worker health.
 *
 * Prerequisites:
 *   npm install express @types/express
 *
 * Run:
 *   CONDUCTOR_SERVER_URL=http://localhost:8080 npx ts-node examples/express-worker-service.ts
 *
 * Then:
 *   curl http://localhost:3000/health
 *   curl -X POST http://localhost:3000/execute -H 'Content-Type: application/json' -d '{"name":"World"}'
 */
import {
  OrkesClients,
  ConductorWorkflow,
  TaskHandler,
  worker,
  simpleTask,
} from "../src/sdk";
import type { Task } from "../src/open-api";
import type { WorkflowExecutor } from "../src/sdk/clients/workflow";

// ── Workers ─────────────────────────────────────────────────────────
const _greetWorker = worker({ taskDefName: "svc_greet", registerTaskDef: true })(
  async (task: Task) => {
    const name = (task.inputData?.name as string) ?? "World";
    return {
      status: "COMPLETED",
      outputData: { greeting: `Hello, ${name}!`, timestamp: new Date().toISOString() },
    };
  }
);

const _processWorker = worker({ taskDefName: "svc_process", registerTaskDef: true })(
  async (task: Task) => {
    const data = task.inputData?.data as string;
    return {
      status: "COMPLETED",
      outputData: { processed: true, length: data?.length ?? 0 },
    };
  }
);

// ── Express server setup ────────────────────────────────────────────
async function createExpressApp(
  workflowClient: WorkflowExecutor,
  wf: ConductorWorkflow
) {
  // Dynamic import so Express is only required when running this example
  const express = (await import("express")).default;
  const app = express();
  app.use(express.json());

  // Health endpoint
  app.get("/health", (_req, res) => {
    res.json({ status: "UP", workers: ["svc_greet", "svc_process"] });
  });

  // Execute workflow endpoint
  app.post("/execute", async (req, res) => {
    try {
      const run = await wf.execute(req.body);
      res.json({
        workflowId: run.workflowId,
        status: run.status,
        output: run.output,
      });
    } catch (err) {
      res.status(500).json({
        error: err instanceof Error ? err.message : "Unknown error",
      });
    }
  });

  // Start workflow async (returns workflow ID)
  app.post("/start", async (req, res) => {
    try {
      const workflowId = await wf.startWorkflow(req.body);
      res.json({ workflowId });
    } catch (err) {
      res.status(500).json({
        error: err instanceof Error ? err.message : "Unknown error",
      });
    }
  });

  // Get workflow status
  app.get("/workflow/:id", async (req, res) => {
    try {
      const status = await workflowClient.getWorkflow(req.params.id, true);
      res.json({
        workflowId: status.workflowId,
        status: status.status,
        output: status.output,
      });
    } catch (err) {
      res.status(500).json({
        error: err instanceof Error ? err.message : "Unknown error",
      });
    }
  });

  return app;
}

// ── Main ────────────────────────────────────────────────────────────
async function main() {
  const clients = await OrkesClients.from();
  const workflowClient = clients.getWorkflowClient();
  const client = clients.getClient();

  // Register workflow
  const wf = new ConductorWorkflow(workflowClient, "express_service_workflow")
    .description("Workflow triggered by Express HTTP endpoints")
    .add(
      simpleTask("greet_ref", "svc_greet", {
        name: "${workflow.input.name}",
      })
    )
    .add(
      simpleTask("process_ref", "svc_process", {
        data: "${greet_ref.output.greeting}",
      })
    )
    .outputParameters({
      greeting: "${greet_ref.output.greeting}",
      processed: "${process_ref.output.processed}",
    });

  await wf.register(true);

  // Start workers
  const handler = new TaskHandler({ client, scanForDecorated: true });
  await handler.startWorkers();

  // Start Express server
  const app = await createExpressApp(workflowClient, wf);
  const PORT = process.env.PORT ?? 3000;

  const server = app.listen(PORT, () => {
    console.log(`Express server running on http://localhost:${PORT}`);
    console.log(`Workers polling for tasks...`);
    console.log(`\nEndpoints:`);
    console.log(`  GET  /health        — Health check`);
    console.log(`  POST /execute       — Execute workflow (sync)`);
    console.log(`  POST /start         — Start workflow (async)`);
    console.log(`  GET  /workflow/:id  — Get workflow status`);
    console.log(`\nExample:`);
    console.log(
      `  curl -X POST http://localhost:${PORT}/execute -H 'Content-Type: application/json' -d '{"name":"Conductor"}'`
    );
  });

  // Graceful shutdown
  const shutdown = async () => {
    console.log("\nShutting down...");
    server.close();
    await handler.stopWorkers();
    process.exit(0);
  };
  process.on("SIGINT", shutdown);
  process.on("SIGTERM", shutdown);
}

main().catch((err) => {
  console.error(err);
  process.exit(1);
});