كل تطبيق ويب جدي يتجاوز عاجلاً أم آجلاً نموذج "طلب/استجابة" المتزامن. إرسال البريد، معالجة الصور المرفوعة، استدعاء واجهات الذكاء الاصطناعي البطيئة، مزامنة البيانات بين الخدمات، تشغيل التقارير المجدولة — لا شيء من هذا ينتمي داخل معالج HTTP. كل هذه المهام يجب أن تُنفَّذ في الخلفية بشكل موثوق، وأن تنجو من إعادة النشر، وتعيد المحاولة عند الفشل، وتتوسع أفقياً عبر عدة عمال.

المسار التقليدي لذلك في TypeScript كان BullMQ على Redis مع لوحة تحكم مخصصة، أو خدمات مُدارة بالكامل مثل Inngest و Trigger.dev. Hatchet يأخذ زاوية مختلفة: مشروع مفتوح المصدر، قابل للاستضافة الذاتية، مبني مباشرة على Postgres. لا Redis. لا Kafka. نفس قاعدة البيانات التي تعتمد عليها لبيانات تطبيقك تنسّق أيضًا عملك الخلفي، مع حالة دائمة، خطوات تُنفَّذ مرة واحدة بالضبط، تحديد للمعدل، تحكم في التزامن، وجدولة cron مدمجة.

في هذا الدليل، ستبني خط أنابيب لمعالجة الوسائط والإشعارات لتطبيق Next.js: عندما يرفع المستخدم صورة، يُنسّق Hatchet سير عمل متعدد الخطوات يتحقق من الملف، يولّد الصور المصغّرة، يستدعي وصف صورة بالذكاء الاصطناعي، يخزّن النتائج، ويرسل إشعارًا. ستتعلّم خلال ذلك كيف يُمثّل Hatchet المهام، سير العمل، إعادات المحاولة، تحديد المعدل، الجدولة عبر cron، ومحفّزات الأحداث — وكيفية نشر أسطول من العمال في الإنتاج.

المتطلبات الأساسية

قبل البدء، تأكد من توفر:

Node.js 20+ مُثبَّت
Docker Desktop يعمل محلياً (لمحرّك Hatchet مع Postgres)
معرفة أساسية بـ TypeScript و async/await
إلمام بـ Next.js App Router (route handlers و server actions)
محرر أكواد مثل VS Code

لست بحاجة إلى حساب Hatchet Cloud مدفوع في هذا الدليل — كل شيء يعمل محلياً عبر Docker.

ما ستبنيه

خط أنابيب لمعالجة الوسائط يتكوّن من:

مهمة validateUpload — تتحقق من الحجم ونوع الملف والأبعاد
مهمة generateThumbnails — تنتج نسخًا صغيرة ومتوسطة وكبيرة
مهمة captionImage — تستدعي خدمة وصف بالذكاء الاصطناعي مع إعادات محاولة
مهمة persistMetadata — تحفظ النتائج في مخطط Postgres الخاص بتطبيقك
مهمة notifyUser — ترسل إشعار push أو بريد إلكتروني
سير عمل media-pipeline — رسم DAG يربط المهام مع فروع متوازية
سير عمل تنظيف مجدوَل يعمل ليلاً
route handler في Next.js يُشغّل خط الأنابيب عند الرفع

في النهاية ستفهم كيف يختلف Hatchet عن الأنظمة المعتمدة على قوائم الانتظار فقط، ومتى تختار المهام المنفردة مقابل سير العمل الكامل، وكيف تُشغّل العمّال في الإنتاج.

لماذا Hatchet بدلاً من BullMQ أو Inngest

توضيح قصير قبل أن نكتب الكود:

BullMQ قائمة انتظار قائمة على Redis. ممتازة لقوائم FIFO/الأولوية، لكن يجب أن تبني فوقها بنفسك سير العمل الدائم، إعادات المحاولة المتقدمة، لوحات التحكم، وأدوات المراقبة.
Inngest / Trigger.dev منصّات تنفيذ دائم بأولوية SaaS. تجربة المطوّر فيها رائعة، لكنها تُقيّدك بالمزوّد وتُضيف فاتورة تكبر مع الاستخدام.
Hatchet مفتوح المصدر، قابل للاستضافة الذاتية، ويستخدم Postgres كمصدر الحقيقة. تحصل على حالة دائمة، سير عمل بنمط DAG، توزيع المهام، تحديد المعدل، ومفاتيح التزامن بقاعدة بيانات إضافية واحدة — غالبًا تلك التي تشغّلها بالفعل.

إذا كانت بنيتك تحتوي على Postgres بالفعل وترغب في تجنّب إضافة Redis ومزوّد SaaS في الوقت نفسه، فإن Hatchet خيار افتراضي قوي.

الخطوة 1: تشغيل Hatchet محلياً عبر Docker

أنشئ مجلد مشروع جديد وشغّل حزمة Hatchet المحلية. يوفّر Hatchet ملف docker-compose.yml واحد يُشغّل Postgres و RabbitMQ (يُستخدم داخلياً كقناة إشعارات منخفضة التأخير) ومحرّك Hatchet ولوحة التحكم.

mkdir hatchet-media-pipeline && cd hatchet-media-pipeline
curl -L https://hatchet.run/install/docker-compose.yml -o docker-compose.yml
docker compose up -d

عندما تنتهي الحزمة من الإقلاع، تصبح لوحة Hatchet متاحة على http://localhost:8080. سجّل الدخول ببيانات الاعتماد الافتراضية المطبوعة في الطرفية، أنشئ tenant، ثم ولّد رمز API من صفحة Settings → API Tokens. احفظه كمتغير بيئة:

echo "HATCHET_CLIENT_TOKEN=your_token_here" > .env
echo "HATCHET_CLIENT_TLS_STRATEGY=none" >> .env

علم tls_strategy=none مهم للتطوير المحلي — يستخدم Hatchet في الإنتاج gRPC فوق TLS، لكن الحزمة المحلية تعمل بدون تشفير.

الخطوة 2: تثبيت TypeScript SDK

ابدأ مشروع TypeScript وأضف الـ SDK الرسمي:

npm init -y
npm install @hatchet-dev/typescript-sdk
npm install -D typescript tsx @types/node dotenv
npx tsc --init

حدّث ملف tsconfig.json ليعمل بسلاسة مع Node الحديث و ESM:

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "ESNext",
    "moduleResolution": "Bundler",
    "esModuleInterop": true,
    "strict": true,
    "skipLibCheck": true,
    "outDir": "dist",
    "rootDir": "src"
  },
  "include": ["src/**/*"]
}

أنشئ نقطة الإعداد المشتركة لـ Hatchet:

// src/hatchet.ts
import "dotenv/config";
import { Hatchet } from "@hatchet-dev/typescript-sdk";
 
export const hatchet = Hatchet.init();

تقرأ Hatchet.init() رمز API وإعدادات الاتصال من متغيرات البيئة. لا حاجة لتهيئة إضافية للحزمة المحلية.

الخطوة 3: تعريف أوّل مهمة

المهام هي أصغر وحدة عمل في Hatchet. كل مهمة لها مدخلات ومخرجات مكتوبة بصرامة، إعادات محاولة تلقائية، وحالة دائمة. أنشئ مهمة التحقق من الرفع:

// src/tasks/validate-upload.ts
import { hatchet } from "../hatchet";
 
type Input = {
  fileUrl: string;
  mimeType: string;
  sizeBytes: number;
};
 
type Output = {
  ok: boolean;
  width: number;
  height: number;
};
 
export const validateUpload = hatchet.task({
  name: "validateUpload",
  retries: 2,
  fn: async (input: Input): Promise<Output> => {
    if (input.sizeBytes > 25_000_000) {
      throw new Error("file too large: must be under 25 MB");
    }
    if (!["image/png", "image/jpeg", "image/webp"].includes(input.mimeType)) {
      throw new Error(`unsupported mime type: ${input.mimeType}`);
    }
    const dims = await probeImage(input.fileUrl);
    return { ok: true, width: dims.width, height: dims.height };
  },
});
 
async function probeImage(url: string) {
  return { width: 1920, height: 1080 };
}

عدّة نقاط جديرة بالملاحظة:

مدخلات ومخرجات مكتوبة بصرامة. الـ SDK آمن من حيث الأنواع من البداية إلى النهاية، لذا عندما تستهلك مهام أخرى هذا الناتج تحصل على إكمال تلقائي صحيح.
retries: 2. سيُعيد Hatchet المحاولة مرتين عند رمي الأخطاء قبل اعتبار المهمة فاشلة.
دالة async نقية. لا أنابيب قوائم انتظار، لا callbacks بنمط done(). رمي الخطأ يُفشل المهمة، وإرجاع قيمة يُنجحها.

الخطوة 4: تشغيل عامل

تعريف المهمة وحده لا يفعل شيئاً حتى يلتقطه عامل. العمّال عمليات طويلة الأمد تشترك في مهمة أو أكثر وتنفّذها فور جدولتها.

// src/worker.ts
import { hatchet } from "./hatchet";
import { validateUpload } from "./tasks/validate-upload";
 
async function main() {
  const worker = await hatchet.worker("media-worker", {
    workflows: [validateUpload],
    slots: 10,
  });
  await worker.start();
}
 
main().catch((err) => {
  console.error(err);
  process.exit(1);
});

قيمة slots: 10 تخبر العامل أنه يستطيع تنفيذ حتى عشر مهام بالتوازي. في الإنتاج تضبط هذه القيمة لتتطابق مع ملف استخدام CPU والذاكرة للأعمال التي يقوم بها العامل.

شغّله في طرفية منفصلة:

npx tsx src/worker.ts

سيُسجّل العامل نفسه لدى محرّك Hatchet ويصبح مرئياً في لوحة التحكم.

الخطوة 5: تشغيل المهمة

في طرفية ثالثة، شغّل سكربتاً صغيراً يُطلق المهمة وينتظر نتيجتها:

// src/trigger.ts
import { validateUpload } from "./tasks/validate-upload";
 
const result = await validateUpload.run({
  fileUrl: "https://example.com/avatar.png",
  mimeType: "image/png",
  sizeBytes: 1_240_000,
});
 
console.log("validation result:", result);

npx tsx src/trigger.ts

ينبغي أن ترى أبعاد الصورة المتحقَّق منها مطبوعة في الطرفية. افتح لوحة Hatchet وستجد التشغيل، مدته، محاولاته، سجلاته، ومدخلاته ومخرجاته المُكتَوبة — كلها مخزّنة بشكل دائم في Postgres.

الخطوة 6: تركيب المهام داخل سير عمل

يربط سير العمل المهام معاً بشكل DAG، حيث تتلقى كل خطوة مخرجات والديها. هذا هو خط أنابيب الوسائط الكامل:

// src/workflows/media-pipeline.ts
import { hatchet } from "../hatchet";
 
type PipelineInput = {
  userId: string;
  fileUrl: string;
  mimeType: string;
  sizeBytes: number;
};
 
export const mediaPipeline = hatchet.workflow<PipelineInput>({
  name: "mediaPipeline",
  on: { event: "media:uploaded" },
});
 
const validate = mediaPipeline.task({
  name: "validate",
  retries: 2,
  fn: async (input, ctx) => {
    return ctx.runTask("validateUpload", {
      fileUrl: input.fileUrl,
      mimeType: input.mimeType,
      sizeBytes: input.sizeBytes,
    });
  },
});
 
const thumbs = mediaPipeline.task({
  name: "generateThumbnails",
  parents: [validate],
  retries: 3,
  fn: async (input, ctx) => {
    const dims = ctx.parents.validate;
    return generateThumbnails(input.fileUrl, dims);
  },
});
 
const caption = mediaPipeline.task({
  name: "captionImage",
  parents: [validate],
  retries: 5,
  rateLimits: [{ key: "ai-caption", units: 1, dynamic: true }],
  fn: async (input) => {
    return aiCaption(input.fileUrl);
  },
});
 
const persist = mediaPipeline.task({
  name: "persistMetadata",
  parents: [thumbs, caption],
  fn: async (input, ctx) => {
    return saveMetadata({
      userId: input.userId,
      thumbnails: ctx.parents.generateThumbnails,
      caption: ctx.parents.captionImage,
    });
  },
});
 
mediaPipeline.task({
  name: "notifyUser",
  parents: [persist],
  fn: async (input) => sendPushNotification(input.userId, "Your upload is ready"),
});

ما الذي يمنحك إياه هذا:

فروع متوازية. كل من generateThumbnails و captionImage يعتمدان على validate ويُنفّذان معاً عند نجاح التحقق.
تجميع ضمني. لا تبدأ persistMetadata إلا بعد اكتمال كلا الوالدين، مع توفّر مخرجاتهما المُكتَوبة على ctx.parents.
سياسات إعادة محاولة لكل مهمة. استدعاء وصف الصورة الهش يعيد المحاولة حتى خمس مرات، وتوليد الصور المصغرة حتى ثلاث.
تحديد المعدل. يشارك استدعاء الذكاء الاصطناعي في دلو معدل مشترك باسم ai-caption، ليحترم كامل أسطول العمّال سقف معدل واحد.

أضف سير العمل إلى تسجيل العامل في src/worker.ts:

const worker = await hatchet.worker("media-worker", {
  workflows: [validateUpload, mediaPipeline],
  slots: 10,
});

الخطوة 7: تشغيل سير العمل من Next.js

اربط خط الأنابيب بـ route handler في Next.js App Router. ضع الملف تحت app/api/uploads/route.ts:

import { NextResponse } from "next/server";
import { mediaPipeline } from "@/src/workflows/media-pipeline";
 
export async function POST(req: Request) {
  const body = await req.json();
 
  const handle = await mediaPipeline.run({
    userId: body.userId,
    fileUrl: body.fileUrl,
    mimeType: body.mimeType,
    sizeBytes: body.sizeBytes,
  });
 
  return NextResponse.json({ workflowRunId: handle.workflowRunId });
}

تُضيف mediaPipeline.run() تشغيلاً لسير العمل، تحفظ المدخلات بشكل دائم في Postgres، وتعود مباشرة برقم تعريفي للتشغيل. يبقى معالج HTTP لديك سريعاً، ويستمر باقي العمل في الخلفية — حتى لو أعاد Next.js نشر نفسه في منتصف التشغيل.

لسير العمل المُشغَّل من أنظمة خارجية (webhooks لـ Stripe، أحداث S3، أحداث Resend) أصدر حدث Hatchet بدلاً من ذلك:

await hatchet.events.push("media:uploaded", {
  userId: body.userId,
  fileUrl: body.fileUrl,
  mimeType: body.mimeType,
  sizeBytes: body.sizeBytes,
});

لأن mediaPipeline مُعلَن بـ on: { event: "media:uploaded" }، كل دفعة من هذا الحدث تتفرّع تلقائياً إلى تشغيل سير عمل.

الخطوة 8: إضافة ضوابط التزامن

افترض أنك تريد ثلاث تشغيلات متزامنة كحد أقصى لكل مستخدم، حتى لا يتمكن حساب واحد يرفع مئة ملف من تجويع قائمة الانتظار للجميع. يُعبّر Hatchet عن ذلك بشكل تصريحي عبر مفتاح تزامن:

export const mediaPipeline = hatchet.workflow<PipelineInput>({
  name: "mediaPipeline",
  on: { event: "media:uploaded" },
  concurrency: {
    expression: "input.userId",
    maxRuns: 3,
    limitStrategy: "GROUP_ROUND_ROBIN",
  },
});

يُجمّع المحرّك التشغيلات حسب input.userId، يحدّ كل مجموعة بثلاث تشغيلات نشطة، ويُوزّع التشغيلات الإضافية بالتناوب round-robin. بدون أقفال مخصصة. بدون Redis. الحالة تعيش في Postgres.

الخطوة 9: جدولة تنظيف عبر cron

ليست كل المهام الخلفية مدفوعة بالطلبات. أضف تنظيفاً ليلياً يحذف الصور المصغرة اليتيمة الأقدم من سبعة أيام:

// src/workflows/cleanup.ts
import { hatchet } from "../hatchet";
 
export const cleanupOrphans = hatchet.workflow({
  name: "cleanupOrphans",
  on: { cron: "0 3 * * *" },
});
 
cleanupOrphans.task({
  name: "deleteOrphans",
  fn: async () => {
    const removed = await deleteOrphanThumbnails({ olderThanDays: 7 });
    return { removed };
  },
});

سجّل سير العمل على العامل ويتولى Hatchet جدولة cron نيابة عنك، بما في ذلك التنسيق عبر نسخ متعددة من العمّال — تشغيل cron واحد بالضبط لكل فترة، بدون تكرار.

الخطوة 10: اختبار المهام بمعزل

مهام Hatchet دوال async عادية، مما يجعل اختبار الوحدات أمراً تافهاً. اسحب منطق الأعمال إلى وحدة نقية واختبره مباشرة:

// src/lib/validate.ts
export function validateMediaInput(input: {
  mimeType: string;
  sizeBytes: number;
}) {
  if (input.sizeBytes > 25_000_000) throw new Error("file too large");
  const allowed = ["image/png", "image/jpeg", "image/webp"];
  if (!allowed.includes(input.mimeType)) throw new Error("bad mime");
  return true;
}

// src/lib/validate.test.ts
import { describe, it, expect } from "vitest";
import { validateMediaInput } from "./validate";
 
describe("validateMediaInput", () => {
  it("accepts a valid PNG under 25 MB", () => {
    expect(validateMediaInput({ mimeType: "image/png", sizeBytes: 1_000 })).toBe(true);
  });
  it("rejects oversized files", () => {
    expect(() => validateMediaInput({ mimeType: "image/png", sizeBytes: 26_000_000 })).toThrow();
  });
});

غلاف المهمة لا يفعل سوى استدعاء validateMediaInput(). لا حاجة لاختباراتك أن تُحاكي Hatchet نفسه.

الخطوة 11: مراقبة التشغيلات في الإنتاج

لوحة التحكم على http://localhost:8080 هي نفس الواجهة التي تحصل عليها في الإنتاج. لكل تشغيل سير عمل ترى:

خط زمني لكل مهمة، بما في ذلك زمن الانتظار وزمن التنفيذ
المدخلات والمخرجات الكاملة المُكتَوبة (مشفّرة عند التخزين)
سجلات مهيكلة مُصدَرة بواسطة ctx.log()
محاولات إعادة المحاولة ورسائل الأخطاء الخاصة بها
حالة تحديد المعدل ومفاتيح التزامن

للمراقبة البرمجية، اعرض مقاييس Hatchet إلى Prometheus وغذِّها إلى لوحة Grafana إلى جانب مقاييس تطبيقك الحالية. النسب المئوية لزمن الاستجابة لكل مهمة مفيدة بشكل خاص عند مطاردة الخطوات البطيئة داخل DAG.

الخطوة 12: نشر أسطول من العمّال

عادة ما يتكوّن النشر الإنتاجي من ثلاث قطع:

محرّك Hatchet — يُشغَّل عبر مخطط Helm، Fly Machines، Railway، أو عنقود Kubernetes الحالي، موجَّهاً إلى Postgres مُدار (Neon، Supabase، RDS).
بركة من عمليات العمّال — كود TypeScript خاصتك، منشور كحاوية طويلة الأمد (Fly Machines، Render، Railway، ECS) مع تحجيم تلقائي بناءً على عمق قائمة الانتظار.
تطبيقك — Next.js على Vercel، خادم Bun على Coolify، Laravel monolith، أو أي شيء آخر. يحتاج فقط الـ SDK لإضافة التشغيلات ودفع الأحداث.

ضع العامل في Docker بصورة Node صغيرة:

FROM node:20-alpine
WORKDIR /app
COPY package*.json ./
RUN npm ci --omit=dev
COPY dist ./dist
ENV NODE_ENV=production
CMD ["node", "dist/worker.js"]

ابنِ مرة واحدة، انشر بقدر ما تتطلب إنتاجيتك من نسخ. يُعالج Hatchet توزيع العمل، وأنت تركّز على منطق الأعمال.

استكشاف الأخطاء وإصلاحها

بعض المشكلات التي قد تواجهها:

UNAVAILABLE: connection refused عند تشغيل عامل محلياً يعني عادةً أن حزمة Docker لم تكتمل الإقلاع. انتظر حتى يُظهر docker compose logs hatchet-engine رسالة gRPC server listening.
مهام عالقة في حالة PENDING تشير عادةً إلى أنه لا عامل مُسجَّل لذلك سير العمل. تأكد من تضمين سير العمل ضمن مصفوفة workflows: المُمرَّرة إلى hatchet.worker().
إعادات المحاولة لا تنطلق أبداً إذا عادت الدالة بسلام رغم وجود فشل منطقي. ارمِ خطأ للإشارة إلى الفشل — هكذا يقرر Hatchet إعادة المحاولة.
مفاتيح تحديد المعدل لا تُطبَّق عبر الأسطول — تأكد من أن كل عامل يُسجّل نفس تعريف تحديد المعدل. التعريفات المتعارضة تُنشئ دلاء مستقلة.

الخطوات التالية

لديك الآن قائمة مهام دائمة، آمنة الأنواع، مدعومة بـ Postgres تُشغّل خط أنابيب وسائط حقيقياً. لتوسيع المشروع أكثر:

استبدل aiCaption() الوهمية باستدعاء حقيقي لنموذج رؤية — جرّب دليلنا حول Google Gemini API مع TypeScript.
اربط مسار الرفع بطبقة مصادقة قوية باستخدام Better Auth و Next.js.
قارن Hatchet مع Inngest و Trigger.dev لاختيار محرّك التنفيذ الدائم الأنسب لفريقك.

الخلاصة

يدمج Hatchet قائمة الانتظار وإطار العمّال والمجدوِل ومخزن الحالة الدائم في نظام واحد مدعوم بـ Postgres. تكتب دوال TypeScript عادية، تركّبها في DAGs، تُعلن إعادات المحاولة وحدود المعدل كبيانات، وتترك الباقي للمحرّك. النموذج العقلي صغير، البصمة التشغيلية قاعدة بيانات إضافية واحدة، والشفرة المصدرية مفتوحة — مما يجعله افتراضاً جذاباً للفِرَق التي تثق بالفعل في Postgres وتريد أن يبدو العمل الخلفي كبقية قاعدة الكود لديها.

في هذا الدليل، أعددت Hatchet على Docker، عرّفت مهام مكتوبة الأنواع، ركّبتها في سير عمل بفروع متوازية، طبّقت إعادات المحاولة وحدود المعدل، جدوَلت تنظيفاً عبر cron، وشغّلت خط الأنابيب بالكامل من مسار في Next.js. تتوسع نفس الأنماط لتعالج آلاف الوظائف في الثانية عبر أسطول من العمّال، مع نفس لوحة التحكم ونفس Postgres كمصدر وحيد للحقيقة.