Asynchronous Report Generation

[ngoiyaeric]

I understand you want to make report generation asynchronous so users can interact with other parts of the system while reports are being generated. Currently, the submit server action in app/actions.tsx processes requests synchronously, blocking until all agent work completes before returning 1 .

Current Blocking Behavior

The system uses streaming primitives for progressive UI rendering, but the server action itself is synchronous:

• The submit function initializes streaming state but waits for all processing to complete 1
• The researcher agent runs in a synchronous loop until a non-empty response is generated 2
• Resolution search processing is called synchronously 3
• The function only returns after all work is done 4

Required Changes for Asynchronous Processing

To make report generation truly asynchronous, you would need to implement a background job queue system:

1. Job Queue Infrastructure

• Implement a job queue (e.g., Bull, Agenda, or database-backed queue)
• Create a separate worker process to handle report generation jobs
• Store job state in a database for persistence

2. Modified Submit Flow

// Instead of blocking, return immediately with job ID
async function submit(formData?: FormData, skip?: boolean) {
  const jobId = nanoid()
  
  // Enqueue the job for background processing
  await queue.add('report-generation', {
    jobId,
    formData,
    skip
  })
  
  // Return immediately with job ID
  return {
    jobId,
    status: 'processing'
  }
}

3. Status Checking Mechanism

• Create a new server action to check job status
• Implement polling or websocket updates on the client
• Stream partial results as they become available from the worker

4. Worker Process

• Move the current processEvents() and processResolutionSearch() logic to a worker
• Update AIState progressively as results are generated
• Store intermediate results in the database

Notes

This is a significant architectural change that requires implementing a job queue system, worker processes, and status tracking infrastructure. The current streaming approach in app/actions.tsx provides progressive UI updates but still blocks the server action. True asynchronous processing would require background job processing with a queue system like Bull or a database-backed solution. The existing optimizations in OPTIMIZATION_SUMMARY.md focus on reducing re-renders and caching, not on making the actual generation process non-blocking.

Wiki pages you might want to explore:

• AI Agent Orchestration (QueueLab/QCX)
• Submit Action and Tool Orchestration (QueueLab/QCX)

Citations

File: app/actions.tsx (L36-42)

async function submit(formData?: FormData, skip?: boolean) {
  'use server'

  const aiState = getMutableAIState<typeof AI>()
  const uiStream = createStreamableUI()
  const isGenerating = createStreamableValue(true)
  const isCollapsed = createStreamableValue(false)

File: app/actions.tsx (L223-223)

    processResolutionSearch();

File: app/actions.tsx (L516-550)

    while (
      useSpecificAPI
        ? answer.length === 0
        : answer.length === 0 && !errorOccurred
    ) {
      const { fullResponse, hasError, toolResponses } = await researcher(
        currentSystemPrompt,
        uiStream,
        streamText,
        messages,
        mapProvider,
        useSpecificAPI,
        drawnFeatures
      )
      answer = fullResponse
      toolOutputs = toolResponses
      errorOccurred = hasError

      if (toolOutputs.length > 0) {
        toolOutputs.map(output => {
          aiState.update({
            ...aiState.get(),
            messages: [
              ...aiState.get().messages,
              {
                id: groupeId,
                role: 'tool',
                content: JSON.stringify(output.result),
                name: output.toolName,
                type: 'tool'
              }
            ]
          })
        })
      }

File: app/actions.tsx (L619-624)

  return {
    id: nanoid(),
    isGenerating: isGenerating.value,
    component: uiStream.value,
    isCollapsed: isCollapsed.value
  }

[ngoiyaeric]

@coderabbitai plan https://vercel.com/docs/queues perhaps use these Queues or use semaphores in typesrcipt

[coderabbitai]

Coding Plan

Summary

• Implement a lightweight database-backed job queue using the existing PostgreSQL/Supabase infrastructure, avoiding additional dependencies like Redis
• Extract agent processing logic into a context-independent worker module that reports progress via callbacks rather than streaming primitives
• Provide deployment flexibility through a hybrid worker model supporting both cron-triggered API routes and standalone processes
• Maintain backward compatibility by keeping synchronous processing for simple/fast operations while routing complex report generation through the async job queue

Design Choices

Design Choice 1: Queue technology selection

Options Considered:

1. Bull/BullMQ with Redis - Robust features but requires adding Redis infrastructure
2. pg-boss - Postgres-native job queue, fits existing Supabase stack
3. Simple database polling - Minimal approach using only the existing Postgres database

Chosen Option: 3

Rationale: Simple database polling is selected because it requires no additional infrastructure beyond the existing Postgres database, aligns with the codebase's simplicity-first approach, and can be enhanced later if needed.

Design Choice 2: Client update mechanism

Options Considered:

1. Polling via API endpoint - Simple, stateless, works with any deployment
2. Server-Sent Events (SSE) - Efficient for real-time updates but requires persistent connections
3. WebSocket - Full duplex but adds complexity and infrastructure requirements

Chosen Option: 1

Rationale: Polling is selected for simplicity and compatibility with serverless deployments; the polling interval can be tuned for responsiveness.

Design Choice 3: Worker deployment model

Options Considered:

1. Separate long-running process - Traditional worker, requires separate deployment
2. Cron-triggered API route - Worker logic in an API route called periodically
3. Hybrid approach - Support both via a shared worker module

Chosen Option: 3

Rationale: A hybrid approach where the worker logic is encapsulated in a module that can be invoked either by a cron-triggered API route or a standalone process, providing deployment flexibility.

💡 User Tips

Regenerate the plan with different choices with @coderabbitai <feedback>.

Implementation Steps

Phase 1: Job Queue Database Infrastructure

This phase establishes the persistence layer for tracking asynchronous jobs, including their state, progress, and results. The jobs table will serve as both the queue and the status store.

Task 1: Create Jobs Table Schema

Add a new database table to track job state, progress, and results for asynchronous report generation.

• Add jobs table definition in lib/db/schema.ts with columns: id (uuid primary key), user_id (FK to users), chat_id (FK to chats), type (enum: 'report_generation', 'resolution_search'), status (enum: 'pending', 'processing', 'completed', 'failed'), progress (integer 0-100), input (jsonb for serialized form data and parameters), result (jsonb for final output), error (text for failure messages), created_at, updated_at, started_at, completed_at
• Add appropriate indexes on status and user_id for efficient polling queries
• Define Drizzle relations linking jobs to users and chats

Task 2: Create Database Migration

Generate and apply the migration for the new jobs table.

• Create migration file in drizzle/migrations/ using drizzle-kit generate
• Include index creation statements for query optimization
• Follow existing migration patterns from 0000_sweet_metal_master.sql and 0001_add_calendar_notes.sql

Task 3: Implement Job Service Layer

Create a service module with functions to manage job lifecycle operations.

• Create lib/services/jobs.ts with functions: createJob(), getJob(), updateJobStatus(), updateJobProgress(), completeJob(), failJob(), getPendingJobs(), getUserJobs()
• Implement optimistic locking pattern for getPendingJobs() using status update to prevent duplicate processing
• Include type definitions for job input/output structures that mirror the current submit parameters

🤖 Prompt for AI agents

Set up the database persistence layer for the async job queue system. The goal
is to create a `jobs` table, its migration, and a service layer for managing job
lifecycle operations.

**Task 1: Update `lib/db/schema.ts`**
- Add a `jobs` table with the following columns:
  - `id`: uuid, primary key
  - `user_id`: FK to users
  - `chat_id`: FK to chats
  - `type`: enum with values `'report_generation'` and `'resolution_search'`
- `status`: enum with values `'pending'`, `'processing'`, `'completed'`,
`'failed'`
  - `progress`: integer (0–100)
  - `input`: jsonb (serialized form data and parameters)
  - `result`: jsonb (final output)
  - `error`: text (failure messages)
  - `created_at`, `updated_at`, `started_at`, `completed_at` timestamp columns
- Add indexes on `status` and `user_id` for efficient polling queries
- Define Drizzle relations linking jobs to users and chats

**Task 2: Create migration in `drizzle/migrations/`**
- Use drizzle-kit generate to produce the migration file
- Include index creation statements for query optimization
- Follow existing migration file patterns (e.g., `0000_sweet_metal_master.sql`,
`0001_add_calendar_notes.sql`)

**Task 3: Create `lib/services/jobs.ts`**
- Implement the following exported functions: `createJob()`, `getJob()`,
`updateJobStatus()`, `updateJobProgress()`, `completeJob()`, `failJob()`,
`getPendingJobs()`, `getUserJobs()`
- `getPendingJobs()` must use an optimistic locking pattern — atomically update
`status` to `'processing'` and return the claimed job to prevent duplicate
processing
- Include TypeScript type definitions for job input/output structures that
mirror the current `submit` server action parameters

Phase 2: Worker Implementation

This phase extracts the agent processing logic into a context-independent worker module that can execute jobs from the queue without depending on the Server Action request lifecycle.

Task 1: Extract Agent Processing Logic

Refactor processEvents and processResolutionSearch to be callable from a worker context without streaming primitives.

• Create lib/workers/report-processor.ts containing the core processing logic extracted from app/actions.tsx
• Replace createStreamableUI and createStreamableValue dependencies with callback-based progress reporting (e.g., onProgress, onPartialResult)
• The processor should accept serializable inputs and return serializable outputs that can be stored in the jobs table
• Maintain the existing agent call chain: taskManager → inquire/researcher → querySuggestor, but collect results into a data structure rather than streaming UI

Task 2: Implement Worker Entry Point

Create the worker module that polls for pending jobs and executes them.

• Create lib/workers/job-worker.ts with a processNextJob() function that claims and processes one pending job
• Implement a runWorkerLoop() function for continuous processing with configurable poll interval
• Use the job service layer to claim jobs atomically (update status to 'processing' and return the job in one transaction)
• Call the report processor with appropriate callbacks that update job progress via the service layer
• Handle errors gracefully: catch exceptions, call failJob() with error details, continue to next job

Task 3: Create Worker API Route for Cron Triggering

Add an API endpoint that can be called by a cron scheduler to trigger job processing.

• Create app/api/workers/process-jobs/route.ts as a POST endpoint
• Implement authorization check (e.g., secret header or Vercel cron authentication)
• Call processNextJob() in a loop with a time limit (e.g., 25 seconds for Vercel function timeout safety)
• Return processed job count and any errors for monitoring

🤖 Prompt for AI agents

Extract agent processing logic into a self-contained worker module and expose it
via a cron-triggerable API route. The goal is to make job processing independent
of the Server Action request lifecycle.

**Task 1: Create `lib/workers/report-processor.ts`**
- Extract core processing logic from `processEvents` and
`processResolutionSearch` in `app/actions.tsx`
- Remove all `createStreamableUI` and `createStreamableValue` dependencies
- Replace streaming with callback-based progress reporting using parameters like
`onProgress` and `onPartialResult`
- Accept fully serializable inputs and return serializable outputs suitable for
storage in the jobs table
- Maintain the existing agent call chain: taskManager → inquire/researcher →
querySuggestor, collecting results into a data structure instead of streaming UI

**Task 2: Create `lib/workers/job-worker.ts`**
- Implement `processNextJob()`: claims one pending job (atomically via the job
service layer, setting status to `'processing'`), runs the report processor with
progress-updating callbacks, and handles completion/failure
- Implement `runWorkerLoop()`: continuously polls for and processes pending jobs
with a configurable poll interval
- On any exception: catch the error, call `failJob()` with error details, and
continue to the next job without crashing the loop

**Task 3: Create `app/api/workers/process-jobs/route.ts`**
- Expose a POST endpoint
- Implement an authorization check (e.g., validating a secret header or Vercel
cron authentication)
- Call `processNextJob()` in a loop bounded by a time limit (e.g., 25 seconds to
stay within Vercel function timeout)
- Return a response with the count of processed jobs and any errors encountered,
for monitoring purposes

Phase 3: Modified Submit Flow and Status API

This phase modifies the submit action to enqueue jobs instead of processing synchronously and adds an API endpoint for clients to check job status.

Task 1: Modify Submit Action for Job Enqueueing

Update the submit server action to create a job record and return immediately for async processing.

• Modify app/actions.tsx to detect when async processing is appropriate (e.g., complex queries, resolution search)
• For async paths: serialize input parameters, call createJob(), return job ID and status immediately
• For simple/fast operations (e.g., hard-coded FAQ answers, simple inquiries): keep synchronous behavior for responsiveness
• Add a new return type variant that includes jobId and status: 'queued' alongside the existing streaming return shape
• Update AIState handling to work with job-based flow (initial state set at job creation, final state set by worker)

Task 2: Create Job Status API Endpoint

Add an API route for clients to poll job status and retrieve results.

• Create app/api/jobs/[id]/route.ts with GET method
• Implement authentication check using getCurrentUserIdOnServer()
• Verify the requesting user owns the job (or has appropriate permissions)
• Return job status, progress percentage, and result data when completed
• For completed jobs with UI-generating results, include serialized component data that the client can render

Task 3: Handle Job Result Reconstruction

Implement logic to reconstruct the UI state from completed job results.

• Create lib/utils/job-result-renderer.ts with functions to convert serialized job results back into React components
• Map result types to appropriate UI components (BotMessage, SearchResults, GeoJsonLayer, etc.)
• Update getUIStateFromAIState in app/actions.tsx to handle job-based message reconstruction
• Ensure completed job results are persisted to chat history via the existing saveChat mechanism

🤖 Prompt for AI agents

Modify the `submit` server action to enqueue jobs for async processing and add
the necessary API endpoints for status polling and result reconstruction.

**Task 1: Modify `app/actions.tsx`**
- Add logic to detect when async processing is appropriate (e.g., complex
queries, resolution search)
- For **async paths**: serialize input parameters, call `createJob()`, and
return immediately with `jobId` and `status: 'queued'`
- For **simple/fast operations** (e.g., hard-coded FAQ answers, simple
inquiries): keep the existing synchronous behavior
- Add a new return type variant that includes `jobId` and `status: 'queued'`
alongside the existing streaming return shape
- Update AIState handling so initial state is set at job creation and final
state is set by the worker

**Task 2: Create `app/api/jobs/[id]/route.ts`**
- Expose a GET endpoint
- Authenticate the request using `getCurrentUserIdOnServer()`
- Verify the requesting user owns the job before returning data
- Return job status, progress percentage, and result data when completed
- For completed jobs with UI-generating results, include serialized component
data for client-side rendering

**Task 3: Create `lib/utils/job-result-renderer.ts`**
- Implement functions to convert serialized job results back into React
components
- Map result types to appropriate UI components (e.g., BotMessage,
SearchResults, GeoJsonLayer)
- Update `getUIStateFromAIState` in `app/actions.tsx` to handle job-based
message reconstruction
- Ensure completed job results are persisted to chat history using the existing
`saveChat` mechanism

Phase 4: Client-Side Integration

This phase updates the frontend to handle async job states, poll for status updates, and render results when jobs complete.

Task 1: Create Job Polling Hook

Implement a React hook for polling job status with automatic cleanup.

• Create lib/hooks/use-job-status.ts with useJobStatus(jobId) hook
• Implement polling logic with configurable interval (e.g., 1-2 seconds while processing)
• Return job status, progress, result, and error states
• Include automatic polling termination when job reaches terminal state (completed/failed)
• Handle network errors gracefully with retry logic

Task 2: Create Job Progress Component

Build a UI component to display job processing status and progress.

• Create components/job-progress.tsx showing job status, progress bar, and elapsed time
• Handle different job states: queued (waiting), processing (with progress), completed (success), failed (with error message)
• Include option to display partial results as they become available (if stored during processing)
• Style consistently with existing spinner and loading patterns in the codebase

Task 3: Update Chat Interface for Async Jobs

Modify the chat UI to handle both synchronous streaming and asynchronous job-based responses.

• Update components/chat.tsx (or equivalent) to detect job-based responses from submit
• When a job ID is returned, render the JobProgress component and start polling
• On job completion, render the result using the job result renderer
• Ensure the UI remains interactive during job processing (user can view history, start new chats, etc.)
• Handle edge cases: page refresh during processing (restore job polling), job timeout, user cancellation

🤖 Prompt for AI agents

Integrate the async job system into the client-side UI with a polling hook, a
progress display component, and updates to the chat interface.

**Task 1: Create `lib/hooks/use-job-status.ts`**
- Implement a `useJobStatus(jobId)` React hook
- Poll the job status API at a configurable interval (e.g., 1–2 seconds while
the job is processing)
- Return job status, progress, result, and error states
- Automatically stop polling when the job reaches a terminal state (`completed`
or `failed`)
- Handle network errors gracefully with retry logic

**Task 2: Create `components/job-progress.tsx`**
- Build a component that displays job status, a progress bar, and elapsed time
- Handle all job states: queued (waiting indicator), processing (progress bar),
completed (success display), failed (error message)
- Optionally display partial results as they become available if stored during
processing
- Style consistently with existing spinner and loading UI patterns in the
codebase

**Task 3: Update `components/chat.tsx` (or equivalent)**
- Detect when the `submit` action returns a job-based response (i.e., a `jobId`
is present)
- When a job ID is returned, render the `JobProgress` component and begin
polling via `useJobStatus`
- On job completion, render the result using the job result renderer from
`lib/utils/job-result-renderer.ts`
- Keep the UI interactive during processing (user can view history, start new
chats, etc.)
- Handle edge cases: page refresh during processing (restore polling from job
ID), job timeout, and user cancellation

Research

The application is a Next.js project using Drizzle ORM with Supabase PostgreSQL, Vercel AI SDK (ai/rsc) for streaming, and a multi-agent architecture (researcher, task-manager, inquire, writer, query-suggestor, resolution-search). Currently, the submit server action in app/actions.tsx uses fire-and-forget async functions (processEvents, processResolutionSearch) that stream results via createStreamableUI and createStreamableValue, but these are tied to the request lifecycle. There is no existing job queue or background worker infrastructure. Authentication uses Supabase SSR with cookie-based sessions.

---

🚀 Next Steps

🤖 All AI agent prompts combined

Task: 1

Set up the database persistence layer for the async job queue system. The goal
is to create a `jobs` table, its migration, and a service layer for managing job
lifecycle operations.

**Task 1: Update `lib/db/schema.ts`**
- Add a `jobs` table with the following columns:
  - `id`: uuid, primary key
  - `user_id`: FK to users
  - `chat_id`: FK to chats
  - `type`: enum with values `'report_generation'` and `'resolution_search'`
- `status`: enum with values `'pending'`, `'processing'`, `'completed'`,
`'failed'`
  - `progress`: integer (0–100)
  - `input`: jsonb (serialized form data and parameters)
  - `result`: jsonb (final output)
  - `error`: text (failure messages)
  - `created_at`, `updated_at`, `started_at`, `completed_at` timestamp columns
- Add indexes on `status` and `user_id` for efficient polling queries
- Define Drizzle relations linking jobs to users and chats

**Task 2: Create migration in `drizzle/migrations/`**
- Use drizzle-kit generate to produce the migration file
- Include index creation statements for query optimization
- Follow existing migration file patterns (e.g., `0000_sweet_metal_master.sql`,
`0001_add_calendar_notes.sql`)

**Task 3: Create `lib/services/jobs.ts`**
- Implement the following exported functions: `createJob()`, `getJob()`,
`updateJobStatus()`, `updateJobProgress()`, `completeJob()`, `failJob()`,
`getPendingJobs()`, `getUserJobs()`
- `getPendingJobs()` must use an optimistic locking pattern — atomically update
`status` to `'processing'` and return the claimed job to prevent duplicate
processing
- Include TypeScript type definitions for job input/output structures that
mirror the current `submit` server action parameters
===============================================================================

Task: 2

Extract agent processing logic into a self-contained worker module and expose it
via a cron-triggerable API route. The goal is to make job processing independent
of the Server Action request lifecycle.

**Task 1: Create `lib/workers/report-processor.ts`**
- Extract core processing logic from `processEvents` and
`processResolutionSearch` in `app/actions.tsx`
- Remove all `createStreamableUI` and `createStreamableValue` dependencies
- Replace streaming with callback-based progress reporting using parameters like
`onProgress` and `onPartialResult`
- Accept fully serializable inputs and return serializable outputs suitable for
storage in the jobs table
- Maintain the existing agent call chain: taskManager → inquire/researcher →
querySuggestor, collecting results into a data structure instead of streaming UI

**Task 2: Create `lib/workers/job-worker.ts`**
- Implement `processNextJob()`: claims one pending job (atomically via the job
service layer, setting status to `'processing'`), runs the report processor with
progress-updating callbacks, and handles completion/failure
- Implement `runWorkerLoop()`: continuously polls for and processes pending jobs
with a configurable poll interval
- On any exception: catch the error, call `failJob()` with error details, and
continue to the next job without crashing the loop

**Task 3: Create `app/api/workers/process-jobs/route.ts`**
- Expose a POST endpoint
- Implement an authorization check (e.g., validating a secret header or Vercel
cron authentication)
- Call `processNextJob()` in a loop bounded by a time limit (e.g., 25 seconds to
stay within Vercel function timeout)
- Return a response with the count of processed jobs and any errors encountered,
for monitoring purposes
===============================================================================

Task: 3

Modify the `submit` server action to enqueue jobs for async processing and add
the necessary API endpoints for status polling and result reconstruction.

**Task 1: Modify `app/actions.tsx`**
- Add logic to detect when async processing is appropriate (e.g., complex
queries, resolution search)
- For **async paths**: serialize input parameters, call `createJob()`, and
return immediately with `jobId` and `status: 'queued'`
- For **simple/fast operations** (e.g., hard-coded FAQ answers, simple
inquiries): keep the existing synchronous behavior
- Add a new return type variant that includes `jobId` and `status: 'queued'`
alongside the existing streaming return shape
- Update AIState handling so initial state is set at job creation and final
state is set by the worker

**Task 2: Create `app/api/jobs/[id]/route.ts`**
- Expose a GET endpoint
- Authenticate the request using `getCurrentUserIdOnServer()`
- Verify the requesting user owns the job before returning data
- Return job status, progress percentage, and result data when completed
- For completed jobs with UI-generating results, include serialized component
data for client-side rendering

**Task 3: Create `lib/utils/job-result-renderer.ts`**
- Implement functions to convert serialized job results back into React
components
- Map result types to appropriate UI components (e.g., BotMessage,
SearchResults, GeoJsonLayer)
- Update `getUIStateFromAIState` in `app/actions.tsx` to handle job-based
message reconstruction
- Ensure completed job results are persisted to chat history using the existing
`saveChat` mechanism
===============================================================================

Task: 4

Integrate the async job system into the client-side UI with a polling hook, a
progress display component, and updates to the chat interface.

**Task 1: Create `lib/hooks/use-job-status.ts`**
- Implement a `useJobStatus(jobId)` React hook
- Poll the job status API at a configurable interval (e.g., 1–2 seconds while
the job is processing)
- Return job status, progress, result, and error states
- Automatically stop polling when the job reaches a terminal state (`completed`
or `failed`)
- Handle network errors gracefully with retry logic

**Task 2: Create `components/job-progress.tsx`**
- Build a component that displays job status, a progress bar, and elapsed time
- Handle all job states: queued (waiting indicator), processing (progress bar),
completed (success display), failed (error message)
- Optionally display partial results as they become available if stored during
processing
- Style consistently with existing spinner and loading UI patterns in the
codebase

**Task 3: Update `components/chat.tsx` (or equivalent)**
- Detect when the `submit` action returns a job-based response (i.e., a `jobId`
is present)
- When a job ID is returned, render the `JobProgress` component and begin
polling via `useJobStatus`
- On job completion, render the result using the job result renderer from
`lib/utils/job-result-renderer.ts`
- Keep the UI interactive during processing (user can view history, start new
chats, etc.)
- Handle edge cases: page refresh during processing (restore polling from job
ID), job timeout, and user cancellation

💡 Iterate on the plan with: `@coderabbitai` <feedback>

Example Feedback
- `@coderabbitai` You can skip phase 3. Add a simple unit test case for phase 2.
- `@coderabbitai` For design choice 1 go ahead with option 3 and replan.

---

💬 Have feedback or questions? Drop into our discord!

[ngoiyaeric]

https://workflow-sdk.dev/