fix: 修复发布、部署资源时事务使用错误；

Author: maslow

packages/devops-server/src/api/function.ts

@@ -3,7 +3,7 @@ import { Constants } from "../constants"
 import { Globals } from "../lib/globals"
 import { compileTs2js } from 'cloud-function-engine/dist/utils'
 import { CloudFunctionStruct } from "cloud-function-engine"
-import { ObjectId } from 'mongodb'
+import { ClientSession, ObjectId } from 'mongodb'
 import * as assert from 'assert'
 const db = Globals.sys_db
 
@@ -63,8 +63,8 @@ export async function publishFunctions() {
     await session.withTransaction(async () => {
       const _db = app_accessor.db
       const app_coll = _db.collection(Constants.function_collection)
-      await app_coll.deleteMany({})
-      await app_coll.insertMany(data)
+      await app_coll.deleteMany({}, { session })
+      await app_coll.insertMany(data, { session })
     })
   } catch (error) {
     logger.error(error)
@@ -99,7 +99,7 @@ export async function deployFunctions(functions: CloudFunctionStruct[]) {
   try {
     await session.withTransaction(async () => {
       for (const func of data) {
-        await _deployOneFunction(func)
+        await _deployOneFunction(func, session)
       }
     })
   } catch (error) {
@@ -110,12 +110,12 @@ export async function deployFunctions(functions: CloudFunctionStruct[]) {
   }
 }
 
-async function _deployOneFunction(func: CloudFunctionStruct) {
+async function _deployOneFunction(func: CloudFunctionStruct, session: ClientSession) {
 
-  await _deleteFunctionWithSameNameButNotId(func)
+  await _deleteFunctionWithSameNameButNotId(func, session)
 
   const db = Globals.sys_accessor.db
-  const r = await db.collection('__functions').findOne({ _id: new ObjectId(func._id) })
+  const r = await db.collection('__functions').findOne({ _id: new ObjectId(func._id) }, { session })
 
   const data = {
     ...func
@@ -126,7 +126,7 @@ async function _deployOneFunction(func: CloudFunctionStruct) {
     delete data['_id']
     const ret = await db.collection('__functions').updateOne({ _id: r._id }, {
       $set: data
-    })
+    }, { session })
 
     assert(ret.matchedCount, `deploy: update function ${func.name} occurred error`)
     return
@@ -135,20 +135,20 @@ async function _deployOneFunction(func: CloudFunctionStruct) {
   // if new function
   data._id = new ObjectId(data._id) as any
 
-  const ret = await db.collection('__functions').insertOne(data as any)
+  const ret = await db.collection('__functions').insertOne(data as any, { session })
   assert(ret.insertedId, `deploy: add function ${func.name} occurred error`)
 }
 
 /**
  * 删除本地 _id 不同，但 name 相同的云函数（若存在）
  * @param func 
  */
-async function _deleteFunctionWithSameNameButNotId(func: CloudFunctionStruct) {
+async function _deleteFunctionWithSameNameButNotId(func: CloudFunctionStruct, session: ClientSession) {
   const db = Globals.sys_accessor.db
   await db.collection('__functions').findOneAndDelete({
     _id: {
       $ne: new ObjectId(func._id)
     },
     name: func.name
-  })
+  }, { session })
 }

packages/devops-server/src/api/rules.ts

@@ -1,7 +1,7 @@
 import * as assert from 'assert'
 import { Constants } from '../constants'
 import { Globals } from "../lib/globals"
-import { ObjectId } from 'mongodb'
+import { ClientSession, ObjectId } from 'mongodb'
 
 const db = Globals.sys_db
 export interface RuleDocument {
@@ -48,8 +48,8 @@ export async function publishAccessPolicy() {
     await session.withTransaction(async () => {
       const _db = app_accessor.db
       const app_coll = _db.collection(Constants.policy_collection)
-      await app_coll.deleteMany({})
-      await app_coll.insertMany(ret)
+      await app_coll.deleteMany({}, { session })
+      await app_coll.insertMany(ret, { session })
     })
   } catch (error) {
     logger.error(error)
@@ -76,7 +76,7 @@ export async function deployPolicies(policies) {
   try {
     await session.withTransaction(async () => {
       for (const item of data) {
-        await _deployOnePolicy(item)
+        await _deployOnePolicy(item, session)
       }
     })
   } catch (error) {
@@ -87,12 +87,12 @@ export async function deployPolicies(policies) {
   }
 }
 
-async function _deployOnePolicy(policy: any) {
+async function _deployOnePolicy(policy: any, session: ClientSession) {
 
-  await _deletePolicyWithSameNameButNotId(policy)
+  await _deletePolicyWithSameNameButNotId(policy, session)
 
   const db = Globals.sys_accessor.db
-  const r = await db.collection('__policies').findOne({ _id: new ObjectId(policy._id) })
+  const r = await db.collection('__policies').findOne({ _id: new ObjectId(policy._id) }, { session })
 
   const data = {
     ...policy
@@ -104,28 +104,28 @@ async function _deployOnePolicy(policy: any) {
     delete data['_id']
     const ret = await db.collection('__policies').updateOne({ _id: r._id }, {
       $set: data
-    })
+    }, { session })
 
     assert(ret.matchedCount, `deploy: update policy ${policy.name} occurred error`)
     return
   }
 
   // if new
   data._id = new ObjectId(data._id) as any
-  const ret = await db.collection('__policies').insertOne(data as any)
+  const ret = await db.collection('__policies').insertOne(data as any, { session })
   assert(ret.insertedId, `deploy: add policy ${policy.name} occurred error`)
 }
 
 /**
  * 删除本地 _id 不同，但 name 相同的策略（若存在）
  * @param func 
  */
-async function _deletePolicyWithSameNameButNotId(policy: any) {
+async function _deletePolicyWithSameNameButNotId(policy: any, session: ClientSession) {
   const db = Globals.sys_accessor.db
   await db.collection('__policies').findOneAndDelete({
     _id: {
       $ne: new ObjectId(policy._id)
     },
     name: policy.name
-  })
+  }, { session })
 }

packages/devops-server/src/api/trigger.ts

@@ -1,7 +1,7 @@
 
 import { Constants } from "../constants"
 import { Globals } from "../lib/globals"
-import { ObjectId } from 'mongodb'
+import { ClientSession, ObjectId } from 'mongodb'
 import * as assert from 'assert'
 
 const db = Globals.sys_db
@@ -49,8 +49,8 @@ export async function publishTriggers() {
     await session.withTransaction(async () => {
       const _db = app_accessor.db
       const app_coll = _db.collection(Constants.trigger_collection)
-      await app_coll.deleteMany({})
-      await app_coll.insertMany(ret)
+      await app_coll.deleteMany({}, { session })
+      await app_coll.insertMany(ret, { session })
     })
   } catch (error) {
     logger.error(error)
@@ -77,7 +77,7 @@ export async function deployTriggers(triggers: any[]) {
   try {
     await session.withTransaction(async () => {
       for (const func of data) {
-        await _deployOneTrigger(func)
+        await _deployOneTrigger(func, session)
       }
     })
   } catch (error) {
@@ -88,10 +88,10 @@ export async function deployTriggers(triggers: any[]) {
   }
 }
 
-async function _deployOneTrigger(trigger: any) {
+async function _deployOneTrigger(trigger: any, session: ClientSession) {
 
   const db = Globals.sys_accessor.db
-  const r = await db.collection('__triggers').findOne({ _id: new ObjectId(trigger._id) })
+  const r = await db.collection('__triggers').findOne({ _id: new ObjectId(trigger._id) }, { session })
 
   const data = {
     ...trigger
@@ -103,14 +103,14 @@ async function _deployOneTrigger(trigger: any) {
     delete data['_id']
     const ret = await db.collection('__triggers').updateOne({ _id: r._id }, {
       $set: data
-    })
+    }, { session })
 
     assert(ret.matchedCount, `deploy: update trigger ${trigger.name} occurred error`)
     return
   }
 
   // if new function
   data._id = new ObjectId(data._id) as any
-  const ret = await db.collection('__triggers').insertOne(data as any)
+  const ret = await db.collection('__triggers').insertOne(data as any, { session })
   assert(ret.insertedId, `deploy: add trigger ${trigger.name} occurred error`)
 }