Queues & 비동기 작업 관리

import { handleCallback, send } from '@vercel/queue'

export async function enqueueDocument(docId: string, chunks: string[]) {
  await Promise.all(
    chunks.map((chunk, i) =>
      send(
        'embedding-jobs',
        { chunk, docId: `${docId}#${i}` },
        { idempotencyKey: `${docId}#${i}` }
      )
    )
  )
}

export const POST = handleCallback(
  async (job) => {
    const embedding = await generateEmbedding(job.chunk)
    await upsertVector({ docId: job.docId, chunk: job.chunk, embedding })
  },
  {
    visibilityTimeoutSeconds: 600,
    retry: (_error, metadata) => {
      if (metadata.deliveryCount > 5) return { acknowledge: true }
      return { afterSeconds: Math.min(300, 2 ** metadata.deliveryCount * 5) }
    },
  }
)

{
  "functions": {
    "app/api/queues/embed/route.ts": {
      "experimentalTriggers": [{ "type": "queue/v2beta", "topic": "embedding-jobs" }]
    }
  }
}

import { send } from '@vercel/queue'
import { sleep } from 'workflow'

export async function batchAnalysisWorkflow(batchId: string) {
  'use workflow'

  const data = await collectData(batchId)
  const chunks = await splitIntoChunks(data)

  // Queues로 fan-out
  await enqueueProcessingChunks(batchId, chunks)

  // 완료 대기 방식은 별도 status store나 webhook/hook으로 구현합니다.
  await sleep('1h')

  return aggregateResults(batchId)
}

async function enqueueProcessingChunks(batchId: string, chunks: string[]) {
  'use step'

  await Promise.all(
    chunks.map((chunk, index) =>
      send(
        'processing-jobs',
        { batchId, chunk, index },
        { idempotencyKey: `${batchId}:${index}` }
      )
    )
  )
}

import { embed } from 'ai'

export const POST = handleCallback(async (job) => {
  const { embedding } = await embed({
    model: 'openai/text-embedding-3-small',
    value: job.chunk,
    providerOptions: {
      gateway: {
        tags: ['queue:embedding', `batch:${job.batchId}`],
      },
    },
  })
  // ...
})