Causal Consistency fixes

src/controller/org.controller/org.controller.js

@@ -17,10 +17,8 @@ const validateUUID = require('uuid').validate
  */
 async function getOrgs (req, res, next) {
   try {
-    const session = await mongoose.startSession()
     const repo = req.ctx.repositories.getBaseOrgRepository()
     const CONSTANTS = getConstants()
-    let returnValue
 
     // temporary measure to allow tests to work after fixing #920
     // tests required changing the global limit to force pagination
@@ -32,11 +30,7 @@ async function getOrgs (req, res, next) {
     options.sort = { short_name: 'asc' }
     options.page = req.ctx.query.page ? parseInt(req.ctx.query.page) : CONSTANTS.PAGINATOR_PAGE // if 'page' query parameter is not defined, set 'page' to the default page value
 
-    try {
-      returnValue = await repo.getAllOrgs({ ...options, session }, true)
-    } finally {
-      await session.endSession()
-    }
+    const returnValue = await repo.getAllOrgs({ ...options }, true)
 
     logger.info({ uuid: req.ctx.uuid, message: 'The orgs were sent to the user.' })
     return res.status(200).json(returnValue)
@@ -58,7 +52,6 @@ async function getOrgs (req, res, next) {
  */
 async function getOrg (req, res, next) {
   try {
-    const session = await mongoose.startSession()
     const repo = req.ctx.repositories.getBaseOrgRepository()
     const requesterOrgShortName = req.ctx.org
     const identifier = req.ctx.params.identifier
@@ -67,26 +60,21 @@ async function getOrg (req, res, next) {
     let returnValue
 
     try {
-      session.startTransaction()
-      const requesterOrg = await repo.findOneByShortName(requesterOrgShortName, { session }, returnLegacyFormat)
+      const requesterOrg = await repo.findOneByShortName(requesterOrgShortName, {}, returnLegacyFormat)
       const requesterOrgIdentifier = identifierIsUUID ? requesterOrg.UUID : requesterOrgShortName
-      const isSecretariat = await repo.isSecretariat(requesterOrg, { session }, returnLegacyFormat)
+      const isSecretariat = await repo.isSecretariat(requesterOrg, {}, returnLegacyFormat)
 
       if (requesterOrgIdentifier !== identifier && !isSecretariat) {
         logger.info({ uuid: req.ctx.uuid, message: identifier + ' organization can only be viewed by the users of the same organization or the Secretariat.' })
         return res.status(403).json(error.notSameOrgOrSecretariat())
       }
 
-      returnValue = await repo.getOrg(identifier, identifierIsUUID, { session }, returnLegacyFormat)
+      returnValue = await repo.getOrg(identifier, identifierIsUUID, {}, returnLegacyFormat)
     } catch (error) {
-      await session.abortTransaction()
       // Handle the specific error thrown by BaseOrgRepository.createOrg
       if (error.message && error.message.includes('Unknown Org type requested')) {
         return res.status(400).json({ message: error.message })
       }
-      throw error
-    } finally {
-      await session.endSession()
     }
     if (!returnValue) { // an empty result can only happen if the requestor is the Secretariat
       logger.info({ uuid: req.ctx.uuid, message: identifier + ' organization does not exist.' })
@@ -336,7 +324,7 @@ async function updateOrg (req, res, next) {
   const shortNameUrlParameter = req.ctx.params.shortname
   const orgRepository = req.ctx.repositories.getBaseOrgRepository()
 
-  const session = await mongoose.startSession()
+  const session = await mongoose.startSession({ causalConsistency: false })
   let responseMessage
   // Get the query parameters as JSON
   // These are validated by the middleware in org/index.js

src/controller/registry-org.controller/registry-org.controller.js

@@ -71,7 +71,6 @@ async function getAllOrgs (req, res, next) {
  */
 async function getOrg (req, res, next) {
   try {
-    const session = await mongoose.startSession()
     const repo = req.ctx.repositories.getBaseOrgRepository()
     const conversationRepo = req.ctx.repositories.getConversationRepository()
     // User passed in parameter to filter for
@@ -81,32 +80,28 @@ async function getOrg (req, res, next) {
     let returnValue
 
     try {
-      session.startTransaction()
-      const requesterOrg = await repo.findOneByShortName(requesterOrgShortName, { session })
+      const requesterOrg = await repo.findOneByShortName(requesterOrgShortName)
       const requesterOrgIdentifier = identifierIsUUID ? requesterOrg.UUID : requesterOrgShortName
-      const isSecretariat = await repo.isSecretariat(requesterOrg, { session })
+      const isSecretariat = await repo.isSecretariat(requesterOrg)
 
       if (requesterOrgIdentifier !== identifier && !isSecretariat) {
         logger.info({ uuid: req.ctx.uuid, message: identifier + ' organization can only be viewed by the users of the same organization or the Secretariat.' })
         return res.status(403).json(error.notSameOrgOrSecretariat())
       }
 
-      returnValue = await repo.getOrg(identifier, identifierIsUUID, { session })
+      returnValue = await repo.getOrg(identifier, identifierIsUUID)
 
       if (returnValue) {
         // fetch conversation
-        const conversation = await conversationRepo.getAllByTargetUUID(returnValue.UUID, isSecretariat, { session })
+        const conversation = await conversationRepo.getAllByTargetUUID(returnValue.UUID, isSecretariat)
         returnValue.conversation = conversation?.length ? _.map(conversation, c => _.omit(c, ['__v', '_id', 'UUID', 'previous_conversation_uuid', 'next_conversation_uuid', 'target_uuid', 'visibility'])) : undefined
       }
     } catch (error) {
-      await session.abortTransaction()
       // Handle the specific error thrown by BaseOrgRepository.createOrg
       if (error.message && error.message.includes('Unknown Org type requested')) {
         return res.status(400).json({ message: error.message })
       }
       throw error
-    } finally {
-      await session.endSession()
     }
     if (!returnValue) { // an empty result can only happen if the requestor is the Secretariat
       logger.info({ uuid: req.ctx.uuid, message: identifier + ' organization does not exist.' })
@@ -134,7 +129,7 @@ async function getOrg (req, res, next) {
  */
 async function createOrg (req, res, next) {
   try {
-    const session = await mongoose.startSession()
+    const session = await mongoose.startSession({ causalConsistency: false })
     const repo = req.ctx.repositories.getBaseOrgRepository()
     const body = req.ctx.body
     const isSecretariat = await repo.isSecretariatByShortName(req.ctx.org, { session })
@@ -232,7 +227,7 @@ async function createOrg (req, res, next) {
  */
 async function updateOrg (req, res, next) {
   try {
-    const session = await mongoose.startSession()
+    const session = await mongoose.startSession({ causalConsistency: false })
     const shortName = req.ctx.params.shortname
     const repo = req.ctx.repositories.getBaseOrgRepository()
     const userRepo = req.ctx.repositories.getBaseUserRepository()
@@ -318,15 +313,18 @@ async function updateOrg (req, res, next) {
         }
       }
 
+      // Update Org full will cause a write to the Conversations collection, to avoid a read-after-write issue, we need to get the previous conversation data first
+      const previousConversation = await conversationRepo.getAllByTargetUUID(await repo.getOrgUUID(shortName, { session }), isSecretariat, { session }) || []
+
       updatedOrg = await repo.updateOrgFull(shortName, req.ctx.body, { session }, false, requestingUser.UUID, isAdmin, isSecretariat)
       jointApprovalRequired = _.get(updatedOrg, 'joint_approval_required', false)
       _.unset(updatedOrg, 'joint_approval_required')
-
-      await session.commitTransaction()
-      session.startTransaction()
-      // Checking for existing Conversations
-      const existingConversations = await conversationRepo.getAllByTargetUUID(updatedOrg.UUID, isSecretariat, { session }) || []
-      updatedOrg.conversation = existingConversations.map(c => _.omit(c, ['__v', '_id', 'previous_conversation_uuid', 'next_conversation_uuid']))
+      // append previous conversations to any conversations that are in the org already
+      const currentConversations = Array.isArray(updatedOrg?.conversation) ? updatedOrg.conversation : []
+      const prevConversations = Array.isArray(previousConversation) ? previousConversation : []
+      if (updatedOrg) {
+        updatedOrg.conversation = [...currentConversations, ...prevConversations].map(c => _.omit(c, ['__v', '_id', 'previous_conversation_uuid', 'next_conversation_uuid']))
+      }
 
       await session.commitTransaction()
     } catch (updateErr) {
@@ -387,7 +385,7 @@ async function updateOrg (req, res, next) {
  */
 async function deleteOrg (req, res, next) {
   try {
-    const session = await mongoose.startSession()
+    const session = await mongoose.startSession({ causalConsistency: false })
     const repo = req.ctx.repositories.getBaseOrgRepository()
     const shortName = req.ctx.params.identifier
 
@@ -500,7 +498,7 @@ async function getUsers (req, res, next) {
  *              Called by POST /api/registryOrg/:shortname/user
  */
 async function createUserByOrg (req, res, next) {
-  const session = await mongoose.startSession()
+  const session = await mongoose.startSession({ causalConsistency: false })
   try {
     const body = req.ctx.body
     const userRepo = req.ctx.repositories.getBaseUserRepository()

src/controller/registry-user.controller/registry-user.controller.js

@@ -21,9 +21,7 @@ const _ = require('lodash')
 async function getAllUsers (req, res, next) {
   try {
     const CONSTANTS = getConstants()
-    const session = await mongoose.startSession()
     const repo = req.ctx.repositories.getBaseUserRepository()
-    let returnValue
 
     // temporary measure to allow tests to work after fixing #920
     // tests required changing the global limit to force pagination
@@ -35,36 +33,32 @@ async function getAllUsers (req, res, next) {
     options.sort = { short_name: 'asc' }
     options.page = req.ctx.query.page ? parseInt(req.ctx.query.page) : CONSTANTS.PAGINATOR_PAGE // if 'page' query parameter is not defined, set 'page' to the default page value
 
-    try {
-      returnValue = await repo.getAllUsers(options)
-      // Hydrate roles
-      const orgRepo = req.ctx.repositories.getBaseOrgRepository()
-      const distinctOrgUUIDs = [...new Set(returnValue.users.map(u => u.org_UUID))]
-
-      // Fetch all relevant orgs in one go (or in parallel) if possible, but map is easy for now
-      // Since we don't have a "getManyOrgsByUUID", we might need to do it one by one or improve repository
-      // For now, let's iterate and fetch. It's not optimal but safe given repo limitations.
-      // Optimization: We can build a map of orgUUID -> orgObject
-      const orgMap = {}
-      for (const uuid of distinctOrgUUIDs) {
-        // We need the org content to get admins
-        const org = await orgRepo.findOneByUUID(uuid)
-        if (org) {
-          orgMap[uuid] = org
-        }
+    const returnValue = await repo.getAllUsers(options)
+    // Hydrate roles
+    const orgRepo = req.ctx.repositories.getBaseOrgRepository()
+    const distinctOrgUUIDs = [...new Set(returnValue.users.map(u => u.org_UUID))]
+
+    // Fetch all relevant orgs in one go (or in parallel) if possible, but map is easy for now
+    // Since we don't have a "getManyOrgsByUUID", we might need to do it one by one or improve repository
+    // For now, let's iterate and fetch. It's not optimal but safe given repo limitations.
+    // Optimization: We can build a map of orgUUID -> orgObject
+    const orgMap = {}
+    for (const uuid of distinctOrgUUIDs) {
+      // We need the org content to get admins
+      const org = await orgRepo.findOneByUUID(uuid)
+      if (org) {
+        orgMap[uuid] = org
       }
-
-      returnValue.users.forEach(user => {
-        const org = orgMap[user.org_UUID]
-        if (org && org.admins && org.admins.includes(user.UUID)) {
-          user.role = 'ADMIN'
-        }
-        // If not admin, leave as is (undefined or empty or whatever it was)
-      })
-    } finally {
-      await session.endSession()
     }
 
+    returnValue.users.forEach(user => {
+      const org = orgMap[user.org_UUID]
+      if (org && org.admins && org.admins.includes(user.UUID)) {
+        user.role = 'ADMIN'
+      }
+      // If not admin, leave as is (undefined or empty or whatever it was)
+    })
+
     logger.info({ uuid: req.ctx.uuid, message: 'The user information was sent to the secretariat user.' })
     return res.status(200).json(returnValue)
   } catch (err) {
@@ -143,7 +137,7 @@ async function getUser (req, res, next) {
 }
 
 async function createUser (req, res, next) {
-  const session = await mongoose.startSession()
+  const session = await mongoose.startSession({ causalConsistency: false })
   try {
     const orgRepo = req.ctx.repositories.getBaseOrgRepository()
     const userRepo = req.ctx.repositories.getBaseUserRepository()

src/repositories/baseOrgRepository.js

@@ -331,7 +331,6 @@ class BaseOrgRepository extends BaseRepository {
     // In the future we may be able to dynamically detect, but for now we will take a boolean
     let legacyObjectRaw = null
     let registryObjectRaw = null
-    let legacyObject = null
     let registryObject = null
     const legacyOrgRepo = new OrgRepository()
     const ReviewObjectRepository = require('./reviewObjectRepository')
@@ -448,8 +447,9 @@ class BaseOrgRepository extends BaseRepository {
 
     // The legacy way of doing this, the way this is written under the hood there is no other way
     // This await does not return a value, even though there is a return in it. :shrugg:
+    let postUpdate = {}
     if (isSecretariat) {
-      await legacyOrgRepo.updateByOrgUUID(sharedUUID, legacyObjectRaw, options)
+      postUpdate = await legacyOrgRepo.updateByOrgUUID(sharedUUID, legacyObjectRaw, options)
     }
 
     // If we are not a secretariat, then we need to return the uuid of the review object.
@@ -459,12 +459,9 @@ class BaseOrgRepository extends BaseRepository {
 
     if (isLegacyObject) {
       // This gets us the mongoose object that has all the right data in it, the "legacyObjectRaw" is the custom JSON we are sending. NOT the post written object.
-      legacyObject = await legacyOrgRepo.findOneByShortName(
-        legacyObjectRaw.short_name,
-        options
-      )
       // Convert the actual model, back to a json model
-      const legacyObjectRawJson = legacyObject.toObject()
+
+      const legacyObjectRawJson = postUpdate.toObject()
       // Remove private stuff
       delete legacyObjectRawJson.__v
       delete legacyObjectRawJson._id
@@ -785,11 +782,11 @@ class BaseOrgRepository extends BaseRepository {
       updatedRegistryOrg = _.merge(registryOrg, _.omit(registryObjectRaw, jointApprovalFieldsRegistry))
       updatedLegacyOrg = _.merge(legacyOrg, _.omit(legacyObjectRaw, jointApprovalFieldsLegacy))
     }
-
     // handle conversation
     const requestingUser = await userRepo.findUserByUUID(requestingUserUUID, options)
+    const conversationArray = []
     if (conversation) {
-      await conversationRepo.createConversation(registryOrg.UUID, conversation, requestingUser, isSecretariat, options)
+      conversationArray.push(await conversationRepo.createConversation(registryOrg.UUID, conversation, requestingUser, isSecretariat, options))
     }
 
     // ADD AUDIT ENTRY AUTOMATICALLY for the registry object before it gets saved.
@@ -826,6 +823,7 @@ class BaseOrgRepository extends BaseRepository {
         }
         console.log('Audit entry created for registry object')
       } catch (auditError) {
+        console.error('Audit entry creation failed:', auditError)
       }
     }
 
@@ -878,6 +876,7 @@ class BaseOrgRepository extends BaseRepository {
     }
 
     const plainJavascriptRegistryOrg = updatedRegistryOrg.toObject()
+    plainJavascriptRegistryOrg.conversation = conversationArray
     // Remove private things
     delete plainJavascriptRegistryOrg.__v
     delete plainJavascriptRegistryOrg._id

src/repositories/baseRepository.js

@@ -59,7 +59,7 @@ class BaseRepository {
   }
 
   async findOneAndUpdate (query = {}, set = {}, options = {}) {
-    return this.collection.findOneAndUpdate(query, set, options)
+    return this.collection.findOneAndUpdate(query, set, { ...options, new: true })
   }
 
   async findOneAndReplace (query = {}, replacement = {}) {

src/repositories/orgRepository.js

@@ -24,7 +24,8 @@ class OrgRepository extends BaseRepository {
     // The filter to find the document
     const filter = { UUID: orgUUID }
     const updatePayload = { $set: updateData }
-    return this.collection.findOneAndUpdate(filter, updatePayload, executeOptions)
+    const data = await this.collection.findOneAndUpdate(filter, updatePayload, { ...executeOptions, new: true })
+    return data
   }
 
   async isSecretariat (org, options = {}) {