315 lines
15 KiB
JavaScript
315 lines
15 KiB
JavaScript
'use strict';
|
||
|
||
/**
|
||
* Gitea Workflow 核心编排服务。
|
||
*
|
||
* 该服务是 Webhook/管理 API 与领域存储、队列之间的适配层。server.js
|
||
* 只需在路由中注入本服务实例，即可逐步接入，不需要复制状态机逻辑。
|
||
*/
|
||
|
||
const crypto = require('crypto');
|
||
const domain = require('./gitea-workflow-domain');
|
||
const { createGiteaWorkflowStore } = require('./gitea-workflow-store');
|
||
const { createGiteaWorkflowQueue } = require('./gitea-workflow-queue');
|
||
const {
|
||
normalizeWebhookEvent,
|
||
verifyWebhookSignature,
|
||
} = require('./gitea-webhook');
|
||
|
||
function safeText(value, max = 1000) {
|
||
return typeof value === 'string' ? value.trim().slice(0, max) : '';
|
||
}
|
||
|
||
function header(headers, names) {
|
||
const source = headers || {};
|
||
for (const wanted of names) {
|
||
for (const [key, value] of Object.entries(source)) {
|
||
if (key.toLowerCase() === wanted.toLowerCase()) return Array.isArray(value) ? value[0] : value;
|
||
}
|
||
}
|
||
return '';
|
||
}
|
||
|
||
function buildGiteaMcpConfig(config, options = {}) {
|
||
const normalized = domain.normalizeConfig(config || {});
|
||
const host = safeText(options.host || normalized.gitea.host, 500);
|
||
const token = String(options.token ?? normalized.gitea.token ?? '');
|
||
const command = safeText(options.command || normalized.gitea.mcpCommand, 200) || 'gitea-mcp';
|
||
const args = Array.isArray(options.args) ? options.args.slice() : ['-t', 'stdio'];
|
||
if (host && !args.includes('-H')) args.push('-H', host);
|
||
return {
|
||
type: 'stdio',
|
||
command,
|
||
args,
|
||
env: {
|
||
...(options.env || {}),
|
||
GITEA_HOST: host,
|
||
GITEA_ACCESS_TOKEN: token,
|
||
GITEA_TOKEN: token,
|
||
},
|
||
};
|
||
}
|
||
|
||
function parseBody(rawBody) {
|
||
try {
|
||
return JSON.parse(Buffer.isBuffer(rawBody) ? rawBody.toString('utf8') : String(rawBody || ''));
|
||
} catch {
|
||
const error = new Error('Webhook 请求体不是有效 JSON');
|
||
error.code = 'invalid_json';
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
class GiteaWorkflowService {
|
||
constructor(options = {}) {
|
||
this.config = domain.normalizeConfig(options.config || options, options.env || process.env);
|
||
this.store = options.store || createGiteaWorkflowStore({
|
||
filePath: options.statePath || options.filePath,
|
||
now: options.now,
|
||
});
|
||
this.clock = typeof options.now === 'function' ? options.now : Date.now;
|
||
this.runner = typeof options.runner === 'function' ? options.runner : null;
|
||
this.queue = options.queue || createGiteaWorkflowQueue({
|
||
store: this.store,
|
||
runner: this.runner || (async (task) => ({ state: domain.TASK_STATES.FAILED, errorCode: 'runner_not_configured', errorMessage: `未配置 Codex runner（${task.taskId}）。` })),
|
||
autoRecover: options.autoRecover !== false,
|
||
now: this.clock,
|
||
});
|
||
this.queue.on('updated', (task) => this.emitTaskAudit(task));
|
||
}
|
||
|
||
emitTaskAudit(task) {
|
||
if (!task || !task.taskId) return null;
|
||
return this.store.appendAudit({
|
||
taskId: task.taskId,
|
||
sessionKey: task.sessionKey,
|
||
repoKey: task.repoKey,
|
||
action: 'task.updated',
|
||
toState: task.state,
|
||
deliveryId: task.deliveryId,
|
||
turnId: task.turnId,
|
||
errorCode: task.errorCode,
|
||
});
|
||
}
|
||
|
||
verify(rawBody, headers = {}) {
|
||
const secret = this.config.gitea.webhookSecret;
|
||
// 内部部署可以不配置 Secret；配置后仍严格执行 HMAC 验签。
|
||
if (!secret) return true;
|
||
const signature = header(headers, ['x-gitea-signature', 'x-hub-signature-256', 'x-signature']);
|
||
return Boolean(secret && verifyWebhookSignature(rawBody, signature, secret));
|
||
}
|
||
|
||
async ingestWebhook({ rawBody, headers = {} } = {}) {
|
||
if (!this.verify(rawBody, headers)) return { statusCode: 401, payload: { ok: false, code: 'invalid_signature' } };
|
||
let payload;
|
||
try { payload = parseBody(rawBody); } catch (error) { return { statusCode: 400, payload: { ok: false, code: error.code } }; }
|
||
const deliveryId = safeText(header(headers, ['x-gitea-delivery', 'x-delivery-id', 'x-github-delivery']) || payload.delivery_id, 180);
|
||
if (!deliveryId) return { statusCode: 400, payload: { ok: false, code: 'missing_delivery_id' } };
|
||
const deliveryKey = `${this.config.instanceId}:${deliveryId}`;
|
||
const claim = this.store.claimDelivery(deliveryKey, { deliveryId, metadata: { event: header(headers, ['x-gitea-event']) } });
|
||
if (claim.duplicate) {
|
||
return { statusCode: 200, payload: { ok: true, duplicate: true, deliveryId, taskId: claim.record.taskId }, record: claim.record };
|
||
}
|
||
const event = normalizeWebhookEvent({
|
||
payload,
|
||
headers,
|
||
instanceId: this.config.instanceId,
|
||
botIdentity: { login: this.config.botLogin, id: this.config.botId },
|
||
});
|
||
if (event.ignored) {
|
||
this.store.updateDelivery(deliveryKey, { status: 'ignored', reason: event.reason });
|
||
this.store.appendAudit({ action: 'webhook.ignored', actor: 'gitea', deliveryId, metadata: { reason: event.reason, eventName: event.eventName } });
|
||
return { statusCode: 200, payload: { ok: true, ignored: true, reason: event.reason, deliveryId }, event };
|
||
}
|
||
const repository = this.store.upsertRepository({
|
||
instanceId: this.config.instanceId,
|
||
owner: event.repository.owner,
|
||
name: event.repository.name,
|
||
cloneUrl: event.repository.cloneUrl,
|
||
defaultBranch: this.config.gitea.defaultBranchOverride || event.repository.defaultBranch,
|
||
});
|
||
if (repository.enabled === false || repository.status === 'disabled') {
|
||
this.store.updateDelivery(deliveryKey, { status: 'rejected', reason: 'repository_disabled' });
|
||
this.store.appendAudit({ action: 'webhook.rejected', actor: 'gitea', deliveryId, repoKey: repository.key, errorCode: 'repository_disabled' });
|
||
return { statusCode: 200, payload: { ok: true, rejected: true, reason: 'repository_disabled', deliveryId } };
|
||
}
|
||
const sessionInput = {
|
||
instanceId: this.config.instanceId,
|
||
owner: event.repository.owner,
|
||
repo: event.repository.name,
|
||
kind: event.resource.kind,
|
||
number: event.resource.number,
|
||
};
|
||
const session = this.store.upsertSession(sessionInput);
|
||
const task = this.store.createTask({
|
||
taskId: `gitea-task-${crypto.createHash('sha256').update(deliveryKey).digest('hex').slice(0, 24)}`,
|
||
deliveryKey, deliveryId, repoKey: repository.key, resourceKey: session.resourceKey,
|
||
sessionKey: session.sessionKey, commentId: event.comment.id || null,
|
||
actor: event.comment.author.login || null, instruction: event.mention.instruction,
|
||
state: domain.TASK_STATES.RECEIVED,
|
||
// 保留经过规范化的仓库/资源上下文，供工作区与 Codex runner 使用；其中不含 Token/Secret。
|
||
metadata: { eventName: event.eventName, resource: event.resource, event },
|
||
});
|
||
this.store.updateDelivery(deliveryKey, { status: 'accepted', taskId: task.taskId });
|
||
this.store.appendAudit({ action: 'webhook.accepted', actor: 'gitea', deliveryId, taskId: task.taskId, sessionKey: task.sessionKey, repoKey: task.repoKey, fromState: domain.TASK_STATES.RECEIVED, toState: domain.TASK_STATES.QUEUED, commentId: task.commentId });
|
||
if (this.runner) this.queue.setRunner(task.taskId, this.runner);
|
||
this.queue.enqueue(task.taskId).catch(() => undefined);
|
||
return { statusCode: 202, payload: { ok: true, accepted: true, deliveryId, taskId: task.taskId, sessionKey: task.sessionKey }, task, event };
|
||
}
|
||
|
||
/**
|
||
* 接收已经由独立 Webhook receiver 验签和规范化的任务，供 server.js
|
||
* 保持最小挂接点。该方法仍在本地 store 中执行 delivery 幂等。
|
||
*/
|
||
enqueueNormalizedTask(input = {}) {
|
||
const event = input.event || input;
|
||
const deliveryId = safeText(input.deliveryId || event.deliveryId, 180);
|
||
const deliveryKey = safeText(input.deliveryKey || `${this.config.instanceId}:${deliveryId}`, 240);
|
||
const claim = this.store.claimDelivery(deliveryKey, { deliveryId, taskId: input.taskId || null, metadata: { source: 'normalized_receiver' } });
|
||
if (claim.duplicate) {
|
||
return { ok: true, duplicate: true, task: claim.record.taskId ? this.store.getTask(claim.record.taskId) : null, delivery: claim.record };
|
||
}
|
||
const repository = this.store.upsertRepository({
|
||
instanceId: this.config.instanceId,
|
||
owner: event.repository?.owner,
|
||
name: event.repository?.name,
|
||
cloneUrl: event.repository?.cloneUrl,
|
||
defaultBranch: this.config.gitea.defaultBranchOverride || event.repository?.defaultBranch,
|
||
});
|
||
if (repository.enabled === false || repository.status === 'disabled') {
|
||
this.store.updateDelivery(deliveryKey, { status: 'rejected', reason: 'repository_disabled' });
|
||
this.store.appendAudit({ action: 'webhook.rejected', actor: 'gitea', deliveryId, repoKey: repository.key, errorCode: 'repository_disabled' });
|
||
return { ok: false, duplicate: false, rejected: true, code: 'repository_disabled', repository };
|
||
}
|
||
const resource = event.resource || {};
|
||
const session = this.store.upsertSession({ instanceId: this.config.instanceId, owner: repository.owner, repo: repository.name, kind: resource.kind, number: resource.number });
|
||
const task = this.store.createTask({
|
||
taskId: input.taskId || `gitea-task-${crypto.createHash('sha256').update(deliveryKey).digest('hex').slice(0, 24)}`,
|
||
deliveryKey, deliveryId, repoKey: repository.key, resourceKey: session.resourceKey, sessionKey: session.sessionKey,
|
||
commentId: event.comment?.id || null, actor: event.comment?.author?.login || null,
|
||
instruction: event.mention?.instruction || event.comment?.body || '', metadata: { eventName: event.eventName, resource, event },
|
||
state: domain.TASK_STATES.RECEIVED,
|
||
});
|
||
this.store.updateDelivery(deliveryKey, { status: 'accepted', taskId: task.taskId });
|
||
this.store.appendAudit({ action: 'webhook.accepted', actor: 'gitea', deliveryId, taskId: task.taskId, repoKey: task.repoKey, sessionKey: task.sessionKey, commentId: task.commentId });
|
||
if (this.runner) this.queue.setRunner(task.taskId, this.runner);
|
||
this.queue.enqueue(task.taskId).catch(() => undefined);
|
||
return { ok: true, duplicate: false, task, repository, session };
|
||
}
|
||
|
||
/** 运行时挂接 Codex/工作区 runner，并立即尝试领取队列。 */
|
||
setRunner(runner) {
|
||
if (typeof runner !== 'function') throw new TypeError('runner 必须是函数');
|
||
this.runner = runner;
|
||
this.queue.defaultRunner = runner;
|
||
for (const task of this.store.listTasks({ state: domain.TASK_STATES.QUEUED })) this.queue.setRunner(task.taskId, runner);
|
||
this.queue.drain();
|
||
return this;
|
||
}
|
||
|
||
/**
|
||
* 在进程内更新管理页保存的连接配置，无需重启服务。
|
||
* Secret 只从管理适配器传入内存配置，不会写入任务、日志或公开响应。
|
||
*/
|
||
updateConfig(input = {}) {
|
||
const current = this.config || {};
|
||
const currentGitea = current.gitea || {};
|
||
this.config = domain.normalizeConfig({
|
||
...current,
|
||
instanceId: input.instanceId ?? current.instanceId,
|
||
botLogin: input.botLogin ?? current.botLogin,
|
||
botId: input.botId ?? current.botId,
|
||
workspaceRoot: input.workspaceRoot ?? current.workspaceRoot,
|
||
defaultBranchOverride: input.defaultBranchOverride ?? currentGitea.defaultBranchOverride,
|
||
gitea: {
|
||
...currentGitea,
|
||
...(input.gitea || {}),
|
||
host: input.host ?? input.gitea?.host ?? currentGitea.host,
|
||
token: input.botToken ?? input.gitea?.token ?? currentGitea.token,
|
||
webhookSecret: input.webhookSecret ?? input.gitea?.webhookSecret ?? currentGitea.webhookSecret,
|
||
defaultBranchOverride: input.defaultBranchOverride ?? input.gitea?.defaultBranchOverride ?? currentGitea.defaultBranchOverride,
|
||
},
|
||
});
|
||
return this.config;
|
||
}
|
||
|
||
getOverview() {
|
||
const tasks = this.store.listTasks();
|
||
const repositories = this.store.listRepositories();
|
||
return {
|
||
ok: true,
|
||
available: true,
|
||
generatedAt: domain.iso(this.clock),
|
||
control: this.store.getControl(),
|
||
summary: {
|
||
running: tasks.filter((task) => domain.RUNNING_STATES.has(task.state)).length,
|
||
queued: tasks.filter((task) => task.state === domain.TASK_STATES.QUEUED || task.state === domain.TASK_STATES.RETRY_WAIT).length,
|
||
repositories: repositories.length,
|
||
blocked: repositories.filter((repo) => repo.dirty || repo.status === 'blocked').length,
|
||
},
|
||
repositories,
|
||
tasks,
|
||
audits: this.store.listAudits().slice(-200),
|
||
queue: this.queue.getStatus(),
|
||
};
|
||
}
|
||
|
||
listRepositories(query = {}) { return this.store.listRepositories(query); }
|
||
listTasks(query = {}) { return this.store.listTasks(query); }
|
||
getTask(taskId) { return this.store.getTask(taskId); }
|
||
listSessions(query = {}) { return this.store.listSessions(query); }
|
||
listTurns(query = {}) { return this.store.listTurns(query); }
|
||
listAudits(query = {}) { return this.store.listAudits(query); }
|
||
listAudit(query = {}) { return this.listAudits(query); }
|
||
|
||
pause(input = {}) { const result = this.queue.pause(input); this.store.appendAudit({ action: 'admin.pause', actor: input.actor || 'admin', metadata: { reason: input.reason || null } }); return result; }
|
||
resume(input = {}) { const result = this.queue.resume(input); this.store.appendAudit({ action: 'admin.resume', actor: input.actor || 'admin', metadata: { reason: input.reason || null } }); return result; }
|
||
|
||
setRepositoryEnabled(repoKey, enabled, input = {}) {
|
||
const repository = this.store.getRepository(repoKey);
|
||
if (!repository) return { ok: false, code: 'repository_not_found' };
|
||
const updated = this.store.upsertRepository({ ...repository, enabled: Boolean(enabled), status: enabled ? 'active' : 'disabled' });
|
||
this.store.appendAudit({ action: enabled ? 'admin.repository_enabled' : 'admin.repository_disabled', actor: input.actor || 'admin', repoKey, metadata: { reason: input.reason || null } });
|
||
if (enabled) {
|
||
for (const task of this.store.listTasks({ repoKey, state: domain.TASK_STATES.BLOCKED_WORKSPACE })) {
|
||
try { this.queue.enqueue(task.taskId); } catch {}
|
||
}
|
||
}
|
||
return { ok: true, repository: updated };
|
||
}
|
||
disableRepository(input = {}) { return this.setRepositoryEnabled(input.repoKey, false, input); }
|
||
enableRepository(input = {}) { return this.setRepositoryEnabled(input.repoKey, true, input); }
|
||
cancelTask(input = {}) { return this.queue.cancelQueued(input.taskId || input.id, input); }
|
||
abortTask(input = {}) { return this.queue.abortRunning(input.taskId || input.id, input); }
|
||
abortTurn(input = {}) { return this.abortTask(input); }
|
||
buildThreadConfig(options = {}) {
|
||
const session = options.session || {};
|
||
const settings = {
|
||
mode: 'yolo',
|
||
...(this.config.codex.model ? { model: this.config.codex.model } : {}),
|
||
...(this.config.codex.reasoningEffort ? { reasoning_effort: this.config.codex.reasoningEffort } : {}),
|
||
...(this.config.codex.developerInstructions ? { developer_instructions: this.config.codex.developerInstructions } : {}),
|
||
};
|
||
return {
|
||
cwd: options.cwd || null,
|
||
mcp_servers: {
|
||
gitea: buildGiteaMcpConfig(this.config, options),
|
||
},
|
||
collaborationMode: { settings },
|
||
sessionKey: session.sessionKey || null,
|
||
};
|
||
}
|
||
|
||
publicConfig() { return domain.publicConfig(this.config); }
|
||
}
|
||
|
||
function createGiteaWorkflowService(options) { return new GiteaWorkflowService(options); }
|
||
|
||
module.exports = {
|
||
GiteaWorkflowService,
|
||
buildGiteaMcpConfig,
|
||
createGiteaWorkflowService,
|
||
parseBody,
|
||
};
|