feat(CronJob|SQS): adds a cronjob to listen SQS and insert person payload into queue

Author: douglasfdev

src/api/modules/v1/components/Commands/Slinger.ts

@@ -2,8 +2,9 @@ import { FileService } from '$service/file/FileService.ts';
 import { PersonService } from "$service/person/PersonService.ts";
 import {
   everyMinute,
-  every15Minute,
-  start
+  hourly,
+  cron,
+  start,
 } from '$deps';
 
 export class Slinger {
@@ -12,21 +13,23 @@ export class Slinger {
   }
 
   private init() {
-    // this.handleDisareByMinute();
-    this.handleDispareByFiftyMinute();
+    this.dispareFiles();
+    this.disparePersonIntoQueue();
+    this.dispareFilesPayloadIntoQueue();
     start();
+    this.init();
+  }
+
+  private dispareFiles() {
+    cron("* * * * * *", async () => await new FileService().listenFilesFromDB());
   }
 
-  private handleDisareByMinute() {
-    everyMinute(async () => {
-      await new FileService().listenFiles();
-    });
+  private disparePersonIntoQueue() {
+    cron("* * * * * *", async () => await new PersonService().listenAndInsertPersonFromQueue());
   }
 
-  private handleDispareByFiftyMinute() {
-    everyMinute(async () => {
-      await new PersonService().listenAndInsertPersonFromQueue();
-    });
+  private dispareFilesPayloadIntoQueue() {
+    cron("* * * * * *", async () => await new FileService().handlerPersonFromObjectIntoSQS());
   }
 }
 

src/api/modules/v1/repository/File/FileRespository.ts

@@ -55,6 +55,27 @@ class FileRepository extends File {
     );
   }
 
+  public async pendingFilesByName(name: string) {
+    return this.foundPendingFileByName(name);
+  }
+
+  private async foundPendingFileByName(name: string): Promise<Array<Partial<IFileDTO>>> {
+    return this.mysql.buildQuery(
+      `
+        SELECT
+          id,
+          name,
+          CASE
+            WHEN status = 0 THEN 'PENDING'
+            WHEN status = 1 THEN 'LAUNCHED'
+          END as status
+        FROM ${this.table}
+        WHERE name = ?
+        AND status = ?;
+      `, [name, FilenameEnum.PENDING]
+    )
+  }
+
   public updatedAfterListenAll(id: string) {
     return this.updateStatusAfterListenAll(id);
   }

src/api/modules/v1/services/file/FileService.ts

@@ -4,18 +4,28 @@ import {
   FilenameEnum,
   IFileDTO,
   IFileService,
+  log,
 } from "$common";
 import { FileRespository } from "$repositories";
 import { PersonService } from "$service/person/PersonService.ts";
+import { IObjectS3DTO, IPersonDTO } from '$interfaces';
+import { SimpleQueueService } from "$component/AWS/sqs.component.ts";
 
 export class FileService implements IFileService {
-  private s3: typeof S3;
   private fileRepository: typeof FileRespository;
-  private personService: PersonService = new PersonService()
+  private personService: PersonService;
+  private s3: typeof S3;
+  private sQs: SimpleQueueService;
+  private status = {
+    "PENDING": "PENDING",
+    "LAUNCHED": "LAUNCHED"
+  };
 
   constructor() {
     this.s3 = S3;
+    this.sQs = new SimpleQueueService();
     this.fileRepository = FileRespository;
+    this.personService = new PersonService();
   }
 
   public async handlerFilesPerson(files: Array<FormDataFile>): Promise<string> {
@@ -57,40 +67,76 @@ export class FileService implements IFileService {
     return this.fileRepository.list();
   }
 
-  public async listenFiles() {
-    const files = await this.fileRepository.pendingFiles();
+  public async listenFilesFromDB() {
+    const filesPending = await this.fileRepository.pendingFiles();
 
-    if (!files) {
+    if (!filesPending) {
       return;
     }
 
-    const pendings = files.filter(
-      file => file.status === FilenameEnum.PENDING
-    );
+    for (const filePending of filesPending) {
+      await this.personService.listenAndCreatePerson(filePending.name);
+      await this.fileRepository.updatedAfterListenAll(filePending.id)
 
-    for (const pending of pendings) {
-      if (pending.status === FilenameEnum.PENDING) {
-        this.personService.listenAndCreatePerson(pending.name);
-        await this.fileRepository.updatedAfterListenAll(pending.id)
-        continue;
-      };
+      continue;
     }
 
-    return files;
+    return filesPending;
   }
 
-  public async listAllObjectsFromBucket() {
+  public async listAllObjectsFromBucket(): Promise<Array<IObjectS3DTO>> {
     const objects = await this.s3.listObjects() as Array<_Object>;
 
-    const objectsFiltered = objects.map(object => {
-        return {
-          name: object.Key,
-          size: object.Size,
-        }
+    const objectsFiltered: Array<IObjectS3DTO> = objects.map(object => {
+      return {
+        name: object.Key,
+        size: object.Size,
+      }
     });
 
     return objectsFiltered;
   }
+
+  public async handlerPersonFromObjectIntoSQS() {
+    const objects = await this.listAllObjectsFromBucket();
+
+    const personList: Partial<IPersonDTO>[] = []
+
+    for (const object of objects) {
+      const pendingsFiles = await this.fileRepository.pendingFilesByName(object.name as string);
+
+      if (!pendingsFiles) {
+        return;
+      }
+
+      pendingsFiles.forEach(async file => {
+        if (String(file.status) !== this.status.PENDING) {
+          log.success("Done process with success!");
+          return;
+        }
+
+        const getS3Object = await this.s3.readFileFromS3(object.name as string) as Array<string>;
+
+        for (const result of getS3Object) {
+          const [name, age, sex, size, weight] = result.split(',');
+          const person: Partial<IPersonDTO> = {
+            name,
+            age: Number(age),
+            sex,
+            size: Number(size),
+            weight: Number(weight)
+          }
+
+          await this.sQs.handleQueue(person);
+          continue;
+        }
+
+        await this.fileRepository.updatedAfterListenAll(file.id as string);
+      });
+    }
+
+    return personList;
+  }
 }
 
 export default new FileService()

src/api/modules/v1/services/person/PersonService.ts

@@ -114,11 +114,15 @@ export class PersonService {
       const personObject: Partial<IPersonDTO> = JSON.parse(payload);
 
       await this.personRepository.createPersons(personObject);
-      await this.fileRepository.updatedAfterListenAll();
+
       person.push(personObject);
       continue;
     }
 
+    files.forEach(
+      async file => await this.fileRepository.updatedAfterListenAll(file.id as string)
+    )
+
     return person;
   }
 }

src/common/interfaces/IObjectS3DTO.ts

@@ -0,0 +1,4 @@
+export interface IObjectS3DTO {
+  name: string | null | undefined;
+  size: number | null | undefined;
+}

src/common/interfaces/IPerson.ts

@@ -1,7 +1,7 @@
 import { PersonEnum } from "../enums/index.ts";
 
 export interface IPersonDTO {
-  id: string;
+  id?: string;
   name: string;
   age: number;
   sex: string;

src/common/interfaces/index.ts

@@ -2,3 +2,4 @@ export * from "./IPerson.ts";
 export * from "./IFleService.ts";
 export * from "./IFIleController.ts";
 export * from './IFileDTO.ts';
+export * from './IObjectS3DTO.ts';