BossAgentsDigitalStaff 统一 Pipeline 重构方案
1. 现状分析
当前架构：无统一 pipeline
DigitalStaff.vue (3051 行, 20+ 个独立 fetch)
│
├── fetchData() → GET /api/digital-staff/list
├── fetchTasks() → GET /api/digital-staff/tasks
│ → GET /api/digital-staff/task-stats
├── fetchStaffLogs(id) → GET /api/digital-staff/logs
├── fetchStaffReports(s) → GET /api/reports/list
├── confirmSummon() → POST /api/digital-staff/run
├── checkPendingConfirms()→ GET /api/digital-staff/resume
├── createTask() → POST /api/digital-staff/tasks
├── assignTask() → POST /api/digital-staff/tasks/:id/assign
├── completeTask() → POST /api/digital-staff/tasks/:id/complete
├── deleteTask() → DELETE /api/digital-staff/tasks/:id
└── ... 其他 N 个独立 fetch
核心问题
| 问题 | 说明 | 后果 |
|---|---|---|
| 无请求去重 | 切换员工时 fetchStaffReports 重复拉全量报告 | 性能浪费 |
| 无请求取消 | 快速切换员工时旧请求覆盖新状态 | 竞态 → .filter 报错 |
| 无批处理保证 | selectStaff 同时 fire logs + reports，一个成功另一个失败 | 面板数据不一致 |
| 无重试/退避 | 失败立即结束，不回退 | 服务器尖峰时全部失败 |
| 无统一错误处理 | 每个函数自己 try/catch，各自管理 loading 状态 | 错误重复，状态混乱 |
| 状态碎片化 | 18 个独立 ref + 4 个 computed，无一致性契约 | 调试困难 |
2. Pipeline 架构设计
2.1 核心概念
pipeline.request({ type, id, depends: [] })
→ 去重: 相同 type+id 的进行中请求被消除
→ 取消: 旧 type 被新 type 替代时自动 abort 旧请求
→ 分组: depends 用于表达原子性需求（全组成功或回滚）
→ 重试: 失败后按指数退避重试（最多 3 次）
→ 状态更新: 成功/失败后自动更新对应 ref
2.2 API 设计
// composables/usePipeline.js
interface PipelineOptions {
retry?: number // 重试次数，默认 2
retryDelay?: number // 初始退避延迟 ms，默认 500
abortSignal?: AbortSignal
}
interface RequestSpec<T = any> {
type: string // 'staff:list' | 'staff:logs' | 'staff:reports' | 'tasks:list' | ...
id?: string // 逻辑 id，用于去重和取消
group?: string // 批次组，同组内任一失败则整组回滚
handler: () => Promise<T> // 实际 fetch 函数
onSuccess?: (data: T) => void
onError?: (err: Error) => void
}
function usePipeline() {
const active: Ref<Map<string, AbortController>>
const pending: Ref<Set<string>>
const errors: Ref<Map<string, Error>>
function request<T>(spec: RequestSpec<T>, opts?: PipelineOptions): Promise<T>
function cancelGroup(group: string): void // 取消整组进行中的请求
function cancelAll(): void // 取消全部
function getError(type: string): Error | null // 获取最近错误
function clearErrors(): void
}
2.3 DigitalStaff Pipeline 层
// composables/useDigitalStaffPipeline.js
// 在 DigitalStaff.vue 之上加一层 pipeline 封装
export function useDigitalStaffPipeline() {
const pipeline = usePipeline()
// 状态 ref — 组件只消费这些
const state = reactive({
staffList: [],
selectedStaff: null,
staffLogs: [],
staffReports: [],
taskList: [],
taskStats: {},
loading: { staff: false, logs: false, reports: false, tasks: false },
errors: {}
})
// ✅ 原子性操作：selectStaff = logs + reports 要么全成功要么全失败
async function selectStaff(staff) {
// 取消旧的操作
pipeline.cancelGroup('staff-detail')
state.selectedStaff = staff
if (!staff?.id) return
// 统一批次请求: 同 group 内任一失败整组回滚
const results = await Promise.allSettled([
pipeline.request({
type: 'staff:logs',
id: staff.id,
group: 'staff-detail',
handler: () => fetch(`/api/digital-staff/logs?staff_id=${staff.id}`).then(r => r.json()),
onSuccess: (data) => { state.staffLogs = data.data || [] },
}),
pipeline.request({
type: 'staff:reports',
id: staff.id,
group: 'staff-detail',
handler: () => fetch('/api/reports/list').then(r => r.json()),
onSuccess: (data) => { state.staffReports = filterReports(data, staff.name) },
})
])
// 整组失败处理：重置相关状态
if (results.some(r => r.status === 'rejected')) {
state.staffLogs = []
state.staffReports = []
state.errors['staff-detail'] = '部分数据加载失败'
}
}
// ✅ 去重：快速切换 tab 时不重复请求
async function fetchTasks() {
return pipeline.request({
type: 'tasks:list',
handler: async () => {
const [listRes, statsRes] = await Promise.all([
fetch('/api/digital-staff/tasks'),
fetch('/api/digital-staff/task-stats')
])
const listJson = await listRes.json()
const statsJson = await statsRes.json()
state.taskList = (listJson.data?.tasks || []).filter(x => x != null)
state.taskStats = statsJson.data || {}
},
})
}
return { state, selectStaff, fetchTasks, /* ... 其他方法 */, pipeline }
}
3. Pipeline 核心实现
3.1 composables/usePipeline.js
import { ref, reactive, shallowRef } from 'vue'
export function usePipeline(options = {}) {
const defaultRetry = options.retry ?? 2
const defaultRetryDelay = options.retryDelay ?? 500
// 进行中的请求: type → AbortController
const active = shallowRef(new Map())
// 去重 key 集
const pending = ref(new Set())
function dedupKey(spec) {
return spec.group ? `${spec.group}:${spec.type}:${spec.id ?? ''}`
: spec.type + (spec.id ? `:${spec.id}` : '')
}
async function request(spec, opts = {}) {
const retry = opts.retry ?? defaultRetry
const retryDelay = opts.retryDelay ?? defaultRetryDelay
const key = dedupKey(spec)
// 去重：相同 key 的请求进行中则跳过
if (pending.value.has(key)) {
throw new Error(`DUPLICATE: ${key}`)
}
// 取消：旧 group 请求自动 abort
if (spec.group) {
cancelGroup(spec.group)
}
const controller = new AbortController()
pending.value = new Set(pending.value).add(key)
active.value = new Map(active.value).set(key, controller)
let lastError = null
for (let attempt = 0; attempt <= retry; attempt++) {
try {
const result = await spec.handler(controller.signal)
spec.onSuccess?.(result)
return result
} catch (err) {
lastError = err
if (err.name === 'AbortError') break // 主动取消不重试
if (attempt < retry) {
await delay(retryDelay * Math.pow(2, attempt))
continue
}
spec.onError?.(err)
}
}
throw lastError
// cleanup
finally {
pending.value = new Set([...pending.value].filter(k => k !== key))
active.value = new Map([...active.value].filter(([k]) => k !== key))
}
}
function cancelGroup(group) {
for (const [key, controller] of active.value) {
if (key.startsWith(`${group}:`)) {
controller.abort()
}
}
}
function cancelAll() {
for (const controller of active.value.values()) {
controller.abort()
}
pending.value = new Set()
active.value = new Map()
}
return { active, pending, request, cancelGroup, cancelAll }
}
function delay(ms) {
return new Promise(resolve => setTimeout(resolve, ms))
}
3.2 DigitalStaff 接入方式
DigitalStaff.vue 保留模板不变，
🤖
🎁
