mongodb聚合更改未在go中持久化答案

【问题标题】：Mongodb aggregation changes not being persisted in gomongodb聚合更改未在go中持久化
【发布时间】：2022-08-16 03:38:43
【问题描述】：

我正在运行聚合以删除过时的文档，但更改实际上不会影响数据库。查询会忽略已经过期的文档，因此每次查询运行后结果数应该会发生变化，但不会。

func CheckShipmentExpiryDates(c *mongo.Client) (int, error) {
    numberOfExpiredShipments := 0
    coll := c.Database(os.Getenv(\"DATABASE\")).Collection(\"shipments\")
    update := bson.M{\"$set\": bson.M{\"status\": \"EXPIRED\", \"updated_at\": time.Now()}}
    pipeline := []bson.M{
        {\"$lookup\": bson.M{
            \"from\": \"shipment_quotes\",
            \"let\":  bson.M{\"shipmentID\": \"$_id\"},
            \"pipeline\": []bson.M{
                {\"$match\": bson.M{\"$expr\": bson.M{\"$and\": []bson.M{{\"$eq\": []string{\"$shipment_id\", \"$$shipmentID\"}}, {\"$eq\": []string{\"$status\", \"WON\"}}}}}},
            },
            \"as\": \"quotes\",
        }},
        {\"$match\": bson.M{\"expiration_date\": bson.M{\"$exists\": true}}},
        {\"$match\": bson.M{\"$expr\": bson.M{\"$and\": []bson.M{
            {\"$ne\": []string{\"$status\", \"EXPIRED\"}},
            {\"$lt\": []interface{}{\"$expiration_date\", time.Now()}},
            {\"$eq\": []interface{}{bson.M{\"$size\": \"$quotes\"}, 0}},
            {\"expiration_date\": bson.M{\"$type\": 9}},
        }}}},
        update,
    }

    err := c.UseSession(context.TODO(), func(sessionContext mongo.SessionContext) error {
        if err := sessionContext.StartTransaction(); err != nil {
            return err
        }
        cursor, err := coll.Aggregate(sessionContext, pipeline)
        if err != nil {
            _ = sessionContext.AbortTransaction(sessionContext)
            return err
        }

        var shipments []bson.M
        if err := cursor.All(sessionContext, &shipments); err != nil {
            _ = sessionContext.AbortTransaction(sessionContext)
            return err
        }

        fmt.Println(\"~First shipment\'s status\", shipments[0][\"shipment_unique_number\"], shipments[0][\"status\"])

        numberOfExpiredShipments = len(shipments)

        fmt.Println(sessionContext.CommitTransaction(sessionContext))
        return nil
    })

    return numberOfExpiredShipments, err
}

如您所见，我正在记录第一个结果并使用 compass 实时检查数据库，但更改实际上并没有被持久化。查询一遍又一遍地运行，返回相同数量的过期货物。

mc, mongoErr := connection.MongoInit()
    if mongoErr != nil {
        panic(mongoErr)
    }
    utils.InitDB(mc)
    defer func() {
        if err := mc.Disconnect(context.TODO()); err != nil {
            panic(err)
        }
    }()

    n := connection.NewNotificationCenter()
    sseInit(mc, googleApi, n)
    graphSchema, err := schema.InjectSchema(mutationInit(mc, googleApi), queryInit(mc, googleApi))
    if err != nil {
        panic(err)
    }
    restApiUseCase := mutationsRestApiInit(mc, googleApi)
    connection.InjectGraphqlHandler(graphSchema, n, restApiUseCase)

    initIncrementStartdate(mc)
    initShipmentExpiredCron(mc)

func initShipmentExpiredCron(mg *mongo.Client) {
    c := cron.New()
    c.AddFunc(\"*/5 * * * *\", func() {
        expiredShipments, err := utils.CheckShipmentExpiryDates(mg)
        if err != nil {
            log.Println(\"CRON ERROR: An error occured while trying to check the expiry date for each shipment\")
            log.Println(err)
        } else {
            // Print how many shipments are expired
            log.Println(\"CRON SUCCESS: The following number of shipments have expired: \", expiredShipments)
        }
    })
    c.Start()
}

我真的不明白它有什么问题。

我认为问题可能是 $set 在聚合中不起作用？项目中还有许多其他地方 $set 似乎用于聚合和工作。我们以非常抱歉的状态接手了这个项目，所以可能是我误会了。问题是我不能在 updateMany 中使用 $lookup，只能在聚合中使用。
我不知道go，所以我无法遵循所有代码，但是聚合管道不会修改文档/集合/数据库 - 它只是一个查询。您可能会考虑将\"$merge\" 附加到您的管道以实际进行修改。我不确定您可能想要使用哪个 \"$merge\" 选项（如果有的话）。

标签： mongodb go aggregation

【解决方案1】：

rickhg12hs 是对的，我还需要使用$merge。奇怪的是，这使得聚合不返回任何东西，所以现在我不知道已过期的发货数量，但这并不是真正必要的。这是最后的管道

pipeline := []bson.M{{
  "$lookup": bson.M{
    "from": "shipment_quotes",
    "let": bson.M{
      "shipmentID": "$_id"
    },
    "pipeline": []bson.M{{
      "$match": bson.M{
        "$expr": bson.M{
          "$and": []bson.M{{
            "$eq": []string{
              "$shipment_id", "$$shipmentID"
            }}, {
              "$eq": []string{
                "$status", "WON"
              }
            }
          }}
        }
      },
    },
    "as": "quotes",
  }}, {
    "$match": bson.M{
      "expiration_date": bson.M{
        "$exists": true
      }
    }
  }, {
    "$match": bson.M{
      "$expr": bson.M{
        "$and": []bson.M{{
          "$ne": []string{
            "$status", "EXPIRED"
          }
        }, {
          "$lt": []interface{}{
            "$expiration_date", time.Now()
          }
        }, {
          "$eq": []interface{}{
            bson.M{
              "$size": "$quotes"
            }, 0
          }
        }, {
          "expiration_date": bson.M{
            "$type": 9
          }
        },
      }
    }
  }},
  update,
  {
    "$merge": bson.M{
      "into": "shipments",
      "on": "_id"
    }
  },
}

【讨论】：

以$merge（也为$out）作为最后阶段的聚合查询返回一个空光标- 因此，没有返回值。