diff --git a/keyserver/src/push/send.js b/keyserver/src/push/send.js index 36f4241f8..430093b0a 100644 --- a/keyserver/src/push/send.js +++ b/keyserver/src/push/send.js @@ -1,841 +1,835 @@ // @flow import apn from '@parse/node-apn'; import type { ResponseFailure } from '@parse/node-apn'; import invariant from 'invariant'; import _cloneDeep from 'lodash/fp/cloneDeep'; import _flow from 'lodash/fp/flow'; import _mapValues from 'lodash/fp/mapValues'; import _pickBy from 'lodash/fp/pickBy'; import uuidv4 from 'uuid/v4'; import { oldValidUsernameRegex } from 'lib/shared/account-utils'; import { createMessageInfo, sortMessageInfoList, shimUnsupportedRawMessageInfos, } from 'lib/shared/message-utils'; import { messageSpecs } from 'lib/shared/messages/message-specs'; import { notifTextsForMessageInfo } from 'lib/shared/notif-utils'; import { rawThreadInfoFromServerThreadInfo, threadInfoFromRawThreadInfo, } from 'lib/shared/thread-utils'; import type { DeviceType } from 'lib/types/device-types'; import { type RawMessageInfo, type MessageInfo, messageTypes, } from 'lib/types/message-types'; import type { ServerThreadInfo, ThreadInfo } from 'lib/types/thread-types'; import { updateTypes } from 'lib/types/update-types'; import { promiseAll } from 'lib/utils/promises'; import createIDs from '../creators/id-creator'; import { createUpdates } from '../creators/update-creator'; import { dbQuery, SQL, mergeOrConditions } from '../database/database'; import type { CollapsableNotifInfo } from '../fetchers/message-fetchers'; import { fetchCollapsableNotifs } from '../fetchers/message-fetchers'; import { fetchServerThreadInfos } from '../fetchers/thread-fetchers'; import { fetchUserInfos } from '../fetchers/user-fetchers'; import type { Viewer } from '../session/viewer'; import { getAPNsNotificationTopic } from './providers'; import { apnPush, fcmPush, getUnreadCounts, apnMaxNotificationPayloadByteSize, fcmMaxNotificationPayloadByteSize, } from './utils'; type Device = { +deviceType: DeviceType, +deviceToken: string, +codeVersion: ?number, }; type PushUserInfo = { +devices: Device[], +messageInfos: RawMessageInfo[], }; type Delivery = IOSDelivery | AndroidDelivery | { collapsedInto: string }; type NotificationRow = { +dbID: string, +userID: string, +threadID?: ?string, +messageID?: ?string, +collapseKey?: ?string, +deliveries: Delivery[], }; export type PushInfo = { [userID: string]: PushUserInfo }; async function sendPushNotifs(pushInfo: PushInfo) { if (Object.keys(pushInfo).length === 0) { return; } const [ unreadCounts, { usersToCollapsableNotifInfo, serverThreadInfos, userInfos }, dbIDs, ] = await Promise.all([ getUnreadCounts(Object.keys(pushInfo)), fetchInfos(pushInfo), createDBIDs(pushInfo), ]); const deliveryPromises = []; const notifications: Map = new Map(); for (const userID in usersToCollapsableNotifInfo) { const threadInfos = _flow( _mapValues((serverThreadInfo: ServerThreadInfo) => { const rawThreadInfo = rawThreadInfoFromServerThreadInfo( serverThreadInfo, userID, ); if (!rawThreadInfo) { return null; } return threadInfoFromRawThreadInfo(rawThreadInfo, userID, userInfos); }), _pickBy(threadInfo => threadInfo), )(serverThreadInfos); for (const notifInfo of usersToCollapsableNotifInfo[userID]) { const hydrateMessageInfo = (rawMessageInfo: RawMessageInfo) => createMessageInfo(rawMessageInfo, userID, userInfos, threadInfos); const newMessageInfos = []; const newRawMessageInfos = []; for (const newRawMessageInfo of notifInfo.newMessageInfos) { const newMessageInfo = hydrateMessageInfo(newRawMessageInfo); if (newMessageInfo) { newMessageInfos.push(newMessageInfo); newRawMessageInfos.push(newRawMessageInfo); } } if (newMessageInfos.length === 0) { continue; } const existingMessageInfos = notifInfo.existingMessageInfos .map(hydrateMessageInfo) .filter(Boolean); const allMessageInfos = sortMessageInfoList([ ...newMessageInfos, ...existingMessageInfos, ]); const [ firstNewMessageInfo, ...remainingNewMessageInfos ] = newMessageInfos; const threadID = firstNewMessageInfo.threadID; const threadInfo = threadInfos[threadID]; const updateBadge = threadInfo.currentUser.subscription.home; const displayBanner = threadInfo.currentUser.subscription.pushNotifs; const username = userInfos[userID] && userInfos[userID].username; const userWasMentioned = username && threadInfo.currentUser.role && oldValidUsernameRegex.test(username) && firstNewMessageInfo.type === messageTypes.TEXT && new RegExp(`\\B@${username}\\b`, 'i').test(firstNewMessageInfo.text); if (!updateBadge && !displayBanner && !userWasMentioned) { continue; } const badgeOnly = !displayBanner && !userWasMentioned; const dbID = dbIDs.shift(); invariant(dbID, 'should have sufficient DB IDs'); const byDeviceType = getDevicesByDeviceType(pushInfo[userID].devices); const firstMessageID = firstNewMessageInfo.id; invariant(firstMessageID, 'RawMessageInfo.id should be set on server'); const notificationInfo = { source: 'new_message', dbID, userID, threadID, messageID: firstMessageID, collapseKey: notifInfo.collapseKey, }; const iosVersionsToTokens = byDeviceType.get('ios'); if (iosVersionsToTokens) { for (const [codeVer, deviceTokens] of iosVersionsToTokens) { const codeVersion = parseInt(codeVer, 10); // only for Flow const shimmedNewRawMessageInfos = shimUnsupportedRawMessageInfos( newRawMessageInfos, { platform: 'ios', codeVersion }, ); const notification = prepareIOSNotification( allMessageInfos, shimmedNewRawMessageInfos, threadInfo, notifInfo.collapseKey, badgeOnly, unreadCounts[userID], codeVersion, ); deliveryPromises.push( sendIOSNotification(notification, [...deviceTokens], { ...notificationInfo, codeVersion, }), ); } } const androidVersionsToTokens = byDeviceType.get('android'); if (androidVersionsToTokens) { for (const [codeVer, deviceTokens] of androidVersionsToTokens) { const codeVersion = parseInt(codeVer, 10); // only for Flow const shimmedNewRawMessageInfos = shimUnsupportedRawMessageInfos( newRawMessageInfos, { platform: 'android', codeVersion }, ); const notification = prepareAndroidNotification( allMessageInfos, shimmedNewRawMessageInfos, threadInfo, notifInfo.collapseKey, badgeOnly, unreadCounts[userID], dbID, codeVersion, ); deliveryPromises.push( sendAndroidNotification(notification, [...deviceTokens], { ...notificationInfo, codeVersion, }), ); } } for (const newMessageInfo of remainingNewMessageInfos) { const newDBID = dbIDs.shift(); invariant(newDBID, 'should have sufficient DB IDs'); const messageID = newMessageInfo.id; invariant(messageID, 'RawMessageInfo.id should be set on server'); notifications.set(newDBID, { dbID: newDBID, userID, threadID: newMessageInfo.threadID, messageID, collapseKey: notifInfo.collapseKey, deliveries: [{ collapsedInto: dbID }], }); } } } const cleanUpPromises = []; if (dbIDs.length > 0) { const query = SQL`DELETE FROM ids WHERE id IN (${dbIDs})`; cleanUpPromises.push(dbQuery(query)); } const [deliveryResults] = await Promise.all([ Promise.all(deliveryPromises), Promise.all(cleanUpPromises), ]); await saveNotifResults(deliveryResults, notifications, true); } // The results in deliveryResults will be combined with the rows // in rowsToSave and then written to the notifications table async function saveNotifResults( deliveryResults: $ReadOnlyArray, inputRowsToSave: Map, rescindable: boolean, ) { const rowsToSave = new Map(inputRowsToSave); const allInvalidTokens = []; for (const deliveryResult of deliveryResults) { const { info, delivery, invalidTokens } = deliveryResult; const { dbID, userID } = info; const curNotifRow = rowsToSave.get(dbID); if (curNotifRow) { curNotifRow.deliveries.push(delivery); } else { // Ternary expressions for Flow const threadID = info.threadID ? info.threadID : null; const messageID = info.messageID ? info.messageID : null; const collapseKey = info.collapseKey ? info.collapseKey : null; rowsToSave.set(dbID, { dbID, userID, threadID, messageID, collapseKey, deliveries: [delivery], }); } if (invalidTokens) { allInvalidTokens.push({ userID, tokens: invalidTokens, }); } } const notificationRows = []; for (const notification of rowsToSave.values()) { notificationRows.push([ notification.dbID, notification.userID, notification.threadID, notification.messageID, notification.collapseKey, JSON.stringify(notification.deliveries), Number(!rescindable), ]); } const dbPromises = []; if (allInvalidTokens.length > 0) { dbPromises.push(removeInvalidTokens(allInvalidTokens)); } if (notificationRows.length > 0) { const query = SQL` INSERT INTO notifications (id, user, thread, message, collapse_key, delivery, rescinded) VALUES ${notificationRows} `; dbPromises.push(dbQuery(query)); } if (dbPromises.length > 0) { await Promise.all(dbPromises); } } async function fetchInfos(pushInfo: PushInfo) { const usersToCollapsableNotifInfo = await fetchCollapsableNotifs(pushInfo); const threadIDs = new Set(); const threadWithChangedNamesToMessages = new Map(); const addThreadIDsFromMessageInfos = (rawMessageInfo: RawMessageInfo) => { const threadID = rawMessageInfo.threadID; threadIDs.add(threadID); const messageSpec = messageSpecs[rawMessageInfo.type]; if (messageSpec.threadIDs) { for (const id of messageSpec.threadIDs(rawMessageInfo)) { threadIDs.add(id); } } if ( rawMessageInfo.type === messageTypes.CHANGE_SETTINGS && rawMessageInfo.field === 'name' ) { const messages = threadWithChangedNamesToMessages.get(threadID); if (messages) { messages.push(rawMessageInfo.id); } else { threadWithChangedNamesToMessages.set(threadID, [rawMessageInfo.id]); } } }; for (const userID in usersToCollapsableNotifInfo) { for (const notifInfo of usersToCollapsableNotifInfo[userID]) { for (const rawMessageInfo of notifInfo.existingMessageInfos) { addThreadIDsFromMessageInfos(rawMessageInfo); } for (const rawMessageInfo of notifInfo.newMessageInfos) { addThreadIDsFromMessageInfos(rawMessageInfo); } } } const promises = {}; // These threadInfos won't have currentUser set promises.threadResult = fetchServerThreadInfos( SQL`t.id IN (${[...threadIDs]})`, ); if (threadWithChangedNamesToMessages.size > 0) { const typesThatAffectName = [ messageTypes.CHANGE_SETTINGS, messageTypes.CREATE_THREAD, ]; const oldNameQuery = SQL` SELECT IF( JSON_TYPE(JSON_EXTRACT(m.content, "$.name")) = 'NULL', "", JSON_UNQUOTE(JSON_EXTRACT(m.content, "$.name")) ) AS name, m.thread FROM ( SELECT MAX(id) AS id FROM messages WHERE type IN (${typesThatAffectName}) AND JSON_EXTRACT(content, "$.name") IS NOT NULL AND`; const threadClauses = []; for (const [threadID, messages] of threadWithChangedNamesToMessages) { threadClauses.push( SQL`(thread = ${threadID} AND id NOT IN (${messages}))`, ); } oldNameQuery.append(mergeOrConditions(threadClauses)); oldNameQuery.append(SQL` GROUP BY thread ) x LEFT JOIN messages m ON m.id = x.id `); promises.oldNames = dbQuery(oldNameQuery); } const { threadResult, oldNames } = await promiseAll(promises); const serverThreadInfos = threadResult.threadInfos; if (oldNames) { const [result] = oldNames; for (const row of result) { const threadID = row.thread.toString(); serverThreadInfos[threadID].name = row.name; } } const userInfos = await fetchNotifUserInfos( serverThreadInfos, usersToCollapsableNotifInfo, ); return { usersToCollapsableNotifInfo, serverThreadInfos, userInfos }; } async function fetchNotifUserInfos( serverThreadInfos: { +[threadID: string]: ServerThreadInfo }, usersToCollapsableNotifInfo: { +[userID: string]: CollapsableNotifInfo[] }, ) { const missingUserIDs = new Set(); for (const threadID in serverThreadInfos) { const serverThreadInfo = serverThreadInfos[threadID]; for (const member of serverThreadInfo.members) { missingUserIDs.add(member.id); } } const addUserIDsFromMessageInfos = (rawMessageInfo: RawMessageInfo) => { missingUserIDs.add(rawMessageInfo.creatorID); const userIDs = messageSpecs[rawMessageInfo.type].userIDs?.(rawMessageInfo) ?? []; for (const userID of userIDs) { missingUserIDs.add(userID); } }; for (const userID in usersToCollapsableNotifInfo) { for (const notifInfo of usersToCollapsableNotifInfo[userID]) { for (const rawMessageInfo of notifInfo.existingMessageInfos) { addUserIDsFromMessageInfos(rawMessageInfo); } for (const rawMessageInfo of notifInfo.newMessageInfos) { addUserIDsFromMessageInfos(rawMessageInfo); } } } return await fetchUserInfos([...missingUserIDs]); } async function createDBIDs(pushInfo: PushInfo): Promise { let numIDsNeeded = 0; for (const userID in pushInfo) { numIDsNeeded += pushInfo[userID].messageInfos.length; } return await createIDs('notifications', numIDsNeeded); } function getDevicesByDeviceType( devices: Device[], ): Map>> { const byDeviceType = new Map(); for (const device of devices) { let innerMap = byDeviceType.get(device.deviceType); if (!innerMap) { innerMap = new Map(); byDeviceType.set(device.deviceType, innerMap); } const codeVersion: number = device.codeVersion !== null && device.codeVersion !== undefined ? device.codeVersion : -1; let innerMostSet = innerMap.get(codeVersion); if (!innerMostSet) { innerMostSet = new Set(); innerMap.set(codeVersion, innerMostSet); } innerMostSet.add(device.deviceToken); } return byDeviceType; } function prepareIOSNotification( allMessageInfos: MessageInfo[], newRawMessageInfos: RawMessageInfo[], threadInfo: ThreadInfo, collapseKey: ?string, badgeOnly: boolean, unreadCount: number, codeVersion: number, ): apn.Notification { const uniqueID = uuidv4(); const notification = new apn.Notification(); notification.topic = getAPNsNotificationTopic(codeVersion); const { merged, ...rest } = notifTextsForMessageInfo( allMessageInfos, threadInfo, ); if (!badgeOnly) { notification.body = merged; notification.sound = 'default'; } notification.payload = { ...notification.payload, ...rest, }; notification.badge = unreadCount; notification.threadId = threadInfo.id; notification.id = uniqueID; notification.pushType = 'alert'; notification.payload.id = uniqueID; notification.payload.threadID = threadInfo.id; if (codeVersion > 1000) { notification.mutableContent = true; } if (collapseKey) { notification.collapseId = collapseKey; } const messageInfos = JSON.stringify(newRawMessageInfos); // We make a copy before checking notification's length, because calling // length compiles the notification and makes it immutable. Further // changes to its properties won't be reflected in the final plaintext // data that is sent. const copyWithMessageInfos = _cloneDeep(notification); copyWithMessageInfos.payload = { ...copyWithMessageInfos.payload, messageInfos, }; if (copyWithMessageInfos.length() <= apnMaxNotificationPayloadByteSize) { notification.payload.messageInfos = messageInfos; return notification; } const notificationCopy = _cloneDeep(notification); if (notificationCopy.length() > apnMaxNotificationPayloadByteSize) { console.warn( `iOS notification ${uniqueID} exceeds size limit, even with messageInfos omitted`, ); } return notification; } function prepareAndroidNotification( allMessageInfos: MessageInfo[], newRawMessageInfos: RawMessageInfo[], threadInfo: ThreadInfo, collapseKey: ?string, badgeOnly: boolean, unreadCount: number, dbID: string, codeVersion: number, ): Object { const notifID = collapseKey ? collapseKey : dbID; const { merged, ...rest } = notifTextsForMessageInfo( allMessageInfos, threadInfo, ); const notification = { data: { badge: unreadCount.toString(), ...rest, threadID: threadInfo.id, }, }; // The reason we only include `badgeOnly` for newer clients is because older // clients don't know how to parse it. The reason we only include `id` for // newer clients is that if the older clients see that field, they assume // the notif has a full payload, and then crash when trying to parse it. // By skipping `id` we allow old clients to still handle in-app notifs and // badge updating. if (!badgeOnly || codeVersion >= 69) { notification.data = { ...notification.data, id: notifID, badgeOnly: badgeOnly ? '1' : '0', }; } const messageInfos = JSON.stringify(newRawMessageInfos); const copyWithMessageInfos = { ...notification, data: { ...notification.data, messageInfos }, }; if ( Buffer.byteLength(JSON.stringify(copyWithMessageInfos)) <= fcmMaxNotificationPayloadByteSize ) { return copyWithMessageInfos; } if ( Buffer.byteLength(JSON.stringify(notification)) > fcmMaxNotificationPayloadByteSize ) { console.warn( `Android notification ${notifID} exceeds size limit, even with messageInfos omitted`, ); } return notification; } type NotificationInfo = | { +source: 'new_message', +dbID: string, +userID: string, +threadID: string, +messageID: string, +collapseKey: ?string, +codeVersion: number, } | { +source: 'mark_as_unread' | 'mark_as_read' | 'activity_update', +dbID: string, +userID: string, +codeVersion: number, }; type IOSDelivery = { source: $PropertyType, deviceType: 'ios', iosID: string, deviceTokens: $ReadOnlyArray, codeVersion: number, errors?: $ReadOnlyArray, }; type IOSResult = { info: NotificationInfo, delivery: IOSDelivery, invalidTokens?: $ReadOnlyArray, }; async function sendIOSNotification( notification: apn.Notification, deviceTokens: $ReadOnlyArray, notificationInfo: NotificationInfo, ): Promise { const { source, codeVersion } = notificationInfo; const response = await apnPush({ notification, deviceTokens, codeVersion }); const delivery: IOSDelivery = { source, deviceType: 'ios', iosID: notification.id, deviceTokens, codeVersion, }; if (response.errors) { delivery.errors = response.errors; } const result: IOSResult = { info: notificationInfo, delivery, }; if (response.invalidTokens) { result.invalidTokens = response.invalidTokens; } return result; } type AndroidDelivery = { source: $PropertyType, deviceType: 'android', androidIDs: $ReadOnlyArray, deviceTokens: $ReadOnlyArray, codeVersion: number, errors?: $ReadOnlyArray, }; type AndroidResult = { info: NotificationInfo, delivery: AndroidDelivery, invalidTokens?: $ReadOnlyArray, }; async function sendAndroidNotification( notification: Object, deviceTokens: $ReadOnlyArray, notificationInfo: NotificationInfo, ): Promise { const collapseKey = notificationInfo.collapseKey ? notificationInfo.collapseKey : null; // for Flow... const { source, codeVersion } = notificationInfo; const response = await fcmPush({ notification, deviceTokens, collapseKey, codeVersion, }); const androidIDs = response.fcmIDs ? response.fcmIDs : []; const delivery: AndroidDelivery = { source, deviceType: 'android', androidIDs, deviceTokens, codeVersion, }; if (response.errors) { delivery.errors = response.errors; } const result: AndroidResult = { info: notificationInfo, delivery, }; if (response.invalidTokens) { result.invalidTokens = response.invalidTokens; } return result; } type InvalidToken = { +userID: string, +tokens: $ReadOnlyArray, }; async function removeInvalidTokens( invalidTokens: $ReadOnlyArray, ): Promise { const sqlTuples = invalidTokens.map( invalidTokenUser => SQL`( user = ${invalidTokenUser.userID} AND device_token IN (${invalidTokenUser.tokens}) )`, ); const sqlCondition = mergeOrConditions(sqlTuples); const selectQuery = SQL` SELECT id, user, device_token FROM cookies WHERE `; selectQuery.append(sqlCondition); const [result] = await dbQuery(selectQuery); const userCookiePairsToInvalidDeviceTokens = new Map(); for (const row of result) { const userCookiePair = `${row.user}|${row.id}`; const existing = userCookiePairsToInvalidDeviceTokens.get(userCookiePair); if (existing) { existing.add(row.device_token); } else { userCookiePairsToInvalidDeviceTokens.set( userCookiePair, new Set([row.device_token]), ); } } const time = Date.now(); const promises = []; for (const entry of userCookiePairsToInvalidDeviceTokens) { const [userCookiePair, deviceTokens] = entry; const [userID, cookieID] = userCookiePair.split('|'); const updateDatas = [...deviceTokens].map(deviceToken => ({ type: updateTypes.BAD_DEVICE_TOKEN, userID, time, deviceToken, targetCookie: cookieID, })); promises.push(createUpdates(updateDatas)); } const updateQuery = SQL` UPDATE cookies SET device_token = NULL WHERE `; updateQuery.append(sqlCondition); promises.push(dbQuery(updateQuery)); await Promise.all(promises); } async function updateBadgeCount( viewer: Viewer, source: 'mark_as_unread' | 'mark_as_read' | 'activity_update', - excludeDeviceTokens: $ReadOnlyArray, ) { const { userID } = viewer; const deviceTokenQuery = SQL` SELECT platform, device_token, versions FROM cookies WHERE user = ${userID} AND device_token IS NOT NULL `; if (viewer.data.cookieID) { deviceTokenQuery.append(SQL`AND id != ${viewer.cookieID} `); } - if (excludeDeviceTokens.length > 0) { - deviceTokenQuery.append( - SQL`AND device_token NOT IN (${excludeDeviceTokens}) `, - ); - } const [unreadCounts, [deviceTokenResult], [dbID]] = await Promise.all([ getUnreadCounts([userID]), dbQuery(deviceTokenQuery), createIDs('notifications', 1), ]); const unreadCount = unreadCounts[userID]; const devices = deviceTokenResult.map(row => ({ deviceType: row.platform, deviceToken: row.device_token, codeVersion: JSON.parse(row.versions)?.codeVersion, })); const byDeviceType = getDevicesByDeviceType(devices); const deliveryPromises = []; const iosVersionsToTokens = byDeviceType.get('ios'); if (iosVersionsToTokens) { for (const [codeVer, deviceTokens] of iosVersionsToTokens) { const codeVersion = parseInt(codeVer, 10); // only for Flow const notification = new apn.Notification(); notification.topic = getAPNsNotificationTopic(codeVersion); notification.badge = unreadCount; notification.pushType = 'alert'; deliveryPromises.push( sendIOSNotification(notification, [...deviceTokens], { source, dbID, userID, codeVersion, }), ); } } const androidVersionsToTokens = byDeviceType.get('android'); if (androidVersionsToTokens) { for (const [codeVer, deviceTokens] of androidVersionsToTokens) { const codeVersion = parseInt(codeVer, 10); // only for Flow const notificationData = codeVersion < 69 ? { badge: unreadCount.toString() } : { badge: unreadCount.toString(), badgeOnly: '1' }; const notification = { data: notificationData }; deliveryPromises.push( sendAndroidNotification(notification, [...deviceTokens], { source, dbID, userID, codeVersion, }), ); } } const deliveryResults = await Promise.all(deliveryPromises); await saveNotifResults(deliveryResults, new Map(), false); } export { sendPushNotifs, updateBadgeCount }; diff --git a/keyserver/src/updaters/activity-updaters.js b/keyserver/src/updaters/activity-updaters.js index dd422135e..152f9a918 100644 --- a/keyserver/src/updaters/activity-updaters.js +++ b/keyserver/src/updaters/activity-updaters.js @@ -1,518 +1,518 @@ // @flow import invariant from 'invariant'; import _difference from 'lodash/fp/difference'; import _max from 'lodash/fp/max'; import { localIDPrefix } from 'lib/shared/message-utils'; import type { UpdateActivityResult, UpdateActivityRequest, SetThreadUnreadStatusRequest, SetThreadUnreadStatusResult, } from 'lib/types/activity-types'; import { messageTypes } from 'lib/types/message-types'; import { threadPermissions } from 'lib/types/thread-types'; import { updateTypes } from 'lib/types/update-types'; import { ServerError } from 'lib/utils/errors'; import { createUpdates } from '../creators/update-creator'; import { dbQuery, SQL, mergeOrConditions } from '../database/database'; import type { SQLStatementType } from '../database/types'; import { deleteActivityForViewerSession } from '../deleters/activity-deleters'; import { checkThread, getValidThreads, } from '../fetchers/thread-permission-fetchers'; import { rescindPushNotifs } from '../push/rescind'; import { updateBadgeCount } from '../push/send'; import type { Viewer } from '../session/viewer'; import { earliestFocusedTimeConsideredExpired } from '../shared/focused-times'; type PartialThreadStatus = { +focusActive: boolean, +threadID: string, +newLastReadMessage: ?number, }; type ThreadStatus = | { +focusActive: true, +threadID: string, +newLastReadMessage: number, +curLastReadMessage: number, +rescindCondition: SQLStatementType, } | { +focusActive: false, +threadID: string, +newLastReadMessage: ?number, +curLastReadMessage: number, +rescindCondition: ?SQLStatementType, +newerMessageFromOtherAuthor: boolean, }; async function activityUpdater( viewer: Viewer, request: UpdateActivityRequest, ): Promise { if (!viewer.loggedIn) { throw new ServerError('not_logged_in'); } const focusUpdatesByThreadID = new Map(); for (const activityUpdate of request.updates) { const threadID = activityUpdate.threadID; const updatesForThreadID = focusUpdatesByThreadID.get(threadID) ?? []; if (!focusUpdatesByThreadID.has(threadID)) { focusUpdatesByThreadID.set(threadID, updatesForThreadID); } updatesForThreadID.push(activityUpdate); } const unverifiedThreadIDs: $ReadOnlySet = new Set( request.updates.map(update => update.threadID), ); const verifiedThreadsData = await getValidThreads( viewer, [...unverifiedThreadIDs], [ { check: 'permission', permission: threadPermissions.VISIBLE, }, ], ); if (verifiedThreadsData.length === 0) { return { unfocusedToUnread: [] }; } const memberThreadIDs = new Set(); const verifiedThreadIDs = []; for (const threadData of verifiedThreadsData) { if (threadData.role > 0) { memberThreadIDs.add(threadData.threadID); } verifiedThreadIDs.push(threadData.threadID); } const partialThreadStatuses: PartialThreadStatus[] = []; for (const threadID of verifiedThreadIDs) { const focusUpdates = focusUpdatesByThreadID.get(threadID); invariant(focusUpdates, `no focusUpdate for thread ID ${threadID}`); const focusActive = !focusUpdates.some(update => !update.focus); const newLastReadMessage = _max( focusUpdates .filter( update => update.latestMessage && !update.latestMessage.startsWith(localIDPrefix), ) .map(update => parseInt(update.latestMessage)), ); partialThreadStatuses.push({ threadID, focusActive, newLastReadMessage, }); } // We update the focused rows before we check for new messages so we can // guarantee that any messages that may have set the thread to unread before // we set it to focused are caught and overriden await updateFocusedRows(viewer, partialThreadStatuses); if (memberThreadIDs.size === 0) { return { unfocusedToUnread: [] }; } const memberPartialThreadStatuses = partialThreadStatuses.filter( partialStatus => memberThreadIDs.has(partialStatus.threadID), ); const unfocusedLatestMessages = new Map(); for (const partialThreadStatus of memberPartialThreadStatuses) { const { threadID, focusActive, newLastReadMessage } = partialThreadStatus; if (!focusActive) { unfocusedLatestMessages.set(threadID, newLastReadMessage ?? 0); } } const [ unfocusedThreadsWithNewerMessages, lastMessageInfos, ] = await Promise.all([ checkForNewerMessages(viewer, unfocusedLatestMessages), fetchLastMessageInfo(viewer, [...memberThreadIDs]), ]); const threadStatuses: ThreadStatus[] = []; for (const partialThreadStatus of memberPartialThreadStatuses) { const { threadID, focusActive, newLastReadMessage } = partialThreadStatus; const lastMessageInfo = lastMessageInfos.get(threadID); invariant( lastMessageInfo !== undefined, `no lastMessageInfo for thread ID ${threadID}`, ); const { lastMessage, lastReadMessage: curLastReadMessage, } = lastMessageInfo; if (focusActive) { threadStatuses.push({ focusActive: true, threadID, newLastReadMessage: newLastReadMessage ? Math.max(lastMessage, newLastReadMessage) : lastMessage, curLastReadMessage, rescindCondition: SQL`n.thread = ${threadID}`, }); } else { threadStatuses.push({ focusActive: false, threadID, newLastReadMessage, curLastReadMessage, rescindCondition: newLastReadMessage ? SQL`(n.thread = ${threadID} AND n.message <= ${newLastReadMessage})` : null, newerMessageFromOtherAuthor: unfocusedThreadsWithNewerMessages.has( threadID, ), }); } } // The following block determines whether to enqueue updates for a given // (user, thread) pair and whether to propagate badge count notifs to all of // that user's devices const setUnread: Array<{ +threadID: string, +unread: boolean }> = []; for (const threadStatus of threadStatuses) { const { threadID, curLastReadMessage } = threadStatus; if (!threadStatus.focusActive) { const { newLastReadMessage, newerMessageFromOtherAuthor } = threadStatus; if (newerMessageFromOtherAuthor) { setUnread.push({ threadID, unread: true }); } else if (!newLastReadMessage) { // This is a rare edge case. It should only be possible for threads that // have zero messages on both the client and server, which shouldn't // happen. In this case we'll set the thread to read, just in case... console.warn(`thread ID ${threadID} appears to have no messages`); setUnread.push({ threadID, unread: false }); } else if (newLastReadMessage > curLastReadMessage) { setUnread.push({ threadID, unread: false }); } } else { const { newLastReadMessage } = threadStatus; if (newLastReadMessage > curLastReadMessage) { setUnread.push({ threadID, unread: false }); } } } const time = Date.now(); const updateDatas = setUnread.map(({ threadID, unread }) => ({ type: updateTypes.UPDATE_THREAD_READ_STATUS, userID: viewer.userID, time, threadID, unread, })); const latestMessages = new Map(); for (const threadStatus of threadStatuses) { const { threadID, newLastReadMessage, curLastReadMessage } = threadStatus; if (newLastReadMessage && newLastReadMessage > curLastReadMessage) { latestMessages.set(threadID, newLastReadMessage); } } await Promise.all([ updateLastReadMessage(viewer, latestMessages), createUpdates(updateDatas, { viewer, updatesForCurrentSession: 'ignore' }), ]); // We do this afterwards so the badge count is correct const rescindConditions = threadStatuses .map(({ rescindCondition }) => rescindCondition) .filter(Boolean); let rescindCondition; if (rescindConditions.length > 0) { rescindCondition = SQL`n.user = ${viewer.userID} AND `; rescindCondition.append(mergeOrConditions(rescindConditions)); } await rescindAndUpdateBadgeCounts( viewer, rescindCondition, updateDatas.length > 0 ? 'activity_update' : null, ); return { unfocusedToUnread: [...unfocusedThreadsWithNewerMessages] }; } async function updateFocusedRows( viewer: Viewer, partialThreadStatuses: $ReadOnlyArray, ): Promise { const threadIDs = partialThreadStatuses .filter(threadStatus => threadStatus.focusActive) .map(({ threadID }) => threadID); const time = Date.now(); if (threadIDs.length > 0) { const focusedInsertRows = threadIDs.map(threadID => [ viewer.userID, viewer.session, threadID, time, ]); const query = SQL` INSERT INTO focused (user, session, thread, time) VALUES ${focusedInsertRows} ON DUPLICATE KEY UPDATE time = VALUE(time) `; await dbQuery(query); } if (viewer.hasSessionInfo) { await deleteActivityForViewerSession(viewer, time); } } // To protect against a possible race condition, we reset the thread to unread // if the latest message ID on the client at the time that focus was dropped // is no longer the latest message ID. // Returns the set of unfocused threads that should be set to unread on // the client because a new message arrived since they were unfocused. async function checkForNewerMessages( viewer: Viewer, latestMessages: Map, ): Promise> { if (latestMessages.size === 0 || !viewer.loggedIn) { return new Set(); } const unfocusedThreadIDs = [...latestMessages.keys()]; const focusedElsewhereThreadIDs = await checkThreadsFocused( viewer, unfocusedThreadIDs, ); const unreadCandidates = _difference(unfocusedThreadIDs)( focusedElsewhereThreadIDs, ); if (unreadCandidates.length === 0) { return new Set(); } const knowOfExtractString = `$.${threadPermissions.KNOW_OF}.value`; const query = SQL` SELECT m.thread, MAX(m.id) AS latest_message FROM messages m LEFT JOIN memberships stm ON m.type = ${messageTypes.CREATE_SUB_THREAD} AND stm.thread = m.content AND stm.user = ${viewer.userID} WHERE m.thread IN (${unreadCandidates}) AND m.user != ${viewer.userID} AND ( m.type != ${messageTypes.CREATE_SUB_THREAD} OR JSON_EXTRACT(stm.permissions, ${knowOfExtractString}) IS TRUE ) GROUP BY m.thread `; const [result] = await dbQuery(query); const threadsWithNewerMessages = new Set(); for (const row of result) { const threadID = row.thread.toString(); const serverLatestMessage = row.latest_message; const clientLatestMessage = latestMessages.get(threadID); if (clientLatestMessage < serverLatestMessage) { threadsWithNewerMessages.add(threadID); } } return threadsWithNewerMessages; } async function checkThreadsFocused( viewer: Viewer, threadIDs: $ReadOnlyArray, ): Promise { const time = earliestFocusedTimeConsideredExpired(); const query = SQL` SELECT thread FROM focused WHERE time > ${time} AND user = ${viewer.userID} AND thread IN (${threadIDs}) GROUP BY thread `; const [result] = await dbQuery(query); const focusedThreadIDs = []; for (const row of result) { focusedThreadIDs.push(row.thread.toString()); } return focusedThreadIDs; } async function updateLastReadMessage( viewer: Viewer, lastReadMessages: $ReadOnlyMap, ) { if (lastReadMessages.size === 0) { return; } const query = SQL` UPDATE memberships SET last_read_message = GREATEST(last_read_message, CASE `; lastReadMessages.forEach((lastMessage, threadID) => { query.append(SQL` WHEN thread = ${threadID} THEN ${lastMessage} `); }); query.append(SQL` ELSE last_read_message END) WHERE thread IN (${[...lastReadMessages.keys()]}) AND user = ${viewer.userID} `); return await dbQuery(query); } async function setThreadUnreadStatus( viewer: Viewer, request: SetThreadUnreadStatusRequest, ): Promise { if (!viewer.loggedIn) { throw new ServerError('not_logged_in'); } const isMemberAndCanViewThread = await checkThread(viewer, request.threadID, [ { check: 'is_member', }, { check: 'permission', permission: threadPermissions.VISIBLE, }, ]); if (!isMemberAndCanViewThread) { throw new ServerError('invalid_parameters'); } const resetThreadToUnread = await shouldResetThreadToUnread(viewer, request); if (!resetThreadToUnread) { const lastReadMessage = request.unread ? SQL`0` : SQL`GREATEST(m.last_read_message, ${request.latestMessage ?? 0})`; const update = SQL` UPDATE memberships m SET m.last_read_message = `; update.append(lastReadMessage); update.append(SQL` WHERE m.thread = ${request.threadID} AND m.user = ${viewer.userID} `); const queryPromise = dbQuery(update); const time = Date.now(); const updatesPromise = createUpdates( [ { type: updateTypes.UPDATE_THREAD_READ_STATUS, userID: viewer.userID, time: time, threadID: request.threadID, unread: request.unread, }, ], { viewer, updatesForCurrentSession: 'ignore' }, ); await Promise.all([updatesPromise, queryPromise]); } let rescindCondition; if (!request.unread) { rescindCondition = SQL` n.user = ${viewer.userID} AND n.thread = ${request.threadID} AND n.message <= ${request.latestMessage} `; } await rescindAndUpdateBadgeCounts( viewer, rescindCondition, request.unread ? 'mark_as_unread' : 'mark_as_read', ); return { resetToUnread: resetThreadToUnread, }; } async function rescindAndUpdateBadgeCounts( viewer: Viewer, rescindCondition: ?SQLStatementType, badgeCountUpdateSource: ?( | 'activity_update' | 'mark_as_unread' | 'mark_as_read' ), ) { - const excludeDeviceTokens = []; + const notificationPromises = []; if (rescindCondition) { - const handledDeviceTokens = await rescindPushNotifs(rescindCondition); - excludeDeviceTokens.push(...handledDeviceTokens); + notificationPromises.push(rescindPushNotifs(rescindCondition)); } if (badgeCountUpdateSource) { - await updateBadgeCount(viewer, badgeCountUpdateSource, excludeDeviceTokens); + notificationPromises.push(updateBadgeCount(viewer, badgeCountUpdateSource)); } + await Promise.all(notificationPromises); } async function shouldResetThreadToUnread( viewer: Viewer, request: SetThreadUnreadStatusRequest, ): Promise { if (request.unread) { return false; } const threadsWithNewerMessages = await checkForNewerMessages( viewer, new Map([[request.threadID, parseInt(request.latestMessage) || 0]]), ); return threadsWithNewerMessages.has(request.threadID); } type LastMessageInfo = { +lastMessage: number, +lastReadMessage: number, }; async function fetchLastMessageInfo( viewer: Viewer, threadIDs: $ReadOnlyArray, ) { const query = SQL` SELECT thread, last_message, last_read_message FROM memberships WHERE user = ${viewer.userID} AND thread IN (${threadIDs}) `; const [result] = await dbQuery(query); const lastMessages = new Map(); for (const row of result) { const threadID = row.thread.toString(); const lastMessage = row.last_message; const lastReadMessage = row.last_read_message; lastMessages.set(threadID, { lastMessage, lastReadMessage }); } return lastMessages; } export { activityUpdater, setThreadUnreadStatus };