10592362701

Committed 28 Aug 2024 07:41AM UTC coverage: 32.008% (-0.007%) from 32.015%

Build # 10592362701

Build Type

Pull #9646

github

Committed by

jimrandomh

Commit Message

Add a setting (default "Hide Downvoted") for inline-react visibility, in the triple-dot menu

Pull Request Pull Request #9646: Add a setting (default "Hide Downvoted") for inline-react visibility, in the triple-dot menu

Run Details

5020 of 40814 branches covered (12.3%)

Branch coverage included in aggregate %.

24 of 53 new or added lines in 5 files covered. (45.28%)

328 existing lines in 6 files now uncovered.

29223 of 66167 relevant lines covered (44.17%)

189.23 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

8.95

/packages/lesswrong/server/serverSentEvents.ts

import type { Express, Response } from 'express';
import { getUserFromReq } from './vulcan-lib/apollo-server/context';
import { Notifications } from "../lib/collections/notifications/collection";
import { getSiteUrl } from "../lib/vulcan-lib/utils";
import { DatabaseServerSetting } from './databaseSettings';
import maxBy from 'lodash/maxBy';
import moment from 'moment';
import { getConfirmedCoauthorIds } from '../lib/collections/posts/helpers';
import { ActiveDialogue, ActiveDialogueServer, ServerSentEventsMessage, TypingIndicatorMessage } from '../components/hooks/useUnreadNotifications';
import TypingIndicatorsRepo from './repos/TypingIndicatorsRepo';
import UsersRepo from './repos/UsersRepo';
import { isEAForum } from '../lib/instanceSettings';

const disableServerSentEvents = new DatabaseServerSetting<boolean>("disableServerSentEvents", false);

interface ConnectionInfo {
  newestNotificationTimestamp: Date|null,
  res: Response
}

const openConnections: Record<string, ConnectionInfo[]> = {};

export function addServerSentEventsEndpoint(app: Express) {
  app.get('/api/notificationEvents', async (req, res) => {
    const parsedUrl = new URL(req.url, getSiteUrl())
    const apiVersionStr = parsedUrl.searchParams.get("version") ?? "1";
    const apiVersion = parseInt(apiVersionStr);
    const currentUser = getUserFromReq(req)

    // Can't subscribe to notifications if logged out
    if (!currentUser) {
      if (apiVersion===1) {
        // Wait awhile before closing the connection. A previous version of this
        // code would see the connection close, and try to reconnect after only
        // 3s, resulting in a very high number of reconnection requests. By
        // keeping the connection open for awhile (as we would for a logged-in
        // user's notifications channel), we slow down those requests.
        setTimeout(() => {
          // Status 204 ("no content") means the client should not rerequest,
          // according to the server-sent events spec.
          res.status(204).send();
        }, 120000);
      } else {
        res.status(204).send();
      }
      return;
    }
    
    
    res.setHeader('Cache-Control', 'no-cache');
    res.setHeader('Content-Type', 'text/event-stream');
    res.setHeader('Connection', 'keep-alive');
    res.flushHeaders(); // flush the headers to establish SSE with client
    
    // If the disableServerSentEvents setting is set, send a message to request
    // that this client not try to reconnect. Only if apiVersion>=2.
    if (apiVersion>=2 && disableServerSentEvents.get()) {
      res.write(`data: {"stop":true}\n\n`);
      res.end();
      return;
    }
    
    const userId = currentUser._id;
    if (!(userId in openConnections)) {
      openConnections[userId] = [{res, newestNotificationTimestamp: null}];
    } else {
      openConnections[userId].push({res, newestNotificationTimestamp: null});
    }

    // If client closes connection, stop sending events
    res.on('close', () => {
      openConnections[userId] = openConnections[userId].filter(r => r.res!==res);
      if (!openConnections[userId].length)
        delete openConnections[userId];
      res.end();
    });
  });
  
  setInterval(checkForNotifications, 1000);
  if (!isEAForum) {
    setInterval(checkForTypingIndicators, 1000);
    setInterval(checkForActiveDialoguePartners, 1000);
  }
}

export function sendSseMessageToUser(userId: string, message: ServerSentEventsMessage) {
  const userConnections = openConnections[userId];
  if (!userConnections) {
    // TODO: do we want to log an error here?  Probably not, it'll be happening reasonably often for innocous reasons
    // eslint-disable-next-line no-console
    console.log(`No connections found for user id ${userId}`, message);
    return;
  }

  for (let userConnection of userConnections) {
    userConnection.res.write(`data: ${JSON.stringify(message)}\n\n`);
  }
}

let lastNotificationCheck = new Date();
let lastTypingIndicatorsCheck = new Date();
let lastActiveDialoguePartnersMessage = new Date();

async function checkForNotifications() {
  const numOpenConnections = Object.keys(openConnections).length;
  if (!numOpenConnections) {
    return;
  }

  const newNotifications = await Notifications.find({
    createdAt: {$gt: lastNotificationCheck}
  }, {
    projection: {userId:1, createdAt:1}
  }).fetch();
  
  // TODO: Handle waitingForBatch
  // If a notification is batched, then delivering the batch means clearing the
  // waitingForBatch flag, but I don't think it updates the `createdAt` flag,
  // which will cause batched notifications to get missed.

  if (newNotifications.length > 0) {
    // Take the newest createdAt of a notification we saw, or one second ago,
    // whichever is earlier, as the cutoff date for the next query. The
    // one-second-ago case is to handle potential concurrency issues in the
    // database where, if two notifications are created at close to the same
    // time, then having received the one with the newer timestamp as a result
    // does not guarantee that the one with the older timestamp has actually
    // committed.
    // (This concurrency issue was inferred from theory, we did not observe a
    // problem happening in practice and expect that the problem would occur
    // rarely if this was wrong)
    const newestNotificationDate: Date = maxBy(newNotifications, n=>new Date(n.createdAt))!.createdAt;
    const oneSecondAgo = moment().subtract(1, 'seconds').toDate();
    if (newestNotificationDate > oneSecondAgo) {
      lastNotificationCheck = oneSecondAgo;
    } else {
      lastNotificationCheck = newestNotificationDate;
    }
  }

  const usersWithNewNotifications: Record<string,Date> = {};
  for (let notification of newNotifications) {
    const userId = notification.userId;
    if (userId in usersWithNewNotifications) {
      usersWithNewNotifications[userId] = dateMax(usersWithNewNotifications[userId], notification.createdAt);
    } else {
      usersWithNewNotifications[userId] = notification.createdAt;
    }
  }
  
  for (let userId of Object.keys(usersWithNewNotifications)) {
    const newTimestamp = usersWithNewNotifications[userId];
    if (openConnections[userId]) {
      for (let connection of openConnections[userId]) {
        if (!connection.newestNotificationTimestamp
          || connection.newestNotificationTimestamp < newTimestamp
        ) {
          const message: ServerSentEventsMessage = {
            eventType: "notificationCheck",
            newestNotificationTime: newTimestamp.toISOString(),
          };
          connection.res.write(`data: ${JSON.stringify(message)}\n\n`);
          connection.newestNotificationTimestamp = newTimestamp;
        }
      }
    }
  }
}

function dateMax(a: Date, b: Date) {
  if (a.getTime() > b.getTime())
    return a;
  else
    return b;
}


async function checkForTypingIndicators() {
  const numOpenConnections = Object.keys(openConnections).length;
  if (!numOpenConnections) {
    return;
  }

  const typingIndicatorInfos = await new TypingIndicatorsRepo().getRecentTypingIndicators(lastTypingIndicatorsCheck)

  if (typingIndicatorInfos.length > 0) {
    // Take the newest lastUpdated of a typingIndicator we saw, or one second ago,
    // whichever is earlier, as the cutoff date for the next query. 
    // See checkForNotifications for more details.
    const newestTypingIndicatorDate: Date = maxBy(typingIndicatorInfos, n=>new Date(n.lastUpdated))!.lastUpdated;
    const oneSecondAgo = moment().subtract(1, 'seconds').toDate();
    if (newestTypingIndicatorDate > oneSecondAgo) {
      lastTypingIndicatorsCheck = oneSecondAgo;
    } else {
      lastTypingIndicatorsCheck = newestTypingIndicatorDate;
    }
  }

  const results: Record<string, TypingIndicatorInfo[]> = {};
  for (const curr of typingIndicatorInfos) {
    // Get all userIds that have permission to type on the post
    const userIdsToNotify = [curr.postUserId, ...getConfirmedCoauthorIds(curr)].filter((userId) => userId !== curr.userId);
  
    for (const userIdToNotify of userIdsToNotify) {
      const {_id, userId, documentId, lastUpdated} = curr; // filter to just the fields in TypingIndicatorInfo
      if (results[userIdToNotify]) {
        results[userIdToNotify].push({_id, userId, documentId, lastUpdated});
      } else {
        results[userIdToNotify] = [{_id, userId, documentId, lastUpdated}];
      }
    }
  }
  
  for (let userId of Object.keys(results)) {
    if (openConnections[userId]) {
      for (let connection of openConnections[userId]) {
        const message: TypingIndicatorMessage = {
          eventType: "typingIndicator", 
          typingIndicators: results[userId],
        }
        connection.res.write(`data: ${JSON.stringify(message)}\n\n`)
      }
    }
  }
}

const isRecentlyActive = (editedAt: Date | undefined, minutes: number): boolean => {
  if (!editedAt) {
    return false;
  }

  const currentTime = new Date().getTime();
  const editedTime = new Date(editedAt).getTime();

  return (currentTime - editedTime) <= minutes * 60 * 1000;
}

async function checkForActiveDialoguePartners() {
  const numOpenConnections = Object.keys(openConnections).length;
  if (!numOpenConnections) {
    return;
  }

  const userIds = Object.keys(openConnections);

  const activeDialogues: ActiveDialogueServer[] = await new UsersRepo().getActiveDialogues(userIds);

  const allUsersDialoguesData: Record<string, ActiveDialogue[]> = {};
  for (let dialogue of activeDialogues) {
    const coauthorUserIds = dialogue.coauthorStatuses.map((status: any) => status.userId);
    const allUserIds = [dialogue.userId, ...coauthorUserIds];
    for (let userId of allUserIds) {
      const editedAt = dialogue?.mostRecentEditedAt;
      const data = {
        postId: dialogue._id,
        title: dialogue.title,
        userIds: dialogue.activeUserIds.filter((id => id !== userId)),
        mostRecentEditedAt: editedAt,
        anyoneRecentlyActive: isRecentlyActive(editedAt, 15) // within the last 15 min
      }
      if (allUsersDialoguesData[userId]) {
        allUsersDialoguesData[userId].push(data);
      } else {
        allUsersDialoguesData[userId] = [data];
      }
    }
  }

  for (let userId of userIds) {
    const userDialoguesData = allUsersDialoguesData[userId];
    const message = {
      eventType: "activeDialoguePartners",
      data: userDialoguesData ?? []
    };

    const messageString = `data: ${JSON.stringify(message)}\n\n`;

    if (openConnections[userId]) {
      for (let connection of openConnections[userId]) {
        connection.res.write(messageString);
        connection.newestNotificationTimestamp = new Date();
      } 
    }
  }
}

1	import type { Express, Response } from 'express';	54✔
2	import { getUserFromReq } from './vulcan-lib/apollo-server/context';	18✔
3	import { Notifications } from "../lib/collections/notifications/collection";	18✔
4	import { getSiteUrl } from "../lib/vulcan-lib/utils";	18✔
5	import { DatabaseServerSetting } from './databaseSettings';	18✔
6	import maxBy from 'lodash/maxBy';	18✔
7	import moment from 'moment';	18✔
8	import { getConfirmedCoauthorIds } from '../lib/collections/posts/helpers';	18✔
9	import { ActiveDialogue, ActiveDialogueServer, ServerSentEventsMessage, TypingIndicatorMessage } from '../components/hooks/useUnreadNotifications';
10	import TypingIndicatorsRepo from './repos/TypingIndicatorsRepo';	18✔
11	import UsersRepo from './repos/UsersRepo';	18✔
12	import { isEAForum } from '../lib/instanceSettings';	18✔
13
14	const disableServerSentEvents = new DatabaseServerSetting<boolean>("disableServerSentEvents", false);	18✔
15
16	interface ConnectionInfo {
17	newestNotificationTimestamp: Date\|null,
18	res: Response
19	}
20
21	const openConnections: Record<string, ConnectionInfo[]> = {};	18✔
22
23	export function addServerSentEventsEndpoint(app: Express) {
24	app.get('/api/notificationEvents', async (req, res) => {	×
25	const parsedUrl = new URL(req.url, getSiteUrl())	×
26	const apiVersionStr = parsedUrl.searchParams.get("version") ?? "1";	×
27	const apiVersion = parseInt(apiVersionStr);	×
28	const currentUser = getUserFromReq(req)	×
29
30	// Can't subscribe to notifications if logged out
31	if (!currentUser) {	×
32	if (apiVersion===1) {	×
33	// Wait awhile before closing the connection. A previous version of this
34	// code would see the connection close, and try to reconnect after only
35	// 3s, resulting in a very high number of reconnection requests. By
36	// keeping the connection open for awhile (as we would for a logged-in
37	// user's notifications channel), we slow down those requests.
38	setTimeout(() => {	×
39	// Status 204 ("no content") means the client should not rerequest,
40	// according to the server-sent events spec.
41	res.status(204).send();	×
42	}, 120000);
43	} else {
44	res.status(204).send();	×
45	}
46	return;	×
47	}
48
49
50	res.setHeader('Cache-Control', 'no-cache');	×
51	res.setHeader('Content-Type', 'text/event-stream');	×
52	res.setHeader('Connection', 'keep-alive');	×
53	res.flushHeaders(); // flush the headers to establish SSE with client	×
54
55	// If the disableServerSentEvents setting is set, send a message to request
56	// that this client not try to reconnect. Only if apiVersion>=2.
57	if (apiVersion>=2 && disableServerSentEvents.get()) {	×
58	res.write(`data: {"stop":true}\n\n`);	×
59	res.end();	×
60	return;	×
61	}
62
63	const userId = currentUser._id;	×
64	if (!(userId in openConnections)) {	×
65	openConnections[userId] = [{res, newestNotificationTimestamp: null}];	×
66	} else {
67	openConnections[userId].push({res, newestNotificationTimestamp: null});	×
68	}
69
70	// If client closes connection, stop sending events
71	res.on('close', () => {	×
72	openConnections[userId] = openConnections[userId].filter(r => r.res!==res);	×
73	if (!openConnections[userId].length)	×
74	delete openConnections[userId];	×
75	res.end();	×
76	});
77	});
78
79	setInterval(checkForNotifications, 1000);	×
80	if (!isEAForum) {	×
81	setInterval(checkForTypingIndicators, 1000);	×
82	setInterval(checkForActiveDialoguePartners, 1000);	×
83	}
84	}
85
86	export function sendSseMessageToUser(userId: string, message: ServerSentEventsMessage) {
UNCOV 87	const userConnections = openConnections[userId];	×
UNCOV 88	if (!userConnections) {	×
89	// TODO: do we want to log an error here? Probably not, it'll be happening reasonably often for innocous reasons
90	// eslint-disable-next-line no-console
UNCOV 91	console.log(`No connections found for user id ${userId}`, message);	×
UNCOV 92	return;	×
93	}
94
UNCOV 95	for (let userConnection of userConnections) {	×
UNCOV 96	userConnection.res.write(`data: ${JSON.stringify(message)}\n\n`);	×
97	}
98	}
99
100	let lastNotificationCheck = new Date();	18✔
101	let lastTypingIndicatorsCheck = new Date();	18✔
102	let lastActiveDialoguePartnersMessage = new Date();	18✔
103
104	async function checkForNotifications() {
UNCOV 105	const numOpenConnections = Object.keys(openConnections).length;	×
UNCOV 106	if (!numOpenConnections) {	×
UNCOV 107	return;	×
108	}
109
UNCOV 110	const newNotifications = await Notifications.find({	×
111	createdAt: {$gt: lastNotificationCheck}
112	}, {
113	projection: {userId:1, createdAt:1}
114	}).fetch();
115
116	// TODO: Handle waitingForBatch
117	// If a notification is batched, then delivering the batch means clearing the
118	// waitingForBatch flag, but I don't think it updates the `createdAt` flag,
119	// which will cause batched notifications to get missed.
120
UNCOV 121	if (newNotifications.length > 0) {	×
122	// Take the newest createdAt of a notification we saw, or one second ago,
123	// whichever is earlier, as the cutoff date for the next query. The
124	// one-second-ago case is to handle potential concurrency issues in the
125	// database where, if two notifications are created at close to the same
126	// time, then having received the one with the newer timestamp as a result
127	// does not guarantee that the one with the older timestamp has actually
128	// committed.
129	// (This concurrency issue was inferred from theory, we did not observe a
130	// problem happening in practice and expect that the problem would occur
131	// rarely if this was wrong)
UNCOV 132	const newestNotificationDate: Date = maxBy(newNotifications, n=>new Date(n.createdAt))!.createdAt;	×
UNCOV 133	const oneSecondAgo = moment().subtract(1, 'seconds').toDate();	×
UNCOV 134	if (newestNotificationDate > oneSecondAgo) {	×
UNCOV 135	lastNotificationCheck = oneSecondAgo;	×
136	} else {
UNCOV 137	lastNotificationCheck = newestNotificationDate;	×
138	}
139	}
140
141	const usersWithNewNotifications: Record<string,Date> = {};	×
UNCOV 142	for (let notification of newNotifications) {	×
143	const userId = notification.userId;	×
UNCOV 144	if (userId in usersWithNewNotifications) {	×
UNCOV 145	usersWithNewNotifications[userId] = dateMax(usersWithNewNotifications[userId], notification.createdAt);	×
146	} else {
147	usersWithNewNotifications[userId] = notification.createdAt;	×
148	}
149	}
150
151	for (let userId of Object.keys(usersWithNewNotifications)) {	×
UNCOV 152	const newTimestamp = usersWithNewNotifications[userId];	×
153	if (openConnections[userId]) {	×
UNCOV 154	for (let connection of openConnections[userId]) {	×
UNCOV 155	if (!connection.newestNotificationTimestamp	×
156	\|\| connection.newestNotificationTimestamp < newTimestamp
157	) {
158	const message: ServerSentEventsMessage = {	×
159	eventType: "notificationCheck",
160	newestNotificationTime: newTimestamp.toISOString(),
161	};
UNCOV 162	connection.res.write(`data: ${JSON.stringify(message)}\n\n`);	×
UNCOV 163	connection.newestNotificationTimestamp = newTimestamp;	×
164	}
165	}
166	}
167	}
168	}
169
170	function dateMax(a: Date, b: Date) {
UNCOV 171	if (a.getTime() > b.getTime())	×
UNCOV 172	return a;	×
173	else
UNCOV 174	return b;	×
175	}
176
177
178	async function checkForTypingIndicators() {
UNCOV 179	const numOpenConnections = Object.keys(openConnections).length;	×
180	if (!numOpenConnections) {	×
UNCOV 181	return;	×
182	}
183
UNCOV 184	const typingIndicatorInfos = await new TypingIndicatorsRepo().getRecentTypingIndicators(lastTypingIndicatorsCheck)	×
185
186	if (typingIndicatorInfos.length > 0) {	×
187	// Take the newest lastUpdated of a typingIndicator we saw, or one second ago,
188	// whichever is earlier, as the cutoff date for the next query.
189	// See checkForNotifications for more details.
190	const newestTypingIndicatorDate: Date = maxBy(typingIndicatorInfos, n=>new Date(n.lastUpdated))!.lastUpdated;	×
UNCOV 191	const oneSecondAgo = moment().subtract(1, 'seconds').toDate();	×
192	if (newestTypingIndicatorDate > oneSecondAgo) {	×
UNCOV 193	lastTypingIndicatorsCheck = oneSecondAgo;	×
194	} else {
UNCOV 195	lastTypingIndicatorsCheck = newestTypingIndicatorDate;	×
196	}
197	}
198
199	const results: Record<string, TypingIndicatorInfo[]> = {};	×
UNCOV 200	for (const curr of typingIndicatorInfos) {	×
201	// Get all userIds that have permission to type on the post
UNCOV 202	const userIdsToNotify = [curr.postUserId, ...getConfirmedCoauthorIds(curr)].filter((userId) => userId !== curr.userId);	×
203
UNCOV 204	for (const userIdToNotify of userIdsToNotify) {	×
205	const {_id, userId, documentId, lastUpdated} = curr; // filter to just the fields in TypingIndicatorInfo	×
206	if (results[userIdToNotify]) {	×
UNCOV 207	results[userIdToNotify].push({_id, userId, documentId, lastUpdated});	×
208	} else {
UNCOV 209	results[userIdToNotify] = [{_id, userId, documentId, lastUpdated}];	×
210	}
211	}
212	}
213
UNCOV 214	for (let userId of Object.keys(results)) {	×
215	if (openConnections[userId]) {	×
UNCOV 216	for (let connection of openConnections[userId]) {	×
UNCOV 217	const message: TypingIndicatorMessage = {	×
218	eventType: "typingIndicator",
219	typingIndicators: results[userId],
220	}
221	connection.res.write(`data: ${JSON.stringify(message)}\n\n`)	×
222	}
223	}
224	}
225	}
226
227	const isRecentlyActive = (editedAt: Date \| undefined, minutes: number): boolean => {	18✔
UNCOV 228	if (!editedAt) {	×
UNCOV 229	return false;	×
230	}
231
UNCOV 232	const currentTime = new Date().getTime();	×
UNCOV 233	const editedTime = new Date(editedAt).getTime();	×
234
235	return (currentTime - editedTime) <= minutes * 60 * 1000;	×
236	}
237
238	async function checkForActiveDialoguePartners() {
239	const numOpenConnections = Object.keys(openConnections).length;	×
UNCOV 240	if (!numOpenConnections) {	×
241	return;	×
242	}
243
UNCOV 244	const userIds = Object.keys(openConnections);	×
245
246	const activeDialogues: ActiveDialogueServer[] = await new UsersRepo().getActiveDialogues(userIds);	×
247
UNCOV 248	const allUsersDialoguesData: Record<string, ActiveDialogue[]> = {};	×
UNCOV 249	for (let dialogue of activeDialogues) {	×
250	const coauthorUserIds = dialogue.coauthorStatuses.map((status: any) => status.userId);	×
UNCOV 251	const allUserIds = [dialogue.userId, ...coauthorUserIds];	×
252	for (let userId of allUserIds) {	×
UNCOV 253	const editedAt = dialogue?.mostRecentEditedAt;	×
254	const data = {	×
255	postId: dialogue._id,
256	title: dialogue.title,
257	userIds: dialogue.activeUserIds.filter((id => id !== userId)),	×
258	mostRecentEditedAt: editedAt,
259	anyoneRecentlyActive: isRecentlyActive(editedAt, 15) // within the last 15 min
260	}
UNCOV 261	if (allUsersDialoguesData[userId]) {	×
UNCOV 262	allUsersDialoguesData[userId].push(data);	×
263	} else {
UNCOV 264	allUsersDialoguesData[userId] = [data];	×
265	}
266	}
267	}
268
UNCOV 269	for (let userId of userIds) {	×
270	const userDialoguesData = allUsersDialoguesData[userId];	×
UNCOV 271	const message = {	×
272	eventType: "activeDialoguePartners",
273	data: userDialoguesData ?? []	×
274	};
275
276	const messageString = `data: ${JSON.stringify(message)}\n\n`;	×
277
UNCOV 278	if (openConnections[userId]) {	×
UNCOV 279	for (let connection of openConnections[userId]) {	×
UNCOV 280	connection.res.write(messageString);	×
UNCOV 281	connection.newestNotificationTimestamp = new Date();	×
282	}
283	}
284	}
285	}

ForumMagnum / ForumMagnum / 10592362701

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous