بناء مسارات عمل متينة مع Temporal.io و TypeScript

brew install temporal

temporal server start-dev

mkdir temporal-order-processing
cd temporal-order-processing
npm init -y

npm install @temporalio/client @temporalio/worker @temporalio/workflow @temporalio/activity
npm install -D typescript ts-node @types/node

npx tsc --init

{
  "compilerOptions": {
    "target": "ES2020",
    "module": "commonjs",
    "lib": ["ES2020"],
    "strict": true,
    "esModuleInterop": true,
    "outDir": "./dist",
    "rootDir": "./src",
    "skipLibCheck": true,
    "resolveJsonModule": true
  },
  "include": ["src/**/*"],
  "exclude": ["node_modules", "dist"]
}

mkdir -p src/{workflows,activities,workers,client,shared}

// src/shared/types.ts
export interface OrderItem {
  productId: string;
  quantity: number;
  price: number;
}
 
export interface OrderInput {
  orderId: string;
  customerId: string;
  items: OrderItem[];
  totalAmount: number;
  paymentMethodId: string;
}
 
export interface OrderResult {
  orderId: string;
  status: "completed" | "failed" | "cancelled";
  chargeId?: string;
  trackingNumber?: string;
  message: string;
}

// src/activities/order-activities.ts
import { OrderInput } from "../shared/types";
 
export async function validateOrder(input: OrderInput): Promise<boolean> {
  console.log(`التحقق من الطلب ${input.orderId}`);
  await new Promise((resolve) => setTimeout(resolve, 100));
 
  if (input.items.length === 0) {
    throw new Error("يجب أن يحتوي الطلب على عنصر واحد على الأقل");
  }
  if (input.totalAmount <= 0) {
    throw new Error("يجب أن يكون إجمالي الطلب أكبر من الصفر");
  }
 
  return true;
}
 
export async function chargePayment(
  orderId: string,
  amount: number,
  paymentMethodId: string
): Promise<string> {
  console.log(`خصم ${amount} للطلب ${orderId}`);
  await new Promise((resolve) => setTimeout(resolve, 500));
 
  const chargeId = `ch_${Date.now()}_${orderId}`;
  return chargeId;
}
 
export async function sendConfirmationEmail(
  orderId: string,
  customerId: string,
  chargeId: string
): Promise<void> {
  console.log(`إرسال بريد التأكيد للطلب ${orderId}`);
  await new Promise((resolve) => setTimeout(resolve, 200));
}
 
export async function scheduleShipping(
  orderId: string,
  itemCount: number
): Promise<string> {
  console.log(`جدولة شحن الطلب ${orderId}`);
  await new Promise((resolve) => setTimeout(resolve, 300));
 
  return `TRK${Date.now()}`;
}

// src/workflows/order-workflow.ts
import {
  proxyActivities,
  defineSignal,
  defineQuery,
  setHandler,
} from "@temporalio/workflow";
import type * as activities from "../activities/order-activities";
import type { OrderInput, OrderResult } from "../shared/types";
 
const {
  validateOrder,
  chargePayment,
  sendConfirmationEmail,
  scheduleShipping,
} = proxyActivities<typeof activities>({
  startToCloseTimeout: "30 seconds",
  retry: {
    maximumAttempts: 3,
    initialInterval: "1 second",
    maximumInterval: "10 seconds",
    backoffCoefficient: 2,
  },
});
 
export const cancelOrderSignal = defineSignal<[string]>("cancelOrder");
export const getOrderStatusQuery = defineQuery<string>("getOrderStatus");
 
export async function orderWorkflow(input: OrderInput): Promise<OrderResult> {
  let cancelled = false;
  let cancellationReason = "";
  let currentStatus = "pending";
 
  setHandler(cancelOrderSignal, (reason: string) => {
    cancelled = true;
    cancellationReason = reason;
  });
 
  setHandler(getOrderStatusQuery, () => currentStatus);
 
  try {
    currentStatus = "validating";
    await validateOrder(input);
 
    if (cancelled) {
      return {
        orderId: input.orderId,
        status: "cancelled",
        message: `تم إلغاء الطلب قبل الدفع: ${cancellationReason}`,
      };
    }
 
    currentStatus = "charging";
    const chargeId = await chargePayment(
      input.orderId,
      input.totalAmount,
      input.paymentMethodId
    );
 
    currentStatus = "confirming";
    await sendConfirmationEmail(input.orderId, input.customerId, chargeId);
 
    currentStatus = "shipping";
    const trackingNumber = await scheduleShipping(
      input.orderId,
      input.items.length
    );
 
    currentStatus = "completed";
 
    return {
      orderId: input.orderId,
      status: "completed",
      chargeId,
      trackingNumber,
      message: "تمت معالجة الطلب بنجاح",
    };
  } catch (error) {
    currentStatus = "failed";
    return {
      orderId: input.orderId,
      status: "failed",
      message: error instanceof Error ? error.message : "حدث خطأ غير معروف",
    };
  }
}

// src/workers/worker.ts
import { Worker, NativeConnection } from "@temporalio/worker";
import * as activities from "../activities/order-activities";
import path from "path";
 
async function run() {
  const connection = await NativeConnection.connect({
    address: "localhost:7233",
  });
 
  const worker = await Worker.create({
    connection,
    namespace: "default",
    taskQueue: "order-processing",
    workflowsPath: path.join(__dirname, "../workflows"),
    activities,
  });
 
  console.log("العامل يعمل — يستطلع المهام على القائمة: order-processing");
  await worker.run();
}
 
run().catch((err) => {
  console.error("فشل تشغيل العامل:", err);
  process.exit(1);
});

// src/client/start-workflow.ts
import { Client, Connection } from "@temporalio/client";
import {
  orderWorkflow,
  getOrderStatusQuery,
} from "../workflows/order-workflow";
import { OrderInput } from "../shared/types";
 
async function main() {
  const connection = await Connection.connect({ address: "localhost:7233" });
  const client = new Client({ connection, namespace: "default" });
 
  const orderInput: OrderInput = {
    orderId: "ORDER-001",
    customerId: "CUST-123",
    items: [
      { productId: "PROD-A", quantity: 2, price: 29.99 },
    ],
    totalAmount: 59.98,
    paymentMethodId: "pm_card_visa",
  };
 
  const handle = await client.workflow.start(orderWorkflow, {
    taskQueue: "order-processing",
    workflowId: `order-${orderInput.orderId}`,
    args: [orderInput],
  });
 
  console.log(`بدأ مسار العمل: ${handle.workflowId}`);
 
  // استطلاع الحالة حتى اكتمال مسار العمل
  let done = false;
  while (!done) {
    const status = await handle.query(getOrderStatusQuery);
    console.log(`الحالة الحالية: ${status}`);
    if (status === "completed" || status === "failed") done = true;
    await new Promise((r) => setTimeout(r, 1000));
  }
 
  const result = await handle.result();
  console.log("النتيجة النهائية:", result);
 
  await connection.close();
}
 
main().catch(console.error);

{
  "scripts": {
    "worker": "ts-node src/workers/worker.ts",
    "start": "ts-node src/client/start-workflow.ts"
  }
}

temporal server start-dev

npm run worker

npm run start

proxyActivities({
  startToCloseTimeout: "30 seconds",   // أقصى وقت لمحاولة واحدة
  scheduleToCloseTimeout: "5 minutes", // أقصى وقت إجمالي شاملاً جميع المحاولات
  retry: {
    maximumAttempts: 3,
    initialInterval: "1 second",       // انتظار 1 ثانية قبل أول إعادة محاولة
    maximumInterval: "30 seconds",     // لا تنتظر أكثر من 30 ثانية
    backoffCoefficient: 2,             // مضاعف أسي: 1s، 2s، 4s
    nonRetryableErrorTypes: ["PaymentDeclinedError"],
  },
})

import { sleep } from "@temporalio/workflow";
 
export async function subscriptionRenewalWorkflow(userId: string) {
  // انتظر 30 يوماً قبل التجديد
  await sleep("30 days");
 
  await chargeRenewalFee(userId);
  await sendRenewalConfirmation(userId);
 
  // إعادة الجدولة للدورة القادمة
  await sleep("30 days");
}

// src/__tests__/order-workflow.test.ts
import { TestWorkflowEnvironment } from "@temporalio/testing";
import { Worker } from "@temporalio/worker";
import { orderWorkflow } from "../workflows/order-workflow";
import * as activities from "../activities/order-activities";
 
describe("مسار عمل الطلبات", () => {
  let testEnv: TestWorkflowEnvironment;
 
  beforeAll(async () => {
    testEnv = await TestWorkflowEnvironment.createLocal();
  });
 
  afterAll(async () => {
    await testEnv.teardown();
  });
 
  it("يكتمل بنجاح لطلب صالح", async () => {
    const { client, nativeConnection } = testEnv;
 
    const worker = await Worker.create({
      connection: nativeConnection,
      namespace: "default",
      taskQueue: "test-queue",
      workflowsPath: require.resolve("../workflows/order-workflow"),
      activities,
    });
 
    const result = await worker.runUntil(
      client.workflow.execute(orderWorkflow, {
        taskQueue: "test-queue",
        workflowId: "test-001",
        args: [{
          orderId: "TEST-001",
          customerId: "CUST-001",
          items: [{ productId: "P1", quantity: 1, price: 10 }],
          totalAmount: 10,
          paymentMethodId: "pm_test",
        }],
      })
    );
 
    expect(result.status).toBe("completed");
    expect(result.trackingNumber).toBeDefined();
  });
});