Merge pull request #2533 from withspectrum/last-merge

Last merge

Author: brianlovin

iris/models/directMessageThread.js

@@ -1,6 +1,6 @@
 //@flow
 const { db } = require('./db');
-import { NEW_DOCUMENTS } from './utils';
+import { NEW_DOCUMENTS, eachAsyncNewValue } from './utils';
 
 export type DBDirectMessageThread = {
   createdAt: Date,
@@ -95,14 +95,7 @@ const listenToUpdatedDirectMessageThreads = (userId: string) => (
       right: ['id', 'createdAt', 'threadId', 'lastActive', 'lastSeen'],
     })
     .zip()
-    .run({ cursor: true }, (err, cursor) => {
-      if (err) throw err;
-      cursor.each((err, data) => {
-        if (err) throw err;
-        // Call the passed callback with the notification
-        cb(data);
-      });
-    });
+    .run(eachAsyncNewValue(cb));
 };
 
 // prettier-ignore

iris/models/message.js

@@ -6,7 +6,7 @@ import {
   processReputationEventQueue,
   _adminProcessToxicMessageQueue,
 } from 'shared/bull/queues';
-import { NEW_DOCUMENTS } from './utils';
+import { NEW_DOCUMENTS, eachAsyncNewValue } from './utils';
 import { setThreadLastActive } from './thread';
 
 export type MessageTypes = 'text' | 'media';
@@ -159,16 +159,8 @@ export const listenToNewMessagesInThread = (threadId: string) => (
     .changes({
       includeInitial: false,
     })
-    .filter(NEW_DOCUMENTS)
-    .run({ cursor: true }, (err, cursor) => {
-      if (err) throw err;
-      cursor.each((err, data) => {
-        // TODO(@mxstbr): Maybe we need to cursor.close here?
-        if (err) throw err;
-        // Call the passed callback with the message directly
-        cb(data.new_val);
-      });
-    });
+    .filter(NEW_DOCUMENTS)('new_val')
+    .run(eachAsyncNewValue(cb));
 };
 
 export const getMessageCount = (threadId: string): Promise<number> => {

iris/models/notification.js

@@ -1,6 +1,6 @@
 // @flow
 const { db } = require('./db');
-import { NEW_DOCUMENTS } from './utils';
+import { NEW_DOCUMENTS, eachAsyncNewValue } from './utils';
 
 export const getNotificationsByUser = (
   userId: string,
@@ -78,17 +78,7 @@ export const listenToNewNotifications = (userId: string) => (
     })
     .zip()
     .filter(row => row('context')('type').ne('DIRECT_MESSAGE_THREAD'))
-    .run({ cursor: true }, (err, cursor) => {
-      if (err) throw err;
-      cursor.each((err, data) => {
-        if (err) throw err;
-        // For some reason this can be called without data, in which case
-        // we don't want to call the callback with it obviously
-        if (!data) return;
-        // Call the passed callback with the notification
-        cb(data);
-      });
-    });
+    .run(eachAsyncNewValue(cb));
 };
 
 export const listenToNewDirectMessageNotifications = (userId: string) => (
@@ -107,12 +97,5 @@ export const listenToNewDirectMessageNotifications = (userId: string) => (
     })
     .zip()
     .filter(row => row('context')('type').eq('DIRECT_MESSAGE_THREAD'))
-    .run({ cursor: true }, (err, cursor) => {
-      if (err) throw err;
-      cursor.each((err, data) => {
-        if (err) throw err;
-        // Call the passed callback with the notification
-        cb(data);
-      });
-    });
+    .run(eachAsyncNewValue(cb));
 };

iris/models/thread.js

@@ -6,7 +6,7 @@ import {
   sendThreadNotificationQueue,
   _adminProcessToxicThreadQueue,
 } from 'shared/bull/queues';
-const { NEW_DOCUMENTS, parseRange } = require('./utils');
+const { NEW_DOCUMENTS, parseRange, eachAsyncNewValue } = require('./utils');
 import { deleteMessagesInThread } from '../models/message';
 import { turnOffAllThreadNotifications } from '../models/usersThreads';
 import type { PaginationOptions } from '../utils/paginate-arrays';
@@ -517,12 +517,5 @@ export const listenToUpdatedThreads = (channelIds: Array<string>) => (
     .filter(thread =>
       db.expr(channelIds).contains(thread('new_val')('channelId'))
     )('new_val')
-    .run({ cursor: true }, (err, cursor) => {
-      if (err) throw err;
-      cursor.each((err, data) => {
-        if (err) throw err;
-        // Call the passed callback with the notification
-        cb(data);
-      });
-    });
+    .run(eachAsyncNewValue(cb));
 };

iris/models/utils.js

@@ -5,6 +5,24 @@ export const NEW_DOCUMENTS = db
   .eq(null)
   .and(db.not(db.row('new_val').eq(null)));
 
+export const eachAsyncNewValue = (cb: Function) => (
+  err?: Error,
+  cursor: Cursor
+) => {
+  if (err) throw err;
+  cursor
+    .eachAsync(data => {
+      // Call the passed callback with the message directly
+      cb(data);
+    })
+    .catch(err => {
+      console.error(err);
+      try {
+        cursor.close();
+      } catch (err) {}
+    });
+};
+
 export const listenToNewDocumentsIn = (table, cb) => {
   return (
     db
@@ -14,14 +32,7 @@ export const listenToNewDocumentsIn = (table, cb) => {
       })
       // Filter to only include newly inserted messages in the changefeed
       .filter(NEW_DOCUMENTS)
-      .run({ cursor: true }, (err, cursor) => {
-        if (err) throw err;
-        cursor.each((err, data) => {
-          if (err) throw err;
-          // Call the passed callback with the message directly
-          cb(data.new_val);
-        });
-      })
+      .run(eachAsyncNewValue(cb))
   );
 };
 

iris/mutations/message/addMessage.js

@@ -48,13 +48,12 @@ export default async (
       }".`
     );
 
-  const thread = await loaders.thread.load(message.threadId);
-
-  if (thread.isLocked) throw new UserError("Can't reply in a locked thread.");
-
   let contextPermissions;
+  let thread;
   // Make sure that we have permission to send a message in the community
   if (message.threadType === 'story') {
+    thread = await loaders.thread.load(message.threadId);
+    if (thread.isLocked) throw new UserError("Can't reply in a locked thread.");
     const permissions = await loaders.userPermissionsInCommunity.load([
       currentUser.id,
       thread.communityId,