blink-queue

MCP 工具

blink_queue_enqueue · blink_queue_schedule · blink_queue_list · blink_queue_stats · blink_queue_cancel · blink_queue_get · blink_queue_dlq_retry · blink_queue_create_queue

快速开始

# Check backend is deployed (required for queue delivery)
blink backend status

# Enqueue a task
blink queue enqueue send-welcome-email --payload '{"to":"[email protected]"}'

# Create a cron schedule
blink queue schedule create daily-digest "0 9 * * *" --payload '{"reportType":"daily"}'

# List tasks
blink queue list --status pending

# View stats
blink queue stats

最低 SDK 版本：@blinkdotnew/sdk >= 2.9.0

重要规则

1. 必须部署后端 — 先部署后端；否则队列无法投递
2. 先编写处理程序 — 在入队前，将 POST /api/queue 添加到 backend/index.ts
3. 调度名称 = taskName — 处理程序接收的调度 name 会作为 taskName
4. 修改处理程序后重新部署 — blink backend deploy
5. 返回 2xx — 非 2xx 响应会触发重试，之后进入死信队列

步骤 1 — 将处理程序添加到 backend/index.ts

import { createClient } from '@blinkdotnew/sdk'

app.post('/api/queue', async (c) => {
  const blink = createClient({ projectId: c.env.BLINK_PROJECT_ID, secretKey: c.env.BLINK_SECRET_KEY })
  // Your queue URL is public. Verify every delivery came from Blink Queue before
  // acting on it — one check covers both enqueued tasks and cron ticks. Read the
  // RAW body: the signature is over the exact bytes, so parse only after verifying.
  const body = await c.req.text()
  const ok = await blink.queue.verify({
    signature: c.req.header('upstash-signature') ?? '',
    body,
    signingKey: c.env.BLINK_QUEUE_SIGNING_KEY,
    nextSigningKey: c.env.BLINK_QUEUE_SIGNING_KEY_NEXT,  // platform-injected; used during key rotation
  })
  if (!ok) return c.json({ error: 'invalid signature' }, 401)

  const { taskName, payload } = JSON.parse(body)

  switch (taskName) {
    case 'send-welcome-email':
      await sendEmail(payload.to, payload.displayName)
      return c.json({ ok: true })

    case 'daily-digest':
      await generateDigest(payload.reportType)
      return c.json({ ok: true })

    default:
      return c.json({ error: `Unknown task: ${taskName}` }, 400)
  }
})

BLINK_QUEUE_SIGNING_KEY 会自动注入后端环境变量，无需手动设置。verify() 会返回 true/false，仅在缺少密钥时抛出错误。

不同来源对应的负载：

来源	投递的请求体
blink.queue.enqueue(taskName, payload)	{ taskName, taskId, payload }
blink.queue.schedule(name, cron, payload)	{ taskName: name, payload }（不含 taskId）

步骤 2 — 部署，然后入队

blink backend deploy
blink queue enqueue send-welcome-email --payload '{"to":"[email protected]"}' --queue emails --delay 10s

SDK 方法

// Enqueue
await blink.queue.enqueue('send-welcome-email', { to: '[email protected]' })
await blink.queue.enqueue('send-welcome-email', { to: '[email protected]' }, {
  queue: 'emails', delay: '5m', retries: 3,
})

// Schedule — name IS the taskName
await blink.queue.schedule('daily-digest', '0 9 * * *', { type: 'daily' })
await blink.queue.schedule('daily-digest', '0 9 * * *', { type: 'daily' }, {
  timezone: 'America/New_York',
})

// Manage schedules
await blink.queue.pauseSchedule('daily-digest')
await blink.queue.resumeSchedule('daily-digest')
await blink.queue.deleteSchedule('daily-digest')

// Named queues (create once, idempotent)
await blink.queue.createQueue('emails', { parallelism: 10 })

// Inspect & manage
const tasks = await blink.queue.list({ status: 'pending', queue: 'emails' })
await blink.queue.cancel(taskId)
const dead = await blink.queue.listDead()
await blink.queue.retryDead(taskId)
const stats = await blink.queue.stats()

Cron 模式

Cron	触发时间
0 9 * * *	每天早上 9 点
0 * * * *	每小时
*/15 * * * *	每 15 分钟
0 9 * * 1	每周一早上 9 点

常见错误

错误	原因	修复方法
任务未投递	后端未部署	blink backend deploy
Unknown task: X	处理程序中缺少对应分支	添加匹配的分支
任务不断重试	处理程序返回非 2xx 响应	返回 c.json({ ok: true })
cancel 返回 409	任务不处于待处理状态	只能取消 'pending' 任务

计费

1 个积分 = 1,000 个任务（每个任务 0.001 个积分）。重试和读取免费。仅限 Pro+ 方案。