🔔 Implementing notifications (#231)

tdrive/backend/node/config/custom-environment-variables.json

@@ -83,6 +83,18 @@
       "secretKey": "STORAGE_S3_SECRET_KEY"
     }
   },
+  "email-pusher":{
+    "email_interface": "EMAIL_INTERFACE",
+    "endpoint":"EMAIL_ENDPOINT",
+    "api_key":"EMAIL_API_KEY",
+    "sender":"EMAIL_SENDER",
+    "smtp_user": "EMAIL_SMTP_USER",
+    "smtp_password": "EMAIL_SMTP_PASSWORD",
+    "smtp_host": "EMAIL_SMTP_HOST",
+    "smtp_port": "EMAIL_SMTP_PORT",
+    "smtp_tls": "EMAIL_SMTP_SECURE",
+    "debug": "EMAIL_DEBUG"
+  },
   "message-queue": {
     "type": "PUBSUB_TYPE",
     "amqp": {

tdrive/backend/node/config/default.json

@@ -138,7 +138,8 @@
   "email-pusher":{
      "endpoint":"https://api.smtp2go.com/v3/email/send",
      "api_key":"secret",
-     "sender":"noreply@twake.app"
+     "sender":"noreply@twake.app",
+     "debug": true
   },
   "applications":{
      "grid":[

tdrive/backend/node/package.json

@@ -103,9 +103,9 @@
   "dependencies": {
     "@elastic/elasticsearch": "7",
     "@fastify/caching": "^7.0.0",
+    "@fastify/cookie": "^6.0.0",
     "@fastify/formbody": "^6.0.0",
     "@fastify/static": "^5.0.1",
-    "@fastify/cookie": "^6.0.0",
     "@ffprobe-installer/ffprobe": "^1.4.1",
     "@sentry/node": "^6.19.7",
     "@sentry/tracing": "^6.19.7",
@@ -161,6 +161,7 @@
     "node-cache": "^5.1.2",
     "node-cron": "^3.0.0",
     "node-fetch": "^2.6.7",
+    "nodemailer": "^6.9.7",
     "openid-client": "^5.4.0",
     "ora": "^5.4.0",
     "pdf-parse": "^1.1.1",

tdrive/backend/node/src/core/platform/services/cron/index.ts

@@ -1,5 +1,5 @@
 import * as cron from "node-cron";
-import uuid from "uuid";
+import { v4 as uuidv4 } from "uuid";
 
 import { TdriveService } from "../../framework";
 import { CronAPI, CronJob, CronExpression, CronTask } from "./api";
@@ -16,7 +16,7 @@ export default class CronService extends TdriveService<CronAPI> implements CronA
   schedule(expression: CronExpression, job: CronJob, description?: string): CronTask {
     this.logger.debug(`Submit new job with name ${description}`);
     const task: CronTask = {
-      id: uuid.v4(),
+      id: uuidv4(),
       description: description || "",
       startDate: Date.now(),
       lastRun: 0,

tdrive/backend/node/src/core/platform/services/email-pusher/index.ts

@@ -7,7 +7,9 @@ import {
   EmailPusherEmailType,
   EmailPusherPayload,
   EmailPusherResponseType,
+  SMTPClientConfigType,
 } from "./types";
+import nodemailer from "nodemailer";
 import * as Eta from "eta";
 import { convert } from "html-to-text";
 import path from "path";
@@ -21,9 +23,12 @@ export default class EmailPusherClass
   readonly name = "email-pusher";
   readonly version: "1.0.0";
   logger: TdriveLogger = getLogger("email-pusher-service");
+  interface: string;
   apiKey: string;
   apiUrl: string;
   sender: string;
+  transporter: any;
+  debug: boolean;
 
   api(): EmailPusherAPI {
     return this;
@@ -33,10 +38,26 @@ export default class EmailPusherClass
     Eta.configure({
       views: path.join(__dirname, "templates"),
     });
-
+    this.interface = this.configuration.get<string>("email_interface", "");
-    this.apiUrl = this.configuration.get<string>("endpoint", "");
+    if (this.interface === "smtp") {
-    this.apiKey = this.configuration.get<string>("api_key", "");
+      const smtpConfig: SMTPClientConfigType = {
-    this.sender = this.configuration.get<string>("sender", "");
+        host: this.configuration.get<string>("smtp_host", ""),
+        port: this.configuration.get<number>("smtp_port", 25),
+        requireTLS: this.configuration.get<boolean>("smtp_tls", true),
+        auth: {
+          user: this.configuration.get<string>("smtp_user", ""),
+          pass: this.configuration.get<string>("smtp_password", ""),
+        },
+      };
+      this.transporter = nodemailer.createTransport(smtpConfig);
+      this.sender = this.configuration.get<string>("smtp_from", "");
+    } else {
+      this.transporter = null;
+      this.apiUrl = this.configuration.get<string>("endpoint", "");
+      this.apiKey = this.configuration.get<string>("api_key", "");
+      this.sender = this.configuration.get<string>("sender", "");
+      this.debug = this.configuration.get<boolean>("debug", false);
+    }
 
     return this;
   }
@@ -108,21 +129,40 @@ export default class EmailPusherClass
         sender: this.sender,
       };
 
-      const { data } = await axios.post<EmailPusherEmailType, EmailPusherResponseType>(
+      if (this.debug) {
-        `${this.apiUrl}`,
+        this.logger.info("EMAIL::SENT ", { emailObject });
-        emailObject,
+      } else {
-      );
+        if (this.interface === "smtp") {
+          try {
+            const info = await this.transporter.sendMail({
+              from: `"Sender Name" <${this.sender}>`,
+              to: to,
+              subject: subject,
+              text: text_body,
+              html: html_body,
+            });
 
-      if (data.error && data.error.length) {
+            this.logger.info("Message sent: %s", info.response);
-        throw Error(data.error);
+          } catch (err) {
-      }
+            this.logger.error({ error: `${err}` }, "Failed to send email");
+          }
+        } else {
+          const { data } = await axios.post<EmailPusherEmailType, EmailPusherResponseType>(
+            `${this.apiUrl}`,
+            emailObject,
+          );
+          if (data.error && data.error.length) {
+            throw Error(data.error);
+          }
 
-      if (data.failed === 1 && data.failures.length) {
+          if (data.failed === 1 && data.failures.length) {
-        throw Error(data.failures.join(""));
+            throw Error(data.failures.join(""));
-      }
+          }
 
-      if (data.succeeded) {
+          if (data.succeeded) {
-        this.logger.info("email sent");
+            this.logger.info("email sent");
+          }
+        }
       }
     } catch (error) {
       this.logger.error({ error: `${error}` }, "Failed to send email");

tdrive/backend/node/src/core/platform/services/email-pusher/templates/en/notification-document.eta

@@ -0,0 +1,43 @@
+<% layout('./_structure') %>
+<% it.title = 'Missed notifications from company ' +  it.company.name  %>
+
+<%~ includeFile("../common/_body.eta", {
+  paragraphs: [
+  `
+  <table border="0" cellpadding="0" cellspacing="0" role="presentation" style="vertical-align:top;" width="100%" >
+      <tbody>
+          <tr>
+              <td align="center" style="font-size:0px;padding:0 0 8px;word-break:break-word;">
+                  <div style="font-family:Ubuntu, Helvetica, Arial, sans-serif;font-size:24px;font-weight:800;line-height:29px;text-align:center;color:#000000;">
+                    While you were away
+                  </div>
+              </td>
+          </tr>
+      </tbody>
+  </table>
+  `,
+  `
+  <table border="0" cellpadding="0" cellspacing="0" role="presentation" style="vertical-align:top;" width="100%" >
+      <tbody>
+          <tr>
+              <td align="center" style="font-size:0px;padding:0 0 16px;word-break:break-word;" >
+                  <div style="font-family:Ubuntu, Helvetica, Arial, sans-serif;font-size:16px;line-height:29px;text-align:center;color:#000000;">
+                    A new document has been shared with you! 📄
+                  </div>
+              </td>
+          </tr>
+      </tbody>
+  </table>
+  `,
+  ...it.notifications.map(notification =>
+    includeFile("./notification-document/notification.eta", notification)
+  ),
+  `
+  <div style="font-family:Ubuntu, Helvetica, Arial, sans-serif;font-size:13px;line-height:1;text-align:left;color:#000000;">
+    <a class="main-button" href="https://web.tdrive.app/">
+    ${it.notifications.length>1 ? `See all ${it.notifications.length} messages` : `See on Tdrive`}
+    </a>
+  </div>
+  `
+  ]
+}) %>

tdrive/backend/node/src/core/platform/services/email-pusher/templates/en/notification-document.subject.eta

@@ -0,0 +1 @@
+Missed notifications in your company <%= it.company.name %>

tdrive/backend/node/src/core/platform/services/email-pusher/types.ts

@@ -1,12 +1,14 @@
+import { DriveFile } from "src/services/documents/entities/drive-file";
 import Company from "../../../../services/user/entities/company";
-import Workspace from "../../../../services/workspaces/entities/workspace";
 import User from "../../../../services/user/entities/user";
 
 export type EmailBuilderDataPayload = {
-  user: User;
+  sender: User;
+  receiver: User;
   company: Company;
-  notifications: {
+  notifications?: {
-    workspace: Workspace;
+    type: string;
+    item: DriveFile;
   }[];
 };
 
@@ -16,7 +18,7 @@ export type EmailBuilderRenderedResult = {
   subject: string;
 };
 
-export type EmailBuilderTemplateName = "notification-digest";
+export type EmailBuilderTemplateName = "notification-digest" | "notification-document";
 
 export type EmailPusherPayload = {
   subject: string;
@@ -43,3 +45,15 @@ export type EmailPusherResponseType = {
 };
 
 export type EmailLanguageType = "en" | "fr";
+
+export type SMTPClientConfigType = {
+  host: string;
+  port: number;
+  secure?: boolean;
+  requireTLS: boolean;
+  auth: {
+    user: string;
+    pass: string;
+  };
+  logger?: boolean;
+};

tdrive/backend/node/src/services/documents/services/engine/index.ts

@@ -1,10 +1,47 @@
 import globalResolver from "../../../global-resolver";
+import { logger } from "../../../../core/platform/framework";
+import { localEventBus } from "../../../../core/platform/framework/event-bus";
 import { Initializable } from "../../../../core/platform/framework";
+import { DocumentEvents, NotificationPayloadType } from "../../types";
 import { DocumentsProcessor } from "./extract-keywords";
+import Repository from "../../../../core/platform/services/database/services/orm/repository/repository";
 import { DriveFile, TYPE } from "../../entities/drive-file";
 import { DocumentsFinishedProcess } from "./save-keywords";
-
 export class DocumentsEngine implements Initializable {
+  private documentRepository: Repository<DriveFile>;
+
+  async DispatchDocumentEvent(e: NotificationPayloadType, event: string) {
+    const sender = await globalResolver.services.users.get({ id: e.notificationEmitter });
+    const receiver = await globalResolver.services.users.get({ id: e.notificationReceiver });
+    const company = await globalResolver.services.companies.getCompany({
+      id: e.context.company.id,
+    });
+    try {
+      const { html, text, subject } = await globalResolver.platformServices.emailPusher.build(
+        "notification-document",
+        receiver.language || "en",
+        {
+          sender,
+          receiver,
+          company,
+          notifications: [
+            {
+              type: event,
+              item: e.item,
+            },
+          ],
+        },
+      );
+      await globalResolver.platformServices.emailPusher.send(receiver.email_canonical, {
+        subject,
+        html,
+        text,
+      });
+    } catch (error) {
+      logger.error(error);
+    }
+  }
+
   async init(): Promise<this> {
     const repository = await globalResolver.database.getRepository<DriveFile>(TYPE, DriveFile);
 
@@ -13,6 +50,25 @@ export class DocumentsEngine implements Initializable {
       new DocumentsFinishedProcess(repository),
     );
 
+    localEventBus.subscribe(DocumentEvents.DOCUMENT_SAHRED, async (e: NotificationPayloadType) => {
+      await this.DispatchDocumentEvent(e, DocumentEvents.DOCUMENT_SAHRED);
+    });
+
+    localEventBus.subscribe(
+      DocumentEvents.DOCUMENT_VERSION_UPDATED,
+      async (e: NotificationPayloadType) => {
+        await this.DispatchDocumentEvent(e, DocumentEvents.DOCUMENT_VERSION_UPDATED);
+      },
+    );
+
     return this;
   }
+
+  notifyDocumentShared(notificationPayload: NotificationPayloadType) {
+    localEventBus.publish(DocumentEvents.DOCUMENT_SAHRED, notificationPayload);
+  }
+
+  notifyDocumentVersionUpdated(notificationPayload: NotificationPayloadType) {
+    localEventBus.publish(DocumentEvents.DOCUMENT_VERSION_UPDATED, notificationPayload);
+  }
 }

tdrive/backend/node/src/services/documents/services/index.ts

@@ -50,7 +50,6 @@ import {
   getItemScope,
 } from "./access-check";
 import { websocketEventBus } from "../../../core/platform/services/realtime/bus";
-
 import archiver from "archiver";
 import internal from "stream";
 import {
@@ -416,10 +415,26 @@ export class DocumentsService {
           } else {
             oldParent = item.parent_id;
           }
-
           if (key === "access_info") {
+            const sharedWith = content.access_info.entities.filter(
+              info =>
+                !item.access_info.entities.find(entity => entity.id === info.id) &&
+                info.type === "user",
+            );
+
             item.access_info = content.access_info;
-            item.access_info.entities.forEach(info => {
+
+            if (sharedWith.length > 0) {
+              // Notify the user that the document has been shared with them
+              gr.services.documents.engine.notifyDocumentShared({
+                context,
+                item,
+                notificationEmitter: context.user.id,
+                notificationReceiver: sharedWith[0].id,
+              });
+            }
+
+            item.access_info.entities.forEach(async info => {
               if (!info.grantor) {
                 info.grantor = context.user.id;
               }
@@ -717,6 +732,14 @@ export class DocumentsService {
 
       await this.repository.save(item);
 
+      // Notify the user that the document versions have been updated
+      gr.services.documents.engine.notifyDocumentVersionUpdated({
+        context,
+        item,
+        notificationEmitter: context.user.id,
+        notificationReceiver: item.creator,
+      });
+
       this.notifyWebsocket(item.parent_id, context);
       await updateItemSize(item.parent_id, this.repository, context);
 

tdrive/backend/node/src/services/documents/types.ts

@@ -91,3 +91,15 @@ export type DriveTdriveTab = {
   item_id: string;
   level: "read" | "write";
 };
+
+export enum DocumentEvents {
+  DOCUMENT_SAHRED = "document_shared",
+  DOCUMENT_VERSION_UPDATED = "document_version_updated",
+}
+
+export type NotificationPayloadType = {
+  context: CompanyExecutionContext;
+  item: DriveFile;
+  notificationEmitter: string;
+  notificationReceiver: string;
+};

tdrive/backend/node/src/utils/types.ts

@@ -104,6 +104,15 @@ export interface ResourceEventsPayload {
   workspace?: WorkspacePrimaryKey;
 }
 
+export interface DocumentEventsPayload {
+  user: User;
+  actor?: User;
+  document?: {
+    sender: string;
+  };
+  company?: { id: string };
+}
+
 export interface PaginationQueryParameters {
   // It is offset for MongoDB and pointer to the next page for ScyllaDB
   page_token?: string;

tdrive/backend/node/test/e2e/documents/documents-notifications.spec.ts

@@ -0,0 +1,104 @@
+import { describe, beforeEach, it, expect, afterAll, jest } from "@jest/globals";
+import { init, TestPlatform } from "../setup";
+import TestHelpers from "../common/common_test_helpers";
+import { DocumentsEngine } from "../../../src/services/documents/services/engine";
+import { deserialize } from "class-transformer";
+import { File } from "../../../src/services/files/entities/file";
+import { ResourceUpdateResponse } from "../../../src/utils/types";
+import { e2e_createDocument, e2e_createDocumentFile, e2e_createVersion } from "./utils";
+import { DriveFileMockClass } from "../common/entities/mock_entities";
+import { TestDbService } from "../utils.prepare.db";
+
+describe("the Drive feature", () => {
+  let platform: TestPlatform;
+  const notifyDocumentShared = jest.spyOn(DocumentsEngine.prototype, "notifyDocumentShared");
+  const notifyDocumentVersionUpdated = jest.spyOn(
+    DocumentsEngine.prototype,
+    "notifyDocumentVersionUpdated",
+  );
+
+  beforeEach(async () => {
+    platform = await init({
+      services: [
+        "webserver",
+        "database",
+        "applications",
+        "search",
+        "storage",
+        "message-queue",
+        "user",
+        "search",
+        "files",
+        "websocket",
+        "messages",
+        "auth",
+        "realtime",
+        "channels",
+        "counter",
+        "statistics",
+        "platform-services",
+        "documents",
+      ],
+    });
+  }, 300000000);
+
+  afterAll(async () => {
+    await platform?.tearDown();
+    // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+    // @ts-ignore
+    platform = null;
+  });
+
+  const createItem = async (): Promise<DriveFileMockClass> => {
+    await TestDbService.getInstance(platform, true);
+    const scope: "personal" | "shared" = "shared";
+    const item = {
+      name: "new test file",
+      parent_id: "root",
+      company_id: platform.workspace.company_id,
+      scope,
+    };
+
+    const version = {};
+
+    const response = await e2e_createDocument(platform, item, version);
+    return deserialize<DriveFileMockClass>(DriveFileMockClass, response.body);
+  };
+
+  it("Did notify the user after sharing a file.", async () => {
+    // jest.setTimeout(20000);
+    //given:: user uploaded one doc and give permission to another user
+    const oneUser = await TestHelpers.getInstance(platform, true, { companyRole: "admin" });
+    const anotherUser = await TestHelpers.getInstance(platform, true, { companyRole: "admin" });
+    //upload files
+    const doc = await oneUser.uploadRandomFileAndCreateDocument();
+    await new Promise(r => setTimeout(r, 3000));
+    //give permissions to the file
+    doc.access_info.entities.push({
+      type: "user",
+      id: anotherUser.user.id,
+      level: "read",
+      // eslint-disable-next-line @typescript-eslint/ban-ts-comment
+      // @ts-ignore
+      grantor: null,
+    });
+    await oneUser.updateDocument(doc.id, doc);
+
+    expect(notifyDocumentShared).toHaveBeenCalled();
+  });
+
+  it("Did notify the user after creating a new version for a file.", async () => {
+    const item = await createItem();
+    const fileUploadResponse = await e2e_createDocumentFile(platform);
+    const fileUploadResult = deserialize<ResourceUpdateResponse<File>>(
+      ResourceUpdateResponse,
+      fileUploadResponse.body,
+    );
+
+    const file_metadata = { external_id: fileUploadResult.resource.id };
+
+    await e2e_createVersion(platform, item.id, { filename: "file2", file_metadata });
+
+    expect(notifyDocumentVersionUpdated).toHaveBeenCalled();
+  });
+});