301 lines
11 KiB
JavaScript
301 lines
11 KiB
JavaScript
'use strict';
|
||
|
||
/**
|
||
* Gitea Workflow 持久化仓库。
|
||
*
|
||
* 使用单个 JSON 文件保存 MVP 领域状态，所有写入都通过同目录临时文件
|
||
* + rename 完成，进程崩溃时不会留下半截 JSON。业务层可以替换为数据库，
|
||
* 但应保持本文件暴露的幂等键和查询契约。
|
||
*/
|
||
|
||
const crypto = require('crypto');
|
||
const fs = require('fs');
|
||
const path = require('path');
|
||
|
||
const domain = require('./gitea-workflow-domain');
|
||
|
||
const MAX_AUDITS = 10_000;
|
||
const MAX_DELIVERIES = 20_000;
|
||
|
||
function clone(value) {
|
||
return domain.clone(value);
|
||
}
|
||
|
||
function atomicWriteJson(filePath, value) {
|
||
const target = path.resolve(filePath);
|
||
fs.mkdirSync(path.dirname(target), { recursive: true });
|
||
const temp = `${target}.${process.pid}.${crypto.randomBytes(6).toString('hex')}.tmp`;
|
||
fs.writeFileSync(temp, `${JSON.stringify(value, null, 2)}\n`, { mode: 0o600 });
|
||
try {
|
||
fs.renameSync(temp, target);
|
||
} finally {
|
||
try { if (fs.existsSync(temp)) fs.unlinkSync(temp); } catch {}
|
||
}
|
||
}
|
||
|
||
function emptyState() {
|
||
return {
|
||
schemaVersion: 1,
|
||
updatedAt: domain.iso(),
|
||
control: { paused: false, reason: null, actor: 'system', version: 0, updatedAt: domain.iso() },
|
||
repositories: {},
|
||
sessions: {},
|
||
tasks: {},
|
||
turns: {},
|
||
deliveries: {},
|
||
audits: [],
|
||
};
|
||
}
|
||
|
||
function sanitizeMetadata(value, depth = 0) {
|
||
if (depth > 4) return '[truncated]';
|
||
if (Array.isArray(value)) return value.slice(0, 100).map((item) => sanitizeMetadata(item, depth + 1));
|
||
if (!value || typeof value !== 'object') return typeof value === 'string' ? value.slice(0, 2000) : value;
|
||
const output = {};
|
||
for (const [key, item] of Object.entries(value)) {
|
||
if (/token|secret|password|authorization|api[-_]?key/i.test(key)) {
|
||
output[key] = '[redacted]';
|
||
} else {
|
||
output[key] = sanitizeMetadata(item, depth + 1);
|
||
}
|
||
}
|
||
return output;
|
||
}
|
||
|
||
class GiteaWorkflowStore {
|
||
constructor(options = {}) {
|
||
this.filePath = options.filePath ? path.resolve(options.filePath) : null;
|
||
this.clock = typeof options.now === 'function' ? options.now : Date.now;
|
||
this.state = emptyState();
|
||
this.load();
|
||
}
|
||
|
||
load() {
|
||
if (!this.filePath || !fs.existsSync(this.filePath)) return this.state;
|
||
try {
|
||
const parsed = JSON.parse(fs.readFileSync(this.filePath, 'utf8'));
|
||
if (parsed && typeof parsed === 'object' && !Array.isArray(parsed)) {
|
||
const fresh = emptyState();
|
||
this.state = {
|
||
...fresh,
|
||
...parsed,
|
||
control: { ...fresh.control, ...(parsed.control || {}) },
|
||
repositories: parsed.repositories && typeof parsed.repositories === 'object' ? parsed.repositories : {},
|
||
sessions: parsed.sessions && typeof parsed.sessions === 'object' ? parsed.sessions : {},
|
||
tasks: parsed.tasks && typeof parsed.tasks === 'object' ? parsed.tasks : {},
|
||
turns: parsed.turns && typeof parsed.turns === 'object' ? parsed.turns : {},
|
||
deliveries: parsed.deliveries && typeof parsed.deliveries === 'object' ? parsed.deliveries : {},
|
||
audits: Array.isArray(parsed.audits) ? parsed.audits : [],
|
||
};
|
||
}
|
||
} catch {
|
||
// 启动时损坏的状态文件不能让 Webhook 进程直接退出；保留空状态并在下次写入时修复。
|
||
this.state = emptyState();
|
||
}
|
||
return this.state;
|
||
}
|
||
|
||
persist() {
|
||
this.state.updatedAt = domain.iso(this.clock);
|
||
if (this.filePath) atomicWriteJson(this.filePath, this.state);
|
||
return this.snapshot();
|
||
}
|
||
|
||
snapshot() { return clone(this.state); }
|
||
|
||
getControl() { return clone(this.state.control); }
|
||
|
||
setPaused(paused, input = {}) {
|
||
const next = Boolean(paused);
|
||
const previous = this.state.control;
|
||
this.state.control = {
|
||
paused: next,
|
||
reason: typeof input.reason === 'string' ? input.reason.slice(0, 1000) : null,
|
||
actor: typeof input.actor === 'string' ? input.actor.slice(0, 160) : 'system',
|
||
version: Number(previous.version || 0) + (previous.paused === next ? 0 : 1),
|
||
updatedAt: domain.iso(this.clock),
|
||
};
|
||
this.persist();
|
||
return this.getControl();
|
||
}
|
||
|
||
upsertRepository(input) {
|
||
const key = input.key || domain.repoKeyFor(input);
|
||
const previous = this.state.repositories[key];
|
||
const record = domain.createRepositoryRecord({ ...previous, ...input, key }, { now: this.clock });
|
||
this.state.repositories[key] = record;
|
||
this.persist();
|
||
return clone(record);
|
||
}
|
||
|
||
getRepository(key) { return clone(this.state.repositories[String(key)] || null); }
|
||
listRepositories(query = {}) {
|
||
return Object.values(this.state.repositories).filter((item) => (
|
||
(!query.instanceId || item.instanceId === query.instanceId)
|
||
&& (!query.status || item.status === query.status)
|
||
&& (query.enabled === undefined ? true : item.enabled === Boolean(query.enabled))
|
||
)).map(clone);
|
||
}
|
||
|
||
upsertSession(input) {
|
||
const key = input.sessionKey || domain.sessionKeyFor(input);
|
||
const previous = this.state.sessions[key];
|
||
const record = domain.createSessionRecord({ ...previous, ...input, sessionKey: key }, { now: this.clock });
|
||
this.state.sessions[key] = record;
|
||
this.persist();
|
||
return clone(record);
|
||
}
|
||
|
||
getSession(key) { return clone(this.state.sessions[String(key)] || null); }
|
||
listSessions(query = {}) {
|
||
return Object.values(this.state.sessions).filter((item) => (
|
||
(!query.repoKey || item.resourceKey?.startsWith(String(query.repoKey)))
|
||
&& (!query.status || item.status === query.status)
|
||
)).map(clone);
|
||
}
|
||
|
||
createTask(input) {
|
||
const record = domain.createTaskRecord(input, { now: this.clock });
|
||
if (this.state.tasks[record.taskId]) return clone(this.state.tasks[record.taskId]);
|
||
this.state.tasks[record.taskId] = record;
|
||
this.persist();
|
||
return clone(record);
|
||
}
|
||
|
||
upsertTask(input) {
|
||
const id = String(input.taskId || input.id || '');
|
||
if (!id) throw new TypeError('任务缺少 taskId');
|
||
const previous = this.state.tasks[id];
|
||
const record = domain.createTaskRecord({ ...previous, ...input, taskId: id }, { now: this.clock });
|
||
this.state.tasks[id] = record;
|
||
this.persist();
|
||
return clone(record);
|
||
}
|
||
|
||
transitionTask(taskId, nextState, options = {}) {
|
||
const current = this.state.tasks[String(taskId)];
|
||
if (!current) return null;
|
||
if (options.expectedVersion !== undefined
|
||
&& Number(current.stateVersion || 0) !== Number(options.expectedVersion)) {
|
||
const error = new Error('任务版本已变化，请刷新后重试。');
|
||
error.code = 'version_conflict';
|
||
error.expectedVersion = options.expectedVersion;
|
||
error.actualVersion = current.stateVersion || 0;
|
||
throw error;
|
||
}
|
||
const next = domain.transitionTask(current, nextState, { ...options, now: options.now || this.clock });
|
||
this.state.tasks[String(taskId)] = next;
|
||
this.persist();
|
||
return clone(next);
|
||
}
|
||
|
||
getTask(taskId) { return clone(this.state.tasks[String(taskId)] || null); }
|
||
listTasks(query = {}) {
|
||
return Object.values(this.state.tasks).filter((item) => (
|
||
(!query.repoKey || item.repoKey === query.repoKey)
|
||
&& (!query.sessionKey || item.sessionKey === query.sessionKey)
|
||
&& (!query.resourceKey || item.resourceKey === query.resourceKey)
|
||
&& (!query.state && !query.status || item.state === (query.state || query.status))
|
||
)).sort((a, b) => String(a.createdAt).localeCompare(String(b.createdAt))).map(clone);
|
||
}
|
||
|
||
upsertTurn(input) {
|
||
const id = String(input.turnId || '');
|
||
if (!id) throw new TypeError('turn 缺少 turnId');
|
||
const record = { ...(this.state.turns[id] || {}), ...clone(input), turnId: id, updatedAt: domain.iso(this.clock) };
|
||
this.state.turns[id] = record;
|
||
this.persist();
|
||
return clone(record);
|
||
}
|
||
|
||
getTurn(turnId) { return clone(this.state.turns[String(turnId)] || null); }
|
||
listTurns(query = {}) {
|
||
return Object.values(this.state.turns).filter((item) => (!query.taskId || item.taskId === query.taskId)
|
||
&& (!query.sessionKey || item.sessionKey === query.sessionKey)).map(clone);
|
||
}
|
||
|
||
getDelivery(deliveryKey) { return clone(this.state.deliveries[String(deliveryKey)] || null); }
|
||
|
||
/** 原子语义由单进程事件循环保证；重复 delivery 永远返回原记录。 */
|
||
claimDelivery(deliveryKey, value = {}) {
|
||
const key = String(deliveryKey || '');
|
||
if (!key) throw new TypeError('deliveryKey 不能为空');
|
||
const existing = this.state.deliveries[key];
|
||
if (existing) return { duplicate: true, record: clone(existing) };
|
||
const record = {
|
||
deliveryKey: key,
|
||
deliveryId: value.deliveryId || key.split(':').slice(1).join(':'),
|
||
status: value.status || 'processing',
|
||
taskId: value.taskId || null,
|
||
createdAt: value.createdAt || domain.iso(this.clock),
|
||
updatedAt: value.updatedAt || domain.iso(this.clock),
|
||
metadata: sanitizeMetadata(value.metadata || {}),
|
||
};
|
||
this.state.deliveries[key] = record;
|
||
const keys = Object.keys(this.state.deliveries);
|
||
while (keys.length > MAX_DELIVERIES) delete this.state.deliveries[keys.shift()];
|
||
this.persist();
|
||
return { duplicate: false, record: clone(record) };
|
||
}
|
||
|
||
updateDelivery(deliveryKey, patch = {}) {
|
||
const key = String(deliveryKey);
|
||
if (!this.state.deliveries[key]) return null;
|
||
this.state.deliveries[key] = { ...this.state.deliveries[key], ...sanitizeMetadata(patch), updatedAt: domain.iso(this.clock) };
|
||
this.persist();
|
||
return clone(this.state.deliveries[key]);
|
||
}
|
||
|
||
appendAudit(input = {}) {
|
||
const item = {
|
||
eventId: input.eventId || crypto.randomUUID(),
|
||
taskId: input.taskId || null,
|
||
sessionKey: input.sessionKey || null,
|
||
repoKey: input.repoKey || null,
|
||
actor: input.actor || 'system',
|
||
action: input.action || 'unknown',
|
||
fromState: input.fromState || null,
|
||
toState: input.toState || null,
|
||
deliveryId: input.deliveryId || null,
|
||
turnId: input.turnId || null,
|
||
commentId: input.commentId ?? null,
|
||
errorCode: input.errorCode || null,
|
||
timestamp: input.timestamp || domain.iso(this.clock),
|
||
metadata: sanitizeMetadata(input.metadata || {}),
|
||
};
|
||
this.state.audits.push(item);
|
||
if (this.state.audits.length > MAX_AUDITS) this.state.audits.splice(0, this.state.audits.length - MAX_AUDITS);
|
||
this.persist();
|
||
return clone(item);
|
||
}
|
||
|
||
listAudits(query = {}) {
|
||
return this.state.audits.filter((item) => (!query.taskId || item.taskId === query.taskId)
|
||
&& (!query.sessionKey || item.sessionKey === query.sessionKey)
|
||
&& (!query.repoKey || item.repoKey === query.repoKey)
|
||
&& (!query.action || item.action === query.action)
|
||
&& (!query.from || String(item.timestamp) >= String(query.from))
|
||
&& (!query.to || String(item.timestamp) <= String(query.to))).map(clone);
|
||
}
|
||
|
||
recover(options = {}) {
|
||
const before = Object.values(this.state.tasks);
|
||
const after = domain.recoverTasks(before, { ...options, now: options.now || this.clock });
|
||
for (const task of after) this.state.tasks[task.taskId] = task;
|
||
this.persist();
|
||
return after.map(clone);
|
||
}
|
||
}
|
||
|
||
function createGiteaWorkflowStore(options) {
|
||
return new GiteaWorkflowStore(options);
|
||
}
|
||
|
||
module.exports = {
|
||
GiteaWorkflowStore,
|
||
atomicWriteJson,
|
||
createGiteaWorkflowStore,
|
||
emptyState,
|
||
sanitizeMetadata,
|
||
};
|