124 lines
6.6 KiB
JavaScript
124 lines
6.6 KiB
JavaScript
import { createHash } from 'node:crypto'
|
|
import { mkdir, open, readFile, rename, unlink } from 'node:fs/promises'
|
|
import { dirname } from 'node:path'
|
|
|
|
const hash = value => createHash('sha256').update(value).digest('hex')
|
|
const safe = value => typeof value === 'string' && /^[\w-]{1,100}$/.test(value) ? value : 'unavailable'
|
|
export const success = response => response?.http === 200 && response.body?.code === '200'
|
|
export function requireSuccess(response) {
|
|
if (!success(response)) throw new Error(`HTTP=${response?.http}; code=${safe(response?.body?.code)}; requestId=${safe(response?.requestId)}; traceId=${safe(response?.traceId)}`)
|
|
return response.body.data
|
|
}
|
|
|
|
export async function withFlowJournal(path, binding, action) {
|
|
await mkdir(dirname(path), { recursive: true })
|
|
const lock = await open(`${path}.lock`, 'wx', 0o600)
|
|
try {
|
|
let state
|
|
try { state = JSON.parse(await readFile(path, 'utf8')) }
|
|
catch (error) {
|
|
if (error.code !== 'ENOENT') throw new Error('全流程检查点损坏；禁止新建替代批次')
|
|
state = { version: 1, binding, stages: {}, requests: [], status: 'NEW' }
|
|
}
|
|
if (state.version !== 1 || state.binding !== binding) throw new Error('全流程检查点绑定变化')
|
|
const save = async patch => {
|
|
const next = { ...state, ...patch }
|
|
const file = await open(`${path}.next`, 'wx', 0o600)
|
|
try { await file.writeFile(JSON.stringify(next, null, 2)); await file.sync() } finally { await file.close() }
|
|
await rename(`${path}.next`, path)
|
|
Object.assign(state, next)
|
|
}
|
|
await save({})
|
|
return await action(state, save)
|
|
} finally { await lock.close(); await unlink(`${path}.lock`) }
|
|
}
|
|
|
|
async function bodyDigest(body) {
|
|
if (!(body instanceof FormData)) return hash(JSON.stringify(body ?? null))
|
|
const parts = []
|
|
for (const [key, value] of body) parts.push([key, typeof value === 'string' ? value
|
|
: { name: value.name, type: value.type, sha256: hash(Buffer.from(await value.arrayBuffer())) }])
|
|
return hash(JSON.stringify(parts))
|
|
}
|
|
|
|
// Only public business HTTP. No fixture runner, SQL, deployment or automatic network retry.
|
|
export function createFlowContext({ settings, state, save, fetcher = fetch }) {
|
|
const sessions = []
|
|
const context = {
|
|
marker: '', options: {}, context: {}, createdResources: {}, steps: [],
|
|
registerSession(session) { sessions.push(session) },
|
|
async request(method, path, body, token = context.token, extraHeaders = {}) {
|
|
return send(method, path, body, token, extraHeaders, false)
|
|
},
|
|
async requestWithBody(method, path, body, token = context.token, extraHeaders = {}) {
|
|
return send(method, path, body, token, extraHeaders, false)
|
|
},
|
|
async requestMultipart(method, path, file, token = context.token, extraHeaders = {}) {
|
|
const form = file instanceof FormData ? file : new FormData()
|
|
if (!(file instanceof FormData)) {
|
|
if (!file || file.content === undefined) throw new Error('缺少上传内容')
|
|
form.append('file', new Blob([file.content], { type: file.type || 'application/octet-stream' }), file.name || 'acceptance.bin')
|
|
}
|
|
return send(method, path, form, token, extraHeaders, false)
|
|
},
|
|
async requestBinary(method, path, token = context.token) { return send(method, path, undefined, token, {}, true) },
|
|
apiSuccess(module, name, response) {
|
|
const data = requireSuccess(response)
|
|
context.steps.push({ module, name, passed: true })
|
|
return data
|
|
},
|
|
assert(module, name, response, condition) {
|
|
context.steps.push({ module, name, passed: Boolean(condition) })
|
|
if (!condition) throw new Error(`${module}: ${name}`)
|
|
},
|
|
async login(role) {
|
|
const data = requireSuccess(await send('POST', '/api/auth/login', settings.credentials[role], null, {}, false))
|
|
if (!data?.accessToken || !data.refreshToken) throw new Error(`${role}: 登录令牌不完整`)
|
|
sessions.push(data)
|
|
return data
|
|
},
|
|
async close() {
|
|
const failures = []
|
|
for (const session of sessions) {
|
|
try { requireSuccess(await send('POST', '/api/auth/logout', { refreshToken: session.refreshToken }, session.accessToken, {}, false)) }
|
|
catch { failures.push('SESSION_LOGOUT_FAILED') }
|
|
}
|
|
return failures
|
|
},
|
|
}
|
|
async function send(method, path, body, token, extraHeaders, binary) {
|
|
if (!['GET', 'POST', 'PUT'].includes(method) || !/^\/api\/(auth\/(login|logout)$|platform\/|merchant\/|pda\/)/.test(path)
|
|
|| path.includes('..') || path.includes('#')) throw new Error('全流程请求路径或方法不允许')
|
|
const auth = path === '/api/auth/login' || path === '/api/auth/logout'
|
|
if (path !== '/api/auth/logout' && state.requests.length >= 2500) throw new Error('全流程请求超过安全上限')
|
|
const entry = { sequence: state.requests.length + 1, method, path, startedAt: new Date().toISOString(),
|
|
status: 'INTENT', ...(auth ? {} : { bodyHash: await bodyDigest(body) }) }
|
|
await save({ requests: [...state.requests, entry] })
|
|
let result
|
|
try {
|
|
const response = await fetcher(settings.apiBase.replace(/\/api$/, '') + path, {
|
|
method, redirect: 'error', signal: AbortSignal.timeout(45000),
|
|
headers: { ...(body !== undefined && !(body instanceof FormData) ? { 'Content-Type': 'application/json' } : {}),
|
|
...extraHeaders, ...(token ? { Authorization: `Bearer ${token}` } : {}) },
|
|
...(body !== undefined ? { body: body instanceof FormData ? body : JSON.stringify(body) } : {}),
|
|
})
|
|
if (binary && response.ok && !response.headers.get('content-type')?.includes('json')) {
|
|
const bytes = new Uint8Array(await response.arrayBuffer())
|
|
result = { http: response.status, bytes, buffer: Buffer.from(bytes), contentType: response.headers.get('content-type'), headers: Object.fromEntries(response.headers) }
|
|
} else {
|
|
const envelope = await response.json()
|
|
result = { http: response.status, body: envelope, requestId: envelope?.requestId || response.headers.get('x-request-id'),
|
|
traceId: envelope?.traceId || response.headers.get('x-trace-id') }
|
|
}
|
|
} catch {
|
|
await save({ requests: state.requests.map(row => row.sequence === entry.sequence ? { ...row, status: 'UNKNOWN' } : row) })
|
|
throw new Error(`HTTP结果不确定：${method} ${path}；禁止自动重试`)
|
|
}
|
|
await save({ requests: state.requests.map(row => row.sequence === entry.sequence ? { ...row,
|
|
status: success(result) || result.bytes ? 'DONE' : 'REJECTED', http: result.http,
|
|
code: safe(result.body?.code), requestId: safe(result.requestId), traceId: safe(result.traceId) } : row) })
|
|
return result
|
|
}
|
|
return context
|
|
}
|