User service gRPC method and notifications

TODO.txt

@@ -1,5 +1,11 @@
 
 To Do:
+[] Finish user service gRPC server implementation
+[] Add necessary notifications send from item service
+[] Create load testing/benchmarking scripts
+[] Add flag to turn off sending emails during benchmarking (to prevent costing money)
+[] Create a run script to run all services as detached processes at once
+[] Add NGINX to reverse proxy requests
 
 [] Add https://github.com/inxilpro/node-app-root-path to make imports cleaner and easier to manage
 [] Investigate wether https://www.npmjs.com/package/lusca is worth using or not. It's in the TS boilerplate repository

package.json

@@ -10,7 +10,8 @@
     "items-service": "nodemon ./dist/items-service/main.js",
     "items-gql-service": "nodemon ./dist/items-qgl-service/main.js",
     "user-service": "nodemon ./dist/user-service/main.js",
-    "notification-service": "nodemon ./dist/notification-service/main.js"
+    "notification-service": "nodemon ./dist/notification-service/main.js",
+    "all-services": "npm run session-service & npm run items-service & npm run user-service & npm run notification-service"
   },
   "nodemonConfig": {
     "delay": "2500"

protos/user.proto

@@ -0,0 +1,21 @@
+syntax = "proto3";
+
+package user;
+
+
+service UserService {
+  rpc GetUserById (GetUserByIdRequest) returns (GetUserByIdResponse) {}
+}
+
+
+message GetUserByIdRequest {
+  int32 userId = 1;
+}
+
+message GetUserByIdResponse {
+  int32 userId = 1;
+  string email = 2;
+  string firstName = 3;
+  string lastName = 4;
+  string role = 5;
+}

services/items-service/config/grpc_config.ts

@@ -7,19 +7,28 @@ import path from "path";
 import { PackageDefinition } from "@grpc/proto-loader";
 
 
-// Get path to proto file
-const PROTO_PATH: string = path.join(__dirname, "../../../protos/session.proto");
+// Get path to proto files
+const SESSION_PROTO_PATH: string = path.join(__dirname, "../../../protos/session.proto");
+const USER_PROTO_PATH: string = path.join(__dirname, "../../../protos/user.proto");
 
-// Get session proto file
-const sessionPackageDefinition: PackageDefinition = loadSync(PROTO_PATH, { keepCase: true });
+// Get session and user proto files
+const sessionPackageDefinition: PackageDefinition = loadSync(SESSION_PROTO_PATH, { keepCase: true });
+const userPackageDefinition: PackageDefinition = loadSync(USER_PROTO_PATH, { keepCase: true });
 
 // Load session package and get gRPC client to session service
 // @ts-ignore gRPC proto file is dynamically imported, definition is not generated till runtime
-const loadedSessionGrpcPackage = grpc.loadPackageDefinition(sessionPackageDefinition).session
+const loadedSessionGrpcPackage = grpc.loadPackageDefinition(sessionPackageDefinition).session;
 // @ts-ignore
 const sessionServiceGrpcClient = new loadedSessionGrpcPackage.SessionService(process.env.SESSION_SERVICE_GRPC_URL, grpc.credentials.createInsecure());
 
-// Promisify gRPC client
+// Load user package and get gRPC client to user service
+// @ts-ignore gRPC proto file is dynamically imported, definition is not generated till runtime
+const loadedUserGrpcPackage = grpc.loadPackageDefinition(userPackageDefinition).user;
+// @ts-ignore
+const userServiceGrpcClient = new loadedUserGrpcPackage.UserService(process.env.USER_SERVICE_GRPC_URL, grpc.credentials.createInsecure());
+
+// Promisify gRPC clients
 grpcPromise.promisifyAll(sessionServiceGrpcClient);
+grpcPromise.promisifyAll(userServiceGrpcClient);
 
-export { sessionServiceGrpcClient };
+export { sessionServiceGrpcClient, userServiceGrpcClient };

services/items-service/middleware/authorization.ts

@@ -18,15 +18,15 @@ const verifySessionToken = async (req: Request, _res: Response, next: NextFuncti
 
     if (!authHeader) {
       // Unauthorized if Authentication header is missing from request
-      throw new ApiError("Authentication header is not present", HttpStatus.UNAUTHORIZED);
+      next(new ApiError("Authentication header is not present", HttpStatus.UNAUTHORIZED));
     }
 
     // Parse session token from Authorization header
     const sessionToken: string = authHeader.replace("Bearer ", "");
 
     if (!sessionToken) {
       // Unauthorized if session token is missing from request
-      throw new ApiError("Session token is missing from Authentication header", HttpStatus.UNAUTHORIZED);
+      next(new ApiError("Session token is missing from Authentication header", HttpStatus.UNAUTHORIZED));
     }
 
     // Make gRPC call to session service to validate session token

services/items-service/routes/items.ts

@@ -4,6 +4,8 @@ import HttpStatus from "http-status-codes";
 
 // Config
 import { sendNotificationToQueue } from "../config/rabbitmq_config";
+import { userServiceGrpcClient } from "../config/grpc_config";
+import logger from "../config/logger_config";
 
 // Middleware
 import { isAuthenticated, isAuthenticatedAdmin } from "../middleware/authorization";
@@ -214,13 +216,34 @@ router.put("/:itemId", [
     await sendNotificationToQueue(emailNotification);
 
     // Get user id of who the item was created by
-    // const { created_by_user_id: createdByUserId } = result.rows[0];
+    const { created_by_user_id: createdByUserId } = result.rows[0];
 
-    // if (userId !== createdByUserId) {
+    if (userId !== createdByUserId) {
       // User modified an item created by a different user, send owner of item an email
-      // TODO: Make gRPC call to user-service to get email address for user
-      // TODO: Send email to this user to notify them that their item has been modified
-    // }
+
+      try {
+        // Make gRPC call to user-service to get email address for user
+        const userInfo = await userServiceGrpcClient.getUserById().sendMessage({ userId: createdByUserId });
+
+        // Get email and firstName from user info
+        const { email, firstName } = userInfo;
+
+        // Create the content of the email notification
+        const emailNotification: IEmailNotification = {
+          email,
+          firstName,
+          subject: "Another user has modified your item!",
+          messageHeader: `Your item has been modified, now named ${name}`,
+          messageBody: `The description for ${name} is: ${description}.`
+        };
+
+        // Send email to this user to notify them that their item has been modified
+        await sendNotificationToQueue(emailNotification);
+      }
+      catch (error) {
+        logger.error(error);
+      }
+    }
 
     res.json(result.rows[0]);
   }
@@ -270,13 +293,34 @@ router.delete("/:itemId", [
     await sendNotificationToQueue(emailNotification);
 
     // Get user id of who the item was created by
-    // const { createdbyuserId: createdByUserId } = result.rows[0];
+    const { created_by_user_id: createdByUserId } = result.rows[0];
 
-    // if (userId !== createdByUserId) {
+    if (userId !== createdByUserId) {
       // User deleted an item created by a different user, send owner of item an email
-      // TODO: Make gRPC call to user-service to get email address for user
-      // TODO: Send email to this user to notify them that their item has been modified
-    // }
+
+      try {
+        // Make gRPC call to user-service to get email address for user
+        const userInfo = await userServiceGrpcClient.getUserById().sendMessage({ userId: createdByUserId });
+
+        // Get email and firstName from user info
+        const { email, firstName } = userInfo;
+
+        // Create the content of the email notification
+        const emailNotification: IEmailNotification = {
+          email,
+          firstName,
+          subject: "Another user has deleted your item!",
+          messageHeader: `Your item named ${name} has been deleted by another user`,
+          messageBody: `The description for ${name} was: ${description}.`
+        };
+
+        // Send email to this user to notify them that their item has been modified
+        await sendNotificationToQueue(emailNotification);
+      }
+      catch (error) {
+        logger.error(error);
+      }
+    }
 
     res.sendStatus(200);
   }

services/session-service/config/grpc_config.ts

@@ -15,10 +15,10 @@ import { validateSession, createSession, replaceSession, removeSession } from ".
 
 
 // Get path to proto file
-const PROTO_PATH: string = path.join(__dirname, "../../../protos/session.proto");
+const SESSION_PROTO_PATH: string = path.join(__dirname, "../../../protos/session.proto");
 
 // Load proto file
-const sessionPackageDefinition: PackageDefinition = loadSync(PROTO_PATH, { keepCase: true });
+const sessionPackageDefinition: PackageDefinition = loadSync(SESSION_PROTO_PATH, { keepCase: true });
 
 // Get proto package definition
 // @ts-ignore gRPC proto file is dynamically imported, definition is not generated till runtime
@@ -37,4 +37,4 @@ server.bind(process.env.SESSION_SERVICE_GRPC_BIND_URL, grpc.ServerCredentials.cr
 // Start gRPC server
 server.start();
 
-logger.info(`Session gRPC server listening on port: ${process.env.SESSION_SERVICE_GRPC_BIND_URL}.`);
+logger.info(`gRPC server listening on: ${process.env.SESSION_SERVICE_GRPC_BIND_URL}.`);

services/session-service/grpc/session.ts

@@ -3,7 +3,6 @@ import { signJwt, verifyJwt } from "../repository/jwt";
 import { getSessionToken, setSessionToken, removeSessionToken } from "../repository/session_manager";
 
 
-// TODO: What is the `call` type
 const validateSession = async (call: any, callback: Function): Promise<void> => {
   const sessionToken: string = call.request?.sessionToken;
 
@@ -12,7 +11,7 @@ const validateSession = async (call: any, callback: Function): Promise<void> =>
 
   if (!sessionValues) {
     // If token is invalid, no session values are recieved
-    callback("Invalid token", { userId: null, email: null, firstName: null, lastName: null, role: null });
+    callback("Invalid token", null);
   }
   else {
     // Get user id from session values
@@ -23,14 +22,14 @@ const validateSession = async (call: any, callback: Function): Promise<void> =>
 
     if (!cacheSessionToken) {
       // If no token found for user in redis, session has expired
-      callback("Token has expired or has been removed", { userId: null, email: null, firstName: null, lastName: null, role: null });
+      callback("Token has expired or has been removed", null);
     }
     else if (cacheSessionToken !== sessionToken) {
       // If redis token does not match passed token, remove it from redis
       await removeSessionToken(userId);
 
       // Return with an error and no session values
-      callback("Tokens do not match for user", { userId: null, email: null, firstName: null, lastName: null, role: null });
+      callback("Tokens do not match for user", null);
     }
     else {
       // Re-add token to redis to reset expiration time
@@ -42,7 +41,6 @@ const validateSession = async (call: any, callback: Function): Promise<void> =>
   }
 }
 
-// TODO: What is the `call` type
 const createSession = async (call: any, callback: Function): Promise<void> => {
   const { userId, email, firstName, lastName, role } = call.request;
 
@@ -52,14 +50,14 @@ const createSession = async (call: any, callback: Function): Promise<void> => {
   // Check if token creation was successful
   if (!sessionToken) {
     // Failed to create toke, return error
-    callback("Error creating JWT", { sessionToken: "" });
+    callback("Error creating JWT", null);
   }
   else {
     // Add session token to redis
     const result: string = await setSessionToken(userId, sessionToken);
 
     if (!result) {
-      callback("Error adding session to Redis", { sessionToken: "" });
+      callback("Error adding session to Redis", null);
     }
     else {
       // Return new JWT
@@ -75,7 +73,7 @@ const replaceSession = async (call: any, callback: Function): Promise<void> => {
   const result: number = await removeSessionToken(userId);
 
   if (!(result === 0 || result === 1)) {
-    callback("Error removing session token from Redis", { sessionToken: "" });
+    callback("Error removing session token from Redis", null);
   }
   else {
     // Create new session token with updated session values
@@ -84,14 +82,14 @@ const replaceSession = async (call: any, callback: Function): Promise<void> => {
     // Check if token creation was successful
     if (!sessionToken) {
       // Failed to create toke, return error
-      callback("Error creating JWT", { sessionToken: "" });
+      callback("Error creating JWT", null);
     }
     else {
       // Add new session token to redis
       const result = await setSessionToken(userId, sessionToken);
 
       if (!result) {
-        callback("Error adding session to Redis", { sessionToken: "" });
+        callback("Error adding session to Redis", null);
       }
       else {
         // Return new JWT
@@ -116,7 +114,6 @@ const removeSession = async (call: any, callback: Function): Promise<void> => {
   }
 }
 
-
 export {
   validateSession,
   createSession,

services/user-service/config/grpc_config.ts

@@ -3,15 +3,51 @@ import { loadSync } from "@grpc/proto-loader";
 import grpcPromise from "grpc-promise";
 import path from "path";
 
+// Config
+import logger from "./logger_config";
+
 // Types
+import { Server } from "grpc";
 import { PackageDefinition } from "@grpc/proto-loader";
 
+/** Start gRPC Server **/
+
+// Import gRPC method implementations
+import { getUserById } from "../grpc/user";
+
+// Get path to user proto file
+const USER_PROTO_PATH: string = path.join(__dirname, "../../../protos/user.proto");
+
+// Load user proto file
+const userPackageDefinition: PackageDefinition = loadSync(USER_PROTO_PATH, { keepCase: true });
+
+// Get proto package definition
+// @ts-ignore gRPC proto file is dynamically imported, definition is not generated till runtime
+const loadedUserGrpcPackage = grpc.loadPackageDefinition(userPackageDefinition).user;
+
+// Start gRPC server
+const server: Server = new Server();
+
+// Configure user service grpc methods
+// @ts-ignore
+server.addService(loadedUserGrpcPackage.UserService.service, { getUserById });
+
+// Bind gRPC server to url
+server.bind(process.env.USER_SERVICE_GRPC_BIND_URL, grpc.ServerCredentials.createInsecure());
+
+// Start gRPC server
+server.start();
+
+logger.info(`gRPC server listening on: ${process.env.USER_SERVICE_GRPC_BIND_URL}.`);
+
+
+/** Get gRPC client for session service **/
 
 // Get path to proto file
-const PROTO_PATH: string = path.join(__dirname, "../../../protos/session.proto");
+const SESSION_PROTO_PATH: string = path.join(__dirname, "../../../protos/session.proto");
 
 // Get session proto file
-const sessionPackageDefinition: PackageDefinition = loadSync(PROTO_PATH, { keepCase: true });
+const sessionPackageDefinition: PackageDefinition = loadSync(SESSION_PROTO_PATH, { keepCase: true });
 
 // Load session package and get gRPC client to session service
 // @ts-ignore gRPC proto file is dynamically imported, definition is not generated till runtime

services/user-service/grpc/user.ts

@@ -0,0 +1,32 @@
+// Repository
+import { getUserById as getUserByIdQuery } from "../repository/queries";
+
+// Types
+import { QueryResult } from "pg";
+
+
+const getUserById = async (call: any, callback: Function): Promise<void> => {
+  try {
+    const userId: number = call.request?.userId;
+
+    // Get user info from id
+    const result: QueryResult<any> = await getUserByIdQuery(String(userId));
+
+    // Check if we were able to find a user
+    if (result.rowCount !== 1) {
+      callback("No user found for this id", null);
+    }
+
+    // Get user info from query result
+    const { email, first_name: firstName, last_name: lastName, role } = result.rows[0];
+
+    // Return user info to caller
+    callback(null, { userId, email, firstName, lastName, role });
+  }
+  catch (error) {
+    // Return error to caller
+    callback("Error getting user info", null);
+  }
+}
+
+export { getUserById };