Merge pull request #954 from funmusicplace/various-fixes

feat: queue and job for moving files from minio to backblaze

client/src/components/Admin/CallServerTasks.tsx

@@ -48,6 +48,7 @@ export const Admin: React.FC = () => {
         <label>{t("whatTaskCall")}</label>
         <SelectEl {...methods.register("jobName")}>
           <option value="cleanUpFiles">cleanUpFiles</option>
+          <option value="moveBucketToBackblaze">moveBucketToBackblaze</option>
           <option value="initiateUserNotifications">
             initiateUserNotifications
           </option>

src/jobs/optimize-image.ts

@@ -21,7 +21,7 @@ const { defaultOptions, config: sharpConfig } = tempSharpConfig;
 
 const { SIGHTENGINE_USER, SIGHTENGINE_SECRET } = process.env;
 
-const sleep = (ms: number) =>
+export const sleep = (ms: number) =>
   new Promise((resolve) => {
     return setTimeout(resolve, ms);
   });

src/jobs/queue-worker.ts

@@ -17,6 +17,10 @@ import sendMail from "./send-mail";
 import "../queues/send-mail-queue";
 
 import { REDIS_CONFIG } from "../config/redis";
+import {
+  moveFilesToBackblazeJob,
+  moveFilesToBackBlazeQueue,
+} from "../queues/moving-files-to-backblaze";
 
 export const logger = winston.createLogger({
   level: "info",
@@ -49,6 +53,7 @@ yargs // eslint-disable-line
     generateAlbumQueueWorker();
     sendMailQueue();
     cleanUpFilesQueue();
+    moveFilesToBackBlazeWorker();
   })
   .help().argv;
 
@@ -145,6 +150,26 @@ export async function cleanUpFilesQueue() {
     cleanUpOldFilesJob,
     workerOptions
   );
+  logger.info("Move Files From Minio to BackBlaze worker started");
+
+  worker.on("completed", (job: any) => {
+    logger.info("completed:clean-up-old-files");
+  });
+
+  worker.on("failed", (job: any, err: any) => {
+    logger.error("failed:clean-up-old-files", err);
+  });
+
+  worker.on("error", (err: any) => {
+    logger.error("error:clean-up-old-files", err);
+  });
+}
+export async function moveFilesToBackBlazeWorker() {
+  const worker = new Worker(
+    "move-file-to-backblaze",
+    moveFilesToBackblazeJob,
+    workerOptions
+  );
   logger.info("Clean Up Files worker started");
 
   worker.on("completed", (job: any) => {

src/queues/moving-files-to-backblaze.ts

@@ -0,0 +1,87 @@
+import { Job, Queue, QueueEvents } from "bullmq";
+import { REDIS_CONFIG } from "../config/redis";
+import { promises as fsPromises } from "fs";
+import logger from "../logger";
+import {
+  getBufferFromMinio,
+  getObjectList,
+  minioClient,
+  removeObjectsFromBucket,
+  trackGroupFormatBucket,
+  uploadWrapper,
+} from "../utils/minio";
+import { sleep } from "../jobs/optimize-image";
+
+const queueOptions = {
+  prefix: "mirlo",
+  connection: REDIS_CONFIG,
+};
+
+export const moveFilesToBackBlazeQueue = new Queue(
+  "move-file-to-backblaze",
+  queueOptions
+);
+
+export const moveFilesToBackBlazeQueueEvents = new QueueEvents(
+  "move-file-to-backblaze",
+  queueOptions
+);
+
+moveFilesToBackBlazeQueueEvents.on(
+  "completed",
+  async (result: { jobId: string; returnvalue?: any }) => {
+    logger.info(
+      `Job with id ${JSON.stringify(
+        result.jobId
+      )} has been completed, ${JSON.stringify(result.returnvalue)}`
+    );
+  }
+);
+
+moveFilesToBackBlazeQueueEvents.on(
+  "stalled",
+  async (result: { jobId: string }) => {
+    logger.info(`jobId ${result.jobId} stalled: that's a bummer`);
+  }
+);
+
+moveFilesToBackBlazeQueueEvents.on("error", async (error) => {
+  logger.error(`jobId ${JSON.stringify(error)} had an error`);
+});
+
+export const startMovingFiles = async (bucketName: string) => {
+  const files = await getObjectList(bucketName, "");
+  files.map(async (file, i) => {
+    try {
+      console.log("adding to queue", file.name);
+      moveFilesToBackBlazeQueue.add(
+        "move-file-to-backblaze",
+        {
+          bucketName,
+          fileName: file.name,
+        },
+        {
+          delay: i * 1500,
+        }
+      );
+    } catch (e) {
+      console.error(e);
+      logger.error("Error adding to queue");
+    }
+  });
+};
+
+export const moveFilesToBackblazeJob = async (job: Job) => {
+  const bucketName = job.data.bucketName;
+  const fileName = job.data.fileName;
+
+  logger.info(`moving file: bucket: ${bucketName}, ${fileName}`);
+  try {
+    const stream = await minioClient.getObject(bucketName, fileName);
+    await uploadWrapper(bucketName, fileName, stream);
+  } catch (e) {
+    console.error(e);
+    logger.error("Error moving file");
+  }
+  logger.info(`done transfering ${bucketName}/${fileName}`);
+};

src/routers/v1/admin/tasks.ts

@@ -2,6 +2,7 @@ import { NextFunction, Request, Response } from "express";
 import { userAuthenticated, userHasPermission } from "../../../auth/passport";
 import cleanUpFiles from "../../../jobs/tasks/clean-up-files";
 import initiateUserNotifcations from "../../../jobs/tasks/initiate-user-notifications";
+import { startMovingFiles } from "../../../queues/moving-files-to-backblaze";
 
 export default function () {
   const operations = {
@@ -17,6 +18,14 @@ export default function () {
           await cleanUpFiles(jobParam);
           result[jobName] = "Success";
         }
+        if (
+          jobName === "moveBucketToBackblaze" &&
+          typeof jobParam === "string" &&
+          ["artist-avatars"].includes(jobParam)
+        ) {
+          await startMovingFiles(jobParam);
+          result[jobName] = "Success";
+        }
         if (jobName === "initiateUserNotifications") {
           await initiateUserNotifcations();
           result[jobName] = "Success";