import { spawn } from 'node:child_process' import { createWriteStream } from 'node:fs' import { EventEmitter } from 'node:events' import { createGzip } from 'node:zlib' import { pipeline } from 'node:stream/promises' import { DateTime } from 'luxon' import logger from '@adonisjs/core/services/logger' import AuditLog, { type AuditStatus } from '#models/audit_log' import UserError from '#exceptions/user_error' export interface JobLine { ts: number stream: 'out' | 'err' | 'info' text: string } export interface RunOptions { cwd?: string env?: Record /** * Exit codes considered as a success (default: [0]). */ okCodes?: number[] /** * Text shown in the log instead of the real command (to hide secrets). */ display?: string /** * Writes stdout to this file instead of the log. */ stdoutFile?: string gzipStdout?: boolean timeoutMs?: number } export interface JobContext { log(text: string, stream?: JobLine['stream']): void run(command: string, args: string[], options?: RunOptions): Promise } export interface JobMeta { action: string target: string | null username: string | null ip: string | null /** * Two jobs with the same lock key cannot run at the same time * (for example two deployments of the same project). */ lockKey?: string } /** * Output kept in the action log, to keep the SQLite file small. */ const MAX_STORED_OUTPUT = 200_000 /** * Finished jobs stay in memory for a while so that a page opened late * can still replay the whole output. */ const KEEP_FINISHED_MS = 15 * 60_000 export class Job extends EventEmitter { lines: JobLine[] = [] status: AuditStatus = 'running' constructor( readonly id: number, readonly meta: JobMeta ) { super() this.setMaxListeners(50) } push(line: JobLine) { this.lines.push(line) this.emit('line', line) } get output() { return this.lines.map((line) => line.text).join('\n') } } /** * Runs the long actions (deployments, backups, project creation) in the * background. Each job is recorded in the action log and its output is * streamed to the browser. */ class JobRunner { #jobs = new Map() #locks = new Set() get(id: number) { return this.#jobs.get(id) } isLocked(lockKey: string) { return this.#locks.has(lockKey) } async start(meta: JobMeta, task: (ctx: JobContext) => Promise): Promise { if (meta.lockKey && this.#locks.has(meta.lockKey)) { throw new UserError('Une action est déjà en cours sur cet élément, attends la fin.', 409) } if (meta.lockKey) this.#locks.add(meta.lockKey) const entry = await AuditLog.create({ action: meta.action, target: meta.target, username: meta.username, ip: meta.ip, status: 'running', }) const job = new Job(entry.id, meta) this.#jobs.set(job.id, job) const ctx: JobContext = { log: (text, stream = 'info') => { for (const line of text.split('\n')) job.push({ ts: Date.now(), stream, text: line }) }, run: (command, args, options) => this.#run(job, command, args, options), } /** * Not awaited: the HTTP request returns the job id right away. */ void (async () => { try { await task(ctx) job.status = 'success' ctx.log('✔ Terminé') } catch (error) { job.status = 'error' ctx.log(`✖ ${error instanceof Error ? error.message : String(error)}`, 'err') } finally { if (meta.lockKey) this.#locks.delete(meta.lockKey) await this.#finish(job, entry) job.emit('end', job.status) setTimeout(() => this.#jobs.delete(job.id), KEEP_FINISHED_MS).unref() } })() return job } async #finish(job: Job, entry: AuditLog) { try { const output = job.output entry.status = job.status entry.output = output.length > MAX_STORED_OUTPUT ? `[…] ${output.slice(output.length - MAX_STORED_OUTPUT)}` : output entry.finishedAt = DateTime.now() await entry.save() } catch (error) { logger.error({ err: error }, 'cannot save the job output') } } #run(job: Job, command: string, args: string[], options: RunOptions = {}) { const okCodes = options.okCodes ?? [0] job.push({ ts: Date.now(), stream: 'info', text: `$ ${options.display ?? [command, ...args.map(quote)].join(' ')}`, }) return new Promise((resolve, reject) => { const child = spawn(command, args, { cwd: options.cwd, env: cleanEnv(options.env), stdio: ['ignore', 'pipe', 'pipe'], }) const timeout = setTimeout( () => { job.push({ ts: Date.now(), stream: 'err', text: 'Délai dépassé, commande arrêtée.' }) child.kill('SIGKILL') }, options.timeoutMs ?? 15 * 60_000 ) let fileWritten: Promise = Promise.resolve() if (options.stdoutFile) { const file = createWriteStream(options.stdoutFile) fileWritten = options.gzipStdout ? pipeline(child.stdout, createGzip(), file) : pipeline(child.stdout, file) } else { lineReader(child.stdout, (text) => job.push({ ts: Date.now(), stream: 'out', text })) } lineReader(child.stderr, (text) => job.push({ ts: Date.now(), stream: 'err', text })) child.on('error', (error) => { clearTimeout(timeout) reject(new Error(`Impossible de lancer "${command}" : ${error.message}`)) }) child.on('close', async (code) => { clearTimeout(timeout) try { await fileWritten } catch (error) { return reject(error) } if (code !== null && okCodes.includes(code)) return resolve() reject(new Error(`La commande "${command}" a échoué (code ${code ?? 'tué'}).`)) }) }) } } /** * Variables passed to the commands. The dashboard's own variables (PORT, * APP_KEY…) are left out: "docker compose" would use them to fill the * ${VAR} of the projects' compose files. */ const INHERITED_ENV = [ 'PATH', 'HOME', 'TZ', 'DOCKER_HOST', 'DOCKER_CONFIG', // Windows (local development) 'USERPROFILE', 'APPDATA', 'LOCALAPPDATA', 'ProgramData', 'ProgramFiles', 'SystemRoot', 'TEMP', 'TMP', ] export function cleanEnv(extra: Record = {}) { const env: Record = {} for (const key of INHERITED_ENV) { if (process.env[key] !== undefined) env[key] = process.env[key]! } /** * Never wait for a password: fail right away instead. */ env.GIT_TERMINAL_PROMPT = '0' return { ...env, ...extra } } /** * Calls the callback for each complete line of the stream. */ export function lineReader(stream: NodeJS.ReadableStream, onLine: (line: string) => void) { let rest = '' stream.setEncoding('utf8') stream.on('data', (chunk: string) => { const lines = (rest + chunk).split(/\r?\n/) rest = lines.pop() ?? '' for (const line of lines) onLine(line) }) stream.on('end', () => { if (rest) onLine(rest) rest = '' }) } function quote(arg: string) { return /^[\w@%+=:,./-]+$/.test(arg) ? arg : `'${arg.replace(/'/g, `'\\''`)}'` } const jobRunner = new JobRunner() export default jobRunner