Frecency thread feeds (second try)

api/models/thread.js

@@ -71,14 +71,27 @@ export const getThreadsByChannel = (channelId: string, options: PaginationOption
 };
 
 // prettier-ignore
-export const getThreadsByChannels = (channelIds: Array<string>, options: PaginationOptions): Promise<Array<DBThread>> => {
-  const { first, after } = options
-
+type GetThreadsByChannelPaginationOptions = {
+  first: number,
+  after: number,
+  sort: 'new' | 'trending'
+};
+
+export const getThreadsByChannels = (
+  channelIds: Array<string>,
+  options: GetThreadsByChannelPaginationOptions
+): Promise<Array<DBThread>> => {
+  const { first, after, sort } = options;
+
+  let order = [db.desc('lastActive'), db.desc('createdAt')];
+  // If we want the top threads, first sort by the score and then lastActive
+  if (sort === 'trending') order.unshift(db.desc('score'));
+
   return db
     .table('threads')
     .getAll(...channelIds, { index: 'channelId' })
     .filter(thread => db.not(thread.hasFields('deletedAt')))
-    .orderBy(db.desc('lastActive'), db.desc('createdAt'))
+    .orderBy(...order)
     .skip(after || 0)
     .limit(first || 999999)
     .run();
@@ -447,10 +460,10 @@ export const setThreadLock = (threadId: string, value: boolean, userId: string,
       .run()
       .then(async () => {
         const thread = await getThreadById(threadId)
-        
-        const event = value 
-          ? byModerator 
-            ? events.THREAD_LOCKED_BY_MODERATOR 
+
+        const event = value
+          ? byModerator
+            ? events.THREAD_LOCKED_BY_MODERATOR
             : events.THREAD_LOCKED
           : byModerator
             ? events.THREAD_UNLOCKED_BY_MODERATOR

api/mutations/message/addMessage.js

@@ -13,7 +13,7 @@ import { trackUserThreadLastSeenQueue } from 'shared/bull/queues';
 import type { FileUpload } from 'shared/types';
 import { events } from 'shared/analytics';
 import { isAuthedResolver as requireAuth } from '../../utils/permissions';
-import { trackQueue } from 'shared/bull/queues';
+import { trackQueue, calculateThreadScoreQueue } from 'shared/bull/queues';
 
 type Input = {
   message: {
@@ -344,6 +344,14 @@ export default requireAuth(async (_: any, args: Input, ctx: GraphQLContext) => {
         timestamp: Date.now(),
       });
 
+      calculateThreadScoreQueue.add(
+        {
+          threadId: message.threadId,
+        },
+        {
+          jobId: message.threadId,
+        }
+      );
       return {
         ...dbMessage,
         contextPermissions,

api/mutations/message/deleteMessage.js

@@ -13,7 +13,7 @@ import { getUserPermissionsInChannel } from '../../models/usersChannels';
 import { getUserPermissionsInCommunity } from '../../models/usersCommunities';
 import { events } from 'shared/analytics';
 import { isAuthedResolver as requireAuth } from '../../utils/permissions';
-import { trackQueue } from 'shared/bull/queues';
+import { trackQueue, calculateThreadScoreQueue } from 'shared/bull/queues';
 
 type Input = {
   id: string,
@@ -116,5 +116,15 @@ export default requireAuth(async (_: any, args: Input, ctx: GraphQLContext) => {
         message.senderId
       );
     })
+    .then(() => {
+      return calculateThreadScoreQueue.add(
+        {
+          threadId: message.threadId,
+        },
+        {
+          jobId: message.threadId,
+        }
+      );
+    })
     .then(() => true);
 });

api/mutations/thread/addThreadReaction.js

@@ -3,6 +3,7 @@ import type { GraphQLContext } from '../../';
 import { addThreadReaction } from '../../models/threadReaction';
 import { getThreadById } from '../../models/thread';
 import { isAuthedResolver as requireAuth } from '../../utils/permissions';
+import { calculateThreadScoreQueue } from 'shared/bull/queues';
 
 type Input = {
   input: {
@@ -13,8 +14,16 @@ type Input = {
 
 export default requireAuth(
   async (_: any, args: Input, { user, loaders }: GraphQLContext) => {
-    return await addThreadReaction(args.input, user.id).then(
-      async () => await getThreadById(args.input.threadId)
-    );
+    return await addThreadReaction(args.input, user.id).then(() => {
+      calculateThreadScoreQueue.add(
+        {
+          threadId: args.input.threadId,
+        },
+        {
+          jobId: args.input.threadId,
+        }
+      );
+      return getThreadById(args.input.threadId);
+    });
   }
 );

api/mutations/thread/removeThreadReaction.js

@@ -3,6 +3,7 @@ import type { GraphQLContext } from '../../';
 import { removeThreadReaction } from '../../models/threadReaction';
 import { getThreadById } from '../../models/thread';
 import { isAuthedResolver as requireAuth } from '../../utils/permissions';
+import { calculateThreadScoreQueue } from 'shared/bull/queues';
 
 type Input = {
   input: {
@@ -12,8 +13,16 @@ type Input = {
 
 export default requireAuth(
   async (_: any, args: Input, { user, loaders }: GraphQLContext) => {
-    return await removeThreadReaction(args.input.threadId, user.id).then(
-      async () => await getThreadById(args.input.threadId)
-    );
+    return await removeThreadReaction(args.input.threadId, user.id).then(() => {
+      calculateThreadScoreQueue.add(
+        {
+          threadId: args.input.threadId,
+        },
+        {
+          jobId: args.input.threadId,
+        }
+      );
+      return getThreadById(args.input.threadId);
+    });
   }
 );

api/queries/community/threadConnection.js

@@ -10,9 +10,15 @@ import {
 import { getThreadsByChannels } from '../../models/thread';
 import { canViewCommunity } from '../../utils/permissions';
 
+export type CommunityThreadConnectionPaginationOptions = {
+  after: string,
+  first: number,
+  sort: 'new' | 'trending',
+};
+
 // prettier-ignore
-export default async (root: DBCommunity, args: PaginationOptions, ctx: GraphQLContext) => {
-  const { first = 10, after } = args
+export default async (root: DBCommunity, args: CommunityThreadConnectionPaginationOptions, ctx: GraphQLContext) => {
+  const { first = 10, after, sort = 'new' } = args
   const { user, loaders } = ctx
   const { id } = root
 
@@ -29,7 +35,7 @@ export default async (root: DBCommunity, args: PaginationOptions, ctx: GraphQLCo
   // Get the index from the encoded cursor, asdf234gsdf-2 => ["-2", "2"]
   const lastDigits = cursor.match(/-(\d+)$/);
   const lastThreadIndex =
-    lastDigits && lastDigits.length > 0 && parseInt(lastDigits[1], 10);
+    lastDigits && lastDigits.length > 0 && parseInt(lastDigits[1], 10) || 0;
   const currentUser = user;
 
   // if the user is signed in, only return stories for the channels
@@ -44,10 +50,10 @@ export default async (root: DBCommunity, args: PaginationOptions, ctx: GraphQLCo
     channels = await getPublicChannelsByCommunity(id);
   }
 
-  // $FlowFixMe
   const threads = await getThreadsByChannels(channels, {
     first,
     after: lastThreadIndex,
+    sort,
   });
 
   return {

api/queries/thread/rootThread.js

@@ -1,4 +1,5 @@
 // @flow
+import { calculateThreadScoreQueue } from 'shared/bull/queues';
 import type { GraphQLContext } from '../../';
 
 export default async (
@@ -50,6 +51,23 @@ export default async (
       (!communityPermissions || !communityPermissions.isMember)
     )
       return null;
+
+    // If the threads score hasn't been updated in the past
+    // 24 hours add a new job to the queue to update it
+    if (
+      (!thread.score && !thread.scoreUpdatedAt) ||
+      (thread.scoreUpdatedAt &&
+        Date.now() > new Date(thread.scoreUpdatedAt).getTime() + 86400000)
+    ) {
+      calculateThreadScoreQueue.add(
+        {
+          threadId: thread.id,
+        },
+        {
+          jobId: thread.id,
+        }
+      );
+    }
     return thread;
   }
 };

api/types/Community.js

@@ -130,6 +130,11 @@ const Community = /* GraphQL */ `
     message: String
   }
 
+  enum CommunityThreadConnectionSort {
+    new
+    trending
+  }
+
   type Community {
     id: ID!
     createdAt: Date
@@ -153,6 +158,7 @@ const Community = /* GraphQL */ `
     threadConnection(
       first: Int = 10
       after: String
+      sort: CommunityThreadConnectionSort = new
     ): CommunityThreadsConnection @cost(complexity: 2, multiplier: "first")
     metaData: CommunityMetaData @cost(complexity: 10)
     invoices: [Invoice] @cost(complexity: 1)

mercury/constants.js

@@ -2,6 +2,7 @@
 
 // queues
 export const PROCESS_REPUTATION_EVENT = 'process reputation event';
+export const CALCULATE_THREAD_SCORE = 'calculate thread score';
 
 // reputation event types
 export const THREAD_CREATED = 'thread created';

mercury/index.js

@@ -1,7 +1,8 @@
 const debug = require('debug')('mercury');
 const createWorker = require('../shared/bull/create-worker');
 import processReputationEvent from './queues/processReputationEvent';
-import { PROCESS_REPUTATION_EVENT } from './constants';
+import calculateThreadScore from './queues/calculateThreadScore';
+import { PROCESS_REPUTATION_EVENT, CALCULATE_THREAD_SCORE } from './constants';
 
 const PORT = process.env.PORT || 3005;
 
@@ -10,6 +11,7 @@ debug('Logging with debug enabled!');
 
 const server = createWorker({
   [PROCESS_REPUTATION_EVENT]: processReputationEvent,
+  [CALCULATE_THREAD_SCORE]: calculateThreadScore,
 });
 
 debug(

mercury/models/thread.js

@@ -1,9 +1,109 @@
 // @flow
 const { db } = require('./db');
+import type { DBThread } from 'shared/types';
 
-export const getThread = (id: string): Promise<Object> => {
+export const getThread = (id: string): Promise<DBThread> => {
   return db
     .table('threads')
     .get(id)
     .run();
 };
+
+export const getParticipantCount = (threadId: string): Promise<number> => {
+  return db
+    .table('usersThreads')
+    .getAll(threadId, { index: 'threadId' })
+    .filter({ isParticipant: true })
+    .count()
+    .default(0)
+    .run();
+};
+
+const timeRanges = {
+  hourly: { start: 3600, end: 0 },
+  daily: { start: 86400, end: 3600 },
+  weekly: { start: 604800, end: 86400 },
+  rest: { start: Date.now(), end: 604800 },
+};
+
+export const getParticipantCountByTime = (
+  threadId: string,
+  range: 'hourly' | 'daily' | 'weekly' | 'rest'
+): Promise<number> => {
+  return db
+    .table('messages')
+    .getAll(threadId, { index: 'threadId' })
+    .filter(
+      db.row
+        .hasFields('deletedAt')
+        .not()
+        .and(
+          db
+            .row('timestamp')
+            .ge(db.now().sub(timeRanges[range].start))
+            .and(db.row('timestamp').le(db.now().sub(timeRanges[range].end)))
+        )
+    )
+    .map(rec => rec('senderId'))
+    .distinct()
+    .count()
+    .default(0)
+    .run();
+};
+
+export const getReactionCountByTime = (
+  threadId: string,
+  range: 'hourly' | 'daily' | 'weekly' | 'rest'
+): Promise<number> => {
+  return db
+    .table('threadReactions')
+    .getAll(threadId, { index: 'threadId' })
+    .filter(
+      db.row
+        .hasFields('deletedAt')
+        .not()
+        .and(
+          db
+            .row('createdAt')
+            .ge(db.now().sub(timeRanges[range].start))
+            .and(db.row('createdAt').le(db.now().sub(timeRanges[range].end)))
+        )
+    )
+    .count()
+    .default(0)
+    .run();
+};
+
+export const getMessageCount = (threadId: string): Promise<number> => {
+  return db
+    .table('messages')
+    .getAll(threadId, { index: 'threadId' })
+    .filter(db.row.hasFields('deletedAt').not())
+    .count()
+    .default(0)
+    .run();
+};
+
+export const getReactionCount = (threadId: string): Promise<number> => {
+  return db
+    .table('threadReactions')
+    .getAll(threadId, { index: 'threadId' })
+    .filter(row => row.hasFields('deletedAt').not())
+    .count()
+    .default(0)
+    .run();
+};
+
+export const storeThreadScore = (
+  threadId: string,
+  score: number
+): Promise<any> => {
+  return db
+    .table('threads')
+    .get(threadId)
+    .update({
+      score,
+      scoreUpdatedAt: new Date(),
+    })
+    .run();
+};