feat: initial commit for HookPilot
This commit is contained in:
吴璨
@@ -0,0 +1,28 @@
|
||||
{
|
||||
"name": "@hookpilot/server",
|
||||
"version": "0.1.0",
|
||||
"private": true,
|
||||
"type": "module",
|
||||
"main": "dist/index.js",
|
||||
"engines": {
|
||||
"node": ">=22.13"
|
||||
},
|
||||
"scripts": {
|
||||
"dev": "tsx watch src/index.ts",
|
||||
"build": "tsc -p tsconfig.json",
|
||||
"start": "node dist/index.js",
|
||||
"typecheck": "tsc -p tsconfig.json --noEmit",
|
||||
"reset-password": "node dist/cli.js reset-password"
|
||||
},
|
||||
"dependencies": {
|
||||
"@fastify/cookie": "^11.1.2",
|
||||
"@fastify/formbody": "^9.0.0",
|
||||
"@fastify/static": "^10.1.5",
|
||||
"fastify": "^5.12.5"
|
||||
},
|
||||
"devDependencies": {
|
||||
"@types/node": "^22.20.4",
|
||||
"tsx": "^4.23.15",
|
||||
"typescript": "~5.9.3"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,106 @@
|
||||
import fs from 'node:fs';
|
||||
import path from 'node:path';
|
||||
import cookie from '@fastify/cookie';
|
||||
import fastifyStatic from '@fastify/static';
|
||||
import Fastify, { type FastifyError, type FastifyInstance } from 'fastify';
|
||||
import type { AppConfig } from './config.js';
|
||||
import type { Db } from './db.js';
|
||||
import type { Scheduler } from './executor/scheduler.js';
|
||||
import type { ExecutionStore } from './executor/store.js';
|
||||
import type { HookRepo } from './hooks/repo.js';
|
||||
import { authRoutes, registerSessionGuard } from './routes/auth.js';
|
||||
import { executionRoutes } from './routes/executions.js';
|
||||
import { hookRoutes } from './routes/hooks.js';
|
||||
import { systemRoutes } from './routes/system.js';
|
||||
import { webhookRoutes } from './routes/webhook.js';
|
||||
import type { SettingsStore } from './settings.js';
|
||||
import { HttpError } from './util/http-error.js';
|
||||
|
||||
declare module 'fastify' {
|
||||
interface FastifyRequest {
|
||||
sessionId: string;
|
||||
}
|
||||
interface FastifyContextConfig {
|
||||
/** 无需登录即可访问的管理 API */
|
||||
public?: boolean;
|
||||
}
|
||||
}
|
||||
|
||||
export interface AppContext {
|
||||
config: AppConfig;
|
||||
db: Db;
|
||||
settings: SettingsStore;
|
||||
hooks: HookRepo;
|
||||
store: ExecutionStore;
|
||||
scheduler: Scheduler;
|
||||
}
|
||||
|
||||
/** 日志中去掉 query 里的 token，避免凭证写入日志 */
|
||||
function redactUrl(url: string): string {
|
||||
return url.replace(/([?&]token=)[^&]*/gi, '$1***');
|
||||
}
|
||||
|
||||
export async function buildApp(ctx: AppContext): Promise<FastifyInstance> {
|
||||
const app = Fastify({
|
||||
trustProxy: ctx.config.trustProxy,
|
||||
bodyLimit: ctx.config.bodyLimit,
|
||||
logger: {
|
||||
level: ctx.config.logLevel,
|
||||
serializers: {
|
||||
req: (req) => ({ method: req.method, url: redactUrl(req.url), remoteAddress: req.ip }),
|
||||
},
|
||||
},
|
||||
});
|
||||
|
||||
app.decorateRequest('sessionId', '');
|
||||
await app.register(cookie);
|
||||
|
||||
await app.register(
|
||||
async (api) => {
|
||||
registerSessionGuard(api, ctx);
|
||||
api.setErrorHandler((err: FastifyError, req, reply) => {
|
||||
if (err instanceof HttpError) {
|
||||
return reply.code(err.statusCode).send({ error: err.message, details: err.details });
|
||||
}
|
||||
if (err.validation) return reply.code(400).send({ error: `请求参数无效：${err.message}` });
|
||||
if (err.statusCode && err.statusCode < 500) return reply.code(err.statusCode).send({ error: err.message });
|
||||
req.log.error({ err }, '管理 API 处理失败');
|
||||
return reply.code(500).send({ error: '服务器内部错误' });
|
||||
});
|
||||
await authRoutes(api, ctx);
|
||||
await hookRoutes(api, ctx);
|
||||
await executionRoutes(api, ctx);
|
||||
await systemRoutes(api, ctx);
|
||||
},
|
||||
{ prefix: '/api' },
|
||||
);
|
||||
|
||||
await app.register(async (scope) => webhookRoutes(scope, ctx));
|
||||
|
||||
const indexFile = path.join(ctx.config.webDir, 'index.html');
|
||||
const hasWeb = fs.existsSync(indexFile);
|
||||
if (hasWeb) {
|
||||
await app.register(fastifyStatic, {
|
||||
root: ctx.config.webDir,
|
||||
wildcard: false,
|
||||
setHeaders: (res, filePath) => {
|
||||
const immutable = filePath.includes(`${path.sep}assets${path.sep}`);
|
||||
res.header('Cache-Control', immutable ? 'public, max-age=31536000, immutable' : 'no-cache');
|
||||
},
|
||||
});
|
||||
} else {
|
||||
app.log.warn({ webDir: ctx.config.webDir }, '未找到前端构建产物，仅提供 API（开发时请使用 Vite 开发服务器）');
|
||||
}
|
||||
|
||||
// 前端路由回退到 index.html；API 与 Webhook 路径保持 JSON 404
|
||||
app.setNotFoundHandler((req, reply) => {
|
||||
const url = req.url.split('?')[0]!;
|
||||
const isBackend = url.startsWith('/api/') || url === '/api' || url.startsWith('/hooks/');
|
||||
if (hasWeb && req.method === 'GET' && !isBackend) {
|
||||
return reply.header('Cache-Control', 'no-cache').sendFile('index.html');
|
||||
}
|
||||
return reply.code(404).send({ error: '资源不存在' });
|
||||
});
|
||||
|
||||
return app;
|
||||
}
|
||||
@@ -0,0 +1,70 @@
|
||||
import type { Db } from '../db.js';
|
||||
import { randomToken, sha256 } from '../util/crypto.js';
|
||||
import { hashPassword } from './password.js';
|
||||
|
||||
export const SESSION_COOKIE = 'hp_session';
|
||||
export const MIN_PASSWORD_LENGTH = 8;
|
||||
|
||||
export interface AdminRow {
|
||||
id: number;
|
||||
username: string;
|
||||
password_hash: string;
|
||||
updated_at: number;
|
||||
}
|
||||
|
||||
export function getAdmin(db: Db): AdminRow | undefined {
|
||||
return db.get<AdminRow>('SELECT * FROM admin WHERE id = 1');
|
||||
}
|
||||
|
||||
export async function setAdminCredentials(db: Db, username: string, password: string): Promise<void> {
|
||||
const hash = await hashPassword(password);
|
||||
db.run(
|
||||
`INSERT INTO admin (id, username, password_hash, updated_at) VALUES (1, ?, ?, ?)
|
||||
ON CONFLICT (id) DO UPDATE SET username = excluded.username, password_hash = excluded.password_hash, updated_at = excluded.updated_at`,
|
||||
username,
|
||||
hash,
|
||||
Date.now(),
|
||||
);
|
||||
}
|
||||
|
||||
/**
|
||||
* 首次启动时创建管理员。未通过环境变量指定密码时生成随机密码并输出到日志。
|
||||
* 返回生成的随机密码（若有）。
|
||||
*/
|
||||
export async function ensureAdmin(db: Db, username: string, password: string | undefined): Promise<string | null> {
|
||||
if (getAdmin(db)) return null;
|
||||
const actual = password ?? randomToken(12);
|
||||
await setAdminCredentials(db, username, actual);
|
||||
return password ? null : actual;
|
||||
}
|
||||
|
||||
export function createSession(db: Db, ttlMs: number, ip: string, userAgent: string): string {
|
||||
const token = randomToken(32);
|
||||
const now = Date.now();
|
||||
db.run(
|
||||
'INSERT INTO sessions (id, created_at, expires_at, ip, user_agent) VALUES (?, ?, ?, ?, ?)',
|
||||
sha256(token),
|
||||
now,
|
||||
now + ttlMs,
|
||||
ip,
|
||||
userAgent.slice(0, 300),
|
||||
);
|
||||
return token;
|
||||
}
|
||||
|
||||
export function findSession(db: Db, token: string): { id: string } | undefined {
|
||||
return db.get<{ id: string }>('SELECT id FROM sessions WHERE id = ? AND expires_at > ?', sha256(token), Date.now());
|
||||
}
|
||||
|
||||
export function deleteSession(db: Db, id: string): void {
|
||||
db.run('DELETE FROM sessions WHERE id = ?', id);
|
||||
}
|
||||
|
||||
export function deleteOtherSessions(db: Db, keepId: string | null): void {
|
||||
if (keepId) db.run('DELETE FROM sessions WHERE id <> ?', keepId);
|
||||
else db.run('DELETE FROM sessions');
|
||||
}
|
||||
|
||||
export function purgeExpiredSessions(db: Db): void {
|
||||
db.run('DELETE FROM sessions WHERE expires_at <= ?', Date.now());
|
||||
}
|
||||
@@ -0,0 +1,28 @@
|
||||
import { randomBytes, scrypt, timingSafeEqual, type ScryptOptions } from 'node:crypto';
|
||||
|
||||
const PARAMS = { N: 16384, r: 8, p: 1 };
|
||||
const KEY_LENGTH = 64;
|
||||
|
||||
function scryptAsync(password: string, salt: Buffer, keyLength: number, options: ScryptOptions): Promise<Buffer> {
|
||||
return new Promise((resolve, reject) => {
|
||||
scrypt(password, salt, keyLength, options, (err, key) => (err ? reject(err) : resolve(key)));
|
||||
});
|
||||
}
|
||||
|
||||
export async function hashPassword(password: string): Promise<string> {
|
||||
const salt = randomBytes(16);
|
||||
const key = await scryptAsync(password, salt, KEY_LENGTH, PARAMS);
|
||||
return ['scrypt', PARAMS.N, PARAMS.r, PARAMS.p, salt.toString('base64'), key.toString('base64')].join('$');
|
||||
}
|
||||
|
||||
export async function verifyPassword(password: string, stored: string): Promise<boolean> {
|
||||
const [scheme, n, r, p, saltB64, keyB64] = stored.split('$');
|
||||
if (scheme !== 'scrypt' || !saltB64 || !keyB64) return false;
|
||||
const expected = Buffer.from(keyB64, 'base64');
|
||||
const actual = await scryptAsync(password, Buffer.from(saltB64, 'base64'), expected.length, {
|
||||
N: Number(n),
|
||||
r: Number(r),
|
||||
p: Number(p),
|
||||
});
|
||||
return actual.length === expected.length && timingSafeEqual(actual, expected);
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
const WINDOW_MS = 15 * 60 * 1000;
|
||||
const MAX_FAILURES = 10;
|
||||
|
||||
interface Entry {
|
||||
failures: number;
|
||||
firstAt: number;
|
||||
lockedUntil: number;
|
||||
}
|
||||
|
||||
/** 登录失败限流：同一来源 15 分钟内失败 10 次后锁定 15 分钟 */
|
||||
export class LoginLimiter {
|
||||
private readonly entries = new Map<string, Entry>();
|
||||
|
||||
/** 返回剩余锁定秒数，未锁定返回 0 */
|
||||
lockedFor(ip: string): number {
|
||||
const entry = this.entries.get(ip);
|
||||
if (!entry) return 0;
|
||||
const now = Date.now();
|
||||
if (entry.lockedUntil > now) return Math.ceil((entry.lockedUntil - now) / 1000);
|
||||
if (now - entry.firstAt > WINDOW_MS) this.entries.delete(ip);
|
||||
return 0;
|
||||
}
|
||||
|
||||
fail(ip: string): void {
|
||||
const now = Date.now();
|
||||
let entry = this.entries.get(ip);
|
||||
if (!entry || now - entry.firstAt > WINDOW_MS) {
|
||||
entry = { failures: 0, firstAt: now, lockedUntil: 0 };
|
||||
this.entries.set(ip, entry);
|
||||
}
|
||||
entry.failures++;
|
||||
if (entry.failures >= MAX_FAILURES) entry.lockedUntil = now + WINDOW_MS;
|
||||
if (this.entries.size > 10_000) this.prune(now);
|
||||
}
|
||||
|
||||
succeed(ip: string): void {
|
||||
this.entries.delete(ip);
|
||||
}
|
||||
|
||||
private prune(now: number): void {
|
||||
for (const [ip, entry] of this.entries) {
|
||||
if (entry.lockedUntil <= now && now - entry.firstAt > WINDOW_MS) this.entries.delete(ip);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,42 @@
|
||||
import './warnings.js';
|
||||
import fs from 'node:fs';
|
||||
import { deleteOtherSessions, getAdmin, MIN_PASSWORD_LENGTH, setAdminCredentials } from './auth/admin.js';
|
||||
import { config, dataPath } from './config.js';
|
||||
import { Db } from './db.js';
|
||||
import { randomToken } from './util/crypto.js';
|
||||
|
||||
const USAGE = `用法：
|
||||
node dist/cli.js reset-password [用户名]
|
||||
|
||||
重置管理员密码并使所有登录会话失效。
|
||||
新密码取自环境变量 HOOKPILOT_NEW_PASSWORD，未设置时随机生成并输出。`;
|
||||
|
||||
async function resetPassword(username: string | undefined): Promise<void> {
|
||||
fs.mkdirSync(config.dataDir, { recursive: true, mode: 0o700 });
|
||||
const db = new Db(dataPath('hookpilot.db'));
|
||||
try {
|
||||
const provided = process.env.HOOKPILOT_NEW_PASSWORD;
|
||||
if (provided !== undefined && provided.length < MIN_PASSWORD_LENGTH) {
|
||||
throw new Error(`HOOKPILOT_NEW_PASSWORD 至少需要 ${MIN_PASSWORD_LENGTH} 个字符`);
|
||||
}
|
||||
const password = provided ?? randomToken(12);
|
||||
const name = username ?? getAdmin(db)?.username ?? config.initialAdminUser;
|
||||
await setAdminCredentials(db, name, password);
|
||||
deleteOtherSessions(db, null);
|
||||
console.log(`管理员 ${name} 的密码已重置。`);
|
||||
if (!provided) console.log(`新密码：${password}`);
|
||||
} finally {
|
||||
db.close();
|
||||
}
|
||||
}
|
||||
|
||||
const [command, arg] = process.argv.slice(2);
|
||||
if (command === 'reset-password') {
|
||||
resetPassword(arg).catch((err) => {
|
||||
console.error(err instanceof Error ? err.message : err);
|
||||
process.exit(1);
|
||||
});
|
||||
} else {
|
||||
console.log(USAGE);
|
||||
process.exit(command ? 1 : 0);
|
||||
}
|
||||
@@ -0,0 +1,62 @@
|
||||
import path from 'node:path';
|
||||
import { fileURLToPath } from 'node:url';
|
||||
|
||||
const here = path.dirname(fileURLToPath(import.meta.url));
|
||||
const projectRoot = path.resolve(here, '..', '..');
|
||||
const isWindows = process.platform === 'win32';
|
||||
|
||||
function env(name: string): string | undefined {
|
||||
const value = process.env[name];
|
||||
return value === undefined || value.trim() === '' ? undefined : value.trim();
|
||||
}
|
||||
|
||||
function intEnv(name: string, fallback: number): number {
|
||||
const raw = env(name);
|
||||
if (raw === undefined) return fallback;
|
||||
const value = Number(raw);
|
||||
if (!Number.isInteger(value) || value <= 0) {
|
||||
throw new Error(`环境变量 ${name} 必须是正整数，当前值：${raw}`);
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
||||
function trustProxyEnv(raw: string): boolean | string {
|
||||
if (raw === 'true') return true;
|
||||
if (raw === 'false') return false;
|
||||
return raw;
|
||||
}
|
||||
|
||||
function cookieSecureEnv(raw: string): 'auto' | boolean {
|
||||
if (raw === 'true') return true;
|
||||
if (raw === 'false') return false;
|
||||
return 'auto';
|
||||
}
|
||||
|
||||
export const config = {
|
||||
host: env('HOOKPILOT_HOST') ?? '127.0.0.1',
|
||||
port: intEnv('HOOKPILOT_PORT', 8787),
|
||||
dataDir: path.resolve(env('HOOKPILOT_DATA_DIR') ?? path.join(projectRoot, 'data')),
|
||||
webDir: path.resolve(env('HOOKPILOT_WEB_DIR') ?? path.join(projectRoot, 'web', 'dist')),
|
||||
logLevel: env('HOOKPILOT_LOG_LEVEL') ?? 'info',
|
||||
/** 反向代理信任设置，决定如何解析客户端 IP 与协议 */
|
||||
trustProxy: trustProxyEnv(env('HOOKPILOT_TRUST_PROXY') ?? 'loopback'),
|
||||
/** Webhook 与管理 API 的请求体上限，默认 1 MiB */
|
||||
bodyLimit: intEnv('HOOKPILOT_BODY_LIMIT', 1024 * 1024),
|
||||
cookieSecure: cookieSecureEnv(env('HOOKPILOT_COOKIE_SECURE') ?? 'auto'),
|
||||
sessionTtlMs: intEnv('HOOKPILOT_SESSION_TTL_HOURS', 168) * 3600 * 1000,
|
||||
bash: env('HOOKPILOT_BASH') ?? (isWindows ? 'bash' : '/bin/bash'),
|
||||
/** 脚本执行时的 PATH，子进程不继承服务自身的完整环境 */
|
||||
execPath:
|
||||
env('HOOKPILOT_EXEC_PATH') ??
|
||||
(isWindows ? (process.env.PATH ?? '') : '/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin'),
|
||||
initialAdminUser: env('HOOKPILOT_ADMIN_USER') ?? 'admin',
|
||||
initialAdminPassword: env('HOOKPILOT_ADMIN_PASSWORD'),
|
||||
/** 终止任务时 SIGTERM 与 SIGKILL 之间的等待时间 */
|
||||
killGraceMs: 3000,
|
||||
};
|
||||
|
||||
export type AppConfig = typeof config;
|
||||
|
||||
export function dataPath(...parts: string[]): string {
|
||||
return path.join(config.dataDir, ...parts);
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
import { createRequire } from 'node:module';
|
||||
import type { DatabaseSync as DatabaseSyncType, StatementSync } from 'node:sqlite';
|
||||
|
||||
// ESM 会在任何代码执行前链接内置模块，静态 import 会让 warnings.ts 的过滤来不及生效，因此在运行时加载
|
||||
const { DatabaseSync } = createRequire(import.meta.url)('node:sqlite') as typeof import('node:sqlite');
|
||||
|
||||
export type SqlParam = null | number | bigint | string | Uint8Array;
|
||||
|
||||
const MIGRATIONS: string[] = [
|
||||
`
|
||||
CREATE TABLE admin (
|
||||
id INTEGER PRIMARY KEY CHECK (id = 1),
|
||||
username TEXT NOT NULL,
|
||||
password_hash TEXT NOT NULL,
|
||||
updated_at INTEGER NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE sessions (
|
||||
id TEXT PRIMARY KEY,
|
||||
created_at INTEGER NOT NULL,
|
||||
expires_at INTEGER NOT NULL,
|
||||
ip TEXT NOT NULL DEFAULT '',
|
||||
user_agent TEXT NOT NULL DEFAULT ''
|
||||
);
|
||||
|
||||
CREATE TABLE hooks (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
slug TEXT NOT NULL UNIQUE,
|
||||
name TEXT NOT NULL,
|
||||
description TEXT NOT NULL DEFAULT '',
|
||||
group_name TEXT NOT NULL DEFAULT '',
|
||||
method TEXT NOT NULL DEFAULT 'POST' CHECK (method IN ('GET', 'POST')),
|
||||
enabled INTEGER NOT NULL DEFAULT 1,
|
||||
token_hash TEXT NOT NULL,
|
||||
token_hint TEXT NOT NULL DEFAULT '',
|
||||
params TEXT NOT NULL DEFAULT '[]',
|
||||
current_version_id INTEGER,
|
||||
working_dir TEXT NOT NULL DEFAULT '',
|
||||
timeout_sec INTEGER NOT NULL DEFAULT 30,
|
||||
response_mode TEXT NOT NULL DEFAULT 'async' CHECK (response_mode IN ('sync', 'async')),
|
||||
concurrency INTEGER NOT NULL DEFAULT 1,
|
||||
created_at INTEGER NOT NULL,
|
||||
updated_at INTEGER NOT NULL
|
||||
);
|
||||
|
||||
CREATE TABLE script_versions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
hook_id INTEGER NOT NULL REFERENCES hooks(id) ON DELETE CASCADE,
|
||||
version INTEGER NOT NULL,
|
||||
content TEXT NOT NULL,
|
||||
created_at INTEGER NOT NULL,
|
||||
UNIQUE (hook_id, version)
|
||||
);
|
||||
|
||||
-- 任务表兼作持久化队列，保存参数、请求与执行配置快照
|
||||
CREATE TABLE executions (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
hook_id INTEGER NOT NULL REFERENCES hooks(id) ON DELETE CASCADE,
|
||||
script_version_id INTEGER NOT NULL,
|
||||
trigger_type TEXT NOT NULL,
|
||||
source_execution_id INTEGER,
|
||||
status TEXT NOT NULL,
|
||||
env TEXT NOT NULL,
|
||||
param_defs TEXT NOT NULL,
|
||||
request TEXT NOT NULL,
|
||||
config TEXT NOT NULL,
|
||||
request_ip TEXT NOT NULL DEFAULT '',
|
||||
exit_code INTEGER,
|
||||
signal TEXT,
|
||||
error TEXT,
|
||||
stdout_bytes INTEGER NOT NULL DEFAULT 0,
|
||||
stderr_bytes INTEGER NOT NULL DEFAULT 0,
|
||||
output_truncated INTEGER NOT NULL DEFAULT 0,
|
||||
created_at INTEGER NOT NULL,
|
||||
started_at INTEGER,
|
||||
finished_at INTEGER
|
||||
);
|
||||
CREATE INDEX idx_executions_status ON executions (status, id);
|
||||
CREATE INDEX idx_executions_hook ON executions (hook_id, id);
|
||||
CREATE INDEX idx_executions_finished ON executions (finished_at);
|
||||
|
||||
CREATE TABLE execution_output (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
execution_id INTEGER NOT NULL REFERENCES executions(id) ON DELETE CASCADE,
|
||||
stream TEXT NOT NULL CHECK (stream IN ('stdout', 'stderr')),
|
||||
byte_offset INTEGER NOT NULL,
|
||||
data TEXT NOT NULL
|
||||
);
|
||||
CREATE INDEX idx_execution_output ON execution_output (execution_id, stream, byte_offset);
|
||||
|
||||
CREATE TABLE settings (
|
||||
key TEXT PRIMARY KEY,
|
||||
value TEXT NOT NULL
|
||||
);
|
||||
`,
|
||||
];
|
||||
|
||||
export class Db {
|
||||
readonly raw: DatabaseSyncType;
|
||||
private readonly statements = new Map<string, StatementSync>();
|
||||
private inTransaction = false;
|
||||
|
||||
constructor(file: string) {
|
||||
this.raw = new DatabaseSync(file);
|
||||
this.raw.exec('PRAGMA journal_mode = WAL');
|
||||
this.raw.exec('PRAGMA synchronous = NORMAL');
|
||||
this.raw.exec('PRAGMA foreign_keys = ON');
|
||||
this.raw.exec('PRAGMA busy_timeout = 5000');
|
||||
this.migrate();
|
||||
}
|
||||
|
||||
private prepare(sql: string): StatementSync {
|
||||
let statement = this.statements.get(sql);
|
||||
if (!statement) {
|
||||
statement = this.raw.prepare(sql);
|
||||
this.statements.set(sql, statement);
|
||||
}
|
||||
return statement;
|
||||
}
|
||||
|
||||
get<T>(sql: string, ...params: SqlParam[]): T | undefined {
|
||||
return this.prepare(sql).get(...params) as T | undefined;
|
||||
}
|
||||
|
||||
all<T>(sql: string, ...params: SqlParam[]): T[] {
|
||||
return this.prepare(sql).all(...params) as T[];
|
||||
}
|
||||
|
||||
run(sql: string, ...params: SqlParam[]): { changes: number; lastInsertRowid: number } {
|
||||
const result = this.prepare(sql).run(...params);
|
||||
return { changes: Number(result.changes), lastInsertRowid: Number(result.lastInsertRowid) };
|
||||
}
|
||||
|
||||
exec(sql: string): void {
|
||||
this.raw.exec(sql);
|
||||
}
|
||||
|
||||
/** 同步事务；嵌套调用时并入外层事务 */
|
||||
transaction<T>(fn: () => T): T {
|
||||
if (this.inTransaction) return fn();
|
||||
this.raw.exec('BEGIN IMMEDIATE');
|
||||
this.inTransaction = true;
|
||||
try {
|
||||
const result = fn();
|
||||
this.raw.exec('COMMIT');
|
||||
return result;
|
||||
} catch (err) {
|
||||
this.raw.exec('ROLLBACK');
|
||||
throw err;
|
||||
} finally {
|
||||
this.inTransaction = false;
|
||||
}
|
||||
}
|
||||
|
||||
close(): void {
|
||||
this.raw.close();
|
||||
}
|
||||
|
||||
private migrate(): void {
|
||||
const current = this.get<{ user_version: number }>('PRAGMA user_version')?.user_version ?? 0;
|
||||
for (let i = current; i < MIGRATIONS.length; i++) {
|
||||
this.transaction(() => {
|
||||
this.raw.exec(MIGRATIONS[i]!);
|
||||
this.raw.exec(`PRAGMA user_version = ${i + 1}`);
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,300 @@
|
||||
import { spawn, type ChildProcess } from 'node:child_process';
|
||||
import fs from 'node:fs/promises';
|
||||
import os from 'node:os';
|
||||
import path from 'node:path';
|
||||
import { StringDecoder } from 'node:string_decoder';
|
||||
import type { RequestSnapshot } from '../hooks/params.js';
|
||||
import type { ExecutionOutcome, FinalStatus, OutputStream } from './store.js';
|
||||
|
||||
const isWindows = process.platform === 'win32';
|
||||
/** 主进程退出后等待输出管道关闭的时间；后台子进程仍占用管道时不再无限等待 */
|
||||
const PIPE_CLOSE_WAIT_MS = 2000;
|
||||
const FLUSH_INTERVAL_MS = 250;
|
||||
const FLUSH_THRESHOLD_CHARS = 64 * 1024;
|
||||
const WINDOWS_ENV_PASSTHROUGH = ['SYSTEMROOT', 'WINDIR', 'TEMP', 'TMP', 'USERPROFILE', 'COMSPEC', 'PATHEXT', 'APPDATA', 'LOCALAPPDATA'];
|
||||
|
||||
export type KillReason = 'timeout' | 'killed' | 'shutdown';
|
||||
|
||||
export interface RunSpec {
|
||||
executionId: number;
|
||||
hookSlug: string;
|
||||
trigger: string;
|
||||
scriptPath: string;
|
||||
/** 参数映射得到的 HOOK_ 变量 */
|
||||
env: Record<string, string>;
|
||||
request: RequestSnapshot;
|
||||
workingDir: string;
|
||||
timeoutSec: number;
|
||||
outputLimitBytes: number;
|
||||
}
|
||||
|
||||
export interface RunnerOptions {
|
||||
bash: string;
|
||||
execPath: string;
|
||||
dataDir: string;
|
||||
killGraceMs: number;
|
||||
writeOutput: (stream: OutputStream, offset: number, data: string) => void;
|
||||
onError: (err: unknown) => void;
|
||||
}
|
||||
|
||||
/**
|
||||
* 按块缓存输出并写入数据库。超出上限后继续读取并丢弃，避免管道写满阻塞脚本。
|
||||
* 偏移量按已保存文本的 UTF-8 字节数计算，块边界不会切断多字节字符。
|
||||
*/
|
||||
class OutputCollector {
|
||||
private readonly decoders = { stdout: new StringDecoder('utf8'), stderr: new StringDecoder('utf8') };
|
||||
private readonly pending = { stdout: '', stderr: '' };
|
||||
private readonly offsets = { stdout: 0, stderr: 0 };
|
||||
private stored = 0;
|
||||
private timer: NodeJS.Timeout | null = null;
|
||||
private closed = false;
|
||||
truncated = false;
|
||||
|
||||
constructor(
|
||||
private readonly limit: number,
|
||||
private readonly write: RunnerOptions['writeOutput'],
|
||||
private readonly onError: RunnerOptions['onError'],
|
||||
) {}
|
||||
|
||||
push(stream: OutputStream, chunk: Buffer): void {
|
||||
if (this.closed || this.truncated) return;
|
||||
let data = chunk;
|
||||
const remaining = this.limit - this.stored;
|
||||
if (data.length > remaining) {
|
||||
data = data.subarray(0, remaining);
|
||||
this.truncated = true;
|
||||
}
|
||||
this.stored += data.length;
|
||||
this.pending[stream] += this.decoders[stream].write(data);
|
||||
if (this.pending[stream].length >= FLUSH_THRESHOLD_CHARS) this.flush(stream);
|
||||
else if (!this.timer) {
|
||||
this.timer = setTimeout(() => {
|
||||
this.timer = null;
|
||||
this.flushAll();
|
||||
}, FLUSH_INTERVAL_MS);
|
||||
}
|
||||
}
|
||||
|
||||
flushAll(): void {
|
||||
this.flush('stdout');
|
||||
this.flush('stderr');
|
||||
}
|
||||
|
||||
close(): void {
|
||||
if (this.closed) return;
|
||||
if (this.timer) clearTimeout(this.timer);
|
||||
this.timer = null;
|
||||
this.pending.stdout += this.decoders.stdout.end();
|
||||
this.pending.stderr += this.decoders.stderr.end();
|
||||
this.flushAll();
|
||||
this.closed = true;
|
||||
}
|
||||
|
||||
bytes(stream: OutputStream): number {
|
||||
return this.offsets[stream];
|
||||
}
|
||||
|
||||
private flush(stream: OutputStream): void {
|
||||
const text = this.pending[stream];
|
||||
if (!text) return;
|
||||
this.pending[stream] = '';
|
||||
const offset = this.offsets[stream];
|
||||
this.offsets[stream] += Buffer.byteLength(text);
|
||||
try {
|
||||
this.write(stream, offset, text);
|
||||
} catch (err) {
|
||||
this.onError(err);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
export class Runner {
|
||||
private child: ChildProcess | null = null;
|
||||
private killReason: KillReason | null = null;
|
||||
|
||||
constructor(
|
||||
private readonly spec: RunSpec,
|
||||
private readonly options: RunnerOptions,
|
||||
) {}
|
||||
|
||||
/** 终止任务进程组：先 SIGTERM，宽限期后 SIGKILL（即使主进程已退出，也清理残留子进程） */
|
||||
kill(reason: KillReason): void {
|
||||
if (this.killReason) return;
|
||||
this.killReason = reason;
|
||||
if (!this.child || this.child.pid === undefined) return;
|
||||
this.signalGroup('SIGTERM');
|
||||
setTimeout(() => this.signalGroup('SIGKILL'), this.options.killGraceMs).unref();
|
||||
}
|
||||
|
||||
async run(): Promise<ExecutionOutcome> {
|
||||
const output = new OutputCollector(this.spec.outputLimitBytes, this.options.writeOutput, this.options.onError);
|
||||
const runDir = path.join(this.options.dataDir, 'runs', String(this.spec.executionId));
|
||||
try {
|
||||
if (this.killReason) return this.outcome(output, null, null, null);
|
||||
|
||||
await fs.mkdir(runDir, { recursive: true, mode: 0o700 });
|
||||
const requestFile = path.join(runDir, 'request.json');
|
||||
await fs.writeFile(requestFile, JSON.stringify(this.spec.request, null, 2), { mode: 0o600 });
|
||||
|
||||
const cwd = this.spec.workingDir || path.join(this.options.dataDir, 'workspace');
|
||||
const stat = await fs.stat(cwd).catch(() => null);
|
||||
if (!stat?.isDirectory()) return this.outcome(output, null, null, `工作目录不存在或不可访问：${cwd}`);
|
||||
|
||||
return await this.spawnAndWait(cwd, this.buildEnv(requestFile), output);
|
||||
} catch (err) {
|
||||
return this.outcome(output, null, null, err instanceof Error ? err.message : String(err));
|
||||
} finally {
|
||||
output.close();
|
||||
await fs.rm(runDir, { recursive: true, force: true }).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
/** 显式构造子进程环境，不继承服务自身的配置与凭证 */
|
||||
private buildEnv(requestFile: string): Record<string, string> {
|
||||
let username = process.env.USER ?? '';
|
||||
try {
|
||||
username = os.userInfo().username;
|
||||
} catch {
|
||||
// 部分容器环境没有 passwd 记录
|
||||
}
|
||||
const env: Record<string, string> = {
|
||||
PATH: this.options.execPath,
|
||||
HOME: process.env.HOME ?? os.homedir(),
|
||||
USER: username,
|
||||
LOGNAME: username,
|
||||
SHELL: this.options.bash,
|
||||
LANG: process.env.LANG || 'C.UTF-8',
|
||||
};
|
||||
if (process.env.TZ) env.TZ = process.env.TZ;
|
||||
if (isWindows) {
|
||||
for (const key of WINDOWS_ENV_PASSTHROUGH) {
|
||||
const value = process.env[key];
|
||||
if (value) env[key] = value;
|
||||
}
|
||||
}
|
||||
for (const [key, value] of Object.entries(this.spec.env)) {
|
||||
if (key.startsWith('HOOK_')) env[key] = value;
|
||||
}
|
||||
env.HOOK_EXECUTION_ID = String(this.spec.executionId);
|
||||
env.HOOK_SLUG = this.spec.hookSlug;
|
||||
env.HOOK_TRIGGER = this.spec.trigger;
|
||||
env.HOOK_REQUEST_FILE = requestFile;
|
||||
return env;
|
||||
}
|
||||
|
||||
private spawnAndWait(cwd: string, env: Record<string, string>, output: OutputCollector): Promise<ExecutionOutcome> {
|
||||
return new Promise((resolve) => {
|
||||
let child: ChildProcess;
|
||||
try {
|
||||
// 请求数据只通过环境变量传入，脚本文件原样执行，不做任何文本替换
|
||||
child = spawn(this.options.bash, [this.spec.scriptPath], {
|
||||
cwd,
|
||||
env,
|
||||
detached: !isWindows,
|
||||
stdio: ['ignore', 'pipe', 'pipe'],
|
||||
windowsHide: true,
|
||||
});
|
||||
} catch (err) {
|
||||
resolve(this.outcome(output, null, null, `无法启动 Bash：${err instanceof Error ? err.message : String(err)}`));
|
||||
return;
|
||||
}
|
||||
this.child = child;
|
||||
|
||||
// 准备阶段收到的终止请求在进程启动后立即执行
|
||||
if (this.killReason) {
|
||||
const reason = this.killReason;
|
||||
this.killReason = null;
|
||||
this.kill(reason);
|
||||
}
|
||||
|
||||
let exitCode: number | null = null;
|
||||
let signal: string | null = null;
|
||||
let spawnError: string | null = null;
|
||||
let settled = false;
|
||||
let closeTimer: NodeJS.Timeout | undefined;
|
||||
const timeout = setTimeout(() => this.kill('timeout'), this.spec.timeoutSec * 1000);
|
||||
|
||||
const finish = () => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timeout);
|
||||
if (closeTimer) clearTimeout(closeTimer);
|
||||
output.flushAll();
|
||||
resolve(this.outcome(output, exitCode, signal, spawnError ? `无法启动 Bash：${spawnError}` : null));
|
||||
};
|
||||
|
||||
child.stdout?.on('data', (chunk: Buffer) => output.push('stdout', chunk));
|
||||
child.stderr?.on('data', (chunk: Buffer) => output.push('stderr', chunk));
|
||||
child.on('error', (err) => {
|
||||
spawnError = err.message;
|
||||
if (child.pid === undefined) finish();
|
||||
});
|
||||
child.on('exit', (code, sig) => {
|
||||
exitCode = code;
|
||||
signal = sig;
|
||||
closeTimer = setTimeout(() => {
|
||||
child.stdout?.destroy();
|
||||
child.stderr?.destroy();
|
||||
finish();
|
||||
}, PIPE_CLOSE_WAIT_MS);
|
||||
});
|
||||
child.on('close', (code, sig) => {
|
||||
if (exitCode === null && signal === null) {
|
||||
exitCode = code;
|
||||
signal = sig;
|
||||
}
|
||||
finish();
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
private signalGroup(signal: NodeJS.Signals): void {
|
||||
const pid = this.child?.pid;
|
||||
if (pid === undefined) return;
|
||||
if (isWindows) {
|
||||
spawn('taskkill', ['/pid', String(pid), '/T', '/F'], { windowsHide: true, stdio: 'ignore' }).on('error', () => undefined);
|
||||
return;
|
||||
}
|
||||
try {
|
||||
process.kill(-pid, signal);
|
||||
} catch {
|
||||
// 进程组已全部退出
|
||||
}
|
||||
}
|
||||
|
||||
private outcome(
|
||||
output: OutputCollector,
|
||||
exitCode: number | null,
|
||||
signal: string | null,
|
||||
error: string | null,
|
||||
): ExecutionOutcome {
|
||||
let status: FinalStatus;
|
||||
let message = error;
|
||||
switch (this.killReason) {
|
||||
case 'timeout':
|
||||
status = 'timeout';
|
||||
message = `执行超过 ${this.spec.timeoutSec} 秒，已终止进程组`;
|
||||
break;
|
||||
case 'killed':
|
||||
status = 'killed';
|
||||
message = '已被管理员手动终止';
|
||||
break;
|
||||
case 'shutdown':
|
||||
status = 'interrupted';
|
||||
message = '服务停止，执行被中断';
|
||||
break;
|
||||
default:
|
||||
status = !error && exitCode === 0 ? 'success' : 'failed';
|
||||
if (!message && signal) message = `进程被信号 ${signal} 结束`;
|
||||
}
|
||||
return {
|
||||
status,
|
||||
exitCode,
|
||||
signal,
|
||||
error: message,
|
||||
stdoutBytes: output.bytes('stdout'),
|
||||
stderrBytes: output.bytes('stderr'),
|
||||
truncated: output.truncated,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,187 @@
|
||||
import type { AppConfig } from '../config.js';
|
||||
import type { HookRepo } from '../hooks/repo.js';
|
||||
import type { RequestSnapshot } from '../hooks/params.js';
|
||||
import type { SettingsStore } from '../settings.js';
|
||||
import { Runner } from './runner.js';
|
||||
import type { ExecConfig, ExecutionOutcome, ExecutionStore, NewExecution } from './store.js';
|
||||
|
||||
interface RunningTask {
|
||||
hookId: number;
|
||||
runner: Runner;
|
||||
done: Promise<void>;
|
||||
}
|
||||
|
||||
export interface SchedulerDeps {
|
||||
config: AppConfig;
|
||||
store: ExecutionStore;
|
||||
hooks: HookRepo;
|
||||
settings: SettingsStore;
|
||||
log: { error: (obj: unknown, msg: string) => void };
|
||||
}
|
||||
|
||||
function failed(error: string): ExecutionOutcome {
|
||||
return { status: 'failed', exitCode: null, signal: null, error, stdoutBytes: 0, stderrBytes: 0, truncated: false };
|
||||
}
|
||||
|
||||
/**
|
||||
* 任务调度：SQLite 中的排队任务按接收顺序启动，
|
||||
* 同时受全局并发上限与各接口并发数限制。
|
||||
*/
|
||||
export class Scheduler {
|
||||
private readonly running = new Map<number, RunningTask>();
|
||||
private readonly waiters = new Map<number, (() => void)[]>();
|
||||
private scheduled = false;
|
||||
private stopping = false;
|
||||
|
||||
constructor(private readonly deps: SchedulerDeps) {}
|
||||
|
||||
/** 启动时调用，返回被标记为中断的任务数 */
|
||||
recover(): number {
|
||||
return this.deps.store.markInterrupted();
|
||||
}
|
||||
|
||||
enqueue(input: NewExecution): number {
|
||||
const id = this.deps.store.insert(input);
|
||||
this.schedule();
|
||||
return id;
|
||||
}
|
||||
|
||||
schedule(): void {
|
||||
if (this.scheduled || this.stopping) return;
|
||||
this.scheduled = true;
|
||||
setImmediate(() => {
|
||||
this.scheduled = false;
|
||||
try {
|
||||
this.tick();
|
||||
} catch (err) {
|
||||
this.deps.log.error({ err }, '任务调度失败');
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/** 取消排队任务或终止运行中的任务 */
|
||||
cancel(id: number): 'cancelled' | 'killing' | 'inactive' {
|
||||
if (this.deps.store.cancelQueued(id)) {
|
||||
this.notify(id);
|
||||
this.schedule();
|
||||
return 'cancelled';
|
||||
}
|
||||
const task = this.running.get(id);
|
||||
if (task) {
|
||||
task.runner.kill('killed');
|
||||
return 'killing';
|
||||
}
|
||||
return 'inactive';
|
||||
}
|
||||
|
||||
cancelQueuedForHook(hookId: number, reason: string): number {
|
||||
const ids = this.deps.store.cancelQueuedForHook(hookId, reason);
|
||||
ids.forEach((id) => this.notify(id));
|
||||
return ids.length;
|
||||
}
|
||||
|
||||
hasRunning(hookId: number): boolean {
|
||||
for (const task of this.running.values()) if (task.hookId === hookId) return true;
|
||||
return false;
|
||||
}
|
||||
|
||||
/** 等待任务进入终态（成功、失败、取消等） */
|
||||
waitFor(id: number): Promise<void> {
|
||||
const row = this.deps.store.get(id);
|
||||
if (!row || (row.status !== 'queued' && row.status !== 'running')) return Promise.resolve();
|
||||
return new Promise((resolve) => {
|
||||
const list = this.waiters.get(id) ?? [];
|
||||
list.push(resolve);
|
||||
this.waiters.set(id, list);
|
||||
});
|
||||
}
|
||||
|
||||
/** 服务停止：终止运行中的任务并标记为中断，排队任务保留到下次启动 */
|
||||
async shutdown(): Promise<void> {
|
||||
this.stopping = true;
|
||||
const tasks = [...this.running.values()];
|
||||
tasks.forEach((task) => task.runner.kill('shutdown'));
|
||||
const deadline = new Promise((resolve) => setTimeout(resolve, this.deps.config.killGraceMs + 2000).unref());
|
||||
await Promise.race([Promise.all(tasks.map((t) => t.done)), deadline]);
|
||||
}
|
||||
|
||||
private tick(): void {
|
||||
if (this.stopping) return;
|
||||
const limit = this.deps.settings.get().globalConcurrency;
|
||||
if (this.running.size >= limit) return;
|
||||
|
||||
const perHook = new Map<number, number>();
|
||||
for (const task of this.running.values()) perHook.set(task.hookId, (perHook.get(task.hookId) ?? 0) + 1);
|
||||
|
||||
for (const item of this.deps.store.queued()) {
|
||||
if (this.running.size >= limit) break;
|
||||
const count = perHook.get(item.hook_id) ?? 0;
|
||||
if (count >= item.concurrency) continue;
|
||||
if (this.start(item.id)) perHook.set(item.hook_id, count + 1);
|
||||
}
|
||||
}
|
||||
|
||||
private start(id: number): boolean {
|
||||
const { store, hooks, settings, config, log } = this.deps;
|
||||
if (!store.markRunning(id)) return false;
|
||||
|
||||
let runner: Runner;
|
||||
let hookId: number;
|
||||
try {
|
||||
const execution = store.get(id)!;
|
||||
hookId = execution.hook_id;
|
||||
const hook = hooks.get(execution.hook_id);
|
||||
const version = hooks.version(execution.script_version_id);
|
||||
if (!hook || !version) throw new Error('接口或脚本版本不存在');
|
||||
const execConfig = JSON.parse(execution.config) as ExecConfig;
|
||||
runner = new Runner(
|
||||
{
|
||||
executionId: id,
|
||||
hookSlug: hook.slug,
|
||||
trigger: execution.trigger_type,
|
||||
scriptPath: hooks.ensureScriptFile(version),
|
||||
env: JSON.parse(execution.env) as Record<string, string>,
|
||||
request: JSON.parse(execution.request) as RequestSnapshot,
|
||||
workingDir: execConfig.workingDir,
|
||||
timeoutSec: execConfig.timeoutSec,
|
||||
outputLimitBytes: settings.get().outputLimitBytes,
|
||||
},
|
||||
{
|
||||
bash: config.bash,
|
||||
execPath: config.execPath,
|
||||
dataDir: config.dataDir,
|
||||
killGraceMs: config.killGraceMs,
|
||||
writeOutput: (stream, offset, data) => store.appendOutput(id, stream, offset, data),
|
||||
onError: (err) => log.error({ err, executionId: id }, '写入任务输出失败'),
|
||||
},
|
||||
);
|
||||
} catch (err) {
|
||||
store.finish(id, failed(err instanceof Error ? err.message : String(err)));
|
||||
this.notify(id);
|
||||
return false;
|
||||
}
|
||||
|
||||
const done = runner
|
||||
.run()
|
||||
.catch((err: unknown) => failed(err instanceof Error ? err.message : String(err)))
|
||||
.then((outcome) => {
|
||||
try {
|
||||
store.finish(id, outcome);
|
||||
} catch (err) {
|
||||
log.error({ err, executionId: id }, '更新任务状态失败');
|
||||
}
|
||||
this.running.delete(id);
|
||||
this.notify(id);
|
||||
this.schedule();
|
||||
});
|
||||
this.running.set(id, { hookId, runner, done });
|
||||
return true;
|
||||
}
|
||||
|
||||
private notify(id: number): void {
|
||||
const list = this.waiters.get(id);
|
||||
if (!list) return;
|
||||
this.waiters.delete(id);
|
||||
list.forEach((resolve) => resolve());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,285 @@
|
||||
import type { Db, SqlParam } from '../db.js';
|
||||
import type { ParamDef, RequestSnapshot } from '../hooks/params.js';
|
||||
|
||||
export type ExecutionStatus =
|
||||
| 'queued'
|
||||
| 'running'
|
||||
| 'success'
|
||||
| 'failed'
|
||||
| 'timeout'
|
||||
| 'cancelled'
|
||||
| 'killed'
|
||||
| 'interrupted';
|
||||
export type FinalStatus = Exclude<ExecutionStatus, 'queued' | 'running'>;
|
||||
export type TriggerType = 'webhook' | 'test' | 'rerun';
|
||||
export type OutputStream = 'stdout' | 'stderr';
|
||||
|
||||
export const EXECUTION_STATUSES: ExecutionStatus[] = [
|
||||
'queued',
|
||||
'running',
|
||||
'success',
|
||||
'failed',
|
||||
'timeout',
|
||||
'cancelled',
|
||||
'killed',
|
||||
'interrupted',
|
||||
];
|
||||
export const TRIGGER_TYPES: TriggerType[] = ['webhook', 'test', 'rerun'];
|
||||
|
||||
export interface ExecConfig {
|
||||
workingDir: string;
|
||||
timeoutSec: number;
|
||||
responseMode: 'sync' | 'async';
|
||||
}
|
||||
|
||||
export interface ExecutionRow {
|
||||
id: number;
|
||||
hook_id: number;
|
||||
script_version_id: number;
|
||||
trigger_type: TriggerType;
|
||||
source_execution_id: number | null;
|
||||
status: ExecutionStatus;
|
||||
env: string;
|
||||
param_defs: string;
|
||||
request: string;
|
||||
config: string;
|
||||
request_ip: string;
|
||||
exit_code: number | null;
|
||||
signal: string | null;
|
||||
error: string | null;
|
||||
stdout_bytes: number;
|
||||
stderr_bytes: number;
|
||||
output_truncated: number;
|
||||
created_at: number;
|
||||
started_at: number | null;
|
||||
finished_at: number | null;
|
||||
}
|
||||
|
||||
/** 列表与详情查询附带的接口与版本信息 */
|
||||
export interface ExecutionListRow extends ExecutionRow {
|
||||
hook_slug: string;
|
||||
hook_name: string;
|
||||
script_version: number;
|
||||
}
|
||||
|
||||
export interface NewExecution {
|
||||
hookId: number;
|
||||
scriptVersionId: number;
|
||||
trigger: TriggerType;
|
||||
sourceExecutionId: number | null;
|
||||
env: Record<string, string>;
|
||||
paramDefs: ParamDef[];
|
||||
request: RequestSnapshot;
|
||||
config: ExecConfig;
|
||||
requestIp: string;
|
||||
}
|
||||
|
||||
export interface ExecutionOutcome {
|
||||
status: FinalStatus;
|
||||
exitCode: number | null;
|
||||
signal: string | null;
|
||||
error: string | null;
|
||||
stdoutBytes: number;
|
||||
stderrBytes: number;
|
||||
truncated: boolean;
|
||||
}
|
||||
|
||||
export interface ExecutionFilter {
|
||||
hookId?: number;
|
||||
status?: ExecutionStatus;
|
||||
trigger?: TriggerType;
|
||||
}
|
||||
|
||||
const LIST_SELECT = `
|
||||
SELECT e.*, h.slug AS hook_slug, h.name AS hook_name, v.version AS script_version
|
||||
FROM executions e
|
||||
JOIN hooks h ON h.id = e.hook_id
|
||||
JOIN script_versions v ON v.id = e.script_version_id
|
||||
`;
|
||||
|
||||
export class ExecutionStore {
|
||||
constructor(private readonly db: Db) {}
|
||||
|
||||
insert(input: NewExecution): number {
|
||||
const { lastInsertRowid } = this.db.run(
|
||||
`INSERT INTO executions (hook_id, script_version_id, trigger_type, source_execution_id, status, env, param_defs,
|
||||
request, config, request_ip, created_at)
|
||||
VALUES (?, ?, ?, ?, 'queued', ?, ?, ?, ?, ?, ?)`,
|
||||
input.hookId,
|
||||
input.scriptVersionId,
|
||||
input.trigger,
|
||||
input.sourceExecutionId,
|
||||
JSON.stringify(input.env),
|
||||
JSON.stringify(input.paramDefs),
|
||||
JSON.stringify(input.request),
|
||||
JSON.stringify(input.config),
|
||||
input.requestIp,
|
||||
Date.now(),
|
||||
);
|
||||
return lastInsertRowid;
|
||||
}
|
||||
|
||||
get(id: number): ExecutionRow | undefined {
|
||||
return this.db.get<ExecutionRow>('SELECT * FROM executions WHERE id = ?', id);
|
||||
}
|
||||
|
||||
getDetailed(id: number): ExecutionListRow | undefined {
|
||||
return this.db.get<ExecutionListRow>(`${LIST_SELECT} WHERE e.id = ?`, id);
|
||||
}
|
||||
|
||||
list(filter: ExecutionFilter, limit: number, offset: number): { items: ExecutionListRow[]; total: number } {
|
||||
const where: string[] = [];
|
||||
const params: SqlParam[] = [];
|
||||
if (filter.hookId) {
|
||||
where.push('e.hook_id = ?');
|
||||
params.push(filter.hookId);
|
||||
}
|
||||
if (filter.status) {
|
||||
where.push('e.status = ?');
|
||||
params.push(filter.status);
|
||||
}
|
||||
if (filter.trigger) {
|
||||
where.push('e.trigger_type = ?');
|
||||
params.push(filter.trigger);
|
||||
}
|
||||
const clause = where.length ? `WHERE ${where.join(' AND ')}` : '';
|
||||
const total = this.db.get<{ n: number }>(`SELECT COUNT(*) AS n FROM executions e ${clause}`, ...params)?.n ?? 0;
|
||||
const items = this.db.all<ExecutionListRow>(
|
||||
`${LIST_SELECT} ${clause} ORDER BY e.id DESC LIMIT ? OFFSET ?`,
|
||||
...params,
|
||||
limit,
|
||||
offset,
|
||||
);
|
||||
return { items, total };
|
||||
}
|
||||
|
||||
recent(limit: number, statuses?: ExecutionStatus[]): ExecutionListRow[] {
|
||||
if (!statuses?.length) return this.db.all<ExecutionListRow>(`${LIST_SELECT} ORDER BY e.id DESC LIMIT ?`, limit);
|
||||
const marks = statuses.map(() => '?').join(', ');
|
||||
return this.db.all<ExecutionListRow>(
|
||||
`${LIST_SELECT} WHERE e.status IN (${marks}) ORDER BY e.id DESC LIMIT ?`,
|
||||
...statuses,
|
||||
limit,
|
||||
);
|
||||
}
|
||||
|
||||
countByStatus(since?: number): Record<string, number> {
|
||||
const rows = since
|
||||
? this.db.all<{ status: string; n: number }>(
|
||||
'SELECT status, COUNT(*) AS n FROM executions WHERE created_at >= ? GROUP BY status',
|
||||
since,
|
||||
)
|
||||
: this.db.all<{ status: string; n: number }>('SELECT status, COUNT(*) AS n FROM executions GROUP BY status');
|
||||
return Object.fromEntries(rows.map((r) => [r.status, r.n]));
|
||||
}
|
||||
|
||||
/** 按接收顺序列出排队任务及其接口当前的并发配置 */
|
||||
queued(): { id: number; hook_id: number; concurrency: number }[] {
|
||||
return this.db.all(
|
||||
`SELECT e.id, e.hook_id, h.concurrency FROM executions e JOIN hooks h ON h.id = e.hook_id
|
||||
WHERE e.status = 'queued' ORDER BY e.id`,
|
||||
);
|
||||
}
|
||||
|
||||
markRunning(id: number): boolean {
|
||||
return (
|
||||
this.db.run("UPDATE executions SET status = 'running', started_at = ? WHERE id = ? AND status = 'queued'", Date.now(), id)
|
||||
.changes > 0
|
||||
);
|
||||
}
|
||||
|
||||
finish(id: number, outcome: ExecutionOutcome): void {
|
||||
this.db.run(
|
||||
`UPDATE executions SET status = ?, exit_code = ?, signal = ?, error = ?, stdout_bytes = ?, stderr_bytes = ?,
|
||||
output_truncated = ?, finished_at = ?
|
||||
WHERE id = ?`,
|
||||
outcome.status,
|
||||
outcome.exitCode,
|
||||
outcome.signal,
|
||||
outcome.error,
|
||||
outcome.stdoutBytes,
|
||||
outcome.stderrBytes,
|
||||
outcome.truncated ? 1 : 0,
|
||||
Date.now(),
|
||||
id,
|
||||
);
|
||||
}
|
||||
|
||||
cancelQueued(id: number): boolean {
|
||||
return (
|
||||
this.db.run(
|
||||
"UPDATE executions SET status = 'cancelled', finished_at = ?, error = '排队中被取消' WHERE id = ? AND status = 'queued'",
|
||||
Date.now(),
|
||||
id,
|
||||
).changes > 0
|
||||
);
|
||||
}
|
||||
|
||||
cancelQueuedForHook(hookId: number, reason: string): number[] {
|
||||
return this.db.transaction(() => {
|
||||
const ids = this.db
|
||||
.all<{ id: number }>("SELECT id FROM executions WHERE hook_id = ? AND status = 'queued'", hookId)
|
||||
.map((r) => r.id);
|
||||
this.db.run(
|
||||
"UPDATE executions SET status = 'cancelled', finished_at = ?, error = ? WHERE hook_id = ? AND status = 'queued'",
|
||||
Date.now(),
|
||||
reason,
|
||||
hookId,
|
||||
);
|
||||
return ids;
|
||||
});
|
||||
}
|
||||
|
||||
/** 服务启动时调用：上次异常退出时仍在运行的任务标记为中断，不自动重跑 */
|
||||
markInterrupted(): number {
|
||||
return this.db.run(
|
||||
"UPDATE executions SET status = 'interrupted', finished_at = ?, error = '服务重启，执行被中断' WHERE status = 'running'",
|
||||
Date.now(),
|
||||
).changes;
|
||||
}
|
||||
|
||||
appendOutput(id: number, stream: OutputStream, offset: number, data: string): void {
|
||||
this.db.run(
|
||||
'INSERT INTO execution_output (execution_id, stream, byte_offset, data) VALUES (?, ?, ?, ?)',
|
||||
id,
|
||||
stream,
|
||||
offset,
|
||||
data,
|
||||
);
|
||||
}
|
||||
|
||||
/** 从指定字节偏移量起读取输出，单次最多返回约 maxBytes 字节 */
|
||||
readOutput(id: number, stream: OutputStream, offset: number, maxBytes: number): { data: string; nextOffset: number; more: boolean } {
|
||||
const batch = 200;
|
||||
const rows = this.db.all<{ byte_offset: number; data: string }>(
|
||||
`SELECT byte_offset, data FROM execution_output
|
||||
WHERE execution_id = ? AND stream = ? AND byte_offset >= ? ORDER BY byte_offset LIMIT ?`,
|
||||
id,
|
||||
stream,
|
||||
offset,
|
||||
batch,
|
||||
);
|
||||
let data = '';
|
||||
let size = 0;
|
||||
let nextOffset = offset;
|
||||
let more = rows.length === batch;
|
||||
for (const row of rows) {
|
||||
const length = Buffer.byteLength(row.data);
|
||||
if (size > 0 && size + length > maxBytes) {
|
||||
more = true;
|
||||
break;
|
||||
}
|
||||
data += row.data;
|
||||
size += length;
|
||||
nextOffset = row.byte_offset + length;
|
||||
}
|
||||
return { data, nextOffset, more };
|
||||
}
|
||||
|
||||
purgeFinishedBefore(cutoff: number): number {
|
||||
return this.db.run(
|
||||
"DELETE FROM executions WHERE finished_at IS NOT NULL AND finished_at < ? AND status NOT IN ('queued', 'running')",
|
||||
cutoff,
|
||||
).changes;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,326 @@
|
||||
import net from 'node:net';
|
||||
import { badRequest } from '../util/http-error.js';
|
||||
|
||||
export const PARAM_SOURCES = ['json', 'form', 'query', 'header'] as const;
|
||||
export const PARAM_TYPES = ['string', 'number', 'boolean', 'ip'] as const;
|
||||
export type ParamSource = (typeof PARAM_SOURCES)[number];
|
||||
export type ParamType = (typeof PARAM_TYPES)[number];
|
||||
|
||||
export interface ParamDef {
|
||||
/** 脚本中的环境变量名，统一以 HOOK_ 开头 */
|
||||
env: string;
|
||||
source: ParamSource;
|
||||
/** JSON 使用点路径（如 data.ip、items.0.name），其余为字段名或请求头名 */
|
||||
key: string;
|
||||
type: ParamType;
|
||||
required: boolean;
|
||||
default: string;
|
||||
sensitive: boolean;
|
||||
description: string;
|
||||
}
|
||||
|
||||
export interface RequestSnapshot {
|
||||
method: string;
|
||||
contentType: string;
|
||||
query: Record<string, string | string[]>;
|
||||
headers: Record<string, string>;
|
||||
bodyKind: 'json' | 'form' | 'text' | 'none';
|
||||
body: unknown;
|
||||
}
|
||||
|
||||
export interface ParamValue {
|
||||
env: string;
|
||||
value: string;
|
||||
origin: 'request' | 'default' | 'empty';
|
||||
sensitive: boolean;
|
||||
}
|
||||
|
||||
export interface ExtractResult {
|
||||
env: Record<string, string>;
|
||||
values: ParamValue[];
|
||||
errors: string[];
|
||||
}
|
||||
|
||||
/** 平台自动注入的变量，不允许参数占用 */
|
||||
export const RESERVED_ENV = new Set(['HOOK_REQUEST_FILE', 'HOOK_EXECUTION_ID', 'HOOK_SLUG', 'HOOK_TRIGGER']);
|
||||
/** 鉴权相关请求头：不写入执行记录，也不能作为参数来源 */
|
||||
export const SECRET_HEADERS = new Set(['authorization', 'proxy-authorization', 'cookie', 'x-hook-token']);
|
||||
export const TOKEN_QUERY_KEY = 'token';
|
||||
export const MASK = '******';
|
||||
|
||||
const ENV_PATTERN = /^HOOK_[A-Z0-9_]+$/;
|
||||
const SOURCE_LABELS: Record<ParamSource, string> = { json: 'JSON', form: '表单', query: 'Query', header: 'Header' };
|
||||
const TRUE_VALUES = new Set(['true', '1', 'yes', 'on']);
|
||||
const FALSE_VALUES = new Set(['false', '0', 'no', 'off']);
|
||||
const NUMBER_PATTERN = /^[-+]?(\d+(\.\d*)?|\.\d+)([eE][-+]?\d+)?$/;
|
||||
|
||||
type Coerced = { ok: true; value: string } | { ok: false; reason: string };
|
||||
|
||||
function coerce(raw: unknown, type: ParamType): Coerced {
|
||||
switch (type) {
|
||||
case 'string':
|
||||
if (typeof raw === 'string') return { ok: true, value: raw };
|
||||
return { ok: true, value: typeof raw === 'object' ? JSON.stringify(raw) : String(raw) };
|
||||
case 'number':
|
||||
if (typeof raw === 'number' && Number.isFinite(raw)) return { ok: true, value: String(raw) };
|
||||
if (typeof raw === 'string' && NUMBER_PATTERN.test(raw.trim())) return { ok: true, value: raw.trim() };
|
||||
return { ok: false, reason: '不是有效的数字' };
|
||||
case 'boolean': {
|
||||
if (typeof raw === 'boolean') return { ok: true, value: String(raw) };
|
||||
const text = typeof raw === 'string' ? raw.trim().toLowerCase() : '';
|
||||
if (TRUE_VALUES.has(text)) return { ok: true, value: 'true' };
|
||||
if (FALSE_VALUES.has(text)) return { ok: true, value: 'false' };
|
||||
return { ok: false, reason: '不是有效的布尔值（可用 true/false、1/0、yes/no、on/off）' };
|
||||
}
|
||||
case 'ip':
|
||||
if (typeof raw === 'string' && net.isIP(raw.trim()) !== 0) return { ok: true, value: raw.trim() };
|
||||
return { ok: false, reason: '不是合法的 IP 地址' };
|
||||
}
|
||||
}
|
||||
|
||||
export function normalizeParamDefs(input: unknown, method: string): ParamDef[] {
|
||||
if (!Array.isArray(input)) throw badRequest('参数配置格式错误');
|
||||
if (input.length > 100) throw badRequest('参数数量不能超过 100 个');
|
||||
const seen = new Set<string>();
|
||||
|
||||
return input.map((item, index) => {
|
||||
const label = `第 ${index + 1} 个参数`;
|
||||
if (!item || typeof item !== 'object') throw badRequest(`${label}格式错误`);
|
||||
const raw = item as Record<string, unknown>;
|
||||
|
||||
const env = String(raw.env ?? '').trim().toUpperCase();
|
||||
if (!ENV_PATTERN.test(env)) {
|
||||
throw badRequest(`${label}：变量名需以 HOOK_ 开头，且只包含大写字母、数字和下划线`);
|
||||
}
|
||||
if (RESERVED_ENV.has(env)) throw badRequest(`${label}：${env} 是平台保留变量`);
|
||||
if (seen.has(env)) throw badRequest(`${label}：变量名 ${env} 重复`);
|
||||
seen.add(env);
|
||||
|
||||
const source = raw.source as ParamSource;
|
||||
if (!PARAM_SOURCES.includes(source)) throw badRequest(`${label}：参数来源无效`);
|
||||
if (method === 'GET' && (source === 'json' || source === 'form')) {
|
||||
throw badRequest(`${label}：GET 接口只能从 Query 或 Header 提取参数`);
|
||||
}
|
||||
|
||||
let key = String(raw.key ?? '').trim();
|
||||
if (!key || key.length > 200) throw badRequest(`${label}：请填写来源字段名（最多 200 字符）`);
|
||||
if (source === 'header') {
|
||||
key = key.toLowerCase();
|
||||
if (SECRET_HEADERS.has(key)) throw badRequest(`${label}：不能从鉴权相关请求头 ${key} 提取参数`);
|
||||
}
|
||||
if (source === 'query' && key === TOKEN_QUERY_KEY) {
|
||||
throw badRequest(`${label}：Query 字段 token 用于接口鉴权，不能作为参数`);
|
||||
}
|
||||
|
||||
const type = (raw.type ?? 'string') as ParamType;
|
||||
if (!PARAM_TYPES.includes(type)) throw badRequest(`${label}：参数类型无效`);
|
||||
|
||||
const defaultValue = raw.default === undefined || raw.default === null ? '' : String(raw.default);
|
||||
if (defaultValue !== '') {
|
||||
const result = coerce(defaultValue, type);
|
||||
if (!result.ok) throw badRequest(`${label}：默认值${result.reason}`);
|
||||
if (result.value.includes('\0')) throw badRequest(`${label}：默认值包含非法字符`);
|
||||
}
|
||||
|
||||
return {
|
||||
env,
|
||||
source,
|
||||
key,
|
||||
type,
|
||||
required: Boolean(raw.required),
|
||||
default: defaultValue,
|
||||
sensitive: Boolean(raw.sensitive),
|
||||
description: String(raw.description ?? '').slice(0, 500),
|
||||
};
|
||||
});
|
||||
}
|
||||
|
||||
function splitPath(path: string): string[] {
|
||||
return path.split('.').filter((segment) => segment !== '');
|
||||
}
|
||||
|
||||
function readPath(value: unknown, path: string): unknown {
|
||||
let current = value;
|
||||
for (const segment of splitPath(path)) {
|
||||
if (Array.isArray(current) && /^\d+$/.test(segment)) {
|
||||
current = current[Number(segment)];
|
||||
} else if (current !== null && typeof current === 'object' && Object.hasOwn(current, segment)) {
|
||||
current = (current as Record<string, unknown>)[segment];
|
||||
} else {
|
||||
return undefined;
|
||||
}
|
||||
}
|
||||
return current;
|
||||
}
|
||||
|
||||
function first(value: unknown): unknown {
|
||||
return Array.isArray(value) ? value[0] : value;
|
||||
}
|
||||
|
||||
function lookup(def: ParamDef, request: RequestSnapshot): unknown {
|
||||
switch (def.source) {
|
||||
case 'json':
|
||||
return request.bodyKind === 'json' ? readPath(request.body, def.key) : undefined;
|
||||
case 'form': {
|
||||
if (request.bodyKind !== 'form' || !request.body || typeof request.body !== 'object') return undefined;
|
||||
const body = request.body as Record<string, unknown>;
|
||||
return Object.hasOwn(body, def.key) ? first(body[def.key]) : undefined;
|
||||
}
|
||||
case 'query':
|
||||
return Object.hasOwn(request.query, def.key) ? first(request.query[def.key]) : undefined;
|
||||
case 'header':
|
||||
return Object.hasOwn(request.headers, def.key) ? request.headers[def.key] : undefined;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 按参数配置从请求中提取并校验参数。
|
||||
* 每个参数只从管理员指定的来源读取，不在 Body、Query、Header 之间隐式回退。
|
||||
*/
|
||||
export function extractParams(defs: ParamDef[], request: RequestSnapshot): ExtractResult {
|
||||
const env: Record<string, string> = {};
|
||||
const values: ParamValue[] = [];
|
||||
const errors: string[] = [];
|
||||
|
||||
for (const def of defs) {
|
||||
const raw = lookup(def, request);
|
||||
const missing = raw === undefined || raw === null || raw === '';
|
||||
let value = '';
|
||||
let origin: ParamValue['origin'] = 'request';
|
||||
|
||||
if (missing) {
|
||||
if (def.default !== '') {
|
||||
const result = coerce(def.default, def.type);
|
||||
value = result.ok ? result.value : def.default;
|
||||
origin = 'default';
|
||||
} else if (def.required) {
|
||||
errors.push(`${def.env}：缺少必填参数（${SOURCE_LABELS[def.source]} 字段 ${def.key}）`);
|
||||
continue;
|
||||
} else {
|
||||
origin = 'empty';
|
||||
}
|
||||
} else {
|
||||
const result = coerce(raw, def.type);
|
||||
if (!result.ok) {
|
||||
errors.push(`${def.env}：${result.reason}`);
|
||||
continue;
|
||||
}
|
||||
value = result.value;
|
||||
}
|
||||
|
||||
if (value.includes('\0')) {
|
||||
errors.push(`${def.env}：包含非法的空字符`);
|
||||
continue;
|
||||
}
|
||||
env[def.env] = value;
|
||||
values.push({ env: def.env, value, origin, sensitive: def.sensitive });
|
||||
}
|
||||
|
||||
return { env, values, errors };
|
||||
}
|
||||
|
||||
export function maskEnv(env: Record<string, string>, defs: ParamDef[]): Record<string, string> {
|
||||
const sensitive = new Set(defs.filter((d) => d.sensitive).map((d) => d.env));
|
||||
const masked: Record<string, string> = {};
|
||||
for (const [key, value] of Object.entries(env)) {
|
||||
masked[key] = sensitive.has(key) && value !== '' ? MASK : value;
|
||||
}
|
||||
return masked;
|
||||
}
|
||||
|
||||
function maskPath(value: unknown, path: string): void {
|
||||
const segments = splitPath(path);
|
||||
let current = value;
|
||||
for (let i = 0; i < segments.length; i++) {
|
||||
if (current === null || typeof current !== 'object') return;
|
||||
const segment = segments[i]!;
|
||||
const container = current as Record<string, unknown>;
|
||||
if (!Object.hasOwn(container, segment)) return;
|
||||
if (i === segments.length - 1) container[segment] = MASK;
|
||||
else current = container[segment];
|
||||
}
|
||||
}
|
||||
|
||||
/** 返回隐藏了敏感参数原值的请求快照副本，用于在管理界面展示 */
|
||||
export function maskRequest(request: RequestSnapshot, defs: ParamDef[]): RequestSnapshot {
|
||||
const copy = structuredClone(request);
|
||||
for (const def of defs) {
|
||||
if (!def.sensitive) continue;
|
||||
switch (def.source) {
|
||||
case 'json':
|
||||
if (copy.bodyKind === 'json') maskPath(copy.body, def.key);
|
||||
break;
|
||||
case 'form':
|
||||
if (copy.bodyKind === 'form' && copy.body && typeof copy.body === 'object') {
|
||||
const body = copy.body as Record<string, unknown>;
|
||||
if (Object.hasOwn(body, def.key)) body[def.key] = MASK;
|
||||
}
|
||||
break;
|
||||
case 'query':
|
||||
if (Object.hasOwn(copy.query, def.key)) copy.query[def.key] = MASK;
|
||||
break;
|
||||
case 'header':
|
||||
if (Object.hasOwn(copy.headers, def.key)) copy.headers[def.key] = MASK;
|
||||
break;
|
||||
}
|
||||
}
|
||||
return copy;
|
||||
}
|
||||
|
||||
/** 由原始字符串组装请求快照，用于在线测试 */
|
||||
export function buildSnapshot(input: {
|
||||
method: string;
|
||||
query: string;
|
||||
headers: Record<string, string>;
|
||||
bodyType: RequestSnapshot['bodyKind'];
|
||||
body: string;
|
||||
}): RequestSnapshot {
|
||||
const query: Record<string, string | string[]> = {};
|
||||
for (const [key, value] of new URLSearchParams(input.query.replace(/^\?/, ''))) {
|
||||
if (key === TOKEN_QUERY_KEY) continue;
|
||||
appendValue(query, key, value);
|
||||
}
|
||||
|
||||
const headers: Record<string, string> = {};
|
||||
for (const [name, value] of Object.entries(input.headers)) {
|
||||
const key = name.trim().toLowerCase();
|
||||
if (!key || SECRET_HEADERS.has(key)) continue;
|
||||
headers[key] = String(value);
|
||||
}
|
||||
|
||||
let body: unknown = null;
|
||||
let contentType = '';
|
||||
switch (input.bodyType) {
|
||||
case 'json':
|
||||
contentType = 'application/json';
|
||||
try {
|
||||
body = JSON.parse(input.body);
|
||||
} catch {
|
||||
throw badRequest('请求体不是合法的 JSON');
|
||||
}
|
||||
break;
|
||||
case 'form': {
|
||||
contentType = 'application/x-www-form-urlencoded';
|
||||
const form: Record<string, string | string[]> = {};
|
||||
for (const [key, value] of new URLSearchParams(input.body)) appendValue(form, key, value);
|
||||
body = form;
|
||||
break;
|
||||
}
|
||||
case 'text':
|
||||
contentType = 'text/plain';
|
||||
body = input.body;
|
||||
break;
|
||||
case 'none':
|
||||
break;
|
||||
}
|
||||
if (contentType) headers['content-type'] = contentType;
|
||||
|
||||
return { method: input.method, contentType, query, headers, bodyKind: input.bodyType, body };
|
||||
}
|
||||
|
||||
function appendValue(target: Record<string, string | string[]>, key: string, value: string): void {
|
||||
const existing = target[key];
|
||||
if (existing === undefined) target[key] = value;
|
||||
else if (Array.isArray(existing)) existing.push(value);
|
||||
else target[key] = [existing, value];
|
||||
}
|
||||
@@ -0,0 +1,315 @@
|
||||
import fs from 'node:fs';
|
||||
import path from 'node:path';
|
||||
import type { Db } from '../db.js';
|
||||
import { generateHookToken, sha256, tokenHint } from '../util/crypto.js';
|
||||
import { badRequest, conflict, notFound } from '../util/http-error.js';
|
||||
import { normalizeParamDefs, type ParamDef } from './params.js';
|
||||
|
||||
export type HttpMethod = 'GET' | 'POST';
|
||||
export type ResponseMode = 'sync' | 'async';
|
||||
|
||||
export interface HookRow {
|
||||
id: number;
|
||||
slug: string;
|
||||
name: string;
|
||||
description: string;
|
||||
group_name: string;
|
||||
method: HttpMethod;
|
||||
enabled: number;
|
||||
token_hash: string;
|
||||
token_hint: string;
|
||||
params: string;
|
||||
current_version_id: number | null;
|
||||
working_dir: string;
|
||||
timeout_sec: number;
|
||||
response_mode: ResponseMode;
|
||||
concurrency: number;
|
||||
created_at: number;
|
||||
updated_at: number;
|
||||
}
|
||||
|
||||
export interface HookListRow extends HookRow {
|
||||
current_version: number | null;
|
||||
last_status: string | null;
|
||||
last_at: number | null;
|
||||
}
|
||||
|
||||
export interface ScriptVersionRow {
|
||||
id: number;
|
||||
hook_id: number;
|
||||
version: number;
|
||||
content: string;
|
||||
created_at: number;
|
||||
}
|
||||
|
||||
export interface HookInput {
|
||||
slug: string;
|
||||
name: string;
|
||||
description: string;
|
||||
group: string;
|
||||
method: HttpMethod;
|
||||
params: ParamDef[];
|
||||
script: string;
|
||||
workingDir: string;
|
||||
timeoutSec: number;
|
||||
responseMode: ResponseMode;
|
||||
concurrency: number;
|
||||
}
|
||||
|
||||
const SLUG_PATTERN = /^[a-z0-9][a-z0-9_-]{0,63}$/;
|
||||
const MAX_SCRIPT_BYTES = 256 * 1024;
|
||||
|
||||
function intInRange(value: unknown, min: number, max: number, label: string): number {
|
||||
const n = Number(value);
|
||||
if (!Number.isInteger(n) || n < min || n > max) throw badRequest(`${label}需为 ${min} 到 ${max} 之间的整数`);
|
||||
return n;
|
||||
}
|
||||
|
||||
export function validateSlug(value: unknown): string {
|
||||
const slug = String(value ?? '').trim();
|
||||
if (!SLUG_PATTERN.test(slug)) {
|
||||
throw badRequest('接口标识只能包含小写字母、数字、下划线和连字符，以字母或数字开头，最多 64 个字符');
|
||||
}
|
||||
return slug;
|
||||
}
|
||||
|
||||
export function normalizeHookInput(body: unknown): HookInput {
|
||||
if (!body || typeof body !== 'object') throw badRequest('请求内容格式错误');
|
||||
const raw = body as Record<string, unknown>;
|
||||
|
||||
const name = String(raw.name ?? '').trim();
|
||||
if (!name || name.length > 100) throw badRequest('请填写接口名称（最多 100 个字符）');
|
||||
|
||||
const method = String(raw.method ?? 'POST').toUpperCase();
|
||||
if (method !== 'GET' && method !== 'POST') throw badRequest('请求方式只能是 GET 或 POST');
|
||||
|
||||
const script = String(raw.script ?? '').replace(/\r\n?/g, '\n');
|
||||
if (!script.trim()) throw badRequest('脚本内容不能为空');
|
||||
if (Buffer.byteLength(script) > MAX_SCRIPT_BYTES) throw badRequest('脚本内容不能超过 256 KiB');
|
||||
|
||||
const workingDir = String(raw.workingDir ?? '').trim();
|
||||
if (workingDir && !path.isAbsolute(workingDir)) throw badRequest('工作目录必须是绝对路径');
|
||||
|
||||
const responseMode = String(raw.responseMode ?? 'async');
|
||||
if (responseMode !== 'sync' && responseMode !== 'async') throw badRequest('响应模式无效');
|
||||
|
||||
const description = String(raw.description ?? '').trim();
|
||||
if (description.length > 1000) throw badRequest('描述最多 1000 个字符');
|
||||
const group = String(raw.group ?? '').trim();
|
||||
if (group.length > 50) throw badRequest('分组名称最多 50 个字符');
|
||||
|
||||
return {
|
||||
slug: validateSlug(raw.slug),
|
||||
name,
|
||||
description,
|
||||
group,
|
||||
method,
|
||||
params: normalizeParamDefs(raw.params ?? [], method),
|
||||
script,
|
||||
workingDir,
|
||||
timeoutSec: intInRange(raw.timeoutSec ?? 30, 1, 86400, '超时时间（秒）'),
|
||||
responseMode,
|
||||
concurrency: intInRange(raw.concurrency ?? 1, 1, 64, '接口并发数'),
|
||||
};
|
||||
}
|
||||
|
||||
export function parseParamDefs(row: HookRow): ParamDef[] {
|
||||
return JSON.parse(row.params) as ParamDef[];
|
||||
}
|
||||
|
||||
export class HookRepo {
|
||||
constructor(
|
||||
private readonly db: Db,
|
||||
private readonly scriptsDir: string,
|
||||
) {}
|
||||
|
||||
list(): HookListRow[] {
|
||||
return this.db.all<HookListRow>(`
|
||||
SELECT h.*, v.version AS current_version,
|
||||
(SELECT e.status FROM executions e WHERE e.hook_id = h.id ORDER BY e.id DESC LIMIT 1) AS last_status,
|
||||
(SELECT e.created_at FROM executions e WHERE e.hook_id = h.id ORDER BY e.id DESC LIMIT 1) AS last_at
|
||||
FROM hooks h
|
||||
LEFT JOIN script_versions v ON v.id = h.current_version_id
|
||||
ORDER BY h.group_name, h.name, h.id
|
||||
`);
|
||||
}
|
||||
|
||||
get(id: number): HookRow | undefined {
|
||||
return this.db.get<HookRow>('SELECT * FROM hooks WHERE id = ?', id);
|
||||
}
|
||||
|
||||
require(id: number): HookRow {
|
||||
const hook = this.get(id);
|
||||
if (!hook) throw notFound('接口不存在');
|
||||
return hook;
|
||||
}
|
||||
|
||||
getBySlug(slug: string): HookRow | undefined {
|
||||
return this.db.get<HookRow>('SELECT * FROM hooks WHERE slug = ?', slug);
|
||||
}
|
||||
|
||||
groups(): string[] {
|
||||
return this.db
|
||||
.all<{ group_name: string }>("SELECT DISTINCT group_name FROM hooks WHERE group_name <> '' ORDER BY group_name")
|
||||
.map((r) => r.group_name);
|
||||
}
|
||||
|
||||
create(input: HookInput, enabled = true): { hook: HookRow; token: string } {
|
||||
if (this.getBySlug(input.slug)) throw conflict(`接口标识 ${input.slug} 已被使用`);
|
||||
const token = generateHookToken();
|
||||
const now = Date.now();
|
||||
const hook = this.db.transaction(() => {
|
||||
const { lastInsertRowid: id } = this.db.run(
|
||||
`INSERT INTO hooks (slug, name, description, group_name, method, enabled, token_hash, token_hint, params,
|
||||
working_dir, timeout_sec, response_mode, concurrency, created_at, updated_at)
|
||||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
|
||||
input.slug,
|
||||
input.name,
|
||||
input.description,
|
||||
input.group,
|
||||
input.method,
|
||||
enabled ? 1 : 0,
|
||||
sha256(token),
|
||||
tokenHint(token),
|
||||
JSON.stringify(input.params),
|
||||
input.workingDir,
|
||||
input.timeoutSec,
|
||||
input.responseMode,
|
||||
input.concurrency,
|
||||
now,
|
||||
now,
|
||||
);
|
||||
const versionId = this.addVersion(id, input.script);
|
||||
this.db.run('UPDATE hooks SET current_version_id = ? WHERE id = ?', versionId, id);
|
||||
return this.get(id)!;
|
||||
});
|
||||
return { hook, token };
|
||||
}
|
||||
|
||||
/** 更新配置；脚本内容变化时生成新版本，已入队任务仍使用各自固定的版本 */
|
||||
update(id: number, input: HookInput): HookRow {
|
||||
const current = this.require(id);
|
||||
if (input.slug !== current.slug && this.getBySlug(input.slug)) {
|
||||
throw conflict(`接口标识 ${input.slug} 已被使用`);
|
||||
}
|
||||
return this.db.transaction(() => {
|
||||
let versionId = current.current_version_id;
|
||||
const currentVersion = versionId ? this.version(versionId) : undefined;
|
||||
if (!currentVersion || currentVersion.content !== input.script) {
|
||||
versionId = this.addVersion(id, input.script);
|
||||
}
|
||||
this.db.run(
|
||||
`UPDATE hooks SET slug = ?, name = ?, description = ?, group_name = ?, method = ?, params = ?,
|
||||
working_dir = ?, timeout_sec = ?, response_mode = ?, concurrency = ?, current_version_id = ?, updated_at = ?
|
||||
WHERE id = ?`,
|
||||
input.slug,
|
||||
input.name,
|
||||
input.description,
|
||||
input.group,
|
||||
input.method,
|
||||
JSON.stringify(input.params),
|
||||
input.workingDir,
|
||||
input.timeoutSec,
|
||||
input.responseMode,
|
||||
input.concurrency,
|
||||
versionId,
|
||||
Date.now(),
|
||||
id,
|
||||
);
|
||||
return this.get(id)!;
|
||||
});
|
||||
}
|
||||
|
||||
copy(id: number, slug: string, name: string): { hook: HookRow; token: string } {
|
||||
const source = this.require(id);
|
||||
const script = source.current_version_id ? this.version(source.current_version_id)?.content : undefined;
|
||||
return this.create(
|
||||
{
|
||||
slug: validateSlug(slug),
|
||||
name: name.trim() || `${source.name}（副本）`,
|
||||
description: source.description,
|
||||
group: source.group_name,
|
||||
method: source.method,
|
||||
params: parseParamDefs(source),
|
||||
script: script ?? '#!/usr/bin/env bash\n',
|
||||
workingDir: source.working_dir,
|
||||
timeoutSec: source.timeout_sec,
|
||||
responseMode: source.response_mode,
|
||||
concurrency: source.concurrency,
|
||||
},
|
||||
false,
|
||||
);
|
||||
}
|
||||
|
||||
delete(id: number): void {
|
||||
this.require(id);
|
||||
this.db.transaction(() => {
|
||||
this.db.run('DELETE FROM executions WHERE hook_id = ?', id);
|
||||
this.db.run('DELETE FROM script_versions WHERE hook_id = ?', id);
|
||||
this.db.run('DELETE FROM hooks WHERE id = ?', id);
|
||||
});
|
||||
fs.rmSync(path.join(this.scriptsDir, String(id)), { recursive: true, force: true });
|
||||
}
|
||||
|
||||
setEnabled(id: number, enabled: boolean): HookRow {
|
||||
this.require(id);
|
||||
this.db.run('UPDATE hooks SET enabled = ?, updated_at = ? WHERE id = ?', enabled ? 1 : 0, Date.now(), id);
|
||||
return this.get(id)!;
|
||||
}
|
||||
|
||||
regenerateToken(id: number): string {
|
||||
this.require(id);
|
||||
const token = generateHookToken();
|
||||
this.db.run(
|
||||
'UPDATE hooks SET token_hash = ?, token_hint = ?, updated_at = ? WHERE id = ?',
|
||||
sha256(token),
|
||||
tokenHint(token),
|
||||
Date.now(),
|
||||
id,
|
||||
);
|
||||
return token;
|
||||
}
|
||||
|
||||
versions(hookId: number): Omit<ScriptVersionRow, 'content'>[] {
|
||||
return this.db.all(
|
||||
'SELECT id, hook_id, version, created_at FROM script_versions WHERE hook_id = ? ORDER BY version DESC',
|
||||
hookId,
|
||||
);
|
||||
}
|
||||
|
||||
version(id: number): ScriptVersionRow | undefined {
|
||||
return this.db.get<ScriptVersionRow>('SELECT * FROM script_versions WHERE id = ?', id);
|
||||
}
|
||||
|
||||
/** 确保版本对应的脚本文件存在且与数据库内容一致，返回文件路径 */
|
||||
ensureScriptFile(version: ScriptVersionRow): string {
|
||||
const dir = path.join(this.scriptsDir, String(version.hook_id));
|
||||
const file = path.join(dir, `v${version.version}.sh`);
|
||||
let existing: string | null = null;
|
||||
try {
|
||||
existing = fs.readFileSync(file, 'utf8');
|
||||
} catch {
|
||||
existing = null;
|
||||
}
|
||||
if (existing !== version.content) {
|
||||
fs.mkdirSync(dir, { recursive: true, mode: 0o700 });
|
||||
fs.writeFileSync(file, version.content, { mode: 0o700 });
|
||||
}
|
||||
return file;
|
||||
}
|
||||
|
||||
private addVersion(hookId: number, content: string): number {
|
||||
const latest = this.db.get<{ v: number | null }>(
|
||||
'SELECT MAX(version) AS v FROM script_versions WHERE hook_id = ?',
|
||||
hookId,
|
||||
);
|
||||
const { lastInsertRowid } = this.db.run(
|
||||
'INSERT INTO script_versions (hook_id, version, content, created_at) VALUES (?, ?, ?, ?)',
|
||||
hookId,
|
||||
(latest?.v ?? 0) + 1,
|
||||
content,
|
||||
Date.now(),
|
||||
);
|
||||
return lastInsertRowid;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
import './warnings.js';
|
||||
import fs from 'node:fs';
|
||||
import { ensureAdmin, purgeExpiredSessions } from './auth/admin.js';
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import { buildApp } from './app.js';
|
||||
import { config, dataPath } from './config.js';
|
||||
import { Db } from './db.js';
|
||||
import { Scheduler } from './executor/scheduler.js';
|
||||
import { ExecutionStore } from './executor/store.js';
|
||||
import { HookRepo } from './hooks/repo.js';
|
||||
import { SettingsStore } from './settings.js';
|
||||
|
||||
const MAINTENANCE_INTERVAL_MS = 3600 * 1000;
|
||||
|
||||
async function main(): Promise<void> {
|
||||
fs.mkdirSync(config.dataDir, { recursive: true, mode: 0o700 });
|
||||
fs.mkdirSync(dataPath('scripts'), { recursive: true, mode: 0o700 });
|
||||
fs.mkdirSync(dataPath('workspace'), { recursive: true });
|
||||
// 任务临时目录只在执行期间存在，启动时清理上次残留
|
||||
fs.rmSync(dataPath('runs'), { recursive: true, force: true });
|
||||
fs.mkdirSync(dataPath('runs'), { recursive: true, mode: 0o700 });
|
||||
|
||||
const db = new Db(dataPath('hookpilot.db'));
|
||||
const settings = new SettingsStore(db);
|
||||
const hooks = new HookRepo(db, dataPath('scripts'));
|
||||
const store = new ExecutionStore(db);
|
||||
|
||||
// 路由注册时就会取用调度器，因此先创建调度器；其日志在应用构建完成后才会被调用
|
||||
let app: FastifyInstance | null = null;
|
||||
const scheduler = new Scheduler({
|
||||
config,
|
||||
store,
|
||||
hooks,
|
||||
settings,
|
||||
log: { error: (obj: unknown, msg: string) => app?.log.error(obj, msg) },
|
||||
});
|
||||
app = await buildApp({ config, db, settings, hooks, store, scheduler });
|
||||
|
||||
const generatedPassword = await ensureAdmin(db, config.initialAdminUser, config.initialAdminPassword);
|
||||
if (generatedPassword) {
|
||||
app.log.warn(`已创建管理员账户 ${config.initialAdminUser}，初始密码：${generatedPassword}（请登录后立即修改）`);
|
||||
}
|
||||
|
||||
const interrupted = scheduler.recover();
|
||||
if (interrupted) app.log.warn(`${interrupted} 个任务在上次服务停止时仍在运行，已标记为中断`);
|
||||
|
||||
const maintenance = () => {
|
||||
try {
|
||||
const cutoff = Date.now() - settings.get().retentionDays * 86400 * 1000;
|
||||
const purged = store.purgeFinishedBefore(cutoff);
|
||||
purgeExpiredSessions(db);
|
||||
if (purged) app.log.info(`已清理 ${purged} 条过期执行记录`);
|
||||
} catch (err) {
|
||||
app.log.error({ err }, '定期清理失败');
|
||||
}
|
||||
};
|
||||
maintenance();
|
||||
const timer = setInterval(maintenance, MAINTENANCE_INTERVAL_MS);
|
||||
timer.unref();
|
||||
|
||||
await app.listen({ host: config.host, port: config.port });
|
||||
scheduler.schedule();
|
||||
|
||||
let shuttingDown = false;
|
||||
const shutdown = async (signal: string) => {
|
||||
if (shuttingDown) return;
|
||||
shuttingDown = true;
|
||||
app.log.info(`收到 ${signal}，正在停止服务`);
|
||||
clearInterval(timer);
|
||||
await scheduler.shutdown();
|
||||
await app.close();
|
||||
db.close();
|
||||
process.exit(0);
|
||||
};
|
||||
process.on('SIGINT', () => void shutdown('SIGINT'));
|
||||
process.on('SIGTERM', () => void shutdown('SIGTERM'));
|
||||
}
|
||||
|
||||
main().catch((err) => {
|
||||
console.error(err);
|
||||
process.exit(1);
|
||||
});
|
||||
@@ -0,0 +1,98 @@
|
||||
import type { FastifyInstance, FastifyReply, FastifyRequest } from 'fastify';
|
||||
import type { AppContext } from '../app.js';
|
||||
import {
|
||||
createSession,
|
||||
deleteOtherSessions,
|
||||
deleteSession,
|
||||
findSession,
|
||||
getAdmin,
|
||||
MIN_PASSWORD_LENGTH,
|
||||
SESSION_COOKIE,
|
||||
setAdminCredentials,
|
||||
} from '../auth/admin.js';
|
||||
import { verifyPassword } from '../auth/password.js';
|
||||
import { LoginLimiter } from '../auth/rate-limit.js';
|
||||
import { badRequest, HttpError } from '../util/http-error.js';
|
||||
|
||||
function cookieSecure(ctx: AppContext, req: FastifyRequest): boolean {
|
||||
return ctx.config.cookieSecure === 'auto' ? req.protocol === 'https' : ctx.config.cookieSecure;
|
||||
}
|
||||
|
||||
/** 管理 API 鉴权：除标记为 public 的路由外，都要求有效的登录会话 */
|
||||
export function registerSessionGuard(app: FastifyInstance, ctx: AppContext): void {
|
||||
app.addHook('onRequest', async (req: FastifyRequest, reply: FastifyReply) => {
|
||||
if (req.routeOptions.config?.public) return;
|
||||
const token = req.cookies[SESSION_COOKIE];
|
||||
const session = token ? findSession(ctx.db, token) : undefined;
|
||||
if (!session) {
|
||||
return reply.code(401).send({ error: '未登录或登录已过期' });
|
||||
}
|
||||
req.sessionId = session.id;
|
||||
});
|
||||
}
|
||||
|
||||
export async function authRoutes(app: FastifyInstance, ctx: AppContext): Promise<void> {
|
||||
const limiter = new LoginLimiter();
|
||||
|
||||
app.post<{ Body: { username?: string; password?: string } }>(
|
||||
'/auth/login',
|
||||
{ config: { public: true } },
|
||||
async (req, reply) => {
|
||||
const locked = limiter.lockedFor(req.ip);
|
||||
if (locked > 0) throw new HttpError(429, `登录失败次数过多，请 ${Math.ceil(locked / 60)} 分钟后再试`);
|
||||
|
||||
const username = String(req.body?.username ?? '');
|
||||
const password = String(req.body?.password ?? '');
|
||||
const admin = getAdmin(ctx.db);
|
||||
const ok = !!admin && admin.username === username && (await verifyPassword(password, admin.password_hash));
|
||||
if (!ok) {
|
||||
limiter.fail(req.ip);
|
||||
req.log.warn({ ip: req.ip }, '管理员登录失败');
|
||||
throw new HttpError(401, '用户名或密码错误');
|
||||
}
|
||||
limiter.succeed(req.ip);
|
||||
|
||||
const token = createSession(ctx.db, ctx.config.sessionTtlMs, req.ip, String(req.headers['user-agent'] ?? ''));
|
||||
reply.setCookie(SESSION_COOKIE, token, {
|
||||
path: '/',
|
||||
httpOnly: true,
|
||||
sameSite: 'strict',
|
||||
secure: cookieSecure(ctx, req),
|
||||
maxAge: Math.floor(ctx.config.sessionTtlMs / 1000),
|
||||
});
|
||||
return { username: admin!.username };
|
||||
},
|
||||
);
|
||||
|
||||
app.post('/auth/logout', async (req, reply) => {
|
||||
deleteSession(ctx.db, req.sessionId);
|
||||
reply.clearCookie(SESSION_COOKIE, { path: '/' });
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
app.get('/auth/me', async () => {
|
||||
return { username: getAdmin(ctx.db)?.username ?? '' };
|
||||
});
|
||||
|
||||
/** 修改用户名或密码；修改密码后其他会话全部失效 */
|
||||
app.put<{ Body: { currentPassword?: string; username?: string; newPassword?: string } }>(
|
||||
'/auth/account',
|
||||
async (req) => {
|
||||
const admin = getAdmin(ctx.db)!;
|
||||
const currentPassword = String(req.body?.currentPassword ?? '');
|
||||
if (!(await verifyPassword(currentPassword, admin.password_hash))) throw badRequest('当前密码不正确');
|
||||
|
||||
const username = req.body?.username === undefined ? admin.username : String(req.body.username).trim();
|
||||
if (!/^[A-Za-z0-9_.-]{1,32}$/.test(username)) throw badRequest('用户名只能包含字母、数字、下划线、点和连字符，最多 32 个字符');
|
||||
|
||||
const newPassword = req.body?.newPassword ? String(req.body.newPassword) : '';
|
||||
if (newPassword && newPassword.length < MIN_PASSWORD_LENGTH) {
|
||||
throw badRequest(`新密码至少 ${MIN_PASSWORD_LENGTH} 个字符`);
|
||||
}
|
||||
|
||||
await setAdminCredentials(ctx.db, username, newPassword || currentPassword);
|
||||
if (newPassword) deleteOtherSessions(ctx.db, req.sessionId);
|
||||
return { username };
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
import { maskEnv, maskRequest, type ParamDef, type RequestSnapshot } from '../hooks/params.js';
|
||||
import { parseParamDefs, type HookListRow, type HookRow, type ScriptVersionRow } from '../hooks/repo.js';
|
||||
import type { ExecConfig, ExecutionListRow } from '../executor/store.js';
|
||||
|
||||
export function hookDto(row: HookRow, script: ScriptVersionRow | undefined) {
|
||||
return {
|
||||
id: row.id,
|
||||
slug: row.slug,
|
||||
name: row.name,
|
||||
description: row.description,
|
||||
group: row.group_name,
|
||||
method: row.method,
|
||||
enabled: row.enabled === 1,
|
||||
tokenHint: row.token_hint,
|
||||
params: parseParamDefs(row),
|
||||
workingDir: row.working_dir,
|
||||
timeoutSec: row.timeout_sec,
|
||||
responseMode: row.response_mode,
|
||||
concurrency: row.concurrency,
|
||||
currentVersion: script ? { id: script.id, version: script.version, createdAt: script.created_at } : null,
|
||||
script: script?.content ?? '',
|
||||
createdAt: row.created_at,
|
||||
updatedAt: row.updated_at,
|
||||
};
|
||||
}
|
||||
|
||||
export function hookListDto(row: HookListRow) {
|
||||
return {
|
||||
id: row.id,
|
||||
slug: row.slug,
|
||||
name: row.name,
|
||||
description: row.description,
|
||||
group: row.group_name,
|
||||
method: row.method,
|
||||
enabled: row.enabled === 1,
|
||||
responseMode: row.response_mode,
|
||||
concurrency: row.concurrency,
|
||||
currentVersion: row.current_version,
|
||||
lastStatus: row.last_status,
|
||||
lastAt: row.last_at,
|
||||
updatedAt: row.updated_at,
|
||||
};
|
||||
}
|
||||
|
||||
function paramList(row: ExecutionListRow, truncate: number | null) {
|
||||
const defs = JSON.parse(row.param_defs) as ParamDef[];
|
||||
const sensitive = new Set(defs.filter((d) => d.sensitive).map((d) => d.env));
|
||||
const masked = maskEnv(JSON.parse(row.env) as Record<string, string>, defs);
|
||||
return Object.entries(masked).map(([env, value]) => ({
|
||||
env,
|
||||
value: truncate !== null && value.length > truncate ? `${value.slice(0, truncate)}…` : value,
|
||||
sensitive: sensitive.has(env),
|
||||
}));
|
||||
}
|
||||
|
||||
function durationMs(row: ExecutionListRow): number | null {
|
||||
if (!row.started_at) return null;
|
||||
return (row.finished_at ?? Date.now()) - row.started_at;
|
||||
}
|
||||
|
||||
export function executionSummaryDto(row: ExecutionListRow) {
|
||||
return {
|
||||
id: row.id,
|
||||
hookId: row.hook_id,
|
||||
hookSlug: row.hook_slug,
|
||||
hookName: row.hook_name,
|
||||
scriptVersionId: row.script_version_id,
|
||||
scriptVersion: row.script_version,
|
||||
trigger: row.trigger_type,
|
||||
sourceExecutionId: row.source_execution_id,
|
||||
status: row.status,
|
||||
requestIp: row.request_ip,
|
||||
params: paramList(row, 80),
|
||||
exitCode: row.exit_code,
|
||||
signal: row.signal,
|
||||
error: row.error,
|
||||
stdoutBytes: row.stdout_bytes,
|
||||
stderrBytes: row.stderr_bytes,
|
||||
truncated: row.output_truncated === 1,
|
||||
createdAt: row.created_at,
|
||||
startedAt: row.started_at,
|
||||
finishedAt: row.finished_at,
|
||||
durationMs: durationMs(row),
|
||||
};
|
||||
}
|
||||
|
||||
/** 执行详情：敏感参数与请求中对应字段均以掩码展示 */
|
||||
export function executionDetailDto(row: ExecutionListRow, script: ScriptVersionRow | undefined) {
|
||||
const defs = JSON.parse(row.param_defs) as ParamDef[];
|
||||
return {
|
||||
...executionSummaryDto(row),
|
||||
params: paramList(row, null),
|
||||
paramDefs: defs,
|
||||
request: maskRequest(JSON.parse(row.request) as RequestSnapshot, defs),
|
||||
config: JSON.parse(row.config) as ExecConfig,
|
||||
script: script?.content ?? '',
|
||||
};
|
||||
}
|
||||
@@ -0,0 +1,121 @@
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { AppContext } from '../app.js';
|
||||
import {
|
||||
EXECUTION_STATUSES,
|
||||
TRIGGER_TYPES,
|
||||
type ExecutionStatus,
|
||||
type ExecConfig,
|
||||
type TriggerType,
|
||||
} from '../executor/store.js';
|
||||
import { badRequest, notFound } from '../util/http-error.js';
|
||||
import { executionDetailDto, executionSummaryDto } from './dto.js';
|
||||
|
||||
const idParams = {
|
||||
type: 'object',
|
||||
properties: { id: { type: 'integer', minimum: 1 } },
|
||||
required: ['id'],
|
||||
} as const;
|
||||
|
||||
const LOG_CHUNK_BYTES = 256 * 1024;
|
||||
|
||||
export async function executionRoutes(app: FastifyInstance, ctx: AppContext): Promise<void> {
|
||||
const { store, hooks, scheduler } = ctx;
|
||||
|
||||
app.get<{
|
||||
Querystring: { hookId?: number; status?: ExecutionStatus; trigger?: TriggerType; page?: number; pageSize?: number };
|
||||
}>(
|
||||
'/executions',
|
||||
{
|
||||
schema: {
|
||||
querystring: {
|
||||
type: 'object',
|
||||
properties: {
|
||||
hookId: { type: 'integer', minimum: 1 },
|
||||
status: { type: 'string', enum: EXECUTION_STATUSES },
|
||||
trigger: { type: 'string', enum: TRIGGER_TYPES },
|
||||
page: { type: 'integer', minimum: 1, default: 1 },
|
||||
pageSize: { type: 'integer', minimum: 1, maximum: 200, default: 20 },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
async (req) => {
|
||||
const page = req.query.page ?? 1;
|
||||
const pageSize = req.query.pageSize ?? 20;
|
||||
const { items, total } = store.list(
|
||||
{ hookId: req.query.hookId, status: req.query.status, trigger: req.query.trigger },
|
||||
pageSize,
|
||||
(page - 1) * pageSize,
|
||||
);
|
||||
return { items: items.map(executionSummaryDto), total, page, pageSize };
|
||||
},
|
||||
);
|
||||
|
||||
app.get<{ Params: { id: number } }>('/executions/:id', { schema: { params: idParams } }, async (req) => {
|
||||
const row = store.getDetailed(req.params.id);
|
||||
if (!row) throw notFound('执行记录不存在');
|
||||
return executionDetailDto(row, hooks.version(row.script_version_id));
|
||||
});
|
||||
|
||||
/** 按字节偏移量增量读取 stdout / stderr，执行期间可持续轮询 */
|
||||
app.get<{ Params: { id: number }; Querystring: { stdout?: number; stderr?: number } }>(
|
||||
'/executions/:id/logs',
|
||||
{
|
||||
schema: {
|
||||
params: idParams,
|
||||
querystring: {
|
||||
type: 'object',
|
||||
properties: {
|
||||
stdout: { type: 'integer', minimum: 0, default: 0 },
|
||||
stderr: { type: 'integer', minimum: 0, default: 0 },
|
||||
},
|
||||
},
|
||||
},
|
||||
},
|
||||
async (req) => {
|
||||
const row = store.get(req.params.id);
|
||||
if (!row) throw notFound('执行记录不存在');
|
||||
const stdout = store.readOutput(row.id, 'stdout', req.query.stdout ?? 0, LOG_CHUNK_BYTES);
|
||||
const stderr = store.readOutput(row.id, 'stderr', req.query.stderr ?? 0, LOG_CHUNK_BYTES);
|
||||
return {
|
||||
status: row.status,
|
||||
finished: row.status !== 'queued' && row.status !== 'running',
|
||||
truncated: row.output_truncated === 1,
|
||||
stdout,
|
||||
stderr,
|
||||
};
|
||||
},
|
||||
);
|
||||
|
||||
app.post<{ Params: { id: number } }>('/executions/:id/cancel', { schema: { params: idParams } }, async (req) => {
|
||||
if (!store.get(req.params.id)) throw notFound('执行记录不存在');
|
||||
const result = scheduler.cancel(req.params.id);
|
||||
if (result === 'inactive') throw badRequest('任务已结束，无法取消');
|
||||
return { result };
|
||||
});
|
||||
|
||||
/** 使用原脚本版本与参数创建新任务；执行策略取接口当前配置 */
|
||||
app.post<{ Params: { id: number } }>('/executions/:id/rerun', { schema: { params: idParams } }, async (req, reply) => {
|
||||
const source = store.get(req.params.id);
|
||||
if (!source) throw notFound('执行记录不存在');
|
||||
const hook = hooks.require(source.hook_id);
|
||||
const config: ExecConfig = {
|
||||
workingDir: hook.working_dir,
|
||||
timeoutSec: hook.timeout_sec,
|
||||
responseMode: hook.response_mode,
|
||||
};
|
||||
const executionId = scheduler.enqueue({
|
||||
hookId: hook.id,
|
||||
scriptVersionId: source.script_version_id,
|
||||
trigger: 'rerun',
|
||||
sourceExecutionId: source.id,
|
||||
env: JSON.parse(source.env) as Record<string, string>,
|
||||
paramDefs: JSON.parse(source.param_defs),
|
||||
request: JSON.parse(source.request),
|
||||
config,
|
||||
requestIp: req.ip,
|
||||
});
|
||||
reply.code(202);
|
||||
return { executionId };
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,161 @@
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { AppContext } from '../app.js';
|
||||
import { buildSnapshot, extractParams, normalizeParamDefs, type RequestSnapshot } from '../hooks/params.js';
|
||||
import { normalizeHookInput, parseParamDefs } from '../hooks/repo.js';
|
||||
import { badRequest, conflict, notFound } from '../util/http-error.js';
|
||||
import { hookDto, hookListDto } from './dto.js';
|
||||
|
||||
const idParams = {
|
||||
type: 'object',
|
||||
properties: { id: { type: 'integer', minimum: 1 } },
|
||||
required: ['id'],
|
||||
} as const;
|
||||
|
||||
interface TestBody {
|
||||
dryRun?: boolean;
|
||||
/** 仅预览时使用：以尚未保存的参数配置预览映射结果 */
|
||||
params?: unknown;
|
||||
request?: {
|
||||
query?: string;
|
||||
headers?: Record<string, string>;
|
||||
bodyType?: RequestSnapshot['bodyKind'];
|
||||
body?: string;
|
||||
};
|
||||
}
|
||||
|
||||
const BODY_TYPES: RequestSnapshot['bodyKind'][] = ['json', 'form', 'text', 'none'];
|
||||
|
||||
export async function hookRoutes(app: FastifyInstance, ctx: AppContext): Promise<void> {
|
||||
const { hooks, scheduler } = ctx;
|
||||
|
||||
const detail = (id: number) => {
|
||||
const hook = hooks.require(id);
|
||||
return hookDto(hook, hook.current_version_id ? hooks.version(hook.current_version_id) : undefined);
|
||||
};
|
||||
|
||||
app.get('/hooks', async () => {
|
||||
return { items: hooks.list().map(hookListDto), groups: hooks.groups() };
|
||||
});
|
||||
|
||||
app.post('/hooks', async (req, reply) => {
|
||||
const { hook, token } = hooks.create(normalizeHookInput(req.body));
|
||||
reply.code(201);
|
||||
return { hook: detail(hook.id), token };
|
||||
});
|
||||
|
||||
app.get<{ Params: { id: number } }>('/hooks/:id', { schema: { params: idParams } }, async (req) => {
|
||||
return detail(req.params.id);
|
||||
});
|
||||
|
||||
app.put<{ Params: { id: number } }>('/hooks/:id', { schema: { params: idParams } }, async (req) => {
|
||||
hooks.update(req.params.id, normalizeHookInput(req.body));
|
||||
scheduler.schedule();
|
||||
return detail(req.params.id);
|
||||
});
|
||||
|
||||
app.delete<{ Params: { id: number } }>('/hooks/:id', { schema: { params: idParams } }, async (req) => {
|
||||
hooks.require(req.params.id);
|
||||
if (scheduler.hasRunning(req.params.id)) throw conflict('该接口有正在运行的任务，请先终止后再删除');
|
||||
hooks.delete(req.params.id);
|
||||
return { ok: true };
|
||||
});
|
||||
|
||||
app.post<{ Params: { id: number }; Body: { slug?: string; name?: string } }>(
|
||||
'/hooks/:id/copy',
|
||||
{ schema: { params: idParams } },
|
||||
async (req, reply) => {
|
||||
const { hook, token } = hooks.copy(req.params.id, String(req.body?.slug ?? ''), String(req.body?.name ?? ''));
|
||||
reply.code(201);
|
||||
return { hook: detail(hook.id), token };
|
||||
},
|
||||
);
|
||||
|
||||
/** 启用或停用；停用时取消该接口尚未开始的任务 */
|
||||
app.post<{ Params: { id: number }; Body: { enabled?: boolean } }>(
|
||||
'/hooks/:id/enabled',
|
||||
{ schema: { params: idParams } },
|
||||
async (req) => {
|
||||
const enabled = req.body?.enabled === true;
|
||||
hooks.setEnabled(req.params.id, enabled);
|
||||
const cancelled = enabled ? 0 : scheduler.cancelQueuedForHook(req.params.id, '接口已停用，排队任务被取消');
|
||||
return { hook: detail(req.params.id), cancelled };
|
||||
},
|
||||
);
|
||||
|
||||
app.post<{ Params: { id: number } }>('/hooks/:id/token', { schema: { params: idParams } }, async (req) => {
|
||||
return { token: hooks.regenerateToken(req.params.id) };
|
||||
});
|
||||
|
||||
app.get<{ Params: { id: number } }>('/hooks/:id/versions', { schema: { params: idParams } }, async (req) => {
|
||||
const hook = hooks.require(req.params.id);
|
||||
return {
|
||||
currentVersionId: hook.current_version_id,
|
||||
items: hooks.versions(hook.id).map((v) => ({ id: v.id, version: v.version, createdAt: v.created_at })),
|
||||
};
|
||||
});
|
||||
|
||||
app.get<{ Params: { id: number; versionId: number } }>(
|
||||
'/hooks/:id/versions/:versionId',
|
||||
{
|
||||
schema: {
|
||||
params: {
|
||||
type: 'object',
|
||||
properties: { id: { type: 'integer' }, versionId: { type: 'integer' } },
|
||||
required: ['id', 'versionId'],
|
||||
},
|
||||
},
|
||||
},
|
||||
async (req) => {
|
||||
const version = hooks.version(req.params.versionId);
|
||||
if (!version || version.hook_id !== req.params.id) throw notFound('脚本版本不存在');
|
||||
return { id: version.id, version: version.version, content: version.content, createdAt: version.created_at };
|
||||
},
|
||||
);
|
||||
|
||||
/**
|
||||
* 在线测试：dryRun 只预览参数映射；否则使用当前保存的脚本版本真实执行。
|
||||
* 测试任务跳过 Token 校验，其余流程与外部调用一致。
|
||||
*/
|
||||
app.post<{ Params: { id: number }; Body: TestBody }>(
|
||||
'/hooks/:id/test',
|
||||
{ schema: { params: idParams } },
|
||||
async (req, reply) => {
|
||||
const hook = hooks.require(req.params.id);
|
||||
const body = req.body ?? {};
|
||||
const input = body.request ?? {};
|
||||
const bodyType = input.bodyType ?? 'none';
|
||||
if (!BODY_TYPES.includes(bodyType)) throw badRequest('请求体类型无效');
|
||||
|
||||
const snapshot = buildSnapshot({
|
||||
method: hook.method,
|
||||
query: String(input.query ?? ''),
|
||||
headers: input.headers && typeof input.headers === 'object' ? input.headers : {},
|
||||
bodyType,
|
||||
body: String(input.body ?? ''),
|
||||
});
|
||||
const defs =
|
||||
body.dryRun && body.params !== undefined ? normalizeParamDefs(body.params, hook.method) : parseParamDefs(hook);
|
||||
const result = extractParams(defs, snapshot);
|
||||
|
||||
if (body.dryRun) {
|
||||
return { ok: result.errors.length === 0, values: result.values, errors: result.errors, request: snapshot };
|
||||
}
|
||||
if (result.errors.length) throw badRequest('参数校验未通过，脚本未执行', result.errors);
|
||||
if (!hook.current_version_id) throw badRequest('接口尚未保存脚本');
|
||||
|
||||
const executionId = scheduler.enqueue({
|
||||
hookId: hook.id,
|
||||
scriptVersionId: hook.current_version_id,
|
||||
trigger: 'test',
|
||||
sourceExecutionId: null,
|
||||
env: result.env,
|
||||
paramDefs: defs,
|
||||
request: snapshot,
|
||||
config: { workingDir: hook.working_dir, timeoutSec: hook.timeout_sec, responseMode: hook.response_mode },
|
||||
requestIp: req.ip,
|
||||
});
|
||||
reply.code(202);
|
||||
return { executionId };
|
||||
},
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
import type { FastifyInstance } from 'fastify';
|
||||
import type { AppContext } from '../app.js';
|
||||
import { executionSummaryDto } from './dto.js';
|
||||
|
||||
const FAILURE_STATUSES = ['failed', 'timeout', 'killed', 'interrupted'] as const;
|
||||
|
||||
export async function systemRoutes(app: FastifyInstance, ctx: AppContext): Promise<void> {
|
||||
const { db, store, settings, scheduler } = ctx;
|
||||
|
||||
app.get('/overview', async () => {
|
||||
const hookCounts = db.get<{ total: number; enabled: number }>(
|
||||
'SELECT COUNT(*) AS total, COALESCE(SUM(enabled), 0) AS enabled FROM hooks',
|
||||
)!;
|
||||
const active = store.countByStatus();
|
||||
const last24h = store.countByStatus(Date.now() - 24 * 3600 * 1000);
|
||||
return {
|
||||
hooks: hookCounts,
|
||||
running: active.running ?? 0,
|
||||
queued: active.queued ?? 0,
|
||||
last24h: {
|
||||
success: last24h.success ?? 0,
|
||||
failed: FAILURE_STATUSES.reduce((sum, s) => sum + (last24h[s] ?? 0), 0),
|
||||
total: Object.values(last24h).reduce((sum, n) => sum + n, 0),
|
||||
},
|
||||
recent: store.recent(10).map(executionSummaryDto),
|
||||
recentFailures: store.recent(5, [...FAILURE_STATUSES]).map(executionSummaryDto),
|
||||
};
|
||||
});
|
||||
|
||||
app.get('/settings', async () => settings.get());
|
||||
|
||||
app.put('/settings', async (req) => {
|
||||
const updated = settings.update((req.body ?? {}) as Record<string, unknown>);
|
||||
scheduler.schedule();
|
||||
return updated;
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,139 @@
|
||||
import formbody from '@fastify/formbody';
|
||||
import type { FastifyError, FastifyInstance, FastifyRequest } from 'fastify';
|
||||
import type { AppContext } from '../app.js';
|
||||
import type { ExecutionStatus } from '../executor/store.js';
|
||||
import { extractParams, SECRET_HEADERS, TOKEN_QUERY_KEY, type RequestSnapshot } from '../hooks/params.js';
|
||||
import { parseParamDefs } from '../hooks/repo.js';
|
||||
import { safeEqual, sha256 } from '../util/crypto.js';
|
||||
|
||||
/** 同步模式下各终态对应的 HTTP 状态码 */
|
||||
const SYNC_STATUS_CODES: Partial<Record<ExecutionStatus, number>> = {
|
||||
success: 200,
|
||||
failed: 500,
|
||||
timeout: 504,
|
||||
killed: 409,
|
||||
cancelled: 409,
|
||||
interrupted: 503,
|
||||
};
|
||||
|
||||
function extractToken(req: FastifyRequest): string {
|
||||
const auth = req.headers.authorization;
|
||||
if (typeof auth === 'string' && /^bearer\s+/i.test(auth)) return auth.replace(/^bearer\s+/i, '').trim();
|
||||
const header = req.headers['x-hook-token'];
|
||||
if (typeof header === 'string' && header) return header.trim();
|
||||
const query = (req.query as Record<string, unknown>)[TOKEN_QUERY_KEY];
|
||||
return typeof query === 'string' ? query : '';
|
||||
}
|
||||
|
||||
/** 请求快照：不包含鉴权头、Cookie 与 query 中的 token */
|
||||
function snapshotRequest(req: FastifyRequest): RequestSnapshot {
|
||||
const contentType = String(req.headers['content-type'] ?? '').toLowerCase();
|
||||
|
||||
const query: Record<string, string | string[]> = {};
|
||||
for (const [key, value] of Object.entries((req.query ?? {}) as Record<string, unknown>)) {
|
||||
if (key === TOKEN_QUERY_KEY || value === undefined) continue;
|
||||
query[key] = Array.isArray(value) ? value.map(String) : String(value);
|
||||
}
|
||||
|
||||
const headers: Record<string, string> = {};
|
||||
for (const [key, value] of Object.entries(req.headers)) {
|
||||
if (SECRET_HEADERS.has(key) || value === undefined) continue;
|
||||
headers[key] = Array.isArray(value) ? value.join(', ') : String(value);
|
||||
}
|
||||
|
||||
let body: unknown = req.body ?? null;
|
||||
let bodyKind: RequestSnapshot['bodyKind'];
|
||||
if (body === null || body === '') {
|
||||
body = null;
|
||||
bodyKind = 'none';
|
||||
} else if (contentType.includes('json') && typeof body === 'object') {
|
||||
bodyKind = 'json';
|
||||
} else if (contentType.includes('application/x-www-form-urlencoded') && typeof body === 'object') {
|
||||
bodyKind = 'form';
|
||||
body = { ...(body as Record<string, unknown>) };
|
||||
} else {
|
||||
bodyKind = 'text';
|
||||
body = Buffer.isBuffer(body) ? body.toString('utf8') : typeof body === 'string' ? body : JSON.stringify(body);
|
||||
}
|
||||
|
||||
return { method: req.method, contentType, query, headers, bodyKind, body };
|
||||
}
|
||||
|
||||
export async function webhookRoutes(app: FastifyInstance, ctx: AppContext): Promise<void> {
|
||||
const { hooks, scheduler, store } = ctx;
|
||||
|
||||
await app.register(formbody);
|
||||
// 其他类型的请求体按文本接收，脚本可通过 HOOK_REQUEST_FILE 读取
|
||||
app.addContentTypeParser('*', { parseAs: 'string' }, (_req, body, done) => done(null, body));
|
||||
|
||||
app.setErrorHandler((err: FastifyError, req, reply) => {
|
||||
const statusCode = err.statusCode && err.statusCode >= 400 && err.statusCode < 500 ? err.statusCode : 500;
|
||||
if (statusCode === 500) req.log.error({ err }, 'Webhook 处理失败');
|
||||
reply.code(statusCode).send({
|
||||
ok: false,
|
||||
error: statusCode === 413 ? 'payload_too_large' : statusCode === 500 ? 'internal_error' : 'bad_request',
|
||||
message: statusCode === 500 ? '服务器内部错误' : err.message,
|
||||
});
|
||||
});
|
||||
|
||||
app.route<{ Params: { slug: string } }>({
|
||||
method: ['GET', 'POST', 'PUT', 'PATCH', 'DELETE'],
|
||||
url: '/hooks/:slug',
|
||||
handler: async (req, reply) => {
|
||||
// 1. 查找配置，检查启用状态与请求方式
|
||||
const hook = hooks.getBySlug(req.params.slug);
|
||||
if (!hook) return reply.code(404).send({ ok: false, error: 'not_found', message: '接口不存在' });
|
||||
if (!hook.enabled) return reply.code(403).send({ ok: false, error: 'disabled', message: '接口已停用' });
|
||||
if (req.method !== hook.method) {
|
||||
reply.header('Allow', hook.method);
|
||||
return reply.code(405).send({ ok: false, error: 'method_not_allowed', message: `该接口只接受 ${hook.method} 请求` });
|
||||
}
|
||||
|
||||
// 2. 校验 Token（请求大小由 bodyLimit 限制）
|
||||
const token = extractToken(req);
|
||||
if (!token || !safeEqual(sha256(token), hook.token_hash)) {
|
||||
return reply.code(401).send({ ok: false, error: 'unauthorized', message: 'Token 无效' });
|
||||
}
|
||||
|
||||
// 3. 提取并校验参数
|
||||
const snapshot = snapshotRequest(req);
|
||||
const defs = parseParamDefs(hook);
|
||||
const result = extractParams(defs, snapshot);
|
||||
if (result.errors.length) {
|
||||
return reply.code(400).send({ ok: false, error: 'invalid_params', message: '参数校验未通过', details: result.errors });
|
||||
}
|
||||
if (!hook.current_version_id) {
|
||||
return reply.code(500).send({ ok: false, error: 'no_script', message: '接口尚未配置脚本' });
|
||||
}
|
||||
|
||||
// 4. 持久化任务，固定参数与脚本版本；5. 由调度器按并发限制执行
|
||||
const executionId = scheduler.enqueue({
|
||||
hookId: hook.id,
|
||||
scriptVersionId: hook.current_version_id,
|
||||
trigger: 'webhook',
|
||||
sourceExecutionId: null,
|
||||
env: result.env,
|
||||
paramDefs: defs,
|
||||
request: snapshot,
|
||||
config: { workingDir: hook.working_dir, timeoutSec: hook.timeout_sec, responseMode: hook.response_mode },
|
||||
requestIp: req.ip,
|
||||
});
|
||||
|
||||
// 7. 异步模式在任务持久化后立即返回
|
||||
if (hook.response_mode === 'async') {
|
||||
return reply.code(202).send({ ok: true, executionId, status: 'queued' });
|
||||
}
|
||||
|
||||
await scheduler.waitFor(executionId);
|
||||
const row = store.get(executionId)!;
|
||||
const durationMs = row.started_at && row.finished_at ? row.finished_at - row.started_at : null;
|
||||
return reply.code(SYNC_STATUS_CODES[row.status] ?? 500).send({
|
||||
ok: row.status === 'success',
|
||||
executionId,
|
||||
status: row.status,
|
||||
exitCode: row.exit_code,
|
||||
durationMs,
|
||||
});
|
||||
},
|
||||
});
|
||||
}
|
||||
@@ -0,0 +1,72 @@
|
||||
import type { Db } from './db.js';
|
||||
import { badRequest } from './util/http-error.js';
|
||||
|
||||
export interface Settings {
|
||||
/** 全局同时运行的任务上限 */
|
||||
globalConcurrency: number;
|
||||
/** 执行记录保留天数 */
|
||||
retentionDays: number;
|
||||
/** 单次执行输出保留上限（字节，stdout 与 stderr 合计） */
|
||||
outputLimitBytes: number;
|
||||
}
|
||||
|
||||
export const DEFAULT_SETTINGS: Settings = {
|
||||
globalConcurrency: 4,
|
||||
retentionDays: 30,
|
||||
outputLimitBytes: 5 * 1024 * 1024,
|
||||
};
|
||||
|
||||
const RANGES: Record<keyof Settings, [min: number, max: number, label: string]> = {
|
||||
globalConcurrency: [1, 64, '全局并发上限'],
|
||||
retentionDays: [1, 3650, '日志保留天数'],
|
||||
outputLimitBytes: [64 * 1024, 100 * 1024 * 1024, '任务输出大小上限'],
|
||||
};
|
||||
|
||||
const KEYS = Object.keys(DEFAULT_SETTINGS) as (keyof Settings)[];
|
||||
|
||||
export class SettingsStore {
|
||||
private cache: Settings;
|
||||
|
||||
constructor(private readonly db: Db) {
|
||||
this.cache = this.load();
|
||||
}
|
||||
|
||||
get(): Settings {
|
||||
return { ...this.cache };
|
||||
}
|
||||
|
||||
update(patch: Record<string, unknown>): Settings {
|
||||
const next = { ...this.cache };
|
||||
for (const key of KEYS) {
|
||||
if (patch[key] === undefined) continue;
|
||||
const value = Number(patch[key]);
|
||||
const [min, max, label] = RANGES[key];
|
||||
if (!Number.isInteger(value) || value < min || value > max) {
|
||||
throw badRequest(`${label}需为 ${min} 到 ${max} 之间的整数`);
|
||||
}
|
||||
next[key] = value;
|
||||
}
|
||||
this.db.transaction(() => {
|
||||
for (const key of KEYS) {
|
||||
this.db.run(
|
||||
'INSERT INTO settings (key, value) VALUES (?, ?) ON CONFLICT (key) DO UPDATE SET value = excluded.value',
|
||||
key,
|
||||
JSON.stringify(next[key]),
|
||||
);
|
||||
}
|
||||
});
|
||||
this.cache = next;
|
||||
return this.get();
|
||||
}
|
||||
|
||||
private load(): Settings {
|
||||
const settings = { ...DEFAULT_SETTINGS };
|
||||
for (const row of this.db.all<{ key: string; value: string }>('SELECT key, value FROM settings')) {
|
||||
if ((KEYS as string[]).includes(row.key)) {
|
||||
const value = Number(JSON.parse(row.value));
|
||||
if (Number.isInteger(value)) settings[row.key as keyof Settings] = value;
|
||||
}
|
||||
}
|
||||
return settings;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,24 @@
|
||||
import { createHash, randomBytes, timingSafeEqual } from 'node:crypto';
|
||||
|
||||
export function sha256(value: string): string {
|
||||
return createHash('sha256').update(value).digest('hex');
|
||||
}
|
||||
|
||||
export function randomToken(bytes = 24): string {
|
||||
return randomBytes(bytes).toString('base64url');
|
||||
}
|
||||
|
||||
export function safeEqual(a: string, b: string): boolean {
|
||||
const left = Buffer.from(a);
|
||||
const right = Buffer.from(b);
|
||||
return left.length === right.length && timingSafeEqual(left, right);
|
||||
}
|
||||
|
||||
/** Webhook 调用凭证，数据库只保存其哈希 */
|
||||
export function generateHookToken(): string {
|
||||
return `hp_${randomToken(24)}`;
|
||||
}
|
||||
|
||||
export function tokenHint(token: string): string {
|
||||
return token.slice(-4);
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
export class HttpError extends Error {
|
||||
constructor(
|
||||
public readonly statusCode: number,
|
||||
message: string,
|
||||
public readonly details?: unknown,
|
||||
) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
|
||||
export const badRequest = (message: string, details?: unknown) => new HttpError(400, message, details);
|
||||
export const notFound = (message = '资源不存在') => new HttpError(404, message);
|
||||
export const conflict = (message: string) => new HttpError(409, message);
|
||||
@@ -0,0 +1,11 @@
|
||||
// node:sqlite 在 Node 22 中仍标记为实验特性，只屏蔽它的启动警告，其余警告照常输出。
|
||||
// 必须作为入口的第一个 import，才能在 node:sqlite 加载前生效。
|
||||
const originalEmitWarning = process.emitWarning.bind(process) as (...args: unknown[]) => void;
|
||||
|
||||
process.emitWarning = ((warning: string | Error, ...rest: unknown[]) => {
|
||||
const text = typeof warning === 'string' ? warning : warning?.message;
|
||||
if (typeof text === 'string' && text.includes('SQLite is an experimental feature')) return;
|
||||
originalEmitWarning(warning, ...rest);
|
||||
}) as typeof process.emitWarning;
|
||||
|
||||
export {};
|
||||
@@ -0,0 +1,18 @@
|
||||
{
|
||||
"compilerOptions": {
|
||||
"target": "ES2023",
|
||||
"lib": ["ES2023"],
|
||||
"module": "NodeNext",
|
||||
"moduleResolution": "NodeNext",
|
||||
"outDir": "dist",
|
||||
"rootDir": "src",
|
||||
"strict": true,
|
||||
"noImplicitOverride": true,
|
||||
"noFallthroughCasesInSwitch": true,
|
||||
"esModuleInterop": true,
|
||||
"skipLibCheck": true,
|
||||
"sourceMap": true,
|
||||
"types": ["node"]
|
||||
},
|
||||
"include": ["src"]
|
||||
}
|
||||
Reference in New Issue
Block a user