Anjing Webshells

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



import { Queue, Worker } from "bullmq";
import scrapeData from "@/app/dbConfig/scrapeData";
import Keyword from "@/models/keywordModel";
import Prompt from "@/models/promptModel";
import redisClient from "./redisClient";
import { timeOperation } from "@/lib/timeOperation";

interface OrganizationJobData {
    API: string;
    GPT: string;
    secondaryKeywords: string[];
    keyword: string;
    customOutlinePrompt?: string;
    customSectionPrompt?: string;
    selectPrompt: string;
    _id: string;
}

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

const workerOrg = new Worker(
  "longArticlesQueueOrg",
  async (job) => {
    const jobStartTime = performance.now();
    const jobData = job.data as OrganizationJobData;

    try {

      // Get prompt data
      const prompt = await Prompt.findById(jobData.selectPrompt);
      if (!prompt?.multiStepPrompt) {
        throw new Error("Invalid prompt configuration");
      }

      // Build prompts with keyword replacement
      const outlinePrompt = prompt.multiStepPrompt.outline.replaceAll(
        "{{KEYWORD}}", 
        jobData.keyword
      ) + (jobData.customOutlinePrompt || "");

      const sectionPrompt = prompt.multiStepPrompt.section.replaceAll(
        "{{KEYWORD}}", 
        jobData.keyword
      ) + (jobData.customSectionPrompt || "");

      // Generate content
      const generatedContent = await scrapeData(
        outlinePrompt,
        sectionPrompt,
        jobData.API,
        jobData.GPT,
        jobData.secondaryKeywords,
        jobData.keyword,
        jobData.customOutlinePrompt,
        jobData.customSectionPrompt
      );

      // Update keyword with generated content
      const updatedKeyword = await Keyword.findByIdAndUpdate(
        jobData._id,
        {
          response: generatedContent?.trim(),
          status: "completed",
        },
        { new: true }
      );

      timeOperation(
        `SUCCESS: Organization article generation for keyword: '${jobData.keyword}'`, 
        jobStartTime
      );
      return updatedKeyword;

    } catch (error: any) {
      timeOperation(
        `FAILED: Organization article generation for keyword:'${jobData.keyword} , error: ${error}'`, 
        jobStartTime
      );

      // Update keyword with error status
      if (jobData.keyword) {
        try {
          await Keyword.findOneAndUpdate(
            { keyword: jobData.keyword, organizationBatch: jobData.batchId },
            {
              response: error.message || "An error occurred while generating the article.",
              status: "failed",
            },
            { new: true }
          );
        } catch (updateError) {
          console.error('Error updating keyword on job failure:', updateError);
        }
      }
      await job.moveToFailed(
        { message: error.message || "Job failed due to an error." }, 
        true
      );
      await job.remove();
      return Promise.reject(error);
    }
  },
  {
    connection: redisClient,
    limiter: {
      max: 1,
      duration: 500,
    },
    concurrency: 1,
  }
);

// Graceful shutdown handler
const gracefulShutdown = async (signal: string) => {
  console.log(`Received ${signal}, shutting down organization queue gracefully...`);

  try {
    await workerOrg.close();
    await longArticlesQueueOrg.close();
    await longArticlesQueueOrg.drain();
    await longArticlesQueueOrg.obliterate({ force: true });
  } catch (error) {
    console.error("Error during organization queue shutdown:", error);
  }
};

// Handle shutdown signals
process.on("SIGINT", () => gracefulShutdown("SIGINT"));
process.on("SIGTERM", () => gracefulShutdown("SIGTERM"));
process.on("exit", () => gracefulShutdown("exit"));
process.on("uncaughtException", (err) => {
  console.error("Uncaught Exception in organization queue:", err);
  gracefulShutdown("uncaughtException");
});

export { longArticlesQueueOrg, workerOrg };