Merge pull request #10 from lvchkn/refactoring-1.6

Refactoring - v1.6

README.md

@@ -4,18 +4,20 @@
 
 ### How to run locally
 
-Create .env file in the app's root directory (use .env.example as a reference).
+1. Create .env file in the app's root directory (use .env.example as a reference).
 
-Start the compose stack:
+1. Start the compose stack:
 
-```bash
- docker compose up
-```
+   ```bash
+   docker compose up
+   ```
 
-Then run `npm i` and either of the following options:
+1. Run the following commands:
 
-- `npm run local` - run with ts-node
-- `npm run local:watch` - run in watch mode with nodemon and ts-node.
+   ```bash
+   npm i
+   npm run start
+   ```
 
 #### Run tests:
 

db/mongo.ts

@@ -12,12 +12,30 @@ export async function connectToMongoDb() {
   try {
     await client.connect();
     console.log("Successfully connected to Mongo!");
+    await createIndexes();
   } catch (error) {
     console.error("Error occurred while connecting to Mongo!", error);
     await client.close();
   }
 }
 
+async function createIndexes(): Promise<void> {
+  const collectionName = "releases";
+  const db = client.db("rymdata");
+
+  const options = {
+    collation: {
+      locale: "en",
+      strength: 2,
+    },
+  };
+
+  await db.collection(collectionName).createIndex({ artist: 1 }, options);
+  await db.collection(collectionName).createIndex({ album: 1 }, options);
+  await db.collection(collectionName).createIndex({ genres: 1 }, options);
+  await db.collection(collectionName).createIndex({ year: 1 }, options);
+}
+
 export function getClient() {
   return client;
 }

db/releasesReadRepo.ts

@@ -1,50 +1,53 @@
 import { Filter, WithId } from "mongodb";
-import { Release } from "../parser.js";
-import { FilterOptions } from "../routers/releasesRouter.js";
 import { getClient } from "./mongo.js";
+import { Release } from "../models/release.js";
+import { Query } from "../services/queryProcessor.js";
 
 const client = getClient();
 const db = client.db("rymdata");
 const releasesCollectionName = "releases";
 
-function parseFilters(filterOptions: FilterOptions): Filter<Release> {
+function parseFilters(query: Query): Filter<Release> {
+  const { artists, albums, genres, years } = query;
   const filter: Filter<Release> = {};
 
-  if (filterOptions.artists) {
+  if (artists.length > 0) {
     filter.artist = {
-      $in: filterOptions.artists,
+      $in: artists,
     };
   }
 
-  if (filterOptions.albums) {
+  if (albums.length > 0) {
     filter.album = {
-      $in: filterOptions.albums,
+      $in: albums,
     };
   }
 
-  if (filterOptions.genres) {
+  if (genres.length > 0) {
     filter.genres = {
-      $all: filterOptions.genres,
+      $all: genres,
     };
   }
 
-  if (filterOptions.years) {
+  if (years.length > 0) {
     filter.year = {
-      $in: filterOptions.years,
+      $in: years,
     };
   }
 
   return filter;
 }
 
-export async function getAllReleases(
-  filterOptions: FilterOptions
-): Promise<WithId<Release>[]> {
-  const filters = parseFilters(filterOptions);
+export async function getAllReleases(query: Query): Promise<WithId<Release>[]> {
+  const filters = parseFilters(query);
 
   const releasesCursor = db
     .collection<Release>(releasesCollectionName)
-    .find(filters);
+    .find(filters)
+    .collation({
+      locale: "en",
+      strength: 2,
+    });
   const releases = releasesCursor.toArray();
 
   return releases;

db/releasesWriteRepo.ts

@@ -3,8 +3,8 @@ import {
   BulkWriteResult,
   MongoBulkWriteError,
 } from "mongodb";
-import { Release } from "../parser.js";
 import { getClient } from "./mongo.js";
+import { Release } from "../models/release.js";
 
 const client = getClient();
 const db = client.db("rymdata");

db/tasksRepo.ts

@@ -1,15 +1,11 @@
 import { WithId } from "mongodb";
 import { getClient } from "./mongo.js";
+import { Task } from "../models/task.js";
 
 const client = getClient();
 const db = client.db("rymdata");
 const tasksCollectionName = "tasks";
 
-export interface Task {
-  id: string;
-  status: string;
-}
-
 export async function getTask(taskId: string): Promise<WithId<Task>> {
   const task = await db
     .collection<Task>(tasksCollectionName)

models/parseRequest.ts

@@ -0,0 +1,7 @@
+export interface ParseRequest {
+  profile: string;
+  tag: string;
+  fromPage: number;
+  toPage: number;
+  isTest?: boolean;
+}

models/release.ts

@@ -0,0 +1,6 @@
+export interface Release {
+  artist: string;
+  album: string;
+  genres: string[];
+  year: number;
+}

models/task.ts

@@ -0,0 +1,6 @@
+export type TaskStatus = "Pending" | "Completed";
+
+export interface Task {
+  id: string;
+  status: TaskStatus;
+}

mq/consumer.ts

@@ -1,51 +1,50 @@
-import { ConsumeMessage } from "amqplib";
-import { Task, upsertTask } from "../db/tasksRepo.js";
-import { parseAndSave } from "../parser.js";
+import { Channel, Connection, ConsumeMessage } from "amqplib";
+import { upsertTask } from "../db/tasksRepo.js";
 import { connectToRabbitMq } from "./amqp.js";
+import { save } from "../services/dataSaver.js";
+import { Task } from "../models/task.js";
 
-export async function startConsumer(): Promise<void> {
-  const connection = await connectToRabbitMq();
-  const channel = await connection.createChannel();
+let channel: Channel;
+let connection: Connection;
 
-  process.once("SIGINT", async () => {
-    await channel.close();
-    await connection.close();
-  });
+process.once("SIGINT", async () => {
+  await channel.close();
+  await connection.close();
+});
+
+export async function startConsumer(): Promise<void> {
+  connection = await connectToRabbitMq();
+  channel = await connection.createChannel();
 
   const { queue } = await channel.assertQueue("parse-tasks-queue", {
     durable: true,
   });
 
   await channel.prefetch(2);
+  await channel.consume(queue, onMessageHandler, { noAck: false });
+}
+
+async function onMessageHandler(message: ConsumeMessage | null): Promise<void> {
+  if (message === null) {
+    console.log("Error while receiving the message");
+    return;
+  }
+
+  const json = message.content.toString();
+  console.log("Message received!");
+
+  const parseRequest = JSON.parse(json);
+  const result = await save(parseRequest);
+
+  console.log(JSON.stringify(result));
+
+  const task: Task = {
+    id: message.properties.messageId,
+    status: "Completed",
+  };
+
+  const upsertTaskResult = await upsertTask(task);
+  console.log("upsertTaskResult completed:", JSON.stringify(upsertTaskResult));
 
-  await channel.consume(
-    queue,
-    async (message: ConsumeMessage | null) => {
-      if (message !== null) {
-        const json = message.content.toString();
-        console.log("Message received!");
-
-        const parseRequest = JSON.parse(json);
-        const result = await parseAndSave(parseRequest);
-
-        console.log(JSON.stringify(result));
-
-        const task: Task = {
-          id: message.properties.messageId,
-          status: "Completed",
-        };
-
-        const upsertTaskResult = await upsertTask(task);
-        console.log(
-          "upsertTaskResult completed:",
-          JSON.stringify(upsertTaskResult)
-        );
-
-        channel.ack(message);
-      } else {
-        console.log("Error while receiving the message");
-      }
-    },
-    { noAck: false }
-  );
+  channel.ack(message);
 }

mq/publisher.ts

@@ -2,7 +2,8 @@ import { Connection } from "amqplib";
 import { v4 as uuidv4 } from "uuid";
 import { exit } from "process";
 import { connectToRabbitMq } from "./amqp.js";
-import { Task, upsertTask } from "../db/tasksRepo.js";
+import { upsertTask } from "../db/tasksRepo.js";
+import { Task } from "../models/task.js";
 
 let connection: Connection | null = null;
 
@@ -32,7 +33,7 @@ export async function publishMessage(message) {
 
     const task: Task = {
       id: uuid,
-      status: "pending",
+      status: "Pending",
     };
 
     const upsertTaskResult = await upsertTask(task);