539 lines
24 KiB
JavaScript
539 lines
24 KiB
JavaScript
'use strict';
|
||
|
||
/**
|
||
* Gitea Workflow 管理面适配器。
|
||
*
|
||
* 该模块不实现 Webhook、调度器或 Codex turn，只把核心 workflow service
|
||
* 暴露为稳定的查询/控制契约，并在核心服务尚未挂接时提供可恢复的只读状态。
|
||
*/
|
||
|
||
const fs = require('fs');
|
||
const path = require('path');
|
||
const crypto = require('crypto');
|
||
|
||
const TERMINAL_STATES = new Set([
|
||
'succeeded', 'succeeded_with_rest_fallback', 'failed', 'failed_reply',
|
||
'cancelled', 'aborted', 'ignored', 'duplicate', 'rejected',
|
||
]);
|
||
const RUNNING_STATES = new Set(['preparing', 'running', 'aborting']);
|
||
|
||
function nowIso() {
|
||
return new Date().toISOString();
|
||
}
|
||
|
||
function clone(value) {
|
||
return value == null ? value : JSON.parse(JSON.stringify(value));
|
||
}
|
||
|
||
function safeText(value, max = 2000) {
|
||
return String(value == null ? '' : value).slice(0, max);
|
||
}
|
||
|
||
function requestId(value) {
|
||
const text = safeText(value, 120).trim();
|
||
return text || crypto.randomUUID();
|
||
}
|
||
|
||
function stripGlobalConcurrency(settings = {}) {
|
||
const next = { ...settings };
|
||
delete next.maxConcurrency;
|
||
return next;
|
||
}
|
||
|
||
function createEmptyState() {
|
||
return {
|
||
version: 1,
|
||
control: {
|
||
globalPaused: false,
|
||
pauseReason: null,
|
||
updatedBy: 'system',
|
||
updatedAt: nowIso(),
|
||
version: 0,
|
||
},
|
||
repositories: [],
|
||
tasks: [],
|
||
audits: [],
|
||
logs: [],
|
||
controlRequests: {},
|
||
settings: {
|
||
host: '', botLogin: 'ccweb-bot', workspaceRoot: '', defaultBranch: '',
|
||
webhookSecretConfigured: false, botTokenConfigured: false,
|
||
updatedAt: nowIso(), updatedBy: 'system',
|
||
},
|
||
updatedAt: nowIso(),
|
||
};
|
||
}
|
||
|
||
function readState(filePath) {
|
||
if (!filePath) return createEmptyState();
|
||
try {
|
||
if (!fs.existsSync(filePath)) return createEmptyState();
|
||
const stat = fs.statSync(filePath);
|
||
if (stat.size > 8 * 1024 * 1024) throw new Error('workflow 管理状态文件超过 8MB 上限');
|
||
const parsed = JSON.parse(fs.readFileSync(filePath, 'utf8'));
|
||
return {
|
||
...createEmptyState(),
|
||
...parsed,
|
||
control: { ...createEmptyState().control, ...(parsed.control || {}) },
|
||
repositories: Array.isArray(parsed.repositories) ? parsed.repositories : [],
|
||
tasks: Array.isArray(parsed.tasks) ? parsed.tasks : [],
|
||
audits: Array.isArray(parsed.audits) ? parsed.audits : [],
|
||
logs: Array.isArray(parsed.logs) ? parsed.logs : [],
|
||
controlRequests: parsed.controlRequests && typeof parsed.controlRequests === 'object' ? parsed.controlRequests : {},
|
||
settings: stripGlobalConcurrency({ ...createEmptyState().settings, ...(parsed.settings || {}) }),
|
||
};
|
||
} catch {
|
||
return createEmptyState();
|
||
}
|
||
}
|
||
|
||
function writeState(filePath, state) {
|
||
if (!filePath) return;
|
||
fs.mkdirSync(path.dirname(filePath), { recursive: true });
|
||
const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`;
|
||
fs.writeFileSync(tempPath, JSON.stringify(state, null, 2), { mode: 0o600 });
|
||
fs.renameSync(tempPath, filePath);
|
||
}
|
||
|
||
function normalizeRepository(repository = {}) {
|
||
const status = String(repository.status || (repository.enabled === false ? 'disabled' : 'active'));
|
||
return {
|
||
repoKey: safeText(repository.repoKey || repository.key || '', 240),
|
||
instanceId: safeText(repository.instanceId || 'default', 80),
|
||
owner: safeText(repository.owner || '', 120),
|
||
name: safeText(repository.name || repository.repo || '', 160),
|
||
cloneUrl: safeText(repository.cloneUrl || '', 500),
|
||
workspacePath: safeText(repository.workspacePath || repository.path || '', 1000),
|
||
defaultBranch: safeText(repository.defaultBranch || repository.default_branch || '', 160),
|
||
status,
|
||
enabled: status !== 'disabled',
|
||
registeredAt: repository.registeredAt || repository.createdAt || null,
|
||
lastFetchAt: repository.lastFetchAt || null,
|
||
dirty: !!(repository.dirty || repository.workspaceDirty || status === 'blocked'),
|
||
dirtyReason: safeText(repository.dirtyReason || repository.blockedReason || repository.workspaceReason || '', 1000) || null,
|
||
dirtySessionKey: safeText(repository.dirtySessionKey || '', 300) || null,
|
||
activeTaskId: safeText(repository.activeTaskId || '', 120) || null,
|
||
queuedCount: Number(repository.queuedCount || 0),
|
||
version: Number(repository.version || 0),
|
||
};
|
||
}
|
||
|
||
function normalizeTask(task = {}) {
|
||
const state = safeText(task.state || task.status || 'queued', 64) || 'queued';
|
||
return {
|
||
taskId: safeText(task.taskId || task.id || '', 120),
|
||
repoKey: safeText(task.repoKey || '', 240),
|
||
resourceKey: safeText(task.resourceKey || '', 300),
|
||
sessionKey: safeText(task.sessionKey || '', 300),
|
||
commentId: task.commentId ?? null,
|
||
author: safeText(task.author || task.actor || '', 160),
|
||
instruction: safeText(task.instruction || task.prompt || '', 4000),
|
||
state,
|
||
stateLabel: safeText(task.stateLabel || '', 120) || null,
|
||
attempt: Number(task.attempt || 0),
|
||
replyAttempt: Number(task.replyAttempt || 0),
|
||
threadId: safeText(task.threadId || '', 240) || null,
|
||
turnId: safeText(task.turnId || '', 240) || null,
|
||
turnState: safeText(task.turnState || task.turn?.status || '', 80) || null,
|
||
turnStartedAt: task.turnStartedAt || task.turn?.startedAt || null,
|
||
turnUpdatedAt: task.turnUpdatedAt || task.turn?.updatedAt || null,
|
||
logs: Array.isArray(task.logs) ? task.logs.slice(-100).map((entry) => ({
|
||
timestamp: entry.timestamp || entry.ts || null,
|
||
level: safeText(entry.level || 'info', 20),
|
||
message: safeText(entry.message || entry.event || '', 1000),
|
||
})) : [],
|
||
dirtyReason: safeText(task.dirtyReason || task.workspaceReason || (state === 'blocked_workspace' ? task.errorMessage || task.error : ''), 1000) || null,
|
||
errorCode: safeText(task.errorCode || '', 120) || null,
|
||
errorMessage: safeText(task.errorMessage || task.error || '', 1000) || null,
|
||
createdAt: task.createdAt || null,
|
||
updatedAt: task.updatedAt || null,
|
||
nextRetryAt: task.nextRetryAt || null,
|
||
terminal: TERMINAL_STATES.has(state),
|
||
};
|
||
}
|
||
|
||
function normalizeAudit(entry = {}) {
|
||
return {
|
||
eventId: safeText(entry.eventId || entry.id || '', 120),
|
||
taskId: safeText(entry.taskId || '', 120) || null,
|
||
repoKey: safeText(entry.repoKey || '', 240) || null,
|
||
action: safeText(entry.action || entry.event || '', 160),
|
||
actor: safeText(entry.actor || entry.operator || 'system', 160),
|
||
reason: safeText(entry.reason || '', 1000) || null,
|
||
fromState: safeText(entry.fromState || '', 64) || null,
|
||
toState: safeText(entry.toState || '', 64) || null,
|
||
deliveryId: safeText(entry.deliveryId || '', 160) || null,
|
||
turnId: safeText(entry.turnId || '', 240) || null,
|
||
commentId: entry.commentId ?? null,
|
||
errorCode: safeText(entry.errorCode || '', 120) || null,
|
||
timestamp: entry.timestamp || entry.createdAt || entry.ts || null,
|
||
metadata: entry.metadata && typeof entry.metadata === 'object' ? clone(entry.metadata) : {},
|
||
};
|
||
}
|
||
|
||
function normalizeLog(entry = {}) {
|
||
return {
|
||
timestamp: entry.timestamp || entry.ts || null,
|
||
level: safeText(entry.level || 'info', 20),
|
||
taskId: safeText(entry.taskId || '', 120) || null,
|
||
repoKey: safeText(entry.repoKey || '', 240) || null,
|
||
event: safeText(entry.event || entry.action || '', 160),
|
||
message: safeText(entry.message || entry.error || '', 1600),
|
||
};
|
||
}
|
||
|
||
function pickService(service, names, ...args) {
|
||
if (!service) return undefined;
|
||
for (const name of names) {
|
||
if (typeof service[name] !== 'function') continue;
|
||
return service[name](...args);
|
||
}
|
||
return undefined;
|
||
}
|
||
|
||
function collectionItems(value) {
|
||
if (Array.isArray(value)) return value;
|
||
if (Array.isArray(value?.items)) return value.items;
|
||
return [];
|
||
}
|
||
|
||
function createGiteaWorkflowManagement(options = {}) {
|
||
const statePath = options.statePath || null;
|
||
let service = options.workflowService || null;
|
||
const clock = options.now || nowIso;
|
||
let state = readState(statePath);
|
||
const secretsPath = options.secretsPath || null;
|
||
const onConfigUpdate = typeof options.onConfigUpdate === 'function' ? options.onConfigUpdate : null;
|
||
|
||
function readSecrets() {
|
||
if (!secretsPath) return {};
|
||
try { return JSON.parse(fs.readFileSync(secretsPath, 'utf8')); } catch { return {}; }
|
||
}
|
||
function writeSecrets(next) {
|
||
if (!secretsPath) return;
|
||
fs.mkdirSync(path.dirname(secretsPath), { recursive: true });
|
||
const temp = `${secretsPath}.${process.pid}.${Date.now()}.tmp`;
|
||
fs.writeFileSync(temp, JSON.stringify(next, null, 2), { mode: 0o600 });
|
||
fs.renameSync(temp, secretsPath);
|
||
}
|
||
function configuration() {
|
||
const settings = stripGlobalConcurrency({ ...state.settings });
|
||
const secrets = readSecrets();
|
||
settings.webhookSecretConfigured = Boolean(secrets.webhookSecret || settings.webhookSecretConfigured);
|
||
settings.botTokenConfigured = Boolean(secrets.botToken || settings.botTokenConfigured);
|
||
return { settings, secrets: { webhookSecretConfigured: settings.webhookSecretConfigured, botTokenConfigured: settings.botTokenConfigured } };
|
||
}
|
||
function runtimeConfiguration() {
|
||
const publicConfig = configuration();
|
||
const secrets = readSecrets();
|
||
return { settings: stripGlobalConcurrency(publicConfig.settings), secrets: { webhookSecret: secrets.webhookSecret || '', botToken: secrets.botToken || '' } };
|
||
}
|
||
function updateConfiguration(input = {}) {
|
||
const actor = safeText(input.actor || 'admin', 160);
|
||
const reason = safeText(input.reason || '', 1000);
|
||
if (!reason) throw Object.assign(new Error('配置变更必须填写操作原因。'), { statusCode: 400, code: 'reason_required' });
|
||
const current = runtimeConfiguration();
|
||
const requestedWorkspace = safeText(input.workspaceRoot ?? state.settings.workspaceRoot, 1200);
|
||
const requestedHost = safeText(input.host ?? state.settings.host, 500);
|
||
if (requestedHost) {
|
||
let parsedHost;
|
||
try { parsedHost = new URL(requestedHost); } catch {
|
||
throw Object.assign(new Error('Gitea 地址必须是完整的 http/https URL。'), { statusCode: 400, code: 'invalid_gitea_host' });
|
||
}
|
||
if (!['http:', 'https:'].includes(parsedHost.protocol) || parsedHost.username || parsedHost.password) {
|
||
throw Object.assign(new Error('Gitea 地址只允许 http/https，且不能携带凭据。'), { statusCode: 400, code: 'invalid_gitea_host' });
|
||
}
|
||
}
|
||
if (requestedWorkspace && (!path.isAbsolute(requestedWorkspace) || path.resolve(requestedWorkspace) === path.parse(requestedWorkspace).root)) {
|
||
throw Object.assign(new Error('工作区根目录必须是非根的绝对路径。'), { statusCode: 400, code: 'invalid_workspace_root' });
|
||
}
|
||
const nextSettings = {
|
||
...state.settings,
|
||
host: requestedHost,
|
||
botLogin: safeText(input.botLogin ?? state.settings.botLogin, 120) || 'ccweb-bot',
|
||
workspaceRoot: requestedWorkspace,
|
||
defaultBranch: safeText(input.defaultBranch ?? state.settings.defaultBranch, 200),
|
||
updatedAt: clock(), updatedBy: actor,
|
||
};
|
||
const secrets = { ...current.secrets };
|
||
if (Object.prototype.hasOwnProperty.call(input, 'webhookSecret') && String(input.webhookSecret || '').trim()) secrets.webhookSecret = String(input.webhookSecret).trim();
|
||
if (Object.prototype.hasOwnProperty.call(input, 'botToken') && String(input.botToken || '').trim()) secrets.botToken = String(input.botToken).trim();
|
||
nextSettings.webhookSecretConfigured = Boolean(secrets.webhookSecret);
|
||
nextSettings.botTokenConfigured = Boolean(secrets.botToken);
|
||
state.settings = stripGlobalConcurrency(nextSettings);
|
||
writeSecrets(secrets);
|
||
appendAudit({ action: 'admin.configuration_updated', actor, reason, metadata: { changed: Object.keys(input).filter((key) => !['webhookSecret', 'botToken', 'maxConcurrency'].includes(key)) } });
|
||
if (onConfigUpdate) onConfigUpdate({ settings: clone(state.settings), secrets: clone(secrets) });
|
||
persist();
|
||
return { settings: clone(state.settings), secrets: { webhookSecretConfigured: nextSettings.webhookSecretConfigured, botTokenConfigured: nextSettings.botTokenConfigured } };
|
||
}
|
||
|
||
function persist() {
|
||
state.updatedAt = clock();
|
||
writeState(statePath, state);
|
||
}
|
||
|
||
function appendAudit({ action, actor = 'system', reason = '', taskId = null, repoKey = null, fromState = null, toState = null, metadata = {} }) {
|
||
const item = normalizeAudit({
|
||
eventId: crypto.randomUUID(), taskId, repoKey, action, actor, reason,
|
||
fromState, toState, metadata, timestamp: clock(),
|
||
});
|
||
state.audits.push(item);
|
||
state.audits = state.audits.slice(-2000);
|
||
persist();
|
||
return item;
|
||
}
|
||
|
||
function sourceOverview() {
|
||
const value = pickService(service, ['getManagementOverview', 'getOverview', 'overview']);
|
||
return value && typeof value === 'object' ? value : null;
|
||
}
|
||
|
||
function snapshot() {
|
||
const overview = sourceOverview() || {};
|
||
const repositories = pickService(service, ['listRepositories', 'getRepositories']) || overview.repositories || state.repositories;
|
||
const tasks = pickService(service, ['listTasks', 'getTasks'], {}) || overview.tasks || state.tasks;
|
||
const auditSource = pickService(service, ['listAudit', 'listAudits', 'getAudit']);
|
||
const logSource = pickService(service, ['listLogs', 'getLogs']);
|
||
const audits = auditSource
|
||
? [...state.audits, ...collectionItems(auditSource)]
|
||
: (overview.audits || state.audits);
|
||
const logs = logSource
|
||
? [...state.logs, ...collectionItems(logSource)]
|
||
: (overview.logs || state.logs);
|
||
const normalizedTasks = collectionItems(tasks).map(normalizeTask);
|
||
const normalizedRepos = collectionItems(repositories).map(normalizeRepository);
|
||
const tasksByRepo = new Map();
|
||
for (const task of normalizedTasks) {
|
||
if (!task.repoKey) continue;
|
||
const list = tasksByRepo.get(task.repoKey) || [];
|
||
list.push(task);
|
||
tasksByRepo.set(task.repoKey, list);
|
||
}
|
||
for (const repo of normalizedRepos) {
|
||
const repoTasks = tasksByRepo.get(repo.repoKey) || [];
|
||
repo.queuedCount = repoTasks.filter((task) => task.state === 'queued' || task.state === 'retry_wait').length;
|
||
repo.activeTaskId = repo.activeTaskId || repoTasks.find((task) => RUNNING_STATES.has(task.state))?.taskId || null;
|
||
const blockedTask = repoTasks.find((task) => task.state === 'blocked_workspace');
|
||
if (blockedTask) {
|
||
repo.dirty = true;
|
||
repo.dirtyReason = repo.dirtyReason || blockedTask.dirtyReason || blockedTask.errorMessage || '工作区阻塞';
|
||
repo.status = 'blocked';
|
||
}
|
||
}
|
||
const runningCount = normalizedTasks.filter((task) => RUNNING_STATES.has(task.state)).length;
|
||
const queuedCount = normalizedTasks.filter((task) => task.state === 'queued' || task.state === 'retry_wait').length;
|
||
const control = {
|
||
...state.control,
|
||
...(overview.control || {}),
|
||
globalPaused: overview.globalPaused ?? overview.control?.globalPaused ?? state.control.globalPaused,
|
||
};
|
||
const taskLogs = normalizedTasks.flatMap((task) => task.logs.map((entry) => ({ ...entry, taskId: task.taskId, repoKey: task.repoKey })));
|
||
const normalizedLogs = (Array.isArray(logs) ? logs : []).map(normalizeLog);
|
||
const auditLogs = collectionItems(audits).map((entry) => ({
|
||
timestamp: entry.timestamp || entry.createdAt || entry.ts || null,
|
||
level: entry.errorCode ? 'warn' : 'info',
|
||
taskId: entry.taskId || null,
|
||
repoKey: entry.repoKey || null,
|
||
event: entry.action || entry.event || 'workflow',
|
||
message: [entry.action || entry.event || 'workflow', entry.errorCode, entry.toState].filter(Boolean).join(' · '),
|
||
})).map(normalizeLog);
|
||
return {
|
||
ok: true,
|
||
available: !!service,
|
||
generatedAt: clock(),
|
||
control,
|
||
summary: {
|
||
running: Number(overview.summary?.running ?? runningCount),
|
||
queued: Number(overview.summary?.queued ?? queuedCount),
|
||
repositories: normalizedRepos.length,
|
||
blocked: normalizedRepos.filter((repo) => repo.dirty || repo.status === 'blocked').length,
|
||
},
|
||
repositories: normalizedRepos,
|
||
tasks: normalizedTasks,
|
||
audits: Array.from(new Map(collectionItems(audits).map(normalizeAudit)
|
||
.map((entry) => [entry.eventId || `${entry.timestamp}:${entry.action}`, entry])).values()).slice(-200),
|
||
logs: [...normalizedLogs, ...taskLogs.map(normalizeLog), ...auditLogs].slice(-300),
|
||
};
|
||
}
|
||
|
||
function listTasks(query = {}) {
|
||
const all = snapshot().tasks;
|
||
const filtered = all.filter((task) => (!query.repoKey || task.repoKey === query.repoKey)
|
||
&& (!query.resourceKey || task.resourceKey === query.resourceKey)
|
||
&& (!query.state || task.state === query.state));
|
||
const offset = Math.max(0, Number.parseInt(query.offset || 0, 10) || 0);
|
||
const limit = Math.min(200, Math.max(1, Number.parseInt(query.limit || 50, 10) || 50));
|
||
return { items: filtered.slice(offset, offset + limit), total: filtered.length, offset, limit };
|
||
}
|
||
|
||
function resolveTask(taskId) {
|
||
const serviceTask = pickService(service, ['getTask', 'findTask'], taskId);
|
||
if (serviceTask) return normalizeTask(serviceTask);
|
||
return snapshot().tasks.find((task) => task.taskId === taskId) || null;
|
||
}
|
||
|
||
function controlAction(action, input = {}) {
|
||
const actor = safeText(input.actor || 'admin', 160);
|
||
const reason = safeText(input.reason || '', 1000);
|
||
if (!reason) return { ok: false, statusCode: 400, code: 'reason_required', message: '控制操作必须填写原因。' };
|
||
const rid = requestId(input.requestId);
|
||
if (state.controlRequests[rid]) {
|
||
return { ...clone(state.controlRequests[rid]), idempotent: true, overview: snapshot() };
|
||
}
|
||
const previous = snapshot();
|
||
const methodNames = {
|
||
pause: ['pause', 'pauseAll', 'setGlobalPaused'],
|
||
resume: ['resume', 'resumeAll', 'setGlobalPaused'],
|
||
disable: ['disableRepository', 'setRepositoryEnabled', 'setRepositoryStatus'],
|
||
enable: ['enableRepository', 'setRepositoryEnabled', 'setRepositoryStatus'],
|
||
cancel: ['cancelTask', 'cancelQueuedTask'],
|
||
abort: ['abortTask', 'abortTurn', 'cancelRunningTask'],
|
||
}[action] || [];
|
||
const serviceArgs = action === 'pause' || action === 'resume'
|
||
? [{ actor, reason, requestId: rid }]
|
||
: [{ ...input, actor, reason, requestId: rid }];
|
||
let result;
|
||
try {
|
||
result = pickService(service, methodNames, ...serviceArgs);
|
||
} catch (error) {
|
||
return { ok: false, statusCode: 503, code: 'workflow_service_error', message: safeText(error.message || error, 500) };
|
||
}
|
||
if (!service) return { ok: false, statusCode: 503, code: 'workflow_service_unavailable', message: 'Workflow 核心服务尚未挂接，控制操作未执行。' };
|
||
if (result === undefined) return { ok: false, statusCode: 503, code: 'workflow_control_unsupported', message: `核心服务未提供 ${action} 操作。` };
|
||
if (result && result.ok === false) {
|
||
return { ...clone(result), ok: false, statusCode: result.code === 'task_not_found' || result.code === 'repository_not_found' ? 404 : 400 };
|
||
}
|
||
const taskId = input.taskId || null;
|
||
const repoKey = input.repoKey || null;
|
||
appendAudit({
|
||
action: `admin.${action}`, actor, reason, taskId, repoKey,
|
||
fromState: taskId ? resolveTask(taskId)?.state : null,
|
||
toState: taskId ? resolveTask(taskId)?.state : null,
|
||
metadata: { requestId: rid, result: result && typeof result === 'object' ? result : null },
|
||
});
|
||
const response = { ok: true, requestId: rid, action, result: result === true ? {} : clone(result), previous: previous.summary };
|
||
state.controlRequests[rid] = response;
|
||
const requestIds = Object.keys(state.controlRequests);
|
||
while (requestIds.length > 500) delete state.controlRequests[requestIds.shift()];
|
||
persist();
|
||
return { ...clone(response), overview: snapshot() };
|
||
}
|
||
|
||
return Object.freeze({
|
||
overview: snapshot,
|
||
listTasks,
|
||
getTask(taskId) { return resolveTask(taskId); },
|
||
listAudit(query = {}) {
|
||
const items = snapshot().audits.filter((entry) => (!query.taskId || entry.taskId === query.taskId)
|
||
&& (!query.repoKey || entry.repoKey === query.repoKey));
|
||
return { items: items.slice(-Math.min(500, Number(query.limit || 200))), total: items.length };
|
||
},
|
||
control: controlAction,
|
||
configuration,
|
||
runtimeConfiguration,
|
||
updateConfiguration,
|
||
appendAudit,
|
||
setState(next) {
|
||
const cloned = clone(next) || {};
|
||
const nextState = { ...state, ...cloned };
|
||
if (cloned.settings) nextState.settings = stripGlobalConcurrency({ ...state.settings, ...cloned.settings });
|
||
state = nextState;
|
||
persist();
|
||
},
|
||
setWorkflowService(nextService) { service = nextService || null; },
|
||
getState() { return clone(state); },
|
||
});
|
||
}
|
||
|
||
function readRequestBody(req, maxBytes = 256 * 1024) {
|
||
return new Promise((resolve, reject) => {
|
||
let total = 0;
|
||
const chunks = [];
|
||
req.on('data', (chunk) => {
|
||
total += chunk.length;
|
||
if (total > maxBytes) {
|
||
reject(Object.assign(new Error('请求体过大'), { statusCode: 413 }));
|
||
req.destroy();
|
||
return;
|
||
}
|
||
chunks.push(chunk);
|
||
});
|
||
req.on('end', () => {
|
||
try {
|
||
const text = Buffer.concat(chunks).toString('utf8').trim();
|
||
resolve(text ? JSON.parse(text) : {});
|
||
} catch (error) {
|
||
reject(Object.assign(new Error('请求体不是有效 JSON'), { statusCode: 400, cause: error }));
|
||
}
|
||
});
|
||
req.on('error', reject);
|
||
});
|
||
}
|
||
|
||
function sendJson(res, statusCode, payload) {
|
||
if (res.headersSent) return;
|
||
res.writeHead(statusCode, { 'Content-Type': 'application/json; charset=utf-8', 'Cache-Control': 'no-cache' });
|
||
res.end(JSON.stringify(payload));
|
||
}
|
||
|
||
async function handleGiteaWorkflowManagementApi(req, res, url, management, options = {}) {
|
||
const pathname = url.pathname.replace(/\/+$/, '') || '/';
|
||
if (typeof options.authenticate === 'function' && !options.authenticate(req)) {
|
||
return sendJson(res, 401, { ok: false, code: 'unauthorized', message: 'Not authenticated' });
|
||
}
|
||
if (!management) return sendJson(res, 503, { ok: false, code: 'workflow_service_unavailable', message: 'Workflow 管理服务不可用。' });
|
||
try {
|
||
if (req.method === 'GET' && pathname === '/api/gitea-workflow/overview') return sendJson(res, 200, management.overview());
|
||
if (req.method === 'GET' && pathname === '/api/gitea-workflow/config') return sendJson(res, 200, { ok: true, ...management.configuration() });
|
||
if (req.method === 'PUT' && pathname === '/api/gitea-workflow/config') {
|
||
const body = await readRequestBody(req);
|
||
return sendJson(res, 200, { ok: true, ...management.updateConfiguration(body) });
|
||
}
|
||
if (req.method === 'GET' && pathname === '/api/gitea-workflow/tasks') {
|
||
return sendJson(res, 200, { ok: true, ...management.listTasks(Object.fromEntries(url.searchParams.entries())) });
|
||
}
|
||
const taskMatch = pathname.match(/^\/api\/gitea-workflow\/tasks\/([^/]+)$/);
|
||
if (req.method === 'GET' && taskMatch) {
|
||
const task = management.getTask(decodeURIComponent(taskMatch[1]));
|
||
return task ? sendJson(res, 200, { ok: true, task }) : sendJson(res, 404, { ok: false, code: 'task_not_found' });
|
||
}
|
||
if (req.method === 'GET' && pathname === '/api/gitea-workflow/audit') {
|
||
return sendJson(res, 200, { ok: true, ...management.listAudit(Object.fromEntries(url.searchParams.entries())) });
|
||
}
|
||
const controlMatch = pathname.match(/^\/api\/gitea-workflow\/control\/(pause|resume)$/);
|
||
if (req.method === 'POST' && controlMatch) {
|
||
const body = await readRequestBody(req);
|
||
const result = management.control(controlMatch[1], body);
|
||
return sendJson(res, result.statusCode || (result.ok ? 200 : 400), result);
|
||
}
|
||
const repoMatch = pathname.match(/^\/api\/gitea-workflow\/repos\/(.+)\/(disable|enable)$/);
|
||
if (req.method === 'POST' && repoMatch) {
|
||
const body = await readRequestBody(req);
|
||
const result = management.control(repoMatch[2], { ...body, repoKey: decodeURIComponent(repoMatch[1]) });
|
||
return sendJson(res, result.statusCode || (result.ok ? 200 : 400), result);
|
||
}
|
||
const taskControlMatch = pathname.match(/^\/api\/gitea-workflow\/tasks\/([^/]+)\/(cancel|abort)$/);
|
||
if (req.method === 'POST' && taskControlMatch) {
|
||
const body = await readRequestBody(req);
|
||
const result = management.control(taskControlMatch[2], { ...body, taskId: decodeURIComponent(taskControlMatch[1]) });
|
||
return sendJson(res, result.statusCode || (result.ok ? 200 : 400), result);
|
||
}
|
||
return sendJson(res, 404, { ok: false, code: 'not_found' });
|
||
} catch (error) {
|
||
return sendJson(res, error.statusCode || 500, {
|
||
ok: false,
|
||
code: error.code || 'workflow_api_error',
|
||
message: safeText(error.message || error, 500),
|
||
});
|
||
}
|
||
}
|
||
|
||
module.exports = {
|
||
createGiteaWorkflowManagement,
|
||
handleGiteaWorkflowManagementApi,
|
||
normalizeRepository,
|
||
normalizeTask,
|
||
normalizeAudit,
|
||
};
|