Skip to content

feat: Integrate Vercel Queues for asynchronous resolution search #702

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
ngoiyaeric wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat: Integrate Vercel Queues for asynchronous resolution search #702

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 8 additions & 91 deletions app/actions.tsx
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,8 @@ import type { FeatureCollection } from 'geojson'
import { Spinner } from '@/components/ui/spinner'
import { Section } from '@/components/section'
import { FollowupPanel } from '@/components/followup-panel'
import { inquire, researcher, taskManager, querySuggestor, resolutionSearch, type DrawnFeature } from '@/lib/agents'
import { inquire, researcher, taskManager, querySuggestor, type DrawnFeature } from '@/lib/agents'
import { send } from '@vercel/queue';
import { writer } from '@/lib/agents/writer'
import { saveChat, getSystemPrompt } from '@/lib/actions/chat'
import { Chat, AIMessage } from '@/lib/types'
Expand Down Expand Up @@ -101,96 +102,12 @@ async function submit(formData?: FormData, skip?: boolean) {

async function processResolutionSearch() {
try {
const streamResult = await resolutionSearch(messages, timezone, drawnFeatures, location);

let fullSummary = '';
for await (const partialObject of streamResult.partialObjectStream) {
if (partialObject.summary) {
fullSummary = partialObject.summary;
summaryStream.update(fullSummary);
}
}

const analysisResult = await streamResult.object;
summaryStream.done(analysisResult.summary || 'Analysis complete.');

if (analysisResult.geoJson) {
uiStream.append(
<GeoJsonLayer
id={groupeId}
data={analysisResult.geoJson as FeatureCollection}
/>
);
}

messages.push({ role: 'assistant', content: analysisResult.summary || 'Analysis complete.' });

const sanitizedMessages: CoreMessage[] = messages.map((m: any) => {
if (Array.isArray(m.content)) {
return {
...m,
content: m.content.filter((part: any) => part.type !== 'image')
} as CoreMessage
}
return m
})

const currentMessages = aiState.get().messages;
const sanitizedHistory = currentMessages.map((m: any) => {
if (m.role === "user" && Array.isArray(m.content)) {
return {
...m,
content: m.content.map((part: any) =>
part.type === "image" ? { ...part, image: "IMAGE_PROCESSED" } : part
)
}
}
return m
});
const relatedQueries = await querySuggestor(uiStream, sanitizedMessages);
uiStream.append(
<Section title="Follow-up">
<FollowupPanel />
</Section>
);

await new Promise(resolve => setTimeout(resolve, 500));

aiState.done({
...aiState.get(),
messages: [
...aiState.get().messages,
{
id: groupeId,
role: 'assistant',
content: analysisResult.summary || 'Analysis complete.',
type: 'response'
},
{
id: groupeId,
role: 'assistant',
content: JSON.stringify({
...analysisResult,
image: dataUrl,
mapboxImage: mapboxDataUrl,
googleImage: googleDataUrl
}),
type: 'resolution_search_result'
},
{
id: groupeId,
role: 'assistant',
content: JSON.stringify(relatedQueries),
type: 'related'
},
{
id: groupeId,
role: 'assistant',
content: 'followup',
type: 'followup'
}
]
});
await send('resolution-search', { messages, timezone, drawnFeatures, location });

@coderabbitai coderabbitai Bot Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win

Consider an idempotency key for the enqueue call.

send() supports deduplication: idempotencyKey provides deduplication for the full retention window. Since processResolutionSearch is fire-and-forget and not obviously guarded against duplicate invocation (e.g., double form submits, client retries), passing a stable idempotencyKey (derived from the request/groupeId) would prevent duplicate resolution-search jobs from being enqueued and processed.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@app/actions.tsx` at line 105, The resolution-search enqueue in
processResolutionSearch currently has no deduplication, so duplicate requests
can create multiple jobs. Update the send() call in processResolutionSearch to
pass a stable idempotencyKey derived from the request or groupeId, and keep it
consistent for equivalent submissions so retries or double submits are deduped
by send(). Use the existing identifiers processResolutionSearch and
send('resolution-search', ...) to locate the call site.

summaryStream.update('Resolution search task enqueued. Please check back later for results.');
summaryStream.done();
isGenerating.done(false);
uiStream.done();
return; // Exit early as the actual processing will happen in the queue
Comment on lines +105 to +110

@qodo-code-review qodo-code-review Bot Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Action required

2. Queued results never surfaced 🐞 Bug ≡ Correctness

The resolution_search server action now only enqueues a job and closes UI streams without writing
any assistant response/resolution_search_result messages to aiState, so chats won’t be saved
and there is no in-repo mechanism to later render the background result. Users will see “enqueued”
once, but the actual analysis output can’t appear in history/reload/share flows.
Agent Prompt 
## Issue description
The new async queue path enqueues work and exits without updating `aiState` with an assistant `response` (required for persistence) or any placeholder/correlation data needed to later attach the real `resolution_search_result`.

## Issue Context
- `onSetAIState` only persists chats when there is at least one message with `type === 'response'`.
- UI reconstruction expects `resolution_search_result` messages to render the carousel + GeoJSON.

## Fix Focus Areas
- app/actions.tsx[54-138]
- app/actions.tsx[515-520]
- app/actions.tsx[659-681]
- app/api/queues/resolution-search/route.ts[5-17]

## Expected fix shape
- When enqueuing, include a correlation id (e.g., `chatId` + `groupeId` and/or userId) in the queue message.
- Update `aiState` with an assistant `response` like “Queued resolution search…” so the chat is persistable.
- Persist the queue result (in the callback) keyed by that correlation id.
- Add a retrieval mechanism (polling endpoint/server action) that loads the persisted result and appends a `resolution_search_result` message (and any related/followup messages) into `aiState` for UI rendering.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

} catch (error) {
console.error('Error in resolution search:', error);
summaryStream.error(error);
Expand Down
18 changes: 18 additions & 0 deletions app/api/queues/resolution-search/route.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
import { handleCallback } from '@vercel/queue';
import { resolutionSearch } from '@/lib/agents/resolution-search';
import { CoreMessage } from 'ai';

export const POST = handleCallback(async (message: { messages: CoreMessage[], timezone: string, drawnFeatures?: any[], location?: { lat: number, lng: number } }, metadata) => {

@coderabbitai coderabbitai Bot Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win

Type the message payload with DrawnFeature[] instead of any[].

drawnFeatures?: any[] loses the type contract that app/actions.tsx establishes via import { ..., type DrawnFeature } from '@/lib/agents'. Importing and reusing DrawnFeature here keeps the producer/consumer payload contract in sync and catches shape drift at compile time.

♻️ Proposed fix 
 import { handleCallback } from '`@vercel/queue`';
 import { resolutionSearch } from '`@/lib/agents/resolution-search`';
 import { CoreMessage } from 'ai';
+import type { DrawnFeature } from '`@/lib/agents/resolution-search`';

-export const POST = handleCallback(async (message: { messages: CoreMessage[], timezone: string, drawnFeatures?: any[], location?: { lat: number, lng: number } }, metadata) => {
+export const POST = handleCallback(async (message: { messages: CoreMessage[], timezone: string, drawnFeatures?: DrawnFeature[], location?: { lat: number, lng: number } }, metadata) => {
📝 Committable suggestion

‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.

Suggested change
export const POST = handleCallback(async (message: { messages: CoreMessage[], timezone: string, drawnFeatures?: any[], location?: { lat: number, lng: number } }, metadata) => {
import { handleCallback } from '`@vercel/queue`';
import { resolutionSearch } from '`@/lib/agents/resolution-search`';
import { CoreMessage } from 'ai';
import type { DrawnFeature } from '`@/lib/agents/resolution-search`';
export const POST = handleCallback(async (message: { messages: CoreMessage[], timezone: string, drawnFeatures?: DrawnFeature[], location?: { lat: number, lng: number } }, metadata) => {
🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@app/api/queues/resolution-search/route.ts` at line 5, The POST handler in
resolution-search/route.ts is typing drawnFeatures as any[], which breaks the
shared payload contract. Update the message parameter in the handleCallback
callback to reuse the DrawnFeature type from `@/lib/agents` instead of any[],
matching the type used by app/actions.tsx. This keeps the producer and consumer
in sync and preserves compile-time validation of the drawnFeatures shape.

console.log(`Processing resolution search message ${metadata.messageId} for topic ${metadata.topicName}`);
try {
// Execute the resolutionSearch logic asynchronously
const result = await resolutionSearch(message.messages, message.timezone, message.drawnFeatures, message.location);
console.log(`Resolution search message ${metadata.messageId} processed successfully.`);
// In a real application, you would likely store this result in a database
// or notify the original requestor through a webhook or websocket.
return { status: 'success', result };
Comment on lines +5 to +13

@qodo-code-review qodo-code-review Bot Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Action required

1. Non-serializable queue result 🐞 Bug ≡ Correctness

The queue callback returns the streamObject result wrapper from resolutionSearch instead of
materializing the final structured object, so the returned value is not a plain JSON payload and the
analysis result is never concretely produced/consumed. This can break the callback contract and
prevents any usable result from being stored or forwarded.
Agent Prompt 
## Issue description
`resolutionSearch()` returns a `streamObject(...)` result (stream wrapper). The queue callback currently returns that wrapper directly (`{ status: 'success', result }`), which is not the final structured analysis and is unlikely to be JSON-serializable.

## Issue Context
Other call sites in the repo consume `partialObjectStream` and/or `await result.object` to obtain the final structured object.

## Fix Focus Areas
- app/api/queues/resolution-search/route.ts[5-17]
- lib/agents/resolution-search.tsx[41-90]

## Expected fix shape
- In the queue callback:
  - `const stream = await resolutionSearch(...)`
  - `const analysis = await stream.object` (and optionally iterate `partialObjectStream` if needed)
  - Persist `analysis` somewhere (DB/kv) keyed by correlation id, or publish/notify.
  - Return a small serializable ack (e.g., `{ status: 'success' }`) rather than the stream wrapper.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

Comment on lines +5 to +13

@coderabbitai coderabbitai Bot Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🗄️ Data Integrity & Integration | 🔵 Trivial | 🏗️ Heavy lift

Implement the TODO: persist or forward the result.

The comment on lines 11-12 correctly flags that result needs to be stored or relayed back to the requester — right now it's just returned from the handler, which only acknowledges the message per Return normally = acknowledge // Throw = retry with backoff; the value itself has no destination. This is the other half of the enqueue-side gap flagged in app/actions.tsx (lines 103-121) — without persistence (DB row, cache entry, webhook/websocket push), the resolution search result is unrecoverable once this consumer returns.

Want me to sketch a persistence + notification path (e.g., write to a datastore keyed by a job id returned from send, then have the client poll or subscribe)?

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@app/api/queues/resolution-search/route.ts` around lines 5 - 13, The POST
handler in handleCallback currently computes resolutionSearch and only returns
the result, so the data is lost once the callback acknowledges the message.
Update the resolution-search flow in POST/resolutionSearch to persist the result
or forward it to the requester using a durable path such as a database/cache
keyed by the message/job id, or a webhook/websocket notification. Make sure the
enqueue/send side in app/actions.tsx and the consumer side agree on a shared
identifier so the caller can later fetch or receive the stored result instead of
relying on the handler return value.

} catch (error) {
console.error(`Error processing resolution search message ${metadata.messageId}:`, error);
throw error; // Re-throw to trigger retry mechanism
}
Comment on lines +14 to +17

@qodo-code-review qodo-code-review Bot Jul 6, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Remediation recommended

3. Invalid timezone retry storm 🐞 Bug ☼ Reliability

The queue worker rethrows all errors to trigger retries, but resolutionSearch uses a user-provided
timezone in toLocaleString, which throws on invalid timezones; a permanently bad payload will be
retried repeatedly. This can cause avoidable queue backlogs and repeated failures from a single
malformed request.
Agent Prompt 
## Issue description
User-provided `timezone` is used in `toLocaleString({ timeZone: timezone })` and may throw if invalid. The queue handler rethrows, which means the queue will retry a permanently invalid payload.

## Issue Context
Timezone is read from `formData` and passed through the queue message.

## Fix Focus Areas
- app/actions.tsx[55-61]
- lib/agents/resolution-search.tsx[42-51]
- app/api/queues/resolution-search/route.ts[14-17]

## Expected fix shape
- Validate/sanitize timezone before enqueuing and/or in the worker:
  - If invalid, default to `'UTC'`.
  - Optionally treat invalid timezone as non-retryable (ack with error stored for the requester) instead of throwing.
- Consider catching `RangeError` from `toLocaleString` inside `resolutionSearch` and falling back to `'UTC'`.

ⓘ Copy this prompt and use it to remediate the issue with your preferred AI generation tools

});
9 changes: 9 additions & 0 deletions vercel.json
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
{
"functions": {
"app/api/queues/resolution-search/route.ts": {
"experimentalTriggers": [
{ "type": "queue/v2beta", "topic": "resolution-search" }
]
}
}
}