yuuza/outline
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Notifications refactor (#5151

Browse Source 
* Ongoing

* refactor

* test

* Add cleanup task

* refactor
This commit is contained in:
Tom Moor
2023-04-08 09:22:49 -04:00
committed by GitHub
parent c97110e72b
commit 9c9ceef8ee
28 changed files with 1122 additions and 901 deletions
Download Patch File Download Diff File Expand all files Collapse all files

51
server/queues/processors/EmailsProcessor.ts Normal file
View File

@@ -0,0 +1,51 @@
import { NotificationEventType } from "@shared/types";
import CollectionCreatedEmail from "@server/emails/templates/CollectionCreatedEmail";
import CommentCreatedEmail from "@server/emails/templates/CommentCreatedEmail";
import CommentMentionedEmail from "@server/emails/templates/CommentMentionedEmail";
import DocumentMentionedEmail from "@server/emails/templates/DocumentMentionedEmail";
import DocumentPublishedOrUpdatedEmail from "@server/emails/templates/DocumentPublishedOrUpdatedEmail";
import { Notification } from "@server/models";
import { Event, NotificationEvent } from "@server/types";
import BaseProcessor from "./BaseProcessor";
export default class NotificationsProcessor extends BaseProcessor {
static applicableEvents: Event["name"][] = ["notifications.create"];
async perform(event: NotificationEvent) {
const notification = await Notification.scope([
"withTeam",
"withUser",
"withActor",
]).findByPk(event.modelId);
if (!notification) {
return;
}
switch (notification.event) {
case NotificationEventType.UpdateDocument:
case NotificationEventType.PublishDocument: {
await new DocumentPublishedOrUpdatedEmail(notification).schedule();
return;
}
case NotificationEventType.MentionedInDocument: {
await new DocumentMentionedEmail(notification).schedule();
return;
}
case NotificationEventType.MentionedInComment: {
await new CommentMentionedEmail(notification).schedule();
return;
}
case NotificationEventType.CreateCollection: {
await new CollectionCreatedEmail(notification).schedule();
return;
}
case NotificationEventType.CreateComment: {
await new CommentCreatedEmail(notification).schedule();
}
}
}
}

346
server/queues/processors/NotificationsProcessor.ts
View File

@@ -1,26 +1,4 @@
import { subHours } from "date-fns";
import { differenceBy } from "lodash";
import { Op } from "sequelize";
import { NotificationEventType } from "@shared/types";
import { Minute } from "@shared/utils/time";
import subscriptionCreator from "@server/commands/subscriptionCreator";
import { sequelize } from "@server/database/sequelize";
import CollectionCreatedEmail from "@server/emails/templates/CollectionCreatedEmail";
import DocumentMentionedEmail from "@server/emails/templates/DocumentMentionedEmail";
import DocumentPublishedOrUpdatedEmail from "@server/emails/templates/DocumentPublishedOrUpdatedEmail";
import env from "@server/env";
import Logger from "@server/logging/Logger";
import {
Document,
Team,
Collection,
Notification,
Revision,
User,
View,
} from "@server/models";
import DocumentHelper from "@server/models/helpers/DocumentHelper";
import NotificationHelper from "@server/models/helpers/NotificationHelper";
import {
CollectionEvent,
RevisionEvent,
@@ -28,8 +6,11 @@ import {
DocumentEvent,
CommentEvent,
} from "@server/types";
import CommentCreatedNotificationTask from "../tasks/CommentCreatedNotificationTask";
import CommentUpdatedNotificationTask from "../tasks/CommentUpdatedNotificationTask";
import CollectionCreatedNotificationsTask from "../tasks/CollectionCreatedNotificationsTask";
import CommentCreatedNotificationsTask from "../tasks/CommentCreatedNotificationsTask";
import CommentUpdatedNotificationsTask from "../tasks/CommentUpdatedNotificationsTask";
import DocumentPublishedNotificationsTask from "../tasks/DocumentPublishedNotificationsTask";
import RevisionCreatedNotificationsTask from "../tasks/RevisionCreatedNotificationsTask";
import BaseProcessor from "./BaseProcessor";
export default class NotificationsProcessor extends BaseProcessor {
@@ -57,20 +38,8 @@ export default class NotificationsProcessor extends BaseProcessor {
}
}
async commentCreated(event: CommentEvent) {
await CommentCreatedNotificationTask.schedule(event, {
delay: Minute,
});
}
async commentUpdated(event: CommentEvent) {
await CommentUpdatedNotificationTask.schedule(event, {
delay: Minute,
});
}
async documentPublished(event: DocumentEvent) {
// never send notifications when batch importing documents
// never send notifications when batch importing
if (
"data" in event &&
"source" in event.data &&
@@ -79,304 +48,35 @@ export default class NotificationsProcessor extends BaseProcessor {
return;
}
const [collection, document, team] = await Promise.all([
Collection.findByPk(event.collectionId),
Document.findByPk(event.documentId, { includeState: true }),
Team.findByPk(event.teamId),
]);
if (!document || !team || !collection) {
return;
}
await this.createDocumentSubscriptions(document, event);
// Send notifications to mentioned users first
const mentions = DocumentHelper.parseMentions(document);
const userIdsSentNotifications: string[] = [];
for (const mention of mentions) {
const [recipient, actor] = await Promise.all([
User.findByPk(mention.modelId),
User.findByPk(mention.actorId),
]);
if (
recipient &&
actor &&
recipient.id !== actor.id &&
recipient.subscribedToEventType(NotificationEventType.Mentioned)
) {
const notification = await Notification.create({
event: event.name,
userId: recipient.id,
actorId: document.updatedBy.id,
teamId: team.id,
documentId: document.id,
});
userIdsSentNotifications.push(recipient.id);
await new DocumentMentionedEmail(
{
to: recipient.email,
documentId: event.documentId,
actorName: actor.name,
teamUrl: team.url,
mentionId: mention.id,
},
{ notificationId: notification.id }
).schedule();
}
}
const recipients = (
await NotificationHelper.getDocumentNotificationRecipients(
document,
NotificationEventType.PublishDocument,
document.lastModifiedById,
false
)
).filter((recipient) => !userIdsSentNotifications.includes(recipient.id));
for (const recipient of recipients) {
const notify = await this.shouldNotify(document, recipient);
if (notify) {
const notification = await Notification.create({
event: event.name,
userId: recipient.id,
actorId: document.updatedBy.id,
teamId: team.id,
documentId: document.id,
});
await new DocumentPublishedOrUpdatedEmail(
{
to: recipient.email,
userId: recipient.id,
eventType: NotificationEventType.PublishDocument,
documentId: document.id,
teamUrl: team.url,
actorName: document.updatedBy.name,
collectionName: collection.name,
},
{ notificationId: notification.id }
).schedule();
}
}
await DocumentPublishedNotificationsTask.schedule(event);
}
async revisionCreated(event: RevisionEvent) {
const [collection, document, revision, team] = await Promise.all([
Collection.findByPk(event.collectionId),
Document.findByPk(event.documentId, { includeState: true }),
Revision.findByPk(event.modelId),
Team.findByPk(event.teamId),
]);
if (!document || !team || !revision || !collection) {
return;
}
await this.createDocumentSubscriptions(document, event);
// Send notifications to mentioned users first
const prev = await revision.previous();
const oldMentions = prev ? DocumentHelper.parseMentions(prev) : [];
const newMentions = DocumentHelper.parseMentions(document);
const mentions = differenceBy(newMentions, oldMentions, "id");
const userIdsSentNotifications: string[] = [];
for (const mention of mentions) {
const [recipient, actor] = await Promise.all([
User.findByPk(mention.modelId),
User.findByPk(mention.actorId),
]);
if (
recipient &&
actor &&
recipient.id !== actor.id &&
recipient.subscribedToEventType(NotificationEventType.Mentioned)
) {
const notification = await Notification.create({
event: event.name,
userId: recipient.id,
actorId: document.updatedBy.id,
teamId: team.id,
documentId: document.id,
});
userIdsSentNotifications.push(recipient.id);
await new DocumentMentionedEmail(
{
to: recipient.email,
documentId: event.documentId,
actorName: actor.name,
teamUrl: team.url,
mentionId: mention.id,
},
{ notificationId: notification.id }
).schedule();
}
}
const recipients = (
await NotificationHelper.getDocumentNotificationRecipients(
document,
NotificationEventType.UpdateDocument,
document.lastModifiedById,
true
)
).filter((recipient) => !userIdsSentNotifications.includes(recipient.id));
if (!recipients.length) {
return;
}
// generate the diff html for the email
const before = await revision.previous();
const content = await DocumentHelper.toEmailDiff(before, revision, {
includeTitle: false,
centered: false,
signedUrls: 86400 * 4,
});
if (!content) {
return;
}
for (const recipient of recipients) {
const notify = await this.shouldNotify(document, recipient);
if (notify) {
const notification = await Notification.create({
event: event.name,
userId: recipient.id,
actorId: document.updatedBy.id,
teamId: team.id,
documentId: document.id,
});
await new DocumentPublishedOrUpdatedEmail(
{
to: recipient.email,
userId: recipient.id,
eventType: NotificationEventType.UpdateDocument,
documentId: document.id,
teamUrl: team.url,
actorName: document.updatedBy.name,
collectionName: collection.name,
content,
},
{ notificationId: notification.id }
).schedule();
}
}
await RevisionCreatedNotificationsTask.schedule(event);
}
async collectionCreated(event: CollectionEvent) {
const collection = await Collection.scope("withUser").findByPk(
event.collectionId
);
if (!collection || !collection.permission) {
// never send notifications when batch importing
if (
"data" in event &&
"source" in event.data &&
event.data.source === "import"
) {
return;
}
const recipients = await NotificationHelper.getCollectionNotificationRecipients(
collection,
NotificationEventType.CreateCollection
);
for (const recipient of recipients) {
// Suppress notifications for suspended users
if (recipient.isSuspended || !recipient.email) {
continue;
}
await new CollectionCreatedEmail({
to: recipient.email,
userId: recipient.id,
collectionId: collection.id,
}).schedule();
}
await CollectionCreatedNotificationsTask.schedule(event);
}
private shouldNotify = async (
document: Document,
user: User
): Promise<boolean> => {
// Deliver only a single notification in a 12 hour window
const notification = await Notification.findOne({
order: [["createdAt", "DESC"]],
where: {
userId: user.id,
documentId: document.id,
emailedAt: {
[Op.not]: null,
[Op.gte]: subHours(new Date(), 12),
},
},
async commentCreated(event: CommentEvent) {
await CommentCreatedNotificationsTask.schedule(event, {
delay: Minute,
});
}
if (notification) {
if (env.ENVIRONMENT === "development") {
Logger.info(
"processor",
`would have suppressed notification to ${user.id}, but not in development`
);
} else {
Logger.info(
"processor",
`suppressing notification to ${user.id} as recently notified`
);
return false;
}
}
// If this recipient has viewed the document since the last update was made
// then we can avoid sending them a useless notification, yay.
const view = await View.findOne({
where: {
userId: user.id,
documentId: document.id,
updatedAt: {
[Op.gt]: document.updatedAt,
},
},
async commentUpdated(event: CommentEvent) {
await CommentUpdatedNotificationsTask.schedule(event, {
delay: Minute,
});
if (view) {
Logger.info(
"processor",
`suppressing notification to ${user.id} because update viewed`
);
return false;
}
return true;
};
/**
* Create any new subscriptions that might be missing for collaborators in the
* document on publish and revision creation. This does mean that there is a
* short period of time where the user is not subscribed after editing until a
* revision is created.
*
* @param document The document to create subscriptions for
* @param event The event that triggered the subscription creation
*/
private createDocumentSubscriptions = async (
document: Document,
event: DocumentEvent | RevisionEvent
): Promise<void> => {
await sequelize.transaction(async (transaction) => {
const users = await document.collaborators({ transaction });
for (const user of users) {
await subscriptionCreator({
user,
documentId: document.id,
event: "documents.update",
resubscribe: false,
transaction,
ip: event.ip,
});
}
});
};
}
}

52
server/queues/tasks/CleanupOldNotificationsTask.ts Normal file
View File

@@ -0,0 +1,52 @@
import { subMonths } from "date-fns";
import { Op } from "sequelize";
import Logger from "@server/logging/Logger";
import { Notification } from "@server/models";
import BaseTask, { TaskPriority, TaskSchedule } from "./BaseTask";
type Props = Record<string, never>;
export default class CleanupOldNotificationsTask extends BaseTask<Props> {
static cron = TaskSchedule.Daily;
public async perform() {
Logger.info("task", `Permanently destroying old notifications…`);
let count;
count = await Notification.destroy({
where: {
createdAt: {
[Op.lt]: subMonths(new Date(), 12),
},
},
});
Logger.info(
"task",
`Destroyed ${count} notifications older than 12 months…`
);
count = await Notification.destroy({
where: {
viewedAt: {
[Op.ne]: null,
},
createdAt: {
[Op.lt]: subMonths(new Date(), 6),
},
},
});
Logger.info(
"task",
`Destroyed ${count} viewed notifications older than 6 months…`
);
}
public get options() {
return {
attempts: 1,
priority: TaskPriority.Background,
};
}
}

44
server/queues/tasks/CollectionCreatedNotificationsTask.ts Normal file
View File

@@ -0,0 +1,44 @@
import { NotificationEventType } from "@shared/types";
import { Collection, Notification } from "@server/models";
import NotificationHelper from "@server/models/helpers/NotificationHelper";
import { CollectionEvent } from "@server/types";
import BaseTask, { TaskPriority } from "./BaseTask";
export default class CollectionCreatedNotificationsTask extends BaseTask<
CollectionEvent
> {
public async perform(event: CollectionEvent) {
const collection = await Collection.findByPk(event.collectionId);
// We only send notifications for collections visible to the entire team
if (!collection || !collection.permission) {
return;
}
const recipients = await NotificationHelper.getCollectionNotificationRecipients(
collection,
NotificationEventType.CreateCollection
);
for (const recipient of recipients) {
// Suppress notifications for suspended users
if (recipient.isSuspended || !recipient.email) {
continue;
}
await Notification.create({
event: NotificationEventType.CreateCollection,
userId: recipient.id,
collectionId: collection.id,
actorId: collection.createdById,
teamId: collection.teamId,
});
}
}
public get options() {
return {
priority: TaskPriority.Background,
};
}
}

141
server/queues/tasks/CommentCreatedNotificationTask.ts
View File

@@ -1,141 +0,0 @@
import { Node } from "prosemirror-model";
import { NotificationEventType } from "@shared/types";
import subscriptionCreator from "@server/commands/subscriptionCreator";
import { sequelize } from "@server/database/sequelize";
import { schema } from "@server/editor";
import CommentCreatedEmail from "@server/emails/templates/CommentCreatedEmail";
import CommentMentionedEmail from "@server/emails/templates/CommentMentionedEmail";
import { Comment, Document, Notification, Team, User } from "@server/models";
import DocumentHelper from "@server/models/helpers/DocumentHelper";
import NotificationHelper from "@server/models/helpers/NotificationHelper";
import ProsemirrorHelper from "@server/models/helpers/ProsemirrorHelper";
import { CommentEvent } from "@server/types";
import BaseTask, { TaskPriority } from "./BaseTask";
export default class CommentCreatedNotificationTask extends BaseTask<
CommentEvent
> {
public async perform(event: CommentEvent) {
const [document, comment, team] = await Promise.all([
Document.scope("withCollection").findOne({
where: {
id: event.documentId,
},
}),
Comment.findByPk(event.modelId),
Team.findByPk(event.teamId),
]);
if (!document || !comment || !team) {
return;
}
// Commenting on a doc automatically creates a subscription to the doc
// if they haven't previously had one.
await sequelize.transaction(async (transaction) => {
await subscriptionCreator({
user: comment.createdBy,
documentId: document.id,
event: "documents.update",
resubscribe: false,
transaction,
ip: event.ip,
});
});
let content = ProsemirrorHelper.toHTML(
Node.fromJSON(schema, comment.data),
{
centered: false,
}
);
if (!content) {
return;
}
content = await DocumentHelper.attachmentsToSignedUrls(
content,
event.teamId,
86400 * 4
);
const mentions = ProsemirrorHelper.parseMentions(
ProsemirrorHelper.toProsemirror(comment.data)
);
const userIdsSentNotifications: string[] = [];
for (const mention of mentions) {
const [recipient, actor] = await Promise.all([
User.findByPk(mention.modelId),
User.findByPk(mention.actorId),
]);
if (
recipient &&
actor &&
recipient.id !== actor.id &&
recipient.subscribedToEventType(NotificationEventType.Mentioned)
) {
const notification = await Notification.create({
event: event.name,
userId: recipient.id,
actorId: actor.id,
teamId: team.id,
documentId: document.id,
});
userIdsSentNotifications.push(recipient.id);
await new CommentMentionedEmail(
{
to: recipient.email,
userId: recipient.id,
documentId: document.id,
teamUrl: team.url,
actorName: comment.createdBy.name,
commentId: comment.id,
content,
collectionName: document.collection?.name,
},
{ notificationId: notification.id }
).schedule();
}
}
const recipients = (
await NotificationHelper.getCommentNotificationRecipients(
document,
comment,
comment.createdById
)
).filter((recipient) => !userIdsSentNotifications.includes(recipient.id));
for (const recipient of recipients) {
const notification = await Notification.create({
event: event.name,
userId: recipient.id,
actorId: comment.createdById,
teamId: team.id,
documentId: document.id,
});
await new CommentCreatedEmail(
{
to: recipient.email,
userId: recipient.id,
documentId: document.id,
teamUrl: team.url,
isReply: !!comment.parentCommentId,
actorName: comment.createdBy.name,
commentId: comment.id,
content,
collectionName: document.collection?.name,
},
{ notificationId: notification.id }
).schedule();
}
}
public get options() {
return {
attempts: 1,
priority: TaskPriority.Background,
};
}
}

91
server/queues/tasks/CommentCreatedNotificationsTask.ts Normal file
View File

@@ -0,0 +1,91 @@
import { NotificationEventType } from "@shared/types";
import subscriptionCreator from "@server/commands/subscriptionCreator";
import { sequelize } from "@server/database/sequelize";
import { Comment, Document, Notification, User } from "@server/models";
import NotificationHelper from "@server/models/helpers/NotificationHelper";
import ProsemirrorHelper from "@server/models/helpers/ProsemirrorHelper";
import { CommentEvent } from "@server/types";
import BaseTask, { TaskPriority } from "./BaseTask";
export default class CommentCreatedNotificationsTask extends BaseTask<
CommentEvent
> {
public async perform(event: CommentEvent) {
const [document, comment] = await Promise.all([
Document.scope("withCollection").findOne({
where: {
id: event.documentId,
},
}),
Comment.findByPk(event.modelId),
]);
if (!document || !comment) {
return;
}
// Commenting on a doc automatically creates a subscription to the doc
// if they haven't previously had one.
await sequelize.transaction(async (transaction) => {
await subscriptionCreator({
user: comment.createdBy,
documentId: document.id,
event: "documents.update",
resubscribe: false,
transaction,
ip: event.ip,
});
});
const mentions = ProsemirrorHelper.parseMentions(
ProsemirrorHelper.toProsemirror(comment.data)
);
const userIdsMentioned: string[] = [];
for (const mention of mentions) {
const recipient = await User.findByPk(mention.modelId);
if (
recipient &&
recipient.id !== mention.actorId &&
recipient.subscribedToEventType(
NotificationEventType.MentionedInComment
)
) {
await Notification.create({
event: NotificationEventType.MentionedInComment,
userId: recipient.id,
actorId: mention.actorId,
teamId: document.teamId,
commentId: comment.id,
documentId: document.id,
});
userIdsMentioned.push(recipient.id);
}
}
const recipients = (
await NotificationHelper.getCommentNotificationRecipients(
document,
comment,
comment.createdById
)
).filter((recipient) => !userIdsMentioned.includes(recipient.id));
for (const recipient of recipients) {
await Notification.create({
event: NotificationEventType.CreateComment,
userId: recipient.id,
actorId: comment.createdById,
teamId: document.teamId,
commentId: comment.id,
documentId: document.id,
});
}
}
public get options() {
return {
priority: TaskPriority.Background,
};
}
}

91
server/queues/tasks/CommentUpdatedNotificationTask.ts
View File

@@ -1,91 +0,0 @@
import { NotificationEventType } from "@shared/types";
import CommentMentionedEmail from "@server/emails/templates/CommentMentionedEmail";
import { Comment, Document, Notification, Team, User } from "@server/models";
import DocumentHelper from "@server/models/helpers/DocumentHelper";
import ProsemirrorHelper from "@server/models/helpers/ProsemirrorHelper";
import { CommentEvent, CommentUpdateEvent } from "@server/types";
import BaseTask, { TaskPriority } from "./BaseTask";
export default class CommentUpdatedNotificationTask extends BaseTask<
CommentEvent
> {
public async perform(event: CommentUpdateEvent) {
const [document, comment, team] = await Promise.all([
Document.scope("withCollection").findOne({
where: {
id: event.documentId,
},
}),
Comment.findByPk(event.modelId),
Team.findByPk(event.teamId),
]);
if (!document || !comment || !team) {
return;
}
const mentions = ProsemirrorHelper.parseMentions(
ProsemirrorHelper.toProsemirror(comment.data)
).filter((mention) => event.data.newMentionIds.includes(mention.id));
if (mentions.length === 0) {
return;
}
let content = ProsemirrorHelper.toHTML(
ProsemirrorHelper.toProsemirror(comment.data),
{
centered: false,
}
);
if (!content) {
return;
}
content = await DocumentHelper.attachmentsToSignedUrls(
content,
event.teamId,
86400 * 4
);
for (const mention of mentions) {
const [recipient, actor] = await Promise.all([
User.findByPk(mention.modelId),
User.findByPk(mention.actorId),
]);
if (
recipient &&
actor &&
recipient.id !== actor.id &&
recipient.subscribedToEventType(NotificationEventType.Mentioned)
) {
const notification = await Notification.create({
event: event.name,
userId: recipient.id,
actorId: actor.id,
teamId: team.id,
documentId: document.id,
});
await new CommentMentionedEmail(
{
to: recipient.email,
userId: recipient.id,
documentId: document.id,
teamUrl: team.url,
actorName: comment.createdBy.name,
commentId: comment.id,
content,
collectionName: document.collection?.name,
},
{ notificationId: notification.id }
).schedule();
}
}
}
public get options() {
return {
attempts: 1,
priority: TaskPriority.Background,
};
}
}

57
server/queues/tasks/CommentUpdatedNotificationsTask.ts Normal file
View File

@@ -0,0 +1,57 @@
import { NotificationEventType } from "@shared/types";
import { Comment, Document, Notification, User } from "@server/models";
import ProsemirrorHelper from "@server/models/helpers/ProsemirrorHelper";
import { CommentEvent, CommentUpdateEvent } from "@server/types";
import BaseTask, { TaskPriority } from "./BaseTask";
export default class CommentUpdatedNotificationsTask extends BaseTask<
CommentEvent
> {
public async perform(event: CommentUpdateEvent) {
const [document, comment] = await Promise.all([
Document.scope("withCollection").findOne({
where: {
id: event.documentId,
},
}),
Comment.findByPk(event.modelId),
]);
if (!document || !comment) {
return;
}
const mentions = ProsemirrorHelper.parseMentions(
ProsemirrorHelper.toProsemirror(comment.data)
).filter((mention) => event.data.newMentionIds.includes(mention.id));
if (mentions.length === 0) {
return;
}
for (const mention of mentions) {
const recipient = await User.findByPk(mention.modelId);
if (
recipient &&
recipient.id !== mention.actorId &&
recipient.subscribedToEventType(
NotificationEventType.MentionedInComment
)
) {
await Notification.create({
event: NotificationEventType.MentionedInComment,
userId: recipient.id,
actorId: mention.actorId,
teamId: document.teamId,
documentId: document.id,
});
}
}
}
public get options() {
return {
attempts: 1,
priority: TaskPriority.Background,
};
}
}

137
server/queues/tasks/DocumentPublishedNotificationsTask.test.ts Normal file
View File

@@ -0,0 +1,137 @@
import { NotificationEventType } from "@shared/types";
import { Notification } from "@server/models";
import {
buildDocument,
buildCollection,
buildUser,
} from "@server/test/factories";
import { setupTestDatabase } from "@server/test/support";
import DocumentPublishedNotificationsTask from "./DocumentPublishedNotificationsTask";
const ip = "127.0.0.1";
setupTestDatabase();
beforeEach(async () => {
jest.resetAllMocks();
});
describe("documents.publish", () => {
test("should not send a notification to author", async () => {
const spy = jest.spyOn(Notification, "create");
const user = await buildUser();
const document = await buildDocument({
teamId: user.teamId,
lastModifiedById: user.id,
});
user.setNotificationEventType(NotificationEventType.PublishDocument);
await user.save();
const processor = new DocumentPublishedNotificationsTask();
await processor.perform({
name: "documents.publish",
documentId: document.id,
collectionId: document.collectionId,
teamId: document.teamId,
actorId: document.createdById,
data: {
title: document.title,
},
ip,
});
expect(spy).not.toHaveBeenCalled();
});
test("should send a notification to other users in team", async () => {
const spy = jest.spyOn(Notification, "create");
const user = await buildUser();
const document = await buildDocument({
teamId: user.teamId,
});
user.setNotificationEventType(NotificationEventType.PublishDocument);
await user.save();
const processor = new DocumentPublishedNotificationsTask();
await processor.perform({
name: "documents.publish",
documentId: document.id,
collectionId: document.collectionId,
teamId: document.teamId,
actorId: document.createdById,
data: {
title: document.title,
},
ip,
});
expect(spy).toHaveBeenCalled();
});
test("should send only one notification in a 12-hour window", async () => {
const spy = jest.spyOn(Notification, "create");
const user = await buildUser();
const document = await buildDocument({
teamId: user.teamId,
createdById: user.id,
lastModifiedById: user.id,
});
const recipient = await buildUser({
teamId: user.teamId,
});
user.setNotificationEventType(NotificationEventType.PublishDocument);
await user.save();
await Notification.create({
event: NotificationEventType.PublishDocument,
actorId: user.id,
userId: recipient.id,
documentId: document.id,
teamId: recipient.teamId,
emailedAt: new Date(),
});
const processor = new DocumentPublishedNotificationsTask();
await processor.perform({
name: "documents.publish",
documentId: document.id,
collectionId: document.collectionId,
teamId: document.teamId,
actorId: document.createdById,
data: {
title: document.title,
},
ip,
});
expect(spy).toHaveBeenCalledTimes(1);
});
test("should not send a notification to users without collection access", async () => {
const spy = jest.spyOn(Notification, "create");
const user = await buildUser();
const collection = await buildCollection({
teamId: user.teamId,
permission: null,
});
const document = await buildDocument({
teamId: user.teamId,
collectionId: collection.id,
});
user.setNotificationEventType(NotificationEventType.PublishDocument);
await user.save();
const processor = new DocumentPublishedNotificationsTask();
await processor.perform({
name: "documents.publish",
documentId: document.id,
collectionId: document.collectionId,
teamId: document.teamId,
actorId: document.createdById,
data: {
title: document.title,
},
ip,
});
expect(spy).not.toHaveBeenCalled();
});
});

72
server/queues/tasks/DocumentPublishedNotificationsTask.ts Normal file
View File

@@ -0,0 +1,72 @@
import { NotificationEventType } from "@shared/types";
import { createSubscriptionsForDocument } from "@server/commands/subscriptionCreator";
import { Document, Notification, User } from "@server/models";
import DocumentHelper from "@server/models/helpers/DocumentHelper";
import NotificationHelper from "@server/models/helpers/NotificationHelper";
import { DocumentEvent } from "@server/types";
import BaseTask, { TaskPriority } from "./BaseTask";
export default class DocumentPublishedNotificationsTask extends BaseTask<
DocumentEvent
> {
public async perform(event: DocumentEvent) {
const document = await Document.findByPk(event.documentId, {
includeState: true,
});
if (!document) {
return;
}
await createSubscriptionsForDocument(document, event);
// Send notifications to mentioned users first
const mentions = DocumentHelper.parseMentions(document);
const userIdsMentioned: string[] = [];
for (const mention of mentions) {
const recipient = await User.findByPk(mention.modelId);
if (
recipient &&
recipient.id !== mention.actorId &&
recipient.subscribedToEventType(
NotificationEventType.MentionedInDocument
)
) {
await Notification.create({
event: NotificationEventType.MentionedInDocument,
userId: recipient.id,
actorId: document.updatedBy.id,
teamId: document.teamId,
documentId: document.id,
});
userIdsMentioned.push(recipient.id);
}
}
const recipients = (
await NotificationHelper.getDocumentNotificationRecipients(
document,
NotificationEventType.PublishDocument,
document.lastModifiedById,
false
)
).filter((recipient) => !userIdsMentioned.includes(recipient.id));
for (const recipient of recipients) {
await Notification.create({
event: NotificationEventType.PublishDocument,
userId: recipient.id,
actorId: document.updatedBy.id,
teamId: document.teamId,
documentId: document.id,
});
}
}
public get options() {
return {
priority: TaskPriority.Background,
};
}
}

284
server/queues/processors/NotificationsProcessor.test.ts → server/queues/tasks/RevisionCreatedNotificationsTask.test.ts
View File

@@ -1,5 +1,3 @@
import { NotificationEventType } from "@shared/types";
import DocumentPublishedOrUpdatedEmail from "@server/emails/templates/DocumentPublishedOrUpdatedEmail";
import {
View,
Subscription,
@@ -7,13 +5,9 @@ import {
Notification,
Revision,
} from "@server/models";
import {
buildDocument,
buildCollection,
buildUser,
} from "@server/test/factories";
import { buildDocument, buildUser } from "@server/test/factories";
import { setupTestDatabase } from "@server/test/support";
import NotificationsProcessor from "./NotificationsProcessor";
import RevisionCreatedNotificationsTask from "./RevisionCreatedNotificationsTask";
const ip = "127.0.0.1";
@@ -23,145 +17,9 @@ beforeEach(async () => {
jest.resetAllMocks();
});
describe("documents.publish", () => {
test("should not send a notification to author", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const user = await buildUser();
const document = await buildDocument({
teamId: user.teamId,
lastModifiedById: user.id,
});
user.setNotificationEventType(NotificationEventType.PublishDocument);
await user.save();
const processor = new NotificationsProcessor();
await processor.perform({
name: "documents.publish",
documentId: document.id,
collectionId: document.collectionId,
teamId: document.teamId,
actorId: document.createdById,
data: {
title: document.title,
},
ip,
});
expect(schedule).not.toHaveBeenCalled();
});
test("should send a notification to other users in team", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const user = await buildUser();
const document = await buildDocument({
teamId: user.teamId,
});
user.setNotificationEventType(NotificationEventType.PublishDocument);
await user.save();
const processor = new NotificationsProcessor();
await processor.perform({
name: "documents.publish",
documentId: document.id,
collectionId: document.collectionId,
teamId: document.teamId,
actorId: document.createdById,
data: {
title: document.title,
},
ip,
});
expect(schedule).toHaveBeenCalled();
});
test("should send only one notification in a 12-hour window", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const user = await buildUser();
const document = await buildDocument({
teamId: user.teamId,
createdById: user.id,
lastModifiedById: user.id,
});
const recipient = await buildUser({
teamId: user.teamId,
});
user.setNotificationEventType(NotificationEventType.PublishDocument);
await user.save();
await Notification.create({
actorId: user.id,
userId: recipient.id,
documentId: document.id,
teamId: recipient.teamId,
event: "documents.publish",
emailedAt: new Date(),
});
const processor = new NotificationsProcessor();
await processor.perform({
name: "documents.publish",
documentId: document.id,
collectionId: document.collectionId,
teamId: document.teamId,
actorId: document.createdById,
data: {
title: document.title,
},
ip,
});
expect(schedule).not.toHaveBeenCalled();
});
test("should not send a notification to users without collection access", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const user = await buildUser();
const collection = await buildCollection({
teamId: user.teamId,
permission: null,
});
const document = await buildDocument({
teamId: user.teamId,
collectionId: collection.id,
});
user.setNotificationEventType(NotificationEventType.PublishDocument);
await user.save();
const processor = new NotificationsProcessor();
await processor.perform({
name: "documents.publish",
documentId: document.id,
collectionId: document.collectionId,
teamId: document.teamId,
actorId: document.createdById,
data: {
title: document.title,
},
ip,
});
expect(schedule).not.toHaveBeenCalled();
});
});
describe("revisions.create", () => {
test("should send a notification to other collaborators", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const document = await buildDocument();
await Revision.createFromDocument(document);
@@ -172,8 +30,8 @@ describe("revisions.create", () => {
document.collaboratorIds = [collaborator.id];
await document.save();
const processor = new NotificationsProcessor();
await processor.perform({
const task = new RevisionCreatedNotificationsTask();
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -182,14 +40,11 @@ describe("revisions.create", () => {
modelId: revision.id,
ip,
});
expect(schedule).toHaveBeenCalled();
expect(spy).toHaveBeenCalled();
});
test("should not send a notification if viewed since update", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const document = await buildDocument();
await Revision.createFromDocument(document);
document.text = "Updated body content";
@@ -204,8 +59,8 @@ describe("revisions.create", () => {
documentId: document.id,
});
const processor = new NotificationsProcessor();
await processor.perform({
const task = new RevisionCreatedNotificationsTask();
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -214,14 +69,11 @@ describe("revisions.create", () => {
modelId: revision.id,
ip,
});
expect(schedule).not.toHaveBeenCalled();
expect(spy).not.toHaveBeenCalled();
});
test("should not send a notification to last editor", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const user = await buildUser();
const document = await buildDocument({
teamId: user.teamId,
@@ -232,8 +84,8 @@ describe("revisions.create", () => {
document.updatedAt = new Date();
const revision = await Revision.createFromDocument(document);
const processor = new NotificationsProcessor();
await processor.perform({
const task = new RevisionCreatedNotificationsTask();
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -242,14 +94,11 @@ describe("revisions.create", () => {
modelId: revision.id,
ip,
});
expect(schedule).not.toHaveBeenCalled();
expect(spy).not.toHaveBeenCalled();
});
test("should send a notification for subscriptions, even to collaborator", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const document = await buildDocument();
await Revision.createFromDocument(document);
document.text = "Updated body content";
@@ -269,9 +118,9 @@ describe("revisions.create", () => {
enabled: true,
});
const processor = new NotificationsProcessor();
const task = new RevisionCreatedNotificationsTask();
await processor.perform({
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -281,7 +130,7 @@ describe("revisions.create", () => {
ip,
});
expect(schedule).toHaveBeenCalled();
expect(spy).toHaveBeenCalled();
});
test("should create subscriptions for collaborator", async () => {
@@ -298,9 +147,9 @@ describe("revisions.create", () => {
collaboratorIds: [collaborator0.id, collaborator1.id, collaborator2.id],
});
const processor = new NotificationsProcessor();
const task = new RevisionCreatedNotificationsTask();
await processor.perform({
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -330,10 +179,7 @@ describe("revisions.create", () => {
});
test("should not send multiple emails", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const collaborator0 = await buildUser();
const collaborator1 = await buildUser({ teamId: collaborator0.teamId });
const collaborator2 = await buildUser({ teamId: collaborator0.teamId });
@@ -350,22 +196,10 @@ describe("revisions.create", () => {
collaboratorIds: [collaborator0.id, collaborator1.id, collaborator2.id],
});
const processor = new NotificationsProcessor();
// Changing document will emit a `documents.update` event.
await processor.perform({
name: "documents.update",
documentId: document.id,
collectionId: document.collectionId,
createdAt: document.updatedAt.toString(),
teamId: document.teamId,
data: { title: document.title, autosave: false, done: true },
actorId: collaborator2.id,
ip,
});
const task = new RevisionCreatedNotificationsTask();
// Those changes will also emit a `revisions.create` event.
await processor.perform({
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -377,14 +211,11 @@ describe("revisions.create", () => {
// This should send out 2 emails, one for each collaborator that did not
// participate in the edit
expect(schedule).toHaveBeenCalledTimes(2);
expect(spy).toHaveBeenCalledTimes(2);
});
test("should not create subscriptions if previously unsubscribed", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const collaborator0 = await buildUser();
const collaborator1 = await buildUser({ teamId: collaborator0.teamId });
const collaborator2 = await buildUser({ teamId: collaborator0.teamId });
@@ -411,9 +242,9 @@ describe("revisions.create", () => {
// `collaborator2` would no longer like to be notified.
await subscription2.destroy();
const processor = new NotificationsProcessor();
const task = new RevisionCreatedNotificationsTask();
await processor.perform({
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -440,14 +271,11 @@ describe("revisions.create", () => {
// One notification as one collaborator performed edit and the other is
// unsubscribed
expect(schedule).toHaveBeenCalledTimes(1);
expect(spy).toHaveBeenCalledTimes(1);
});
test("should send a notification for subscriptions to non-collaborators", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const document = await buildDocument();
const collaborator = await buildUser({ teamId: document.teamId });
const subscriber = await buildUser({ teamId: document.teamId });
@@ -470,9 +298,9 @@ describe("revisions.create", () => {
enabled: true,
});
const processor = new NotificationsProcessor();
const task = new RevisionCreatedNotificationsTask();
await processor.perform({
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -482,14 +310,12 @@ describe("revisions.create", () => {
ip,
});
expect(schedule).toHaveBeenCalled();
expect(spy).toHaveBeenCalled();
});
test("should not send a notification for subscriptions to collaborators if unsubscribed", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const document = await buildDocument();
await Revision.createFromDocument(document);
document.text = "Updated body content";
@@ -514,9 +340,9 @@ describe("revisions.create", () => {
subscription.destroy();
const processor = new NotificationsProcessor();
const task = new RevisionCreatedNotificationsTask();
await processor.perform({
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -527,14 +353,12 @@ describe("revisions.create", () => {
});
// Should send notification to `collaborator` and not `subscriber`.
expect(schedule).toHaveBeenCalledTimes(1);
expect(spy).toHaveBeenCalledTimes(1);
});
test("should not send a notification for subscriptions to members outside of the team", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const document = await buildDocument();
await Revision.createFromDocument(document);
document.text = "Updated body content";
@@ -562,9 +386,9 @@ describe("revisions.create", () => {
enabled: true,
});
const processor = new NotificationsProcessor();
const task = new RevisionCreatedNotificationsTask();
await processor.perform({
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -575,14 +399,12 @@ describe("revisions.create", () => {
});
// Should send notification to `collaborator` and not `subscriber`.
expect(schedule).toHaveBeenCalledTimes(1);
expect(spy).toHaveBeenCalledTimes(1);
});
test("should not send a notification if viewed since update", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const document = await buildDocument();
const revision = await Revision.createFromDocument(document);
const collaborator = await buildUser({ teamId: document.teamId });
@@ -594,9 +416,9 @@ describe("revisions.create", () => {
documentId: document.id,
});
const processor = new NotificationsProcessor();
const task = new RevisionCreatedNotificationsTask();
await processor.perform({
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -605,14 +427,12 @@ describe("revisions.create", () => {
modelId: revision.id,
ip,
});
expect(schedule).not.toHaveBeenCalled();
expect(spy).not.toHaveBeenCalled();
});
test("should not send a notification to last editor", async () => {
const schedule = jest.spyOn(
DocumentPublishedOrUpdatedEmail.prototype,
"schedule"
);
const spy = jest.spyOn(Notification, "create");
const user = await buildUser();
const document = await buildDocument({
teamId: user.teamId,
@@ -620,8 +440,8 @@ describe("revisions.create", () => {
});
const revision = await Revision.createFromDocument(document);
const processor = new NotificationsProcessor();
await processor.perform({
const task = new RevisionCreatedNotificationsTask();
await task.perform({
name: "revisions.create",
documentId: document.id,
collectionId: document.collectionId,
@@ -630,6 +450,6 @@ describe("revisions.create", () => {
modelId: revision.id,
ip,
});
expect(schedule).not.toHaveBeenCalled();
expect(spy).not.toHaveBeenCalled();
});
});

145
server/queues/tasks/RevisionCreatedNotificationsTask.ts Normal file
View File

@@ -0,0 +1,145 @@
import { subHours } from "date-fns";
import { differenceBy } from "lodash";
import { Op } from "sequelize";
import { NotificationEventType } from "@shared/types";
import { createSubscriptionsForDocument } from "@server/commands/subscriptionCreator";
import env from "@server/env";
import Logger from "@server/logging/Logger";
import { Document, Revision, Notification, User, View } from "@server/models";
import DocumentHelper from "@server/models/helpers/DocumentHelper";
import NotificationHelper from "@server/models/helpers/NotificationHelper";
import { RevisionEvent } from "@server/types";
import BaseTask, { TaskPriority } from "./BaseTask";
export default class RevisionCreatedNotificationsTask extends BaseTask<
RevisionEvent
> {
public async perform(event: RevisionEvent) {
const [document, revision] = await Promise.all([
Document.findByPk(event.documentId, { includeState: true }),
Revision.findByPk(event.modelId),
]);
if (!document || !revision) {
return;
}
await createSubscriptionsForDocument(document, event);
// Send notifications to mentioned users first
const before = await revision.previous();
const oldMentions = before ? DocumentHelper.parseMentions(before) : [];
const newMentions = DocumentHelper.parseMentions(document);
const mentions = differenceBy(newMentions, oldMentions, "id");
const userIdsMentioned: string[] = [];
for (const mention of mentions) {
const recipient = await User.findByPk(mention.modelId);
if (
recipient &&
recipient.id !== mention.actorId &&
recipient.subscribedToEventType(
NotificationEventType.MentionedInDocument
)
) {
await Notification.create({
event: NotificationEventType.MentionedInDocument,
userId: recipient.id,
revisionId: event.modelId,
actorId: document.updatedBy.id,
teamId: document.teamId,
documentId: document.id,
});
userIdsMentioned.push(recipient.id);
}
}
const recipients = (
await NotificationHelper.getDocumentNotificationRecipients(
document,
NotificationEventType.UpdateDocument,
document.lastModifiedById,
true
)
).filter((recipient) => !userIdsMentioned.includes(recipient.id));
if (!recipients.length) {
return;
}
for (const recipient of recipients) {
const notify = await this.shouldNotify(document, recipient);
if (notify) {
await Notification.create({
event: NotificationEventType.UpdateDocument,
userId: recipient.id,
revisionId: event.modelId,
actorId: document.updatedBy.id,
teamId: document.teamId,
documentId: document.id,
});
}
}
}
private shouldNotify = async (
document: Document,
user: User
): Promise<boolean> => {
// Create only a single notification in a 6 hour window
const notification = await Notification.findOne({
order: [["createdAt", "DESC"]],
where: {
userId: user.id,
documentId: document.id,
emailedAt: {
[Op.not]: null,
[Op.gte]: subHours(new Date(), 6),
},
},
});
if (notification) {
if (env.ENVIRONMENT === "development") {
Logger.info(
"processor",
`would have suppressed notification to ${user.id}, but not in development`
);
} else {
Logger.info(
"processor",
`suppressing notification to ${user.id} as recently notified`
);
return false;
}
}
// If this recipient has viewed the document since the last update was made
// then we can avoid sending them a useless notification, yay.
const view = await View.findOne({
where: {
userId: user.id,
documentId: document.id,
updatedAt: {
[Op.gt]: document.updatedAt,
},
},
});
if (view) {
Logger.info(
"processor",
`suppressing notification to ${user.id} because update viewed`
);
return false;
}
return true;
};
public get options() {
return {
priority: TaskPriority.Background,
};
}
}