RankFlow AI — Job Queue & Workflow Engine Documentation

Version: 1.0.0
Systems: Inngest (durable workflows) + BullMQ (background jobs) + Redis
Purpose: Orchestrate all automation: onboarding, citations, posts, reports, monitoring

---

1. Architecture Overview#

┌─────────────────────────────────────────────────────────────┐
│                    Workflow Orchestration                   │
│                                                              │
│  ┌─────────────┐      ┌─────────────┐      ┌───────────┐  │
│  │   Inngest   │      │   BullMQ    │      │   Redis   │  │
│  │  (Complex)  │      │  (Simple)   │      │  (Queue   │  │
│  │             │      │             │      │  + Cache)  │  │
│  │ • Onboarding│      │ • GBP Post  │      │           │  │
│  │ • Citation  │      │ • Social    │      │           │  │
│  │   Builder   │      │   Post      │      │           │  │
│  │ • Site      │      │ • Review    │      │           │  │
│  │   Evolution │      │   Monitor   │      │           │  │
│  │ • Monthly   │      │ • NAP Check │      │           │  │
│  │   Report    │      │ • Email     │      │           │  │
│  │ • Content   │      │   Send      │      │           │  │
│  │   Generate  │      │ • Token     │      │           │  │
│  │             │      │   Refresh   │      │           │  │
│  └─────────────┘      └─────────────┘      └───────────┘  │
│        │                      │                            │
│        └──────────────────────┘                            │
│                      │                                       │
│                      ↓                                       │
│              ┌─────────────┐                                │
│              │   Workers   │                                │
│              │  (VPS/Docker)│                                │
│              └─────────────┘                                │
└─────────────────────────────────────────────────────────────┘

When to Use What#

---

2. Inngest Workflow Engine#

Client Setup#

// src/lib/inngest.ts
import { Inngest } from "inngest";

export const inngest = new Inngest({
  id: "rankflow",
  eventKey: process.env.INNGEST_EVENT_KEY,
});

Workflow Functions#

Example: Onboarding Workflow#

// src/server/inngest/functions/onboarding.ts

export const onboardingWorkflow = inngest.createFunction(
  {
    id: "onboarding-pipeline",
    retries: 3,
    concurrency: { limit: 5 },
  },
  { event: "skill/01-practice-onboard" },
  async ({ event, step }) => {
    const { practice_id, user_id } = event.data;
    
    // Step 1: Validate practice exists
    const practice = await step.run("validate-practice", async () => {
      return await db.practice.findUnique({ 
        where: { id: practice_id },
        include: { locations: true },
      });
    });
    
    // Step 2: Generate landing page content
    const siteContent = await step.run("generate-site", async () => {
      return await site.generate(practice_id);
    });
    
    // Step 3: Deploy landing page
    await step.run("deploy-site", async () => {
      return await site.deploy(practice_id, siteContent);
    });
    
    // Step 4: Notify client to connect GBP
    await step.run("notify-gbp-auth", async () => {
      const authUrl = await gbpAuth.getAuthUrl(practice_id);
      await email.send({
        to: practice.owner.email,
        subject: "Connect your Google Business Profile",
        body: `Please connect: ${authUrl}`,
      });
    });
    
    // Step 5: WAIT for OAuth callback (can wait up to 7 days)
    const oauthResult = await step.waitForEvent("gbp/oauth-callback", {
      timeout: "7d",
      match: "data.practice_id",
    });
    
    // Step 6: Generate first GBP post
    await step.run("create-first-post", async () => {
      const content = await ai.generate({
        task: "gbp_post",
        context: { practice, location: practice.locations[0] },
      });
      await bullmq.queues.gbpPostPublish.add("first-post", {
        practiceId: practice_id,
        content,
      });
    });
    
    // Step 7: Schedule recurring posts
    await step.run("schedule-recurring", async () => {
      await scheduleRecurringJobs(practice_id);
    });
    
    // Step 8: Sleep 1 day, then send welcome report
    await step.sleep("1d");
    
    await step.run("send-welcome-report", async () => {
      const report = await generateWelcomeReport(practice_id);
      await email.send({
        to: practice.owner.email,
        subject: "Your RankFlow Welcome Report",
        body: report,
      });
    });
    
    return { practice_id, status: "onboarded" };
  }
);

Example: Citation Builder Workflow#

// src/server/inngest/functions/citation-build.ts

export const citationBuilder = inngest.createFunction(
  {
    id: "citation-builder",
    retries: 3,
    concurrency: { limit: 3 }, // Conservative — directory sites are slow
  },
  { event: "skill/13-citation-submit" },
  async ({ event, step }) => {
    const { practice_id, location_id } = event.data;
    
    // Step 1: Get practice and location
    const [practice, location] = await step.run("get-data", async () => {
      return await Promise.all([
        db.practice.findUnique({ where: { id: practice_id } }),
        db.location.findUnique({ where: { id: location_id } }),
      ]);
    });
    
    // Step 2: Get active directories
    const directories = await step.run("get-directories", async () => {
      return await db.citationDirectory.findMany({ where: { isActive: true } });
    });
    
    // Step 3: Generate unique descriptions for each directory
    const descriptions = await step.run("generate-descriptions", async () => {
      return await Promise.all(
        directories.map(async (dir) => ({
          directoryId: dir.id,
          description: await ai.generate({
            task: "citation_description",
            context: { practice, location, directory: dir },
          }),
        }))
      );
    });
    
    // Step 4: Submit to each directory (parallel with concurrency limit)
    const results = await Promise.all(
      directories.map(async (dir, i) => {
        return await step.run(`submit-${dir.name}`, async () => {
          try {
            if (dir.submissionType === "API") {
              return await submitViaApi(dir, practice, location, descriptions[i]);
            } else {
              return await submitViaBrowser(dir, practice, location, descriptions[i]);
            }
          } catch (error) {
            return { success: false, error: (error as Error).message };
          }
        });
      })
    );
    
    // Step 5: Store results
    await step.run("store-results", async () => {
      await db.citation.createMany({
        data: results.map((r, i) => ({
          practiceId: practice_id,
          locationId: location_id,
          directoryName: directories[i].name,
          status: r.success ? "SUBMITTED" : "FAILED",
          submittedAt: r.success ? new Date() : null,
          errorMessage: r.error || null,
        })),
      });
    });
    
    // Step 6: Schedule NAP verification (1 week later)
    await step.sleep("7d");
    
    await step.run("schedule-verification", async () => {
      await inngest.send({
        name: "skill/14-citation-verify-nap",
        data: { practice_id, location_id },
      });
    });
    
    return { 
      submitted: results.filter(r => r.success).length, 
      failed: results.filter(r => !r.success).length 
    };
  }
);

Inngest Event Types#

// src/server/inngest/events.ts

export type RankFlowEvents = {
  "skill/01-practice-onboard": {
    data: { practice_id: string; user_id: string };
  };
  "skill/13-citation-submit": {
    data: { practice_id: string; location_id: string };
  };
  "skill/14-citation-verify-nap": {
    data: { practice_id: string; location_id: string };
  };
  "skill/32-report-generate": {
    data: { practice_id: string; period_start: string; period_end: string };
  };
  "gbp/oauth-callback": {
    data: { practice_id: string; account_email: string };
  };
  "social/oauth-callback": {
    data: { practice_id: string; platform: string; connection_id: string };
  };
};

---

3. BullMQ Job Queue#

Queue Setup#

// src/server/bullmq/queue.ts

import { Queue } from "bullmq";
import { redis } from "@/lib/redis";

export const queues = {
  gbpPostPublish: new Queue("gbp-post-publish", { connection: redis }),
  socialPostPublish: new Queue("social-post-publish", { connection: redis }),
  reviewMonitor: new Queue("review-monitor", { connection: redis }),
  napCheck: new Queue("nap-check", { connection: redis }),
  emailSend: new Queue("email-send", { connection: redis }),
  tokenRefresh: new Queue("token-refresh", { connection: redis }),
  citationSubmit: new Queue("citation-submit", { connection: redis }),
  pdfGenerate: new Queue("pdf-generate", { connection: redis }),
};

Worker Setup#

// src/server/bullmq/worker.ts

import { Worker } from "bullmq";
import { redis } from "@/lib/redis";
import { logger } from "@/lib/logger";

export function createWorkers() {
  const workers = [
    new Worker("gbp-post-publish", gbpPostProcessor, { 
      connection: redis, 
      concurrency: 3,
      limiter: { max: 30, duration: 60000 }, // 30/min per queue
    }),
    
    new Worker("social-post-publish", socialPostProcessor, { 
      connection: redis, 
      concurrency: 5,
      limiter: { max: 25, duration: 60000 }, // Platform limits
    }),
    
    new Worker("review-monitor", reviewMonitorProcessor, { 
      connection: redis, 
      concurrency: 2,
    }),
    
    new Worker("nap-check", napCheckProcessor, { 
      connection: redis, 
      concurrency: 2,
    }),
    
    new Worker("email-send", emailSendProcessor, { 
      connection: redis, 
      concurrency: 5,
    }),
    
    new Worker("token-refresh", tokenRefreshProcessor, { 
      connection: redis, 
      concurrency: 1, // Sequential to avoid rate limits
    }),
    
    new Worker("citation-submit", citationSubmitProcessor, {
      connection: redis,
      concurrency: 3,
    }),
    
    new Worker("pdf-generate", pdfGenerateProcessor, {
      connection: redis,
      concurrency: 2,
    }),
  ];
  
  return workers;
}

Scheduled Recurring Jobs#

// src/server/bullmq/scheduler.ts

export async function scheduleRecurringJobs() {
  // GBP post scheduler — 2-3x per week per client
  await queues.gbpPostPublish.add(
    "schedule-posts",
    {},
    { repeat: { pattern: "0 9 * * 1,3,5" } } // Mon, Wed, Fri at 9 AM IST
  );
  
  // Review monitor — daily at 8 AM
  await queues.reviewMonitor.add(
    "check-reviews",
    {},
    { repeat: { pattern: "0 8 * * *" } }
  );
  
  // NAP check — 1st of month at 2 AM
  await queues.napCheck.add(
    "check-nap",
    {},
    { repeat: { pattern: "0 2 1 * *" } }
  );
  
  // Token refresh — daily at 3 AM
  await queues.tokenRefresh.add(
    "refresh-tokens",
    {},
    { repeat: { pattern: "0 3 * * *" } }
  );
  
  // Monthly report — 1st of month at 4 AM
  await queues.emailSend.add(
    "monthly-reports",
    {},
    { repeat: { pattern: "0 4 1 * *" } }
  );
  
  // Site evolution — every Monday at 2 AM
  await queues.gbpPostPublish.add(
    "site-evolution",
    {},
    { repeat: { pattern: "0 2 * * 1" } }
  );
}

---

4. Job Types & Scheduling#

Complete Job Registry#

---

5. Retry Policies & Dead Letter Queue#

Default Retry Config#

const DEFAULT_RETRY = {
  attempts: 3,
  backoff: {
    type: "exponential" as const,
    delay: 5000, // 5s initial
  },
};

const JOB_RETRY_CONFIGS: Record<string, { attempts: number; delay: number }> = {
  "gbp-post-publish": { attempts: 3, delay: 5000 },
  "social-post-publish": { attempts: 3, delay: 10000 },
  "review-monitor": { attempts: 5, delay: 60000 },
  "nap-check": { attempts: 3, delay: 300000 }, // 5min
  "citation-submit": { attempts: 5, delay: 60000 },
  "email-send": { attempts: 3, delay: 5000 },
  "token-refresh": { attempts: 5, delay: 60000 },
  "pdf-generate": { attempts: 2, delay: 10000 },
};

Dead Letter Queue#

// src/server/bullmq/dead-letter.ts

import { Queue } from "bullmq";

export const deadLetterQueue = new Queue("dead-letter", { connection: redis });

export async function handleFailedJob(job: Job, err: Error) {
  if (job.attemptsMade >= (JOB_RETRY_CONFIGS[job.name]?.attempts || 3)) {
    await deadLetterQueue.add(job.name, {
      originalJobId: job.id,
      payload: job.data,
      error: err.message,
      stack: err.stack,
      failedAt: new Date().toISOString(),
      attemptsMade: job.attemptsMade,
    });
    
    // Notify admin
    await email.send({
      to: "admin@rankflow.ai",
      subject: `Job Failed: ${job.name}`,
      body: `Job ${job.id} failed after ${job.attemptsMade} attempts. Error: ${err.message}`,
    });
    
    logger.error({
      jobId: job.id,
      jobName: job.name,
      error: err.message,
      attempts: job.attemptsMade,
    }, "Job moved to dead letter queue");
  }
}

Manual Retry from Admin#

// Admin can retry dead letter jobs via dashboard
export async function retryDeadLetterJob(jobId: string) {
  const job = await deadLetterQueue.getJob(jobId);
  if (!job) throw new Error("Job not found in dead letter queue");
  
  const { payload, originalJobId } = job.data;
  
  // Re-queue to original queue
  const targetQueue = queues[job.name as keyof typeof queues];
  if (targetQueue) {
    await targetQueue.add(job.name, payload, {
      jobId: originalJobId, // Preserve original ID
    });
    await job.remove(); // Remove from DLQ
  }
}

---

6. Observability & Monitoring#

Structured Logging#

// Every job logs:
logger.info({
  event: "job_started",
  jobId: job.id,
  jobName: job.name,
  practiceId: job.data.practiceId,
  queue: queue.name,
  attempt: job.attemptsMade + 1,
}, "Job started");

logger.info({
  event: "job_completed",
  jobId: job.id,
  jobName: job.name,
  practiceId: job.data.practiceId,
  durationMs: Date.now() - startTime,
}, "Job completed");

logger.error({
  event: "job_failed",
  jobId: job.id,
  jobName: job.name,
  error: err.message,
  attempt: job.attemptsMade,
}, "Job failed");

Metrics to Track#

Health Check Endpoint#

// src/app/api/health/jobs/route.ts

export async function GET() {
  const queueStatuses = await Promise.all(
    Object.entries(queues).map(async ([name, queue]) => {
      const [waiting, active, completed, failed] = await Promise.all([
        queue.getWaitingCount(),
        queue.getActiveCount(),
        queue.getCompletedCount(),
        queue.getFailedCount(),
      ]);
      return { name, waiting, active, completed, failed };
    })
  );
  
  return NextResponse.json({
    timestamp: new Date().toISOString(),
    queues: queueStatuses,
    overall: {
      totalWaiting: queueStatuses.reduce((s, q) => s + q.waiting, 0),
      totalActive: queueStatuses.reduce((s, q) => s + q.active, 0),
      totalFailed: queueStatuses.reduce((s, q) => s + q.failed, 0),
    },
  });
}

---

7. Scalability Strategy#

Horizontal Scaling#

Web Server (Vercel/Dokploy)          Job Workers (VPS/Docker)
┌─────────────────────┐               ┌─────────────────────┐
│ Next.js App         │               │ Worker Container 1  │
│ - API routes        │               │ - GBP posts         │
│ - tRPC handlers     │               │ - Social posts      │
│ - Inngest handler   │               │ - Review monitor    │
└─────────────────────┘               ├─────────────────────┤
                                      │ Worker Container 2  │
                                      │ - Citation submit   │
                                      │ - NAP check         │
                                      │ - PDF generate      │
                                      ├─────────────────────┤
                                      │ Worker Container N  │
                                      │ (scale by queue depth)│
                                      └─────────────────────┘

Scaling Rules#

Concurrency Limits#

---

8. Idempotency & Safety#

Idempotency Keys#

// All jobs must be idempotent
export function generateJobId(practiceId: string, jobType: string, date?: string): string {
  return `${practiceId}:${jobType}:${date || new Date().toISOString().slice(0, 10)}`;
}

// Example: GBP post for client on 2025-01-15
await queues.gbpPostPublish.add(
  "weekly-post",
  { practiceId: "abc", content: "..." },
  { jobId: generateJobId("abc", "gbp-post", "2025-01-15") }
);
// If re-run, same jobId prevents duplicate

Safety Checks#

Graceful Shutdown#

// src/server/bullmq/graceful-shutdown.ts

export async function gracefulShutdown(workers: Worker[]) {
  logger.info("Starting graceful shutdown...");
  
  // Stop accepting new jobs
  for (const worker of workers) {
    await worker.pause();
  }
  
  // Wait for active jobs to complete (30s timeout)
  await Promise.all(
    workers.map(worker => worker.close())
  );
  
  logger.info("All workers closed. Shutdown complete.");
}

---

End of Job Queue & Workflow Engine Documentation