message: [FIX] Pending update
diff: @@ -160,22 +160,17 @@ export default class RoomSubscription {
  const msgCollection = db.collections.get('messages');
  const threadsCollection = db.collections.get('threads');
  const threadMessagesCollection = db.collections.get('thread_messages');
  - let messageRecord;
  - let threadRecord;
  - let threadMessageRecord;
  // Create or update message
  try {
  - messageRecord = await msgCollection.find(message._id);
  - } catch (error) {
  - // Do nothing
  - }
  - if (messageRecord) {
  - const update = messageRecord.prepareUpdate((m) => {
  + const messageRecord = await msgCollection.find(message._id);
  + if (!messageRecord._hasPendingUpdate) {
  + const update = messageRecord.prepareUpdate(protectedFunction((m) => {
  Object.assign(m, message);
  - });
  + }));
  this._messagesBatch[message._id] = update;
  - } else {
  + }
  + } catch {
  const create = msgCollection.prepareCreate(protectedFunction((m) => {
  m._raw = sanitizedRaw({ id: message._id }, msgCollection.schema);
  m.subscription.id = this.rid;
  @@ -187,17 +182,14 @@ export default class RoomSubscription {
  // Create or update thread
  if (message.tlm) {
  try {
  - threadRecord = await threadsCollection.find(message._id);
  - } catch (error) {
  - // Do nothing
  - }
  -
  - if (threadRecord) {
  + const threadRecord = await threadsCollection.find(message._id);
  + if (!threadRecord._hasPendingUpdate) {
  const updateThread = threadRecord.prepareUpdate(protectedFunction((t) => {
  Object.assign(t, message);
  }));
  this._threadsBatch[message._id] = updateThread;
  - } else {
  + }
  + } catch {
  const createThread = threadsCollection.prepareCreate(protectedFunction((t) => {
  t._raw = sanitizedRaw({ id: message._id }, threadsCollection.schema);
  t.subscription.id = this.rid;
  @@ -210,19 +202,16 @@ export default class RoomSubscription {
  // Create or update thread message
  if (message.tmid) {
  try {
  - threadMessageRecord = await threadMessagesCollection.find(message._id);
  - } catch (error) {
  - // Do nothing
  - }
  -
  - if (threadMessageRecord) {
  + const threadMessageRecord = await threadMessagesCollection.find(message._id);
  + if (!threadMessageRecord._hasPendingUpdate) {
  const updateThreadMessage = threadMessageRecord.prepareUpdate(protectedFunction((tm) => {
  Object.assign(tm, message);
  tm.rid = message.tmid;
  delete tm.tmid;
  }));
  this._threadMessagesBatch[message._id] = updateThreadMessage;
  - } else {
  + }
  + } catch {
  const createThreadMessage = threadMessagesCollection.prepareCreate(protectedFunction((tm) => {
  tm._raw = sanitizedRaw({ id: message._id }, threadMessagesCollection.schema);
  Object.assign(tm, message);
  