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 }