Anjing Webshells

Website : rimsha.abasa.com
: / var / www / talha_silentcontent / src / queues /
Filename : longArticleQueue.ts
back
// @ts-nocheck

import { Queue, RedisClient, Worker } from "bullmq";
// import { getRedisConnection, disconnectRedis } from "./redisClient";
import longGenerate from "@/app/dbConfig/longGenerate";
import scrapeData from "@/app/dbConfig/scrapeData";
import Keyword from "@/models/keywordModel";
import Redis from "ioredis"
import redisClient from "./redisClient";
import { timeOperation } from "@/lib/timeOperation";


const longArticlesQueue = new Queue("longArticlesQueue", {
  connection: redisClient,
  defaultJobOptions: {
    removeOnComplete: true,
    removeOnFail: true,
  },
});

const worker = new Worker(
  "longArticlesQueue",
  async (job) => {
    const jobStartTime = performance.now();


    const {
      outlinePrompt,
      sectionPrompt,
      API,
      GPT,
      secondaryKeywords,
      keyword,
      customOutlinePrompt,
      customSectionPrompt,
      _id,
    } = job.data;
    

    try {
      console.log("Job Data: ", job.data);

      

      // Process the job and generate the article content
      const generatedContent = await scrapeData(
        outlinePrompt,
        sectionPrompt,
        API,
        GPT,
        secondaryKeywords,
        keyword,
        customOutlinePrompt,
        customSectionPrompt
      );

      // Update the Keyword document with the generated content and set status to "completed"
      const updatedKeyword = await Keyword.findByIdAndUpdate(
        _id,
        {
          response: generatedContent!.trim(),
          status: "completed",
        },
        { new: true }
      );

      // Log the updated Keyword document if needed
      console.log(
        "Updated Keyword:",
        updatedKeyword._id,
        updatedKeyword.keyword,
        updatedKeyword.status
      );

      // Calculate and log the total time for this job
      timeOperation(`SUCCESS: Article generation for keyword: '${keyword}, took time:'`, jobStartTime);
      return;
    } catch (error) {
      // Handle errors and reject the job if necessary
      console.log('error in job', error.message);
      timeOperation(`FAILED: Article generation for keyword:'${keyword}', took time:`, jobStartTime);
      try {
        const updatedKeyword = await Keyword.findByIdAndUpdate(
          _id,
          {
            response: error.message || "An error occurred while generating the article.",
            status: "failed",
          },
          { new: true }
        );
        
        console.log(
          "Updated Keyword on error:",
          updatedKeyword._id,
          updatedKeyword.keyword,
          updatedKeyword.status
        );

        await job.moveToFailed({ message: error.message || "Job failed due to an error." }, true);
      } catch (updateError) {
        console.error('Error updating keyword on job failure:', updateError);
      }

      await job.remove();
      return Promise.reject(error);
    }
  },
  {
    connection: redisClient,
    limiter: {
      max: 1,          
      duration: 500,
    },
    concurrency: 1,   //more concurrency added to handle more tasks in parallel
  }
);


const gracefulShutdown = async (signal: string) => {
  console.log(`Received ${signal}, shutting down gracefully...`);

  try {
    console.log("Stopping worker...");
    await worker.close({force: true});
    console.log("Worker stopped.");

    console.log("Closing queue...");
    await longArticlesQueue.close();
    console.log("Queue closed.");

    console.log("Draining jobs...");
    await longArticlesQueue.drain();
    console.log("Queue drained.");

    console.log("Clearing jobs...");
    await longArticlesQueue.obliterate({ force: true });
    console.log("All jobs cleared.");

    console.log("Closing Redis connection...");
    await redisClient.quit();
    console.log("Redis connection closed.");
  } catch (error) {
    console.error("Error during shutdown:", error);
  } finally {
    process.exit(0);
  }
};

// Handle shutdown signals
process.on("SIGINT", () => gracefulShutdown("SIGINT")); // CTRL + C
process.on("SIGTERM", () => gracefulShutdown("SIGTERM")); // External stop signals
process.on("exit", () => gracefulShutdown("exit"));
process.on("uncaughtException", (err) => {
  console.error("Uncaught Exception:", err);
  gracefulShutdown("uncaughtException");
});

export { longArticlesQueue, worker };