任务队列（@weifuwujs/tasks）

后台任务的最小完整形态：至少一次投递 + 幂等键去重 + 指数退避/死信 + 优雅停止， 四驱动（memory/sqlite/postgres/redis）共用同一套契约测试。适合发欢迎邮件、生成报表、清理数据这类 "失败要重试、重复要无害"的工作。

不适合的：秒级定时精度、cron 表达式、跨任务的分布式锁（见文末"边界"）。

十分钟接入

import { createApp } from '@weifuwujs/weifuwu'
import { createTasks } from '@weifuwujs/tasks'
import { sqliteTaskStore } from '@weifuwujs/tasks/sqlite'

const store = sqliteTaskStore('data/tasks.db') // 构造即迁移（幂等）
const tasks = createTasks({ store })

tasks.handle('welcome-email', async (payload: { email: string }) => {
  console.log('send welcome to', payload.email)
})

const app = createApp()
tasks.install(app) // 启动 worker；app.close()/SIGTERM 时优雅停止

await tasks.enqueue(
  'welcome-email',
  { email: '[email protected]' },
  { idempotencyKey: 'welcome:user-1' },
)

最小可跑示例（零依赖，node 直接运行；test/batteries/readme-examples.test.ts 会真跑它）：

import assert from 'node:assert/strict'
import { createTasks, memoryTaskStore } from '@weifuwujs/tasks'

const tasks = createTasks({ store: memoryTaskStore() })
const handled: string[] = []
tasks.handle('welcome-email', async (payload) => { handled.push((payload as { email: string }).email) })

await tasks.enqueue('welcome-email', { email: '[email protected]' }, { idempotencyKey: 'welcome:1' })
await tasks.enqueue('welcome-email', { email: '[email protected]' }, { idempotencyKey: 'welcome:1' })
await tasks.drain()

assert.equal(handled.length, 1)                          // 幂等键：重复入队只执行一次
assert.equal((await tasks.list({ statuses: ['done'] })).length, 1)  // 已完成 1 个
assert.equal((await tasks.list({ statuses: ['pending'] })).length, 0)  // 没有待办（重复入队被幂等键去重）

周期任务（every：首次立即 + 完成后排下一次 + 不选主）：

import assert from 'node:assert/strict'
import { createTasks, memoryTaskStore } from '@weifuwujs/tasks'

const tasks = createTasks({ store: memoryTaskStore() })
const runs: string[] = []
tasks.handle('hourly-report', async () => { runs.push('run') })

tasks.every('hourly-report', 60_000)   // 周期任务：首次立即执行，之后每 60s 排一次

await tasks.runOnce()
assert.equal(runs.length, 1)                                   // 首次立即，不等下一个周期

await tasks.runOnce()
assert.equal(runs.length, 1)                                   // 周期未到：不重复执行
const pending = await tasks.list({ statuses: ['pending'] })
assert.ok(pending[0]!.runAt.getTime() > Date.now())            // 已排回下一次（也是「删除 ≠ 暂停」的执行体）

示例标注：ts exec = 由测试真跑；ts fragment = 片段（依赖前后文，不独立运行）；自足示例带包标签（如 ts tasks），由测试编译。

四步：选驱动 → createTasks → handle → install(app)。之后 enqueue 只落库， 由进程内 worker 轮询执行；不需要额外的调度器。

API 速览

createTasks(options)

| 选项 | 默认 | 说明 | | --- | --- | --- | | store | 必填 | 四驱动之一或你自己的 TaskStore | | concurrency | 1 | 单进程并发执行数 | | pollIntervalMs | 1000 | 轮询间隔 | | visibilityTimeoutMs | 60000 | 可见性超时：执行中崩溃后多久可被重领 | | retry | { baseMs: 1000, maxMs: 900000 } | 指数退避（base * 2^(attempt-1)，封顶） | | warnAfterMs | 30000 | 单任务超时只告警不中断 | | autoStart | true | 进程内 worker 默认开启；测试可关 | | now / workerId / logger | — | 注入时钟、worker 标识、日志器 |

tasks 实例：handle / enqueue / every / install / start / stop / runOnce / drain / retry / list / prune / pendingNames。

enqueue(name, payload?, options?) → { id, created, task }： options = { runAt?, delay?, idempotencyKey?, maxAttempts? }；created: false 表示幂等命中（返回已有任务，不覆盖）。

四驱动

| 驱动 | 入口 | 建表 | 说明 | | --- | --- | --- | --- | | memory | @weifuwujs/tasks | 无需 | 单进程测试/演示 | | sqlite | @weifuwujs/tasks/sqlite | weifuwu-tasks migrate --store sqlite:data/tasks.db | 默认生产驱动；单进程写（多进程用 PG） | | postgres | @weifuwujs/tasks/postgres | weifuwu-tasks migrate --store postgres://… | FOR UPDATE SKIP LOCKED，多进程安全；默认读 DATABASE_URL | | redis | @weifuwujs/tasks/redis | 无需（tasks migrate 为 no-op 提示） | ZSET + 单段 Lua 原子领取；默认读 REDIS_URL |

切换驱动只改一行：

createTasks({ store: postgresTaskStore(process.env.DATABASE_URL!) })

自定义驱动

实现 TaskStore 接口后，用同一套契约测试验证（memory/sqlite/postgres/redis 四驱动跑的就是它）：

import { memoryTaskStore } from '@weifuwujs/tasks'
import { tasksStoreContract } from '@weifuwujs/tasks/testing'

// 你的驱动实现 TaskStore 后，用同一套契约验证
tasksStoreContract('my-store', () => memoryTaskStore())

工厂会被套件多次调用：每次须返回独立的 store（sqlite 驱动用 ':memory:' 或唯一路径）—— 复用同一文件会状态泄漏并假红。

周期任务（every）

tasks.handle('hourly-report', async () => { /* 生成报表 */ })
tasks.every('hourly-report', 3600_000) // 注意：必须在 handle 之后、start 之前注册

三段语义（有跨驱动契约测试钉住：memory / sqlite / postgres / redis 同一套 tasksFacadeContract）：

1. 首次立即到期（不是等一个周期）；
2. 完成后才排下一次：runAt = now() + intervalMs（执行期间不会叠加同名的下一次）；
3. 宕机后 catch-up 只补一次：跨 N 个周期重启也只补跑一遍，不逐周期补跑。

边界：

• 同名注册 = last wins（间隔与 payload 都取最后一次 every）；
• 非法 intervalMs（0 / 负数 / NaN / Infinity）→ INVALID_ARGUMENT 且零入队（不静默排期）；
• 不选主：多个 worker 共用一个库时，每个进程都会各自调度 —— 并发窗口下同一个周期任务可能被重复投递并重复执行。 队列保证至少一次、不做隐式去重；要「只跑一次」用业务幂等或显式 idempotencyKey（见上节）。

周期任务运维（CLI）

weifuwu-tasks list --name hourly-report --status pending --json   # 同一周期任务有几份 pending
weifuwu-tasks prune --status pending --older-than 0s --json       # 清掉重复副本（deleted 计数）
weifuwu-tasks prune --older-than 7d --dry-run                     # 破坏性清理先预览：零删除

• 重复投递看得见：多 worker 并发投递会在库里留下同名 pending 副本，list --name <任务> 的数字就是现场证据；
• --dry-run 零删除：报告 {dryRun, matched, deleted: 0}（人句子模式为「将清理 N 个任务（dry-run，零删除）」）， 计数与真删共用同一条谓词，所以预览值就是真删值；
• 删除 ≠ 暂停：周期任务的排期由进程内的 every 注册驱动 —— 删掉 pending 副本后，下一次 tick 的 ensureScheduled 会把它排回来。要停周期任务请停进程或改代码，CLI 不提供「暂停」。
• 非法参数不静默：--status 只接受 pending/running/done/dead、--limit 只接受正整数 —— 否则 USAGE exit 2 且 stdout 为空（不会静默当「无匹配」让你以为清理过了）。

至少一次与幂等

• attempts 在领取时 +1：即使进程硬崩溃、complete 从未执行，本次尝试也被计数，不会无限重跑；
• 锁过期回队：running 超过 visibilityTimeoutMs 未完成 → 回 pending 重领； attempts 已达 maxAttempts → 置 dead（lastError: visibility timeout exceeded）；
• 失败退避：fail(id, { error, retryAt }) 带 retryAt → pending + 指数退避；retryAt: null → dead；
• 幂等键：enqueue 携 idempotencyKey 时，重复投递返回已有任务（created: false），不覆盖原任务。 分布式下四驱动都用唯一约束/Lua/HSETNX 保证原子；
• 未知 handler：任务保持 pending 并告警一次（不判死），等代码部署补齐后再执行。

推荐幂等键命名：<动作>:<业务主键>（如 welcome:<userId>）。

auth 接线 recipe：欢迎邮件

注册成功发一封欢迎邮件，重复触发（hooks 重投、注册重试）只发一次：

import { createAuth } from '@weifuwujs/auth'
import { sqliteAuthStore } from '@weifuwujs/auth/sqlite'
import { createTasks } from '@weifuwujs/tasks'
import { sqliteTaskStore } from '@weifuwujs/tasks/sqlite'

const tasks = createTasks({ store: sqliteTaskStore('data/tasks.db') })
tasks.handle('welcome-email', async (payload: { userId: string }) => {
  // 接你的邮件服务；失败会按退避重试，幂等键保证只发一次
  console.log('welcome', payload.userId)
})

export const auth = createAuth({
  store: sqliteAuthStore('data/auth.db'),
  hooks: {
    onRegister: (user) => {
      void tasks.enqueue(
        'welcome-email',
        { userId: user.id },
        { idempotencyKey: `welcome:${user.id}` },
      )
    },
  },
})

完整可跑版本见 test/tasks-recipes.test.ts。

生命周期

• tasks.install(app)：进程内 worker 随应用启动；app.close()（含 CLI 的 SIGINT/SIGTERM）时优雅停止： 停止取新任务，超时未完成的任务归还（release，保留下次重试，不丢）；
• tasks work：独立 worker 进程，加载与 weifuwu start 相同的构建产物：

weifuwu-tasks work                 # 常驻 worker（Ctrl+C / SIGTERM 停止）
weifuwu-tasks work --store postgres://…

• 多进程部署：web 进程设 autoStart: false，worker 进程跑 tasks work（或单独进程 createTasks().start()）。

CLI 运维

weifuwu-tasks migrate --store sqlite:data/tasks.db   # 建表（幂等）
weifuwu-tasks list --status dead --json              # 查看（--name/--limit/--json）
weifuwu-tasks retry <id>                             # 手动重驱死信
weifuwu-tasks retry --all-dead
weifuwu-tasks prune --older-than 7d                  # 清理 done/dead（默认 7d）
weifuwu-tasks prune --older-than 7d --dry-run        # 预览：零删除，报告会删多少
weifuwu-tasks prune --status pending --older-than 0s # 清掉排期中的重复副本（见下节：这不是暂停）
weifuwu-tasks stats --json                           # 聚合：四状态计数 / nextRunAt / 重复副本名单

--store 与 auth 同惯例：sqlite:<path>（默认 sqlite:data/tasks.db）、postgres://…、redis://…。

stats 是单次聚合（memory/redis 单趟遍历，sqlite/postgres 聚合 SQL）：2000 任务下全量 list --json 要 784,682 B，而 stats --json 只要 906 B（866×，npm run tasks:evidence 可复跑）。 优先级与全栈指标本轮不做（取证裁决见 docs/tasks-ops-ruling.md）。

CLI 惯例（1.3 起五包统一，见 docs/alignment.md）：无命令 / 未知命令 → USAGE exit 2 且不写 stdout； --help → stdout + exit 0（此前会报未知子命令）；退出码 0 成功 / 1 运行时错误 / 2 用法错误。

测试辅助（/testing）

自定义 TaskStore 驱动与内置四驱动共跑同一套契约：

import { tasksStoreContract } from '@weifuwujs/tasks/testing'

tasksStoreContract('my-store', () => myTaskStore())
// 第三个参数可选：{ skip: true } 或 { skip: '未实现清理' }

契约覆盖：enqueue/get/幂等键、claim 原子性与 attempts、退避/死信、锁过期回收、prune 边界。

边界（不做与规模）

• 不做：cron 表达式、任务优先级、跨任务分布式锁、强杀超时任务（warnAfterMs 只告警）； redis/sqlite 驱动不承诺多进程精确调度（redis 领取原子，但 list/prune 为 O(n) 扫描）；
• 规模假设：P1 面向中小规模（单表/单 ZSET，数万级任务）；更大规模再谈分片与索引（P2）。