This commit is contained in:
Kfir Dayan 2023-07-09 00:00:11 +03:00
parent a0334bf886
commit 3e91c159d7
18 changed files with 222 additions and 333 deletions

View file

@ -10,56 +10,62 @@ This README.md file provides instructions for setting up and running various com
```shell
docker-compose up -d
```
```
## SQS/Terraform
Before proceeding, ensure that you have Terraform installed and have configured AWS CLI with the keys of an authorized user.
1. Navigate to the /infra/terraform directory.
2. Run the following command to initialize Terraform:
```terraform
terraform init
```
3. Run the following command to apply the Terraform configuration:
```terraform
terraform apply
```
## NotificationService
Before running the NotificationService, ensure that you have Node.js, TypeScript (tsc), and npm installed.
Navigate to the /notification-service directory.
Run the following command to install the dependencies:
```npm
npm install
```
Run the following command to build and run (Dem mode)the project:
```npm
npm run dev
```
## TodoService
Before running the TodoService, ensure that you have Node.js, TypeScript (tsc) and npm installed.
Navigate to the /todo-service directory.
Run the following command to install the dependencies:
```npm
npm install
```
Run the following command to build and run (Dem mode)the project:
```npm
npm run dev
```
## Requests
Examples are located in request.http - can be run in VSCode with the REST Client extension.

View file

@ -1,4 +1,4 @@
version: '3.8'
version: "3.8"
services:
mongodb:
image: arm64v8/mongo:4.0

View file

@ -1 +0,0 @@
requirepass your_password

View file

@ -9,4 +9,4 @@ resource "aws_sqs_queue" "queue" {
message_retention_seconds = 345600
visibility_timeout_seconds = 30
receive_wait_time_seconds = 0
}
}

View file

@ -1,10 +1,3 @@
## AWS-SDK ##
AWS_ACCESS_KEY_ID={AWS_ACCESS_KEY_ID}
AWS_SECRET_ACCESS_KEY={AWS_SECRET_ACCESS_KEY}
AWS_REGION={AWS_REGION}
AWS_SQS_URL=https://sqs.{AWS_REGION}.amazonaws.com/{ACCOUNT_ID}/
AWS_SQS_QUEUE_NAME={NAME_OF_QUEUE}
## MONGO USER ##
DATABASE_URL=mongodb://{USER}:{PASSWORD}@localhost:27017/{DB_NAME}
MONGO_DB_NAME={USER_DB_NAME}

View file

@ -1,122 +0,0 @@
// import aws from 'aws-sdk';
// import { ITodo } from '../interfaces/ITodo';
// import { MongoDb } from '../mongodb/MongoDb';
// const env = require('dotenv').config().parsed;
// export class Sqs {
// sqs: aws.SQS;
// queueUrl: string;
// constructor() {
// aws.config.update({
// accessKeyId: env.AWS_ACCESS_KEY_ID,
// secretAccessKey: env.AWS_SECRET_ACCESS_KEY,
// });
// this.sqs = new aws.SQS();
// this.queueUrl = env.AWS_SQS_URL + env.AWS_SQS_QUEUE_NAME;
// if (this.sqs) {
// console.log("SQS connected");
// }
// }
// async create(payload: ITodo, delayTimeForQueue: number) {
// let reQueueTime = 0;
// if (delayTimeForQueue > 900) {
// reQueueTime = delayTimeForQueue - 900;
// delayTimeForQueue = 900;
// }
// const params = {
// DelaySeconds: delayTimeForQueue,
// MessageAttributes: {},
// MessageBody: JSON.stringify({ payload, delayTimeForQueue, reQueueTime }),
// QueueUrl: this.queueUrl,
// };
// try {
// const data = await this.sqs.sendMessage(params).promise();
// console.log("Message sent to the queue", data.MessageId);
// return data;
// } catch (error) {
// console.log("Error", error);
// throw error;
// }
// }
// async startConsumer() {
// while (true) {
// const message = await this.getNextQueue();
// if (message) {
// const { payload, delayTimeForQueue, reQueueTime } = JSON.parse(message.Body) as {
// payload: ITodo;
// delayTimeForQueue: number;
// reQueueTime: number;
// };
// console.log("Received notification:", payload);
// if (reQueueTime === 0) {
// try {
// await MongoDb.updateTodoStatus(payload);
// } catch {
// await this.create(payload, delayTimeForQueue);
// await this.deleteMessage(message.ReceiptHandle);
// console.log("Published new queue with delay, THE DB IS DOWN!:", delayTimeForQueue);
// }
// await this.deleteMessage(message.ReceiptHandle);
// } else if (reQueueTime >= 900) {
// const newDelayTime = 900;
// const newReQueueTime = reQueueTime - 900;
// await this.create(payload, newDelayTime);
// await this.deleteMessage(message.ReceiptHandle);
// console.log("Published new queue with delay:", newDelayTime);
// } else {
// const newDelayTime = reQueueTime;
// await this.create(payload, newDelayTime);
// await this.deleteMessage(message.ReceiptHandle);
// console.log("Published new queue with delay:", newDelayTime);
// }
// }
// }
// }
// private getNextQueue = async () => {
// const params = {
// QueueUrl: this.queueUrl,
// MaxNumberOfMessages: 1,
// VisibilityTimeout: 30,
// WaitTimeSeconds: 20, // Increase the WaitTimeSeconds for long polling
// };
// try {
// const data = await this.sqs.receiveMessage(params).promise();
// const message = data.Messages ? data.Messages[0] : null;
// return message;
// } catch (error) {
// console.error("Error retrieving message from SQS:", error);
// return null;
// }
// }
// private deleteMessage = async (receiptHandle: string) => {
// const params = {
// QueueUrl: this.queueUrl,
// ReceiptHandle: receiptHandle,
// };
// try {
// await this.sqs.deleteMessage(params).promise();
// console.log("Message deleted");
// } catch (error) {
// console.error("Error deleting message from SQS:", error);
// }
// }
// }

View file

@ -1,4 +1,4 @@
import { ObjectId } from 'mongodb';
import { ObjectId } from "mongodb";
export interface ITodo {
_id: ObjectId;
@ -8,4 +8,4 @@ export interface ITodo {
createAt: Date;
updateAt: Date;
status: string;
}
}

View file

@ -1,7 +1,7 @@
import { MongoClient, ObjectId } from 'mongodb';
import { ITodo } from '../interfaces/ITodo';
import { MongoClient, ObjectId } from "mongodb";
import { ITodo } from "../interfaces/ITodo";
const env = require('dotenv').config().parsed;
const env = require("dotenv").config().parsed;
export class MongoDbModel {
client: MongoClient;
@ -14,10 +14,10 @@ export class MongoDbModel {
try {
await this.client.connect();
const db = this.client.db(env.MONGO_DB_NAME);
const todosCollection = db.collection('todos');
const todosCollection = db.collection("todos");
const result = await todosCollection.updateOne(
{ _id: new ObjectId(todo._id) },
{ $set: { status: 'completed' } }
{ $set: { status: "completed" } }
);
console.log(`Updated status of Todo ${todo._id} to completed`);
} catch (error) {
@ -25,5 +25,5 @@ export class MongoDbModel {
} finally {
await this.client.close();
}
}
}
};
}

View file

@ -1,8 +1,8 @@
import amqp, { ConsumeMessage } from 'amqplib';
import { ITodo } from '../interfaces/ITodo';
import { MongoDbModel } from '../mongodb/MongoDb';
import amqp, { ConsumeMessage } from "amqplib";
import { ITodo } from "../interfaces/ITodo";
import { MongoDbModel } from "../mongodb/MongoDb";
const env = require('dotenv').config().parsed;
const env = require("dotenv").config().parsed;
export class RabbitMQ {
channel: amqp.Channel;
@ -15,17 +15,19 @@ export class RabbitMQ {
this.mongoClient = new MongoDbModel();
this.connect().then(() => {
console.log('RabbitMQ connected');
}).catch((error) => {
console.error('Error connecting to RabbitMQ:', error);
});
this.connect()
.then(() => {
console.log("RabbitMQ connected");
})
.catch((error) => {
console.error("Error connecting to RabbitMQ:", error);
});
}
async connect() {
try {
const connection = await amqp.connect({
protocol: 'amqp',
protocol: "amqp",
hostname: env.RABBITMQ_HOST,
port: parseInt(env.RABBITMQ_PORT),
username: env.RABBITMQ_USERNAME,
@ -34,22 +36,29 @@ export class RabbitMQ {
this.channel = await connection.createChannel();
await this.channel.assertQueue(this.queueName);
console.log('Channel and queue asserted successfully #####');
console.log("Channel and queue asserted successfully #####");
} catch (error) {
console.error('Error connecting to RabbitMQ:', error);
console.error("Error connecting to RabbitMQ:", error);
throw error;
}
}
async create(payload: ITodo, delayTimeForQueue: number) {
const message = JSON.stringify({ payload, delayTimeForQueue });
const options = { persistent: true, expiration: delayTimeForQueue.toString() };
const options = {
persistent: true,
expiration: delayTimeForQueue.toString(),
};
try {
await this.channel.sendToQueue(this.queueName, Buffer.from(message), options);
console.log('Message sent to the queue');
await this.channel.sendToQueue(
this.queueName,
Buffer.from(message),
options
);
console.log("Message sent to the queue");
} catch (error) {
console.error('Error sending message to RabbitMQ:', error);
console.error("Error sending message to RabbitMQ:", error);
throw error;
}
}
@ -61,13 +70,12 @@ export class RabbitMQ {
if (message) {
const todo = JSON.parse(message.content.toString());
this.channel.ack(message);
console.log('Received notification:', todo);
console.log("Received notification:", todo);
}
});
}
// try {
// try {
// console.log('Consumer started, waiting for messages...');
// this.channel.consume(this.queueName, async (message) => {
// if (message) {
@ -94,4 +102,4 @@ export class RabbitMQ {
// } catch (error) {
// console.error('Error consuming messages from RabbitMQ:', error);
// }
}
}

View file

@ -1,50 +0,0 @@
import aws from 'aws-sdk';
import { ITodo } from '../schemas/todoSchema';
const env = require('dotenv').config().parsed;
export class Sqs {
sqs: aws.SQS;
queueUrl: string;
constructor() {
aws.config.update({
accessKeyId: env.AWS_ACCESS_KEY_ID,
secretAccessKey: env.AWS_SECRET_ACCESS_KEY,
});
this.sqs = new aws.SQS();
this.queueUrl = env.AWS_SQS_URL + env.AWS_SQS_QUEUE_NAME;
if (this.sqs) {
console.log("SQS connected")
}
}
async create(payload: ITodo, delayTimeForQueue: number) {
let reQueueTime = 0;
if (delayTimeForQueue > 900) {
reQueueTime = delayTimeForQueue - 900;
delayTimeForQueue = 900;
}
const params = {
DelaySeconds: delayTimeForQueue,
MessageAttributes: {},
MessageBody: JSON.stringify({ payload, delayTimeForQueue, reQueueTime }),
QueueUrl: this.queueUrl,
};
return new Promise((resolve, reject) => {
this.sqs.sendMessage(params, (err, data) => {
if (err) {
console.log("Error", err);
reject(err);
} else {
console.log("Message sent to the queue", data.MessageId)
resolve(data);
}
});
});
}
}

View file

@ -1,11 +1,11 @@
import { NextFunction, Request, Response } from 'express';
import { ApiError } from '../utils/ApiError';
import { ITodo } from '../schemas/todoSchema';
import { TodoModel } from '../models/todoModel';
import { NextFunction, Request, Response } from "express";
import { ApiError } from "../utils/ApiError";
import { ITodo } from "../schemas/todoSchema";
import { TodoModel } from "../models/todoModel";
// import { Sqs } from '../aws/Sqs';
import { RabbitMQ } from '../rabbitmq/RabbitMQ';
import { RabbitMQ } from "../rabbitmq/RabbitMQ";
const env = require('dotenv').config().parsed;
const env = require("dotenv").config().parsed;
export class TodoController {
private todoModel: TodoModel;
@ -23,7 +23,7 @@ export class TodoController {
}
return res.json(todos);
}
};
public getOne = async (req: Request, res: Response, next: NextFunction) => {
const todo: ITodo | ApiError = await this.todoModel.findOne(req.params.id);
@ -31,9 +31,13 @@ export class TodoController {
return next(todo);
}
return res.json(todo);
}
};
public createOne = async (req: Request, res: Response, next: NextFunction) => {
public createOne = async (
req: Request,
res: Response,
next: NextFunction
) => {
try {
const todo: ITodo | ApiError = await this.todoModel.create(req.body);
if (todo instanceof ApiError) {
@ -43,31 +47,50 @@ export class TodoController {
return res.json(todo);
} catch {
const err = new ApiError('Internal server error', 500, 'Internal Server Error');
next(err)
const err = new ApiError(
"Internal server error",
500,
"Internal Server Error"
);
next(err);
}
}
};
public updateOne = async (req: Request, res: Response, next: NextFunction) => {
const todo: ITodo | ApiError = await this.todoModel.update(req.body, req.params.id);
public updateOne = async (
req: Request,
res: Response,
next: NextFunction
) => {
const todo: ITodo | ApiError = await this.todoModel.update(
req.body,
req.params.id
);
if (todo instanceof ApiError) {
return next(todo);
}
return res.json(todo);
}
};
public deleteOne = async (req: Request, res: Response, next: NextFunction) => {
public deleteOne = async (
req: Request,
res: Response,
next: NextFunction
) => {
const { id } = req.params;
const todo: boolean | ApiError = await this.todoModel.remove(id);
if (!todo) {
const error = new ApiError('Todo not found', 404, 'Not Found');
const error = new ApiError("Todo not found", 404, "Not Found");
return next(error);
}
return res.json(todo);
}
};
public removeAll = async (req: Request, res: Response, next: NextFunction) => {
public removeAll = async (
req: Request,
res: Response,
next: NextFunction
) => {
const todos: boolean | ApiError = await this.todoModel.removeAll();
return res.json(todos);
}
};
}

View file

@ -1,11 +1,10 @@
import express from 'express';
import mongoose from 'mongoose';
import todoRouter from './routes/todoRouter';
import express from "express";
import mongoose from "mongoose";
import todoRouter from "./routes/todoRouter";
import { ApiError } from './utils/ApiError';
import { ApiError } from "./utils/ApiError";
const env = require('dotenv').config().parsed;
const env = require("dotenv").config().parsed;
class TodoApp {
app: express.Application;
@ -22,10 +21,10 @@ class TodoApp {
}
private setRoutes() {
this.app.use('/todo', todoRouter);
this.app.all('*', (req, res, next) => {
const error = new ApiError('Are you lost?', 404, 'Not Found');
next(error)
this.app.use("/todo", todoRouter);
this.app.all("*", (req, res, next) => {
const error = new ApiError("Are you lost?", 404, "Not Found");
next(error);
});
}
@ -44,12 +43,12 @@ class TodoApp {
mongoose.connect(env.DATABASE_URL);
const db = mongoose.connection;
// Check for DB connection
db.on('error', () => {
console.error.bind(console, 'MongoDB connection error:')
db.on("error", () => {
console.error.bind(console, "MongoDB connection error:");
process.exit(1);
});
db.once('open', () => {
console.log('Connected to MongoDB');
db.once("open", () => {
console.log("Connected to MongoDB");
});
}
}

View file

@ -1,40 +1,52 @@
import { Request, Response, NextFunction } from 'express';
import { ApiError } from '../utils/ApiError';
import { DateTime } from 'luxon';
import moment from 'moment-timezone';
import { Request, Response, NextFunction } from "express";
import { ApiError } from "../utils/ApiError";
import { DateTime } from "luxon";
import moment from "moment-timezone";
const createTodoMiddleWare = async (req: Request, res: Response, next: NextFunction) => {
const createTodoMiddleWare = async (
req: Request,
res: Response,
next: NextFunction
) => {
const { title, description, due_date } = req.body;
if (!title || !due_date) {
const error = new ApiError(`${!title ? 'title' : 'due_date'} is required`, 400, 'Bad Request');
const error = new ApiError(
`${!title ? "title" : "due_date"} is required`,
400,
"Bad Request"
);
return next(error);
}
try {
if(new Date(due_date) < new Date()) {
const error = new ApiError(`due_date must be greater than current date`, 400, 'Bad Request');
if (new Date(due_date) < new Date()) {
const error = new ApiError(
`due_date must be greater than current date`,
400,
"Bad Request"
);
return next(error);
}
}
} catch {
const error = new ApiError(`due_date must be ISO `, 400, 'Bad Request');
const error = new ApiError(`due_date must be ISO `, 400, "Bad Request");
return next(error);
}
if (!description) {
req.body.description = '';
req.body.description = "";
}
//check if date is valid, this is valid: 2023-07-08T14:00:00.000Z
const date = DateTime.fromISO(due_date);
if (!date.isValid) {
const error = new ApiError(`due_date must be a valid date Format`, 400, 'Bad Request');
const error = new ApiError(
`due_date must be a valid date Format`,
400,
"Bad Request"
);
return next(error);
}
next();
}
};
export {
createTodoMiddleWare
}
export { createTodoMiddleWare };

View file

@ -1,66 +1,77 @@
import { Todo, ITodo } from '../schemas/todoSchema';
import { ApiError } from '../utils/ApiError';
import { Todo, ITodo } from "../schemas/todoSchema";
import { ApiError } from "../utils/ApiError";
interface Todo {
title: string,
description: string,
due_date: Date
title: string;
description: string;
due_date: Date;
}
export class TodoModel {
public findAll = async (): Promise<ITodo[]> => {
const todos = await Todo.find();
return todos;
}
};
public findOne = async (_id: string): Promise<ITodo | ApiError> => {
try {
const todo = await Todo.findById(_id);
if (!todo) {
const error = new ApiError('Todo not found', 404, 'Not Found');
const error = new ApiError("Todo not found", 404, "Not Found");
return error;
}
return todo;
} catch {
const error = new ApiError('Internal server error', 500, 'Internal Server Error');
const error = new ApiError(
"Internal server error",
500,
"Internal Server Error"
);
return error;
}
}
};
public create = async (params: Todo): Promise<ITodo | ApiError> => {
let todo = new Todo(params);
todo = await todo.save();
if (!todo) {
const error = new ApiError('Internal server error', 500, 'Internal Server Error');
const error = new ApiError(
"Internal server error",
500,
"Internal Server Error"
);
return error;
}
return todo;
}
};
public update = async (params: Todo, _id: string): Promise<ITodo | ApiError> => {
public update = async (
params: Todo,
_id: string
): Promise<ITodo | ApiError> => {
const todo = await Todo.findOne({ _id });
if (!todo) {
const error = new ApiError('Todo not found', 404, 'Not Found');
const error = new ApiError("Todo not found", 404, "Not Found");
return error;
}
todo.title = params.title;
todo.description = params.description;
todo.due_date = params.due_date;
await todo.save();
return todo;
}
};
public remove = async (_id: string): Promise<boolean | ApiError> => {
const result = await Todo.deleteOne({ _id });
return result.deletedCount > 0 ? true : new ApiError('Todo not found', 404, 'Not Found');
}
return result.deletedCount > 0
? true
: new ApiError("Todo not found", 404, "Not Found");
};
public removeAll = async (): Promise<boolean> => {
await Todo.deleteMany({});
return true;
}
};
}

View file

@ -1,7 +1,7 @@
import amqp, { Options, ConsumeMessage, Channel } from 'amqplib';
import { ITodo } from '../schemas/todoSchema';
import amqp, { Options, ConsumeMessage, Channel } from "amqplib";
import { ITodo } from "../schemas/todoSchema";
const env = require('dotenv').config().parsed;
const env = require("dotenv").config().parsed;
export class RabbitMQ {
connection: amqp.Connection;
@ -9,22 +9,23 @@ export class RabbitMQ {
queue: string;
exchange: string;
constructor() {
this.queue = env.RABBITMQ_QUEUE_NAME;
this.exchange = 'delayed_exchange';
this.exchange = "delayed_exchange";
this.connect().then(() => {
console.log('RabbitMQ connected');
}).catch((error) => {
console.error('Error connecting to RabbitMQ:', error);
});
this.connect()
.then(() => {
console.log("RabbitMQ connected");
})
.catch((error) => {
console.error("Error connecting to RabbitMQ:", error);
});
}
async connect() {
try {
this.connection = await amqp.connect({
protocol: 'amqp',
protocol: "amqp",
hostname: env.RABBITMQ_HOST,
port: parseInt(env.RABBITMQ_PORT),
username: env.RABBITMQ_USERNAME,
@ -33,38 +34,41 @@ export class RabbitMQ {
this.channel = await this.connection.createChannel();
await this.channel.assertQueue(this.queue);
await this.channel.assertExchange(this.exchange, 'x-delayed-message', {
await this.channel.assertExchange(this.exchange, "x-delayed-message", {
durable: true,
arguments: {
'x-delayed-type': 'direct'
}
"x-delayed-type": "direct",
},
});
await this.channel.bindQueue(this.queue, this.exchange, this.queue);
console.log('Channel and queue asserted successfully #####');
console.log("Channel and queue asserted successfully #####");
} catch (error) {
console.error('Error connecting to RabbitMQ:', error);
console.error("Error connecting to RabbitMQ:", error);
throw error;
}
}
async create(payload: ITodo) {
const delayTimeForQueue = this.calculateDelayTimeForQueue(payload);
console.log("The Queue will be delayed for: ", delayTimeForQueue, " ms");
const message = JSON.stringify({ payload });
const options: Options.Publish = { headers: { 'x-delay': delayTimeForQueue } };
const options: Options.Publish = {
headers: { "x-delay": delayTimeForQueue },
};
try {
await this.channel.publish(this.exchange, this.queue, Buffer.from(message),
options
)
console.log(`Queue name is: ${this.queue}`)
await this.channel.publish(
this.exchange,
this.queue,
Buffer.from(message),
options
);
console.log(`Queue name is: ${this.queue}`);
} catch (error) {
console.error('Error sending message to RabbitMQ:', error);
console.error("Error sending message to RabbitMQ:", error);
throw error;
}
}
calculateDelayTimeForQueue(payload: ITodo) {

View file

@ -1,6 +1,6 @@
import { Router } from 'express';
import { TodoController } from '../controllers/todoController';
import { createTodoMiddleWare } from '../middleware/createTodoMiddleWare';
import { Router } from "express";
import { TodoController } from "../controllers/todoController";
import { createTodoMiddleWare } from "../middleware/createTodoMiddleWare";
class TodoRouter {
router: Router;
@ -13,19 +13,21 @@ class TodoRouter {
}
private setRoutes() {
this.router.get('/', this.todoController.getAll);
this.router.get('/:id', this.todoController.getOne);
this.router.post('/', createTodoMiddleWare, this.todoController.createOne);
this.router.put('/:id', createTodoMiddleWare, this.todoController.updateOne);
this.router.delete('/:id', this.todoController.deleteOne);
this.router.delete('/', this.todoController.removeAll)
this.router.get("/", this.todoController.getAll);
this.router.get("/:id", this.todoController.getOne);
this.router.post("/", createTodoMiddleWare, this.todoController.createOne);
this.router.put(
"/:id",
createTodoMiddleWare,
this.todoController.updateOne
);
this.router.delete("/:id", this.todoController.deleteOne);
this.router.delete("/", this.todoController.removeAll);
}
public getRouter() {
return this.router;
}
}
export default new TodoRouter().getRouter();

View file

@ -1,4 +1,4 @@
import mongoose, { Schema, Document } from 'mongoose';
import mongoose, { Schema, Document } from "mongoose";
interface ITodo extends Document {
title: string;
@ -11,13 +11,13 @@ interface ITodo extends Document {
const TodoSchema: Schema<ITodo> = new Schema({
title: { type: String, required: true },
description: { type: String, default: '' },
description: { type: String, default: "" },
due_date: { type: Date, required: true },
createAt: { type: Date, default: Date.now },
updateAt: { type: Date, default: Date.now },
status: { type: String, enum: ['pending', 'completed'], default: 'pending' },
status: { type: String, enum: ["pending", "completed"], default: "pending" },
});
const Todo = mongoose.model<ITodo>('Todo', TodoSchema, 'todos');
const Todo = mongoose.model<ITodo>("Todo", TodoSchema, "todos");
export { Todo, ITodo };

View file

@ -1,7 +1,11 @@
class ApiError extends Error {
constructor(message: string, private statusCode: number = 500,private status: string = 'Internal Server Error') {
constructor(
message: string,
private statusCode: number = 500,
private status: string = "Internal Server Error"
) {
super(message);
}
}
export { ApiError };
export { ApiError };