Aggregation Pipeline — MongoDB

خط التجميع (Aggregation Pipeline)

find يجلب المستندات. Aggregation يحوّلها: يفلتر، ويجمّع، ويحسب، ويربط مجموعات — مثل GROUP BY و JOIN في SQL. البيانات تمر عبر مراحل بالترتيب، وكل مرحلة تستقبل ناتج التي قبلها.

orders ← $match (المدفوعة) ← $group (لكل عميل) ← $sort (الأكثر إنفاقاً) ← $limit 5

أهم المراحل

| المرحلة | ما يعادلها في SQL | الوظيفة |

|---|---|---|

| $match | WHERE | فلترة |

| $group | GROUP BY | تجميع مع $sum و $avg و $min و $max و $push |

| $project / $addFields | SELECT | اختيار حقول وحسابها |

| $sort / $limit / $skip | ORDER BY / LIMIT | |

| $lookup | LEFT JOIN | جلب مستندات من مجموعة أخرى |

| $unwind | — | تفكيك مصفوفة: مستند لكل عنصر |

| $count / $facet | — | عد / عدة تقارير في استعلام واحد |

مثال: أفضل العملاء

const pipeline = [
  { $match: { status: "paid" } },
  { $group: { _id: "$customer", spent: { $sum: "$total" }, orders: { $sum: 1 } } },
  { $sort: { spent: -1 } },
  { $limit: 5 },
  { $lookup: { from: "users", localField: "_id", foreignField: "_id", as: "customer" } },
  { $unwind: "$customer" },
  { $project: { _id: 0, name: "$customer.name", spent: 1, orders: 1 } },
];
console.log(pipeline.map((stage) => Object.keys(stage)[0]).join(" ← "));
// const top = await Order.aggregate(pipeline);

- "$total" بعلامة $ تعني "قيمة الحقل total".

- _id في $group هو مفتاح التجميع (null = مجموعة واحدة للكل).

ترتيب المراحل = الأداء

- ضع $match أولاً: يستفيد من الفهارس ويقلل المستندات لكل ما بعده.

- $limit قبل $lookup عندما يمكن: ربط 5 مستندات أرخص من ربط 50,000.

- $project مبكراً لإسقاط الحقول الكبيرة غير المحتاجة.

تفكيك المصفوفات

عدد مرات بيع كل منتج من طلبات فيها مصفوفة items:

const salesPerProduct = [
  { $unwind: "$items" },
  { $group: { _id: "$items.name", qty: { $sum: "$items.qty" }, revenue: { $sum: { $multiply: ["$items.price", "$items.qty"] } } } },
  { $sort: { revenue: -1 } },
];
console.log(JSON.stringify(salesPerProduct[1].$group.revenue));

ملاحظات Mongoose

- Model.aggregate() لا يحوّل الأنواع تلقائياً كما يفعل find: معرّف نصي من الطلب يجب تحويله: new mongoose.Types.ObjectId(id).

- لا يطبق select: false ولا وسائط المخطط؛ حدد الحقول بنفسك في $project حتى لا تكشف حقولاً حساسة بعد $lookup.

- حد الذاكرة لكل مرحلة 100MB؛ للعمليات الكبيرة { allowDiskUse: true }.

تمرين

اكتب خطاً يعيد لكل شهر في 2026: عدد الطلبات المدفوعة ومجموعها ومتوسطها (استخدم $month في $group)، مرتبة زمنياً.

// npm install mongoose
import mongoose from "mongoose";

await mongoose.connect(process.env.MONGODB_URI ?? "mongodb://127.0.0.1:27017/kodax_demo");
mongoose.set("sanitizeFilter", true);   // يحيّد عوامل $ القادمة من مدخلات المستخدم

const { Schema } = mongoose;
const Customer = mongoose.models.Customer5 ?? mongoose.model("Customer5", new Schema({ name: String, city: String }));
const Order = mongoose.models.Order5 ?? mongoose.model("Order5", new Schema({
  customer: { type: Schema.Types.ObjectId, ref: "Customer5", index: true },
  status: { type: String, enum: ["paid", "pending"], index: true },
  total: Number,
  items: [{ name: String, price: Number, qty: Number }],
  createdAt: { type: Date, default: Date.now },
}));

try {
  await Promise.all([Customer.deleteMany({}), Order.deleteMany({})]);
  const [sara, omar] = await Customer.create([{ name: "سارة", city: "بغداد" }, { name: "عمر", city: "البصرة" }]);
  await Order.create([
    { customer: sara._id, status: "paid", total: 120, items: [{ name: "قلم", price: 1.5, qty: 10 }, { name: "حقيبة", price: 105, qty: 1 }] },
    { customer: sara._id, status: "paid", total: 45, items: [{ name: "دفتر", price: 3, qty: 15 }] },
    { customer: omar._id, status: "pending", total: 300, items: [{ name: "سماعة", price: 300, qty: 1 }] },
  ]);

  // أفضل العملاء: $match أولاً (يستخدم الفهرس)، و $limit قبل $lookup
  const topCustomers = await Order.aggregate([
    { $match: { status: "paid" } },
    { $group: { _id: "$customer", spent: { $sum: "$total" }, orders: { $sum: 1 } } },
    { $sort: { spent: -1 } },
    { $limit: 5 },
    { $lookup: { from: Customer.collection.name, localField: "_id", foreignField: "_id", as: "customer" } },
    { $unwind: "$customer" },
    { $project: { _id: 0, name: "$customer.name", city: "$customer.city", spent: 1, orders: 1 } },
  ]);
  console.log("أفضل العملاء:", topCustomers);

  // المبيعات لكل منتج: تفكيك مصفوفة العناصر
  const perProduct = await Order.aggregate([
    { $match: { status: "paid" } },
    { $unwind: "$items" },
    { $group: { _id: "$items.name", qty: { $sum: "$items.qty" }, revenue: { $sum: { $multiply: ["$items.price", "$items.qty"] } } } },
    { $sort: { revenue: -1 } },
  ]);
  console.log("لكل منتج:", perProduct);

  // aggregate لا يحوّل الأنواع: المعرّف النصي يجب تحويله
  const idFromRequest = String(sara._id);
  const saraTotal = await Order.aggregate([
    { $match: { customer: new mongoose.Types.ObjectId(idFromRequest) } },
    { $group: { _id: null, total: { $sum: "$total" } } },
  ]);
  console.log("مجموع سارة:", saraTotal[0]?.total);
} finally {
  await mongoose.disconnect();
}