MongoDB如何维护产品与变体集合关联的汇总集合同步？

Hey there! Let's figure out how to keep your summary collection in sync with product and variation without messy app-level joins. I’ve got a few practical solutions depending on your needs and database setup (I’m assuming MongoDB here since you mentioned "collections"—adjust as needed for your specific database):

1. Database Triggers (Real-Time & Low-Code) #

If you’re using a managed MongoDB service like Atlas, triggers are the easiest way to automate sync. You can set up triggers that fire whenever data in product or variation is inserted, updated, or deleted, and automatically adjust the summary collection accordingly.

For example, here’s a simplified trigger function that handles all core operations:

exports = async function(changeEvent) {
  const db = context.services.get("YourClusterName").db("YourDatabase");
  const summaryColl = db.collection("summary");
  const changedColl = changeEvent.ns.coll;

  // Handle product collection changes
  if (changedColl === "product") {
    const productId = changeEvent.documentKey._id;
    switch (changeEvent.operationType) {
      case "insert":
        // Create a new summary entry for the new product
        await summaryColl.insertOne({
          productId: productId,
          productName: changeEvent.fullDocument.name,
          category: changeEvent.fullDocument.category,
          variationList: []
        });
        break;
      case "update":
        // Update product details in the summary
        await summaryColl.updateOne(
          { productId: productId },
          { $set: { 
              productName: changeEvent.fullDocument.name, 
              category: changeEvent.fullDocument.category 
            } }
        );
        break;
      case "delete":
        // Remove the summary entry when the product is deleted
        await summaryColl.deleteOne({ productId: productId });
        break;
    }
  }

  // Handle variation collection changes
  else if (changedColl === "variation") {
    const variation = changeEvent.fullDocument;
    switch (changeEvent.operationType) {
      case "insert":
        // Add the new variation to the product's summary
        await summaryColl.updateOne(
          { productId: variation.productId },
          { $push: { variationList: { sku: variation.sku, price: variation.price } } }
        );
        break;
      case "update":
        // Update the specific variation in the summary
        await summaryColl.updateOne(
          { productId: variation.productId, "variationList.sku": variation.sku },
          { $set: { "variationList.$.price": variation.price } }
        );
        break;
      case "delete":
        // Remove the variation from the product's summary
        await summaryColl.updateOne(
          { productId: variation.productId },
          { $pull: { variationList: { sku: variation.sku } } }
        );
        break;
    }
  }
};

2. Custom Sync Service with Change Streams (Flexible & Self-Hosted) #

If you need more control over the sync logic (like adding complex business rules), you can build a lightweight service that listens to your database’s change streams. This works for any MongoDB version 3.6+.

Here’s a quick Node.js example to get you started:

const { MongoClient } = require('mongodb');

async function startSync() {
  const client = await MongoClient.connect("your-mongodb-connection-string");
  const db = client.db("your-database");

  // Watch for changes in the product collection
  const productStream = db.collection("product").watch();
  productStream.on("change", async (change) => {
    const summaryColl = db.collection("summary");
    // Reuse the same logic from the trigger example here
  });

  // Watch for changes in the variation collection
  const variationStream = db.collection("variation").watch();
  variationStream.on("change", async (change) => {
    // Same variation handling logic as above
  });

  console.log("Sync service running...");
}

startSync().catch(err => console.error("Sync failed:", err));

Just add error retry logic and logging to make this production-ready.

3. Periodic Aggregation Sync (For Non-Real-Time Needs) #

If real-time sync isn’t critical, you can run a scheduled aggregation job to rebuild the summary collection periodically (e.g., nightly). This is great if you don’t mind a slight delay in updates and want to avoid ongoing sync overhead.

Use MongoDB’s $lookup to join products and variations, then write the results to summary:

db.product.aggregate([
  {
    $lookup: {
      from: "variation",
      localField: "_id",
      foreignField: "productId",
      as: "variationList"
    }
  },
  {
    $project: {
      productId: "$_id",
      productName: "$name",
      category: "$category",
      variationList: {
        $map: {
          input: "$variationList",
          as: "var",
          in: { sku: "$$var.sku", price: "$$var.price" }
        }
      }
    }
  },
  {
    $merge: {
      into: "summary",
      on: "productId",
      whenMatched: "replace",
      whenNotMatched: "insert"
    }
  }
]);

$merge is better than $out here because it only updates changed documents instead of replacing the entire collection.

Key Tips to Avoid Headaches #

• Atomic Operations: Always use atomic database operations (like updateOne instead of find + save) to prevent data inconsistencies if sync is interrupted.
• Indexing: Add indexes on variation.productId and summary.productId to speed up sync queries—this is crucial if you have large datasets.
• Error Handling: For triggers or sync services, add retry logic for failed operations and log errors so you can debug issues quickly.

内容的提问来源于stack exchange，提问作者s.frz