Điều phối Workflow Temporal: Xây dựng máy trạng thái phân tán bền vững trong TypeScript

mkdir temporal-order-processing
cd temporal-order-processing
npm init -y
npm install @temporalio/client @temporalio/worker @temporalio/workflow @temporalio/testing typescript ts-node
npm install -D @types/node
npx tsc --init

{
  "compilerOptions": {
    "target": "es2020",
    "module": "commonjs",
    "rootDir": "./",
    "outDir": "./dist",
    "esModuleInterop": true,
    "forceConsistentCasingInFileNames": true,
    "strict": true,
    "skipLibCheck": true,
    "experimentalDecorators": true,
    "emitDecoratorMetadata": true
  }
}

// activities.ts
import { ApplicationFailure } from '@temporalio/workflow';

export async function processPayment(orderId: string, amount: number): Promise<string> {
  console.log(`Processing payment for order ${orderId}, amount ${amount}...`);
  // Simulate external payment gateway call
  await new Promise(resolve => setTimeout(resolve, 1000));
  if (Math.random() < 0.1) { // 10% chance of failure
    console.error(`Payment failed for order ${orderId}`);
    throw ApplicationFailure.create({
      message: `Payment gateway error for order ${orderId}`,
      nonRetryable: false, // Allow retries
    });
  }
  console.log(`Payment successful for order ${orderId}. Transaction ID: tx-${orderId}`);
  return `tx-${orderId}`;
}

export async function deductInventory(orderId: string, itemId: string, quantity: number): Promise<boolean> {
  console.log(`Deducting inventory for order ${orderId}, item ${itemId}, quantity ${quantity}...`);
  // Simulate external inventory service call
  await new Promise(resolve => setTimeout(resolve, 800));
  if (Math.random() < 0.05) { // 5% chance of failure
    console.error(`Inventory deduction failed for order ${orderId}`);
    throw ApplicationFailure.create({
      message: `Inventory service error for order ${orderId}`,
      nonRetryable: false,
    });
  }
  console.log(`Inventory deducted for order ${orderId}.`);
  return true;
}

export async function shipOrder(orderId: string, address: string): Promise<string> {
  console.log(`Shipping order ${orderId} to ${address}...`);
  // Simulate external shipping service call
  await new Promise(resolve => setTimeout(resolve, 1500));
  if (Math.random() < 0.02) { // 2% chance of failure
    console.error(`Shipping failed for order ${orderId}`);
    throw ApplicationFailure.create({
      message: `Shipping service error for order ${orderId}`,
      nonRetryable: false,
    });
  }
  console.log(`Order ${orderId} shipped. Tracking ID: trk-${orderId}`);
  return `trk-${orderId}`;
}

export async function sendConfirmationEmail(orderId: string, email: string): Promise<boolean> {
  console.log(`Sending confirmation email for order ${orderId} to ${email}...`);
  // Simulate email service call
  await new Promise(resolve => setTimeout(resolve, 500));
  console.log(`Confirmation email sent for order ${orderId}.`);
  return true;
}

export async function refundPayment(orderId: string, transactionId: string): Promise<boolean> {
  console.log(`Refunding payment for order ${orderId}, transaction ${transactionId}...`);
  await new Promise(resolve => setTimeout(resolve, 700));
  console.log(`Payment refunded for order ${orderId}.`);
  return true;
}

export async function restoreInventory(orderId: string, itemId: string, quantity: number): Promise<boolean> {
  console.log(`Restoring inventory for order ${orderId}, item ${itemId}, quantity ${quantity}...`);
  await new Promise(resolve => setTimeout(resolve, 600));
  console.log(`Inventory restored for order ${orderId}.`);
  return true;
}

// workflows.ts
import { proxyActivities, ApplicationFailure, workflowInfo } from '@temporalio/workflow';
import * as activities from './activities';

const {
  processPayment,
  deductInventory,
  shipOrder,
  sendConfirmationEmail,
  refundPayment,
  restoreInventory,
} = proxyActivities<typeof activities>({
  startToCloseTimeout: '1 minute', // Max time for an activity to complete
  retry: {
    initialInterval: '1 second',
    backoffCoefficient: 2,
    maximumInterval: '10 seconds',
    maximumAttempts: 5, // Retry up to 5 times
    nonRetryableErrorTypes: ['NonRetryableError'], // Custom non-retryable error
  },
});

export interface OrderDetails {
  orderId: string;
  amount: number;
  itemId: string;
  quantity: number;
  customerEmail: string;
  shippingAddress: string;
}

export async function orderFulfillmentWorkflow(orderDetails: OrderDetails): Promise<string> {
  const { orderId, amount, itemId, quantity, customerEmail, shippingAddress } = orderDetails;
  let paymentTransactionId: string | undefined;
  let inventoryDeducted = false;

  try {
    // Step 1: Process Payment
    console.log(`Workflow ${workflowInfo().workflowId}: Starting payment processing.`);
    paymentTransactionId = await processPayment(orderId, amount);
    console.log(`Workflow ${workflowInfo().workflowId}: Payment processed. Transaction ID: ${paymentTransactionId}`);

    // Step 2: Deduct Inventory
    console.log(`Workflow ${workflowInfo().workflowId}: Starting inventory deduction.`);
    await deductInventory(orderId, itemId, quantity);
    inventoryDeducted = true;
    console.log(`Workflow ${workflowInfo().workflowId}: Inventory deducted.`);

    // Step 3: Ship Order
    console.log(`Workflow ${workflowInfo().workflowId}: Starting order shipping.`);
    const trackingId = await shipOrder(orderId, shippingAddress);
    console.log(`Workflow ${workflowInfo().workflowId}: Order shipped. Tracking ID: ${trackingId}`);

    // Step 4: Send Confirmation Email (best effort, no compensation needed)
    console.log(`Workflow ${workflowInfo().workflowId}: Sending confirmation email.`);
    await sendConfirmationEmail(orderId, customerEmail);
    console.log(`Workflow ${workflowInfo().workflowId}: Confirmation email sent.`);

    return `Order ${orderId} fulfilled successfully. Tracking ID: ${trackingId}`;

  } catch (error) {
    console.error(`Workflow ${workflowInfo().workflowId}: Order fulfillment failed: ${error.message}`);

    // Compensation logic (Saga pattern)
    if (inventoryDeducted) {
      console.log(`Workflow ${workflowInfo().workflowId}: Compensating: Restoring inventory.`);
      await restoreInventory(orderId, itemId, quantity);
    }
    if (paymentTransactionId) {
      console.log(`Workflow ${workflowInfo().workflowId}: Compensating: Refunding payment.`);
      await refundPayment(orderId, paymentTransactionId);
    }

    throw ApplicationFailure.create({
      message: `Order fulfillment failed for ${orderId}: ${error.message}`,
      nonRetryable: true, // Mark workflow failure as non-retryable for the client
    });
  }
}

// worker.ts
import { Worker } from '@temporalio/worker';
import * as activities from './activities';
import * as workflows from './workflows';

async function run() {
  const worker = await Worker.create({
    workflowsPath: require.resolve('./workflows'),
    activities,
    taskQueue: 'order-processing-task-queue',
  });

  console.log('Temporal Worker started...');
  await worker.run();
  console.log('Temporal Worker stopped.');
}

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

// client.ts
import { Connection, Client } from '@temporalio/client';
import { orderFulfillmentWorkflow, OrderDetails } from './workflows';

async function startOrderWorkflow() {
  const connection = await Connection.connect({ address: 'localhost:7233' }); // Connect to Temporal Cluster
  const client = new Client({ connection });

  const orderDetails: OrderDetails = {
    orderId: `order-${Date.now()}`,
    amount: 100.00,
    itemId: 'SKU-123',
    quantity: 1,
    customerEmail: 'customer@example.com',
    shippingAddress: '123 Main St, Anytown, USA',
  };

  console.log(`Starting order fulfillment workflow for order ${orderDetails.orderId}...`);

  const handle = await client.workflow.start(orderFulfillmentWorkflow, {
    args: [orderDetails],
    taskQueue: 'order-processing-task-queue',
    workflowId: `order-fulfillment-${orderDetails.orderId}`,
  });

  console.log(`Workflow started, ID: ${handle.workflowId}, Run ID: ${handle.runId}`);

  // Wait for the workflow to complete
  try {
    const result = await handle.result();
    console.log(`Workflow completed: ${result}`);
  } catch (error) {
    console.error(`Workflow failed: ${error.message}`);
  }
}

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

// workflows.ts (add to existing workflow file)
import { defineSignal, set  } from '@temporalio/workflow';

export const orderApprovalSignal = defineSignal<[boolean]>('orderApproval');

export async function orderFulfillmentWorkflow(orderDetails: OrderDetails): Promise<string> {
  // ... (existing code) ...

  let approved = false;
  set('approved', approved); // Set initial state for query

  // Wait for approval signal
  console.log(`Workflow ${workflowInfo().workflowId}: Waiting for order approval.`);
  await workflow.wait(async () => approved);
  console.log(`Workflow ${workflowInfo().workflowId}: Order approved, proceeding.`);

  // ... (rest of the workflow logic) ...
}

// client.ts (add to existing client file)
// ...
const handle = await client.workflow.start(orderFulfillmentWorkflow, { /* ... */ });

// Later, from another service or UI:
await handle.signal(orderApprovalSignal, true);
console.log(`Signal sent to approve workflow ${handle.workflowId}`);
// ...

// workflows.ts (add to existing workflow file)
import { defineQuery, set, get } from '@temporalio/workflow';

export const getOrderStatusQuery = defineQuery<string>('getOrderStatus');

export async function orderFulfillmentWorkflow(orderDetails: OrderDetails): Promise<string> {
  // ... (existing code) ...

  let currentStatus = 'Initialized';
  set('currentStatus', currentStatus); // Set initial state

  workflow.set  (getOrderStatusQuery, () => get<string>('currentStatus'));

  try {
    currentStatus = 'Payment Processing';
    set('currentStatus', currentStatus);
    paymentTransactionId = await processPayment(orderId, amount);

    currentStatus = 'Inventory Deduction';
    set('currentStatus', currentStatus);
    await deductInventory(orderId, itemId, quantity);

    currentStatus = 'Shipping';
    set('currentStatus', currentStatus);
    const trackingId = await shipOrder(orderId, shippingAddress);

    currentStatus = 'Confirmation Sent';
    set('currentStatus', currentStatus);
    await sendConfirmationEmail(orderId, customerEmail);

    currentStatus = 'Completed';
    set('currentStatus', currentStatus);
    return `Order ${orderId} fulfilled successfully. Tracking ID: ${trackingId}`;

  } catch (error) {
    currentStatus = `Failed: ${error.message}`;
    set('currentStatus', currentStatus);
    // ... compensation logic ...
    throw error;
  }
}

// client.ts (add to existing client file)
// ...
const handle = await client.workflow.start(orderFulfillmentWorkflow, { /* ... */ });

// Later, to query status:
const status = await handle.query(getOrderStatusQuery);
console.log(`Current order status: ${status}`);
// ...

// workflows.ts (add to existing workflow file)
import { proxyActivities, ApplicationFailure, workflowInfo, startChild } from '@temporalio/workflow';
import * as activities from './activities';

// Define a new child workflow
export async function processOrderItemWorkflow(orderId: string, itemId: string, quantity: number): Promise<boolean> {
  console.log(`Child Workflow ${workflowInfo().workflowId}: Processing item ${itemId} for order ${orderId}`);
  await activities.deductInventory(orderId, itemId, quantity);
  // Potentially more item-specific logic
  return true;
}

export async function orderFulfillmentWorkflow(orderDetails: OrderDetails): Promise<string> {
  const { orderId, amount, itemId, quantity, customerEmail, shippingAddress } = orderDetails;
  let paymentTransactionId: string | undefined;
  let itemProcessed = false; // Track if child workflow completed

  try {
    // ... process payment ...

    // Step 2: Deduct Inventory using a child workflow
    console.log(`Workflow ${workflowInfo().workflowId}: Starting child workflow for item processing.`);
    await startChild(processOrderItemWorkflow, {
      args: [orderId, itemId, quantity],
      workflowId: `${orderId}-item-${itemId}-processing`,
      taskQueue: 'order-processing-task-queue',
    });
    itemProcessed = true;
    console.log(`Workflow ${workflowInfo().workflowId}: Child workflow for item processing completed.`);

    // ... ship order, send email ...

    return `Order ${orderId} fulfilled successfully.`;

  } catch (error) {
    console.error(`Workflow ${workflowInfo().workflowId}: Order fulfillment failed: ${error.message}`);

    // Compensation logic
    if (itemProcessed) {
      console.log(`Workflow ${workflowInfo().workflowId}: Compensating: Restoring inventory via activity.`);
      await activities.restoreInventory(orderId, itemId, quantity);
    }
    if (paymentTransactionId) {
      console.log(`Workflow ${workflowInfo().workflowId}: Compensating: Refunding payment.`);
      await activities.refundPayment(orderId, paymentTransactionId);
    }

    throw ApplicationFailure.create({
      message: `Order fulfillment failed for ${orderId}: ${error.message}`,
      nonRetryable: true,
    });
  }
}

// test/workflows.test.ts
import { TestWorkflowEnvironment } from '@temporalio/testing';
import { Worker } from '@temporalio/worker';
import { Connection, Client } from '@temporalio/client';
import { orderFulfillmentWorkflow, OrderDetails } from '../workflows';
import * as activities from '../activities';

let testEnv: TestWorkflowEnvironment;
let worker: Worker;
let client: Client;

beforeAll(async () => {
  testEnv = await TestWorkflowEnvironment.createTimeSkipping();
  client = new Client({ connection: testEnv.nativeConnection });

  worker = await Worker.create({
    connection: testEnv.nativeConnection,
    taskQueue: 'test-task-queue',
    workflowsPath: require.resolve('../workflows'),
    activities,
  });

  await worker.run(); // Start the worker in the background
}, 30000); // Increase timeout for setup

afterAll(async () => {
  await worker?.shutdown();
  await testEnv?.shutdown();
});

describe('orderFulfillmentWorkflow', () => {
  it('should successfully fulfill an order', async () => {
    const orderDetails: OrderDetails = {
      orderId: 'test-order-1',
      amount: 100,
      itemId: 'SKU-TEST-1',
      quantity: 1,
      customerEmail: 'test@example.com',
      shippingAddress: '123 Test St',
    };

    // Mock activities to ensure deterministic behavior and control outcomes
    const mockActivities = {
      processPayment: jest.fn(() => Promise.resolve('tx-test-1')),
      deductInventory: jest.fn(() => Promise.resolve(true)),
      shipOrder: jest.fn(() => Promise.resolve('trk-test-1')),
      sendConfirmationEmail: jest.fn(() => Promise.resolve(true)),
      refundPayment: jest.fn(() => Promise.resolve(true)),
      restoreInventory: jest.fn(() => Promise.resolve(true)),
    };

    // Override activities for this test run
    worker.activities = mockActivities;

    const result = await client.workflow.execute(orderFulfillmentWorkflow, {
      args: [orderDetails],
      workflowId: 'test-order-workflow-1',
      taskQueue: 'test-task-queue',
    });

    expect(result).toContain('fulfilled successfully');
    expect(mockActivities.processPayment).toHaveBeenCalledWith('test-order-1', 100);
    expect(mockActivities.deductInventory).toHaveBeenCalledWith('test-order-1', 'SKU-TEST-1', 1);
    expect(mockActivities.shipOrder).toHaveBeenCalledWith('test-order-1', '123 Test St');
    expect(mockActivities.sendConfirmationEmail).toHaveBeenCalledWith('test@example.com', 'test-order-1');
    expect(mockActivities.refundPayment).not.toHaveBeenCalled();
    expect(mockActivities.restoreInventory).not.toHaveBeenCalled();
  });

  it('should compensate if inventory deduction fails', async () => {
    const orderDetails: OrderDetails = {
      orderId: 'test-order-2',
      amount: 200,
      itemId: 'SKU-TEST-2',
      quantity: 2,
      customerEmail: 'test2@example.com',
      shippingAddress: '456 Test Ave',
    };

    const mockActivities = {
      processPayment: jest.fn(() => Promise.resolve('tx-test-2')),
      deductInventory: jest.fn(() => Promise.reject(new Error('Inventory unavailable'))),
      shipOrder: jest.fn(), // Should not be called
      sendConfirmationEmail: jest.fn(), // Should not be called
      refundPayment: jest.fn(() => Promise.resolve(true)),
      restoreInventory: jest.fn(() => Promise.resolve(true)),
    };

    worker.activities = mockActivities;

>>>