Suggestions for flow structure

[flaviolivolsi]

Hi there! I made a combination of flows that does the following:

1. draft an article
2. generate a variable number of fact checks to perform from the article and then perform them
3. rewrite the article based on the fact checks response

This is the code I wrote in order to do that:

import { FlowProducer, Queue, QueueEvents } from "bullmq";

const flowProducer = new FlowProducer({ connection: redisOptions });

const QUEUE_NAME = {
  GENERATE_FACT_CHECK: "article:generate-fact-check-chunks",
  DRAFT: "article:draft",
  REWRITE: "article:rewrite",
  FACT_CHECK: "article:fact-check",
};

const DEFAULT_JOB_OPTIONS = {
  attempts: 3,
  backoff: { type: "exponential", delay: 3000 },
};

const waitForFlowCompletion = async ({
  flowJob,
  queueName,
  timeout = 60000 * 10,
}) => {
  const queue = new Queue(queueName, { connection: redisOptions });
  const queueEvents = new QueueEvents(queueName, { connection: redisOptions });

  return new Promise((resolve, reject) => {
    const timeoutId = setTimeout(
      () => reject(new Error(`Flow timed out after ${timeout / 1000} seconds`)),
      timeout
    );

    queueEvents.on("completed", async ({ jobId }) => {
      console.log(jobId, queueName, "[flow] completed");
      if (jobId === flowJob.job.id) {
        clearTimeout(timeoutId);
        const job = await queue.getJob(jobId);
        resolve(job.returnvalue);
      }
    });

    queueEvents.on("failed", ({ jobId, failedReason }) => {
      console.log(jobId, queueName, "[flow] failed");
      if (jobId === flowJob.job.id) {
        clearTimeout(timeoutId);
        reject(new Error(failedReason));
      }
    });
  });
};

const createDraftAndChunkFlow = ({ title, teamId, articleId }) => {
  return flowProducer.add({
    name: `draft-and-chunk-generation-flow-${articleId}`,
    queueName: QUEUE_NAME.GENERATE_FACT_CHECK,
    opts: DEFAULT_JOB_OPTIONS,
    data: { title, teamId, articleId },
    children: [
      {
        name: `draft-article-${articleId}`,
        queueName: QUEUE_NAME.DRAFT,
        data: { title, teamId, articleId },
        opts: DEFAULT_JOB_OPTIONS,
      },
    ],
  });
};

const createRewriteAndFactCheckFlow = ({ title, text, chunks, articleId }) => {
  return flowProducer.add({
    name: `rewrite-and-fact-check-flow-${articleId}`,
    queueName: QUEUE_NAME.REWRITE,
    data: { title, text, chunks, articleId },
    opts: DEFAULT_JOB_OPTIONS,
    children: chunks.map((chunk, index) => ({
      name: `fact-check-chunk-${index}-${articleId}`,
      queueName: QUEUE_NAME.FACT_CHECK,
      data: { title, chunk, articleId },
      opts: DEFAULT_JOB_OPTIONS,
    })),
  });
};

export const createFlow = async ({ article, teamId }) => {
  const draftFlow = await createDraftAndChunkFlow({
    title: article.title,
    teamId,
    articleId: article.id,
  });

  const { chunks, text } = await waitForFlowCompletion({
    flowJob: draftFlow,
    queueName: QUEUE_NAME.GENERATE_FACT_CHECK,
  });


  const rewriteFlow = await createRewriteAndFactCheckFlow({
    title: article.title,
    text,
    chunks,
    articleId: article.id,
  });

  try {
    await waitForFlowCompletion({
      flowJob: rewriteFlow,
      queueName: QUEUE_NAME.REWRITE,
    });

    await sql`
      UPDATE articles
      SET workflow_status = 'ready_for_review'
      WHERE id = ${article.id}
    `;
  } catch (error) {
    console.error(`Error finalizing article ${article.id}:`, error);

    await sql`
      UPDATE articles
      SET workflow_status = 'errored'
      WHERE id = ${article.id}
    `;
  }
};

It works, but it's not ideal because it's not a single flow, therefore it will be interrupted halfway in case the server stops for any reason in the middle of the process. Is there any way to do all of this in a single flow? Unfortunately it's not clear to me from reading the docs, a nudge in the right direction would help. Thanks in advance!

[flaviolivolsi]

Ok, my mistake, I completely skipped the chaining flows section of the docs - I'll report on the updated code in case anyone finds it useful

[roggervalf]

hey @flaviolivolsi you should not use colons inside your queueNames as colons are considered as separators, this is a convention in redis but also we are using this separator when a flow is involved #2719