Skip to content

Commit 1a9d33b

Browse files
RantsRoamerRantsRoamer
committed
Some updates
1 parent 3eb34f5 commit 1a9d33b

12 files changed

Lines changed: 1102 additions & 42 deletions

‎config.default.json‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,23 @@
1919
"chatHistorySearch": true,
2020
"subagentMissions": true
2121
},
22+
"automation": {
23+
"legacyHeartbeatEnabled": true
24+
},
25+
"observability": {
26+
"retention": {
27+
"runs": 500,
28+
"events": 5000,
29+
"deliveries": 2000,
30+
"alerts": 1000
31+
},
32+
"alerts": {
33+
"enabled": false,
34+
"minSeverity": "error",
35+
"email": { "enabled": false },
36+
"webhook": { "enabled": false, "url": "", "secret": "" }
37+
}
38+
},
2239
"heartbeat": [],
2340
"skills": { "enabledIds": [] },
2441
"searxng": { "url": "", "enabled": false },

‎lib/auth.js‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -169,6 +169,7 @@ function authMiddleware(req, res, next) {
169169
req.path === '/api/app-name' ||
170170
req.path === '/api/channel/chat' ||
171171
req.path.startsWith('/api/webhook/receive/') ||
172+
req.path.startsWith('/api/pipelines/webhook/') ||
172173
req.path.startsWith('/static/') || (req.path === '/' && req.method === 'GET')) {
173174
return next();
174175
}

‎lib/heartbeat.js‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -136,6 +136,8 @@ function persistLastRun(jobId, timestampMs) {
136136

137137
function tick() {
138138
const config = getConfig();
139+
const legacyEnabled = !(config.automation && config.automation.legacyHeartbeatEnabled === false);
140+
if (!legacyEnabled) return;
139141
const jobs = config.heartbeat || [];
140142
const now = Date.now();
141143
for (const job of jobs) {

‎lib/pipelineObservability.js‎

Lines changed: 261 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,261 @@
1+
'use strict';
2+
3+
const fs = require('fs');
4+
const path = require('path');
5+
const crypto = require('crypto');
6+
const { getConfig } = require('./config.js');
7+
8+
const DATA_DIR = path.join(__dirname, '..', 'data');
9+
const RUNS_PATH = path.join(DATA_DIR, 'pipeline-runs.json');
10+
const EVENTS_PATH = path.join(DATA_DIR, 'pipeline-events.json');
11+
const DELIVERIES_PATH = path.join(DATA_DIR, 'webhook-deliveries.json');
12+
const ALERTS_PATH = path.join(DATA_DIR, 'alerts.json');
13+
14+
const DEFAULT_RETENTION = {
15+
runs: 500,
16+
events: 5000,
17+
deliveries: 2000,
18+
alerts: 1000
19+
};
20+
21+
function ensureDataDir() {
22+
if (!fs.existsSync(DATA_DIR)) fs.mkdirSync(DATA_DIR, { recursive: true });
23+
}
24+
25+
function readArray(filePath) {
26+
try {
27+
if (!fs.existsSync(filePath)) return [];
28+
const parsed = JSON.parse(fs.readFileSync(filePath, 'utf8'));
29+
return Array.isArray(parsed) ? parsed : [];
30+
} catch (_) {
31+
return [];
32+
}
33+
}
34+
35+
function writeArray(filePath, records) {
36+
ensureDataDir();
37+
fs.writeFileSync(filePath, JSON.stringify(records, null, 2), 'utf8');
38+
}
39+
40+
function getRetention() {
41+
const cfg = getConfig();
42+
const raw = (cfg.observability && cfg.observability.retention) || {};
43+
return {
44+
runs: Number(raw.runs) > 0 ? Number(raw.runs) : DEFAULT_RETENTION.runs,
45+
events: Number(raw.events) > 0 ? Number(raw.events) : DEFAULT_RETENTION.events,
46+
deliveries: Number(raw.deliveries) > 0 ? Number(raw.deliveries) : DEFAULT_RETENTION.deliveries,
47+
alerts: Number(raw.alerts) > 0 ? Number(raw.alerts) : DEFAULT_RETENTION.alerts
48+
};
49+
}
50+
51+
function trimToLimit(rows, limit) {
52+
if (!Array.isArray(rows)) return [];
53+
if (!Number.isFinite(limit) || limit <= 0) return rows;
54+
if (rows.length <= limit) return rows;
55+
return rows.slice(rows.length - limit);
56+
}
57+
58+
function appendRecord(filePath, record, limit) {
59+
const rows = readArray(filePath);
60+
rows.push(record);
61+
writeArray(filePath, trimToLimit(rows, limit));
62+
}
63+
64+
function newId(prefix) {
65+
return prefix + '_' + crypto.randomBytes(8).toString('hex');
66+
}
67+
68+
function nowIso() {
69+
return new Date().toISOString();
70+
}
71+
72+
function createRun({ pipelineId, pipelineName, triggerType, source, triggerId, contextPreview }) {
73+
const run = {
74+
id: newId('run'),
75+
pipelineId: pipelineId || '',
76+
pipelineName: pipelineName || '',
77+
triggerType: triggerType || 'manual',
78+
triggerId: triggerId || null,
79+
source: source || 'api',
80+
startedAt: nowIso(),
81+
finishedAt: null,
82+
status: 'running',
83+
nodeCount: 0,
84+
error: null,
85+
contextPreview: contextPreview || {}
86+
};
87+
appendRecord(RUNS_PATH, run, getRetention().runs);
88+
return run;
89+
}
90+
91+
function completeRun(runId, updates) {
92+
const rows = readArray(RUNS_PATH);
93+
const idx = rows.findIndex((r) => r.id === runId);
94+
if (idx === -1) return null;
95+
const next = {
96+
...rows[idx],
97+
finishedAt: nowIso(),
98+
...updates
99+
};
100+
rows[idx] = next;
101+
writeArray(RUNS_PATH, trimToLimit(rows, getRetention().runs));
102+
return next;
103+
}
104+
105+
function appendEvent({ runId, pipelineId, nodeId, nodeType, stage, status, message, payload }) {
106+
const evt = {
107+
id: newId('evt'),
108+
ts: nowIso(),
109+
runId: runId || null,
110+
pipelineId: pipelineId || '',
111+
nodeId: nodeId || null,
112+
nodeType: nodeType || null,
113+
stage: stage || 'run',
114+
status: status || 'info',
115+
message: message || '',
116+
payload: payload || null
117+
};
118+
appendRecord(EVENTS_PATH, evt, getRetention().events);
119+
return evt;
120+
}
121+
122+
function appendDelivery({ runId, pipelineId, nodeId, url, statusCode, ok, error, responseText }) {
123+
const rec = {
124+
id: newId('delivery'),
125+
ts: nowIso(),
126+
runId: runId || null,
127+
pipelineId: pipelineId || '',
128+
nodeId: nodeId || null,
129+
url: url || '',
130+
ok: !!ok,
131+
statusCode: statusCode != null ? Number(statusCode) : null,
132+
error: error || null,
133+
responseText: responseText || ''
134+
};
135+
appendRecord(DELIVERIES_PATH, rec, getRetention().deliveries);
136+
return rec;
137+
}
138+
139+
function appendAlert({ runId, pipelineId, severity, type, message, target }) {
140+
const alert = {
141+
id: newId('alert'),
142+
ts: nowIso(),
143+
runId: runId || null,
144+
pipelineId: pipelineId || '',
145+
severity: severity || 'error',
146+
type: type || 'pipeline_failure',
147+
message: message || '',
148+
target: target || null
149+
};
150+
appendRecord(ALERTS_PATH, alert, getRetention().alerts);
151+
return alert;
152+
}
153+
154+
function parseTimeRange(query) {
155+
const sinceMs = query && query.since ? Date.parse(query.since) : null;
156+
const untilMs = query && query.until ? Date.parse(query.until) : null;
157+
return {
158+
sinceMs: Number.isFinite(sinceMs) ? sinceMs : null,
159+
untilMs: Number.isFinite(untilMs) ? untilMs : null
160+
};
161+
}
162+
163+
function filterByTime(rows, key, query) {
164+
const { sinceMs, untilMs } = parseTimeRange(query);
165+
return rows.filter((row) => {
166+
const t = Date.parse(row[key] || row.ts || '');
167+
if (!Number.isFinite(t)) return true;
168+
if (sinceMs != null && t < sinceMs) return false;
169+
if (untilMs != null && t > untilMs) return false;
170+
return true;
171+
});
172+
}
173+
174+
function paginate(rows, query) {
175+
const limit = Math.min(Math.max(Number(query && query.limit) || 100, 1), 1000);
176+
const offset = Math.max(Number(query && query.offset) || 0, 0);
177+
return {
178+
total: rows.length,
179+
limit,
180+
offset,
181+
items: rows.slice(offset, offset + limit)
182+
};
183+
}
184+
185+
function listRuns(query) {
186+
let rows = readArray(RUNS_PATH).slice().reverse();
187+
rows = filterByTime(rows, 'startedAt', query || {});
188+
if (query && query.pipelineId) rows = rows.filter((r) => r.pipelineId === query.pipelineId);
189+
if (query && query.status) rows = rows.filter((r) => r.status === query.status);
190+
if (query && query.triggerType) rows = rows.filter((r) => r.triggerType === query.triggerType);
191+
return paginate(rows, query || {});
192+
}
193+
194+
function getRun(runId) {
195+
return readArray(RUNS_PATH).find((r) => r.id === runId) || null;
196+
}
197+
198+
function listEvents(query) {
199+
let rows = readArray(EVENTS_PATH).slice().reverse();
200+
rows = filterByTime(rows, 'ts', query || {});
201+
if (query && query.pipelineId) rows = rows.filter((r) => r.pipelineId === query.pipelineId);
202+
if (query && query.runId) rows = rows.filter((r) => r.runId === query.runId);
203+
if (query && query.status) rows = rows.filter((r) => r.status === query.status);
204+
return paginate(rows, query || {});
205+
}
206+
207+
function listDeliveries(query) {
208+
let rows = readArray(DELIVERIES_PATH).slice().reverse();
209+
rows = filterByTime(rows, 'ts', query || {});
210+
if (query && query.pipelineId) rows = rows.filter((r) => r.pipelineId === query.pipelineId);
211+
if (query && query.runId) rows = rows.filter((r) => r.runId === query.runId);
212+
if (query && query.ok != null && query.ok !== '') rows = rows.filter((r) => String(r.ok) === String(query.ok));
213+
return paginate(rows, query || {});
214+
}
215+
216+
function listAlerts(query) {
217+
let rows = readArray(ALERTS_PATH).slice().reverse();
218+
rows = filterByTime(rows, 'ts', query || {});
219+
if (query && query.pipelineId) rows = rows.filter((r) => r.pipelineId === query.pipelineId);
220+
if (query && query.runId) rows = rows.filter((r) => r.runId === query.runId);
221+
if (query && query.severity) rows = rows.filter((r) => r.severity === query.severity);
222+
return paginate(rows, query || {});
223+
}
224+
225+
function getSummary() {
226+
const runs = readArray(RUNS_PATH);
227+
const events = readArray(EVENTS_PATH);
228+
const deliveries = readArray(DELIVERIES_PATH);
229+
const alerts = readArray(ALERTS_PATH);
230+
const last24h = Date.now() - (24 * 60 * 60 * 1000);
231+
const last24Runs = runs.filter((r) => Date.parse(r.startedAt || '') >= last24h);
232+
return {
233+
retention: getRetention(),
234+
counts: {
235+
runs: runs.length,
236+
events: events.length,
237+
deliveries: deliveries.length,
238+
alerts: alerts.length
239+
},
240+
health: {
241+
last24Runs: last24Runs.length,
242+
last24FailedRuns: last24Runs.filter((r) => r.status === 'failed').length,
243+
last24SuccessRuns: last24Runs.filter((r) => r.status === 'success').length,
244+
deliveryFailures: deliveries.filter((d) => !d.ok).length
245+
}
246+
};
247+
}
248+
249+
module.exports = {
250+
createRun,
251+
completeRun,
252+
appendEvent,
253+
appendDelivery,
254+
appendAlert,
255+
listRuns,
256+
getRun,
257+
listEvents,
258+
listDeliveries,
259+
listAlerts,
260+
getSummary
261+
};

0 commit comments

Comments
 (0)